feat: implement acknowledgment tracking for pipeline records

[Mahesh Kamble (ma-gk)]

What this does

Adds at-least-once acknowledgment for pipeline records: a message is deleted from its
source only once every downstream task has finished with the record produced from it. A
crash mid-pipeline leaves the message in the queue for redelivery instead of losing it.

An ack.Ack is created per source record and rides on record.Record.Context, so tasks
that simply transform a record need no wiring. Tasks that change a record's branch count
adjust it explicitly — Fanout for one-to-many, Joined for many-to-one, Drop for
filtered-out, Reject for failed.

Review feedback addressed

Comment	Resolution
Merge the two identical loops in archive/tar.go	Done — single pass over the archive, buffering entries so the fan-out count is still known before anything is sent
Are we deleting on receive and in deleteOnComplete?	No — the eager delete was only the output == nil branch (no consumer, so no downstream ack to wait for). Both paths now share one deleteMessage
Goroutine per message can grow unboundedly	Bounded, but not by a pool — see Goroutine model below. Being straight about this: it is not the concurrency-sized pool that was asked for
Define an OnComplete in an interface for cleanup	Done — ack.Acknowledger with Ack(failed bool). SQS implements it as messageAck; Kafka can adopt the same tracker

Defects found while auditing, and fixed here

Auditing the original design surfaced problems beyond the review comments. All of these
are reachable from pipeline shapes that exist in data-airflow today.

Two deadlocks. The receive loop was gated on a semaphore of Concurrency*5 released
only on downstream completion, which deadlocks against any fan-in that must accumulate
records before it can emit. Separately, tracker.Wait() ran inside Run, but a task's
output channel closes only after Run returns — and join, archive pack and
sample tail/random emit only when their input closes. So the wait waited for a flush
that could not happen until the wait ended. The semaphore is gone; the wait moved to a new
optional task.Finisher hook the pipeline calls after closing the output channel.

Silent data loss in converter and xpath. Both emitted one record per CSV row /
container node without adjusting the ack, so the first output to complete settled the
whole record — the message was deleted while the remaining rows were still in flight. Both
now count before sending, matching split and archive.

Ack leaks that hang the pipeline. Base.SendRecord dropped the ack when output was
nil, so any task terminating a pipeline without its own guard never settled it. Separately,
14 sites across 13 tasks abandoned the record they were holding on a mid-loop error. One
malformed CSV row was enough to hang a pipeline indefinitely. New ack.Reject /
ack.Rejected settle such a record as failed so the broker redelivers it. Leftovers in a
task's input are rejected by the pipeline once every worker has returned — doing it
inside the task instead discarded records that healthy sibling workers would have written
(measured: 997 of 1998 lost with task_concurrency: 4).

join's duration: never fired. The select had a default case, so it never
blocked; the ticker was only serviced between records — never while input was stalled,
which is exactly when a partial batch needs flushing.

Goroutine model

One goroutine per unacknowledged message, parked on the ack, bounded by channel_size
rather than by a pool. The concurrency semaphore is taken after the ack settles, around
the delete only, so releasing it depends on DeleteMessage returning and never on pipeline
progress — that is what keeps it from recreating the deadlock.

A pool sized by concurrency was implemented first and then reverted: it needed a mutex
and a completion hook on Ack (the least-tested code in the tree) to save goroutines that
were never the problem — 10,000 in flight measured at 118 MB peak, dominated by record
data rather than acks.

Compatibility

No pipeline YAML changes. concurrency keeps the meaning it has on main: main ran
s.Concurrency processReceipts goroutines behind an s.Concurrency * 1000 channel; it
now sizes concurrent deletions. Only the trigger moved — from on-receive to on-completion,
which is the feature.

Verification

Against a binary built from origin/main:

• 20 non-SQS fixtures — exit codes match, 15 byte-identical after stripping echo
  timestamps (incl. join, xpath, xpath_with_index, file, file_with_glob, both
  archive and both compress fixtures). The 5 that differ are nondeterministic on main
  alone (random sampling, UUIDs, DAG interleaving) with matching record counts.
• 6 converter goldens — every *_test_results.json reproduced byte-for-byte; xls, eml
  and protobuf outputs identical.
• Race detector clean; a throwaway harness exercised the ack helpers under -race
  (4000 iterations of registration racing settle) and was deleted.

Against LocalStack 4.0 — seven shapes that hung on the previous commit now exit cleanly
with the queue fully drained: terminal jq, terminal replace, sample(tail),
join(number: 1000, duration: 2s), archive(pack), unwritable sink, and the new fan-out
fixture. Plus three mocks of real data-airflow shapes (ocr/pull_ocr_results,
catalog_sources/catalog_listings, keepa/products/hourly) — all three hung before this
change.

Scale: 10,000 messages in 28s at 38 MB; 10,000 records via fan-out at 57 MB; 10,000
messages all unacknowledged simultaneously (archive pack) at 118 MB. Exactly-once
confirmed at 2,000 records — 2,000 distinct, 0 duplicates.

New fixture test/pipelines/sqs_fanout_fanin_dag.yaml covers jq explode, DAG parallel
branches and join fan-in in one DAG, with a batch size deliberately indivisible by the
record count so records are always left buffered when the source stops — the case that
deadlocks. It hangs on the previous commit and produces exactly 29 files / 200 records /
100-100 branch split here.

Before deploying — two operational checks

Deferred acknowledgment holds a message in flight for longer, and every real config's
join is unbounded in practice, so nothing is acked until the source stops.

1. VisibilityTimeout must exceed each pipeline's end_after. Otherwise SQS
   redelivers mid-run and the output contains duplicates. Measured: at a 10s timeout with a
   20s window, output was 100% duplicated; at 300s, exact.
2. FIFO queues need per-message message_group_id, or a bounded join. An
   unacknowledged message blocks its message group, so with few groups throughput caps at
   max_messages per group per run — measured 10 of 40 versus 40 of 40 on main. Raising
   the visibility timeout does not help this; a duration: on the join restores
   40 of 40. Nine of the sixteen SQS configs in data-airflow use FIFO; the tightest is
   keepa/products/hourly (max_messages: 1, end_after: 50m). Caterpillar's own SQS
   writer defaults to a UUID per message, so caterpillar-fed queues are likely unaffected.

Known gaps

• No Go tests. The invariants this rests on — Fanout before send, exactly one
  Done/Fail per branch, Joined on the output — are unguarded, so a future regression
  is silent data loss rather than a test failure. All verification above was manual.
• archive(pack) livelocks if the visibility timeout is shorter than the full drain
  time: it acks nothing until the whole stream is consumed, so redelivery keeps the queue
  non-empty and exit_on_empty never fires. A ChangeMessageVisibility heartbeat would
  fix this and the duplication above, needs no new YAML key, and is the natural follow-up.
• http pagination emits one record per page without a Fanout — same data-loss shape as
  converter/xpath was. Not fixed here; the naive per-page AddBranch panics and it
  needs a branch reserved for the loop.
• heimdall's result.go sends with a background context rather than the record's, so its
  acks leak. Fixing it makes its existing terminal-mode guard a double-ack, so both must
  change together.
• sample.drain can leak the buffered records' acks if crypto/rand fails; fixing it
  properly needs the sampler interface to expose its buffer.

🤖 Generated with Claude Code

[prasadlohakpure]

Nice approach!
Maybe we can move a.Done() as a common utility to base task package. Also other func which we think can be abstracted, like

// === TERMINAL TASK HELPER ===
// CompleteAck marks the record as complete (for terminal tasks with output==nil)
func (b *Base) CompleteAck(ctx context.Context) {
    if a, ok := ack.FromContext(ctx); ok {
        a.Done()
    }
}

Then we can simply call these funcs.

[prasadlohakpure]

For larger record counts, it is going to create unbounded count of go routines as each is going to create a channel in blocking state, instead we can use a optimised approach on callback func, and the sink task then executes it.
Same can be done for kafka as well.

// In getMessages — no goroutines, no waiting, no channels:
for _, m := range receiveMessageOutput.Messages {
    msgAck := ack.New()

    // capture receipt handle; delete runs inline on whichever
    // goroutine calls the final Done/Fail — zero extra goroutines.
    receipt := m.ReceiptHandle
    msgId := m.MessageId
    msgAck.Done(func(failed bool) {
        if !failed {
            s.client.DeleteMessage(ctx, &qs.DeleteMessageInput{
                QueueUrl:      &s.QueueURL,
                ReceiptHandle: receipt,
            })
        }
    })

    s.SendData(ack.WithContext(ctx, msgAck), []byte(*m.Body), output)
}

And then, in the end on termination you can call

msgAck.Done(true)

With this approach, now deletion would not be enqueued.

[prasadlohakpure]

Something same can be done for kafka.
We can define an OnComplete function in interface, which can do cleanup