From 10685bffa38d9c412d43856d588e6d8ebd04e4b0 Mon Sep 17 00:00:00 2001 From: Joshua Temple Date: Sun, 23 Aug 2026 01:51:02 -0400 Subject: [PATCH] docs(source): pin the last gate-pass doc gaps (settle-after-Begin, NextBatch limit, Manual accounting) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1: an extra direct Settle for m around Transactional.Begin is now defined on the Begin surface itself: a caller error with undefined coordination — at best a harmless duplicate mark (at-least-once), at worst a race with End's commit/abort. F2: Batched.NextBatch documents that limit < 1 is treated as one (matching the kafka implementation's clamp). F4: Manual() documents that the in-flight slot is released like any settle — drain accounting treats Manual as settled. --- source/capability.go | 3 ++- source/handler.go | 2 ++ source/kafka/capability.go | 7 +++++-- 3 files changed, 9 insertions(+), 3 deletions(-) diff --git a/source/capability.go b/source/capability.go index 905bfbb..2558a9c 100644 --- a/source/capability.go +++ b/source/capability.go @@ -120,7 +120,8 @@ type OrderedDelivery interface { // substitute for per-message settlement. type Batched interface { // NextBatch returns up to limit messages, blocking for at least one, or - // ctx.Err()/ErrDrained as [Subscription.Next] would. + // ctx.Err()/ErrDrained as [Subscription.Next] would. A limit below one is + // treated as one. NextBatch(ctx context.Context, limit int) ([]Message, error) // SettleBatch applies the single result r to every message in ms in one call. // It is for a caller settling a uniform batch directly; the Hopper settles per diff --git a/source/handler.go b/source/handler.go index 0235e75..c88c7f2 100644 --- a/source/handler.go +++ b/source/handler.go @@ -172,6 +172,8 @@ func Skip() Result { return Result{Action: ActionAck, Class: Drop} } func InProgress() Result { return Result{Action: ActionInProgress} } // Manual returns a Result reporting the handler settled the message itself. +// The message's in-flight slot is released like any other settle — drain +// accounting treats Manual as settled — but the backend takes no action. func Manual() Result { return Result{Action: ActionManual} } // Failed reports whether the result carries a failure (anything that is not a diff --git a/source/kafka/capability.go b/source/kafka/capability.go index b9901c1..b359696 100644 --- a/source/kafka/capability.go +++ b/source/kafka/capability.go @@ -424,8 +424,11 @@ func toRecordHeaders(hs source.Headers) []kgo.RecordHeader { // m's offset is marked only after fn succeeds and only inside the transaction, so // a failed transform never advances the consumed position: the commit-after- // process invariant holds, now atomically with the produce side. Begin is the -// full settle path for m; the caller must not also call [Subscription.Settle] for -// m. +// full settle path for m; the caller must not also call [Subscription.Settle] +// for m. A direct Settle for m during an open Begin, or after Begin returned, +// is a caller error with undefined results — it is not detected or coordinated +// with the transaction: at best it duplicates an already-committed mark +// (harmless at-least-once), at worst it races End's commit/abort. // // It is available only when the inlet was built with [WithTransactional]; // otherwise it reports the capability is absent rather than silently running fn