Every statement on the pipeline's DuckDB connection holds the shared … by turbolytics · Pull Request #283 · turbolytics/sql-flow

turbolytics · 2026-09-13T21:43:44Z

…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 (feat: Go engine rewrite (turbine) — v1 parity scope #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 Slow-moving data: liveness proven under trickle, surge, and silence #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.

What this changes

Verification

go test -short -race ./...
uv run --locked pytest tests/tooling -q
make coverage-page && git status --short docs/coverage is clean
uv run --locked pytest tests/release -q — required when the CLI
surface, a flag, the Dockerfile, or anything stamped at build time
changed
make soak — required when the change touches Arrow, ADBC, DuckDB,
cgo, the consume loop, a handler, a sink, the metrics recording path,
or anything that allocates per message or per request

Soak verdict

Notes for the reviewer

…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.

The bluesky demo's first deploy of v2026.09.14 exited with: error reinitializing handler: checkpoint after truncate: Cannot CHECKPOINT: there are other write transactions active StructuredBatch checkpoints after every batch to reclaim the rows its truncate left (#247). DuckDB refuses a checkpoint while another connection holds an uncommitted UPDATE or DDL; an uncommitted INSERT or DELETE does not refuse it. #281 moved the window to connections of its own. Its watermark save is an UPDATE, and the sqlcommand sink drops and creates its batch table. A batch that re-initialised while either was open failed, and a failed batch stops the process. #283's rule, that every statement on the pipeline's connection holds its lock, stopped reaching the window when it moved, and no invariant said so. Init now skips a refused checkpoint; the next batch's checkpoint reclaims what it left. Any other checkpoint error still fails the batch. Each skip logs at debug and counts in handler_checkpoints_skipped_total, so a write held open for longer than a batch shows as a rising count rather than as unexplained memory growth. The enforced invariant pipeline.batch.independent_of_window_io says a window's sink write or close, however long, never fails the consume loop. The manager harness proves it for manager.watermark. It holds the window's flush open and runs three batches through the structured handler, then holds a close after its delete and watermark update, before the commit, and runs three more. Against the release code the second hold fails batch 1 with the production error. With the fix both holds pass, and both closes land once released. TestStructuredInit_SkipsACheckpointAnotherWriterHolds holds an uncommitted UPDATE on a second connection for 50 batches. Init returns no error, and table storage grows to 4,422 KiB. One batch after the writer commits, storage is back to 0 KiB. TestCoreConsumeLoop_CountsSkippedCheckpoints checks the counter. If this is wrong, a windowed StructuredBatch pipeline still exits when a close coincides with a batch, or skips checkpoints forever and grows the way #247 did, with the counter rising.

…#289) * StructuredBatch skips a checkpoint another connection's write refuses The bluesky demo's first deploy of v2026.09.14 exited with: error reinitializing handler: checkpoint after truncate: Cannot CHECKPOINT: there are other write transactions active StructuredBatch checkpoints after every batch to reclaim the rows its truncate left (#247). DuckDB refuses a checkpoint while another connection holds an uncommitted UPDATE or DDL; an uncommitted INSERT or DELETE does not refuse it. #281 moved the window to connections of its own. Its watermark save is an UPDATE, and the sqlcommand sink drops and creates its batch table. A batch that re-initialised while either was open failed, and a failed batch stops the process. #283's rule, that every statement on the pipeline's connection holds its lock, stopped reaching the window when it moved, and no invariant said so. Init now skips a refused checkpoint; the next batch's checkpoint reclaims what it left. Any other checkpoint error still fails the batch. Each skip logs at debug and counts in handler_checkpoints_skipped_total, so a write held open for longer than a batch shows as a rising count rather than as unexplained memory growth. The enforced invariant pipeline.batch.independent_of_window_io says a window's sink write or close, however long, never fails the consume loop. The manager harness proves it for manager.watermark. It holds the window's flush open and runs three batches through the structured handler, then holds a close after its delete and watermark update, before the commit, and runs three more. Against the release code the second hold fails batch 1 with the production error. With the fix both holds pass, and both closes land once released. TestStructuredInit_SkipsACheckpointAnotherWriterHolds holds an uncommitted UPDATE on a second connection for 50 batches. Init returns no error, and table storage grows to 4,422 KiB. One batch after the writer commits, storage is back to 0 KiB. TestCoreConsumeLoop_CountsSkippedCheckpoints checks the counter. If this is wrong, a windowed StructuredBatch pipeline still exits when a close coincides with a batch, or skips checkpoints forever and grows the way #247 did, with the counter rising. * initHandler keeps its deferred unlock; one helper opens the test database Review of #289: initHandler released the pipeline lock by hand so it could read CheckpointSkipped after Init. A panic inside Init would have left the lock held, and every later statement on the pipeline's connection would block on it. The defer is back, and the flag is read under the lock, where it is the one this Init set. TestStructuredInit_SkipsACheckpointAnotherWriterHolds repeated newADBCConn's libduckdb fallback to reach a second connection. openTestDatabase holds it once, and both use it. go test -short on core, handlers, managers and cli/run: pass.

turbolytics mentioned this pull request Sep 14, 2026

Windows owned by the engine: a watermark manager and a late-row policy #281

Merged

turbolytics marked this pull request as ready for review September 14, 2026 00:51

turbolytics force-pushed the fix/progress-write-lock branch from 6d162e4 to 757672c Compare September 14, 2026 01:17

turbolytics merged commit aae403f into main Sep 14, 2026
5 checks passed

This was referenced Sep 14, 2026

changelog: name the v2026.09.14 section #286

Merged

StructuredBatch skips a checkpoint another connection's write refuses #289

Merged

A keyed Postgres sink on pgx, and the leak loops that found the demo's growth #290

Merged

turbolytics mentioned this pull request Sep 15, 2026

sqlcommand sink through the postgres extension stopped consuming after 462 messages at batch 1, no error #292

Open

Navigation Menu

Every statement on the pipeline's DuckDB connection holds the shared … - #283

Every statement on the pipeline's DuckDB connection holds the shared …#283
turbolytics merged 1 commit into
mainfrom
fix/progress-write-lock

turbolytics commented Sep 13, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

turbolytics commented Sep 13, 2026

What this changes

Verification

Soak verdict

Notes for the reviewer

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant