fix(source/kafka): honor drain contract after Close; make Nak redeliver in-session - #199
Merged
Merged
Conversation
…er in-session P0-1: Next/NextBatch re-checked closed&&inFlight==0 before each poll, so a closed subscription with records still in flight kept polling and yielded NEW records — violating the source.Subscription drain contract when driven outside the Hopper. Once closed they now never poll: already-buffered records are yielded, then the call blocks on a settle-wakeup channel until the last in-flight record settles, then returns ErrDrained (ctx cancellation honored). P0-2 (review option a): plain Nak previously only declined to mark, so the record was skipped for the live group until restart/rebalance while the core contract claimed universal at-least-once. Plain Nak now re-seeks through the same machinery as NakAfter with delay zero (pause partition, seek to the record's offset, resume), making redelivery deterministic within a live subscription. Cross-restart redelivery rides committed offsets and stays documented best-effort. Contract text aligned in one voice across source/doc.go, inlet.go, handler.go ActionNak, kafka package doc, and kafka README divergence table; requeueWithDelay cost model documented (head-of-line pause, buffered-ahead records, concurrent-mark override); poll-error record discard documented. Tests: close-then-Next/NextBatch regressions against the real subscription (fake poller), plain-Nak immediate re-seek, NakAfter delay, duplicate-settle safety, Hopper redelivery storm (bounded goroutines, >=501 deliveries).
joshua-temple
force-pushed
the
feat/source-v1-p0-contract
branch
from
August 23, 2026 02:50
9517260 to
ce69966
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Checkpoint 1/5 — P0-1 (drain) + P0-2 (Nak contract) on the road to source/v1.0.0
Implements the two blocking P0s from the v1-readiness review (
crucible-source-v1-review.md).P0-1 — Kafka
Next/NextBatchviolated the drain contract afterCloseThe pre-poll check
closed && inFlight == 0meant a closed subscription with in-flight records kept callingPollRecordsand handed out new records — a direct violation of thesource.Subscriptiondrain contract for any standalone driver. Now: afterClose, never poll; yield only already-buffered records; when the buffer is empty, block until the last in-flight settle lands (settle-wakeup channel, ctx-honoring), then returnErrDrained.P0-2 — core at-least-once claim vs Kafka plain-
Nakreality (review option (a))Plain
Nakpreviously only declined to mark → skipped for the live group until restart/rebalance whilesource/doc.goclaimed universal at-least-once. PlainNaknow goes through the same machinery asNakAfter(d)with d=0: pause partition → re-seek to the record's offset → resume. Redelivery is deterministic within a live subscription; cross-restart redelivery rides committed offsets and stays documented best-effort (the one divergence).Contract text now in one voice across:
source/doc.go(# Contract),source/inlet.go(Next),source/handler.go(ActionNak),source/kafka/kafka.go(ack model),source/kafka/README.md(table + divergence section). Also documents requeueWithDelay cost model + poll-error record discard.Evidence (all run locally on this branch)
go vet ./... && go build ./...(source/)-racecorego test -race -count=1 ./...go test -fuzz=FuzzOrderedSettle -fuzztime=8s+ batch variantgo test -tags integration -count=1 ./...(RedPanda/Docker 28.4.0)GOWORK=off go vet && go test -race -count=1 ./...(source/kafka)New tests (all pass under
-race -count=1):TestNextAfterCloseYieldsOnlyBufferedThenDrained,TestNextBatchAfterCloseYieldsOnlyBufferedThenDrained,TestNextAfterCloseDoesNotPollWithEmptyBufferAndNoInFlight,TestPlainNakReseeksPartitionImmediately,TestNakAfterDelaysThenReseeks,TestDuplicateSettleIsSafe,TestHopper_RedeliveryStormBoundedResources.Regressions vs baseline: none (full matrix green before and after).
API diff vs baseline: none (behavior/docs/tests only).