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 attempts the underlying worker.Pool first, then cancels and joins the completion watcher. It returns pool timeout and caller-context causes rather than reporting an unobserved join as clean. A failed Stop is terminal and must not be retried; a successfully completed repeated Stop remains nil. Callers waiting on KV-triggered OnComplete callbacks should ensure those complete before Stop, typically by canceling the caller's own runtime context and using a separate live bounded shutdown context for Stop.
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). A successfully completed Stop is idempotent. A failed Stop is terminal and must not be retried. 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. The ctx must be non-nil.
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.
Multiple calls return nil after the first successful Stop. A Stop that reports an unobserved pool or watcher join is terminal and must not be retried; the underlying pool has already claimed shutdown.
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. The ctx must be non-nil.
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