From 0184802e4dfdeef5fa03bee2fc8afd9230331ca7 Mon Sep 17 00:00:00 2001 From: "turbolytics.io" Date: Mon, 14 Sep 2026 10:27:35 -0400 Subject: [PATCH 01/27] Leak loops per component, and a replay of the Bluesky demo at Render'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. --- dev/bench/bluesky/postgres.sql | 15 + dev/bench/bluesky/render-10x-no-postgres.yml | 120 +++++++ dev/bench/bluesky/render-10x-noop-sink.yml | 133 ++++++++ dev/bench/bluesky/render-10x.yml | 147 +++++++++ dev/bench/bluesky/render-ca740f8.yml | 133 ++++++++ dev/bench/record.py | 37 +++ dev/bench/replay-soak.sh | 81 +++++ dev/bench/replay/main.go | 145 +++++++++ .../handlers/structured_demo_leak_test.go | 200 ++++++++++++ internal/leakloop/leakloop.go | 295 ++++++++++++++++++ internal/leakloop/trim_linux.go | 10 + internal/leakloop/trim_other.go | 6 + internal/sinks/sqlcommand_leak_test.go | 188 +++++++++++ internal/websocket/leak_test.go | 115 +++++++ 14 files changed, 1625 insertions(+) create mode 100644 dev/bench/bluesky/postgres.sql create mode 100644 dev/bench/bluesky/render-10x-no-postgres.yml create mode 100644 dev/bench/bluesky/render-10x-noop-sink.yml create mode 100644 dev/bench/bluesky/render-10x.yml create mode 100644 dev/bench/bluesky/render-ca740f8.yml create mode 100644 dev/bench/record.py create mode 100644 dev/bench/replay-soak.sh create mode 100644 dev/bench/replay/main.go create mode 100644 internal/handlers/structured_demo_leak_test.go create mode 100644 internal/leakloop/leakloop.go create mode 100644 internal/leakloop/trim_linux.go create mode 100644 internal/leakloop/trim_other.go create mode 100644 internal/sinks/sqlcommand_leak_test.go create mode 100644 internal/websocket/leak_test.go diff --git a/dev/bench/bluesky/postgres.sql b/dev/bench/bluesky/postgres.sql new file mode 100644 index 00000000..a6af2829 --- /dev/null +++ b/dev/bench/bluesky/postgres.sql @@ -0,0 +1,15 @@ +-- One row per minute per language. The primary key leads with bucket, so a +-- time-range scan uses it and no second index is needed. +-- +-- The pipeline's upsert depends on this being a primary key. The DuckDB +-- postgres extension honours ON CONFLICT against a primary key and fails with a +-- binder error against a unique index alone. +CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ NOT NULL, + lang TEXT NOT NULL, + posts INTEGER NOT NULL, + -- Wall-clock time of the last write. bucket is event time. A reader needs + -- both to tell "the stream is behind" from "the stream stopped". + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (bucket, lang) +); diff --git a/dev/bench/bluesky/render-10x-no-postgres.yml b/dev/bench/bluesky/render-10x-no-postgres.yml new file mode 100644 index 00000000..69f787a7 --- /dev/null +++ b/dev/bench/bluesky/render-10x-no-postgres.yml @@ -0,0 +1,120 @@ +# Variant C of render-10x.yml: no postgres extension. The load and attach +# commands are gone and the window's sink is noop. Everything else is +# render-10x.yml, which says how the replay is paced. + +# The Bluesky demo exactly as Render runs it (sql-flow-bluesky-demo ca740f8, +# render-ca740f8.yml beside this file), compressed ten times in wall clock. +# +# Replay the capture with `replay -speed 10`. The stream clock then advances +# ten seconds per second, so every timer that paces work against it has to +# shrink by the same factor for the ratios between messages, batches, polls +# and flushes to stay what they are on Render: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# The one-minute idle bound in the window predicates is left alone. It only +# fires when the stream stops, and a replay does not stop. +# Counts Bluesky posts per minute by language in a 1-minute tumbling window and +# upserts each closed window into Postgres. +commands: + # Logs and any timestamp rendering are UTC whatever the host's timezone. + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No unique index, on purpose. In-memory DuckDB never frees rows deleted + # from an indexed table, and the manager deletes every closed minute, so + # the index grew the worker from 100 MB to 450 MB in a day on a 512 MB + # plan. Without it, each batch appends its own counts and the collect + # query sums them per minute. Measured flat over eight hours. + # See turbolytics/sql-flow#268. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + manager: + # A window closes against the stream's own clock, not wall clock. + # max(bucket) is the newest window the data has reached, so a window + # closes only once the stream has moved past it. That keeps the result + # identical whether the data arrives live or as a replay. With now() + # here instead, a pipeline running behind real time by more than the + # grace period publishes a window whose rows are still arriving, once + # per poll, and those parts are indistinguishable downstream from an + # at-least-once duplicate. + # + # The second branch is the idleness bound. A stream that goes quiet + # never moves its own clock, so the first branch alone would hold the + # newest window open until data a full grace newer arrived. + # sqlflow_progress.last_arrival is when the newest batch was written, + # so after a grace of silence every open window closes. + # + # The grace outlives the batch wait on purpose: an event from the last + # seconds of a window can reach DuckDB up to a flush interval after the + # window ended. + tumbling_window: + poll_interval_seconds: 1 + # One row per minute and language. The table holds one row per batch, + # and the Postgres upsert below must not see the same key twice in + # one statement. + collect_closed_windows_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + GROUP BY bucket, lang + delete_closed_windows_sql: | + DELETE FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + + sink: + type: noop + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + -- to_timestamp yields TIMESTAMPTZ, so the bucket is an instant and the + -- write into a Postgres TIMESTAMPTZ column needs no interpretation. The + -- example's date_trunc(make_timestamp(time_us)) yields a naive + -- TIMESTAMP, and Postgres reads a naive value in the session timezone. + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + # The window manager does the writing. A pipeline sink would write every + # batch, which is the opposite of aggregating. + sink: + type: noop diff --git a/dev/bench/bluesky/render-10x-noop-sink.yml b/dev/bench/bluesky/render-10x-noop-sink.yml new file mode 100644 index 00000000..a9338d66 --- /dev/null +++ b/dev/bench/bluesky/render-10x-noop-sink.yml @@ -0,0 +1,133 @@ +# Variant A of render-10x.yml: the window's sink is noop. The postgres +# extension is still loaded and attached, and nothing is written through it. +# Everything else is render-10x.yml, which says how the replay is paced. + +# The Bluesky demo exactly as Render runs it (sql-flow-bluesky-demo ca740f8, +# render-ca740f8.yml beside this file), compressed ten times in wall clock. +# +# Replay the capture with `replay -speed 10`. The stream clock then advances +# ten seconds per second, so every timer that paces work against it has to +# shrink by the same factor for the ratios between messages, batches, polls +# and flushes to stay what they are on Render: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# The one-minute idle bound in the window predicates is left alone. It only +# fires when the stream stops, and a replay does not stop. +# Counts Bluesky posts per minute by language in a 1-minute tumbling window and +# upserts each closed window into Postgres. +commands: + # Logs and any timestamp rendering are UTC whatever the host's timezone. + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + - name: load postgres extension + sql: | + INSTALL postgres; + LOAD postgres; + + # No default: a worker without a database must fail at startup, not attach + # to some other host. + - name: attach postgres + sql: | + ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES); + + # Only the fields the handler reads. StructuredBatch loads each batch into + # this table, so an unused field is parsing work for nothing. + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No unique index, on purpose. In-memory DuckDB never frees rows deleted + # from an indexed table, and the manager deletes every closed minute, so + # the index grew the worker from 100 MB to 450 MB in a day on a 512 MB + # plan. Without it, each batch appends its own counts and the collect + # query sums them per minute. Measured flat over eight hours. + # See turbolytics/sql-flow#268. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + manager: + # A window closes against the stream's own clock, not wall clock. + # max(bucket) is the newest window the data has reached, so a window + # closes only once the stream has moved past it. That keeps the result + # identical whether the data arrives live or as a replay. With now() + # here instead, a pipeline running behind real time by more than the + # grace period publishes a window whose rows are still arriving, once + # per poll, and those parts are indistinguishable downstream from an + # at-least-once duplicate. + # + # The second branch is the idleness bound. A stream that goes quiet + # never moves its own clock, so the first branch alone would hold the + # newest window open until data a full grace newer arrived. + # sqlflow_progress.last_arrival is when the newest batch was written, + # so after a grace of silence every open window closes. + # + # The grace outlives the batch wait on purpose: an event from the last + # seconds of a window can reach DuckDB up to a flush interval after the + # window ended. + tumbling_window: + poll_interval_seconds: 1 + # One row per minute and language. The table holds one row per batch, + # and the Postgres upsert below must not see the same key twice in + # one statement. + collect_closed_windows_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + GROUP BY bucket, lang + delete_closed_windows_sql: | + DELETE FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + + sink: + type: noop + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + -- to_timestamp yields TIMESTAMPTZ, so the bucket is an instant and the + -- write into a Postgres TIMESTAMPTZ column needs no interpretation. The + -- example's date_trunc(make_timestamp(time_us)) yields a naive + -- TIMESTAMP, and Postgres reads a naive value in the session timezone. + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + # The window manager does the writing. A pipeline sink would write every + # batch, which is the opposite of aggregating. + sink: + type: noop diff --git a/dev/bench/bluesky/render-10x.yml b/dev/bench/bluesky/render-10x.yml new file mode 100644 index 00000000..49162c8d --- /dev/null +++ b/dev/bench/bluesky/render-10x.yml @@ -0,0 +1,147 @@ +# The Bluesky demo exactly as Render runs it (sql-flow-bluesky-demo ca740f8, +# render-ca740f8.yml beside this file), compressed ten times in wall clock. +# +# Replay the capture with `replay -speed 10`. The stream clock then advances +# ten seconds per second, so every timer that paces work against it has to +# shrink by the same factor for the ratios between messages, batches, polls +# and flushes to stay what they are on Render: +# +# poll_interval_seconds 10 -> 1 +# flush_interval_seconds 30 (the default) -> 3 +# +# The one-minute idle bound in the window predicates is left alone. It only +# fires when the stream stops, and a replay does not stop. +# Counts Bluesky posts per minute by language in a 1-minute tumbling window and +# upserts each closed window into Postgres. +commands: + # Logs and any timestamp rendering are UTC whatever the host's timezone. + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + - name: load postgres extension + sql: | + INSTALL postgres; + LOAD postgres; + + # No default: a worker without a database must fail at startup, not attach + # to some other host. + - name: attach postgres + sql: | + ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES); + + # Only the fields the handler reads. StructuredBatch loads each batch into + # this table, so an unused field is parsing work for nothing. + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No unique index, on purpose. In-memory DuckDB never frees rows deleted + # from an indexed table, and the manager deletes every closed minute, so + # the index grew the worker from 100 MB to 450 MB in a day on a 512 MB + # plan. Without it, each batch appends its own counts and the collect + # query sums them per minute. Measured flat over eight hours. + # See turbolytics/sql-flow#268. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + manager: + # A window closes against the stream's own clock, not wall clock. + # max(bucket) is the newest window the data has reached, so a window + # closes only once the stream has moved past it. That keeps the result + # identical whether the data arrives live or as a replay. With now() + # here instead, a pipeline running behind real time by more than the + # grace period publishes a window whose rows are still arriving, once + # per poll, and those parts are indistinguishable downstream from an + # at-least-once duplicate. + # + # The second branch is the idleness bound. A stream that goes quiet + # never moves its own clock, so the first branch alone would hold the + # newest window open until data a full grace newer arrived. + # sqlflow_progress.last_arrival is when the newest batch was written, + # so after a grace of silence every open window closes. + # + # The grace outlives the batch wait on purpose: an event from the last + # seconds of a window can reach DuckDB up to a flush interval after the + # window ended. + tumbling_window: + poll_interval_seconds: 1 + # One row per minute and language. The table holds one row per batch, + # and the Postgres upsert below must not see the same key twice in + # one statement. + collect_closed_windows_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + GROUP BY bucket, lang + delete_closed_windows_sql: | + DELETE FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + + sink: + type: sqlcommand + sqlcommand: + # ON CONFLICT is what makes at-least-once survivable. The manager + # deletes a window only after the sink accepts it, so a crash + # between the two republishes that window on the next poll. Against + # the target's primary key a bare INSERT fails that republish, and + # because the failed flush also blocks the delete, every later poll + # collects the same rows and fails the same way. + # + # updated_at is listed and selected as now() on purpose. The + # postgres extension sends rows that do not conflict through a + # COPY, and that COPY passes an explicit NULL for any column this + # list omits. An explicit NULL defeats the column's DEFAULT, so + # omitting updated_at fails the NOT NULL on every new minute. + sql: | + INSERT INTO pg.posts_per_minute_by_lang (bucket, lang, posts, updated_at) + SELECT bucket, lang, posts, now() FROM sqlflow_sink_batch + ON CONFLICT (bucket, lang) DO UPDATE + SET posts = EXCLUDED.posts, updated_at = EXCLUDED.updated_at + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + flush_interval_seconds: 3 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + -- to_timestamp yields TIMESTAMPTZ, so the bucket is an instant and the + -- write into a Postgres TIMESTAMPTZ column needs no interpretation. The + -- example's date_trunc(make_timestamp(time_us)) yields a naive + -- TIMESTAMP, and Postgres reads a naive value in the session timezone. + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + # The window manager does the writing. A pipeline sink would write every + # batch, which is the opposite of aggregating. + sink: + type: noop diff --git a/dev/bench/bluesky/render-ca740f8.yml b/dev/bench/bluesky/render-ca740f8.yml new file mode 100644 index 00000000..78911b3c --- /dev/null +++ b/dev/bench/bluesky/render-ca740f8.yml @@ -0,0 +1,133 @@ +# Counts Bluesky posts per minute by language in a 1-minute tumbling window and +# upserts each closed window into Postgres. +commands: + # Logs and any timestamp rendering are UTC whatever the host's timezone. + - name: pin the session timezone + sql: | + SET TimeZone='UTC'; + + - name: load postgres extension + sql: | + INSTALL postgres; + LOAD postgres; + + # No default: a worker without a database must fail at startup, not attach + # to some other host. + - name: attach postgres + sql: | + ATTACH '{{ SQLFLOW_POSTGRES_URI }}' AS pg (TYPE POSTGRES); + + # Only the fields the handler reads. StructuredBatch loads each batch into + # this table, so an unused field is parsing work for nothing. + - name: declare the post schema + sql: | + CREATE TABLE IF NOT EXISTS posts ( + time_us BIGINT, + commit STRUCT( + operation TEXT, + record STRUCT(langs TEXT[]) + ) + ); + +tables: + sql: + - name: posts_per_minute_by_lang + # No unique index, on purpose. In-memory DuckDB never frees rows deleted + # from an indexed table, and the manager deletes every closed minute, so + # the index grew the worker from 100 MB to 450 MB in a day on a 512 MB + # plan. Without it, each batch appends its own counts and the collect + # query sums them per minute. Measured flat over eight hours. + # See turbolytics/sql-flow#268. + sql: | + CREATE TABLE IF NOT EXISTS posts_per_minute_by_lang ( + bucket TIMESTAMPTZ, + lang TEXT, + posts INTEGER + ); + + manager: + # A window closes against the stream's own clock, not wall clock. + # max(bucket) is the newest window the data has reached, so a window + # closes only once the stream has moved past it. That keeps the result + # identical whether the data arrives live or as a replay. With now() + # here instead, a pipeline running behind real time by more than the + # grace period publishes a window whose rows are still arriving, once + # per poll, and those parts are indistinguishable downstream from an + # at-least-once duplicate. + # + # The second branch is the idleness bound. A stream that goes quiet + # never moves its own clock, so the first branch alone would hold the + # newest window open until data a full grace newer arrived. + # sqlflow_progress.last_arrival is when the newest batch was written, + # so after a grace of silence every open window closes. + # + # The grace outlives the batch wait on purpose: an event from the last + # seconds of a window can reach DuckDB up to a flush interval after the + # window ended. + tumbling_window: + poll_interval_seconds: 10 + # One row per minute and language. The table holds one row per batch, + # and the Postgres upsert below must not see the same key twice in + # one statement. + collect_closed_windows_sql: | + SELECT bucket, lang, sum(posts)::INTEGER AS posts + FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + GROUP BY bucket, lang + delete_closed_windows_sql: | + DELETE FROM posts_per_minute_by_lang + WHERE bucket + INTERVAL '1 minute' < (SELECT max(bucket) FROM posts_per_minute_by_lang) - INTERVAL '60 seconds' + OR (SELECT now() - last_arrival FROM sqlflow_progress) > INTERVAL '1 minute' + + sink: + type: sqlcommand + sqlcommand: + # ON CONFLICT is what makes at-least-once survivable. The manager + # deletes a window only after the sink accepts it, so a crash + # between the two republishes that window on the next poll. Against + # the target's primary key a bare INSERT fails that republish, and + # because the failed flush also blocks the delete, every later poll + # collects the same rows and fails the same way. + # + # updated_at is listed and selected as now() on purpose. The + # postgres extension sends rows that do not conflict through a + # COPY, and that COPY passes an explicit NULL for any column this + # list omits. An explicit NULL defeats the column's DEFAULT, so + # omitting updated_at fails the NOT NULL on every new minute. + sql: | + INSERT INTO pg.posts_per_minute_by_lang (bucket, lang, posts, updated_at) + SELECT bucket, lang, posts, now() FROM sqlflow_sink_batch + ON CONFLICT (bucket, lang) DO UPDATE + SET posts = EXCLUDED.posts, updated_at = EXCLUDED.updated_at + +pipeline: + name: bluesky-posts-per-minute-by-lang + batch_size: 500 + + source: + type: websocket + websocket: + uri: "{{ SQLFLOW_JETSTREAM_URI|default('wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post') }}" + + handler: + type: handlers.StructuredBatch + table: posts + sql: | + INSERT INTO posts_per_minute_by_lang + SELECT + -- to_timestamp yields TIMESTAMPTZ, so the bucket is an instant and the + -- write into a Postgres TIMESTAMPTZ column needs no interpretation. The + -- example's date_trunc(make_timestamp(time_us)) yields a naive + -- TIMESTAMP, and Postgres reads a naive value in the session timezone. + time_bucket(INTERVAL '1 minute', to_timestamp(time_us / 1000000)) AS bucket, + coalesce(commit.record.langs[1], 'unknown') AS lang, + count(*) AS posts + FROM posts + WHERE commit.operation = 'create' + GROUP BY bucket, lang + + # The window manager does the writing. A pipeline sink would write every + # batch, which is the opposite of aggregating. + sink: + type: noop diff --git a/dev/bench/record.py b/dev/bench/record.py new file mode 100644 index 00000000..9dc2a34b --- /dev/null +++ b/dev/bench/record.py @@ -0,0 +1,37 @@ +# Record real Jetstream post events to gzipped NDJSON. +# +# uv run --with websockets python dev/bench/record.py +# +# Jetstream replays history from a cursor as fast as the client reads, about +# 5,000 posts a second from here against roughly 30 to 50 live, so two million +# real posts take about seven minutes rather than a day. The capture feeds +# dev/bench/replay and SQLFLOW_LEAK_JETSTREAM. Posts are public user content: +# keep captures out of the repository. +import asyncio, gzip, json, sys, time +import websockets + +hours_back, target, out = float(sys.argv[1]), int(sys.argv[2]), sys.argv[3] +BASE = "wss://jetstream2.us-east.bsky.network/subscribe?wantedCollections=app.bsky.feed.post" + +async def main(): + cursor = int((time.time() - hours_back * 3600) * 1_000_000) + n, start, last_report = 0, time.time(), time.time() + with gzip.open(out, "wt", compresslevel=3) as f: + while n < target: + try: + async with websockets.connect(f"{BASE}&cursor={cursor}", max_size=2**22) as ws: + async for raw in ws: + f.write(raw if raw.endswith("\n") else raw + "\n") + n += 1 + cursor = json.loads(raw)["time_us"] + if time.time() - last_report > 30: + last_report = time.time() + print(f"{n} events, {n/(time.time()-start):.0f}/s, stream at {time.strftime('%H:%M', time.gmtime(cursor/1e6))} UTC", flush=True) + if n >= target: + break + except Exception as e: + print(f"reconnect after {n} events: {e!r}", flush=True) + await asyncio.sleep(2) + print(f"done: {n} events in {time.time()-start:.0f}s -> {out}", flush=True) + +asyncio.run(main()) diff --git a/dev/bench/replay-soak.sh b/dev/bench/replay-soak.sh new file mode 100644 index 00000000..1f4eed58 --- /dev/null +++ b/dev/bench/replay-soak.sh @@ -0,0 +1,81 @@ +#!/usr/bin/env bash +# Run a pipeline container under Render's limits and decompose its memory +# once a minute. +# +# dev/bench/replay-soak.sh