Documentation
¶
Overview ¶
Package freshness holds the data-freshness scheduling substrate: the declared dataset registry (this file) and — in later streams — the dataset state store and the leader-gated evaluator. The declared registry is the complement to the observed lineage graph: it is rebuilt from job manifests on every apply and requires no OpenLineage.
Index ¶
- Constants
- Variables
- func BuildDeclarations(def *schema.Definition, jobID uuid.UUID, jobAlias string) ([]models.DatasetDeclaration, error)
- func DeleteForJobsTx(tx *gorm.DB, jobIDs []uuid.UUID) error
- func EnrichStartParams(ctx context.Context, db *gorm.DB, jobID uuid.UUID, params map[string]string, ...) (map[string]string, error)
- func FreshAt(state models.DatasetState) (time.Time, bool)
- func ReplaceForJobTx(tx *gorm.DB, jobID uuid.UUID, decls []models.DatasetDeclaration) error
- func ValidateGraph(decls []models.DatasetDeclaration) error
- type AdvanceInput
- type AdvanceResult
- type ArrivalAdvance
- type ArrivalObserver
- type ArrivalResult
- type Capturer
- type Config
- type CronSkipDecision
- type Evaluator
- type LeaderCheck
- type Outcome
- type Registry
- type RunAdmitter
- type Store
Constants ¶
const ConsumedWatermarksStartParam = "_consumed_watermarks_start"
ConsumedWatermarksStartParam is the run param holding the watermarks of a run's consumed inputs AS THE RUN STARTED — the start-time truth, written by EnrichStartParams and read back at completion by Capturer.consumedForRun.
It is deliberately a DIFFERENT key from freshnessConsumedWatermarksParam (_consumed_watermarks), which carries a freshness-derived run's DERIVATION-time view and belongs to the evaluator alone. The two look alike — same JSON shape, same dataset keys — but they answer different questions and have different owners:
- _consumed_watermarks is the view the derivation DECISION was made on. The evaluator stamps it in derive() and matches on it in hasActiveOrQueuedRun to recognise a run it has already scheduled for those inputs. It must stay fixed for the life of the run or that dedupe stops recognising its own run.
- _consumed_watermarks_start is the view the run BEGAN on. It is re-taken whenever the run is (re-)created, which for a queued run means at promotion, because that is when it truly starts.
Collapsing them onto one key is what an earlier cut of this did, and it made the promotion refresh overwrite the evaluator's decision view: once promotion deletes the run_queue row, the running row is all hasActiveOrQueuedRun has left to match against, and it no longer matched — so the next tick derived a duplicate run for work already in flight.
Variables ¶
var ( // ErrDatasetMultipleProducers is returned when more than one job declares // that it produces the same dataset. ErrDatasetMultipleProducers = errors.New("dataset produced by multiple jobs") // ErrDatasetUnresolvedConsumes is returned when a consumed dataset is // neither produced by any job nor declared as a source. ErrDatasetUnresolvedConsumes = errors.New("consumed dataset is not produced or declared") // ErrDatasetGraphCycle is returned when the producer→consumer graph contains // a cross-job cycle (a dataset cycle is a derivation cycle). ErrDatasetGraphCycle = errors.New("dataset dependency cycle detected") )
Functions ¶
func BuildDeclarations ¶
func BuildDeclarations(def *schema.Definition, jobID uuid.UUID, jobAlias string) ([]models.DatasetDeclaration, error)
BuildDeclarations projects a validated job definition into the declaration rows that represent its slice of the declared graph: one row per declared source, produced dataset, and consumed dataset. The definition must have passed schema.Definition.Validate first. jobID/jobAlias identify the owning job (jobAlias is denormalized onto each row for cross-job lint).
func DeleteForJobsTx ¶
DeleteForJobsTx removes all declarations for the given jobs inside an existing transaction. Used when jobs are retired/pruned so the declared graph never references a job that no longer exists.
func EnrichStartParams ¶
func EnrichStartParams(ctx context.Context, db *gorm.DB, jobID uuid.UUID, params map[string]string, _ bool) (map[string]string, error)
EnrichStartParams is the run-store start-params hook that freezes a run's consumed-input watermarks onto the run row AT CREATION, under ConsumedWatermarksStartParam.
It exists because the consumed view has to be the one the run actually began with. Capturing it from an asynchronous run_started subscriber cannot promise that: the event only queues work, so the read can land after the run is already executing (an input that advanced in between is then credited to a run that never read it), and a full subscriber buffer drops the event outright, at which point there is no view at all. Enriching the params inside the creation path makes the view part of the INSERT: it is there or the run is not.
Every read here goes through the db handle the run store hands in, never a captured connection. Runs are created by stores built over an open transaction (internal/trigger/event/router.go), and reading a second connection underneath one of those deadlocks the whole database.
The result is read back at completion by Capturer.consumedForRun.
On the queued-concurrency path a run is admitted twice: once when it is enqueued (the view rides the run_queue row) and again when the dequeuer promotes it, which is when the run actually begins. The read is therefore unconditional and the second call OVERWRITES the first: a run that waited in the queue while its inputs advanced began on the newer view, and keeping the admission-time one would attribute its output to inputs it never read — the same misattribution this capture exists to prevent, only in the other direction (freshness would then see the output as behind and derive redundant work).
Because the start-time view has its own key, that refresh costs the evaluator nothing: a freshness-derived run's _consumed_watermarks is never read or written here, so its decision-time view — and the hasActiveOrQueuedRun dedupe that compares it — survives promotion untouched. It is also why fromQueue is no longer branched on. When the two views shared a key the flag was what told admission's "keep what is already there" apart from promotion's "take it again"; with a key of its own there is nothing to keep, and the honest rule is simply that every creation of a run re-reads the view that run starts with.
Register it from the server bootstrap under CAESIUM_FRESHNESS_ENABLED. It is a plain func rather than a run.StartParamsEnricher so this package keeps no dependency on internal/run — the dependency runs the other way.
func FreshAt ¶
func FreshAt(state models.DatasetState) (time.Time, bool)
FreshAt returns the effective freshness time for a dataset — max(advanced_at, verified_at) — and whether it has ever been observed. This is the single clock the evaluator measures the SLO against.
func ReplaceForJobTx ¶
ReplaceForJobTx rebuilds a single job's declarations inside an existing transaction: it hard-deletes the job's current rows and inserts the supplied set. This is the per-apply upsert seam — rebuilding from the manifest means a declaration removed from the manifest is pruned. Passing an empty slice clears the job's declarations (a job that dropped its datasets surface).
func ValidateGraph ¶
func ValidateGraph(decls []models.DatasetDeclaration) error
ValidateGraph runs the cross-job declared-graph checks over the full set of declarations — the applied set plus any persisted declarations from jobs not being replaced:
- exactly one job produces a given dataset (any number consume);
- every consumed dataset resolves to a produced dataset or a declared source (external:true is a declared source); and
- the producer→consumer graph is acyclic across jobs.
It is a pure function over the declaration rows so it can be unit-tested without a database; the caller (internal/jobdef) is responsible for gathering the applied + persisted declarations and excluding replaced jobs.
Types ¶
type AdvanceInput ¶
type AdvanceInput struct {
// Namespace is nullable and unused in v1; Name is the dataset identity.
Namespace *string
Name string
// Watermark is the emitted value. Empty means the producing step declared no
// watermark key (or emitted none): degraded mode, which refreshes verified_at
// against CompletedAt rather than advancing.
Watermark string
// RunID is the producing run; recorded as last_run_id.
RunID uuid.UUID
// RunOrder orders this observation against the run that set the current
// watermark — the completion (or start) time of the producing run, or a
// monotonic sequence surrogate. It gates opaque-string advances so a
// late-finishing older run can't clobber a newer opaque value. Zero falls
// back to CompletedAt.
RunOrder time.Time
// CompletedAt is the producing run's completion time, used for advanced_at /
// verified_at and for degraded-mode advances.
CompletedAt time.Time
// Consumed is the snapshot of this dataset's declared-input watermarks as of
// this producing run. It is persisted onto consumed_watermarks in the SAME
// transaction that ACCEPTS the advance/verify, tied to the accepted run — so
// when this Advance loses the race (a newer run's watermark wins under the
// conflict re-read), this run's input snapshot is NOT written either, and the
// winning run's snapshot stays authoritative. Empty means "leave the column
// untouched" (a produced dataset with no consumed inputs). Never written for
// a dropped outcome (backfill / regression / out-of-order).
Consumed map[string]string
// Backfill marks a backfill run: it never advances a watermark.
Backfill bool
}
AdvanceInput carries one producing observation of a dataset's watermark.
type AdvanceResult ¶
type AdvanceResult struct {
Outcome Outcome
State models.DatasetState
}
AdvanceResult is what the contract decided, plus the resulting state row.
type ArrivalAdvance ¶
ArrivalAdvance records one source declaration matched and advanced/verified by an ingested event. EventID is included for caller logs and future operator surfaces; DatasetState does not yet have a durable arrival_event_id field.
type ArrivalObserver ¶
type ArrivalObserver struct {
// contains filtered or unexported fields
}
ArrivalObserver bridges persisted ingested events into source dataset state. It reads declarations on each event so freshly applied bindings are visible without a separate reload hook.
func DefaultArrivalObserver ¶
func DefaultArrivalObserver() *ArrivalObserver
DefaultArrivalObserver returns the process-wide observer used by REST ingestion controllers.
func NewArrivalObserver ¶
func NewArrivalObserver(conn *gorm.DB) *ArrivalObserver
NewArrivalObserver constructs an arrival observer over the provided DB.
func (*ArrivalObserver) Observe ¶
func (o *ArrivalObserver) Observe(ctx context.Context, evt *models.IngestedEvent) (ArrivalResult, error)
func (*ArrivalObserver) SetBus ¶
func (o *ArrivalObserver) SetBus(bus event.Bus)
SetBus wires the observer to the event bus so an accepted arrival advance publishes dataset_advanced. Call once at startup before serving; the field is read on every Observe.
type ArrivalResult ¶
type ArrivalResult struct {
Advances []ArrivalAdvance
}
type Capturer ¶
type Capturer struct {
// contains filtered or unexported fields
}
Capturer hooks the run lifecycle path (NOT a poll): it subscribes to run_completed and, for each producing step's non-cached success, advances the dataset it declares — calling Store.Advance with the emitted watermark value (or refreshing verified_at in degraded mode when the step declares no watermark key or emits none). It also snapshots each produced dataset's consumed-input watermarks so "is my output up to date with my inputs" is a pure row comparison.
That consumed snapshot is the view the run had when it was CREATED, not the one current at its completion, so an input that advances mid-run is not credited to a run that never saw it. StartParamsEnricher stamps it onto the job_runs row synchronously at creation, under ConsumedWatermarksStartParam, so this subscriber only ever reads it back — falling back to a derived run's decision-time _consumed_watermarks and then to a completion-time read. See consumedForRun.
It reads the declared registry (dataset_declarations, freshness A2) to know which output key is a watermark, and the run's task_runs for the emitted ##caesium::output values. Backfill runs never advance (the monotonic guard is enforced in Store.Advance).
Wiring belongs to the freshness evaluator bootstrap (Stream C, cmd/start), gated by CAESIUM_FRESHNESS_ENABLED; the Capturer itself is inert until Start is called.
func NewCapturer ¶
NewCapturer constructs a Capturer over the event bus and DB connection.
type CronSkipDecision ¶
CronSkipDecision is the P1 skip-when-fresh decision for one cron tick.
func ShouldSkipCronRun ¶
func ShouldSkipCronRun(ctx context.Context, db *gorm.DB, jobID uuid.UUID, now time.Time) (CronSkipDecision, error)
ShouldSkipCronRun decides whether a cron tick can be omitted because every produced dataset is fresh and the job's consumed watermarks have not advanced since the run that produced the current outputs. It records skipped_fresh rows only after the full job-level decision is proven.
type Evaluator ¶
type Evaluator struct {
// contains filtered or unexported fields
}
func NewEvaluator ¶
func (*Evaluator) EvaluateDatasetNames ¶
func (*Evaluator) EvaluateEvent ¶
type Outcome ¶
type Outcome string
Outcome is the result of an Advance/Verify decision — the observable record of what the watermark contract did. Regressions and out-of-order opaque writes are *recorded* (returned) and dropped, exactly as the design requires.
const ( // OutcomeAdvanced — the watermark value changed (increased, for orderable // values; a newer producing run, for opaque values). advanced_at moved. OutcomeAdvanced Outcome = "advanced" // OutcomeVerified — a successful run confirmed the current watermark without // changing it (or emitted no watermark key at all: degraded mode). Only // verified_at moved. OutcomeVerified Outcome = "verified" // OutcomeRegressionDropped — an orderable watermark moved backwards. Recorded, // never advanced. OutcomeRegressionDropped Outcome = "regression_dropped" // OutcomeOutOfOrderDropped — an opaque watermark arrived from a run older than // the one that set the current value. Recorded, never advanced. OutcomeOutOfOrderDropped Outcome = "out_of_order_dropped" // OutcomeBackfillDropped — a backfill run never advances a watermark // (monotonic guard; derivations ignore backfill runs). OutcomeBackfillDropped Outcome = "backfill_dropped" )
type Registry ¶
type Registry struct {
// contains filtered or unexported fields
}
Registry is the typed store over the declared dataset graph (dataset_declarations). It is a projection of the applied manifest set — the importer rebuilds a job's rows on every apply and prunes them on retire — so callers read it as the authoritative declared graph.
func NewRegistry ¶
NewRegistry constructs a Registry over the provided connection.
func (*Registry) ListArrivalSources ¶
ListArrivalSources returns source declarations that carry an arrival binding.
type RunAdmitter ¶
type RunAdmitter interface {
AdmitRun(uuid.UUID, *uuid.UUID, ...runstorage.StartOption) (*runstorage.JobRun, bool, error)
}
type Store ¶
type Store struct {
// contains filtered or unexported fields
}
Store is the durable state store over dataset_states / dataset_derivations. It implements the watermark advance/verify contract that distinguishes "a run succeeded" from "the output advanced".
func (*Store) Advance ¶
func (s *Store) Advance(ctx context.Context, in AdvanceInput) (AdvanceResult, error)
Advance applies one producing observation to a dataset's state under the watermark contract:
- Backfill runs never advance (OutcomeBackfillDropped).
- An empty watermark (degraded mode) refreshes verified_at with CompletedAt.
- An unchanged watermark on a success refreshes verified_at, not advanced_at.
- An orderable watermark (numeric / RFC3339) advances only when it increases; a regression is recorded and dropped.
- An opaque-string watermark advances only when the producing run is newer than the one that set the current value; an out-of-order write is dropped.
Freshness is later evaluated against max(advanced_at, verified_at). The whole decision runs in one transaction against the (namespace, name) natural key.
func (*Store) Get ¶
func (s *Store) Get(ctx context.Context, namespace *string, name string) (models.DatasetState, bool, error)
Get returns the state row for a dataset, or (zero, false, nil) when none exists yet (an unknown dataset the evaluator serves before any run).
func (*Store) RecordConsumed ¶
func (s *Store) RecordConsumed(ctx context.Context, namespace *string, name string, consumed map[string]string) error
RecordConsumed snapshots the consumed-input watermarks onto an EXISTING produced dataset's state as a standalone write, for the consumed-only path where no advance is happening (e.g. an operator/manual surface). The normal producing-run path folds the snapshot into Advance (see AdvanceInput.Consumed) so watermark, last_run_id, and consumed_watermarks are written atomically and a losing run never clobbers the winner's snapshot. This updates ONLY the consumed_watermarks column (never a full-row Save) so a concurrent Advance is not clobbered, and is a no-op when the dataset has no state row yet.
Source Files