Documentation
¶
Overview ¶
Package dispatch provides two bounded-concurrency substrate primitives:
- BoundedDispatcher — unordered bounded-concurrency parallel worker pool with optional KV-twofer-aware completion handling (ADR-048).
- KeyedPool — keyed-ORDERED bounded concurrency: same-key work is serialized on one lane, different keys run in parallel (ADR-072).
Pick by whether per-key ordering matters. If any two work items that share a key must be processed in submit order (e.g. graph-ingest, whose arrival-order merge would corrupt on out-of-order same-entity updates), use KeyedPool. Otherwise use BoundedDispatcher.
What this is ¶
BoundedDispatcher is the framework-provided primitive for the "rules sequence, components parallelize" architecture (CLAUDE.md Orchestration Boundaries section). It's what components compose internally when they need to do parallel work over a known list of items — drone fleet weather-monitor walking all active missions, scenario-orchestrator dispatching ready requirements under DAG gating, manufacturing batch's per-widget station processing, semspec scenario-orchestrator's bounded-concurrency dispatch pattern.
BoundedDispatcher is NOT:
- A workflow engine (no DAG semantics, no branching, no lifecycle — see pkg/lifecycle for those)
- A rule-engine extension (rules don't gain new fan-out primitives; for_each at the rule layer is the at-the-rule- layer fan-out)
- A replacement for pkg/worker.Pool — it WRAPS it. New uses prefer BoundedDispatcher (higher-level, KV-twofer aware); existing pkg/worker.Pool consumers stay as-is.
Use when ¶
- A component does internal parallel work over a list of items
- Bounded concurrency is required (caller picks the worker count)
- Optionally: each work item completes async and the dispatcher should fire OnComplete when KV signals match
Do NOT use for ¶
- At-the-rule-layer fan-out (use rule engine's for_each instead)
- Sequential per-item processing (use a plain loop)
- Unbounded concurrency (use a bare goroutine pool)
Example usage (no completion watcher) ¶
d, err := dispatch.New(ctx, dispatch.Config[*Requirement]{
Workers: c.MaxConcurrent,
QueueSize: 256,
Process: c.processRequirement,
}, dispatch.Deps{
NATSClient: c.natsClient,
Logger: c.logger,
})
if err != nil {
return fmt.Errorf("dispatch new: %w", err)
}
defer func() {
if err := d.Stop(context.Background()); err != nil {
c.logger.Warn("dispatch stop", slog.String("error", err.Error()))
}
}()
for _, req := range filterReady(...) {
if err := d.Submit(req); err != nil {
// ErrQueueFull on overflow; caller chooses
// retry/drop/backpressure-propagate.
}
}
Example usage (with KV-twofer completion watcher) ¶
d, err := dispatch.New(ctx, dispatch.Config[*Requirement]{
Workers: c.MaxConcurrent,
QueueSize: 256,
Process: c.processRequirement,
CompletionKVBucket: "EXECUTION_STATES",
CompletionKeyForWorkItem: func(r *Requirement) string {
return "req." + r.Slug + "." + r.ID
},
OnComplete: c.onRequirementComplete,
}, deps)
In the completion-watcher mode, the dispatcher subscribes to the configured KV bucket BEFORE accepting any Submit. Each Submit registers a tracking entry keyed by CompletionKeyForWorkItem(work) before enqueuing to the underlying pool, so a completion-signal write that arrives between Submit and Process can't slip past the watcher.
Shutdown ¶
Stop drains the underlying worker.Pool first (blocks until in-flight Process calls complete), then stops the completion watcher. Callers waiting on KV-triggered OnComplete callbacks should ensure those complete before Stop — typically by canceling the caller's own context and letting Process see the cancellation.
KeyedPool — keyed-ordered concurrency (ADR-072) ¶
KeyedPool partitions work into N lanes by a caller-supplied key: lane = fnv1a(KeyOf(work)) % Lanes. Each lane is one goroutine draining a bounded queue in order, so items sharing a key process serially in submit order while distinct keys run concurrently. Its Process receives the assigned lane index, so a composer can shard per-lane state (e.g. an applied-sequence guard) without locking. A panic in Process is recovered — the lane survives and the optional OnPanic disposition fires (so a composer can Nak the message).
pool, err := dispatch.NewKeyedPool(ctx, dispatch.KeyedConfig[ingestWork]{
Lanes: c.config.IngestLanes,
QueueDepth: 256,
Name: "graph_ingest",
KeyOf: func(w ingestWork) string { return w.entity.ID },
Process: c.processIngest, // (ctx, lane, work) error
OnPanic: func(w ingestWork, _ any) { _ = w.msg.Nak() },
}, dispatch.KeyedDeps{MetricsRegistry: deps.MetricsRegistry, Logger: c.logger})
SubmitBlocking applies backpressure (blocks on a full lane); non-blocking Submit returns ErrLaneFull. On shutdown, cancel the submit context BEFORE Stop so a producer parked in SubmitBlocking unblocks (ADR-072 M3), then Stop drains the lanes.
See also ¶
- pkg/worker — the underlying worker pool (Pool[T])
- pkg/lifecycle — the workflow-shaped substrate that often pairs with BoundedDispatcher (component-internal fan-out over a Lifecycle workflow's instances)
- ADR-048 — the BoundedDispatcher decision
- ADR-072 — the KeyedPool decision (keyed-concurrent entity ingest)
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // ErrInvalidConfig is returned by New when the Config is // internally inconsistent (e.g. Workers <= 0, missing Process, // CompletionKVBucket set without CompletionKeyForWorkItem + // OnComplete). Catches wiring bugs at construction time. ErrInvalidConfig = errors.New("dispatch: invalid config") // ErrQueueFull is the underlying pkg/worker.ErrQueueFull // re-exported so callers can compare via errors.Is without // reaching into pkg/worker directly. ErrQueueFull = worker.ErrQueueFull // ErrStopped is the underlying pkg/worker.ErrPoolStopped // re-exported for the same reason. ErrStopped = worker.ErrPoolStopped // ErrNATSClientRequired is returned by New when // CompletionKVBucket is configured but Deps.NATSClient is nil // (no way to subscribe to the bucket). ErrNATSClientRequired = errors.New("dispatch: Deps.NATSClient is required when CompletionKVBucket is set") // ErrLaneFull is returned by KeyedPool.Submit (the non-blocking // form) when the target lane's bounded queue is at capacity. // SubmitBlocking waits instead of returning this. Distinct from // ErrQueueFull (the BoundedDispatcher's single-queue variant) so a // caller can tell which primitive rejected the work. ErrLaneFull = errors.New("dispatch: lane queue full") )
Package error sentinels. Callers compare with errors.Is.
Functions ¶
This section is empty.
Types ¶
type BoundedDispatcher ¶
type BoundedDispatcher[W any] struct { // contains filtered or unexported fields }
BoundedDispatcher is the framework's bounded-concurrency parallel work primitive. Components compose it for internal fan-out over known work lists; optionally with KV-twofer-aware completion handling when the work involves async writes by other processors.
Generic over work type W — work items can be any Go type. The underlying pkg/worker.Pool[W] enforces the bounded queue + non-blocking submit + graceful shutdown semantics; this wrapper adds the optional KV-twofer-completion overlay.
Concurrency: Submit is safe to call from concurrent goroutines (delegates to pool.Submit which is lock-free via channel send). Stop is idempotent. Stats reads atomic counters via the underlying pool.
func New ¶
New constructs a BoundedDispatcher with the given Config + Deps. Returns ErrInvalidConfig for missing required fields, or ErrNATSClientRequired when CompletionKVBucket is set without a natsclient.
If CompletionKVBucket is configured, New blocks briefly to resolve the bucket via natsclient and start the completion- watcher goroutine. Subsequent Submit calls register their tracking entries against the live watcher.
The pool is started immediately — callers don't need to call Start. Workers are running and ready to receive Submit calls when New returns.
func (*BoundedDispatcher[W]) Stats ¶
func (d *BoundedDispatcher[W]) Stats() worker.PoolStats
Stats returns current dispatcher statistics via the underlying pool. Thread-safe; reads atomic counters.
func (*BoundedDispatcher[W]) Stop ¶
func (d *BoundedDispatcher[W]) Stop(ctx context.Context) error
Stop halts the dispatcher gracefully. Drains the underlying worker pool first (blocks until in-flight Process calls complete or the given context expires), then stops the completion watcher (if configured). Subsequent Submit calls return ErrStopped.
Stop is idempotent. Multiple calls return nil after the first successful Stop.
Callers waiting on KV-triggered OnComplete callbacks should ensure those complete before Stop returns — typically by canceling the caller's own context and letting Process see the cancellation.
func (*BoundedDispatcher[W]) Submit ¶
func (d *BoundedDispatcher[W]) Submit(work W) error
Submit queues a work item. Returns ErrQueueFull if the queue is at capacity. When a completion watcher is active, Submit also registers the work item in the watcher's tracking map BEFORE enqueuing to the pool so a completion signal that arrives between Submit and Process can't slip past the watcher.
Submit is safe to call from concurrent goroutines.
type Config ¶
type Config[W any] struct { // Workers is the bounded concurrency target. The dispatcher // never runs more than Workers Process calls in flight at once. // Must be > 0. Workers int // QueueSize bounds the submit queue. Submit returns // ErrQueueFull when the queue is at capacity. Must be > 0. QueueSize int // Process is called for each submitted work item, in one of // the worker goroutines. Must not be nil. Process func(ctx context.Context, work W) error // CompletionKVBucket — optional. When set, the dispatcher // subscribes to this bucket via natsclient and tracks // completion signals for each submitted work item. The bucket // must exist before New is called; the dispatcher does NOT // create it. Required if CompletionKeyForWorkItem or // OnComplete are set; the three fields are all-or-nothing. CompletionKVBucket string // CompletionKeyForWorkItem — required if CompletionKVBucket is // set. Returns the KV key the dispatcher watches for this work // item's completion. The function must be pure (no per-call // state) because it's called on Submit AND on each KV-watch // update for matching. CompletionKeyForWorkItem func(W) string // OnComplete — required if CompletionKVBucket is set. Called // when CompletionKVBucket has a write at the key returned by // CompletionKeyForWorkItem for some tracked work item. // Invoked from the dispatcher's watcher goroutine; callers // must not block on shared mutexes acquired by Submit. OnComplete func(ctx context.Context, work W) error }
Config parameterizes BoundedDispatcher construction. The zero value is NOT valid — New rejects Workers <= 0, missing Process, and partial completion-watcher configuration.
type Deps ¶
type Deps struct {
NATSClient *natsclient.Client
Logger *slog.Logger
}
Deps carries the framework-provided dependencies the dispatcher needs at runtime. Logger is optional (falls back to slog.Default); NATSClient is required when CompletionKVBucket is configured.
type KeyedConfig ¶
type KeyedConfig[W any] struct { // Lanes is the number of parallel serial lanes. Same-key work is // serialized within a lane; different keys spread across lanes. // Must be > 0. Fixed for the pool's lifetime (lane assignment is a // pure function of the key over this count). Lanes int // QueueDepth bounds each lane's queue. Submit returns ErrLaneFull // when the target lane is at capacity; SubmitBlocking waits. Must // be > 0. QueueDepth int // KeyOf maps a work item to its partition key. Items with equal // keys are processed serially in submit order on one lane. Must be // non-nil and pure (called once per Submit). KeyOf func(W) string // Process handles one work item. It is called by the assigned // lane's single goroutine, with that lane's index — a composer can // use the index to shard per-lane state without locking (the pool // guarantees at most one goroutine per lane). Must be non-nil. Process func(ctx context.Context, lane int, work W) error // OnPanic — optional. When Process panics, the pool recovers it, // keeps the lane goroutine alive, and (if set) calls OnPanic with // the work item and the recovered value so the composer can // dispose of it (e.g. Nak the underlying message). Called from the // lane goroutine. If nil, a recovered panic is logged and dropped. OnPanic func(work W, recovered any) // Name labels this pool's metrics (the `pool` label). Optional; // metrics are only registered when Deps.MetricsRegistry is set. Name string }
KeyedConfig parameterizes KeyedPool construction. The zero value is NOT valid — New rejects Lanes <= 0, QueueDepth <= 0, and a missing KeyOf or Process.
type KeyedDeps ¶
type KeyedDeps struct {
MetricsRegistry *metric.MetricsRegistry
Logger *slog.Logger
}
KeyedDeps carries the framework dependencies. Both are optional: MetricsRegistry nil → metrics are created but not registered (observing is a harmless no-op); Logger nil → slog.Default.
type KeyedPool ¶
type KeyedPool[W any] struct { // contains filtered or unexported fields }
KeyedPool is the framework's keyed-ordered bounded-concurrency primitive: a sibling to BoundedDispatcher (ADR-048) that partitions work into N lanes by a caller-supplied key. All items sharing a key route to the same lane and are processed serially in submit order, while items with different keys spread across lanes and run concurrently. This is distinct from pkg/worker.Pool / BoundedDispatcher, which distribute work across workers with no key affinity or ordering guarantee.
It exists for at-least-once consumers that need per-key ordered concurrency — e.g. graph-ingest, whose arrival-order (full-set- replace) merge requires that all messages for one entity apply in order, but which is otherwise latency-bound on a serial Get+CAS chain (ADR-072, gh#480).
Structure: N separate bounded lane channels, one goroutine per lane. A shared channel + N workers (worker.Pool's shape) cannot preserve per-key order, so this is a distinct type, not a mode on Pool.
Concurrency: Submit / SubmitBlocking are safe to call from concurrent goroutines. Stop is idempotent. The Process function for a given lane is only ever invoked by that lane's single goroutine, so a composer maintaining per-lane state (indexed by the lane number passed into Process) needs no locking on it.
func NewKeyedPool ¶
func NewKeyedPool[W any](ctx context.Context, cfg KeyedConfig[W], deps KeyedDeps) (*KeyedPool[W], error)
NewKeyedPool constructs and starts a KeyedPool. Lane goroutines are running and ready to receive Submit calls when it returns — callers don't call a separate Start.
The ctx is the pool's run context: it is passed to Process, and its cancellation aborts all lanes immediately (in-flight Process calls see the cancellation via their ctx; buffered items are NOT drained). For a graceful drain that DOES finish buffered work, call Stop.
Returns ErrInvalidConfig for a missing or out-of-range required field.
func (*KeyedPool[W]) Stats ¶
func (p *KeyedPool[W]) Stats() KeyedStats
Stats returns current pool statistics. Thread-safe (atomic reads).
func (*KeyedPool[W]) Stop ¶
Stop halts the pool gracefully: it stops accepting new work, drains each lane's buffered items to completion, and returns when all lanes are idle or the given ctx expires (returns ctx.Err()). Idempotent — subsequent calls return nil.
Stop does NOT cancel the pool's run context, so in-flight and buffered Process calls run to completion. If the ctx expires first, the lanes keep draining in the background; the caller has simply stopped waiting.
func (*KeyedPool[W]) Submit ¶
Submit routes work to its lane without blocking. Returns ErrLaneFull if the target lane's queue is at capacity, or ErrStopped after Stop. Safe for concurrent use.
func (*KeyedPool[W]) SubmitBlocking ¶
SubmitBlocking routes work to its lane, blocking until the lane has capacity, the ctx is cancelled (returns ctx.Err()), or the pool is stopped (returns ErrStopped). This is the backpressure form — a full lane blocks the caller rather than dropping.
Composer note (ADR-072 M3): on shutdown, cancel the ctx passed here BEFORE calling Stop, so a caller parked on a full lane unblocks and can dispose of its message; otherwise a synchronous producer (e.g. a NATS consume callback) can wedge teardown. Stop closing drainCh is a backstop that also unblocks this call.
Source Files