refactor: unify queue and workflow runtime

[cmilesio]

What

This PR collapses the overlapping queue and workflow models into one queue-owned runtime. queue.Queue is the canonical application surface, internal/workflow.Engine owns orchestration, drivers own physical delivery, and bus remains as a deprecated compatibility facade.

Fresh review hardening

The three independent review rounds found 3, 6, and 5 primary findings. All 14 are fixed. A final post-fix audit by API, runtime-concurrency, and CI-evidence specialists found five narrower continuity edges; those are fixed too, and the reviewers report no surviving issue.

A separate changed-line audit by NATS, RabbitMQ, Redis, and documentation specialists added the one remaining deterministic production contract: a real NATS subscription failure must leave startup retryable. The other uncovered markers require impossible fixed-shape serialization errors, invalid dependency wiring, global entropy hooks, concrete-client transport fault injection, or artificial process timeouts, so no production-only test seams were added.

Round	Focus	Findings	Result
1	Public API and architecture continuity	3	Fixed and regression-tested
2	Runtime lifecycle, compatibility, and evidence	6	Fixed and regression-tested
3	Observer identity, shutdown, docs, and multi-module CI	5	Fixed and regression-tested

The important before-and-after changes from those reviews are concrete:

Boundary	Before	Now
Nil registration	A nil root or legacy handler could reach the engine registry and panic on delivery	Nil registration is a no-op at root, engine, and deprecated facade boundaries and cannot replace a valid handler
Redis and SQL shutdown	Redis cached one close diagnostic forever; repeated timed SQL shutdown created one waiter goroutine per call	Owned resources close once, cleanup diagnostics are reported once, retries converge, and SQL callers share one drain waiter
NATS subscription startup	A connection could succeed before subscription validation without a real-broker retryability contract	Invalid-subject startup is rejected twice against a real broker without poisoning lifecycle state
Event queue identity	Explicit namespaced queues could split into logical and physical observer buckets, including a whitespace-only edge	Queue, worker, workflow, aggregate, and callback facts report one effective physical queue name
Active delivery metrics	process_failed followed by settlement_failed could decrement Active twice or consume a newer duplicate	One opaque physical settlement identity closes only its own execution; identity-less terminal facts fail closed instead of guessing
Handler panics	A panic could leave Active live and produce no failed-attempt telemetry	process_failed is emitted, correlation state closes, and the original panic value is rethrown so backend semantics stay intact
RabbitMQ retry diagnostics	A replacement publish failure followed by a Nack failure could label the original receipt with the replacement attempt	Republish failure reports attempt N+1 while settlement failure truthfully reports original attempt N
Observer examples	Two manual README examples used the obsolete context-free ObserverFunc signature	Both examples compile against func(context.Context, queue.Event) and a guard pins their exact manual text
Race CI	One root-only race job skipped every nested driver module	Root and all eight driver modules run as visible parallel jobs with GOWORK=off; an always-run fan-in preserves the historical required race check
Generated and integration evidence	Generated assets could drift; skipped or absent repeat scenarios looked green	CI verifies deterministic generation twice, hashes all-backend count evidence, and classifies exact JSON terminal events as pass, fail, skip, or missing

Settlement-aware metrics now have one unambiguous ownership rule:

flowchart TD
  START["process_started with opaque<br/>physical identity"] --> ACTIVE["Open exactly one Active execution"]
  TERMINAL{"Terminal fact"}
  ACTIVE --> TERMINAL
  TERMINAL -->|process success or failure<br/>with same identity| CLOSE["Close that exact execution"]
  TERMINAL -->|settlement failure<br/>with same identity| CLOSE
  TERMINAL -->|identity missing| KEEP["Leave identity-bearing Active unchanged"]
  CLOSE --> COUNTERS["Settlement failure changes neither<br/>Processed nor application Failed"]
  KEEP --> SAFE["Conservative overcount is safer than<br/>undercounting a newer live execution"]

Loading

The nil path changed from forwarding an invalid target:

if handler == nil {
	r.b.Register(jobType, nil)
	return
}

to preserving the existing registry exactly:

if r == nil || handler == nil {
	return
}

The reliability evidence now has an explicit provenance chain:

flowchart LR
  SRC["Integration and root tagged sources"] --> HASH["SHA-256 evidence manifest"]
  FULL["All-backend executed run count"] --> HASH
  HASH --> BADGE["README integration badge"]
  GEN["README, examples, counts,<br/>and benchmark generators"] --> PARITY["Checked-in parity plus<br/>second-run idempotency"]
  JSON["Exact go test JSON event"] --> OUTCOME{"Terminal action"}
  OUTCOME -->|pass| PASS["Executed pass"]
  OUTCOME -->|skip| SKIP["Capability skip"]
  OUTCOME -->|fail| FAIL["Executed failure"]
  OUTCOME -->|absent| MISSING["Missing event and failing job"]

Loading

Architecture before and after

flowchart LR
  subgraph BEFORE["Before: split ownership"]
    direction TB
    BAPP["Application"] --> BQ["queue.Queue facade"]
    BAPP -. legacy calls .-> BB["bus surface"]
    BQ --> BQR["Queue registry, lifecycle,<br/>and observer"]
    BQ --> BWE["bus-owned workflow engine, models,<br/>store, middleware, fake, and observer"]
    BB --> BWE
    BQR --> BRT["Runtime seam"]
    BWE --> BRT
    BRT --> BDR["Queue drivers"]
  end

  subgraph AFTER["After: queue-owned composition"]
    direction TB
    AAPP["Application"] --> AQ["queue.Queue"]
    ALEG["Legacy bus caller"] --> AB["Deprecated bus facade"]
    AB --> AQ
    AQ --> AOWN["Shared registry, middleware,<br/>and lifecycle"]
    AQ --> AWE["Canonical workflow engine"]
    AQ --> AOBS["Unified queue.Event stream"]
    AAPP -. tests .-> AFAKE["Root queue fake"]
    AFAKE --> AWE
    AWE --> AST["Root workflow models and store"]
    AOWN --> ART["Runtime seam"]
    AWE --> ART
    ART --> ADR["Queue drivers"]
  end

Loading

The root queue already exposed chains and batches before this PR. The change is ownership and continuity: those calls now use root queue types and one shared engine instead of crossing into a second public model. The low-level raw runtime route remains for compatibility, but it is not the recommended application path.

Application code before and after

Before, root workflow configuration still depended on public bus models:

store := bus.NewMemoryStore()
observer := bus.ObserverFunc(func(_ context.Context, event bus.Event) {
	_ = event.Kind
})

After, the same application uses one root model. Event.Layer distinguishes queue and workflow facts without creating a second observer pipeline:

store := queue.NewMemoryStore()
observer := queue.ObserverFunc(func(_ context.Context, event queue.Event) {
	if event.Layer == queue.EventLayerWorkflow {
		_ = event.Kind
	}
})

The queue constructor and workflow call stay familiar:

ctx := context.Background()
q, err := queue.NewWorkerpool(
	queue.WithWorkers(4),
	queue.WithStore(store),
	queue.WithObserver(observer),
)
if err != nil {
	return err
}

q.Register("reports:build", func(context.Context, queue.Message) error {
	return nil
})
q.Register("reports:publish", func(context.Context, queue.Message) error {
	return nil
})

_, err = q.Chain(
	queue.NewJob("reports:build").Payload(map[string]string{"id": "rpt_123"}),
	queue.NewJob("reports:publish"),
).OnQueue("critical").Dispatch(ctx)

An option-free bus.New(existingQueue) now returns a compatibility view over this same engine. It shares handlers, middleware, stores, workflow state, observers, and shutdown state. Options that would create conflicting ownership are rejected with bus.ErrQueueOptionsUnsupported and must be applied when the root queue is constructed.

Direct jobs now remain direct

This application call is unchanged:

_, err := q.Dispatch(
	queue.NewJob("emails:send").
		Payload([]byte(`{"id":1}`)).
		OnQueue("critical"),
)

Before, the driver received a workflow protocol message and a JSON wrapper:

physical type: bus:job

{
  "schema_version": 1,
  "dispatch_id": "dsp_0123456789abcdef",
  "kind": "job",
  "job_id": "job_0123456789abcdef",
  "attempt": 0,
  "job": {
    "type": "emails:send",
    "payload": "eyJpZCI6MX0=",
    "options": {
      "Queue": "critical",
      "Delay": 0,
      "Timeout": 0,
      "Retry": 0,
      "Backoff": 0,
      "UniqueFor": 0
    }
  }
}

After, the driver receives the application type and exact application bytes. Correlation travels beside the payload in versioned metadata:

physical type:    emails:send
physical payload: {"id":1}

{
  "schema_version": 1,
  "dispatch_id": "dsp_0123456789abcdef",
  "job_id": "job_0123456789abcdef",
  "queue": "critical"
}

Arbitrary bytes are preserved, and an absent payload stays absent. Chains, batches, callbacks, reserved bus:* types, and the raw runtime route intentionally retain the version-one workflow envelope.

flowchart LR
  A["Queue.Dispatch"] --> B{"Legacy mode or<br/>reserved type?"}
  B -->|Yes| C["Version 1 workflow envelope"]
  B -->|No| D["Application type<br/>and exact payload"]
  D --> E["Versioned correlation metadata<br/>beside the payload"]

  C --> OLD["Old and new workers"]
  E --> NEW["New workers only"]
  OLD --> PIPE["Canonical handler and middleware pipeline"]
  NEW --> PIPE
  PIPE --> SETTLE["Backend settlement boundary"]
  SETTLE --> OBS["Best-effort observer facts"]

Loading

Compatibility is intentionally one-way during rollout. New workers read old and new deliveries; old workers cannot safely consume the new direct format. For non-SQL backends, deploy upgraded workers while producer instances keep queue.WithLegacyDirectEnvelope(), then remove that producer option after old consumers are gone. SQL requires a fleet replacement: quiesce every old SQL worker before any new SQL worker starts, even while legacy producer emission remains enabled.

One runtime behavior map

Concern	Before	Now
Workflow ownership	Root calls crossed into bus models and orchestration	Root queue owns the public model; one internal engine owns orchestration
Compatibility view	A second facade could create separate runtime state	Option-free bus.New(existingQueue) shares the root engine
Observation	Queue and workflow observer models could diverge or label the same work with different queue names	One queue.Event stream uses Event.Layer, one effective physical queue name, and opaque identity for settlement-aware Active gauges
Direct delivery	Application jobs used the bus:job JSON envelope	Drivers receive application type, exact payload, and side metadata
Retry outcome	Application failure and infrastructure failure could both advance attempts	Retryable, permanent, exhausted, and uncommitted outcomes are distinct
Success timing	Success could appear before backend settlement	Success follows the available SQL, SQS, or RabbitMQ settlement boundary
Shutdown	New work and cleanup could race driver closure	Lifecycle leases drain admitted work; timed-out cleanup can be retried
Workflow truth	Duplicate delivery could race contradictory outcomes	Built-in stores enforce first-writer outcomes; provenance-capable paths add receipt-backed recovery
Test doubles	Queue and workflow fakes recorded through separate models	One concurrency-safe fake executes workflows through production rules

SQL receipt-backed handler replay suppression

On the built-in SQL queue plus SQL workflow-store path, the logical workflow outcome and an immutable transition receipt commit in one database transaction. The memory store implements the same contract when complete delivery provenance is supplied, but its receipts are process-local. SQL settlement is separately generation-fenced. If a failed settlement leaves or restores a recoverable row, a later delivery can prove that application work already committed without pretending queue settlement was atomic with it.

flowchart TD
  A["SQL worker claims a delivery<br/>and processing generation"] --> B["Run workflow handler"]
  B --> C["Store transaction commits<br/>workflow outcome plus receipt"]
  C --> D["Attempt successor work<br/>and physical settlement"]
  D --> E{"Settlement commits?"}
  E -->|Yes| F["Publish eligible observer facts<br/>best effort"]
  E -->|No| R{"Delivery remains or<br/>becomes recoverable?"}
  R -->|No| X["No receipt-based replay<br/>on that path"]
  R -->|Yes| G["Redelivery enters recovery"]
  G --> H["Validate receipt version, identity,<br/>fingerprint, and workflow state"]
  H --> I{"Proof valid?"}
  I -->|No| J["Return uncommitted<br/>no ack and no effects"]
  I -->|Yes| K["Suppress duplicate handler execution"]
  K --> L["Restore eligible outcome<br/>and retry settlement"]
  L --> D

Loading

This is not an exactly-once claim. Receipts do not make settlement, successor enqueue, closure callbacks, batch fanout, or observer delivery part of the store transaction. Only the exact physical receipt owner can reconstruct predecessor success facts. Lineage repair is best effort, and chain successor repair remains at least once.

Compatibility and rollout

Area	Impact	Required action
Direct wire format	New workers read both formats; old workers read only legacy envelopes	Deploy upgraded workers while producers keep legacy emission; before rollback, restore legacy emission, drain direct backlog with new workers, and only then return old consumers
SQL queue schema	Adds nullable metadata_json and processing_token	Apply managed schema changes before new workers
SQL worker generations	Old workers cannot honor generation-fenced settlement	Never overlap old and new SQL worker binaries; quiesce old workers before upgrade and new workers before rollback
Workflow SQL schema	Adds bus_workflow_transition_receipts and validates connected MySQL identity columns	Precreate the full table for managed schemas; migrate incompatible MySQL identities while quiescent and construct a fresh store; retain the receipt table on rollback
Redis retry semantics	Retry(0) now means zero instead of Asynq's default 25	Set Retry(25) explicitly if that was the intended policy
Observer source API	Root configuration moves from bus.Observer and bus.Event to queue.Observer and queue.Event	Adapt existing bus observers with queue.ObserverFunc and use Event.Layer; the legacy bus facade still translates events
Settlement-aware metrics	Adds source-compatible busruntime.DeliverySettlementIdentity; identity-less terminal facts cannot close an identity-bearing start	Release and upgrade settlement-aware driver modules with root, and forward the handler context consistently in custom drivers, when exact Active gauges matter
Handler panic telemetry	Panics previously retained backend behavior but could omit a failed-attempt fact	Expect process_failed before the original panic is rethrown; backend panic recovery and retry behavior is unchanged
UniqueFor identity	Versioned logical queue, type, and payload identity replaces volatile workflow envelope IDs and changes SQL uniqueness keys	Upgrade all public workflow producers together; for SQL, quiesce for the largest prior TTL or accept a transient mixed-version duplicate window
Public type identity	Canonical workflow types move from /bus to root /queue	Map reflected paths, gob registrations, DI keys, or type-sensitive persistence where applicable
Chain failure metadata	Built-in FailChain keeps the first terminal cause	Preserve later secondary diagnostics elsewhere instead of using a later failure call to replace the cause
Direct WorkflowStore calls	Built-in stores reject empty workflow or member IDs, empty workflows, and duplicate members; memory-store snapshots no longer alias caller data	Correct ambiguous records before upgrading and stop relying on input or returned-state mutation to change stored state
Additive struct fields	Unkeyed event or database config literals can stop compiling	Move those literals to keyed fields
Go version	No change	None

Schema, durability, and test boundaries

queue.NewSQLStoreWithManagedSchema performs no workflow DDL, so managed deployments must precreate the complete receipt table. Incompatible existing MySQL workflow identities require a quiescent persisted-schema migration and a fresh store; compatible established rows remain readable. Queue database migration remains enabled by default and can be disabled with DisableAutoMigrate. Existing queue rows remain readable after the two nullable queue columns are added. Automatic migration also adds the SQL queue uniqueness-lock expiry index needed for bounded pruning.

• Observer delivery is synchronous, best effort telemetry. Observer panics are isolated. Handler panics emit a failed-attempt fact before being rethrown. Recoverable positive job, chain, and batch facts use deterministic IDs, while dispatch, failure, and other facts use occurrence IDs. A process can settle work and exit before an observer sees the final fact.
• Custom, decorated, and raw-runtime stores keep their contracts but do not gain every private receipt-backed guarantee of the built-in stores.
• Core NATS remains ephemeral. SQS does not yet extend visibility during long handlers. RabbitMQ still requires runtime reconstruction after a closed delivery channel.
• Durable callbacks and continuation intents, a settlement outbox, partial batch fanout recovery, remaining managed-schema cases, and physical database commit/readback ambiguity remain explicit follow-up work in plan.md.

The branch includes literal legacy wire and SQL fixtures, public compile contracts, race coverage, store and lifecycle fault injection, and real SQLite, MySQL, and PostgreSQL recovery scenarios. CI validates all 12 Go modules and runs null, Sync, Workerpool, Redis, NATS, SQS, RabbitMQ, MySQL, PostgreSQL, and SQLite independently. Coverage fans in every buildable module plus all ten backend profiles, rejects incomplete or duplicate inputs before upload, and submits one complete report under the repository's Codecov project and patch policy.

The complete fan-in also drove focused scripted SQL and driver tests for migration races, settlement repair, uniqueness compensation, retry decisions, malformed deliveries, readiness failures, and shutdown boundaries. A follow-up pass proves NATS shutdown waits for accepted handlers, Redis ownership reaches real command semantics, and RabbitMQ immediate retry reaches attempt one, commits, and leaves its broker queue empty. Defensive invariant tests verify ambiguous SQL claim results roll back and absent SQS client results fail closed without presenting those results as production client behavior. Fixed-structure serialization, entropy-source failures, and failures emitted only inside concrete broker clients remain explicit instead of adding production hooks solely for coverage.

Why

The repository had two application models over the same delivery layer. Root queue owned direct jobs and worker lifecycle, while bus separately owned workflow types, stores, middleware, observers, fakes, and orchestration state. The same physical queue could therefore have two registries, two observer pipelines, and different retry or shutdown behavior depending on the facade a caller used.

That duplication also blurred three separate facts: application state committed, the physical delivery settled, and an observer received an event. During retries or finalization failures those facts can diverge. Treating them as one fact risks repeated handlers, contradictory workflow outcomes, premature success events, or lost continuation work.

This PR gives each fact one owner. Durable store state determines workflow truth, the driver boundary determines physical completion, and observation remains telemetry. The result is one ergonomic root API with explicit compatibility and durability boundaries instead of two partially overlapping systems.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 97.03283% with 47 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
driver/natsqueue/queue_nats_impl.go	85.22%	8 Missing and 5 partials ⚠️
driver/rabbitmqqueue/queue_rabbitmq_impl.go	88.23%	8 Missing and 4 partials ⚠️
driver/redisqueue/queue_redis_impl.go	90.58%	4 Missing and 4 partials ⚠️
docs/readme/testcounts/main.go	96.12%	3 Missing and 2 partials ⚠️
driver/natsqueue/worker_nats_impl.go	96.59%	3 Missing and 2 partials ⚠️
driver/rabbitmqqueue/worker_rabbitmq_impl.go	96.69%	3 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!