Documentation
¶
Overview ¶
Package gateddagexec is the executor component for gated-DAG dispatch (ADR-046 Phase 2, gh#357). It composes the pure selection brain (pkg/gateddag) with live framework substrate so a set of work units with depends_on edges is dispatched in dependency order, exactly once per unit, with restart recovery, failure isolation, and stall detection.
Why a component (not a rule action or condition) ¶
Gated-DAG dispatch needs a whole-set, authoritative, re-evaluated-on-every- change view of the unit set. The rule engine evaluates against the changed entity plus one related read (processor/rule/entity_watcher.go → evaluateRulesForEntityState) and a gate on a dependent never re-fires when a prerequisite completes (the engine evaluates rules against the prerequisite, not the dependent). A component watching the unit set provides the whole-set view natively. See ADR-046 "Why a component".
The wedge it kills: derived, never mutated ¶
Each evaluation re-reads the whole unit set authoritatively from the graph and re-derives every unit's status from marker membership (pkg/gateddag is a pure function of markers + edges). There is no projected status field to go stale and nothing to race the markers — the root of the wedge family that reverse-edge propagation could not escape (ADR-046 correctness requirement #1). The only state this component writes is the durable claim marker, which the brain ignores and the executor reads authoritatively for in-flight dedup.
The four load-bearing invariants (do not regress) ¶
An adversarial review of an earlier ADR draft caught it asserting guarantees the substrate does NOT provide. These hold here:
- Single-flight per fan-out, one instance per fan-out. replace_owned is last-write-wins, not CAS, and the owner-lease fence is default-off + cross-incarnation only — neither provides concurrent mutual exclusion. Dedup holds only because reEvaluate runs one pass at a time per instance (evalMu + a single eval goroutine) and exactly one component instance owns a given fan-out (singleton in the flow). If true cross-writer exclusion is ever needed, switch the claim to an ExpectedRevision-based CAS write (see the CAS-UPGRADE POINT in claim.go).
- Claim before dispatch. The claim marker is committed BEFORE the dispatch is published. A crash between publish and claim would re-derive the unit as Ready on restart and double-run.
- Re-eval + recovery ride the lifecycle Watch (which wraps KV WatchAll's bootstrap replay), NOT the pkg/dispatch completion-watcher (which suppresses bootstrap replay). The dispatcher is the bounded-concurrency leg only.
- Stall detection + the periodic backstop are net-new. No production code implements them; they are the core build of this phase, not reuse.
The reset contract (hard requirement on consumers) ¶
To re-dispatch a unit (recovery / re-run), the consumer MUST clear the unit's terminal markers (completed/failed) AND the claim marker, and set the dirtied marker. The brain's Dirtied precedence then re-derives the unit as Ready. As a defensive measure this executor treats a dirtied unit's stale claim as absent (dirtied overrides claim), so a reset that forgot to clear the claim still re-dispatches rather than wedging idle.
Framework / consumer boundary ¶
This component is domain-agnostic. It consumes a resolved depends_on edge set plus completion/failure/reset markers on per-unit entities, and on a dispatchable unit it (1) commits a durable claim marker then (2) publishes a reference (the unit entity ID — never content) to the configured dispatch subject. The consumer derives the edge set, mints fresh-per-run unit IDs, layers any domain gate, and wires its own handler on the dispatch subject to turn the reference into real work. See ADR-046 "Framework/consumer boundary".
Index ¶
- Constants
- Variables
- func CreateGatedDag(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
- func Register(registry *component.Registry) error
- func RegisterPayloads(reg *payloadregistry.Registry) error
- func WorkflowDeclaration() lifecycle.Workflow
- type Component
- func (c *Component) ConfigSchema() component.ConfigSchema
- func (c *Component) DataFlow() component.FlowMetrics
- func (c *Component) Health() component.HealthStatus
- func (c *Component) Initialize() error
- func (c *Component) InputPorts() []component.Port
- func (c *Component) Meta() component.Metadata
- func (c *Component) OutputPorts() []component.Port
- func (c *Component) Start(ctx context.Context) error
- func (c *Component) Stop(timeout time.Duration) error
- type Config
- type DispatchMessage
- type FanOut
Constants ¶
const ( // FailurePolicyContinueOthers keeps independent branches flowing when a unit // fails (its dependents stay Blocked; everything else proceeds). Default. FailurePolicyContinueOthers = "continue_others" // FailurePolicyStopOnFirstFailure halts all new dispatch once any unit has // failed (in-flight units still finish). A blunt circuit-breaker. FailurePolicyStopOnFirstFailure = "stop_on_first_failure" )
Default predicate vocabulary. Consumers may override any of these; the framework defaults give a working out-of-the-box configuration.
const ( PhaseDispatching = "dispatching" PhaseCompleted = "completed" PhaseFailed = "failed" )
FanOut phases. The instance is `dispatching` while units remain, then a terminal phase. Terminal detection is derived from the Transitions table.
const ( // Domain groups gated-DAG executor payloads. Domain = "gateddag" // SchemaVersion is the payload schema version. SchemaVersion = "v1" // CategoryDispatch is the dispatch-reference envelope category. CategoryDispatch = "dispatch" )
Payload registry coordinates for the dispatch envelope.
const FanOutEntityIDPattern = "*.*.gateddag.fanout.instance.*"
FanOutEntityIDPattern matches FanOut instance entities. Six-segment org.platform.domain.system.type.instance shape: org + platform + instance wildcard; domain (`gateddag`), system (`fanout`), type (`instance`) pin the canonical shape.
const FanOutWorkflow = "gateddag-fanout"
FanOutWorkflow is the framework-default lifecycle workflow name the executor watches and self-registers when the consumer does not supply its own.
const PredicateFanOutPhase = "gateddag.fanout.phase"
PredicateFanOutPhase is the FanOut phase predicate. lifecycle.Watch only delivers entries carrying this predicate, so the watched entities MUST carry it (the periodic backstop is the safety net for any entity that does not).
Variables ¶
var Transitions = lifecycle.Transitions{ PhaseDispatching: {PhaseCompleted, PhaseFailed}, PhaseCompleted: {}, PhaseFailed: {}, }
Transitions is the FanOut phase graph:
dispatching ──> completed
└─> failed
completed + failed are terminal (no out-edges).
Functions ¶
func CreateGatedDag ¶
func CreateGatedDag(rawConfig json.RawMessage, deps component.Dependencies) (component.Discoverable, error)
CreateGatedDag is the component.Factory for the gated-DAG executor.
func RegisterPayloads ¶
func RegisterPayloads(reg *payloadregistry.Registry) error
RegisterPayloads registers the gated-DAG executor payload types with the supplied registry. Wired into payloadbuiltins.Register so every binary that can run the executor can decode its dispatch envelope.
func WorkflowDeclaration ¶
WorkflowDeclaration returns the framework FanOut workflow ready for Manager.Register. The executor registers this when FanOutWorkflow is left at its default; a consumer that points Config.FanOutWorkflow at its own workflow registers that one itself.
Types ¶
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component is the gated-DAG executor processor.
func NewComponent ¶
func NewComponent(rawConfig json.RawMessage, deps component.Dependencies) (*Component, error)
NewComponent parses + validates config and wires dependencies.
func (*Component) ConfigSchema ¶
func (c *Component) ConfigSchema() component.ConfigSchema
ConfigSchema returns the configuration schema.
func (*Component) DataFlow ¶
func (c *Component) DataFlow() component.FlowMetrics
DataFlow returns data-flow metrics (the executor is event/timer driven; rate metrics are exposed via the prometheus collectors in metrics.go).
func (*Component) Health ¶
func (c *Component) Health() component.HealthStatus
Health returns current health status.
func (*Component) Initialize ¶
Initialize self-registers the framework FanOut workflow when the config uses the default workflow name, so mgr.Watch resolves it. A consumer that points FanOutWorkflow at its own workflow is responsible for registering that one. The executor requires a lifecycle Manager (it is the Watch substrate); a nil Manager is a wiring error surfaced loudly here rather than a silent no-op.
func (*Component) InputPorts ¶
InputPorts returns the input ports — none (re-eval rides the lifecycle Watch, not a configured port).
func (*Component) OutputPorts ¶
OutputPorts returns the dispatch subject as the single output port.
type Config ¶
type Config struct {
// FanOutWorkflow is the lifecycle.Workflow.Name this executor watches for
// re-eval triggers. Defaults to the framework FanOut workflow; a consumer
// may point it at its own registered workflow. The executor self-registers
// the framework default when this is left at the default (see component.go).
FanOutWorkflow string `json:"fan_out_workflow,omitempty"`
// UnitEntityPrefix is the graph.query.prefix scope read authoritatively each
// evaluation — the blast radius of one fan-out. REQUIRED (no default: it is
// the set of entities this executor will act on).
UnitEntityPrefix string `json:"unit_entity_prefix"`
// DispatchSubject is published (with the unit entity ID as a reference, never
// content) when a unit becomes dispatchable. The consumer wires its own
// handler here. REQUIRED.
DispatchSubject string `json:"dispatch_subject"`
// Marker / edge predicate vocabulary. Must be pairwise distinct after
// defaulting — a collision would make e.g. a completed marker read as a
// claim, silently corrupting dispatch.
CompletedPredicate string `json:"completed_predicate,omitempty"`
FailedPredicate string `json:"failed_predicate,omitempty"`
DirtiedPredicate string `json:"dirtied_predicate,omitempty"`
DependsOnPredicate string `json:"depends_on_predicate,omitempty"`
ClaimPredicate string `json:"claim_predicate,omitempty"`
// Workers / QueueSize bound the dispatch concurrency leg (pkg/dispatch).
Workers int `json:"workers,omitempty"`
QueueSize int `json:"queue_size,omitempty"`
// BackstopInterval is the period of the unconditional re-eval tick that
// closes the missed-watch-event hole and surfaces stalls (invariant #4).
BackstopInterval string `json:"backstop_interval,omitempty"`
// QueryTimeout bounds each authoritative graph.query.prefix read.
QueryTimeout string `json:"query_timeout,omitempty"`
// MaxUnits caps the authoritative whole-set read (QueryPrefixAll bound),
// guarding reply size / memory. A fan-out larger than this logs a truncation
// warning rather than silently dropping units.
MaxUnits int `json:"max_units,omitempty"`
// FailurePolicy selects how a failed unit affects new dispatch. One of
// FailurePolicyContinueOthers (default) or FailurePolicyStopOnFirstFailure.
FailurePolicy string `json:"failure_policy,omitempty"`
}
Config parameterizes the gated-DAG executor. The zero value is NOT valid — NewComponent applies DefaultConfig for unset fields then calls Validate.
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the framework defaults. Required fields (UnitEntityPrefix, DispatchSubject) are intentionally left empty — Validate rejects a config that does not set them.
func (Config) Schema ¶
func (c Config) Schema() component.ConfigSchema
Schema returns the component configuration schema (static metadata used by the registry + operator UI). Mirrors the field set in Config.
type DispatchMessage ¶
type DispatchMessage struct {
// UnitEntityID is the 6-part federated ID of the dispatchable unit. The
// consumer turns this reference into real work (e.g. a publish_agent).
UnitEntityID string `json:"unit_entity_id"`
// FanOutWorkflow is the workflow name of the fan-out instance that owns this
// unit (provenance / routing for the consumer).
FanOutWorkflow string `json:"fan_out_workflow,omitempty"`
}
DispatchMessage is the reference envelope published to Config.DispatchSubject when a unit becomes dispatchable. Per the orchestration rule "rules/dispatch carry references, never content", it carries only identifiers — the consumer retrieves the unit's details from the graph on demand.
It is a registered payload (wrapped in a BaseMessage at publish time) so the publish honors the payload-registry contract even though the immediate consumer may read it raw.
func (*DispatchMessage) MarshalJSON ¶
func (d *DispatchMessage) MarshalJSON() ([]byte, error)
MarshalJSON marshals the payload fields (alias avoids MarshalJSON recursion).
func (*DispatchMessage) Schema ¶
func (d *DispatchMessage) Schema() message.Type
Schema returns the type discriminator for registry routing.
func (*DispatchMessage) UnmarshalJSON ¶
func (d *DispatchMessage) UnmarshalJSON(data []byte) error
UnmarshalJSON unmarshals the payload fields (alias avoids recursion).
func (*DispatchMessage) Validate ¶
func (d *DispatchMessage) Validate() error
Validate checks required fields.
type FanOut ¶
type FanOut struct {
EntityIDField string `json:"entity_id" lifecycle:"id"`
PhaseField string `json:"phase" lifecycle:"phase,predicate=gateddag.fanout.phase"`
}
FanOut is the lifecycle.Participant for a gated-DAG fan-out instance (ADR-047). It is a thin named instance carrying restart recovery + operator gateway visibility; the unit set + edges + markers live on the per-unit entities under Config.UnitEntityPrefix, not here.
func (*FanOut) IsTerminal ¶
IsTerminal reports whether the current phase has no declared out-edges.
func (*FanOut) ParentEntityID ¶
ParentEntityID returns "" — fan-out instances have no parent workflow here.
Source Files