Documentation
¶
Index ¶
- Constants
- func InstanceInputHash(nodeID string, resolvedInputs types.JSONMap, params types.JSONMap) string
- func IsSuccessNode(node *types.AnalysisNode) bool
- func MatchInstance(existing []*types.AnalysisNode, nodeID string, inputHash string) *types.AnalysisNode
- func Materialize(req MaterializeRequest) (*types.AnalysisNode, error)
- func MergeUpstreamInputs(spec *Spec, params types.JSONMap, resolvedInputs types.JSONMap, ...)
- func Probe(req ProbeRequest) (*types.AnalysisNode, error)
- func SpecInstanceHash(spec *Spec, resolvedInputs types.JSONMap, params types.JSONMap) string
- func ToInt(v any, fallback int) int
- func ToJSONMap(v any) types.JSONMap
- func ToJSONSlice(v any) types.JSONSlice
- func ToString(v any) string
- func ToStringSlice(v any) []string
- func ToStringSliceFromAny(items []string) []any
- type MaterializeRequest
- type ProbeRequest
- type Spec
- type WorkspaceLayout
Constants ¶
const CreationSourceScheduler = "scheduler"
CreationSourceScheduler marks rows produced by a runtime scheduler.
Both V2 and V3 stamp the same value so downstream tooling (and the legacy imported rows) can tell scheduler-generated nodes apart from other producers.
Variables ¶
This section is empty.
Functions ¶
func InstanceInputHash ¶
InstanceInputHash is the stable identity of one materialized node instance.
It hashes the runtime node id together with the resolved inputs and the effective params, which is exactly the payload that decides what a node runs. Two schedulers that build the same node therefore compute the same hash, which is what allows them to recognize each other's persisted rows as the same instance (cache hit) instead of materializing a duplicate.
Hashes are persisted on analysis_nodes.input_hash. The payload shape (and the "{}" encoding of empty maps) is part of the contract and must not change without a migration, otherwise historical rows stop matching.
func IsSuccessNode ¶
func IsSuccessNode(node *types.AnalysisNode) bool
IsSuccessNode reports whether a persisted node holds a reusable successful result.
A node that is still "ready" but flagged as a cache hit represents a reused result from a previous run, so it counts as successful for input propagation.
func MatchInstance ¶
func MatchInstance(existing []*types.AnalysisNode, nodeID string, inputHash string) *types.AnalysisNode
MatchInstance finds the persisted node that represents the same instance as the probe.
Matching is by node id first and then by instance hash. Rows persisted before input_hash existed carry an empty hash; for those, a successful (or cache-hit) node with the same node id is accepted, so legacy rows stay reusable. When several instances share a node id (scatter), the exact hash wins.
func Materialize ¶
func Materialize(req MaterializeRequest) (*types.AnalysisNode, error)
Materialize assembles a brand new analysis node row with a dedicated workspace.
Upstream outputs are merged into params / resolved_inputs before the instance identity is computed, because the merged payload is what the node will actually execute with and therefore what must decide cache equality.
func MergeUpstreamInputs ¶
func MergeUpstreamInputs( spec *Spec, params types.JSONMap, resolvedInputs types.JSONMap, incoming []*types.AnalysisEdge, upstream map[string]*types.AnalysisNode, )
MergeUpstreamInputs injects the resolved outputs of successful upstream nodes into the downstream params / resolved_inputs payload.
The runtime engine performs the same merge when an upstream completes, but a node materialized *after* that point is not yet persisted when the propagation happens, so the merge must be repeated here.
Fan-in is honoured: when the target handle is declared "multiple", or the current value is already list-like, produced values accumulate instead of overwriting each other.
func Probe ¶
func Probe(req ProbeRequest) (*types.AnalysisNode, error)
Probe rewrites the volatile fields of an existing node with the freshly compiled template while keeping its identity and workspace.
The returned copy is never persisted directly: it is fed to the fingerprint probe and the cache policy, and its fields are copied back only when the node is actually rerun.
func SpecInstanceHash ¶
SpecInstanceHash computes the instance identity for a spec after its upstream inputs have been merged, using the same payload the builder persists.
func ToJSONMap ¶
ToJSONMap normalises any map-like payload into types.JSONMap.
It is exported so schedulers can reuse the exact conversion the builder applies, keeping their compiler-row parsing in sync with node construction.
func ToJSONSlice ¶
ToJSONSlice normalises any slice-like payload into types.JSONSlice.
func ToString ¶
ToString exposes the tolerant scalar-to-string conversion for scheduler row parsing.
func ToStringSlice ¶
ToStringSlice normalises a mixed array payload into a de-duplicated string slice.
func ToStringSliceFromAny ¶
ToStringSliceFromAny widens a plain string slice for JSONSlice columns.
Types ¶
type MaterializeRequest ¶
type MaterializeRequest struct {
// Analysis is the owning analysis row; its output directory anchors the
// workspace and its project id is copied onto the node.
Analysis *types.Analysis
// Spec is the compiled process template.
Spec *Spec
// Status is the initial node status.
Status string
// ErrorMessage explains a non-ready initial status (for example skipped).
ErrorMessage string
// CreationSource overrides the default creation_source marker.
CreationSource string
// Incoming are the edges that target this node, used to bootstrap inputs.
Incoming []*types.AnalysisEdge
// Upstream maps node id -> persisted node, used to resolve edge sources.
Upstream map[string]*types.AnalysisNode
// WorkspaceDir overrides the derived workspace root when a caller has
// already allocated one.
WorkspaceDir string
// NodeRecordID overrides the generated primary key (tests only).
NodeRecordID int64
}
MaterializeRequest describes a node that does not exist yet for this run.
type ProbeRequest ¶
type ProbeRequest struct {
// Analysis is the owning analysis row.
Analysis *types.Analysis
// Spec is the freshly compiled template to compare against.
Spec *Spec
// Existing is the node persisted by a previous run.
Existing *types.AnalysisNode
// Incoming are the edges that target this node.
Incoming []*types.AnalysisEdge
// Upstream maps node id -> persisted node.
Upstream map[string]*types.AnalysisNode
}
ProbeRequest describes a re-evaluation of an already persisted node.
type Spec ¶
type Spec struct {
// AnalysisNodeID is the compiler-assigned, rerun-stable identity. Empty means
// "generate one".
AnalysisNodeID string
// NodeID is the runtime identity used by edges.
NodeID string
// NodeName is the human readable instance name.
NodeName string
// SampleID identifies the scattered sample, empty for singletons.
SampleID string
// ScriptID is the resolved pipeline_components primary key.
ScriptID int64
// Executor selects the runtime backend (container, local, ...).
Executor string
// InputsPatterns declares input handles and their schemas.
InputsPatterns types.JSONMap
// ResolvedInputs holds the resolved input payload.
ResolvedInputs types.JSONMap
// OutputPatterns declares output handles and their schemas.
OutputPatterns types.JSONMap
// ResolvedOutputs holds the predicted outputs.
ResolvedOutputs types.JSONMap
// Params are the process parameters merged from workflow defaults.
Params types.JSONMap
// UpstreamIDs lists declared upstream node identities.
UpstreamIDs []string
// DownstreamIDs lists declared downstream node identities.
DownstreamIDs []string
// InputValidationErrors lists compile-time input problems.
InputValidationErrors types.JSONSlice
// Retry / MaxRetry configure in-place retries.
Retry int
MaxRetry int
// RerunReason carries a compiler supplied rerun explanation.
RerunReason string
// InputHash overrides the computed instance identity; empty computes it.
InputHash string
}
Spec is the scheduler-neutral description of one process instance to persist.
It is the common denominator the V2 compiler rows and the V3 process specs are both translated into, so the actual row assembly happens exactly once. Fields that only one scheduler produces are simply left at their zero value.
type WorkspaceLayout ¶
type WorkspaceLayout struct {
// Dir is the node root directory.
Dir string
// OutputDir receives produced artifacts.
OutputDir string
// CacheDir holds intermediate cache artifacts.
CacheDir string
// ParamsPath is the generated params.json path.
ParamsPath string
// CommandPath is the generated run.sh path.
CommandPath string
// LogPath is the captured command log path.
LogPath string
}
WorkspaceLayout is the on-disk contract of a single analysis node.
It is intentionally complete: every artifact path a node row stores comes from here, so the runtime preparer, the fingerprint probe and the completion resolver all agree on where files live.
func LayoutFor ¶
func LayoutFor(analysisOutputDir string, nodeRecordID int64) WorkspaceLayout
LayoutFor allocates the workspace layout below the analysis output directory.
The node record primary key doubles as the directory name so the on-disk tree stays traceable from the database row.
func LayoutFromWorkspaceDir ¶
func LayoutFromWorkspaceDir(workspaceDir string) WorkspaceLayout
LayoutFromWorkspaceDir derives the full layout from an already known root.
The directory contract itself lives in utils.NodeLayoutFor so that the DB layer (types.AnalysisNode) can derive the exact same paths without importing this package.
func ResolveLayout ¶
func ResolveLayout(analysisOutputDir string, workspaceOverride string, nodeRecordID int64) WorkspaceLayout
ResolveLayout picks the workspace for a new node.
Precedence:
- an explicit override, used by callers that pre-allocate a workspace;
- analysis.output_dir/<node record id>, the canonical layout;
- empty, when the analysis has no output directory yet - the caller is then responsible for filling the paths in later (see v3's path defaults).
func (WorkspaceLayout) IsZero ¶
func (l WorkspaceLayout) IsZero() bool
IsZero reports whether no layout could be derived, which happens when the analysis has no output directory configured yet.
Source Files