feat: implement acknowledgment tracking for pipeline records

[Mahesh Kamble (ma-gk)]

Description

Implements acknowledgment tracking for pipeline records so an SQS source only deletes a message's receipt once every downstream branch derived from it has finished processing — instead of deleting it as soon as it's received.

• New internal/pkg/pipeline/ack package: an Ack tracks how many pending "branches" a source record has, supports fan-out (AddBranch/Fanout), fan-in (Joined), drop (Drop), and per-branch success/failure (Done/Fail), and rides downstream attached to record.Record.Context.
• pipeline.go's distributeToChannels now calls ack.Fanout for structural DAG fan-out (same record duplicated across parallel branches).
• Tasks that fan records out (archive unpack, jq explode, sample, split, join, kafka) or drop them (filters, sampling) are wired to register/complete the right number of branches.
• sqs source task: replaced the old fixed worker-pool receipt-deletion model with per-message deleteOnComplete, which waits on the message's Ack and only deletes the receipt if every downstream branch succeeded; a failed branch leaves the receipt alone so SQS redelivers it after the visibility timeout.
• Added LocalStack-based test pipelines and a runner script (test/pipelines/sqs_ack_*_test.yaml, run_localstack_ack_scenarios.sh, setup_localstack_sqs.sh) covering baseline, fan-out (split/DAG/jq-explode/archive-unpack), fan-in (join/archive-pack), and drop (jq/sample) scenarios.

Types of changes

• Docs change / refactoring / dependency upgrade
• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)

Checklist

• My code follows the code style of this project.
• My change requires a change to the documentation and I have updated the documentation accordingly.
• I have added tests to cover my changes.

[prasadlohakpure]

loop on L45 and L61 is very similar, can we merge that to a single loop?

[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]

Are we deleting message as soon as it is received, and then again calling s.deleteOnComplete?

[prasadlohakpure]

Calling a go routine for every message can cause unbounded growth in go routines, can we fall back to previous pattern where we are having go routines based on concurrency?

[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