Documentation
¶
Overview ¶
Package signal is the cross-process control-surface for `awf signal` / `awf pause` / `awf cancel`. Per Phase 3 design decision 3, it is a per-run directory of JSON files; the broker is a thin file-reader/writer struct, NOT a generic IPC layer.
Directory layout (per spec §G):
.awf/runs/<run.id>/control/
signal-<name>-<seq>.json # pending signal payloads
pause.json # operator-requested pause (non-terminal)
cancel.json # operator-requested cancel (terminal)
consumed/
signal-<name>-<seq>.json # atomic-renamed here after consumption
Slice 3.5 is POSIX-only; Windows file-lock semantics need verification (Phase 3 design out-of-scope note).
Index ¶
- Variables
- func ConsumedDir(controlDir string) string
- func ControlDir(stateDir, runID string) string
- type Broker
- func (b *Broker) CheckPauseCancel() (*PauseRequest, *CancelRequest, error)
- func (b *Broker) ClearPauseCancel() error
- func (b *Broker) ControlDir() string
- func (b *Broker) Drain() []Delivery
- func (b *Broker) PollInterval() time.Duration
- func (b *Broker) Receive(ctx context.Context, name string, timeout time.Duration) (Delivery, error)
- func (b *Broker) ReceiveMatching(ctx context.Context, name string, timeout time.Duration, match MatchFunc) (Delivery, error)
- func (b *Broker) WriteCancel(req CancelRequest) error
- func (b *Broker) WritePause(req PauseRequest) error
- func (b *Broker) WriteSignal(name string, payload []byte) (int, error)
- type BrokerOption
- type CancelRequest
- type Delivery
- type MatchFunc
- type PauseRequest
Constants ¶
This section is empty.
Variables ¶
var ErrCancelled = errors.New("signal: run cancelled (terminal)")
ErrCancelled is the sentinel engine.Run returns alongside Outcome("") when control-file polling detects cancel.json. The engine has ALREADY appended the terminal run.cancelled event; the CLI exits cleanly. `awf resume` refuses any log with a run.cancelled event.
var ErrPaused = errors.New("signal: run paused (non-terminal)")
ErrPaused is the sentinel engine.Run returns alongside Outcome("") when control-file polling detects pause.json. The CLI's runAndFinish maps this to a clean exit (rc=0) WITHOUT writing run.finished — the run is non- terminal and resumable.
Functions ¶
func ConsumedDir ¶
ConsumedDir returns the consumed/ subdirectory within a control directory.
func ControlDir ¶
ControlDir returns the control-directory path for the given run within a state dir. Mirrors cli/run.go's `<stateDir>/runs/<id>/log` layout pattern.
Types ¶
type Broker ¶
type Broker struct {
// contains filtered or unexported fields
}
Broker is the per-run control-surface. One Broker per run.id, bound to its .awf/runs/<run.id>/control/ directory. The engine constructs an instance at run start; the CLI subcommands (awf signal/pause/cancel) construct an instance pointing at the same directory to write control files.
Concrete struct (no interface) — see Phase 3 design decision 8 + slice 3.5 design Q2. Substitute via constructor (`signal.NewBroker(t.TempDir(), ...)`), not via interface implementation.
Thread-safety: all methods are safe to call from concurrent goroutines. The underlying filesystem operations are the synchronization primitive (POSIX rename is atomic; O_EXCL+O_CREATE is race-free for new files).
func NewBroker ¶
func NewBroker(controlDir string, opts ...BrokerOption) *Broker
NewBroker constructs a Broker bound to controlDir. The directory is created LAZILY — WriteSignal / WritePause / WriteCancel each MkdirAll the controlDir (and consumedDir, for atomic-rename targets) on first call. NewBroker itself does no I/O and never returns an error.
Default poll interval: 100ms (overridable via WithPollInterval).
func (*Broker) CheckPauseCancel ¶
func (b *Broker) CheckPauseCancel() (*PauseRequest, *CancelRequest, error)
CheckPauseCancel reports whether pause.json and cancel.json exist in controlDir. Either bool may be true; both may be true (cancel-wins resolution is the caller's responsibility — engine/controls.go does that).
Returns (pauseReq, cancelReq, err). pauseReq/cancelReq are non-nil iff their respective files exist. err is non-nil only on read errors (file exists but not readable; malformed JSON is silently treated as an empty body). Reads are capped at maxControlFileBytes (L10 fix).
func (*Broker) ClearPauseCancel ¶
ClearPauseCancel removes pause.json and cancel.json. Idempotent (missing files are not errors). Called by cli/resume.go before re-entering the engine — pause is non-terminal but stale pause.json would re-pause on the next commit; resume must clear it to make forward progress.
func (*Broker) ControlDir ¶
ControlDir returns the configured control directory. For tests + diagnostics.
func (*Broker) Drain ¶
Drain returns all pending signals (any name) without blocking. Existed in an earlier draft for use by an at-commit-boundary journal — that draft has been superseded by the background pollControls goroutine architecture (engine/controls.go).
Phase 3 minimum: Drain is NOT called by the engine — early signals are picked up at the matching await's Receive call (via the "drain first" path in Receive). Drain exists for completeness + Phase 6 obs (which may want to project pending signals); tested but not wired in slice 3.5.
func (*Broker) PollInterval ¶
PollInterval returns the configured poll interval. Used by engine.Run's background polling goroutine (engine/controls.go) to set its ticker. Read-only after NewBroker — no setter; reconfigure via a fresh Broker.
func (*Broker) Receive ¶
Receive blocks until a signal of name arrives, ctx cancels, or timeout elapses (0 = no timeout). On delivery, the file is atomic-renamed into consumed/ BEFORE returning (so a crash before the engine commits signal.received leaves the file in consumed/; on resume the engine's Fold re-derives state from the log, not the broker). Returns the Delivery on success.
ctx-cancel: returns (Delivery{}, ctx.Err()). timeout: returns (Delivery{}, context.DeadlineExceeded) — caller maps to
retryable_failure per spec §4.3.
func (*Broker) ReceiveMatching ¶
func (b *Broker) ReceiveMatching(ctx context.Context, name string, timeout time.Duration, match MatchFunc) (Delivery, error)
ReceiveMatching is Receive with a payload predicate: it blocks until a signal of name whose payload satisfies match arrives, ctx cancels, or timeout elapses (0 = no timeout). The EARLIEST-seq matching signal is atomic-renamed into consumed/ before returning; non-matching (and unpredicatable) candidates are LEFT in place for other awaits. Same crash-safety contract as Receive (a crash before the engine commits signal.received leaves the file in consumed/; Fold re-derives state from the log, not the broker).
ctx-cancel: (Delivery{}, ctx.Err()). timeout: (Delivery{}, DeadlineExceeded) — the engine maps to retryable_failure (spec §4.3), identical to "no signal".
func (*Broker) WriteCancel ¶
func (b *Broker) WriteCancel(req CancelRequest) error
WriteCancel writes cancel.json. Idempotent.
func (*Broker) WritePause ¶
func (b *Broker) WritePause(req PauseRequest) error
WritePause writes pause.json. Idempotent — overwrites any existing file.
func (*Broker) WriteSignal ¶
WriteSignal writes signal-<name>-<seq>.json with payload bytes. seq is auto-allocated by re-reading the directory state on EACH attempt — every retry sees concurrent writers' just-committed seqs and bumps past them.
payload may be nil (signal with no payload). Empty file is written; readers distinguish via len(payload) == 0.
Returns the allocated seq on success.
Concurrency: O_EXCL is the atomicity primitive. On ErrExist, the loop re-runs nextSeq() to pick up the up-to-date max, then retries — convergence is O(N) per writer for N concurrent invocations. maxAttempts=50 covers reasonable bursts (CLI operators rarely fire >50 signals concurrently); exhaustion returns an error (slice 3.5 design Q6 + critique C4).
type BrokerOption ¶
type BrokerOption func(*Broker)
BrokerOption configures a Broker. Currently only one option exists; future additions (fsnotify, custom file modes) extend additively.
func WithPollInterval ¶
func WithPollInterval(d time.Duration) BrokerOption
WithPollInterval sets how often Broker.Receive polls the control directory for new signal files. Production default is 100ms; tests use ~1ms for determinism. Out-of-range values (<=0) are silently ignored; production default applies.
type CancelRequest ¶
type CancelRequest struct {
Reason string `json:"reason,omitempty"`
}
CancelRequest is the parsed body of cancel.json.
type Delivery ¶
Delivery is the result of consuming one signal file. Name + Seq identify the source; Payload is the file's raw bytes (empty for no-payload signals).
type MatchFunc ¶
MatchFunc decides whether a buffered signal's payload satisfies a keyed-await `where:` clause. Returns (true, nil) to consume this candidate, (false, nil) to skip it (leave it buffered for another await), or (false, err) when the payload cannot be predicated (e.g. not JSON) — treated as skip-this-candidate by tryConsumeMatching (the engine builds the predicate; see engine/signal_step.go).
type PauseRequest ¶
type PauseRequest struct {
NodePath string `json:"node_path,omitempty"`
Reason string `json:"reason,omitempty"`
}
PauseRequest is the parsed body of pause.json.
Source Files