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