perf(universaldb): time each drain batch statement to localize leader apply stalls

[MasterPtato]

No description provided.

[MasterPtato]

Stack for rivet-dev/actors

• #5704 fix(universaldb): run postgres transaction reads concurrentl …
• #5705 test(gasoline): reproduce slow workflow pulls on postgres
• #5702 fix(universaldb): chunk postgres commit requests that exceed …
• #5703 fix(universaldb): clear postgres key ranges one statement pe …
• #5706 perf(universaldb): time each drain batch statement to locali … 👈
• #5695 fix(universaldb): report the last error when transaction ret …
• main

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

_{change uzzxusqp}

[railway-app]

🚅 Deployed to the actors-pr-5706 environment in rivet-frontend

Service	Status	Web	Updated
frontend-inspector	😴 Sleeping (View Logs)	Web	Sep 11, 2026 at 11:44 pm UTC
frontend-cloud	😴 Sleeping (View Logs)	Web	Sep 11, 2026 at 11:42 pm UTC
kitchen-sink	😴 Sleeping (View Logs)	Web	Sep 11, 2026 at 11:39 pm UTC
ladle	✅ Success (View Logs)	Web	Sep 11, 2026 at 9:36 pm UTC
mcp-hub	✅ Success (View Logs)	Web	Sep 11, 2026 at 9:34 pm UTC
website	❌ Build Failed (View Logs)	Web	Sep 11, 2026 at 9:34 pm UTC

[the-company-company]

🟠 1 medium · 🔵 2 low

_{Reviewed commit 1580368.}

[the-company-company]

🟠 Medium · Account for awaited response delivery

batch_ms is measured only after the watermark publish and the for_each_concurrent(... respond(...)) loop, but neither operation has a phase timer. In multi-node mode Responder::respond awaits async_nats::Client::publish, which can block on client backpressure. A slow batch can therefore still have every reported phase near zero, defeating this instrumentation's purpose and contradicting the “sum roughly batch_ms” comment.

Measure these post-commit awaits (at least reply delivery, and ideally watermark publish) or emit the batch timing before them so the remaining time is explicit.

[the-company-company]

🔵 Low · Fail the harness when a writer fails

The JoinHandle results are discarded. Since write_state unwraps the transaction result inside its spawned task, any failed or panicked writer only produces a JoinError; this loop continues and prints a partial batch report as if the load completed. That can turn a connection, election, or commit regression into misleading timing output.

Await each handle with unwrap() (or otherwise propagate its error) before reporting.

[the-company-company]

🔵 Low · Populate the advertised wide table

Each round writes only concurrency distinct workflows. With 12 rounds and concurrency 96, this scenario visits workflow IDs 0 through 1151 exactly once; the other 7,040 of the advertised 8,192 workflows are never written. The reported range-delete behavior is therefore from a much smaller table than the label and documentation claim.

Increase the rounds/load to populate the intended cardinality, or change the scenario’s cardinality and label to 1,152.

[claude]

Review

This is a pure diagnostics change (per-phase timing fields on drain_batch + a docker-backed, #[ignore]d investigation harness). Low risk overall; a few notes below.

engine/packages/universaldb/src/driver/postgres/resolver/mod.rs

• begin_ms computed differently than every other phase (line 421):

  let begin_ms = batch_start.elapsed().as_millis() as u64 - pool_wait.as_millis() as u64;

  Every other phase (prepare_ms, resolve_ms, atomic_ms, fold_ms, range_delete_ms, apply_ms, commit_ms) uses a fresh Instant::now() right after the previous phase ends and calls .elapsed() on it. begin_ms instead derives its value by subtracting two independently-truncated millisecond counts. It happens to be sound (the total duration up to that point always encompasses pool_wait, so this can't underflow), but it's inconsistent with the pattern used everywhere else in the function and harder to reason about at a glance. Consider let begin_start = Instant::now(); right after pool_wait is captured, then begin_ms = begin_start.elapsed().as_millis() as u64 for consistency.

• A couple of small gaps aren't attributed to any phase: building dedup_nids/dedup_seqs (the loop right after begin_ms) and the job-classification loop before resolve_start are both outside any timer, as is the upsert_count/point_delete_count/range_delete_count/upsert_bytes computation plus unzip() calls between fold_ms and range_delete_start. The trailing comment already says the phases "sum to roughly batch_ms" so this is presumably intentional, but worth double-checking none of these untimed spans is where the reported 7s stalls actually live — otherwise the new fields could all look fast while batch_ms stays high, which would undercut the stated goal of localizing the stall.

engine/packages/universaldb/tests/leader_apply_stall.rs

• Solid adherence to the repo's testing conventions: real Postgres/NATS containers (no mocking), lives under tests/ rather than an inline #[cfg(test)] mod, and the doc comment correctly cites the exact reason a global subscriber is needed instead of thread-local set_default (spawned drain-loop task).
• Minor inconsistency: collector() calls tracing_subscriber::registry().with(BatchLayer(...)).init(), while every other integration test in this same crate (failover.rs, pool_starvation.rs, integration.rs, rocksdb.rs, kv_upsert_scaling.rs) uses .try_init() defensively. It's guarded by OnceLock here so it should only run once in-process, but try_init() would be cheap insurance and matches the established local convention.
• This harness intentionally has no latency assertions (documented as "not to assert a latency bound"), so it doesn't add regression coverage for the stall itself — reasonable for an exploratory tool, just flagging it's not a test in the pass/fail sense.

Other

• No security or correctness concerns; the new fields are plain per-batch counts/durations, not Prometheus metric labels, so the unbounded-cardinality metrics rule doesn't apply here.
• PR description is empty — per CLAUDE.md, PR descriptions should be a short bullet list of what changed.

🤖 Generated with Claude Code