A keyed Postgres sink on pgx, and the leak loops that found the demo's growth - #290
Merged
Merged
Conversation
…s limits The demo on Render still grows after the #277 fix, about 1 MB an hour. A growth that shows per hour is caused per event, and a fast replay changes how many of each event happen per hour, so each loop drives one kind of event and reports growth per event of that kind: - internal/leakloop samples RssAnon, Go retained and duckdb_memory(), fits bytes per event after warm-up, and can malloc_trim before each sample. - The websocket source per message and per reconnect. - StructuredBatch with the demo's nested schema, handler SQL, progress update and window SQL, per message, with and without Postgres attached. - The sqlcommand sink per flush: a local table, and the demo's upsert into Postgres with a growing table, the same keys, and a truncated table, against a plain insert and a read. dev/bench records real Jetstream posts from a cursor, replays them over a websocket paced by their own timestamps, and runs a pipeline container under Render's 0.5 CPU and 512 MB. render-10x.yml is the demo as deployed, compressed ten times in wall clock, with noop-sink and no-postgres variants. SQLFLOW_LEAK_SCALE scales every loop. The defaults run in the -short pass.
A loop inherits the allocator state of every loop before it in the same test binary, so the third StructuredBatch loop started 78 MiB above the first. leakloops.sh builds the test binaries once in a Go and libduckdb container and runs each matching test alone.
…er, and MALLOC_ARENA_MAX=2 The loops report a rate and assert nothing, so they are benchmarks, not tests. They now build only with -tags leakloop, carry no coverage.Covers, and fail rather than skip when SQLFLOW_LEAK_POSTGRES is unset. A passing test the matrix reads as proof must be able to fail. The handler loop had emulated v1.2.0's manager SQL. It is now the handler alone, and internal/managers gets the demo's shape through the real StructuredBatch handler, the progress row, and Watermark.Poll on the manager's connection, with a counting sink and with the demo's sqlcommand upsert into Postgres. leakloop.Replay shifts each post's time_us forward on every pass over the capture, so the stream clock keeps moving and minutes keep closing. The image sets MALLOC_ARENA_MAX=2: two arenas cut the upsert's per-flush growth by two thirds in the loop.
…e that checks the key
…r the harness and the type runner The harness delivers id=1 a second time after its sequence and requires a sink that implements core.KeyedSink to hold ids 1, 2 and 3 once each. A sink that names no key is skipped, and each of the six is exempt with a structural proof. Three harness tests hold the step to its claim: an unkeyed double is skipped, a replacing double passes, a doubling double is caught. Against postgres:16 behind the fault proxy, every sink invariant passes, including honours_context, and every type invariant passes. The type table was measured, not assumed, and the run corrected it: a timestamp inside a list renders as RFC 3339, "2026-09-08T12:00:00.123Z" and "2026-09-08T21:00:00.123456+09:00", not with a space and no zone. The instant claim counts only text bound for a temporal column, so the marks moved from the zoned timestamp rows to two utf8 entries, into timestamp and timestamptz.
… abandoned COPY is the user's Six integration tests against postgres:16: two rows with one key in a batch keep the last; a retry with two buffered batches sharing a key applies them in order and the table holds the second value; an omitted column takes its default on insert and keeps its value on update; the probe refuses a missing table, a missing key column, a key with no exact unique index and a partial unique index, and warns on append onto a unique index; a refused value exits 10; a redial recreates the staging table. The refused-value test caught a misclassification. 1<<40 into a smallint column made pgx abandon the COPY it had started, and the server answered 57014 COPY from stdin failed: unable to encode 1099511627776 into binary format for int2 (OID 21) Class 57 read that as operator intervention: system.sink.unreachable, exit 12, retried for the whole ladder. A 57014 whose message starts "COPY from stdin failed" is now user.sink.encode_failed. Any other 57014, a statement timeout say, keeps the class rule.
…name the cost of ON CONFLICT through the extension
…d end to end and per flush bluesky.postgres.windowed.yml and kafka.postgres.sink.yml use type: postgres, and neither attaches Postgres through DuckDB. The same 200,000 recorded posts through the new example and the sqlcommand version it replaces, into postgres:16: postgres sink 2503 rows, 2026-09-13 12:37 to 13:45 UTC, 179332 posts sqlcommand 2503 rows, 2026-09-13 12:37 to 13:45 UTC, 179332 posts rows in one table and not the other: 0 and 0 Stopping Postgres under the running pipeline exits 12: flushing closed windows: [system.sink.unreachable] sink still failing after 4 attempts: [system.sink.unreachable] postgres sink: connect The per-flush loop, 3,600 flushes of 40 rows into a growing table, run together on Linux: sqlcommand upsert through the extension +42,132 B a flush sqlcommand plain insert +110 B postgres sink upsert -62 B postgres sink append +17 B The window loop through the real manager gains a postgres sink row. Its per-message rate is inside that loop's run-to-run noise, so the per-flush loop above is the comparison.
turbolytics
force-pushed
the
bench/component-leak-loops
branch
from
September 14, 2026 19:18
db43ac9 to
94db41b
Compare
turbolytics
changed the title
turbolytics
marked this pull request as ready for review
September 14, 2026 19:18
…from CI run 34886117062's reports
… only a Postgres target Close dropped the connection and kept every table a failed flush had buffered, so a shutdown with a batch stuck in the retry ladder leaked it. It now releases them. Flush holds a reference of its own across each send, so a Close during a send cannot free the batch being sent, and it removes the head only if Close has not already emptied the buffer. A checked allocator proves the release: 128 bytes held after Close before, 0 after. The sqlcommand warning fired on any ON CONFLICT while any Postgres was attached, including an upsert into a DuckDB table, which sends Postgres nothing. It now reads each ATTACH ... (TYPE POSTGRES) alias and warns only when the upsert's INTO names one. An ATTACH with no AS still warns on any upsert, because validate does not resolve the name it derives. INSERT OR REPLACE into the attachment is warned too: its query log shows the same full-table COPY (SELECT "bucket", "lang", ctid ...) TO STDOUT.
…anager block The four bench configs carried manager: tumbling_window, which #281 removed; main's validate refuses all of them. They reproduced the v1.2.0 image Render runs, and they are recorded at 0184802 for that purpose. They are replaced with the pipeline this branch teaches, compressed ten times in wall clock: demo-10x.yml window with the keyed postgres sink demo-10x-sqlcommand.yml the same window, written through the extension demo-10x-noop-sink.yml the same window, writing nothing Each ran four minutes on the branch image against the 10x replay. The postgres sink and the sqlcommand variant each wrote 1443 rows to 13:16 UTC with no null updated_at, and the noop run held flat. validate is clean on two and warns on the sqlcommand variant. The first sqlcommand run stopped on its first flush with 23502: the statement copied from the example omitted updated_at, and the extension sends an omitted column as NULL. The variant now names it, as Render's did. The postgres sink wrote the same rows into the same table with updated_at omitted, and Postgres applied the default.
…on, and a table named like its staging Review findings 1, 2, 3, 4 and 9 on #290, each reproduced first as a failing integration test through New, with the retry ladder and the probe: 1. A server that holds packets and keeps the socket open blocked a flush past 20 s with a 3 s retry deadline, and the startup probe the same. The ladder checks its deadline only between attempts. Each attempt, a probe or one batch's transaction, is now bounded by the retry deadline: both fail in about 3 s, coded unreachable, and the next flush redials. 2. A connection a pooler closes with a FIN between flushes failed with "begin: conn closed", coded write_failed, which is not retried, so the pipeline stopped with nothing written. A failure that closed the connection, or that pgx reports safe to retry, is now unreachable; the retry redials and delivers. A reset already worked. 3. DROP TABLE IF EXISTS sqlflow_staging on a session with no temp schema resolved the name on search_path and dropped a user's public.sqlflow_staging. Every reference is pg_temp-qualified. 4. A nullable key column passed the probe, and a unique index never matches a null, so a redelivered null key was inserted again: three deliveries, three rows, against idempotent_on_key. The probe refuses a nullable key. 9. A deferrable unique constraint (55000) and an invalid index left by a failed CREATE INDEX CONCURRENTLY (42P10) passed the probe and failed the first flush. The probe requires indimmediate and indisvalid.
…e does Review finding 5 on #290. checkSinks ran only in validate, so a config that skipped validate ran the count overwrite the refusal exists to prevent. config.Window.ReemitOverwrites holds the rule once; validate fails on it and buildManagedTables refuses it before it dials anything, user.config.invalid, exit 10. The retry deadline's doc says it now bounds a postgres attempt too.
Review finding 6 on #290. postgresRows converted the whole batch into [][]any before the COPY started. A postgresCopySource now hands pgx one reused row at a time, and pgx encodes it before asking for the next. Peak Go heap across one flush of 1M rows and 4 columns into Postgres 16, sampled every 5 ms: before +177 MiB, 7,001,967 allocations, 4.40 s after +29 MiB, 6,001,776 allocations, 3.93 s A type the sink cannot convert is refused by a schema check before the transaction opens, so it keeps its own code rather than surfacing mid-COPY as the server's 57014. A value that fails mid-stream returns the source's error, not the abandoned COPY's.
…he first flush Review finding 8 on #290. DuckDB's TIME, INTERVAL and ENUM arrive as time64, month_day_nano_interval and dictionary arrays, and each failed the first flush with type_unsupported, minutes after a deploy. time32 and time64 become pgtype.Time, month_day_nano_interval and duration become pgtype.Interval, and a dictionary writes its decoded value. Postgres keeps microseconds, so nanoseconds are truncated. The same types render inside a JSON container, and postgresConvertible holds the one type set both paths accept. Real DuckDB output through the sink into Postgres 16: TIME '12:34:56.789' time64[us] time 12:34:56.789 INTERVAL 90 MINUTE month_day_nano_interval interval 01:30:00 'ja'::ENUM('en','ja') dictionary<utf8, uint8> text ja The type table declares time64[us], time64[ns], month_day_nano_interval, dictionary and list<time64[us]>, and the runner passes every row.
…ooling, and the spec records the second review Review finding 7 and the untested pooler case on #290, as documentation: the config's doc comment and kafka.postgres.sink.yml say one refused value fails the batch with exit 10 and replays on restart, and the DSN's comment says transaction pooling cannot keep the session staging table. The spec gains a table of the nine findings and what changed for each.
serializeSQL returned col.Value(0), which aliases the record's buffer, and the deferred Release freed it before ReferenceTables parsed it. On Linux the test failed 3 runs in 10 with the AST overwritten by a later query's; CI failed TestReferenceTablesFindsNestedJoin with "invalid character '\u0090'".
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.
What this changes
The Bluesky demo on Render still grows after #277: from about 85 MB to about 105 MB over the 14 hours after the redeploy. The cause is the demo's write,
INSERT ... ON CONFLICTinto an attached Postgres through the DuckDB postgres extension, which copies the whole target table's keys into DuckDB on every flush. This branch carries the benchmarks that found it, and the fix: a keyedpostgressink on pgx that writes at the cost of the batch.The fix, per
docs/superpowers/specs/2026-09-14-postgres-sink-design.mdanddocs/superpowers/plans/2026-09-14-postgres-sink.md:type: postgreswithdsn,table,mode: upsert | append, andkeyfor upsert. One transaction per buffered batch, in arrival order: aCOPYinto a session temp table, thenINSERT ... SELECT DISTINCT ON (key) ... ORDER BY key, __seq DESC ON CONFLICT (key) DO UPDATE. Two rows with one key in a batch keep the last. The retry ladder wraps it.appendonto a table with a unique index.08,53,57and40aresystem.sink.unreachable(exit 12, retried);21,23,28,42,3Dand3Fareuser.sink.invalid(exit 10);22, and a COPY pgx abandoned for a value it cannot encode, areuser.sink.encode_failed(exit 10). That closes A window sink failure exits 1, not 12 #287 for this sink.sqlflow validaterefusesupsertwithlate_rows: reemit, warns onappendwithreemit, and warns on anysqlcommandsink carryingON CONFLICTwhile a command attaches a Postgres.sink.flush.idempotent_on_key: a second delivery of a batch leaves a keyed sink's destination holding it once. The postgres sink proves it; the six other sinks are exempt with a structural proof that they name no key.bluesky.postgres.windowed.ymlandkafka.postgres.sink.ymlmove to the sink.MALLOC_ARENA_MAX=2.The investigation:
A production worker shows growth per hour, but the growth is caused per event: per message, batch, flush, or reconnect. A fast replay changes how many of each event happen per hour, so a leak tied to flushes looks flat in a run of millions of messages. Each loop here drives one kind of event as fast as the component allows and reports growth per event of that kind.
internal/leakloopsamples RssAnon, Go retained (Sys - HeapReleased), andduckdb_memory(), and fits bytes per event after warm-up.SQLFLOW_LEAK_SCALEscales every loop.SQLFLOW_LEAK_MALLOC_TRIMcallsmalloc_trim(0)before each sample.Replaycycles a capture with each post'stime_usshifted forward per pass, so the stream clock keeps moving.The loops build only with
-tags leakloop, carry nocoverage.Covers, and fail rather than skip withoutSQLFLOW_LEAK_POSTGRES. They report a rate and assert nothing, so they are benchmarks, not tests, and they never run in the suite.internal/websocket/leakloop_test.go: per message on one connection, and per reconnect.internal/handlers/structured_leakloop_test.go: StructuredBatch with the demo's nested schema and handler SELECT, per message, nothing downstream.internal/managers/window_leakloop_test.go: the demo's shape through the real StructuredBatch handler, the progress row, andWatermark.Pollon the manager's connection, with a counting sink and with the demo's sqlcommand upsert into Postgres.internal/sinks/sqlcommand_leakloop_test.go: per flush into a local table, and into Postgres with the demo's upsert (growing table, same keys, truncated table), a plain insert, and a read.dev/bench:record.pyrecords real Jetstream posts from a cursor, about 5,000 a second.replayserves a capture over a websocket, paced by each event'stime_us.replay-soak.shruns a pipeline container at Render's 0.5 CPU and 512 MB and decomposes its memory once a minute.bluesky/render-ca740f8.ymlis the demo as deployed.render-10x.ymlis the same pipeline compressed ten times in wall clock, withnoop-sinkandno-postgresvariants.leakloops.shandleakloop.Dockerfilebuild the loops with the tag and run each in its own process on Linux.The image sets
MALLOC_ARENA_MAX=2. It costs no throughput. The Kafka benchmark container ran 2,000,000 messages at batch 5,000 in three alternating pairs:=2, msg/s=2, peak anonThroughput moves within run-to-run noise. Peak anon is 10 to 13 MiB lower with
=2.Findings
Every loop ran on Linux arm64 in Docker with DuckDB v1.5.2, postgres extension
c89234f, and Postgres 16, one process per loop. Loops that read posts used 2,000,000 real Jetstream posts. Native is RssAnon minus Go retained.1. The demo's upsert through the postgres extension grows every flush
3,600 flushes of 40 rows each:
MALLOC_ARENA_MAX=2malloc_trim(0)before each samplemalloc_trim(0)malloc_trim(0)INSERT, a new minute per flushduckdb_memory()stayed at 0.1 MiB and Go retained stayed flat in every row.The extension's query log shows why a growing table costs more. This is one demo-style upsert into a 1,000-row table, with
SET pg_debug_show_queries = trueon the v1.5.2 CLI:Every upsert copies the key columns and
ctidof the whole target table into DuckDB and detects conflicts there. The demo's table gains one row per language every minute and is never trimmed, so every flush reads more than the last.What the rows show:
ON CONFLICTis the cause. A plain insert of the same rows is flat.MALLOC_ARENA_MAX=2grew 66% less with a growing table. Each row is a separate run.malloc_trim(0). It is either still referenced, or held by the jemalloc allocator inside libduckdb, whichmalloc_trimdoes not reach. These loops do not separate the two.2. The v1.2.0 container reproduces the growth, and a noop window sink removes it
render-10x.ymlonturbolytics/sql-flow:v1.2.0at--cpus 0.5 --memory 512m. The replay covers the same 2,000,000 posts that live Jetstream sent over 11.3 hours, in 68 minutes.noop-sinkis the same pipeline with the window's sink set tonoop: the postgres extension stays loaded and attached, and nothing is written through it.The demo stepped up 18 MiB between minutes 8 and 16, then grew 4.9 MiB to the end. The noop-sink run never stepped, and grew 2.8 MiB over the same minutes. While messages flowed, Go retained stayed between 20.3 and 21.7 MiB and
duckdb_memory()between 0.25 and 0.81 MiB in both runs. The demo run wrote 23,403 rows to Postgres.The noop-sink run lost its connection once, at minute 11:
websocket read failed, reconnectingwithconnection reset by peer, thenwebsocket reconnecteda second later. The replay starts every connection at the first post, so that run consumed the first 305,192 posts twice and stopped 135,287 short of the end of the capture. It processed the posts behind the demo's step, 230,000 to 489,000, twice: at minutes 4 to 8 and again at 19 to 24. Native stayed between 45.8 and 48.3 MiB both times.2a. Footguns 3, 4, and 5 in #268 have the same cause
The extension does not send
ON CONFLICTto Postgres. DuckDB runs the upsert itself against a copy of the table's keys and writes the result back. Traced withSET pg_debug_show_queries = trueon the v1.5.2 CLI.A plain insert of three of a table's four columns:
An upsert of two rows with key 1:
The table then held one row,
(1, 10), and nothing reported an error.NOT NULL DEFAULTcolumn fails the upsertCOPY ... FROM STDINhas no column list, so it sends every column. The plain insert names its columns, and Postgres applies the default.pg_constraintwithcontype IN ('p', 'u'). A unique index without a constraint never reaches DuckDB's binder.ON CONFLICT, so its "cannot affect row a second time" check never runs. The log does not show how DuckDB dropped the second row.COPY (SELECT <key columns>, ctid ...) TO STDOUThas no filter on the incoming keys.DuckDB v1.5.5 with extension
41223e5sends the same full-tableCOPY ... TO STDOUT.3. The demo's shape through the real handler and manager
One process per loop, 2,000,000 real posts, Linux arm64, DuckDB v1.5.2. The handler alone, then the handler writing the window table with
Watermark.Pollafter every batch, with a counting sink and with the demo's sqlcommand upsert into Postgres on the sink's own connection:Watermark.Poll, counting sinkThe upsert adds about 11 B a message over the counting sink. At the demo's 500 posts a batch and one closed minute per flush that is about 680 flushes over 2,000,000 posts, or about 21 KB a flush, inside the 12 to 55 KB the sink loop measured on its own. The window loop through the real manager and the sink loop agree.
The handler's own rate varies between runs: the SELECT-only loop ended 2,000,000 messages at 15.0, 33.9, 37.3, and 32.6 MiB native across four runs, and at 10,000,000 messages its growth had fallen from 29 MiB in the first million to 1 MiB in the last. Compare loops within a run of this size, not a single number across runs.
duckdb_memory()read 0 throughout. The existingTestStructuredInvoke_DoesNotLeakNativeMemorybounds a flat three-column schema at 8 MiB over 500,000 messages, about 17 bytes a message, so it does not see this rate.3a. After: the postgres sink under the same loops
The per-flush loop, run together on Linux against the same Postgres 16: 3,600 flushes of one closed minute of 40 rows into a table that grows by that much every flush.
sqlcommandupsert through the extensionsqlcommandplain insert through the extensionpostgressink, upsertpostgressink, appendNative reads negative for the pgx sink because its buffers are Go heap: Go retained exceeds RssAnon by the heap the runtime holds and has not touched.
The window loop through the real manager, 2,000,000 posts, one process each, in one run: counting sink 0.6 to 90.1 MiB native, sqlcommand upsert 17.4 to 146.4 MiB, postgres sink 0.0 to 73.5 MiB. The ranking agrees with the flush loop. The per-message slopes after warm-up do not rank them: the counting sink's moved from 12.7 to 27.6 B between two runs, which is wider than a per-flush effect spread over 3,000 messages a flush.
End to end, the same 200,000 recorded posts through the new
bluesky.postgres.windowed.ymland through thesqlcommandversion it replaces, into Postgres 16:Stopping Postgres under the running pipeline:
4. Small or flat
duckdb_memory()rose to 8.9 MiB, fell to 4.1 at a checkpoint, and ended at 11.3, so about half of that growth is table storage DuckDB accounts for.What is not settled
malloc_trimis a leak or jemalloc retention.What the tests prove
The conformance run against
postgres:16behind the fault proxy passes all 11 sink invariants, includingsink.flush.idempotent_on_keyandsink.flush.honours_context, and all 6 type invariants. The type table is written by hand and was corrected by the run: a timestamp inside a JSON list renders as RFC 3339 (2026-09-08T12:00:00.123Z), not with a space and no zone.Six integration tests prove what this sink adds:
LastRowInABatchWins(1,10), (2,20), (1,30)leaves{1: 30, 2: 20}RetryAppliesBufferedBatchesInOrder21000OmittedColumnsTakeDefaultsAndKeepValuesNOT NULL DEFAULTcolumn the batch omits is filled on insert and unchanged on updateProbeChecksTheTargetARefusedValueIsAUserError1<<40intosmallintexits 10 and is not retried; aNOT NULLthe batch omits names the target and23502ARedialRecreatesTheStagingTableThe refused-value test caught a defect on the way: pgx abandons a
COPYit cannot encode, the server answers57014 COPY from stdin failed, and class 57 read that as unreachable, exit 12, retried. It is nowuser.sink.encode_failed.Three harness tests hold the new step to its claim: an unkeyed double is skipped, a replacing double passes, and a keyed double that appends a second row is caught.
Upstream: the full-table key copy is filed as duckdb/duckdb-postgres#575. A fix there helps
sqlcommandusers; it changes nothing in this sink.Second review: nine findings
An adversarial review ran the sink through
sinks.Newagainst Postgres 16 and broke it nine ways. Each finding reproduced as a failing test first; every one of those tests now passes.AStalledServerFailsTheFlushInsideTheDeadline,...TheProbeInsideTheDeadline8612733write_failed: begin: conn closedand stopped the pipelineAConnectionClosedBetweenFlushesIsRetried(FIN and RST)8612733DROP TABLE IF EXISTS sqlflow_stagingdropped a user'spublic.sqlflow_stagingAUserTableNamedLikeStagingSurvivespg_temp.sqlflow_staging8612733ANullableKeyIsRefusedAtStartup8612733rundid not refuse upsert withlate_rows: reemitTestWindowUpsertWithReemitIsRefusedAtStartupconfig.Window.ReemitOverwrites, read by validate and run; exit 10 before any dialea5656a[][]any: 1M rows x 4 columns held +177 MiBCopyFromSourcestreams one row at a time: +29 MiB, 4.40 s to 3.93 s1208c8ekafka.postgres.sink.yml; a sink-side DLQ is a separate change2b5ed42TIME,INTERVAL,ENUMfailed the first flushTimeIntervalAndEnumConvert, the type tablepgtype.Time,pgtype.Interval, decoded dictionary; real DuckDB output stores12:34:56.789,01:30:00,ja2ecd4c4DeferrableAndInvalidIndexesAreRefusedAtStartupindimmediateandindisvalid8612733Not changed, and said so in the code and spec: the sink needs a direct connection or session pooling, since transaction pooling does not keep the session staging table; a handler whose output columns change per batch recreates the staging table on every flush where they change; and the harness's
idempotent_on_keystep re-delivers an identical row, so replacement is proven byLastRowInABatchWinsandRetryAppliesBufferedBatchesInOrderrather than by the harness.The bench configs under
dev/bench/blueskycarried the removedmanager: tumbling_windowblock from reproducing v1.2.0, and main's validate refused them. They are replaced with the pipeline this branch teaches:demo-10x.yml(window with the postgres sink),demo-10x-sqlcommand.yml(the same window through the extension) anddemo-10x-noop-sink.yml. Each ran four minutes on the branch image. The two writers produced identical tables, 1,443 rows. The v1.2.0 runs in the findings above used the old configs, which are at0184802.Verification
go test -short -race ./...after the second review's fixes: 23 packages ok, 0 FAIL.go vet ./...clean with and without-tags leakloop, andgofmt -lprints nothing.uv run --locked pytest tests/tooling -q: 205 passed.make coverage-page && git status --short docs/coverageis clean after the registry changes were committed.go test -run '^TestIntegration' ./internal/sinksafter the second review's fixes: ok in 114 s. That is every ClickHouse, Kafka, Iceberg and Postgres integration test, including Conformance, Types, the six behaviour tests and the six review tests.bluesky.postgres.windowed.ymlend to end against thesqlcommandversion: identical tables, and exit 12 with Postgres stopped. Output above.uv run --locked pytest tests/release -qon the branch imageturbolytics/sql-flow:postgres-sink(labelv2026.09.14-19-g94db41b,MALLOC_ARENA_MAX=2in its environment): 19 passed.scripts/soak.sh 10on the same image: PASS, verdict below.go test -run '^TestIntegration' ./internal/sinks: ok in 102 s, so the ClickHouse, Kafka and Iceberg integrations beside the new Postgres ones still pass. CI's Integration job passed on the same commit.make coverage-checkagainst CI run 34886117062's own suite reports: passes after committing the regenerateddocs/coverage/status/. The first push failed Coverage because those files were stale;sink.postgresnow reads{unit: covered, integration: covered, release: not_required}, and every invariant it claims is covered at integration.Soak verdict
turbolytics/sql-flow:postgres-sink:The soak runs Kafka through InferredMemBatch into noop, so it gates the engine's per-message path and does not exercise the Postgres sink. The per-flush loop in 3a is the sink's memory evidence.
Notes for the reviewer
go test -short -race ./...and CI do not compile them.user.sink.type_unsupported, the code ClickHouse uses, rather thanencode_failed; a client-side encode failure isencode_failed;sinks.NewgainsWithLoggerand aWarnerinterface so the append warning reaches the log; a keyed sink is detected throughcore.KeyedSinkrather than a subject field.modeis the schema's finding (missing property 'mode'), socheckSinksdoes not repeat it.Reproduce
bluesky/postgres.sqlcreates the demo's table in the target database.Related: #268, #277, #247.