gateddagexec

package
v1.0.0-beta.122 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 1, 2026 License: MIT Imports: 18 Imported by: 0

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:

  1. 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).
  2. 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.
  3. 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.
  4. 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".

Consumer setup checklist (the create path; ADR-046 Phase 2.1)

Before the executor can dispatch, the consumer sets up the fan-out:

  1. (Optional) Set FanOutInstanceID so the framework owns the FanOut lifecycle: the executor then creates the `*.*.gateddag.fanout.instance.*` instance in `dispatching` on Start and auto-transitions it to `completed` when every unit is Done. If unset, no instance lifecycle is owned. FanOut *failure* is always consumer-driven (Manager.Fail) — a stall may be recoverable, so the framework never auto-fails the instance.
  2. Seed each unit as a graph entity under UnitEntityPrefix.
  3. Write the depends_on edges (DependsOnPredicate triples; Object = the prerequisite unit's entity ID) onto the dependents.
  4. On work completion/failure, write the CompletedPredicate / FailedPredicate marker on the unit; to re-run, follow the reset contract above.

Marker and edge predicates are FREE-FORM: graph-ingest validates only the indexing-profile predicate (ADR-054), so the gated-DAG markers need no vocabulary registration. Completions drive dispatch immediately (an internal KV watch over UnitEntityPrefix); the BackstopInterval tick is the correctness floor, not the primary path — raise it (longer net) rather than lower it. A configured StallSubject receives an edge-triggered StallEvent on a wedge.

Index

Constants

View Source 
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.

View Source 
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.

View Source 
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"
	// CategoryStall is the stall-event category (GH #365.3).
	CategoryStall = "stall"
)

Payload registry coordinates for the dispatch envelope.

View Source 
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.

View Source 
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.

View Source 
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

View Source 
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 Register 

func Register(registry *component.Registry) error

Register registers the gated-DAG executor factory with the component registry.

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 + stall envelopes.

func WorkflowDeclaration 

func WorkflowDeclaration() lifecycle.Workflow

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 

func (c *Component) Initialize() error

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 

func (c *Component) InputPorts() []component.Port

InputPorts returns the input ports — none (re-eval rides the lifecycle Watch, not a configured port).

func (*Component) Meta 

func (c *Component) Meta() component.Metadata

Meta returns component metadata.

func (*Component) OutputPorts 

func (c *Component) OutputPorts() []component.Port

OutputPorts returns the dispatch subject as the single output port.

func (*Component) Start 

func (c *Component) Start(ctx context.Context) error

Start builds the executor and begins the eval loop.

func (*Component) Stop 

func (c *Component) Stop(timeout time.Duration) error

Stop gracefully stops the executor.

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.
	// (retry_with_backoff is deferred — ADR-046 Phase 2.1 §4.)
	FailurePolicy string `json:"failure_policy,omitempty"`

	// FanOutInstanceID is the 6-part entity ID of the FanOut lifecycle instance
	// this executor owns (GH #364). OPTIONAL: when set, the executor creates the
	// instance in `dispatching` on Start and auto-transitions it to `completed`
	// when every unit is Done. When empty, no instance lifecycle is owned
	// (beta.117 behavior). FanOut *failure* stays consumer-driven either way.
	FanOutInstanceID string `json:"fan_out_instance_id,omitempty"`

	// StallSubject, when set, receives an edge-triggered StallEvent (on the
	// 0→non-zero stall transition) for active wedge-detection (GH #365.3).
	// OPTIONAL: the gated_dag_stalled_units gauge + WARN log are always emitted.
	StallSubject string `json:"stall_subject,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.

func (Config) Validate 

func (c Config) Validate() error

Validate checks a defaulted config. Call withDefaults first (NewComponent does). Returns the first violation found.

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) EntityID 

func (f *FanOut) EntityID() string

EntityID returns the federated 6-part identifier.

func (*FanOut) IsTerminal 

func (f *FanOut) IsTerminal() bool

IsTerminal reports whether the current phase has no declared out-edges.

func (*FanOut) ParentEntityID 

func (f *FanOut) ParentEntityID() string

ParentEntityID returns "" — fan-out instances have no parent workflow here.

func (*FanOut) Phase 

func (f *FanOut) Phase() string

Phase returns the current phase.

func (*FanOut) Workflow 

func (f *FanOut) Workflow() string

Workflow returns the registered workflow type name.

type StallEvent 

type StallEvent struct {
	// StalledUnits are the held-ready unit IDs with no forward progress.
	StalledUnits []string `json:"stalled_units"`
	// FanOutWorkflow / FanOutInstanceID identify the affected fan-out (provenance).
	FanOutWorkflow   string `json:"fan_out_workflow,omitempty"`
	FanOutInstanceID string `json:"fan_out_instance_id,omitempty"`
}

StallEvent is the edge-triggered wedge-detection event published to Config.StallSubject on the 0→non-zero stall transition (GH #365.3): the fan-out has held-ready units with no forward progress and nothing in-flight (a depends_on cycle, or every non-terminal unit blocked behind a failure). References only — the consumer reads the units' state from the graph.

func (*StallEvent) MarshalJSON 

func (s *StallEvent) MarshalJSON() ([]byte, error)

MarshalJSON marshals the payload fields (alias avoids recursion).

func (*StallEvent) Schema 

func (s *StallEvent) Schema() message.Type

Schema returns the type discriminator for registry routing.

func (*StallEvent) UnmarshalJSON 

func (s *StallEvent) UnmarshalJSON(data []byte) error

UnmarshalJSON unmarshals the payload fields (alias avoids recursion).

func (*StallEvent) Validate 

func (s *StallEvent) Validate() error

Validate checks required fields.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.