Documentation
¶
Overview ¶
Package commandtest provides shared conformance assertions for command.Queue + command.ActiveScanner implementations. Both the in-memory InMemQueue and the PostgreSQL command_queue adapter (adapters/postgres) enroll against the same suite so behavior stays in lock-step.
ref: runtime/audit/ledger/storetest (kernel-side conformance shape) ref: cells/accesscore/internal/ports/conformance (Factory + Features pattern)
Package commandtest provides in-memory implementations of the command package interfaces for use in unit tests and examples.
NOTE: Not suitable for production deployments. Replace with a persistent adapter (e.g., adapters/postgres command store) for durable mode. This package is importable from non-test code intentionally — the guard against production misuse lives at the Cell wiring layer.
Index ¶
- func RunQueueConformance(t *testing.T, factory QueueFactory, features Features)
- type Features
- type InMemQueue
- func (q *InMemQueue) Ack(_ context.Context, commandID string, reason command.AckReason, now time.Time) error
- func (q *InMemQueue) Cancel(_ context.Context, commandID string, now time.Time) error
- func (q *InMemQueue) Dequeue(_ context.Context, targetID string, n int, leaseDuration time.Duration) ([]command.Entry, error)
- func (q *InMemQueue) Enqueue(ctx context.Context, entry command.Entry, opts command.EnqueueOptions) error
- func (q *InMemQueue) ExtendLease(_ context.Context, commandID string, extension time.Duration, now time.Time) error
- func (q *InMemQueue) GetCommand(_ context.Context, id string) (*command.Entry, error)
- func (q *InMemQueue) RepoReady(_ context.Context) error
- func (q *InMemQueue) Report(_ context.Context, commandID string, now time.Time) error
- func (q *InMemQueue) ScanActive(_ context.Context, filter command.ScanFilter) ([]command.Entry, error)
- func (q *InMemQueue) WriteCommand(_ context.Context, entry command.Entry) error
- type QueueFactory
- type TxRunner
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RunQueueConformance ¶
func RunQueueConformance(t *testing.T, factory QueueFactory, features Features)
RunQueueConformance drives the shared state-machine assertions against the implementation produced by factory. Test names are scoped t.Run sub-tests so failures point at a single transition rather than a soup of assertions.
Types ¶
type Features ¶
type Features struct {
// RequiresAmbientTx indicates writes must run inside a txRunner.RunInTx
// scope (true for PG; false for mem). When true, the suite wraps each
// mutating call in RunInTx; when false, it calls the Queue methods directly.
RequiresAmbientTx bool
// SupportsLeaseRenewal indicates ExtendLease is honored (true for both
// today; the flag exists so future read-only mirrors can opt out without
// silently passing).
SupportsLeaseRenewal bool
}
Features captures behavioral differences between in-memory and durable implementations that the conformance suite must accommodate without resorting to t.Skip. Features should describe *what* the implementation offers, never *whether to skip*.
type InMemQueue ¶
type InMemQueue struct {
// Now supplies the clock. Defaults to time.Now if nil.
Now func() time.Time
// contains filtered or unexported fields
}
InMemQueue is a process-local, thread-safe implementation of command.Queue, command.ActiveScanner, and command.Writer backed by a map. It is NOT suitable for multi-replica coordination — use for tests and examples.
Implements:
- command.Queue (Enqueue/Dequeue/Report/Ack/ExtendLease/Cancel)
- command.ActiveScanner (ScanActive/GetCommand)
- command.Writer (WriteCommand — for test seeding)
func NewInMemQueue ¶
func NewInMemQueue() *InMemQueue
NewInMemQueue creates a new InMemQueue with the default wall clock.
func (*InMemQueue) Ack ¶
func (q *InMemQueue) Ack(_ context.Context, commandID string, reason command.AckReason, now time.Time) error
Ack finalizes a command atomically in a single transition step. The AckReason maps directly to a terminal status:
- AckSuccess: current → StatusSucceeded
- AckFailed: current → StatusFailed
- AckTimeout: current → StatusExpired (used by Sweeper on deadline elapse)
- AckRejected: current → StatusCanceled
No chaining: Ack does NOT advance through intermediate states. Acking from StatusSent directly to StatusSucceeded leaves DeliveredAt nil, which is the signal that the device skipped the optional Report step. Already-terminal entries are idempotent only when the requested reason maps to the existing terminal status. A different terminal target is rejected.
func (*InMemQueue) Cancel ¶
Cancel transitions a non-terminal command to StatusCanceled (operator action).
func (*InMemQueue) Dequeue ¶
func (q *InMemQueue) Dequeue(_ context.Context, targetID string, n int, leaseDuration time.Duration) ([]command.Entry, error)
Dequeue returns up to n Pending entries for targetID, oldest first. Each returned entry is advanced to StatusSent (incrementing Attempt) and assigned a lease.
func (*InMemQueue) Enqueue ¶
func (q *InMemQueue) Enqueue(ctx context.Context, entry command.Entry, opts command.EnqueueOptions) error
Enqueue stores an entry atomically with optional authz and idempotency. If entry.ID is empty, a random ID is assigned. If opts.Authz is non-nil, it is called before any write; a non-nil return rejects the enqueue. If opts.IdempotencyKey is non-empty and an entry with that key already exists, this is a no-op (idempotent dedup). InMemQueue stores the key as metadata["_idempotency_key"].
func (*InMemQueue) ExtendLease ¶
func (q *InMemQueue) ExtendLease(_ context.Context, commandID string, extension time.Duration, now time.Time) error
ExtendLease renews the lease for a command. Returns ErrNotFound if the command does not exist in the queue. Returns ErrValidationFailed (lease expired) if the command exists but its lease has expired or was never acquired (e.g. Pending, not yet Dequeued).
func (*InMemQueue) GetCommand ¶
GetCommand returns a single command by ID, or nil if not found. Internal metadata keys (prefix "_") are stripped from the returned copy.
func (*InMemQueue) RepoReady ¶
func (q *InMemQueue) RepoReady(_ context.Context) error
RepoReady always returns nil for the in-memory queue (no external dependency). Satisfies healthz.RepoProber so the same queue value can be registered as the "command_queue_ready" readiness probe in demo/test mode.
func (*InMemQueue) Report ¶
Report advances a command from Sent to Delivered, recording that the device has acknowledged receipt and begun execution. No-op (nil) if the command is already Delivered (idempotent); returns ErrValidationFailed for any other non-Sent status.
func (*InMemQueue) ScanActive ¶
func (q *InMemQueue) ScanActive(_ context.Context, filter command.ScanFilter) ([]command.Entry, error)
ScanActive returns all non-terminal entries matching filter, ordered by CreatedAt ascending. filter.DeviceID="" means scan all devices; filter.Statuses=nil means all non-terminal statuses (Pending/Sent/Delivered). Terminal statuses in filter.Statuses are silently ignored.
func (*InMemQueue) WriteCommand ¶
WriteCommand stores an entry directly (bypasses Enqueue validation). Used by adapter-level tests that need to seed pre-existing entries.
type QueueFactory ¶
type QueueFactory func(t *testing.T) ( q command.Queue, scanner command.ActiveScanner, txRunner TxRunner, now func() time.Time, cleanup func(), )
QueueFactory builds a fresh Queue + ActiveScanner pair (typically the same concrete type) plus the matching TxRunner. The returned cleanup MUST be idempotent and tolerate being called even if no resources were acquired.
The returned clock is the time source the implementation must obey when the suite drives state transitions. Implementations that use wall time should return time.Now-equivalent here so the conformance "now" stays consistent.
type TxRunner ¶
TxRunner is the minimal transactional runner contract the conformance suite depends on. It mirrors kernel/persistence.TxRunner structurally — any implementation of that public interface satisfies this one — but is declared locally so kernel/command/commandtest does not import kernel/persistence (KERNEL-INTERNAL-DAG-01: command package would otherwise gain a cross-owner edge to persistence).
The Go idiom is "accept interfaces, return concrete types"; the conformance suite is the accept-interface site, and a one-method local definition keeps the kernel-internal DAG flat without adding a runtime dependency.
Source Files