Resequencer¶

Resequencer implements the Resequencer EIP. It buffers out-of-order messages and releases them downstream in monotonically increasing sequence-number order. A configurable timeout lets the resequencer give up on missing sequence numbers and skip past them rather than wait forever.

Processors in this file¶

`NewResequencer`¶

YAML form:

steps:
  - resequence:
      starlang: header("X-Seq")
      batch-size: 0
      timeout: 30s
      steps:
        - target: kafka://ordered-output

Field	Type	Default	Description
`starlang`	string	optional	Expression yielding the sequence value. The reifier writes the resolved sequence to the header `HelixSequenceNumber` before the resequencer reads it.
`batch-size`	int	`0` (unlimited)	Maximum number of out-of-order messages to hold. When the buffer hits this size, additional out-of-order messages return an error.
`timeout`	duration	required	How long to wait for a missing sequence number before skipping past it. After the timeout fires, the smallest buffered sequence is forwarded as the new "next expected".
`steps`	list	required	Inner pipeline to receive messages in order.

Constructor: NewResequencer(processor engine.Processor, timeout time.Duration) *Resequencer. The struct exposes SequenceKey (default HelixSequenceNumber) and Capacity (default 0 = unlimited) for direct configuration.

How it works:

Extracts the sequence number from ex.In().Header(SequenceKey). Accepts int, int64, float64, or numeric strings.
On first call, starts a single background goroutine — the timeout checker — that owns the gap timer. The goroutine uses context.Background so it outlives any individual request context.
If the arriving sequence equals nextExpected:
Collects the current exchange plus any consecutive buffered messages (e.g. arrival of seq=5 also releases buffered 6, 7, 8 if present).
Resets the gap timer (cleared if the buffer is now empty).
Releases the collected run sequentially to the inner Processor.
Otherwise (out of order):
Returns an error if Capacity > 0 and the buffer is full.
Stores the exchange under its sequence key in the buffer and ensures the gap timer is armed.
The timeout checker, on timer fire:
Finds the smallest buffered sequence number.
Sets nextExpected to that number.
Releases the consecutive run starting there to Next (which equals Processor).
Resets or stops the gap timer based on remaining buffer state.

On shutdown, Stop() cancels the background goroutine. The checker drains any remaining buffered messages in order before exiting, using context.Background to keep deliveries valid.

Sequence header¶

By default the resequencer reads HelixSequenceNumber. The reifier writes the result of starlang: into that header before each call. You can also feed it from any upstream system that sets a sequence header — e.g. a Kafka consumer that writes the partition offset, or an HTTP service that includes X-Seq and a set-header step that copies it.

Pitfalls¶

Sequence numbers must be monotonically increasing integers starting at 1. (The initial nextExpected is 1.) Numbers with gaps are buffered; numbers below nextExpected are dropped silently (the buffer overwrites the existing key — first-write-wins is not guaranteed).
The buffer is held in memory. Use batch-size to bound it, and tune timeout to your traffic.
A missing sequence stays in nextExpected until the timer fires. Slow producers can cause head-of-line blocking. Reduce timeout if downstream latency matters more than strict ordering.
The background goroutine does not exit until Stop() is called. Tests should always defer Stop() to avoid leaking goroutines.

Examples¶

workflows:
  - id: replay-in-order
    trigger: { uri: kafka://events?group=resequencer }
    steps:
      - set-header:
          name: HelixSequenceNumber
          starlang: toNumber(header("X-Seq"))
      - resequence:
          timeout: 10s
          batch-size: 1000
          steps:
            - target: postgres://events-ordered

Source¶

pkg/processor/resequencer.go