Add batched worker fetch support

[affandar]

Summary

• add provider-native batched worker fetch support
• keep legacy defaults unchanged
• add bounded session-claim behavior for session-aware work

Validation

• cargo check/tests passed in the implementation worktree
• Node binding validation passed against HorizonDB with the patched crates
• HDB benchmark sweep completed with the patched build

Generated by Waldemort.

[pinodeca]

Code review — atv-starter-kit ce-review (Claude Opus 5)

This review was produced by the ce-review skill from atv-starter-kit, running on Claude Opus 5. It dispatches a panel of independent reviewer personas over the diff, then merges and de-duplicates their findings with confidence gating.

Personas that contributed:

Persona	Tier	Why it was selected
correctness-reviewer	always-on	—
testing-reviewer	always-on	—
maintainability-reviewer	always-on	—
project-standards-reviewer	always-on	—
performance-reviewer	conditional	New SQLite queries; PR's stated goal is throughput
api-contract-reviewer	conditional	Adds methods to the public Provider trait and fields to RuntimeOptions
reliability-reviewer	conditional	Touches peek-lock semantics, shutdown, and background tasks
adversarial-reviewer	conditional	+366 lines, data mutations, lock/session semantics

Not selected: learnings-researcher (no docs/solutions/ in this repo) and agent-native-reviewer (library-only surface, no user-facing UI).

84 raw findings were merged and de-duplicated into the 25 below.

---

Is this PR still worth keeping?

Yes. It was filed a while ago, so the first question was whether equivalent functionality had since landed. It has not — main has 16 commits since the merge base (283dcdc, v0.1.28), covering sub-orchestration instance-id collisions (#28), OrchestrationFailed id propagation (#36), replay-safe combinators (#19), new_guid UUID v4 (#26), copyright headers (#22), and the v0.1.29 release. A grep for fetch_work_items, worker_max_inflight, worker_fetch_batch_size, and supports_batched_work_item_fetch across src/** on main returns nothing. The batched-fetch idea is still novel.

Rebased onto origin/main (b6e0255) and force-pushed — bf7b499 → cd750a1. The rebase was clean, no conflicts.

Verification performed

Check	Result
cargo build --all-targets	✅ clean
cargo clippy --all-targets --all-features	✅ 0 warnings from src/ (6 pre-existing unneeded_wildcard_pattern in untouched test files)
cargo nt --no-fail-fast	✅ 1108/1108 passed
./run-tests.sh (two-pass)	✅ both passes green (1108 with features, 1105 without), exit 0
cargo test --doc --all-features	✅ 46 passed

On test_multi_worker_heterogeneous_config

During verification this test failed intermittently, and my first read was that this PR had introduced a duplicate-execution regression. That was wrong, and I want to record the correction explicitly rather than leave a false accusation in the log. Controlled A/B sampling:

	Failures	Rate
This branch	10 / 77	~13%
main	12 / 108	~11%

Two further measurements rule this PR out as a cause:

• SQLITE_LOCKED frequency is unchanged: 60 events (branch) vs 63 (main) over 12 runs each. The new batched fetch transaction does not measurably increase write-lock contention — which also disproves my own finding fix: eliminate race condition in crash_after_dequeue_before_append_completion test #20 below as a practical concern on this workload.
• Ack-failure → test-failure conversion is 1:1 on both: branch 3 ack failures / 3 test failures / 3 co-occurring; main 2 / 2 / 2. The outcome is fully determined by whether an ack deadlocks, independent of branch.

This is the pre-existing bug already filed as #41, with the fix pending in #42 (still open, mergedAt: null; ack_work_item_with_retry is absent from both main and this branch). Mechanism, confirmed from a debug-logged failing run: activity completes → ack_work_item returns SQLite code 6 → worker abandons instead of retrying → item redelivered after 100 ms → handler runs a second time → the test's assert_eq!(a + b, 3) exactly-once assertion sees 4.

No test failure in this suite is attributable to this PR. That said, see finding #2 — the reason nothing broke is largely that the new multi-item path is unreachable at default settings.

---

Verdict: not ready to merge

Two P0s. The first is a behavior change that silently degrades existing users; the second is that the feature is untested and, at defaults, unreachable.

P0

1. worker_max_inflight default contradicts its documentation and throttles existing users — src/runtime/mod.rs:374

The doc comment at src/runtime/mod.rs:190-202 states it "Defaults to worker_concurrency to preserve legacy behavior", but Default hardcodes worker_max_inflight: 2. The semaphore at src/runtime/dispatchers/worker.rs:177 is constructed once, outside the for worker_idx in 0..concurrency loop, so it is shared across every worker slot and becomes a global cap.

Consequence: anyone using RuntimeOptions { worker_concurrency: 8, ..Default::default() } silently drops from 8 concurrent activities to 2. This is a throughput regression shipped by a throughput PR, and it lands on users who did not opt into batching. It also conflicts with the documented 1x1 single-threaded mode for pgrx/embedded use (tests/scenarios/single_thread.rs).

Flagged independently by 8 of 8 reviewers, confidence 1.00 — the only unanimous finding.

Fix: make the field Option<usize> (or use 0 as a sentinel) and resolve it to worker_concurrency at dispatcher start. A literal default cannot express "follow another field".

2. Zero tests for the new batching path — src/runtime/dispatchers/worker.rs:296

process_next_work_batch is a 126-line rewrite of the worker's core dispatch loop, and no test exercises it with more than one item. Because worker_fetch_batch_size defaults to 1, the multi-item branch of both fetch_work_items implementations is unreachable by the entire 1108-test suite. The green suite above therefore says little about this PR.

Minimum bar: a provider-validation case for fetch_work_items (batch of N, partial-failure, session interaction), a dispatcher test with worker_fetch_batch_size > 1, and a 1x1 case in tests/scenarios/single_thread.rs.

P1

3. activity_handles is written but never populated — shutdown drain is dead code — worker.rs:178, 279-283, 331

The vector is created (178), threaded through three call sites (200, 230, 303), drained on shutdown (279-283), and pruned of finished handles (331) — but grep confirms exactly 6 occurrences and no .push anywhere. tokio::spawn at line 376 discards its JoinHandle. Activities are fully detached: shutdown does not wait for them, and the drain loop always sees an empty vector. Flagged by 6 reviewers.

4. Batching is inert by default because of the worker_node_id gate — worker.rs:319

session_batch_allowed = session_config.is_none() || worker_node_id.is_some(), and worker_node_id defaults to None (src/runtime/mod.rs:389). Any runtime using sessions without an explicit node id gets fetch_limit = 1. The gate is unexplained; if it guards a real hazard it needs a comment and a test, and if it does not it should go.

5. InstrumentedProvider does not forward the new capability methods — src/providers/instrumented.rs:79

It overrides fetch_work_item (line 193) but not supports_batched_work_item_fetch or fetch_work_items, so it inherits the trait defaults — including supports_batched_work_item_fetch() -> false. Since src/runtime/mod.rs:990-998 wraps the provider in InstrumentedProvider whenever observability is enabled, turning on observability silently disables batching. A performance feature that vanishes when you instrument to measure it is a particularly unhelpful failure mode.

6. Provider-trait change checklist not followed

.github/copilot-instructions.md requires four steps when the Provider trait changes. None are done: no validation tests in src/provider_validation/, no cases in tests/sqlite_provider_validations.rs, no update to docs/provider-implementation-guide.md, no update to docs/provider-testing-guide.md. Provider is the crate's third-party extension point, so the guide is the contract implementers work from.

7. fetch_work_items contract is under-specified — src/providers/mod.rs:1758

The rustdoc does not state whether partial success is allowed, whether returning fewer than max_items implies the queue is drained, whether ordering is guaranteed, what _max_new_sessions does (it is ignored in both implementations), or what an implementer must do with locks already acquired when a later fetch fails. Third-party implementers cannot write a conforming implementation from this.

8. expect() on permit acquisition permanently kills a worker slot — worker.rs:370

permits.pop().expect(...) panics if the invariant is violated. Inside a dispatcher loop this takes the slot down for the process lifetime. Prefer returning Ok(false) and logging.

9. Default fetch_work_items fallback leaks peek-locks and long-polls per iteration — src/providers/mod.rs:1758-1806

The loop calls fetch_work_item(...)? with ?. On failure at iteration k, the k−1 already-locked items are dropped without being abandoned — they stay locked until timeout. It also passes the full poll_timeout on every iteration, so a batch of 10 against an empty queue can block for 10× the intended poll. Every provider that does not override this inherits both.

10. SQLite batch path can claim a session it did not win the row for — src/providers/sqlite.rs:2098-2145

The session UPSERT (2098-2115) executes before the guarded UPDATE worker_queue ... WHERE id = ?3 AND (lock_token IS NULL OR locked_until <= ?4) (2127-2141). When rows_affected() == 0 — another worker won the row — the code rolls back only in-memory state and continues, then still commits the transaction at the end. The session claim persists for an item this worker never got. The single-item path handles the same case correctly by calling tx.rollback() (1946-1955).

P2

11. new_sessions_claimed is incremented for sessions the worker already owned — claimed_session_ids.insert(sid) at sqlite.rs:2123 does not distinguish new claims from refreshes, so the session budget under-counts available capacity.
12. Permit exhaustion is reported as "no work" — worker.rs:326-329 returns Ok(false) when permits.is_empty(), which the caller treats as an idle signal and backs off on. Permits are also held across the provider long-poll, so an idle fetch pins inflight capacity.
13. Session budget is computed from a SessionTracker snapshot (worker.rs:305-307) that can be stale by the time the fetch runs, allowing over-claim.
14. A panic in the detached activity task is swallowed — worker.rs:470's panic!("unexpected WorkItem in Worker dispatcher") used to abort a tracked task; it now dies silently inside a discarded JoinHandle.
15. ~180 lines duplicated between fetch_work_item and fetch_work_items, already diverged in four behaviors (rollback handling, lock-token format, session-claim ordering, attempt-count derivation). The single-item case should delegate to the batch path with max_items = 1.
16. Undocumented let max_items = max_items.min(8); at sqlite.rs:1993 silently caps the batch. The PR description's throughput sweep above batch size 8 is measuring a clamped value.
17. CHANGELOG.md not updated, despite new public RuntimeOptions fields and Provider methods on a published crate.
18. Documentation now inconsistent with behavior: the worker_concurrency doc comment, docs/observability-guide.md:621, and the RuntimeOptions rustdoc example all describe pre-PR semantics.
19. The atomicity guarantee implied by the batch transaction is not enforced or tested.
20. Up to 16 write statements per transaction (per-row UPDATE + per-row session UPSERT) lengthens SQLite's single-writer lock hold. Measured during verification: no observable increase in SQLITE_LOCKED events vs main (60 vs 63 over 12 runs), so this is theoretical on the tested workload — but it scales with max_items and is worth a note.

P3

21. Ok(found > 0) at worker.rs:381 — found is incremented unconditionally, so this is always true when any item was fetched; the safe_auto caller is the only consumer.
22. The single-fetch special case at providers/mod.rs (max_items > 1 && session.is_some()) is unexplained.
23. process_next_work_batch takes 8 arguments and handles 6 concerns; consider a parameter struct.
24. worker_max_inflight: 0 is silently coerced to 1 by .max(1) rather than rejected.
25. Naming collision: max_items and max_new_sessions differ by one character in a hot signature.

Pre-existing (not introduced here, noted for context)

• Worker abandons work items on retryable ack failures, causing redundant activity re-execution #41 / Retry retryable ack failures instead of abandoning the work item #42 — retryable ack_work_item failures cause abandonment and duplicate activity execution. Discussed above; independently reproduced on main at ~11%.
• RuntimeOptions is not #[non_exhaustive] (src/runtime/mod.rs:156), so every added field is a potential breaking change for users constructing it literally. This PR adds two.
• No index on visible_at / locked_until in worker_queue; the new scan_limit = max_items * 4 widens the scan.
• Cargo.toml:16 comments the sqlite feature as "bundled, no system deps", but sqlx's runtime-tokio-native-tls pulls in openssl-sys, making libssl-dev a hard build requirement. Introduced by 52160f3, unrelated to this PR.

---

Suggested fix order

1. Resolve worker_max_inflight from worker_concurrency (P0-1) — this is the one that hurts existing users.
2. Make batching actually reachable: revisit the worker_node_id gate (Runtime should explicitly handle orphan orchestrator queue messages #4) and forward the capability methods in InstrumentedProvider (Runtime should explicitly handle orphan orchestrator queue messages #5).
3. Either push JoinHandles into activity_handles or delete the vector and the shutdown drain (Stale activity cleanup: TTL or background sweep for undeliverable worker queue items #3).
4. Fix SQLite session-claim ordering (duroxide-pg: concurrent worker startup crashes on migration race condition #10) and quota accounting (Rename Windows x64 npm package: duroxide-win32-x64-msvc → duroxide-windows-x64 #11).
5. Replace expect() with abandon-and-log (Add tryDequeueEvent(queueName) to orchestration context #8); fix partial-failure lock leakage and per-iteration long-poll in the default fallback (Expose event queue depth as orchestration-visible state #9).
6. Add the provider-validation module, both guides, and a 1x1 single-thread case (P0-2, fix: allow libsqlite3-sys >=0.28 #6).
7. Re-run the throughput sweep after lifting or documenting the .min(8) clamp (Implement name and identifier size limits (Phase 1a + Phase 2) #16) — the current numbers above batch size 8 are not measuring what they appear to.

The core idea is sound and worth landing. The blocking issues are the default-value bug and the fact that the feature is currently unreachable and untested at defaults.