Repository navigation
Serve backend reads over a server-to-server endpoint, with a Diesel wrapper - #27
Open
kartikparsoya-eng wants to merge 79 commits into
Open
kartikparsoya-eng wants to merge 79 commits into
kartikparsoya-eng wants to merge 79 commits into
Conversation
A Diesel or similar client sends `SELECT "t"."a", "t"."b" FROM "t" WHERE
("t"."a" = $1) ORDER BY "t"."id" LIMIT $2` with each parameter in
Postgres's binary format and its type OID. The parser now reads
double-quoted identifiers, table-qualified columns (the qualifier must be
the table read), a column list or COUNT(*) in place of `*` (validated,
then dropped: the engine follows whole rows), and `$n` placeholders in
value and LIMIT positions, decoded by OID (integers, floats, numerics,
booleans, text-like types, UUIDs, JSON, timestamps, dates) into the
value the literal would have been. DISTINCT, OFFSET and other types are
refused by name.
A backend that caches what it reads subscribes the SQL its ORM rendered,
binds and all, on a WebSocket at {base}/s2s/v1 and is told accepted or
refused, then touched with the ids of every subscription a committed
transaction changed the rows of. One task owns these subscriptions as
one more consumer of the engine's events beside the group threads, with
its own sink, and maps the engine's subscription ids back to each
connection's own. Accepted waits for hydration, so the backend's first
read of Postgres lands after the engine's and every later change is
reported.
Built with RUSTFLAGS="--cfg tokio_unstable" and --features console, the server starts console-subscriber on 127.0.0.1:6669 and the s2s task is spawned under the name s2s, so tokio-console lists it beside the group, feed and server tasks. Without the cfg the spawn is a plain one.
A subscription that asks for rows gets them instead of touches: the accepted frame carries the table, its key columns, the projection, the order and limit, and every row of the hydrated result; afterwards each transaction sends the rows that entered or changed and the keys of those that left. Rows gathered while the engine hydrates (the storage read landing, a twin's rows, commits in between) are folded by key into the snapshot, so the backend starts from a result that is complete at acceptance. The parser now returns the projection for that.
A client shaping rows needs the alias a column comes out under, not just which column it is; Diesel looks results up by that name.
…ws per table
The parser reads FROM ("a" INNER JOIN ("b" ...) ON (...)) as Diesel
renders it: each JOIN links its right item to a table on its left
through an ON equality and becomes an edge of the engine's tree, INNER
or LEFT; other ON conditions and WHERE conjuncts go to the one table
they name; ORDER BY and LIMIT belong to the root. Tables appearing
twice, RIGHT/FULL/CROSS joins, OR across tables, WHERE on a LEFT-joined
table, ORDER BY on a joined table, unqualified columns in a join and
compound keys are refused by name.
The endpoint plans each subscription through the same planner as a
Zero query (which side is read whole, how a page is read), and in rows
mode ships the hydrated result per table, root first, with each node's
key, parent and join columns, and each transaction's puts and deletes
per table. The backend composes the join.
SyncedConnection<C> implements SimpleConnection, Connection, LoadConnection (every loading mode) and R2D2Connection by delegating to the Postgres connection it wraps, sharing its ANSI transaction state so the two never disagree about depth. Diesel's three entry points for a read, a write and raw SQL now pass through one type the rest of the library can intercept. Integration tests against a local Postgres cover the DSL, sql_query by name, row-by-row loading, nested savepoints and rollback, r2d2 pooling and instrumentation, each compared with or derived from the inner connection's own behaviour.
Diesel derives PgMetadataLookup from GetPgMetadataCache, and needs it whenever a query binds a custom type. Hyperswitch's KV path collects binds itself with the connection as the lookup, so without this the wrapper cannot replace its connection type.
OwnedRow keeps each value's type OID and raw bytes exactly as Postgres sent them, with the column names shared across a result set, and implements Diesel's Row over that, so derive(Queryable), derive(QueryableByName), nested tuples and every FromSql decode a cached row the way they decode a live one. Nothing is re-encoded. capture_all copies a live cursor whole and refuses a half-read result; OwnedCursor yields the copies back the way a live cursor would.
A read on an eventual connection outside a transaction is keyed by its rendered SQL and bind bytes. On a miss the query is subscribed through the Subscriptions trait first, then read from Postgres once, and its rows kept byte for byte; a touch from the engine drops them so the next read refills. A generation per key refuses a fill the read overlapped with a touch. Entries are bounded by bytes and a size eviction ends the subscription; a touch does not. Strong connections, reads inside a transaction, INSERT/UPDATE RETURNING through the load path, and queries the engine refuses go to Postgres as before. One cursor enum carries live and cached rows through Diesel's single Cursor type. Process-wide settings let pools pick up the cache through establish, with the consistency chosen by URL.
EngineClient speaks the engine's S2S protocol on one WebSocket per process, from its own thread and runtime so the wrapper needs no async context: a subscribe sends the key's SQL and bind bytes and blocks on a channel for accepted or refused, with a timeout after which the read is served live. The engine's touched frames become Cache::touched calls through a weak handle. A lost socket refuses every waiting subscribe, clears the cache, and on reconnect resubscribes every held key under its old id. Live tests, skipped without STREAMGRES_S2S_URL, drive a running engine: a commit refills the entry (20 ms end to end locally), an unrelated commit does not touch it, deletes and inserts are changes too, a query outside the engine's subset is served live, raw SQL inside it is subscribed, and evicting an entry unsubscribes it.
Every subscribe asks the engine for rows. When the accepted frame carries them and every projected column has a type the client encodes exactly, the result is kept by primary key (materialized.rs), ordered as the query orders and cut to its limit, each cell encoded to Postgres's binary format for Diesel's own FromSql (encode.rs: integers by width, floats, text-like, UUID, JSON, dates, millisecond timestamps, enums, one-dimensional arrays), and every delta replaces the cache entry in place. The exact column types come from pg_attribute once per table, through the connection the read came in on (types.rs). A column the engine cannot reproduce exactly (numeric, microsecond timestamps, bytea) makes the query fall back to refills from Postgres, and the client records why. A reconnect refills from the engine's snapshot. Live tests: a change reaches the cache with no statement sent to Postgres; ordered, limited and counted results match Postgres and follow changes; a microsecond timestamp column falls back while arrays and jsonb are served from rows.
The store keeps rows per table by primary key and composes the result on each rebuild: root rows in the query's order, a hash join down each edge of the engine's tree, inner edges dropping a parent with no match and left edges extending it with NULLs, then the limit over the joined rows, the projection across tables, or the count. Deltas arrive per table and recompose. Diesel's rendered join shapes are pinned as a test so the engine's grammar and the client stay aligned with them. Live: an inner join through Diesel equals Postgres, follows an insert, a delete and an update on either table with no Postgres read, a left join null-extends, and a count counts joined rows.
A backend subscribes to whatever its code reads, with no screen to bound the set, so one socket per process is replaced by a pool of workers and a per-process cap, both from the environment: STREAMGRES_S2S_CONNECTIONS (default 2), STREAMGRES_S2S_MAX_SUBSCRIPTIONS (default 10000), STREAMGRES_S2S_VERDICT_TIMEOUT_MS (default 5000). EngineClient::connect reads them; with_config takes them explicitly. Each worker keeps its own thread, runtime, ids, waiting subscribes and rows-mode stores. The pool keeps one map from key to worker: a miss goes to the worker holding the key, else to the least loaded connected one; past the cap the engine is not asked and the read is served live, counted as a refusal. A dropped socket now drops only its own keys' cache entries instead of clearing the cache. Live tests: a three-socket pool spreads six keys two each and keeps a changed key where it was; a cap of two refuses the third query, serves it live, and hands a freed slot to the next miss.
The wrapper was developed as its own repository; its history is replayed under crates/streamgres-diesel and the root package becomes a workspace with one lockfile, diesel pinned at 2.2.10 (Hyperswitch's lock). The engine README's S2S section now describes rows mode and points at the crate; the Hyperswitch patch points at the crate's new path.
A read goes via Streamgres (the cache) or via Postgres (live), decided by the most specific of three flags: the query, a with_route scope on the connection, the connection's own route (chosen per pool by URL), then the process default. `query.via(Route::Postgres)` wraps the query in a marker that renders a leading SQL comment; the wrapper strips it at the interception point, so the cache key and the SQL the engine sees are the plain query and flagged and unflagged reads share one entry. Consistency::Eventual/Strong stay as the earlier names.
cache::Limits bounds the cache by bytes, idle time and time to live, read from STREAMGRES_CACHE_MAX_BYTES, STREAMGRES_CACHE_IDLE_MS (default 30 min) and STREAMGRES_CACHE_TTL_MS (default off). Expiry ends the subscription like a size eviction does, and both now bump the key's generation so a fill in flight for an entry that just ended is dropped rather than kept with nothing to update it; the rows-mode fill goes through the same generation check.
One engine streams one database and keeps one cache fresh, so the process settings now pick the cache as well as the route by the URL a connection is established with. Settings::single keeps the one-database shape (every pool shares the cache, the route per URL); Settings::by_url gives each database its own cache attached to its own engine, and a URL with no attachment a passthrough connection. The Hyperswitch patch follows.
A Postgres read is a bypass, not a write-through, so a flow that writes and then reads through the cache could miss its own write until the engine ships it. After any write on a connection (execute, raw SQL, INSERT ... RETURNING, a transaction's COMMIT) that connection's reads go to Postgres for STREAMGRES_READ_YOUR_WRITES_MS (default 1000, 0 off, set_read_your_writes per connection), counted in Stats::after_write. The entry is left alone; the engine's delta still corrects it.
AsyncSyncedConnection (feature `async`) wraps diesel-async's AsyncPgConnection and implements AsyncConnection and PoolableConnection, sharing the routing, cache, read-your-writes window, settings and engine side with the blocking wrapper. Its load future may neither borrow the connection nor carry the query, so the key, route and cache check happen synchronously and the futures a read may need are built up front: the inner load, polled only when the read goes to Postgres, and a one-shot lookup of every table's column types, polled the first time the engine ships rows for a table not yet known. A subscribe runs on a blocking thread. The first read with a user-defined-type bind is live while its OID is resolved and remembered. Subscriptions::subscribe now returns an Outcome: a verdict, or NeedsTypes with the tables whose types the engine side lacks, which the caller looks up its own way and hands to materialize. subscribe_with does both on a blocking connection, so the sync path is unchanged. Tests: six against a fake engine (hit, refill, route flag and scope, transaction, read-your-writes window, refused), one live in rows mode through the catalog load. The microsecond fallback live test now tolerates the documented redundant touch after the fill.
It writes and reads on one connection on purpose, to watch the engine carry the change; with the window on, the read after the insert went to Postgres and never subscribed.
…hey lack
The column types of a table were looked up once per process, on the
first rows-mode subscribe of it, and kept. After a migration added a
column, a query projecting it fell back to refills for good ("column
`kind` is not in the types of `tickets`"), and a new enum type's bind
stayed in refill mode even after the engine restart that made the bind
acceptable. A snapshot naming a column the known types lack now counts
the table as missing, so its types are looked up again; the fetched
ones replace the stale ones. On the async wrapper, whose catalog read
is pipelined ahead of the subscribe, a stale table it has no read for
refuses that one read transiently ("the column types are being
reloaded"), unsubscribes, and the next miss reloads the catalog and is
served from rows; before, that case fell back to refills for good too.
Unit test on the client, and the example's migration scenario covers
the whole path against an engine.
…heck, migrations, lag and scale The query list grows from nine shapes to twenty-three, each saying what it expects of the engine (rows, refills or live) and the scenario test holding it to that: a two-column key, IS NULL, OR, <> and NOT IN, a timestamp range, a hundred-id IN list, empty results that must stay empty, the largest result under a small per-entry cap, a left join with an empty side, and the shapes the engine refuses (array @>, DISTINCT, GROUP BY, MAX) so the refusals are documented rather than discovered. A further shape reads the column a migration adds. The writer draws from a mix of every kind of change: status, comment, insert, delete with cascade, move between organizations, rename under the joins, assignee flipped to NULL and back, multi-row transaction, rollback, and an oversized comment body updated without touching it. EXAMPLE_WRITER_MIX narrows it; the counters are on /stats. The demo checks itself: `example verify`, `/verify` and EXAMPLE_VERIFY_MS read every lookup through the cache and through Postgres and compare, re-reading a differing pair until it agrees or a settle time runs out; a pair that never agrees is a mismatch, kept with both answers. `example lag` measures commit-to-cached-answer under a write rate. `example migrate widen|new-type|drop` runs a schema change of each kind the engine treats differently, and compose restarts the engine after the one that stops it. EXAMPLE_SCALE multiplies the seed. The scenario test adds: a query flag beating the route both ways, a scope, and the read-your-writes window ending; an oversized result served but not kept; sixteen readers under the whole writer mix; the same under an engine restart; and the three migrations. The load tests tell a wrong answer in this process's cache from one in the engine by reading again through a fresh subscription; the first fails, the second is reported and fails under STREAMGRES_STRICT. Seen on main under that test: after a restart under load, a row updated a second or two before the restart is missing from the timestamp-range shape, in a fresh subscription too, until its next write.
…e UTC it names A time in a filter is spelled as a zone-less timestamp (`TIMESTAMP 'epoch' + N microseconds`, so an index on the column answers it), which PostgreSQL reads in the session's zone when the column is `timestamptz`. The engine keeps every time as UTC microseconds, so on a server whose default zone is not UTC a range on such a column missed its rows by the zone's offset: a storage read for "created between 09:00 and 12:00" came back empty under Asia/Kolkata. The miss was hidden most of the time, since a read positioned at a snapshot before the seed had the seed's inserts replayed into its empty result from the delivered-writes buffer; after an engine restart that buffer is empty, so every row of the range shape was missing until its next write. Every connection the storage opens, the reads' and the first, and the replication session now run with `TimeZone=UTC`, set after whatever options the connection string carries so the zone wins; the feed's `timestamptz` text is canonical as a side effect. A live test connects with Asia/Kolkata asked for in the DSN and reads a range on a `timestamptz` and a `timestamp` column; a unit test covers the session options. The runtime logs, at debug, each read's row count at its snapshot and after it is brought up to the engine's position, which is what located this.
The restart test reported engine-side disagreements rather than failing on them while the cause was open; the cause was the engine reading a time range in the server's own zone, now fixed, so both kinds fail, named.
The verdict timeout: a subscribe nobody answers is refused when the wait runs out, with a reason that passes, and its id is withdrawn (unit test on the worker; and against a fake socket that greets and never answers, the read is served live, the timeout is not remembered as a refusal, and the next read asks again). The reconnect cadence: a socket dropped as soon as it is made is tried again every second. The eviction order: once the cache is past half full, a key asked for again and again gets in, displacing an entry read once, and a run of keys each asked for once cannot displace it. A column added while the engine is down: when it returns, a query naming the column is live once while the new type's OID resolves, live once more while the table's types are fetched afresh, then served from engine rows. And the endpoint spoken to directly, frame by frame, with no client library in between: hello, ping, a subscribe in rows mode and in touch mode, the refusals, a change as a delta and as a touch, an unsubscribe that stops the frames, and a frame it cannot read. Writing the last one found that a TRUNCATE of a published table stops the engine the way a dropped column does; the demo README says so.
Anyone who could reach the port could subscribe in rows mode and read every row of every followed table. With STREAMGRES_S2S_TOKEN set, the upgrade must carry Authorization: Bearer <token> (compared in constant time) and is answered 401 otherwise; unset, the endpoint behaves as before. The diesel client sends the same variable's token. One connection now holds at most STREAMGRES_S2S_CONNECTION_SUBSCRIPTIONS subscriptions (10000; 0 lifts it), registering ones counted. The refusal reads "the subscription limit of ...", which the client already treats as transient. A row shared by several rows-mode subscriptions is turned into JSON once per delta instead of once per subscription.
value_json gave a Date or Datetime value in milliseconds while the S2S protocol and the client read microseconds. The catalog maps every Postgres time to Timestamp (an Int in microseconds), so no live column took those arms, but nothing kept it that way. Also: why the materialized key separator cannot collide, with a test; is_multiple_of where clippy asks; async_live.rs formatted.
…failures The per-connection subscription limit bounded nothing while a caller could open any number of connections. STREAMGRES_S2S_MAX_CONNECTIONS (64; 0 lifts it) caps them; past it the upgrade is answered 503. The engine's hello now carries max_subscriptions. The client passes over a socket at that limit and, with every socket full, refuses the subscribe itself, so a read goes live at once instead of after a refused round trip on every miss. A refused upgrade (401, 503) is logged at most once a minute per kind, with a count of the ones held back: a backend with the wrong token retries every second, and so can anyone who reaches the port. The client keeps why its latest connect failed (the token, a 503, a URL or token that cannot go in a request) and puts it in each "not connected" refusal and in EngineClient::connect_errors; before, a rejected client retried in silence and only went live. The bearer scheme is matched without regard to case, and the duplicate check and in-flight count are one pass. The live test sends the token when given one and checks that a missing or wrong one is answered 401.
write_value hands map and set cells to the JSON builder, which now writes in microseconds; it now takes the unit, so each path writes one unit for every time it carries.
…eamgres-demo The wrapper follows the Diesel ecosystem's naming (diesel-async, diesel_migrations): package diesel-streamgres, imported as diesel_streamgres, in crates/diesel-streamgres. The ticket desk becomes streamgres-demo in crates/streamgres-demo, and everything that still said "example" says "demo": the binary (demo seed | serve | bench | verify | lag | migrate, /app/demo in the image), its DEMO_* variables, the compose service, and docker/demo/Dockerfile. The wrapper's tests default to the database diesel_streamgres_test.
Brings in compound join keys (#24), PgBouncer for reads (#23), poke parts sized in bytes (#15), the application server's ASTs (#4) and the fixes since (#16, #20). - Poke parts: main's byte-sized parts replace STREAMGRES_ROWS_PER_PART; a delete still writes its time key through the table's declared types. - Reads: the read endpoint (STREAMGRES_READ_PG_DSN, possibly PgBouncer) gets the same session as before, pinned to UTC with the read timeout as its statement_timeout, on the first connection and on every pooled one. - The composite-key SQL test expects microseconds, the unit times are kept in on this branch.
docs/design.md now describes the wrapper as it is: where it sits between Diesel and Postgres, its modules, the interception surface, the cache and the two ways a result is kept fresh (engine rows, or touched and refilled from Postgres), the engine client, the async adapter, the tests, and how to run it all locally.
The writer kept a clone of the connection's Outgoing for its byte count, and with it a sender of its own queue, so the queue never closed: every disconnect waited out the five-second drain, logged that the backend was not reading, and was aborted before the close frame went out. The writer now holds only the byte count. Also: a zero silence no longer panics the ping interval, and the bearer-token test has its doc comment back.
…ources The style rules want no comments inside function bodies and a `///` on every function and type. A comment that explained one branch now lives on the function, or the branch became a named function with the comment as its doc: the S2S task's request and event handlers, the delta frame builder, the wrapper's frame handlers per tag, its reconnect bookkeeping and the async read's type lookup. No behaviour changes.
… side The same pass over the parser and the lines this branch wrote in the server's client, wire and catalog code. Older code those files carry from main is left as it is. Two stray doc fragments in the parser, left on the wrong functions when `select` moved to its own module, are gone. The binds tests build their arrays and numerics through one helper each instead of byte-by-byte comments. No behaviour changes.
The rest of the style pass, over the files this branch adds: each test's step comments become paragraphs of its doc, in order, and every test, helper and type has a `///`. The demo's handlers and query builders say which lookup and which shape each one is. `TicketSla`'s doc said `demo migrate new-type` adds its column; the seed does, and the doc now says why the query reading it refills from Postgres. No behaviour changes; `demo --help` reads as before.
The end-to-end run showed `/orgs/1/kind/bug` going to Postgres twice after `demo migrate new-type`, not once: the first read teaches the wrapper the type's OID, the second teaches the engine the type and the wrapper the table's new column. The README said one.
An S2S query of thirty `AND`ed `OR` pairs, 2.6 KB of SQL, expands to 2^30 disjuncts when the engine registers it: the engine grew to 13.8 GB and was killed for memory. Twenty pairs survived but held the engine's thread for 21 s, so every other subscriber's changes waited and the feed fell 20 s behind. An `IN` list does the same in proportion: a million values held the thread for 6.5 s and pushed the demo's commit-to-cache p99 from 4 ms to 1.2 s. The endpoint now measures a query before planning it, from its filter trees without expanding them (an `AND` multiplies its parts' alternatives, an `OR` adds them), and refuses one that expands to more than STREAMGRES_S2S_MAX_DISJUNCTS alternatives (1024) or holds more than STREAMGRES_S2S_MAX_LIST_VALUES `IN` and `NOT IN` values (10000); 0 lifts either. The backend reads such a query from Postgres, as it does any refusal.
A single `infinity` in a published timestamp or date column stopped the engine: the change feed's parser had no reading for it (nor for a BC date or a year past 9999), and the snapshot read's cast to `int8` failed in PostgreSQL. Restarted, the engine stopped again on the same row, until someone changed it by hand. The feed's parser now reads everything PostgreSQL prints, without chrono's range: `infinity` and `-infinity` as the extreme `int8` microseconds, which order where PostgreSQL orders them; BC dates; years past 9999; offsets with seconds. A time past what the microseconds count saturates. The snapshot read maps the same way in SQL, and a condition on an infinity renders as `infinity` again. The wrapper encodes the extremes as PostgreSQL's binary infinities, so a cached row decodes exactly as the live one does, errors included.
…xt ping The client looked at how long a socket had been silent only when a ping was due, every third of the window, so a socket that last heard something between two pings was given up on up to a third of a window late. With the default 30 s window, an engine frozen mid-stream (here by SIGSTOP) had its rows served from the cache for 37.7 s while PostgreSQL moved on. The silence is now looked at every thirtieth of the window, a second at most, beside the pings.
An engine killed rather than stopped (out of memory, here) leaves its replication slot behind, and each restart makes a new one. The left slot held 132 MB of WAL within minutes and grows with every write until it is dropped. docs/deploy.md advises max_slot_wal_keep_size as the safety net; the demo's Postgres now sets it.
A backend whose S2S URL pointed at an engine streaming another database with the same schema (a staging engine in production, say) served that database's rows as its own: no error, no warning, the wrong title for ticket 14 from the engine and then from the cache, while Postgres said otherwise. The engine's `hello` now names the database it streams: the cluster's `system_identifier` and the database's name. The first miss through a cache reads the same of the database its reads go to, through its own connection (readable by any role), before it subscribes. A socket whose engine names another database is not asked for rows: each subscribe is refused naming both, and the reads go to Postgres. A key resubscribed after a reconnect is checked when its `accepted` comes. An engine that names no database, one from before this, is trusted.
The engine_live tests count the statements Postgres sees, and the first miss through a cache now also reads which database it reads from; the harness teaches the cache that up front, so the counts are the test's own. The pool and subscription-cap tests built their clients from the default config, which carries no token, so an engine started with one answered 401; they start from the environment's config instead.
The silence window is how long a cache keeps serving a frozen engine's rows: an engine stopped without closing its sockets, or a network gone without a reset, is noticed only when nothing has been heard for that long. Thirty seconds was generous; ten still allows three pings, one every 3.3 s, and an engine that answers them is never taken for dead however busy. Measured against an engine stopped with SIGSTOP, the socket is now given up on 10.0 to 10.2 s in. STREAMGRES_S2S_SILENCE_MS still sets it.
A large transaction or a stalled engine delayed the cache without the
wrapper knowing: a 300k-row transaction left it 1.8 s behind with every
read still served from the cache. The engine now measures how far
behind it is and tells each S2S socket, and the wrapper reads Postgres
past a threshold.
The engine writes a numbered heartbeat into the WAL every 500 ms
(pg_logical_emit_message, not transactional, prefixed with its slot's
name; STREAMGRES_HEARTBEAT_MS, 0 off), from the reads runtime so a
stalled engine thread shows as lag. The feed turns its own heartbeat
into a mark behind every transaction committed before it; once the
engine has applied those, the service sends every consumer a Heartbeat
event behind their events, and the S2S task marks it applied after
queueing their deltas. Everything committed before the newest applied
heartbeat was sent is then on its way to the backends, so that
heartbeat's age bounds how stale they are. The endpoint sends
["lag", {"ms": n}] right after hello, before any answer, and every
500 ms; null before the first heartbeat.
The wrapper adds the time since the last lag frame, so an engine that
stops sending them counts as falling behind too, and reads go live
while the largest lag over the connected sockets is past
STREAMGRES_S2S_MAX_LAG_MS (2000; 0 trusts the cache). A miss answered
with engine rows from a socket found behind keeps them for later reads
and serves this one live. cache::Stats counts these reads as lagging,
and the demo's /stats shows them with the lag. Older wrappers ignore
the frame; an engine that never sends it is trusted, as before.
Measured on the demo with the writer and the verify loop running: a
300k-row transaction raised the lag to 2.7-3.6 s and reads went live
past 2 s; an engine stopped with SIGSTOP was read around after 1.5 s;
verify found no mismatch in either.
Main's plan estimates (#18) added Translated::uncursored. An S2S query never has a keyset cursor, so the endpoint builds it as None.
The check that an engine streams the backend's own database failed open in three places. A wrapper whose first probe of its database failed (a statement timeout, a reset connection) recorded "unknown" for the life of the cache, and an unknown database trusts every engine. The engine, when its own identity read failed, greeted with "database": null, which the wrapper takes for an engine from before the check, and trusts for as long as the socket stays up. The wrapper now records only a database it read: a probe that fails sends that read to Postgres without asking the engine, and the next miss probes again. The engine reads its identity once, since it cannot change while it runs, and keeps it; while it cannot be read it closes each new S2S socket without a greeting (logged at most once a minute), so the backend reconnects and reads Postgres meanwhile. That also takes a storage round trip off every connect.
Three faults in the lag measurement: - The emitter numbered heartbeats from 1 each time it reconnected, and a number at or below the newest applied is not waited on, so after one blip of its connection the reported lag only grew, and backends read Postgres until the new numbers caught up, an hour after an hour's uptime. The numbers now come from Freshness and keep rising. - A message outside a transaction writes no commit record, and logical decoding reads only flushed WAL. On PostgreSQL 16 a lone heartbeat waited 4 to 5 s or more for the WAL writer; the engine's own snapshot rotation, which flushes each second, hid this. On 17 the message is now written with flush => true; before 17 the statement also takes a transaction id, whose commit flushes it (two ids a second). With rotation slowed to a minute and nothing else writing, the lag stays at 20-500 ms on both. - The heartbeats waiting to be applied were unbounded while none was; they are capped at 1024, dropping the second oldest, which can only overstate the lag. A heartbeat failure is now logged again after heartbeats have resumed, not only the first time. Until the first heartbeat is written the lag stays unknown and backends trust the engine, so the engine on a standby, which cannot write one, keeps serving; the doc said otherwise.
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.
The change
Backend services can now read through Streamgres over a server-to-server (S2S) endpoint. The repository also gains a Diesel wrapper that serves a backend's reads from the engine and falls back to Postgres.
src/client/s2s.rs): a token-authenticated WebSocket, separate from the client sync protocol.SELECTs with their binds. The engine answers with the query's rows and later changes, atouchednotice for queries it can watch but not ship, or a refusal with the reason.lagframe: the most it can be behind the database, measured from heartbeats it writes into the WAL (src/sync/heartbeat.rs).IN/NOT IN/= ANY,IS [NOT] NULL,OR, ranges,ORDER BY/LIMIT/OFFSETandCOUNT.crates/diesel-streamgres: a DieselConnectionwrapper, sync and async.touched.docs/design.mddescribes the design.crates/streamgres-demoanddemo/: a docker-compose stack with a ticketing backend and a background writer.DEMO_VERIFY_MS).Tests
tests/s2s_live.rsandtests/pg_live.rsagainst Postgres.cached_reads,async_cached,engine_live,silent_engine,passthrough,shape_coverage,owned_row.tests/scenario.rs.widen,new-typeanddropmigrations and an engine restart.Verified: 872 tests (
cargo test --workspace) and clippy 1.89-D warnings.