feat: durable enqueue outbox dispatch — publish-in-commit via attached bus

[patrickleet]

Stacked on #53 (base = codex/hops-service-create-microsvc-scaffold); rebases onto main once #53 merges. Implements [[specs/durable-enqueue-outbox-dispatch]].

What this does

Connects the producing side of the bus to a service so one bus config drives both consume and produce, with publish-in-commit as the primary path:

let service = distributed::register_handlers!(
    Service::with_repo(repo.queued().aggregate::<Todo>()),
    command handlers::todo_create,
    event   handlers::todo_projection,
);

service
    .with_bus(InMemoryBus::new())   // transport config on the service
    .run(RunOptions::idempotent())  // auto listen (commands) + subscribe (events)
    .await?;

// In a handler:
ctx.commit_outbox(&mut todo, message).await?;
//  bus attached → claim the outbox row in the commit transaction (InFlight,
//                 short lease, attempts=1) → publish immediately via the bus →
//                 complete; publish failure releases for the poller and never
//                 rolls back the committed aggregate.
//  no bus       → commit pending; the polling worker publishes.

The claim-in-transaction model means the after-commit publish needs no separate claim and cannot race the poller; the lease is the crash-handoff to the worker.

Commits

1. CommitReceipt from OutboxCommit::commit — the inserted-ids seam (source-compatible across all 52 callers).
2. BusPublisher — Bus → AsyncMessagePublisher adapter routing Command/Event to send_message/publish_message (the missing piece that lets the dispatcher publish through any *Bus).
3. HasOutboxStore — capability resolving the outbox store through the AggregateRepository → QueuedRepository → leaf wrapper chain.
4. Service::with_bus + Microservice + dispatcher() — transport config on the service (produce side).
5. OutboxCommit::commit_claimed — claim-in-transaction (row inserts InFlight/leased, not poller-claimable).
6. Context::commit_outbox + DynPublisher — publish-in-commit end to end. DynPublisher is an object-safe (boxed-future) shim around AsyncMessagePublisher so Service<D> keeps its type rather than becoming Service<D, P>.
7. Microservice::run — derives listen/subscribe from the registered handlers, executor-agnostic poll-join (no tokio pulled into core). Clone derived for RunOptions.

Tests

cargo test --lib → 236 passed, 0 failed. All integration test targets compile. New coverage: commit_receipt, bus_publisher routing, has_outbox_store wrapper resolution, commit_claimed (in-flight/leased/not-poller-claimable), commit_outbox immediate publish e2e, run() consuming a queued command.

Not in this PR (follow-ups, tracked in tasks/durable-enqueue-outbox-dispatch-impl)

• Docs: README "durable enqueue" framing + closing the Quick Start produce loop.
• Continuous backstop poll loop inside run(): needs an async timer (tokio), which isn't in core. dispatcher() is exposed to drive it; the immediate path covers the happy case, so the loop is the crash backstop only. Open: gate behind a runtime feature vs. a separate worker entrypoint.
• commit_outbox covers AggregateRepository, not yet SnapshotAggregateRepository.
• SQL-backed (postgres/sqlite) end-to-end test of commit_outbox (trait impls compile under those features; only HashMap exercised so far).
• Immediate-publish lease customization (currently defaults; the Microservice setters configure the poller).

🤖 Generated with Claude Code

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 27abf415-5ddb-45c2-bb66-2ca706f32f45

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/durable-enqueue-outbox-dispatch

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}