Documentation
¶
Overview ¶
Package command wires kernel command workers (queue discovery + dispatch registry) into the runtime. The periodic command-expiry sweep is no longer a runtime control shell here — the kernel Sweeper implements reconcile.Reconciler and is driven by a kernel/reconcile.Loop at the cell (see examples/iotdevice/cells/devicecell).
Index ¶
- Constants
- func ClaimKeyFromEntry(e kout.Entry) (key string, ok bool)
- func DiscoverQueueRegistrars(cells []cell.Cell, q kcommand.Queue) (int, error)
- func DiscoverQueueRegistrarsInAssembly(asm *assembly.CoreAssembly, q kcommand.Queue) (int, error)
- func EmitAsync[T any](ctx context.Context, clk clock.Clock, emitter kout.Emitter, ...) error
- type AsyncDispatchFunc
- type CommandID
- type Registry
Constants ¶
const CommandIDMetadataKey = "gocell.command.idempotency_id"
CommandIDMetadataKey is the outbox.Entry metadata key under which an async command's per-instance idempotency identity (commandID) is carried. It is the single source shared by the producer funnel (EmitAsync writes it) and the relay funnel (ClaimKeyFromEntry reads it back).
The "gocell.command." prefix deliberately avoids the kernel-reserved metadata namespace (kout.ReservedMetadataKeys holds observability/principal keys like trace_id / actor_id); a reserved key would be rejected by Entry.Validate at construction. commandID is NOT a principal/observability identity — it is the command instance dedup token — so it lives in the producer-owned business metadata map.
Variables ¶
This section is empty.
Functions ¶
func ClaimKeyFromEntry ¶
ClaimKeyFromEntry extracts an async command entry's idempotency identity and derives the flattened Claimer key. ok=false means the entry is missing its identity slot (empty subject or empty commandID) — the relay MUST fail-closed and dead-letter such an entry rather than silently skipping deduplication.
The tenant dimension is taken from the entry's principal envelope (injected by NewEntry from ctx at emit time, or _notenant when empty); subject from AggregateID; commandID from the CommandIDMetadataKey metadata entry. These flow through the sealed DeriveCommandKey + Flat funnel so the resulting key is node-agnostic and stable.
func DiscoverQueueRegistrars ¶
DiscoverQueueRegistrars injects q into every cell that implements kernel/command.QueueRegistrar. It mirrors the optional-capability discovery style now collapsed into cell.Registrar while keeping command queue ownership in the composition root.
func DiscoverQueueRegistrarsInAssembly ¶
DiscoverQueueRegistrarsInAssembly injects q into QueueRegistrar cells in assembly registration order.
func EmitAsync ¶
func EmitAsync[T any]( ctx context.Context, clk clock.Clock, emitter kout.Emitter, dispatchID CommandID, subject, commandID string, payload T, ) error
EmitAsync is the sole sanctioned emit exit for an async command entry. subject and commandID are required positional params — it is compile-time inexpressible to "emit an async command but omit the idempotency identity slot".
dispatchID becomes the entry's RoutingTopic (the command topic); the relay routes a command entry to its in-process generated DispatchAsync by matching the RoutingTopic against the composition-root dispatcher-map. subject is written to AggregateID and commandID to Metadata[CommandIDMetadataKey]; the relay reads both back via ClaimKeyFromEntry to derive the Claimer dedup key (tenant is taken from the entry's principal envelope, injected by NewEntry from ctx).
WARNING: subject and commandID are adjacent same-typed (string) positional params — transposing them is NOT a compile error and silently mis-partitions the dedup slot across subjects. subject = the dedup aggregate (e.g. deviceID); commandID = the per-instance identity (e.g. the source event id).
Error contract mirrors kout.Emit: every failure is wrapped with context and returned (never swallowed). Callers MUST return or log the error.
Types ¶
type AsyncDispatchFunc ¶
AsyncDispatchFunc is the signature of a generated command DispatchAsync free function: it decodes a claimed command outbox.Entry's payload into the typed *Request, looks the handler up in reg, and invokes it — the async sibling of the synchronous generated Dispatch (ADR docs/architecture/202606040550-1044-adr-command-bus-dispatch-funnel.md §5 ④).
The outbox relay routes a command entry (an entry whose RoutingTopic equals a command DispatchID) to the registered AsyncDispatchFunc via Relay.WithCommandDispatch, instead of publishing it to the broker.
Every generated command package compile-asserts `var _ command.AsyncDispatchFunc = DispatchAsync`, so the generated DispatchAsync is the only value with this exact identity. archtest COMMAND-ASYNC-DISPATCH-CALLER-01 locks the values bound into a WithCommandDispatch map to generated DispatchAsync symbols, the async-path downstream half of the dispatch funnel double-lock (the synchronous half is COMMAND-DISPATCH-REGISTER-CALLER-01). reg is carried as a parameter (rather than captured in a composition-root closure) so the map values stay direct generated-symbol references the archtest can resolve via go/types.
type CommandID ¶
CommandID is the registry key — the command contract id (e.g. "command.devicecommand.enqueue.v1").
It is a type alias (not a defined type) for idutil.SafeID so that:
- idutil.SafeID.Validate() is available for id validation without conversion.
- Generated code can declare `const DispatchID idutil.SafeID = "..."` and pass it directly to RegisterHandler/LookupHandler without a conversion expression (优雅简洁 — the SafeID.Validate guard is the real key guard).
Because it is an alias, CommandID and idutil.SafeID are interchangeable at the type level — the type system does NOT distinguish "a command registry key" from any other SafeID. The funnel boundary (only generated code registers/dispatches) is enforced by archtest COMMAND-DISPATCH-REGISTER-CALLER-01, not by the type.
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry is the sealed in-process command dispatcher core. It maps each CommandID to exactly one handler (boxed as any) so that the generated command_gen.go Dispatch functions can perform type-safe lookups at runtime.
Sealed-construction rationale: mu and handlers are unexported, so business packages holding a *Registry can mutate it only via RegisterHandler and read it only via LookupHandler. Both methods are intended to be called exclusively from generated command code; the caller-allowlist archtest COMMAND-DISPATCH-REGISTER-CALLER-01 enforces that at the call-site level.
The registry stores handlers boxed as any because Go cannot express a heterogeneous map keyed by string with value type "one of many generated Handler interfaces" in the type system — this is the same posture used by the saga *Definition registry and Watermill's command-processor map. The generated Dispatch function performs the type-assert immediately after LookupHandler, keeping the unsafe boundary as narrow as possible.
Thread safety: all operations are protected by an internal RWMutex. RegisterHandler uses a full write-lock; LookupHandler uses a read-lock.
func NewRegistry ¶
func NewRegistry() *Registry
NewRegistry returns a new, empty Registry ready for use.
func (*Registry) LookupHandler ¶
LookupHandler returns the boxed handler for id. If id is not registered, it returns (nil, false). Thread-safe (acquires mu.RLock).
func (*Registry) RegisterHandler ¶
RegisterHandler stores handler under id. Returns an error for any of the following conditions:
- id is empty or contains unsafe characters (idutil.SafeID.Validate fails): KindInvalid / ErrValidationFailed.
- handler is nil or typed-nil (validation.IsNilInterface): KindInvalid / ErrValidationFailed.
- id is already registered (one-to-one command→handler mapping, mirrors Watermill DuplicateCommandHandlerError and the saga registry duplicate check): KindConflict / ErrConflict.
Thread-safe (acquires mu.Lock).
Source Files