[SLOP(claude-opus-4-8-high)] feat(ups): table-backed postgres transport with coalesced doorbell

[MasterPtato]

No description provided.

[MasterPtato]

Stack for rivet-dev/rivet

• #5552 [slopfix] fix(universaldb): match fdb conflict semantics in …
• #5551 [slopfix] fix(universaldb,util,pegboard-envoy): index dedup …
• #5398 feat(kitchen-sink): add churnDb actor and seed driver for co …
• #5397 chore: Clean up
• #5382 [slopfix] docs(self-hosting): correct postgres config format …
• #5356 [slopfix] fix(pegboard): decrement kv_storage_used metric on …
• #5354 [slopfix] refactor(universaldb,ups): single/multi-node udb o …
• #5353 [slopfix] perf(universaldb): fold postgres drain_batch into …
• #5342 [SLOP(claude-opus-4-8-high)] perf(universaldb): batch leader …
• #5341 [slopfix] docs(self-hosting): promote postgres from experime …
• #5340 [slopfix] fix(self-host): repair dev compose generation, kit …
• #5337 [SLOP(claude-opus-4-8-high)] feat(universaldb): graceful pos …
• #5336 [slopfix] test(universaldb): postgres leader failover + abor …
• #5335 [SLOP(claude-opus-4-8-high)] feat(ups): table-backed postgre … 👈
• #5329 [SLOP(claude-opus-4-8-high)] feat(universaldb): postgres lea …
• #5327 [SLOP(claude-opus-4-8)] feat(pegboard-envoy): rate-limit env …
• #5326 [SLOP(claude-opus-4-8)] feat(util): add rate limiter primiti …
• #5322 [SLOP(claude-opus-4-8)] feat(rivetkit-core): add serde durat …
• #5321 Add observation for serde_bare
• #5320 Add observation and ser/de metrics for serde calls
• #5314 [SLOP(claude-opus-4-8-medium)] chore: add profiling cargo pr …
• #5306 [SLOP(claude-opus-4-8)] fix(kitchen-sink): serve built front …
• main

Get stack: forklift get 5335
Push local edits: forklift submit
Merge when ready: forklift merge 5335

_{change xutxrzrv}

[claude]

Code Review: Table-Backed Postgres Transport with Coalesced Doorbell

The architectural direction is good. Moving payload delivery out of the 8000-byte NOTIFY envelope and into a table removes the base64 workaround, decouples delivery correctness from whether a NOTIFY arrives, and the leading/trailing-edge debounced doorbell is a sensible way to bound NOTIFY fan-out. A few things are worth addressing before merge, especially the cursor-based polling scheme.

Correction to a prior review pass on this PR

An earlier automated pass on this PR flagged existing.subscribe() (mod.rs:432) as a compile error, on the theory that scc::hash_map::OccupiedEntry doesn't Deref to the value. I checked this against the rest of the codebase and that claim is wrong: cache/src/req_config.rs:150 calls .subscribe() directly on an Entry::Occupied(broadcast) binding, and pegboard-gateway2/src/shared_state.rs:412-415 mutates fields directly through Entry::Occupied(mut entry). Both are pre-existing, compiling code, confirming OccupiedEntry derefs to V, so existing.subscribe() resolves fine.

Bug: subscriber cursor can permanently skip messages under concurrent publishes to the same subject

current_max_id() (mod.rs:407-420) snapshots MAX(id) from ups_messages, and both subscribe() (mod.rs:560) and PostgresSubscriber::fetch() (mod.rs:693-723) treat id > cursor as "not yet delivered." But id comes from BIGSERIAL/nextval(), which Postgres explicitly does not guarantee to be issued or committed in order across concurrent transactions. A transaction that grabs a lower id can commit after one that grabbed a higher id.

Concretely: transaction A inserts a broadcast row for subject X and gets id=100, then in the same transaction queries ups_queue_subs and inserts N queue rows before committing (try_publish_to_db, mod.rs:493-544), so its commit latency is variable. Transaction B, publishing to the same subject shortly after, gets id=101 and commits quickly with no active queue subs. If B commits before A, any subscriber that reads MAX(id) or advances its cursor during that window sets cursor = 101. When A's row (id=100) finally commits, it will never satisfy id > 101 for that subscriber again; the message is silently dropped, not just delayed. This isn't limited to the initial subscribe race either: fetch() advances self.cursor to the max id seen in every poll result, so an already-running subscriber can hit the same gap on any subsequent poll.

This undermines the "no subscribe/publish race" claim in the comments at mod.rs:557-559. Since try_publish_to_db's duration varies with the number of active queue subscribers on a subject, out-of-order commits are a real risk under concurrent publish load, not just a theoretical edge case. Worth a fix (a small time-based safety margin instead of a hard id-based cursor is the usual approach for this class of problem) plus a regression test with overlapping concurrent publishes to the same subject.

Durability regression: ups_queue_messages is now UNLOGGED

The table DDL (mod.rs:168) changes ups_queue_messages from a logged table to CREATE UNLOGGED TABLE. The comment justifying UNLOGGED (mod.rs:147-150) is scoped to the new broadcast table matching NATS at-most-once semantics, but queue subscribers are a different delivery model (each message claimed exactly once via FOR UPDATE SKIP LOCKED) where callers plausibly expect durability across a Postgres crash or restart. Worth confirming this is intentional. Also note CREATE TABLE IF NOT EXISTS won't convert an already-deployed logged table, so existing clusters and fresh installs would silently diverge in durability behavior.

Missing index for the new GC query

ups_messages GC (mod.rs:183-204) runs every 5 seconds (MESSAGE_GC_INTERVAL) with DELETE FROM ups_messages WHERE created_at < NOW() - ..., but the only index defined is (subject_hash, id) (mod.rs:158-159). Since this table now carries all broadcast pubsub traffic engine-wide, this GC query is a sequential scan of the whole table every 5 seconds under load. Consider an index on created_at, or restructure GC to delete by an id watermark derived from time so it can reuse the existing index.

fetch() has no bound on backlog size

PostgresSubscriber::fetch() (mod.rs:697-723) has no LIMIT and buffers every matching row into an unbounded VecDeque. The previous implementation had natural backpressure via a bounded (1024) broadcast channel with lag reporting; a subscriber that falls behind on a bursty subject can now pull an arbitrarily large result set into memory in one query. Consider a LIMIT per fetch with a loop, or an explicit cap with lag tracking.

Minor: doorbell NOTIFY competes with the main pool

Doorbell::notify_shard (doorbell.rs:119-132) calls self.pool.get() for every NOTIFY, up to about 200/sec per shard times 32 shards under sustained load. That is real contention against the same pool used for publish/fetch/claim queries. A dedicated long-lived connection (or a small reserved sub-pool) for doorbell NOTIFYs would avoid competing with request-path queries during bursts.

Minor: cleanup-task UNLISTEN/remove race can permanently downgrade a subscriber to poll-only latency

In spawn_shard_cleanup_task (mod.rs:465-491), the tx.receiver_count() == 0 check and the subsequent UNLISTEN plus shard_subscriptions.remove_async are not atomic with respect to a concurrent ensure_shard_listen on the same shard. If a new subscriber grabs a receiver from the same tx in that window, the entry still gets removed from the map and the channel gets UNLISTENed. The new subscriber keeps a working rx since it holds a clone of tx, but no future NOTIFY for that shard will reach it again, since poll_connection's map lookup no longer finds an entry and any later subscriber on that shard creates a fresh tx/LISTEN. Correctness is preserved by the poll backstop, but that one subscriber is permanently downgraded to about 1s latency instead of a transient hit. Low severity, worth a comment at minimum, or a fix such as re-checking receiver count after removal.

Test coverage

No tests were added alongside this change. The existing generic driver tests (tests/integration.rs) don't appear to exercise concurrent publishes to the same subject, so they would not catch the cursor race above. The doorbell's coalescing and debounce timing (doorbell.rs) also has no dedicated unit test. Given this touches delivery correctness for a shared infra primitive, both would be valuable additions.

Nit

hash_subject's doc comment (mod.rs:392-399) now correctly notes that hash collisions are possible and resolved via the stored subject text for broadcast messages. ups_queue_subs/ups_queue_messages still key purely on subject_hash/queue_hash with no stored plaintext to disambiguate a collision. Pre-existing gap, not introduced here, but worth a follow-up now that the broadcast path sets a precedent for verifying the real subject.