Managers: a refused poll stops the process, and the manager is a conformance subject - #270
Merged
Merged
Conversation
…ormance subject Fixes #267. The tumbling window manager logged a failed poll and polled again. A window the destination rejected was collected, written and refused every tick for as long as the process lived, with the container reporting healthy throughout. Observed against the live Bluesky firehose: three minutes, a failure every ten seconds, zero rows written, container Up. Start now returns the first failed poll, and run treats a manager's error as the pipeline's: it cancels the consume loop, which drains as it would on SIGTERM, and the process exits with the manager's error. The sink ran its retry ladder before the error reached the manager, so what arrives is final, and a retry in place repeats one the ladder already exhausted. The rows stay in the state table, because Poll deletes only after the sink accepted them, so a restart republishes the window. The manager is its own kind in the registry. It is not a sink and not a handler: it collects, flushes and deletes, and the delete is its commit. So its invariants are the consume loop's checkpoint claims restated for that commit, plus the liveness a poll loop owes on its own: - manager.delete.after_flush: the state table still holds every closed window when Flush runs, and none once the poll returns. - manager.delete.nothing_on_failure: a failed flush deletes nothing. - manager.publish.eventually: the loop publishes a closed window on its own. - manager.failure.exits: a refused poll stops the loop after one attempt. internal/conformance drives a ManagerSubject the way it drives a PipelineSubject: the harness owns a recording sink, fails its flush, and judges the event order and the state table. The tumbling manager supplies build, seed and count. All four are enforced. The status file for the new integration follows from CI's reports.
turbolytics
force-pushed
the
fix/manager-rejected-flush
branch
from
September 12, 2026 19:13
f7da14d to
cd84bc0
Compare
turbolytics
added a commit
that referenced
this pull request
Sep 13, 2026
…lock A window manager failed polls on a healthy pipeline with "Attempting to execute an unsuccessful or closed pending query result". Once on the Bluesky demo at batch 500, and seven times in ten minutes at batch 1 on bluesky.kafka.windowed.yml with the v1.2.0 image. Since #270 a failed poll stops the process, so on main this is a crash at a rate set by batch size. DuckDB closes a pending result the moment another statement runs on its connection. The pipeline, the managers, the progress store and the debug API share one connection and one mutex, and three parties ran statements without the mutex: - The handler reset. Init drops or truncates the batch table after every batch and once at loop start. Unlocked since the Go rewrite (#123), so the race shipped in v1.0.0. At batch 1 it runs once per message. - The progress write. recordProgress released the lock, then ran its UPDATE. Added in #259. - The sqlcommand sink. Its flush runs a drop, an ingest and the user's SQL, as the pipeline sink, a manager sink or the DLQ sink. Each now takes the lock. The sink gets it through sinks.WithConnLock, and sinks.New refuses to build a sqlcommand sink without it, so a new call site that forgets fails at startup instead of racing. No caller flushes a sink while holding the lock, so the sink taking it cannot deadlock. A race has no deterministic reproduction, so the tests check the invariant instead. duckdb.LockChecked wraps a connection and records any statement, commit or rollback that finds the lock free, with the frames that led there. Two tests run single-threaded under it, where a free lock can only mean the caller let go: - core: the consume loop with a handler that touches the connection and a real progress store. Before the fix it named the handler reset at both call sites and the progress write. - run: the real InferredMemBatch handler, sqlcommand sinks in all three roles, the progress store and a tumbling manager, built by the functions the run command calls. Before the fix it named 15 sink statements. Evidence: - Both tests fail on the old code and pass on the new. - Fixing the progress write alone still failed a poll after 2,900 messages on the batch-1 config. That run is how the handler reset was found. - With all three fixed, the same config ran 15,000 messages with no poll failure. bluesky.postgres.windowed.yml ran 12,000 messages with no poll failure and published closed minutes to Postgres through the locked manager sink. - go test -short -race ./... passes. What breaks if this is wrong: a statement still runs unlocked and a manager poll fails, which on main stops the pipeline. A sink that took the lock while its caller held it would deadlock the pipeline instead. Closes #280.
5 tasks
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
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.
Fixes #267.
The defect
Tumbling.Startlogged a failed poll and polled again. A window the destination rejected was collected, written and refused every tick for as long as the process lived, and the container reported healthy throughout. Observed on 2026-09-12 against the live Bluesky firehose: three minutes, a failure every ten seconds, zero rows written, containerUp.The fix
Startreturns the first failed poll. The rows stay in the state table, becausePolldeletes only after the sink accepted them, so a restart republishes the window. There is no retry in place: the sink ran its retry ladder before the error reached the manager, so what arrives is final.runtreats a manager's error as the pipeline's. It cancels the consume loop through aWithCancelCausecontext, the loop drains as it would on SIGTERM, and the process exits with the manager's error rather than the loop's cancellation.The manager as a conformance subject
The manager is its own kind in the registry,
manager, besidepipeline. It is not a sink and not a handler: it collects, flushes and deletes, and the delete is its commit. Its invariants are the consume loop's checkpoint claims restated for that commit, plus the liveness a poll loop owes on its own:manager.delete.after_flushmanager.delete.nothing_on_failuremanager.publish.eventuallymanager.failure.exitsinternal/conformance/manager.godrives aManagerSubjectthe waypipeline.godrives aPipelineSubject: the harness owns a recording sink, fails its flush on demand, and judges the event order and the state table. The tumbling manager's subject supplies build, seed and count in 40 lines. All four invariants areenforced: true.The one unit test that pinned the old behavior,
StartSurvivesAFailedPoll, is replaced byStartReturnsTheFirstFailedPoll.Registry and generator
managerjoinsKINDSandINTEGRATION_KINDSin the generator, the registry tests expect five kinds, anddocs/coverage/integrations/manager.tumbling_window.ymlisconstructed: falselike the pipeline configurations. The page renders the manager column under checkpoint and lifecycle.The status file for the new integration follows from CI's reports in a second commit, the way every new integration has landed.
Verification
go test -short -race ./...TestManagerTumblingWindow_Conformancego vet,gofmtpytest tests/toolingNot in this PR
errs.Retryablereports as retryable. A constraint violation is a user-fixable error and should exit 10. That is a classification change in the sink, andsink.error.classifieshas no cell for any sink yet.