task

package
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Sep 7, 2026 License: Apache-2.0 Imports: 11 Imported by: 0

Documentation

Index

Constants

View Source 
const (

	// MaxOutputBytes caps the total serialised size of collected outputs per
	// task to prevent unbounded memory/DB usage.  Tasks that need to pass
	// larger payloads should use shared storage and pass the reference via the
	// ##caesium::output-ref marker (see OutputRef); the reference itself is
	// small and counts against this cap, the payload it points at does not.
	MaxOutputBytes = 65536 // 64 KB

	// MaxLogSnapshotBytes caps the amount of raw task log text that Caesium
	// persists for completed tasks. This gives the UI a durable snapshot to
	// search and review after the runtime itself has been cleaned up.
	MaxLogSnapshotBytes = 1 << 20 // 1 MiB
)
View Source 
const (

	// MaxPartitionListBytes caps the normalized encoding of the full partition
	// list. Independent of MaxOutputBytes — a large work list must not eat the
	// scalar-output / fan-in budget.
	MaxPartitionListBytes = 256 * 1024

	// MaxPartitionObjectBytes caps one normalized partition object.
	MaxPartitionObjectBytes = 2048

	// MaxPartitionAttributes is the maximum number of free-form scalar
	// attributes a structured partition may carry.
	MaxPartitionAttributes = 16

	// MaxPartitionKeyBytes is the maximum UTF-8 size of a partition key.
	MaxPartitionKeyBytes = 256

	// DefaultMaxPartitions is the default count cap when the executor does not
	// pass one (matches CAESIUM_FANOUT_MAX_PARTITIONS's default).
	DefaultMaxPartitions = 1024

	// PartitionEnv is the default injected env var for the partition key.
	PartitionEnv = "CAESIUM_PARTITION"

	// PartitionJSONEnv is the fixed injected env var for the normalized
	// partition object. It is not renameable via fanOut.env.
	PartitionJSONEnv = "CAESIUM_PARTITION_JSON"
)
View Source 
const OutputNamesIndexEnvPrefix = "CAESIUM_OUTPUT_NAME_INDEX_"

OutputNamesIndexEnvPrefix is the dedicated environment prefix for the generated name index. The full name is CAESIUM_OUTPUT_NAME_INDEX_<NormalizeStepName(step)>. It sits outside CAESIUM_OUTPUT_<STEP>_<KEY> so a user key named caesium_output_names is forwarded as CAESIUM_OUTPUT_<STEP>_CAESIUM_OUTPUT_NAMES without colliding with the sidecar.

Locked to reagents/internal/tf.OutputNamesIndexEnvPrefix.

Variables

View Source 
var ErrFanInAggregateTooLarge = errors.New("task: fan-in aggregate exceeds the output size cap")

ErrFanInAggregateTooLarge is the sentinel behind FanInAggregateTooLargeError, for callers that only need errors.Is.

Functions

func AggregateFanInOutputs 

func AggregateFanInOutputs(producer string, byPartition map[string]map[string]string, succeeded, failed int) (map[string]string, error)

AggregateFanInOutputs folds per-partition instance outputs into the fan-in contract: each scalar key becomes a JSON object keyed by partition value, plus synthetic PARTITION_COUNT / SUCCEEDED / FAILED.

producer names the fan-out step, used only to attribute an over-cap failure. An aggregate that does not fit in MaxOutputBytes returns a *FanInAggregateTooLargeError and a nil map; callers must fail the group rather than publish a partial contract.

func BuildOutputEnv 

func BuildOutputEnv(predecessorOutputs map[string]map[string]string) (map[string]string, error)

BuildOutputEnv constructs CAESIUM_OUTPUT_<STEP>_<KEY>=<VALUE> environment variables from a map of predecessor step names to their output key-value pairs.

Scalar outputs map to a single CAESIUM_OUTPUT_<STEP>_<KEY> var carrying the value verbatim. A reference output (an encoded OutputRef) instead exposes the volume path the downstream container should read — CAESIUM_OUTPUT_<STEP>_<KEY> is set to the path, not the raw JSON — plus a companion CAESIUM_OUTPUT_<STEP>_<KEY>_DIGEST carrying the content digest so a consumer can re-verify the bytes it reads. The container contract stays string-only env vars: a large payload never enters the environment, only its location and digest do.

Every stored key is user data and is forwarded, including caesium_output_names. The generated name index is emitted separately as CAESIUM_OUTPUT_NAME_INDEX_<STEP> whenever any output key would not survive lowercasing the folded suffix. Snake_case-only maps omit the dedicated env, so existing cache keys stay byte-identical.

Any two values that would occupy one environment variable are refused. A map iteration winner would corrupt both the downstream value and its cache identity, so BuildOutputEnv must never guess.

func EncodeOutputNamesIndex 

func EncodeOutputNamesIndex(outputs map[string]string) (string, error)

EncodeOutputNamesIndex returns the JSON sidecar mapping folded env suffixes back to their original output keys. An empty string means every key already survives the fold (ToLower(NormalizeStepName(k)) == k) and the sidecar must be omitted.

func EncodePartitions 

func EncodePartitions(parts []Partition) ([]byte, error)

EncodePartitions returns the lossless normalized JSON array of partitions.

func IsOutputRef 

func IsOutputRef(value string) bool

IsOutputRef reports whether an output-map value is an encoded OutputRef. It is a cheap prefix/substring check used by BuildOutputEnv and the lineage mapper to treat references differently from scalars without fully decoding every value.

func NormalizeDependsOn 

func NormalizeDependsOn(key string, deps []string) ([]string, error)

NormalizeDependsOn canonicalizes one partition's dependsOn list: each entry is trimmed (matching partitionFromKey's treatment of the keys those entries reference), an entry that is empty after trimming is rejected as a producer bug rather than silently dropped, and duplicates are collapsed so the stored edge list and the indegree the scheduler seeds describe the same graph.

It is applied at parse time (so the persisted value is canonical) and again by ValidatePartitionGraph (so partitions constructed by any other path get the identical treatment). It is idempotent.

func NormalizeStepName 

func NormalizeStepName(name string) string

NormalizeStepName converts a step name to an environment-variable-safe prefix. Hyphens and dots are replaced with underscores and the result is uppercased. 

"etl-extract" → "ETL_EXTRACT"
"step.one"    → "STEP_ONE"

func OutputNamesIndexEnv 

func OutputNamesIndexEnv(stepName string) string

OutputNamesIndexEnv is the dedicated env var carrying the generated JSON name index for stepName.

func ParseBranches 

func ParseBranches(logs io.Reader) ([]string, error)

ParseBranches reads container log output and extracts branch selection markers. Each line matching "##caesium::branch <step-name>" adds the step name to the returned slice. Duplicate names are deduplicated while preserving first-seen order.

Lines that do not match the marker prefix are silently ignored.

func ParseOutput 

func ParseOutput(logs io.Reader) (map[string]string, error)

ParseOutput reads container log output and extracts structured key-value pairs from lines matching the ##caesium::output marker protocol.

Multiple marker lines are merged with last-write-wins semantics per key. All values are coerced to strings. If the total serialised output exceeds MaxOutputBytes an error is returned.

Lines that do not match the marker prefix are silently ignored (they are normal log output).

Types

type FanInAggregateTooLargeError 

type FanInAggregateTooLargeError struct {
	// Producer is the fan-out step whose group was being aggregated.
	Producer string
	// Size is the encoded size of the full aggregate, in bytes.
	Size int
	// Cap is MaxOutputBytes at the time of the failure.
	Cap int
}

FanInAggregateTooLargeError reports that a fan-out group's aggregated output does not fit inside MaxOutputBytes. It carries the observed size, the cap, and the producing step so the failure names the step an operator has to fix.

This is deliberately an error and not a degraded result. The aggregate used to be silently replaced by the three counters when it crossed the cap, which dropped EVERY user key: a downstream step's CAESIUM_OUTPUT_<PRODUCER>_<KEY> simply stopped existing, the run was reported successful, and the producer's declared outputSchema was never consulted because the synthesized aggregate does not flow through schema validation. A group whose contract cannot be honored must fail, not quietly change shape.

func (*FanInAggregateTooLargeError) Error 

func (e *FanInAggregateTooLargeError) Error() string

func (*FanInAggregateTooLargeError) Unwrap 

func (e *FanInAggregateTooLargeError) Unwrap() error

type Markers 

type Markers struct {
	Output       map[string]string
	Branches     []string
	Partitions   []Partition
	LogText      string
	LogTruncated bool
}

Markers holds the results of a unified single-pass parse of container logs, extracting both structured output key-value pairs and branch selection markers without buffering the entire log stream in memory.

func CaptureMarkers 

func CaptureMarkers(logs io.Reader, maxSnapshotBytes int) (*Markers, error)

CaptureMarkers reads container log output in a single pass, extracting structured markers while also capturing a bounded raw log snapshot suitable for UI display after the runtime has been cleaned up.

func CaptureMarkersWithLimits 

func CaptureMarkersWithLimits(logs io.Reader, maxSnapshotBytes int, maxRefBytes int64, maxPartitions int) (*Markers, error)

CaptureMarkersWithLimits is CaptureMarkersWithRefLimit plus the executor's effective partition count cap.

func CaptureMarkersWithRefLimit 

func CaptureMarkersWithRefLimit(logs io.Reader, maxSnapshotBytes int, maxRefBytes int64) (*Markers, error)

CaptureMarkersWithRefLimit is CaptureMarkers plus an operator-configured cap on the payload size a large-object reference (##caesium::output-ref) may declare. maxRefBytes <= 0 means unbounded. A reference whose reported size exceeds the cap is dropped (the producer's other outputs still apply); see env.Environment.OutputRefMaxBytes for the rationale.

func ParseMarkers 

func ParseMarkers(logs io.Reader) (*Markers, error)

ParseMarkers reads container log output in a single pass and extracts both structured output (##caesium::output) and branch selection (##caesium::branch) markers. This is more memory-efficient than calling ParseOutput and ParseBranches separately, as it avoids buffering the entire log stream.

func ParseMarkersWithLimits 

func ParseMarkersWithLimits(logs io.Reader, maxRefBytes int64, maxPartitions int) (*Markers, error)

ParseMarkersWithLimits is ParseMarkers plus operator/executor caps on large-object references and the partition count.

type OutputRef 

type OutputRef struct {
	// Ref is outputRefVersion at emit time. It is the first field so a future
	// reader can dispatch on the encoding version before trusting the rest.
	Ref int `json:"caesiumOutputRef"`
	// Path is the location the producing container wrote the payload to (a path
	// inside a mounted BYO volume). It is informational for downstream
	// containers (exposed as CAESIUM_OUTPUT_<STEP>_<KEY>) and deliberately NOT
	// part of the equality decision — only Digest is.
	Path string `json:"path"`
	// Digest is "sha256:" + the lowercase hex SHA-256 of the payload bytes,
	// computed by the producing container. It is the content address; folding
	// it into the hash is what makes the skip value-verified rather than
	// path/heuristic based.
	Digest string `json:"digest"`
	// Size is the payload size in bytes, if the producer reported it. Advisory
	// only (it does not participate in the digest); 0 means unreported.
	Size int64 `json:"size,omitempty"`
}

OutputRef is a content-addressed reference to a large payload a step wrote to a BYO volume/object store instead of inlining it as a structured output. Only the reference (path + digest + size) crosses the step boundary and lands in dqlite; the payload stays on the volume. The Digest is the load-bearing field for caching: it is folded into the consuming step's identity hash (via the encoded value, see Encode), so a step that re-emits a byte-identical payload — hence an identical digest — produces a cache hit. A path change with an unchanged digest does NOT change the hash, which is the point: the value, not its location, decides equality.

func DecodeOutputRef 

func DecodeOutputRef(value string) (OutputRef, bool)

DecodeOutputRef parses an encoded reference value back into an OutputRef. It returns ok=false (not an error) for any value that is not a well-formed reference, so callers can treat non-reference values as plain scalars.

func (OutputRef) Encode 

func (r OutputRef) Encode() string

Encode renders the reference as the canonical string stored under its key in the output map. encoding/json emits the struct fields in declaration order deterministically, so two byte-identical payloads (same digest) encode to the same string — which is exactly what folds into the consuming step's identity hash and yields a cache hit. The path is included for downstream consumers but, because the digest is present and the hash already covers the whole value, equality still tracks content: a moved file with identical bytes keeps the same digest and therefore re-uses the cache.

type Partition 

type Partition struct {
	Key         string            `json:"key"`
	Fingerprint string            `json:"fingerprint,omitempty"`
	DependsOn   []string          `json:"dependsOn,omitempty"`
	Attributes  map[string]string `json:"attributes,omitempty"`
}

Partition is the normalized partition element. A string-form emission is {Key: <value>} with empty Fingerprint/DependsOn/Attributes.

func NormalizePartitions 

func NormalizePartitions(parts []Partition) ([]Partition, error)

NormalizePartitions returns parts with every dependsOn list canonicalized. It is the single entry point callers outside the marker parser (recovery paths, tests) use so the graph and the persisted rows never disagree about keys.

func (Partition) CanonicalJSON 

func (p Partition) CanonicalJSON() ([]byte, error)

CanonicalJSON re-encodes the partition losslessly (sorted keys).

func (Partition) CanonicalObject 

func (p Partition) CanonicalObject() map[string]any

CanonicalObject lifts the partition to the lossless object form used for hashing, injection, and persistence. encoding/json sorts map keys, so the marshaled form is canonical.

func (Partition) EqualPayload 

func (p Partition) EqualPayload(other Partition) bool

EqualPayload reports whether two partitions are byte-identical after canonical encoding (used to distinguish first-seen dedup from conflict).

type PartitionError 

type PartitionError struct {
	Msg string
}

PartitionError is a parse/validation failure of the partition marker protocol. It must fail the producing task; the list is never truncated.

func (*PartitionError) Error 

func (e *PartitionError) Error() string

type PartitionGraph 

type PartitionGraph struct {
	Indegree   map[string]int
	Dependents map[string][]string
	Order      []string // Kahn order; emission order is PartitionIndex, not this
	MaxDepth   int
}

PartitionGraph is the in-group dependsOn graph over a normalized partition list. Indegree is the number of sibling keys an instance waits on (seeded onto outstanding_predecessors). Dependents is the reverse adjacency used for sibling decrement and the skip cascade.

func ValidatePartitionGraph 

func ValidatePartitionGraph(parts []Partition) (*PartitionGraph, error)

ValidatePartitionGraph resolves dependsOn against the emitted key set, rejects dangling keys and self-references, and runs a Kahn pass to detect cycles. Errors name the offending key(s).

type SchemaViolation 

type SchemaViolation struct {
	Key     string `json:"key"`
	Message string `json:"message"`
}

SchemaViolation describes a single output schema validation failure.

func ValidateOutput 

func ValidateOutput(output map[string]string, schemaRaw map[string]any) ([]SchemaViolation, error)

ValidateOutput validates a task's string-valued output map against a JSON Schema. XCom values are all strings; this function coerces them to typed values based on the schema's declared property types before validation.

Returns nil violations (not an error) when the output is valid. Returns a non-empty violations slice when validation fails. Returns an error only when the schema itself cannot be compiled.

func ValidateOutputSchemaBytes 

func ValidateOutputSchemaBytes(output map[string]string, schemaBytes []byte) ([]SchemaViolation, error)

ValidateOutputSchemaBytes validates a task output map against a schema encoded as JSON bytes.

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.