command

type AckReason ¶

type AckReason uint8

AckReason classifies how a command ended. Every reason maps to a terminal status.

ref: JetStream Ack/Nak/Term disposition — only terminal outcomes.

const (
	AckSuccess  AckReason = iota + 1 // device executed successfully → StatusSucceeded
	AckFailed                        // permanent failure → StatusFailed
	AckTimeout                       // deadline elapsed → StatusExpired (used by Sweeper)
	AckRejected                      // device/system rejected the command → StatusCanceled
)

func (r AckReason) String() string

func (r AckReason) TargetStatus() Status

func (r AckReason) Valid() bool

type ActiveScanner ¶

type ActiveScanner interface {
	// ScanActive returns non-terminal entries matching filter, ordered by
	// CreatedAt ascending (FIFO). Implementations MAY impose an adapter-level
	// result cap but MUST document it.
	ScanActive(ctx context.Context, filter ScanFilter) ([]Entry, error)
	// GetCommand returns a single command by ID. Returns ErrCommandNotFound when not found.
	GetCommand(ctx context.Context, id string) (*Entry, error)
}

ActiveScanner queries all non-terminal command entries matching a filter. This is the role-based scan port used by Sweeper (timeout detection) and operational views (list pending/in-flight commands). It is deliberately distinct from Queue.Dequeue (the claim-with-lease primary consumer path): scanners do not advance state or issue leases, and scanners are allowed to see Sent/Delivered entries that Dequeue must hide.

ref: Temporal HistoryService active-timer scan + JetStream ConsumerInfo — separate from the primary Fetch/dispatch path; read-only lens over non-terminal entries.

type AuthzFunc ¶

type AuthzFunc func(ctx context.Context) error

AuthzFunc is a caller-supplied permission hook invoked at the Enqueue boundary. Pass nil to skip (demo/test mode). See docs for T3 DEVICE-ENQUEUE-RBAC.

type EnqueueOptions ¶

type EnqueueOptions struct {
	// IdempotencyKey enforces STATE-AWARE ACTIVE uniqueness: at most one command
	// with this key may be NON-terminal (Pending/Sent/Delivered) at a time. A
	// re-enqueue under the same key is coalesced (no-op, nil error) while a holder
	// is non-terminal, and is ADMITTED once the prior holder reaches a terminal
	// status (Succeeded/Failed/Expired/Canceled) — so a later retry of the same
	// logical command enqueues a fresh entry. Empty = no uniqueness.
	//
	// This is the River UniqueOpts.ByState / Temporal single-open-execution model:
	// the queue (the owner of active-command state) is the correctness authority
	// for "at most one in-flight per identity"; producers MUST NOT reconstruct it
	// from side state. Uniqueness is keyed on Status, so terminal transitions
	// (Ack/Cancel/Sweeper-AckTimeout) release the key automatically — there is no
	// separate release call to forget. PG enforces it with a partial unique index
	// `WHERE … AND status IN (Pending,Sent,Delivered)`; the in-mem store derives it
	// from Status. Both are pinned by the commandtest conformance suite
	// (Enqueue/ActiveKeyBlocksAcrossNonTerminal + Enqueue/KeyReleasedOn*).
	//
	// CAUTION: a command that never reaches terminal would hold its key forever and
	// block all retries — so a producer requesting active-uniqueness MUST also give
	// the command a terminal-guaranteeing deadline (see runtime/command
	// WithActiveUniqueness, which couples the two so the unsafe combination is
	// inexpressible).
	IdempotencyKey string
	// Authz is invoked before any write; return non-nil to reject. Use nil to skip.
	Authz AuthzFunc
}

EnqueueOptions configures a single Enqueue call. Lease duration is not set at enqueue time — it is determined by the leaseDuration parameter of Queue.Dequeue. DefaultLeaseDuration is the recommended default for Dequeue callers.

type Entry ¶

type Entry struct {
	ID          string
	DeviceID    string
	CommandType string // e.g., "reboot", "cert-renew", "config-push"
	Payload     []byte
	Status      Status
	Metadata    map[string]string // extensible key-value pairs

	Timeouts Timeouts

	// Attempt tracks the current delivery attempt (0 = first attempt).
	// Design note: there is no explicit "Retrying" status. Retry is modeled
	// as the same Pending→Sent arc with an incremented Attempt counter. This
	// avoids a combinatorial explosion of states (Retrying×{Sent,Delivered})
	// and keeps the transition table compact. Adapters inspect Attempt to
	// decide whether to retry or transition to Failed.
	//
	// Use ResetForRetry to move a command back to Pending for retry —
	// do NOT directly mutate Status/SentAt fields.
	Attempt     int
	CreatedAt   time.Time
	SentAt      *time.Time // set when Status transitions to Sent
	DeliveredAt *time.Time // set when device ACKs receipt
	CompletedAt *time.Time // set when terminal state reached
}

Entry represents a single L4 command enqueued for device execution.

The lifecycle: Pending → Sent → Delivered → Succeeded/Failed/Expired/Canceled. Queue implementations persist and advance the status through Queue methods.

func NewEntry(id, deviceID, commandType string, payload []byte, timeouts Timeouts, now time.Time) Entry

func (e *Entry) DeadlineFor(phase TimeoutPhase) time.Time

func (e *Entry) LogValue() slog.Value

func (e *Entry) ValidateNew() error

type ExpiryTransition ¶

type ExpiryTransition struct {
	CommandID string
	From      Status
	To        Status // typically StatusExpired
	Reason    string // e.g. "phase_overall_deadline"
}

ExpiryTransition describes a single status change recommended by SweepOnce.

func SweepOnce(entries []Entry, now time.Time) []ExpiryTransition

type Queue ¶

type Queue interface {
	Enqueue(ctx context.Context, entry Entry, opts EnqueueOptions) error
	Dequeue(ctx context.Context, targetID string, n int, leaseDuration time.Duration) ([]Entry, error)
	Report(ctx context.Context, commandID string, now time.Time) error
	Ack(ctx context.Context, commandID string, reason AckReason, now time.Time) error
	ExtendLease(ctx context.Context, commandID string, extension time.Duration, now time.Time) error
	Cancel(ctx context.Context, commandID string, now time.Time) error
}

Queue is the kernel-level L4 command queue facade. Implementations live in adapters/postgres or in-memory (commandtest package).

Event-driven state machine: each state transition is triggered by a distinct Queue method call, not chained inside Ack:

Enqueue   → StatusPending (created)
Dequeue   → StatusPending → StatusSent        (claim + lease, single atomic step)
Report    → StatusSent    → StatusDelivered   (device acknowledged receipt)
Ack       → any non-terminal → terminal       (single atomic step)
ExtendLease → renew existing lease
Cancel    → any non-terminal → StatusCanceled  (operator action)

Ack is single-step atomic: it does NOT chain Pending→Sent→Delivered→terminal. Callers skipping Report (e.g., acking directly from StatusSent) produce a StatusSent → StatusSucceeded transition with DeliveredAt left nil, indicating the device never reported intermediate delivery.

ref: Temporal RecordActivityTaskHeartbeat + RespondActivityTaskCompleted — distinct RPCs for in-progress signal vs terminal; state transitions are recorded at the event, not batched at ack time. ref: JetStream InProgress vs Ack/Nak/Term — distinct client methods for in-flight continuation vs disposition.

type QueueRegistrar ¶

type QueueRegistrar interface {
	RegisterCommandQueue(q Queue)
}

QueueRegistrar is an optional interface a Cell may implement to receive its command.Queue dependency from the runtime; it is the optional injection-direction interface for command queue handles; the cell-side equivalent is collapsed into cell.Registrar. Runtimes SHOULD probe this via type assertion during Cell.Init.

The concrete Queue instance is owned by the composition root (or a runtime/command discovery phase), not the Cell.

The runtime consumer lives in runtime/command.DiscoverQueueRegistrars; see that package for wiring examples. The Sweeper is driven as a reconcile.Reconciler via a kernel/reconcile.Loop.

type ScanFilter ¶

type ScanFilter struct {
	// DeviceID, if non-empty, restricts results to one device. Empty = all devices.
	DeviceID string
	// Statuses, if non-empty, restricts results to the listed statuses.
	// Empty slice = all non-terminal statuses (Pending / Sent / Delivered).
	// Terminal statuses in Statuses are silently ignored — scanners return
	// only non-terminal entries.
	Statuses []Status
}

ScanFilter narrows an ActiveScanner.ScanActive call. Empty fields mean "no filter" (all non-terminal entries across all devices).

type Status ¶

type Status uint8

Status represents the lifecycle state of an L4 command entry. L4 commands target devices with long-latency round-trips (IoT command ack, certificate renewal, etc.).

ref: ThingsBoard RPC status model (QUEUED→SENT→DELIVERED→SUCCESSFUL); ref: Temporal Nexus operations (SCHEDULED→STARTED→terminal, three-tier timeouts).

const (
	// StatusPending: enqueued, awaiting send to device transport.
	//
	// IMPORTANT: iota+1 ensures the zero value (0) is NOT a valid status.
	// A forgotten/uninitialised Entry.Status will not silently appear valid.
	StatusPending   Status = iota + 1 // = 1
	StatusSent                        // transmitted to device transport
	StatusDelivered                   // device ACK'd receipt (not execution)
	StatusSucceeded                   // device confirmed successful execution
	StatusFailed                      // permanent failure (device error or retries exhausted)
	StatusExpired                     // deadline elapsed before completion
	StatusCanceled                    // explicitly canceled by operator/system
)

func (s Status) CanTransitionTo(target Status) bool

func (s Status) IsTerminal() bool

func (s Status) String() string

func (s Status) Valid() bool

func (s Status) ValidTransitions() []Status

type Sweeper ¶

type Sweeper struct {
	// contains filtered or unexported fields
}

Sweeper is a kernel-level command expiry actor. It implements reconcile.Reconciler: each Reconcile call scans non-terminal commands (via ActiveScanner) and terminates expired entries (via Queue.Ack(AckTimeout)). It is the reconcile runtime's first real consumer — a reconcile.Loop drives it on a TickerTrigger cadence (the per-tick control shell that used to live in runtime/command has been deleted).

Time source: the Sweeper holds the cell's business clock (clk). Reconcile — whose signature carries no "now" — sources the sweep time from clk.Now(), following the reconcile convention that a Reconciler owns its own domain clock. Control-plane scheduling (the tick cadence, startup probe, requeue timers) is NOT the Sweeper's concern: it is owned by reconcile.Loop's sealed real-only clock and the injected TickerTrigger clock. The lower-level SweepTick(ctx, now) primitive still takes an explicit "now", keeping business-plane sweep logic deterministically testable without clock injection.

All non-built fields are unexported — callers cannot populate them via `&command.Sweeper{scanner: ...}` literals. The remaining attack surface is the bare zero-value literal `&command.Sweeper{}`, which produces an instance with nil scanner / queue / clk and would panic on the first sweep. The unexported `built` sentinel + head fail-closed (SweepTick and Reconcile) turns that panic into a clean error: only NewSweeper sets `built=true`, so any literal-zero Sweeper short-circuits before dereferencing scanner / clk.

Filter narrows the scan; zero value (default) means "all devices, all non-terminal statuses". Adapters decide whether ScanFilter is honored efficiently (e.g., indexed by device_id) or scanned in memory.

ref: Temporal HistoryService timer scan loop — role-based periodic scan over active timers; disposition (expire vs retry) is a separate decision. ref: kubernetes-sigs/controller-runtime pkg/reconcile/reconcile.go (Reconciler). ref: kernel/outbox.ConsumerBase.built — same sentinel pattern.

func NewSweeper(scanner ActiveScanner, queue Queue, clk clock.Clock, opts ...SweeperOption) (*Sweeper, error)

sweeper, err := command.NewSweeper(scanner, queue, clk,
    command.WithSweeperFilter(command.ScanFilter{DeviceID: "dev-1"}),
)
if err != nil {
    return fmt.Errorf("sweeper: %w", err)
}

func (s *Sweeper) Reconcile(ctx context.Context, _ reconcile.Request) (reconcile.Result, error)

func (s *Sweeper) SweepTick(ctx context.Context, now time.Time) error

func (s *Sweeper) Validate() error

type SweeperOption ¶

type SweeperOption func(*Sweeper)

SweeperOption configures optional Sweeper fields. Pass into NewSweeper as variadic args. The required positional dependencies (scanner, queue) are NOT exposed as options so the type system enforces their presence.

func WithSweeperFilter(f ScanFilter) SweeperOption

type TimeoutPhase ¶

type TimeoutPhase uint8

TimeoutPhase identifies which phase of the command lifecycle a deadline applies to.

const (
	// PhaseScheduleToSend: max duration from creation (Pending) to Sent.
	PhaseScheduleToSend TimeoutPhase = iota + 1
	// PhaseSendToComplete: max duration from Sent to a terminal state.
	PhaseSendToComplete
	// PhaseOverall: absolute max duration from creation to terminal state.
	PhaseOverall
)

type Timeouts ¶

type Timeouts struct {
	// ScheduleToSend: max wait from creation to device transport delivery.
	ScheduleToSend time.Duration
	// SendToComplete: max wait from Sent to terminal state (Succeeded/Failed).
	SendToComplete time.Duration
	// OverallDeadline: absolute max from creation to any terminal state.
	OverallDeadline time.Duration
}

Timeouts configures per-phase deadlines for an L4 command. Zero duration means no timeout for that phase.

Adapter responsibility: the kernel defines timeout configuration and pure deadline calculation (via Entry.DeadlineFor). The adapter MUST run a periodic sweeper (e.g., every 30s–60s) that queries non-terminal commands, computes DeadlineFor for each active phase, and calls Queue.Ack(..., AckTimeout, now) when now exceeds the deadline.

ref: Temporal Nexus operations — ScheduleToCloseTimeout, ScheduleToStartTimeout, StartToCloseTimeout. GoCell simplifies to three tiers.

type Writer ¶

type Writer interface {
	// WriteCommand persists a command entry atomically with business state.
	// Consistency: L4 (DeviceLatent).
	WriteCommand(ctx context.Context, entry Entry) error
}

Writer persists L4 command entries within a transaction. Primarily used by adapter internals and test fixtures to seed entries bypassing Enqueue validation; service/application code MUST use Queue.Enqueue instead.