Help us shape the future of Go! Take the 2026 Go Developer Survey

filament

package module
v0.0.26-dev.2 Latest Latest
Warning

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

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

README

Documentation · Getting started · Contributing · Slack

Filament

Filament is pluggable data replication with checkpointing, batching, and integrity events.

It moves data from sources to sinks using full, incremental, or change data capture replication. Filament keeps progress durable as data moves, verifies batches on both sides of a write, and safely pauses or resumes runs from their checkpoints.

Sources, sinks, state storage, and event transport are replaceable interfaces. Run Filament as a service with its API and web UI, deploy it to your own infrastructure, or embed the engine in a Go application.

Getting started

One-click deployments for Railway and DigitalOcean are coming soon. Until then, you can run Filament locally or follow the documentation for more detailed guidance.

The local environment requires Go, Docker, just, Node.js, and pnpm. On macOS, the repository's Brewfile installs the toolchain:

brew bundle

Start PostgreSQL and NATS, then run Filament:

just infra
just dev

Open http://localhost:5173 to use the web UI.

When you are finished, stop the local infrastructure with:

just infra down

How it works

A Filament pipeline connects a source to a sink and defines how selected resources should be replicated. During a run, Filament extracts records into bounded batches, writes and verifies each batch, and advances durable checkpoints only after successful work.

The runtime is event-driven: the server accepts pipeline and run requests, the control plane schedules and dispatches work, and workers execute individual runs. PostgreSQL stores durable state and NATS JetStream carries lifecycle events between components.

Developing Filament

Filament is written in Go, with a React and TypeScript web UI. Contributions to the engine, connectors, UI, deployment tooling, tests, and documentation are welcome.

See CONTRIBUTING.md for repository setup, code generation, tests, and pull request guidance.

Community

Documentation

Index

Constants

View Source 
const (
	OpInsert = rowmodel.OpInsert
	OpUpdate = rowmodel.OpUpdate
	OpDelete = rowmodel.OpDelete
)

The row operations.

View Source 
const (
	LogicalUnknown     = rowmodel.LogicalUnknown
	LogicalBool        = rowmodel.LogicalBool
	LogicalInt16       = rowmodel.LogicalInt16
	LogicalInt32       = rowmodel.LogicalInt32
	LogicalInt64       = rowmodel.LogicalInt64
	LogicalFloat32     = rowmodel.LogicalFloat32
	LogicalFloat64     = rowmodel.LogicalFloat64
	LogicalDecimal     = rowmodel.LogicalDecimal
	LogicalString      = rowmodel.LogicalString
	LogicalBytes       = rowmodel.LogicalBytes
	LogicalDate        = rowmodel.LogicalDate
	LogicalTime        = rowmodel.LogicalTime
	LogicalTimestamp   = rowmodel.LogicalTimestamp
	LogicalTimestampTZ = rowmodel.LogicalTimestampTZ
	LogicalJSON        = rowmodel.LogicalJSON
	LogicalUUID        = rowmodel.LogicalUUID
	LogicalArray       = rowmodel.LogicalArray
)

The portable column types.

View Source 
const ConnectionSecretPrefix = "filament/"

ConnectionSecretPrefix namespaces the secret refs the connection API mints on a tenant's behalf: connections/<tenant>/<connID>/<field>/v<version>.

View Source 
const DefaultCheckpointEvery = 25

DefaultCheckpointEvery persists a resource's cursor every N written batches when RunOptions.CheckpointEvery is unset.

View Source 
const ScheduleLeaseTTL = 5 * time.Minute

ScheduleLeaseTTL bounds how long a ClaimDue lease is honored before a schedule is eligible to be reclaimed.

Variables

View Source 
var (
	// ErrInvalidToken is returned when an ID (tenant/run/…) can't be used as a
	// subject token. Aliased to the shared sentinel so errors.Is matches across
	// the eventbus boundary.
	ErrInvalidToken = eventbus.ErrInvalidToken

	// ErrNotFound is the sentinel a DataStore returns (wrapped) when a run,
	// resource, or checkpoint does not exist, so callers can branch with
	// errors.Is — e.g. the tracker creating run state on first sight of a fact.
	ErrNotFound = errors.New("not found")
)
View Source 
var ErrVersionConflict = errors.New("version conflict")

ErrVersionConflict indicates an optimistic-lock mismatch.

Functions

func CheckpointKind 

func CheckpointKind(cp Checkpoint) string

CheckpointKind describes a cursor's shape without exposing its value.

func ConnectionSecretRef 

func ConnectionSecretRef(tenant, connID, field string, version int64) string

ConnectionSecretRef builds the canonical ref for a connection-managed secret.

func IsCDC 

func IsCDC(types map[string]IngestionType) bool

IsCDC reports whether types replicates a change stream. CDC never mixes with other types on one route, so any CDC entry means the whole run is CDC.

func IsCDCAppend 

func IsCDCAppend(types map[string]IngestionType) bool

IsCDCAppend reports whether any resource retains the change stream as history.

func IsCDCIngestion 

func IsCDCIngestion(t IngestionType) bool

IsCDCIngestion reports whether an ingestion type reads a change stream.

func IsValidToken 

func IsValidToken(s string) bool

IsValidToken is the boolean form of ValidToken.

func OperationName 

func OperationName(op Operation) string

OperationName renders an Operation for error messages and logs.

func PrimaryKeyForResource 

func PrimaryKeyForResource(ctx context.Context, src Source, resource string) ([]string, error)

PrimaryKeyForResource resolves a resource's primary key from the source's schema or discovery catalog; nil when neither reports one.

func ValidToken 

func ValidToken(s string) error

ValidToken reports whether s is usable as a single subject token, wrapping ErrInvalidToken on the first problem. See eventbus.ValidToken.

func ValidateConnectionSecretRef 

func ValidateConnectionSecretRef(ref string, tenant TenantID) error

ValidateConnectionSecretRef enforces the sole cross-tenant isolation rule for the shared, ref-keyed secret store: a ref in the connection-managed namespace (connections/<tenant>/...) may only be read or written by its owning tenant. Refs outside that namespace (e.g. env-style) carry no tenant and pass through.

func ValidateReplication 

func ValidateReplication(replication ReplicationMode, t IngestionType) error

ValidateReplication enforces the one rule tying edges to connections: CDC runs happen on CDC connections, everything else on standard ones.

func ValidateSinkIngestion 

func ValidateSinkIngestion(spec SinkSpec, t IngestionType) error

ValidateSinkIngestion reports whether the sink spec can serve the write-side policy the ingestion type implies. Any sink may serve append and replace; upsert falls back to the Upsertable capability.

func ValidateSourceIngestion 

func ValidateSourceIngestion(spec ConnectorSpec, t IngestionType) error

ValidateSourceIngestion reports whether the source spec can serve the read-side policy the ingestion type implies.

Types

type ApplyOptions 

type ApplyOptions struct {
	Policy WritePolicy
}

ApplyOptions carries the per-resource write policy governing one Apply.

type ChangeAcknowledger 

type ChangeAcknowledger interface {
	AcknowledgeChanges(ctx context.Context, checkpoints map[string]Checkpoint) error
}

ChangeAcknowledger lets a CDC source release upstream stream retention after the sink commit and the final per-resource checkpoints are both durable. The runner calls it only after observing those checkpoints in the DataStore; implementations must never advance beyond the supplied positions.

type ChangeExtractOpts 

type ChangeExtractOpts struct {
	Resources   []string
	Checkpoints map[string]Checkpoint
	Limit       int
	Observe     SourceObserver
}

ChangeExtractOpts scopes one CDC extraction: resources plus the checkpoints to resume each from.

type ChangeSource 

type ChangeSource interface {
	ExtractChanges(ctx context.Context, sink RecordSink, opts ChangeExtractOpts) error
}

ChangeSource is the optional contract for CDC extraction: emit ordered changes from per-resource checkpoints instead of a full read.

type Checkpoint 

type Checkpoint = rowmodel.Checkpoint

Checkpoint is a resource's resumable cursor: keyed reads plus an immutable Set that returns an updated copy. CheckpointData is the concrete form.

type CheckpointCoverage 

type CheckpointCoverage uint8

CheckpointCoverage describes whether none, some, or all selected resources can resume from a durable source cursor. 

const (
	// CheckpointCoverageNone means no selected resource has a resumable cursor.
	CheckpointCoverageNone CheckpointCoverage = iota
	// CheckpointCoverageSome means only some selected resources have resumable cursors.
	CheckpointCoverageSome
	// CheckpointCoverageAll means every selected resource has a resumable cursor.
	CheckpointCoverageAll
)

func CheckpointCoverageFor 

func CheckpointCoverageFor(resources []string, types map[string]IngestionType) CheckpointCoverage

CheckpointCoverageFor returns the read-side checkpoint coverage for a run.

type CheckpointData 

type CheckpointData = rowmodel.CheckpointData

CheckpointData is the concrete backing for the Checkpoint interface — a resumable cursor. It implements Checkpoint with value-copy semantics on Set.

func NewCheckpoint 

func NewCheckpoint(resource string) *CheckpointData

NewCheckpoint returns an empty checkpoint for a resource.

type CheckpointPolicy 

type CheckpointPolicy string

CheckpointPolicy is when a cursor may be persisted relative to writes. 

const (
	CheckpointNone        CheckpointPolicy = "none"
	CheckpointAfterBatch  CheckpointPolicy = "after_batch"
	CheckpointAfterCommit CheckpointPolicy = "after_commit"
)

Checkpoint timings, weakest to strongest.

type Config 

type Config interface {
	String(key string) string
	Int(key string) int
	Bool(key string) bool
	Duration(key string) time.Duration
	Secret(key string) string
	SecretRef(key string) string
	Sub(key string) Config
	Has(key string) bool
	Raw() map[string]any
}

Config is typed, tolerant read access to a connector's configuration; every accessor misses to a zero value.

func NewConfig 

func NewConfig(m map[string]any) Config

NewConfig adapts a decoded map into a Config. A nil map yields an empty Config where every lookup misses (Has → false, scalars → zero).

type ConfigField 

type ConfigField struct {
	Name     string
	Type     FieldType
	Required bool
	Default  any
	Enum     []EnumOption
	Help     string
	Scope    FieldScope
	Secret   bool
	// VisibleWhen conditionally includes this field based on a sibling field.
	VisibleWhen *FieldCondition
	// Fields describes the members of an object field. It is empty for scalar
	// fields and for intentionally free-form objects.
	Fields []ConfigField
}

ConfigField describes one config field for validation and UI rendering.

type ConfigSchema 

type ConfigSchema struct{ Fields []ConfigField }

ConfigSchema declares the fields a connector's Config accepts.

type ConfigValidatable 

type ConfigValidatable interface {
	Validate(cfg Config) error
}

ConfigValidatable is the optional contract for connector-specific, pure configuration validation. Unlike LiveValidatable it must not access the network, so the API can safely use it while accepting connection settings.

type Connection 

type Connection struct {
	ID         string
	Tenant     string
	Kind       ConnectorKind
	Name       string
	Connector  string
	Config     map[string]any
	SecretRefs map[string]string
	Version    int64
	CreatedAt  int64
	UpdatedAt  int64
	// DeletedAt is unix milliseconds, zero when the connection is live.
	DeletedAt       int64
	CreatedByUserID string
	UpdatedByUserID string
	DeletedByUserID string
}

Connection is the persistence-domain representation of reusable connector config. Config contains no plaintext fields declared as FieldSecret.

type ConnectionFilter 

type ConnectionFilter struct {
	Tenant         string
	Kind           ConnectorKind
	IncludeDeleted bool
	ListOptions
}

ConnectionFilter narrows a connection listing by tenant and/or kind.

type ConnectorKind 

type ConnectorKind int

ConnectorKind identifies which registry owns a reusable connection. 

const (
	ConnectorKindUnspecified ConnectorKind = iota
	ConnectorKindSource
	ConnectorKindSink
)

Connector kinds identify which registry owns a reusable connection.

type ConnectorMaturity 

type ConnectorMaturity string

ConnectorMaturity describes how thoroughly a connector has been tested. 

const (
	// MaturityAlpha is documentation-derived and not yet tested end to end.
	MaturityAlpha ConnectorMaturity = "alpha"
	// MaturityBeta has run end to end, but is not verified for every supported
	// object or production environment.
	MaturityBeta ConnectorMaturity = "beta"
	// MaturityStable is fully production-ready across its supported objects.
	MaturityStable ConnectorMaturity = "stable"
)

Connector maturity levels, from least to most proven.

type ConnectorSpec 

type ConnectorSpec struct {
	Name           string
	DisplayName    string
	Description    string
	DarkLogoURL    string
	LightLogoURL   string
	Version        string
	Maturity       ConnectorMaturity
	Modes          []ReadMode
	SourcePolicies []SourcePolicy
	Config         ConfigSchema
	Resources      ResourceCapabilities
}

ConnectorSpec is a source's self-description: identity, supported modes and policies, config schema, and resource capabilities. It powers the catalog.

type Counter 

type Counter interface {
	Inc()
	Add(float64)
}

Counter is a monotonically increasing metric.

type CursorColumn 

type CursorColumn struct {
	SchemaField
	PrimaryKey       bool
	Eligible         bool
	Recommended      bool
	Rank             int
	Configurable     bool
	SupportsLookback bool
	Warning          string
}

CursorColumn is a schema field annotated with its incremental-cursor capabilities. Rank is connector-specific; lower positive values are better.

type CursorColumnProvider 

type CursorColumnProvider interface {
	CursorColumns(ctx context.Context, resource string) ([]CursorColumn, error)
}

CursorColumnProvider describes which resource columns can safely serve as durable incremental cursors and which one the connector recommends.

type DataStore 

type DataStore interface {
	Ping(ctx context.Context) error // reachability, for readiness probes
	EnsureTenant(ctx context.Context, id TenantID, name string) error

	SaveRun(ctx context.Context, s RunState) error
	// CreateRun persists a new run row or promotes a pre-created RunScheduled
	// row. A row that has progressed past RunScheduled is left untouched and
	// ErrVersionConflict returned, so a racing intake cannot roll a live run
	// back to an earlier status.
	CreateRun(ctx context.Context, s RunState) error
	LoadRun(ctx context.Context, id RunID) (RunState, error)
	// ListRuns returns the page selected by the filter's Limit/Offset plus the
	// total number of runs matching the filter before the page was cut.
	ListRuns(ctx context.Context, f RunFilter) ([]RunState, int, error)
	// DeleteRun removes a run and its resources. Only the scheduler calls it, to
	// reap pre-created RunScheduled rows; deleting a run that ever executed would
	// discard history. Deleting a missing run is a no-op.
	DeleteRun(ctx context.Context, id RunID) error

	UpsertResource(ctx context.Context, rs ResourceState) error // enabled toggle + progress
	ListResources(ctx context.Context, id RunID) ([]ResourceState, error)

	SaveCheckpoint(ctx context.Context, id RunID, cp Checkpoint) error
	LoadCheckpoint(ctx context.Context, id RunID, resource string) (Checkpoint, error)
	SaveResourceCheckpoint(ctx context.Context, state ResourceCheckpointState) error
	LoadResourceCheckpoint(ctx context.Context, key ResourceCheckpointKey) (ResourceCheckpointState, error)
	ListResourceCheckpoints(ctx context.Context, route ResourceCheckpointRoute) ([]ResourceCheckpointState, error)
	DeleteResourceCheckpoint(ctx context.Context, key ResourceCheckpointKey) error

	DedupSeen(ctx context.Context, tenant string, run RunID, seq uint64) (bool, error)

	CreateConnection(ctx context.Context, c Connection) (Connection, error)
	UpdateConnection(ctx context.Context, c Connection) (Connection, error)
	LoadConnection(ctx context.Context, id string) (Connection, error)
	ListConnections(ctx context.Context, f ConnectionFilter) ([]Connection, int, error)
	DeleteConnection(ctx context.Context, id string) error

	CreatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)
	CreatePipelineVersion(ctx context.Context, pipelineID string, v *ingestionv1.PipelineVersion) (*ingestionv1.PipelineVersion, error)
	UpdatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)
	// LoadPipeline includes soft-deleted pipelines; check DeletedAt before
	// mutating or running one.
	LoadPipeline(ctx context.Context, id string) (*ingestionv1.Pipeline, error)
	LoadPipelineVersion(ctx context.Context, pipelineID string, version int64) (*ingestionv1.PipelineVersion, error)
	ListPipelineVersions(ctx context.Context, f PipelineVersionFilter) ([]*ingestionv1.PipelineVersion, int, error)
	ListPipelines(ctx context.Context, f PipelineFilter) ([]*ingestionv1.Pipeline, int, error)
	DeletePipeline(ctx context.Context, id string) error
	Name() string
}

DataStore persists run, resource, checkpoint, and dedup state — the control plane's system of record.

type DiscoverOpts 

type DiscoverOpts struct{ Refresh bool }

DiscoverOpts controls discovery; Refresh bypasses any cached catalog.

type DiscoverResult 

type DiscoverResult struct{ Resources []Resource }

DiscoverResult is the resource catalog a Discoverable source returns.

type Discoverable 

type Discoverable interface {
	Discover(ctx context.Context, opts DiscoverOpts) (DiscoverResult, error)
}

Discoverable is the optional contract for browsing a source's available resources.

type Dispatcher 

type Dispatcher interface {
	Dispatch(ctx context.Context, spec RunSpec) (RunHandle, error) // bus | inline | binary | k8s
	Name() string
}

Dispatcher routes a resolved RunSpec to whatever executes it.

type EnumOption 

type EnumOption struct {
	Value string
	Label string
}

EnumOption is one ordered choice for an enum or list config field.

type ExtractOpts 

type ExtractOpts struct {
	Resources   []string
	Selectors   []string
	Limit       int // 0 = unbounded
	Parallelism int
	Observe     SourceObserver
}

ExtractOpts scopes one extraction: which resources, how fast. Read behavior per resource is carried by the checkpoint plan handed to ExtractFrom, not by a run-wide mode.

type Field 

type Field struct {
	Key   string
	Value any
}

Field is a structured log field.

type FieldCondition 

type FieldCondition struct {
	Field  string
	Values []string
}

FieldCondition controls whether a config field applies based on a sibling's string value.

type FieldScope 

type FieldScope int

FieldScope is the scope for a config field, aligning with either a connection config or a pipeline config. Connection == Only used to establish a connection to connector, reusable for all runs Pipeline == Varies based on pipeline For example: postgres has a dsn to connect (connection scope) and a table the data should land in (pipeline scope) 

const (
	ScopeUnspecified FieldScope = iota
	ScopeConnection
	ScopePipeline
)

The field scopes.

func (FieldScope) IsPipeline 

func (s FieldScope) IsPipeline() bool

IsPipeline reports whether the field is set per pipeline node rather than on the reusable connection.

type FieldType 

type FieldType int

FieldType is a ConfigField's value kind. 

const (
	FieldString FieldType = iota
	FieldInt
	FieldBool
	FieldSecret // rendered masked; stored as a ref
	FieldDuration
	FieldEnum
	FieldObject
	// FieldList is an ordered list of strings. Enum contains the selectable
	// values when the field is rendered as a checkbox group.
	FieldList
)

The config field types.

type Gauge 

type Gauge interface {
	Set(float64)
	Inc()
	Dec()
}

Gauge is a metric that can move in both directions.

type Histogram 

type Histogram interface{ Observe(float64) }

Histogram records a distribution of observed values.

type IncrementalPlanner 

type IncrementalPlanner interface {
	PlanIncremental(ctx context.Context, resources []string, prev map[string]Checkpoint, cursors map[string]ResourceCursorConfig) (map[string]Checkpoint, error)
}

IncrementalPlanner builds a durable per-resource plan from the previous cross-run watermark and versioned cursor configuration.

type IngestionPlan 

type IngestionPlan struct {
	WritePolicies map[string]WritePolicy
	RequiresCDC   bool
}

IngestionPlan is the resolved policy set for a run: per-resource write policies, each bound from its resource's own ingestion type.

func ResolveIngestionPlan 

func ResolveIngestionPlan(ctx context.Context, src Source, snk Sink, spec RunSpec) (IngestionPlan, error)

ResolveIngestionPlan validates each of the run's ingestion types against both connector specs — the source's narrowed by its connection config — and binds one write policy per resource from that resource's own type, discovering primary keys where the policy requires them.

type IngestionType 

type IngestionType string

IngestionType names how a run moves data end to end; it determines both the source's read policy and the sink's write policy. 

const (
	IngestionFullReplace       IngestionType = "full_replace"
	IngestionFullUpsert        IngestionType = "full_upsert"
	IngestionFullAppend        IngestionType = "full_append"
	IngestionIncrementalAppend IngestionType = "incremental_append"
	IngestionIncrementalUpsert IngestionType = "incremental_upsert"
	IngestionIncrementalDelete IngestionType = "incremental_delete"
	IngestionCDCMerge          IngestionType = "cdc_merge"
	IngestionCDCAppend         IngestionType = "cdc_append"
)

The defined ingestion types, named {read}_{write}: what the source reads crossed with how the sink lands it. CDC implies both sides.

func IngestionFor 

func IngestionFor(read ReadMode, write WriteMode) (IngestionType, error)

IngestionFor compiles the independent read and write levers into the engine's internal ingestion type. Unspecified levers default to a full replacement snapshot, except an incremental read defaults to upsert.

func TypeFor 

func TypeFor(types map[string]IngestionType, resource string) IngestionType

TypeFor returns the ingestion type governing resource: its per-resource entry when one exists, otherwise the route default ("" entry).

func (IngestionType) OrDefault 

func (t IngestionType) OrDefault() IngestionType

OrDefault substitutes IngestionFullReplace for the empty type.

func (IngestionType) SourcePolicy 

func (t IngestionType) SourcePolicy() SourcePolicy

SourcePolicy returns the source-side policy this ingestion type implies.

func (IngestionType) WriteCapability 

func (t IngestionType) WriteCapability() WritePolicyCapability

WriteCapability returns the capability a sink must offer to serve this type.

func (IngestionType) WritePolicy 

func (t IngestionType) WritePolicy() WritePolicy

WritePolicy returns the sink-side policy this ingestion type implies.

type Label 

type Label struct{ Key, Value string }

Label is a metric label.

type ListOptions 

type ListOptions struct {
	Search         string
	SortBy         string
	SortDescending bool
	Limit          int
	Offset         int
}

ListOptions controls search, deterministic ordering, and offset pagination for datastore-backed collection RPCs.

type LiveValidatable 

type LiveValidatable interface {
	TestConnection(ctx context.Context, cfg Config) error
}

LiveValidatable is the optional contract for probing connectivity with a config before any run uses it. Both sources and sinks may implement it.

type Logger 

type Logger interface {
	Trace(msg string, kv ...Field)
	Debug(msg string, kv ...Field)
	Info(msg string, kv ...Field)
	Warn(msg string, kv ...Field)
	Error(msg string, err error, kv ...Field)
	With(kv ...Field) Logger
}

Logger is structured leveled logging with field accumulation via With.

type LogicalType 

type LogicalType = rowmodel.LogicalType

LogicalType is a portable, engine-independent column type. A schema-aware source classifies each column into one of these so a sink can map to its own type system; alongside it SchemaField.Native carries the source's exact type for a same-engine round-trip. Decimal precision/scale and array element types live in Native, not here.

type Metric 

type Metric int

Metric is an aggregatable measure over runs. RunCount/RunRecords/RunBytes are sums; RunDuration is the mean duration over terminal runs; RunMemoryUsage/RunCPUUsage are the mean peak working-set bytes / CPU seconds over runs that reported usage. Mirrors metrics.v1.Metric (protos/metrics/v1/metrics.proto). 

const (
	MetricRunCount Metric = iota
	MetricRunRecords
	MetricRunBytes
	MetricRunDuration
	MetricRunMemoryUsage
	MetricRunCPUUsage
)

Metric values, mirroring metrics.v1.Metric's real (non-UNSPECIFIED) entries.

type Metrics 

type Metrics interface {
	Counter(name string, labels ...Label) Counter
	Gauge(name string, labels ...Label) Gauge
	Histogram(name string, labels ...Label) Histogram
}

Metrics vends the three instrument kinds by name and labels.

type MetricsDimension 

type MetricsDimension int

MetricsDimension names a runs column usable for grouping and filtering. Mirrors metrics.v1.Dimension; the zero value means unset ("no grouping", one total series/row with key ""). 

const (
	DimensionUnspecified MetricsDimension = iota
	DimensionTenantID
	DimensionPipelineID
	DimensionStatus
)

MetricsDimension values, mirroring metrics.v1.Dimension.

type MetricsFilter 

type MetricsFilter struct {
	Dimension MetricsDimension
	Values    []string
}

MetricsFilter narrows a query to rows whose dimension matches one of values.

type MetricsGranularity 

type MetricsGranularity int

MetricsGranularity is the bucket width for a timeseries query. Mirrors metrics.v1.MetricsGranularity. 

const (
	GranularityHour MetricsGranularity = iota
	GranularityDay
)

MetricsGranularity values, mirroring metrics.v1.MetricsGranularity's real (non-UNSPECIFIED) entries.

type MetricsStore 

type MetricsStore interface {
	Ping(ctx context.Context) error
	QueryRunTimeseries(ctx context.Context, q RunTimeseriesQuery) ([]RunTimeseries, error)
	QueryRunAggregate(ctx context.Context, q RunAggregateQuery) ([]RunAggregateRow, error)
}

MetricsStore is an optional DataStore capability (see PipelineScheduleStore) for run metrics queries, backing metrics.v1.MetricsService. Ping mirrors DataStore's (readiness probes); a backend that needs cleanup on shutdown additionally implements io.Closer (checked optionally, same as DataStore backends — see cmd/server/main.go).

type Operation 

type Operation = rowmodel.Operation

Operation is the change kind a row carries.

type OverlapPolicy 

type OverlapPolicy int

OverlapPolicy decides what a tick does when the previous run is still in flight. 

const (
	OverlapSkip  OverlapPolicy = iota
	OverlapAllow               // run concurrently
)

The supported overlap policies.

type PipelineFilter 

type PipelineFilter struct {
	Tenant         string
	IncludeDeleted bool
	ListOptions
}

PipelineFilter narrows a pipeline listing by tenant.

type PipelineScheduleStore 

type PipelineScheduleStore interface {
	ScheduleStore
	CreatePipelineWithSchedule(ctx context.Context, p *ingestionv1.Pipeline, schedule *ScheduleState) (*ingestionv1.Pipeline, error)
}

PipelineScheduleStore atomically creates a pipeline and its optional primary schedule.

type PipelineVersionFilter 

type PipelineVersionFilter struct {
	PipelineID string
	ListOptions
}

PipelineVersionFilter narrows a pipeline version listing.

type RateLimited 

type RateLimited interface {
	Limits() RatePolicy
}

RateLimited lets a source declare its own extraction rate ceiling.

type RatePolicy 

type RatePolicy struct {
	RequestsPerSecond float64
	Burst             int
}

RatePolicy is a token-bucket rate limit on source requests.

type ReadMode 

type ReadMode int

ReadMode is how a source reads: full scan, incremental from a cursor, or CDC. 

const (
	ModeFull ReadMode = iota
	ModeIncremental
	ModeCDC
)

The replication modes.

func (ReadMode) String 

func (m ReadMode) String() string

String renders a read mode for messages and logs.

type RecordSchema 

type RecordSchema = rowmodel.Schema

RecordSchema is one resource's column layout, as a source reports it and a Schematized sink consumes it.

type RecordSink 

type RecordSink = arrowbatch.Inlet

RecordSink is where a Source pushes extracted rows — the engine's inlet, not a data Sink. A source opens one RowWriter per (resource, part) it reads and appends rows into it; the pipeline turns the writer's flushes into Batches.

type Ref 

type Ref struct {
	Connector string
	ConfigRef string
	Config    map[string]any

	// SecretRefs maps a connector config field to a Secrets reference.
	SecretRefs map[string]string
}

Ref names a connector (source or sink) together with its config — the indirection a RunRequest carries instead of live instances.

type ReplicationAware 

type ReplicationAware interface {
	Replication(cfg Config) ReplicationMode
}

ReplicationAware lets a source report which replication mode a connection config selects. Sources without the contract are always standard.

type ReplicationMode 

type ReplicationMode string

ReplicationMode is how a connection replicates, decided at source creation: query-based reads or the change stream. 

const (
	ReplicationStandard ReplicationMode = "standard"
	ReplicationCDC      ReplicationMode = "cdc"
)

The replication modes.

func ReplicationOf 

func ReplicationOf(src Source, cfg Config) ReplicationMode

ReplicationOf resolves a connection's replication mode from its source.

type Resource 

type Resource struct {
	Name        string
	Selector    string
	Selectable  bool
	PrimaryKey  []string
	Schema      *RecordSchema // optional
	Estimated   int64
	DisplayName string
	Metadata    map[string]string
}

Resource is one discoverable unit of data (table, stream, endpoint) and the metadata needed to select and plan it.

type ResourceCapabilities 

type ResourceCapabilities struct{ Discoverable, PerResourceCursor bool }

ResourceCapabilities advertises what a source can do per resource.

type ResourceCheckpointKey 

type ResourceCheckpointKey struct {
	PipelineID        string
	PipelineVersionID string
	Route             string
	Resource          string
}

ResourceCheckpointKey identifies durable progress shared by runs of one immutable pipeline route. Resource is deliberately part of the key so each table advances independently.

type ResourceCheckpointRoute 

type ResourceCheckpointRoute struct {
	PipelineID        string
	PipelineVersionID string
	Route             string
}

ResourceCheckpointRoute identifies every resource cursor one pipeline version keeps for one source-to-sink route.

type ResourceCheckpointState 

type ResourceCheckpointState struct {
	Key        ResourceCheckpointKey
	Run        RunID
	Checkpoint Checkpoint
	UpdatedAt  time.Time
}

ResourceCheckpointState is the durable cursor plus its most recent writer. Connector-specific phases and cursor shapes remain opaque inside Checkpoint.

type ResourceCursorConfig 

type ResourceCursorConfig struct {
	Field           string
	LookbackSeconds int64
}

ResourceCursorConfig selects one resource's durable incremental field and optional overlap window for late transactions.

type ResourcePlanner 

type ResourcePlanner interface {
	PlanResources(ctx context.Context, resources, selectors []string) ([]string, error)
}

ResourcePlanner lets a source translate externally selected resources into the concrete resource names it will emit. Selectors remain source-private; the returned names are used by the engine for policy, schema, and checkpoint setup.

type ResourceState 

type ResourceState struct {
	Run      RunID // owning run — the key a DataStore files this under
	Tenant   TenantID
	Resource string
	Enabled  bool
	Status   RunStatus
	Records  int64
	Bytes    int64
	Error    string
}

ResourceState is one resource's persisted progress within a run.

type Resumable 

type Resumable interface {
	ExtractFrom(ctx context.Context, sink RecordSink, opts ExtractOpts, prev map[string]Checkpoint) error
}

Resumable is the optional source contract for checkpointed extraction. prev maps each resource to the cursor to resume from (a plan produced by ResumePlanner, carrying any progress from a prior run); a resource absent from prev — or mapped to nil — is read from the start. ExtractFrom opens one RowWriter per shard and stamps each row's keyset position (RowMeta.Key) so the pipeline can advance and persist the cursor.

type ResumePlanner 

type ResumePlanner interface {
	PlanResume(ctx context.Context, resources []string, prev map[string]Checkpoint) (map[string]Checkpoint, error)
}

ResumePlanner builds the initial per-resource checkpoint plan for a resumable run: the shard layout (PK split) plus any cursor carried over from prev. The engine persists the returned plan before extraction so a later resume sees a complete, stable shard layout. A resource that cannot be resumed (no primary key) maps to nil — the source reads it non-resumably and it is re-read whole on resume.

type RowMeta 

type RowMeta = rowmodel.Meta

RowMeta is what a row carries beside its columns: its operation and the resume metadata the pipeline turns into the batch's checkpoint delta.

type RowWriter 

type RowWriter = arrowbatch.RowWriter

RowWriter is a typed, columnar row appender for one (resource, part). A row is written by calling one Append method per schema field, in schema order, then EndRow; the first append opens the row and EndRow closes it. Each Append must match the field's Arrow type as mapped by arrowbatch.Schema: Bool, Int16, Int32, Int64, Float32, Float64, Decimal (Precision > 0), String (string, json, uuid, array, unknown, unbounded decimal), Bytes, Date (days since epoch), Time (microseconds since midnight), Timestamp (microseconds since epoch, UTC for timestamptz). Null is valid for any field.

A writer belongs to the one goroutine that appends to it. It flushes on its own when a chunk is full and, at the next EndRow, when the pipeline's flush timer has asked; a source whose stream goes idle calls Flush itself so buffered rows do not wait for the next one. Drain flushes and then marks the part complete.

type RunAggregateQuery 

type RunAggregateQuery struct {
	Tenant       TenantID
	Metrics      []Metric
	Since, Until time.Time
	// GroupBy splits the result into one row per dimension value;
	// DimensionUnspecified means one total row with key "".
	GroupBy MetricsDimension
	Filters []MetricsFilter
}

RunAggregateQuery parameterizes a single-window metrics query over runs.

type RunAggregateRow 

type RunAggregateRow struct {
	Key    string
	Values []float64
}

RunAggregateRow is one grouped row's metric values, positionally aligned with the query's Metrics. Key is "" when the query has no GroupBy.

type RunFilter 

type RunFilter struct {
	Tenant            TenantID
	PipelineID        string
	PipelineVersionID *string
	Status            []RunStatus
	Schedule          ScheduleID
	Since             time.Time
	Until             time.Time
	// UpdatedBefore matches runs whose last write is older than it — the
	// staleness probe: heartbeat folds bump UpdatedAt, so a Running run that
	// stops updating has lost its worker.
	UpdatedBefore  time.Time
	Limit          int
	Offset         int
	Search         string
	SortBy         string
	SortDescending bool
}

RunFilter narrows a DataStore run listing; zero fields match everything. Since is inclusive and Until exclusive on StartedAt — a window asks which runs ran in it, so runs that never started fall outside either bound.

type RunHandle 

type RunHandle interface {
	ID() RunID
	Status(ctx context.Context) (RunStatus, error)
	Wait(ctx context.Context) (RunResult, error)
}

RunHandle is a caller's grip on a dispatched run: poll its status or block for the result.

type RunID 

type RunID string

RunID identifies a run.

func (RunID) Valid 

func (r RunID) Valid() error

Valid reports whether the run ID is usable as a subject token (see ValidToken).

type RunOptions 

type RunOptions struct {
	FetchSize           int
	BatchMaxRows        int
	BatchMaxBytes       int64
	RateLimit           *RatePolicy
	SnapshotParallelism int

	// CheckpointEvery is how many written batches advance the cursor before it is
	// persisted (0 → DefaultCheckpointEvery). Larger values cut DataStore writes at
	// the cost of more redundant re-reads on resume (harmless: the sink is idempotent).
	CheckpointEvery int
}

RunOptions tunes throughput knobs for a run; zero values defer to engine defaults.

type RunRequest 

type RunRequest struct {
	Tenant             TenantID
	PipelineID         string
	PipelineVersionID  string
	IdempotencyKey     string
	Source             Ref
	Sink               Ref
	SourceConnectionID string
	SinkConnectionID   string
	Resources          []string
	Selectors          []string
	// IngestionTypes maps each resource to its ingestion type; the "" entry is
	// the route default for resources not explicitly listed.
	IngestionTypes  map[string]IngestionType
	CheckpointRoute string
	CursorConfigs   map[string]ResourceCursorConfig
	Options         RunOptions
	ScheduleID      ScheduleID
	// ScheduledFor is the occurrence this request represents; zero when manual.
	ScheduledFor time.Time
	// WorkerConfiguration is resolved at compile time and stamped here, so a
	// later edit to the pipeline cannot reshape a run already requested.
	WorkerConfiguration WorkerConfiguration `json:",omitzero"`
}

RunRequest is the caller-facing ask for a run, deduplicated by IdempotencyKey; the engine resolves it into a RunSpec.

func (RunRequest) ResourceCheckpointKey 

func (r RunRequest) ResourceCheckpointKey(resource string) (ResourceCheckpointKey, bool)

ResourceCheckpointKey returns the stable cross-run key for resource. False means the request did not originate from a versioned pipeline route.

type RunResult 

type RunResult struct {
	Status  RunStatus
	Records int64
	Bytes   int64
	URIs    []string
	Error   string
}

RunResult is the terminal outcome of a run as reported by a RunHandle.

type RunSpec 

type RunSpec struct {
	Tenant TenantID
	Run    RunID
	// StartedAt is the logical run's first start time. A zero value is stamped by
	// the runner; resumed executions retain the original value.
	StartedAt time.Time
	// ExecutionID identifies one dispatch attempt of a logical run. Dispatchers
	// derive it from the run.requested fact so redelivery is idempotent while a
	// later resume creates fresh worker infrastructure.
	ExecutionID        string
	PipelineID         string
	PipelineVersionID  string
	SourceConnectionID string
	SinkConnectionID   string
	CheckpointRoute    string
	CursorConfigs      map[string]ResourceCursorConfig
	Source             Ref
	Sink               Ref
	Resources          []string
	Selectors          []string
	// IngestionTypes maps each resource to its ingestion type; the "" entry is
	// the route default for resources not explicitly listed.
	IngestionTypes map[string]IngestionType
	Options        RunOptions
	WritePolicies  map[string]WritePolicy
	// WorkerConfiguration is the already-resolved worker shape for this run: the
	// pipeline's configuration with the request's override folded in.
	WorkerConfiguration WorkerConfiguration
}

RunSpec is the fully resolved execution plan for one run — what a Runtime receives after the engine has bound refs, ingestion type, and options.

func (RunSpec) ResourceCheckpointKey 

func (s RunSpec) ResourceCheckpointKey(resource string) (ResourceCheckpointKey, bool)

ResourceCheckpointKey returns the stable cross-run key for resource.

type RunState 

type RunState struct {
	Run       RunID
	Tenant    TenantID
	Status    RunStatus
	Request   RunRequest
	Resources []ResourceState
	Records   int64
	Bytes     int64

	// Lifecycle stamps, first-write-wins in the store. Created on insert,
	// scheduled at the occurrence's fire time, requested when run.requested is
	// emitted, started and finished folded from the run's own facts. Zero is
	// unset.
	CreatedAt   time.Time
	ScheduledAt time.Time
	RequestedAt time.Time
	StartedAt   time.Time
	EndedAt     *time.Time
	UpdatedAt   time.Time

	Error      string
	ScheduleID ScheduleID
	// Folded from run.heartbeat facts: cumulative worker CPU time and the
	// peak working set observed over the run.
	CPUSeconds      float64
	MemoryPeakBytes int64
}

RunState is the persisted record of a run: its request, per-resource progress, and terminal outcome.

type RunStatus 

type RunStatus int

RunStatus is the lifecycle state of a run or resource. 

const (
	RunRequested RunStatus = iota
	RunRunning
	RunCompleted
	RunFailed
	RunCanceled
	RunPaused
	// RunPartial is a run that failed mid-extract but persisted resumable progress.
	// Unlike RunFailed it is not terminal: re-emitting run.requested for the same
	// RunID resumes it from the last checkpoint. Only resumable runs reach it.
	RunPartial
	// RunScheduled is a run pre-created for a schedule's next occurrence, before
	// its fire time. It precedes RunRequested in lifecycle order but is declared
	// last so persisted ordinals stay stable and the zero value stays RunRequested.
	// Owned entirely by the control plane: the scheduler creates it and promotes
	// it to RunRequested at fire; nothing downstream ever sees it.
	RunScheduled
)

Run lifecycle states, in rough order of progression.

type RunTimeseries 

type RunTimeseries struct {
	Key    string
	Points []TimeseriesPoint
}

RunTimeseries is one grouped series: Key is the GroupBy dimension's value ("" when the query has no GroupBy), Points are dense — every bucket in [Since, Until) is present, zero-filled, ascending.

type RunTimeseriesQuery 

type RunTimeseriesQuery struct {
	Tenant          TenantID
	Metrics         []Metric
	Since, Until    time.Time
	Granularity     MetricsGranularity
	TZOffsetMinutes int
	// GroupBy splits the result into one series per dimension value;
	// DimensionUnspecified means one total series with key "".
	GroupBy MetricsDimension
	Filters []MetricsFilter
}

RunTimeseriesQuery parameterizes a bucketed metrics query over runs.

type RunTransitionOptions 

type RunTransitionOptions struct {
	ResetExecution   bool
	PreserveProgress bool
	Ended            bool
}

RunTransitionOptions controls the attempt-local state reset performed by a lifecycle transition. Ended stamps ended_at (first-write-wins) at transition time; set it only on transitions into terminal states, so the row carries an end time even if the async fact fold never lands.

type RunTransitionStore 

type RunTransitionStore interface {
	TransitionRun(
		ctx context.Context,
		id RunID,
		from []RunStatus,
		to RunStatus,
		opts RunTransitionOptions,
	) (RunState, error)
}

RunTransitionStore applies lifecycle commands with a compare-and-swap on the current status. It is separate from DataStore so adapters can reject run signaling explicitly instead of emulating an unsafe LoadRun/SaveRun race.

type Runtime 

type Runtime interface {
	Run(ctx context.Context, spec RunSpec) (RunHandle, error)
	Signal(ctx context.Context, h RunHandle, sig Signal) error
	Name() string
}

Runtime executes runs and delivers lifecycle signals to them.

type ScheduleFilter 

type ScheduleFilter struct {
	Tenant  TenantID
	Enabled *bool
	Limit   int
	Cursor  string
}

ScheduleFilter narrows a schedule listing; zero fields match everything.

type ScheduleID 

type ScheduleID string

ScheduleID identifies a schedule.

type ScheduleSpec 

type ScheduleSpec struct {
	Tenant     TenantID
	Name       string
	PipelineID string
	Cron       string
	Timezone   string
	Overlap    OverlapPolicy
	Enabled    bool
}

ScheduleSpec defines when a pipeline runs and whether occurrences may overlap.

type ScheduleState 

type ScheduleState struct {
	ID        ScheduleID
	Spec      ScheduleSpec
	Enabled   bool
	LastFired *time.Time
	NextFire  *time.Time
	CreatedAt time.Time
}

ScheduleState is a persisted schedule plus derived timing.

type ScheduleStore 

type ScheduleStore interface {
	SaveSchedule(ctx context.Context, s ScheduleState) error
	LoadSchedule(ctx context.Context, id ScheduleID) (ScheduleState, error)
	LoadPipelineSchedule(ctx context.Context, pipelineID string) (ScheduleState, error)
	ListSchedules(ctx context.Context, f ScheduleFilter) ([]ScheduleState, error)
	DeleteSchedule(ctx context.Context, id ScheduleID) error
	ClaimDue(ctx context.Context, now time.Time, limit int) ([]ScheduleState, error) // SELECT … FOR UPDATE SKIP LOCKED
	ReleaseScheduleClaim(ctx context.Context, id ScheduleID) error
}

ScheduleStore persists schedules and hands out due ones under a claim, so concurrent schedulers never double-fire.

type SchemaField 

type SchemaField = rowmodel.Field

SchemaField describes one column: name, nullability, and its portable plus native types.

type SchemaProvider 

type SchemaProvider interface {
	Schema(ctx context.Context, resource string) (RecordSchema, error)
}

SchemaProvider is an optional Source capability: it returns the column schema of a single resource. It is resource-scoped (lighter than the browse-style Discover) so the engine can fetch exactly the requested resources' schemas before extraction to drive a Schematized sink's typed DDL.

type Schematized 

type Schematized interface {
	EnsureSchema(ctx context.Context, resource string, schema RecordSchema) error
}

Schematized is the optional sink contract for typed DDL: materialize a resource's schema before its records arrive.

type Secret 

type Secret struct {
	Tenant TenantID
	Value  []byte
	Meta   map[string]string
}

Secret is plaintext, in-memory only; never logged, never on the bus.

type Secrets 

type Secrets interface {
	Read(ctx context.Context, ref string) (Secret, error)
	Write(ctx context.Context, ref string, s Secret) error
	Delete(ctx context.Context, ref string) error
	Name() string
}

Secrets resolves opaque references to secret material, keeping plaintext out of configs and stores.

type Signal 

type Signal int

Signal is a lifecycle command sent to a running run. 

const (
	SignalPause Signal = iota
	SignalResume
	SignalCancel
)

The run signals.

type Sink 

type Sink interface {
	Spec() SinkSpec // identity + config schema + capabilities (powers the catalog)
	Open(ctx context.Context, run RunSpec) error
	Apply(ctx context.Context, b *arrowbatch.Batch, opts ApplyOptions) (WriteReceipt, error)
	Commit(ctx context.Context) error
	Abort(ctx context.Context) error
	Name() string
}

Sink is the connector contract for writing data: open for a run, apply batches under a write policy, then commit or abort. Apply borrows b for the duration of the call; a sink that keeps it afterward must Retain it and eventually Release that reference.

type SinkCapabilities 

type SinkCapabilities struct {
	Transactional bool
	Upsertable    bool
	Schematized   bool
	// EncodedIntegrity requires Apply to verify the final serialized bytes at
	// its write boundary and return the resulting EncodedCRC as evidence.
	EncodedIntegrity bool
	WritePolicies    []WritePolicyCapability
	// PreferredBatchRows is the sink's preferred rows per Apply, used when the
	// run does not set Options.BatchMaxRows.
	// 0 defers to the engine default.
	PreferredBatchRows int
}

SinkCapabilities advertises the optional contracts and write modes a sink supports, so the engine can match it to an ingestion type.

type SinkFactory 

type SinkFactory func() Sink

SinkFactory constructs a fresh Sink instance per resolve.

type SinkRegistry 

type SinkRegistry interface {
	Register(name string, f SinkFactory)
	Resolve(name string) (Sink, error)
	Specs() []SinkSpec
}

SinkRegistry maps sink names to factories and exposes their specs for the catalog.

type SinkSpec 

type SinkSpec struct {
	Name         string
	DisplayName  string
	Description  string
	DarkLogoURL  string
	LightLogoURL string
	Version      string
	Maturity     ConnectorMaturity
	Config       ConfigSchema
	Capabilities SinkCapabilities

	// SchemaField names the pipeline-scoped config field holding where the
	// sink lands this pipeline's output — a schema, namespace, database, or
	// key prefix — so the server can default it from the source connection's
	// normalized name. Empty for sinks without one.
	SchemaField string
}

SinkSpec is a sink's self-description: identity, config schema, and capabilities. It powers the catalog.

type Source 

type Source interface {
	Spec() ConnectorSpec
	Validate(cfg Config) error
	Configure(ctx context.Context, cfg Config) error
	Extract(ctx context.Context, sink RecordSink, opts ExtractOpts) error
	Teardown(ctx context.Context) error
}

Source is the connector contract for reading data: describe itself, validate and take config, extract into a RecordSink, and tear down.

type SourceFactory 

type SourceFactory func() Source

SourceFactory constructs a fresh Source instance per resolve.

type SourceObserver 

type SourceObserver func(SourceProgress)

SourceObserver receives progress concurrently when a source extracts more than one resource or fans out. Implementations must be concurrency-safe.

func (SourceObserver) Report 

func (o SourceObserver) Report(progress SourceProgress)

Report delivers progress when an observer is configured. A nil observer is a safe no-op so sources can report without branching at every call site.

type SourcePolicy 

type SourcePolicy struct {
	Mode          ReadMode
	EmitsOps      []Operation
	Ordered       bool
	Checkpointing CheckpointPolicy
}

SourcePolicy is the read-side contract an ingestion type implies: mode, emitted operations, ordering, and checkpoint timing.

func SourcePolicies 

func SourcePolicies(types ...IngestionType) []SourcePolicy

SourcePolicies maps each ingestion type to its source policy, preserving order — the shape connector Specs advertise.

func SourcePolicyForIngestion 

func SourcePolicyForIngestion(t IngestionType) SourcePolicy

SourcePolicyForIngestion derives the canonical read-side policy for an ingestion type.

type SourceProgress 

type SourceProgress struct {
	Kind         SourceProgressKind
	Resource     string
	Records      int64
	Bytes        int64
	URI          string
	ParentsTotal int64
	RetryAfter   time.Duration
	Checkpoint   *CheckpointData
	Error        string
}

SourceProgress carries source-local extraction progress to the runner. Fields are populated according to Kind. Checkpoint on WatermarkAdvanced describes an observed cursor only; durable checkpoint persistence remains tracker-owned.

type SourceProgressKind 

type SourceProgressKind uint8

SourceProgressKind identifies a non-terminal extraction signal. Run and resource lifecycle state remains runner-owned; these signals describe work only the source can observe directly. 

const (
	SourceProgressPageFetched SourceProgressKind = iota + 1
	SourceProgressFanOutStarted
	SourceProgressWatermarkAdvanced
	SourceProgressRateLimited
	SourceProgressRetryExhausted
)

Source progress kinds identify the source-local signals an observer can report.

type SourceRegistry 

type SourceRegistry interface {
	Register(name string, f SourceFactory)
	Resolve(name string) (Source, error)
	Specs() []ConnectorSpec
}

SourceRegistry maps source names to factories and exposes their specs for the catalog.

type Span 

type Span interface {
	End()
	SetError(err error)
	SetAttr(key string, v any)
}

Span is one traced operation; End it exactly once.

type StageID 

type StageID string

StageID identifies a Transactional sink's staging area.

type SyncSnapshot 

type SyncSnapshot struct {
	Run       RunState
	Resources []ResourceState
	AtSeq     uint64
}

SyncSnapshot is a consistent read of a run and its resources at bus sequence AtSeq, the anchor a live tail resumes from.

type TenantID 

type TenantID string

TenantID identifies a tenant. 

const DefaultTenantID TenantID = "00000000-0000-0000-0000-000000000000"

DefaultTenantID identifies the built-in tenant used when callers omit one.

func (TenantID) Valid 

func (t TenantID) Valid() error

Valid reports whether the ID is usable as a subject token (see ValidToken). TenantID and RunID become subject tokens, so they must validate before a run is accepted or an event published.

type TimeseriesPoint 

type TimeseriesPoint struct {
	BucketStart time.Time
	Values      []float64
}

TimeseriesPoint is one bucket's metric values, positionally aligned with the query's Metrics.

type Tracer 

type Tracer interface {
	Start(ctx context.Context, name string) (context.Context, Span)
}

Tracer starts spans; the returned context carries the span for nesting.

type Transactional 

type Transactional interface {
	Stage(ctx context.Context) (StageID, error)
	Promote(ctx context.Context, id StageID) error
}

Transactional is the optional sink contract for staged, atomic runs: write into a stage, promote on success.

type Upsertable 

type Upsertable interface {
	Upsert(ctx context.Context, b *arrowbatch.Batch, keys []string) (WriteReceipt, error)
}

Upsertable is the optional sink contract for key-based merge writes.

type VersionPolicy 

type VersionPolicy struct {
	Strategy VersionStrategy
	Field    string
	Logical  LogicalType
	Native   string
}

VersionPolicy tells a sink how to resolve competing values for one primary key. Field and its types are populated for VersionCursor; they are empty for VersionInsertOrder.

func ResolveCursorVersionPolicy 

func ResolveCursorVersionPolicy(ctx context.Context, src Source, resource string, config ResourceCursorConfig) (VersionPolicy, error)

ResolveCursorVersionPolicy binds a resource's configured or recommended incremental cursor into the generic version contract used by sinks. A zero policy means the source does not expose cursor metadata and callers should retain their insertion-order fallback.

func ResolveWriteVersionPolicy 

func ResolveWriteVersionPolicy(
	ctx context.Context,
	src Source,
	resource string,
	config ResourceCursorConfig,
	sourceMode ReadMode,
	writeMode WriteMode,
) (VersionPolicy, error)

ResolveWriteVersionPolicy binds the version strategy for one source-to-sink resource. Non-upsert writes do not need a version; snapshot upserts and incremental sources without cursor metadata use insertion order.

type VersionStrategy 

type VersionStrategy string

VersionStrategy selects the ordering value an insert-based upsert sink uses when several records share the same primary key. 

const (
	// VersionInsertOrder leaves conflict ordering to the sink's insertion and
	// merge semantics. It is the fallback when no row-level cursor is available.
	VersionInsertOrder VersionStrategy = "insert_order"
	// VersionCursor orders rows by the source field used for incremental reads.
	// The field must be present, non-null, and advance on every source update.
	VersionCursor VersionStrategy = "cursor"
)

type WorkerConfiguration 

type WorkerConfiguration struct {
	Resources    WorkerResources
	NodeSelector map[string]string
	Tolerations  []WorkerToleration
}

WorkerConfiguration is how a run's worker is shaped, a subset of a Kubernetes pod spec. It travels as a whole so adding a knob does not change every signature that carries it.

func (WorkerConfiguration) IsZero 

func (w WorkerConfiguration) IsZero() bool

IsZero reports whether nothing is set.

func (WorkerConfiguration) Merge 

func (w WorkerConfiguration) Merge(base WorkerConfiguration) WorkerConfiguration

Merge overlays w onto base and returns the result. Resources merge per key; node selector and tolerations replace wholesale when set, since merging selector keys from two sources would produce a node set neither author asked for.

type WorkerResources 

type WorkerResources struct {
	Requests map[string]string
	Limits   map[string]string
}

WorkerResources sizes the worker that executes a run, shaped like a Kubernetes ResourceRequirements: resource name ("cpu", "memory") to quantity string ("500m", "2Gi"). A missing key is left off the Job so a namespace LimitRange can supply it. Only the Kubernetes dispatcher reads them; in-process execution ignores them entirely.

func (WorkerResources) IsZero 

func (w WorkerResources) IsZero() bool

IsZero reports whether nothing is set.

func (WorkerResources) Merge 

func (w WorkerResources) Merge(base WorkerResources) WorkerResources

Merge overlays w's keys onto base, per map, and returns the result. This is how a per-run override folds over a pipeline's configured default: overriding memory alone leaves the pipeline's CPU in place.

type WorkerToleration 

type WorkerToleration struct {
	Key      string
	Operator string
	Value    string
	Effect   string
}

WorkerToleration mirrors a Kubernetes toleration verbatim: Operator is "Equal" (the default when empty) or "Exists"; Effect is "NoSchedule", "PreferNoSchedule", "NoExecute", or empty to match every effect.

type Workload 

type Workload int

Workload is what a dispatcher can prove about the worker it launched for a run. The reaper reads it to tell a dead run from one that is slow to start or alive but silent. 

const (
	// WorkloadAbsent means no workload exists for the run: it was never
	// dispatched, or the platform has already garbage-collected it.
	WorkloadAbsent Workload = iota
	// WorkloadActive means the workload exists and has not finished.
	WorkloadActive
	// WorkloadFinished means the workload ran to a terminal outcome.
	WorkloadFinished
)

Workload states, from least to most evidence of a completed attempt.

func (Workload) String 

func (w Workload) String() string

type WriteAtomicity 

type WriteAtomicity string

WriteAtomicity is the unit at which a sink's writes become visible. 

const (
	AtomicityBatch    WriteAtomicity = "batch"
	AtomicityResource WriteAtomicity = "resource"
	AtomicityRun      WriteAtomicity = "run"
)

Atomicity units, smallest to largest.

type WriteDurability 

type WriteDurability string

WriteDurability identifies the sink operation after which an acknowledged Apply is recoverable. It is distinct from visibility/atomicity: a sink may make one resource visible at a time while still buffering every Apply until the run's Commit call. 

const (
	// DurabilityAfterApply means a successful Apply is recoverable without the
	// run's Commit call.
	DurabilityAfterApply WriteDurability = "after_apply"
	// DurabilityAfterCommit means Apply only stages data and checkpoints must
	// remain tentative until Commit succeeds.
	DurabilityAfterCommit WriteDurability = "after_commit"
)

type WriteMode 

type WriteMode string

WriteMode is how a sink lands records: append, replace, upsert, delete, or CDC merge. 

const (
	WriteAppend  WriteMode = "append"
	WriteReplace WriteMode = "replace"
	WriteUpsert  WriteMode = "upsert"
	WriteDelete  WriteMode = "delete"
	WriteMerge   WriteMode = "merge"
)

The defined write modes.

func WriteModesFor 

func WriteModesFor(read ReadMode) []WriteMode

WriteModesFor returns the user-selectable write modes compatible with read.

type WritePolicy 

type WritePolicy struct {
	Capability WritePolicyCapability
	Resource   string
	Keys       []string
	Version    VersionPolicy
	Checkpoint CheckpointPolicy
}

WritePolicy binds a capability to one resource's keys and checkpoint timing — the per-resource contract handed to a sink via ApplyOptions.

func WritePolicyForIngestion 

func WritePolicyForIngestion(t IngestionType) WritePolicy

WritePolicyForIngestion derives the canonical sink-side policy for an ingestion type; keys and resource are bound later, per resource.

func (WritePolicy) ValidateBatch 

func (p WritePolicy) ValidateBatch(resource string, batch *arrowbatch.Batch) error

ValidateBatch rejects the first row whose immutable batch operation the policy does not accept.

func (WritePolicy) ValidateOps 

func (p WritePolicy) ValidateOps(resource string, ops []Operation) error

ValidateOps rejects the first row whose operation the policy does not accept. A nil ops (all inserts) is accepted by every policy.

type WritePolicyCapability 

type WritePolicyCapability struct {
	Mode          WriteMode
	RequiresPK    bool
	RequiresOrder bool
	AcceptsOps    []Operation
	Atomicity     WriteAtomicity
	Durability    WriteDurability
}

WritePolicyCapability is what a sink must support to serve a write mode: key/order requirements, accepted operations, and atomicity.

func WriteCapabilities 

func WriteCapabilities(types ...IngestionType) []WritePolicyCapability

WriteCapabilities maps each ingestion type to its required sink capability, preserving order — the shape connector Specs advertise.

func (WritePolicyCapability) Accepts 

func (c WritePolicyCapability) Accepts(op Operation) bool

Accepts reports whether the capability admits op; an empty AcceptsOps admits everything.

type WriteReceipt 

type WriteReceipt struct {
	URI   string
	Bytes int64
	Rows  int
	// WriteCRC is the sink's independently recomputed checksum over the borrowed
	// in-memory batch, immediately before its write boundary.
	WriteCRC uint32
	// EncodedCRC, when non-nil, is the checksum a sink verified over the exact
	// serialized bytes handed to its transport. Sinks advertising
	// SinkCapabilities.EncodedIntegrity must return it or the run fails.
	EncodedCRC *uint32
	Checkpoint *CheckpointData
}

WriteReceipt is what a sink returns per Apply: where the data landed, how much, and the write-side CRC for integrity comparison.

Source Files

View all Source files

Directories

Path Synopsis
api
ingestion/v1
ingestion/v1/ingestionv1connect
metrics/v1
metrics/v1/metricsv1connect
Package app is the public composition root for a Filament ingestion service.
 Package app is the public composition root for a Filament ingestion service.
Package arrowbatch owns Arrow row batches, builders, schema mapping, and in-memory integrity checks.
 Package arrowbatch owns Arrow row batches, builders, schema mapping, and in-memory integrity checks.
Package checkpoint encodes and decodes resumable-read checkpoints: keyset (key-order pages), bitmap (unordered sub-range), and ctid (physical block range).
 Package checkpoint encodes and decodes resumable-read checkpoints: keyset (key-order pages), bitmap (unordered sub-range), and ctid (physical block range).
connectors
clickhouse
Package clickhouse wires the ClickHouse sink onto the default registry.
 Package clickhouse wires the ClickHouse sink onto the default registry.
clickhouse/sink
Package clickhouse implements a typed ClickHouse sink.
 Package clickhouse implements a typed ClickHouse sink.
http
Package httpapi is the generic Tier-2 HTTP connector.
 Package httpapi is the generic Tier-2 HTTP connector.
http/auth
Package auth provides a pluggable Authenticator interface for the HTTP connector.
 Package auth provides a pluggable Authenticator interface for the HTTP connector.
http/errs
Package errs holds the httpapi connector's error taxonomy.
 Package errs holds the httpapi connector's error taxonomy.
http/incremental
Package incremental implements watermark-based incremental extraction.
 Package incremental implements watermark-based incremental extraction.
http/internal/atomicwatermark
Package atomicwatermark provides a lock-free max-watermark cell for incremental extraction.
 Package atomicwatermark provides a lock-free max-watermark cell for incremental extraction.
http/internal/paths
Package paths is the canonical dot-path lookup for the httpapi connector.
 Package paths is the canonical dot-path lookup for the httpapi connector.
http/internal/scalar
Package scalar provides the canonical conversion of HTTP connector scalar values to strings.
 Package scalar provides the canonical conversion of HTTP connector scalar values to strings.
http/manifest
grammar.go enforces the v1 manifest meta-grammar — the shape every manifest.yaml must conform to (object structure, field types, enum values).
 grammar.go enforces the v1 manifest meta-grammar — the shape every manifest.yaml must conform to (object structure, field types, enum values).
http/obs
Package obs holds constructors for non-nil observability dependencies.
 Package obs holds constructors for non-nil observability dependencies.
http/pagination
Package pagination implements pluggable pagination strategies for the HTTP connector.
 Package pagination implements pluggable pagination strategies for the HTTP connector.
http/request
Package request builds outgoing HTTP requests from a v1 manifest.Resource.
 Package request builds outgoing HTTP requests from a v1 manifest.Resource.
http/response
Package response decodes API responses into records, captures parent fields, and surfaces 200-wrapped errors.
 Package response decodes API responses into records, captures parent fields, and surfaces 200-wrapped errors.
http/stream
Package stream implements long-lived streaming response readers for the HTTP connector.
 Package stream implements long-lived streaming response readers for the HTTP connector.
http/template
Package template renders `{{ scope.key }}` placeholders in manifest strings using a typed Scope assembled at request-build time.
 Package template renders `{{ scope.key }}` placeholders in manifest strings using a typed Scope assembled at request-build time.
internal/dbconfig
Package dbconfig contains the connection-input convention shared by database connectors.
 Package dbconfig contains the connection-input convention shared by database connectors.
internal/ndjson
Package ndjson renders Arrow rows as JSON objects, one per line, for sinks that write text such as stdout and object storage.
 Package ndjson renders Arrow rows as JSON objects, one per line, for sinks that write text such as stdout and object storage.
mysql
Package mysql wires the mysql source and sink onto the default registry.
 Package mysql wires the mysql source and sink onto the default registry.
mysql/internal/connection
mysql/sink
mysql/source
Package mysql implements the filament.Source interface as a full-snapshot, incremental-watermark, and CDC reader for MySQL (8.0+).
 Package mysql implements the filament.Source interface as a full-snapshot, incremental-watermark, and CDC reader for MySQL (8.0+).
postgres
Package postgres wires the postgres source and sink onto the default registry.
 Package postgres wires the postgres source and sink onto the default registry.
postgres/internal/connection
postgres/sink
postgres/source
Package postgres implements resumable full-snapshot reads from a PostgreSQL source.
 Package postgres implements resumable full-snapshot reads from a PostgreSQL source.
sample
Package sample implements the filament.Source interface as a synthetic generator (dev/test).
 Package sample implements the filament.Source interface as a synthetic generator (dev/test).
stdout
Package stdout implements the filament.Sink interface by printing batches as NDJSON.
 Package stdout implements the filament.Sink interface by printing batches as NDJSON.
datastore
memory
Package memory implements filament.DataStore and filament.ScheduleStore with in-process maps, mirroring the semantics of datastore/postgres.
 Package memory implements filament.DataStore and filament.ScheduleStore with in-process maps, mirroring the semantics of datastore/postgres.
postgres
Package postgres implements filament.DataStore, filament.ScheduleStore and a pipeline store on PostgreSQL via pgx.
 Package postgres implements filament.DataStore, filament.ScheduleStore and a pipeline store on PostgreSQL via pgx.
postgres/metrics
Package metrics implements filament.MetricsStore over the same Postgres runs table datastore/postgres.Store persists to.
 Package metrics implements filament.MetricsStore over the same Postgres runs table datastore/postgres.Store persists to.
postgres/sqlcgen
Package eventbus is a transport-only, payload-agnostic event bus shared across the system.
 Package eventbus is a transport-only, payload-agnostic event bus shared across the system.
host
Package host runs modules against an eventbus.Bus: it subscribes each Runnable's declared patterns and pumps deliveries to handlers with at-least-once ack/nak.
 Package host runs modules against an eventbus.Bus: it subscribes each Runnable's declared patterns and pumps deliveries to handlers with at-least-once ack/nak.
inproc
Package inproc implements the in-memory eventbus provider: channel-backed delivery, ack/nak redelivery, and replay from an in-memory log.
 Package inproc implements the in-memory eventbus provider: channel-backed delivery, ack/nak redelivery, and replay from an in-memory log.
nats
Package nats implements eventbus.Bus over NATS JetStream.
 Package nats implements eventbus.Bus over NATS JetStream.
Package events defines filament's typed event catalog: every fact the system emits, its payload type, and the machinery facts travel through — subjects (subject.go), wire framing (wire.go), emit paths (emit.go), and typed consumption (mux.go).
 Package events defines filament's typed event catalog: every fact the system emits, its payload type, and the machinery facts travel through — subjects (subject.go), wire framing (wire.go), emit paths (emit.go), and typed consumption (mux.go).
internal
arrowtext
compile
Package compile turns a stored pipeline into per-route run requests.
 Package compile turns a stored pipeline into per-route run requests.
cron
Package cron parses standard 5-field cron expressions and computes the next fire time, with no external dependencies.
 Package cron parses standard 5-field cron expressions and computes the next fire time, with no external dependencies.
modules/dispatch/inproc
Package inproc is the in-process dispatch backend: it mounts the engine module so requested runs execute inside the calling process.
 Package inproc is the in-process dispatch backend: it mounts the engine module so requested runs execute inside the calling process.
modules/dispatch/k8s
Package k8s is the Kubernetes dispatch backend: it turns run.requested facts into worker Jobs.
 Package k8s is the Kubernetes dispatch backend: it turns run.requested facts into worker Jobs.
modules/engine
Package engine is the in-process dispatch backend: it reacts to a run.requested fact and executes the run inside the calling process.
 Package engine is the in-process dispatch backend: it reacts to a run.requested fact and executes the run inside the calling process.
modules/orchestrator
Package orchestrator owns run intake and lifecycle.
 Package orchestrator owns run intake and lifecycle.
modules/reaper
Package reaper is the zombie-run janitor: on each tick it finds runs stuck in Requested or Running with no write in the staleness window — a worker that never got admitted, a lost pod, an OOM-killed worker, a crashed engine — and emits run.failed so the tracker folds them terminal.
 Package reaper is the zombie-run janitor: on each tick it finds runs stuck in Requested or Running with no write in the staleness window — a worker that never got admitted, a lost pod, an OOM-killed worker, a crashed engine — and emits run.failed so the tracker folds them terminal.
modules/scheduler
Package scheduler is the cron firing module: on each tick it claims due schedules, compiles their pipelines into runs, and submits them.
 Package scheduler is the cron firing module: on each tick it claims due schedules, compiles their pipelines into runs, and submits them.
modules/stream
Package stream is the client-facing read side of the plane.
 Package stream is the client-facing read side of the plane.
modules/tracker
Package tracker folds bus facts into DataStore run/resource/checkpoint state.
 Package tracker folds bus facts into DataStore run/resource/checkpoint state.
naming
Package naming normalizes display names into destination identifiers.
 Package naming normalizes display names into destination identifiers.
runs
Package runs holds the run-intake primitive shared by the control-plane modules.
 Package runs holds the run-intake primitive shared by the control-plane modules.
schedule
Package schedule contains schedule-domain validation and timing calculations.
 Package schedule contains schedule-domain validation and timing calculations.
usage
Package usage samples this process's cgroup v2 CPU and memory accounting.
 Package usage samples this process's cgroup v2 CPU and memory accounting.
Package module defines ingestion's Module contract and its dependency wiring.
 Package module defines ingestion's Module contract and its dependency wiring.
Package pipeline implements the builder → writer → integrity extraction loop.
 Package pipeline implements the builder → writer → integrity extraction loop.
Package profiling provides drop-in helpers for live and file-based Go profiling.
 Package profiling provides drop-in helpers for live and file-based Go profiling.
Package registry maps provider names to Source and Sink factories for per-run resolution.
 Package registry maps provider names to Source and Sink factories for per-run resolution.
Package rowmodel defines the connector-neutral row schema and per-row metadata.
 Package rowmodel defines the connector-neutral row schema and per-row metadata.
Package runner executes one Filament ingestion run.
 Package runner executes one Filament ingestion run.
secret
env
Package env is an filament.Secrets provider backed by process environment variables.
 Package env is an filament.Secrets provider backed by process environment variables.
postgres
Package postgres implements filament.Secrets using the shared Filament Postgres schema.
 Package postgres implements filament.Secrets using the shared Filament Postgres schema.
Package ui serves the Filament web UI.
 Package ui serves the Filament web UI.

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.