[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

Current stack:

• [SLOP(claude-opus-4-8-high)] perf(universaldb): batch leader apply and fold follower commit round-trips #5342 vtrtqpyp · 2026-06-26 02:20:21
• [slopfix] docs(self-hosting): promote postgres from experimental to recommended OSS multi-node backend #5341 zsmysquk · 2026-06-26 02:20:19
• [slopfix] fix(self-host): repair dev compose generation, kitchen-sink serverful runner, and postgres/udb bootstrap #5340 wpvkrymu · 2026-06-25 22:53:05
• [SLOP(claude-opus-4-8-high)] feat(universaldb): graceful postgres leader handoff on shutdown #5337 mvxpzxsl · 2026-06-25 20:34:04
• [slopfix] test(universaldb): postgres leader failover + abort resolver task on driver drop #5336 tvmntvqo · 2026-06-25 20:34:03
• [SLOP(claude-opus-4-8-high)] feat(ups): table-backed postgres transport with coalesced doorbell #5335 xutxrzrv · 2026-06-25 20:34:01 👈
• [SLOP(claude-opus-4-8-high)] feat(universaldb): postgres leader-resolver driver overhaul #5329 xrrmomwz · 2026-06-25 00:55:07
• [SLOP(claude-opus-4-8)] feat(pegboard-envoy): rate-limit envoy ws ingress and cap get_pages, trim unused metrics #5327 tnzyzqrl · 2026-06-24 20:57:36
• [SLOP(claude-opus-4-8)] feat(util): add rate limiter primitive and ingress throttles for actor create and gateway websocket #5326 pnyswwtm · 2026-06-24 20:57:35
• [SLOP(claude-opus-4-8)] feat(rivetkit-core): add serde duration and size metrics #5322 nqkrytvr · 2026-06-23 22:49:13
• Add observation for serde_bare #5321 oskoutmw · 2026-06-23 20:23:10
• Add observation and ser/de metrics for serde calls #5320 ztnuwvts · 2026-06-23 20:23:09

Dependencies:

• [SLOP(claude-opus-4-8-medium)] fix(rivetkit-napi): reap completed tasks from event-loop JoinSet (#5315) #5314 xxqwtrzp · 2026-06-22 22:42:01
• main

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

[claude]

Code Review: Table-backed Postgres Transport with Coalesced Doorbell

The architectural direction here is solid. Moving broadcast payloads from NOTIFY (8KB cap, transient) to a table (unbounded, poll-backstopped) eliminates the base64 encoding workaround, removes the per-subject LISTEN proliferation, and makes delivery correctness independent of whether a NOTIFY arrives. Good change overall. A few issues worth addressing:

Bug: existing.subscribe() will not compile (line 432)

scc::hash_map::OccupiedEntry does not implement Deref<Target = V>, so existing.subscribe() on an OccupiedEntry won't resolve to broadcast::Sender::subscribe(). Should be existing.get().subscribe().

Resource leak: Doorbell task runs forever with no cancellation path

Doorbell::new spawns a task holding Arc<Doorbell>. The task runs an unbounded loop with no cancellation token. When PostgresDriver is dropped, the task continues holding Arc<Doorbell> (and transitively Arc<Pool>), keeping the connection pool alive. Per repo guidelines: spawned futures capturing heavy resources must have a guaranteed completion path via a CancellationToken.

Semantic change: ups_queue_messages is now UNLOGGED

The diff changes ups_queue_messages from a logged to an UNLOGGED table. On a Postgres crash, all in-flight queue messages are silently lost. The broadcast table being UNLOGGED is intentional (matching NATS at-most-once), but queue subscribers typically expect at-least-once delivery. This is a behavioral regression worth calling out explicitly.

Missing index on created_at for broadcast GC

The GC query DELETE FROM ups_messages WHERE created_at < NOW() - (...) has no index on created_at — only (subject_hash, id) is indexed. This forces a sequential scan every 5 seconds. An index on created_at is needed. Also consider a LIMIT clause to bound each GC pass under high load.

Minor: cleanup task UNLISTEN race

In spawn_shard_cleanup_task, after token.cancelled() fires and receiver_count() == 0 is confirmed, there is a window before remove_async completes where a concurrent ensure_shard_listen can subscribe to the Occupied entry and receive a receiver for a sender that then disappears from shard_subscriptions. The new subscriber won't receive doorbell wakeups until the 1-second poll backstop. Low severity since the poll backstop covers it.

Nit: double-hashing in shard_for

shard_for takes a subject_hash string (already hex-encoded hash) and hashes it again via DefaultHasher. Simpler to parse the hex digits directly and reduce mod 32.

Nit: quoted numeric literal in SET

format!("SET idle_in_transaction_session_timeout = '{}'", LISTEN_IDLE_IN_TRANSACTION_TIMEOUT_MS) — idle_in_transaction_session_timeout accepts an unquoted integer. Removing the quotes is more idiomatic. No injection risk since it's a compile-time constant either way.