Documentation
¶
Overview ¶
Package runner executes one Filament ingestion run. It is the single implementation of the run sequence: resolve providers, open the sink, drive the pipeline, then commit or abort, publishing lifecycle facts throughout.
Both dispatch backends call RunOne — the engine module runs it in-process, the worker binary runs it once per Job — so where a run executes is a deployment choice and never a behavioral one. State transitions are not written here; they are folded from the published facts by the tracker. The sole exception is the resumable cursor seed in resolveExtractor.
Index ¶
- Constants
- func ExecuteContinuousAttempt(ctx context.Context, deps Deps, spec filament.RunSpec) (err error)
- func LoadContinuousAttempt(ctx context.Context, store filament.StreamRuntimeStore, ...) (filament.RunSpec, error)
- func PlanResources(ctx context.Context, src filament.Source, resources, selectors []string) ([]string, error)
- func ResolveConfigRefs(ctx context.Context, secrets filament.Secrets, spec *filament.RunSpec) error
- func RunContinuous(ctx context.Context, cfg ContinuousConfig) (result error)
- func RunOne(ctx context.Context, deps Deps, spec filament.RunSpec) error
- func ShouldRun(state filament.RunState) bool
- func SpecFromState(s filament.RunState) filament.RunSpec
- type ContinuousConfig
- type Deps
Constants ¶
const DefaultDrainTimeout = 30 * time.Second
DefaultDrainTimeout bounds graceful worker shutdown.
const DefaultLeaseTTL = 2 * time.Minute
DefaultLeaseTTL bounds worker ownership between renewals.
Variables ¶
This section is empty.
Functions ¶
func ExecuteContinuousAttempt ¶ added in v0.6.0
ExecuteContinuousAttempt executes a supervisor-admitted identity exactly once. It never allocates an attempt or falls back to a bounded lifecycle.
func LoadContinuousAttempt ¶ added in v0.6.0
func LoadContinuousAttempt(ctx context.Context, store filament.StreamRuntimeStore, state filament.RunState, executionID string) (filament.RunSpec, error)
LoadContinuousAttempt binds a worker to the immutable execution dispatched to it, refusing stale Jobs even when they refer to the same logical run.
func PlanResources ¶
func PlanResources(ctx context.Context, src filament.Source, resources, selectors []string) ([]string, error)
PlanResources resolves a run's requested resources to concrete source resources. An empty request means "all" for discoverable sources. Sources with private selector semantics can override this through ResourcePlanner.
func ResolveConfigRefs ¶
ResolveConfigRefs resolves opaque config and field references into the process-local RunSpec copy immediately before connector configuration.
func RunContinuous ¶ added in v0.6.0
func RunContinuous(ctx context.Context, cfg ContinuousConfig) (result error)
RunContinuous executes native epochs. The shared emitter publishes activity; runtime persistence, rather than event folding, owns continuous state. Providers must honor cancellation. Forced cancellation abandons tentative work.
func RunOne ¶
RunOne executes a single extraction end-to-end: resolve providers, open the sink, drive the pipeline with the source, then commit or abort. It publishes run.started first and exactly one terminal fact (run.completed | run.failed | run.partial | run.paused | run.canceled).
A non-nil return means execution was not admitted (see admit). Once admitted, outcomes travel as terminal facts and RunOne returns nil.
func ShouldRun ¶
ShouldRun reports whether a persisted run is this attempt's to execute. Only a freshly requested or resumable-partial run qualifies, so a redelivered trigger for a run already running or finished is a no-op rather than duplicated work. Each new CDC catch-up has its own run and resumes from the route checkpoint; a completed run must not be resurrected by a late trigger redelivery.
Types ¶
type ContinuousConfig ¶ added in v0.6.0
type ContinuousConfig struct {
Bus eventbus.Bus
Log filament.Logger
Enabled bool
Spec filament.RunSpec
Store filament.StreamRuntimeStore
Source filament.Source
Sink filament.Sink
// NewSink supplies fresh instances for sinks without reusable sessions.
NewSink func() (filament.Sink, error)
Codecs filament.CodecResolver
Schemas map[string]rowmodel.Schema
Boundary filament.Boundary
LeaseTTL, DrainTimeout time.Duration
// MaxEpochs bounds integration runs; zero runs until stopped or canceled.
MaxEpochs int
// contains filtered or unexported fields
}
ContinuousConfig configures an admitted serial attempt. RunOne handles only bounded runs. Write intent is resolved by the shared stream planner. Schemas default to the configured source.
type Deps ¶
type Deps struct {
Bus eventbus.Bus
DataStore filament.DataStore
StreamStore filament.ContinuousRunStore
Secrets filament.Secrets
Sources filament.SourceRegistry
Sinks filament.SinkRegistry
Log filament.Logger
Tracer filament.Tracer
}
Deps are the process-local dependencies needed to execute one run. Tracer is optional; a nil one skips span reporting.
Source Files