Skip to content

Managers: a refused poll stops the process, and the manager is a conformance subject - #270

Merged
turbolytics merged 3 commits into
mainfrom
fix/manager-rejected-flush
Sep 12, 2026
Merged

Managers: a refused poll stops the process, and the manager is a conformance subject#270
turbolytics merged 3 commits into
mainfrom
fix/manager-rejected-flush

Conversation

@turbolytics

Copy link
Copy Markdown
Owner

Fixes #267.

The defect

Tumbling.Start 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, 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, container Up.

The fix

  • Start returns the first failed poll. The rows stay in the state table, because Poll deletes 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.
  • run treats a manager's error as the pipeline's. It cancels the consume loop through a WithCancelCause context, 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, beside pipeline. 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:

Invariant Class Claim
manager.delete.after_flush safety The state table still holds every closed window when Flush runs, and none once the poll returns. Observed from inside the flush, not inferred from the count afterwards.
manager.delete.nothing_on_failure safety A failed flush deletes nothing.
manager.publish.eventually liveness The loop publishes a closed window on its own.
manager.failure.exits liveness A refused poll stops the loop after one attempt, with the rows still in the table.

internal/conformance/manager.go drives a ManagerSubject the way pipeline.go drives a PipelineSubject: 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 are enforced: true.

The one unit test that pinned the old behavior, StartSurvivesAFailedPoll, is replaced by StartReturnsTheFirstFailedPoll.

Registry and generator

manager joins KINDS and INTEGRATION_KINDS in the generator, the registry tests expect five kinds, and docs/coverage/integrations/manager.tumbling_window.yml is constructed: false like 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

Check Result
go test -short -race ./... clean
TestManagerTumblingWindow_Conformance four cells proven
go vet, gofmt clean
pytest tests/tooling 204 pass, page regenerated

Not in this PR

  • The sqlcommand sink's errors are uncoded, so a refused flush exits 1, which errs.Retryable reports as retryable. A constraint violation is a user-fixable error and should exit 10. That is a classification change in the sink, and sink.error.classifies has no cell for any sink yet.
  • A failed final poll at shutdown still only logs. Its rows are in memory and gone either way.
  • The tumbling window docs on the site should say that a refused publish now stops the process. Separate PR.

…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
turbolytics force-pushed the fix/manager-rejected-flush branch from f7da14d to cd84bc0 Compare September 12, 2026 19:13
@turbolytics
turbolytics merged commit be9412c into main Sep 12, 2026
5 checks passed
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.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Tumbling manager retries a rejected flush forever and the process stays up

1 participant