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

filament

package module
v0.24.0-dev.3 Latest Latest
Warning

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

 Go to latest Published: Oct 19, 2026 License: Apache-2.0 Imports: 19 Imported by: 0

README

Release Tests Ask DeepWiki 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.

Install

curl -fsSL https://getgalaxy.io/filament/install | sh

On macOS via Homebrew

brew install galaxy-io/tap/filament

Prebuilt binaries for Linux and macOS are on the releases page.

Upgrade a direct installation with filament upgrade. For Homebrew installations, use brew upgrade galaxy-io/tap/filament.

Getting started

To run Filament locally, install Go, Docker, just, Node.js, and pnpm. On macOS, the repository's Brewfile installs the toolchain:

brew bundle

For Podman users, set CONTAINER_ENGINE=podman to use Podman with the commands below. See engine configuration.

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

Deploying Filament

Kubernetes with Helm is the recommended way to run Filament in production. See the Kubernetes deployment guide for installation instructions and the Helm chart README for all supported values.

For a simpler hosted deployment without Kubernetes, deploy with 1 click:

Deploy on Railway Deploy to Render Deploy to DO

Read the Railway, Render, and DigitalOcean guides for the deployed resources, runtime behavior, and production checklist.

[!WARNING] The DigitalOcean deployment uses filament/standalone, an all-in-one image that includes the control-plane, server, and SQLite datastore.

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 (
	InputRows                     InputSemantics    = "rows"
	InputChanges                  InputSemantics    = "changes"
	InputMessages                 InputSemantics    = "messages"
	OrderingNone                  Ordering          = "none"
	OrderingPartition             Ordering          = "partition"
	OrderingTransaction           Ordering          = "transaction"
	OrderingGlobalStrict          Ordering          = "global_strict"
	DeliveryReplayableAtLeastOnce DeliveryGuarantee = "replayable_at_least_once"
	DeliveryBestEffort            DeliveryGuarantee = "best_effort"
)

Source input, ordering, and delivery capabilities are negotiated independently.

View Source 
const (
	PositionIncomparable = rowmodel.PositionIncomparable
	PositionEqual        = rowmodel.PositionEqual
	PositionBefore       = rowmodel.PositionBefore
	PositionAfter        = rowmodel.PositionAfter
	ControlUnspecified   = rowmodel.ControlUnspecified
	TxnBegin             = rowmodel.TxnBegin
	TxnEnd               = rowmodel.TxnEnd
	ProgressBoundary     = rowmodel.ProgressBoundary
	MemberActivated      = rowmodel.MemberActivated
)

Neutral comparison outcomes and control kinds retain their rowmodel meanings.

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 EpochCertificateFormatVersion = 1

EpochCertificateFormatVersion identifies the canonical certificate format.

View Source 
const EpochMessageSchemaFormatVersion = 2

EpochMessageSchemaFormatVersion binds accepted message layouts to progress.

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 (
	ErrEpochMismatch        = errors.New("stream: epoch mismatch")
	ErrFenced               = errors.New("stream: fenced")
	ErrLeaseExpired         = errors.New("stream: lease expired")
	ErrEpochConflict        = errors.New("stream: epoch conflict")
	ErrPositionRegression   = errors.New("stream: position regression")
	ErrPositionIncomparable = rowmodel.ErrPositionIncomparable
	ErrIncompleteCoverage   = errors.New("stream: incomplete or mixed coverage")
	ErrMembershipChanged    = errors.New("stream: membership changed")
)

Stream error categories distinguish ownership, epoch binding, progress, and coverage failures.

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 ErrConfigure = errors.New("connector rejected configuration")

ErrConfigure reports a connector rejecting its configuration, either when validating it or when connecting with it. Workers return it as a ConfigureError so the connector's own reason survives intact.

View Source 
var ErrContinuousDisabled = errors.New("continuous execution is disabled")

ErrContinuousDisabled reports that the continuous runtime is unavailable.

View Source 
var ErrRunOverlap = errors.New("run overlaps an active replication route")

ErrRunOverlap means another requested, running, or paused run already owns the same replication route.

View Source 
var ErrTakeoverBlocked = errors.New("stream: takeover blocked")

ErrTakeoverBlocked reports that predecessor quiescence has not been proven.

View Source 
var ErrUnsupported = errors.New("unsupported by connector")

ErrUnsupported reports an optional contract the connector does not offer, such as discovery or a connectivity probe.

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 ValidateContinuous added in v0.18.0 

func ValidateContinuous(source ConnectorSpec, sink SinkSpec) error

ValidateContinuous checks that both specs declare streaming support and that at least one declared streaming write policy is compatible with the source. Registration guarantees a declaring connector implements the native contracts.

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 declares a write policy that can serve the ingestion type. Undeclared policies are rejected.

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 {
	// Epoch is nil for bounded writes. Native sinks reject writes that do not
	// match the currently open epoch; this is not destination fencing.
	Epoch  *EpochRef
	Policy WritePolicy
}

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

type Attempt added in v0.6.0 

type Attempt struct {
	// Claimed records that a worker began executing this attempt.
	Claimed              bool
	Spec                 RunSpec
	Revision             int64
	Termination          AttemptTermination
	Reason               string
	Lease                LeaseToken
	StartedAt, ExpiresAt time.Time
	EndedAt              *time.Time
}

Attempt describes one admitted worker lifetime and its lease interval.

type AttemptRef added in v0.6.0 

type AttemptRef struct {
	RunID                 RunID
	ExecutionID, StreamID string
	Generation, Token     int64
}

AttemptRef identifies a stream worker and its admitted ownership token.

func (AttemptRef) Validate added in v0.6.0 

func (a AttemptRef) Validate() error

Validate requires complete attempt identity and positive generation and token values.

type AttemptTermination added in v0.6.0 

type AttemptTermination string

AttemptTermination describes how a worker ended. Successful session closure, not inferred workload disappearance, is necessary to record clean termination. Detailed takeover proofs land with their owner. 

const (
	AttemptClean    AttemptTermination = "clean"
	AttemptReaped   AttemptTermination = "reaped"
	AttemptUnproven AttemptTermination = "unproven"
)

Termination outcomes distinguish proven closure from inferred or unproven shutdown.

type Boundary added in v0.6.0 

type Boundary struct {
	MaxRecords      int
	MaxBytes        int64
	MaxWait, MaxAge time.Duration
}

Boundary defines soft record, byte, idle, and age targets for one epoch. Complete source transactions may exceed these targets.

type Catalog added in v0.18.0 

type Catalog struct {
	Sources []ConnectorSpec
	Sinks   []SinkSpec
}

Catalog is every registered connector spec. Sources include alias entries with AliasTarget set so readers can tell them from concrete registrations.

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 CodecResolver added in v0.6.0 

type CodecResolver = rowmodel.CodecResolver

CodecResolver finds a position codec by name and version.

type CommittedEpoch added in v0.6.0 

type CommittedEpoch struct {
	Certificate EpochCertificate
	CommittedAt time.Time
}

CommittedEpoch pairs a persisted certificate with its datastore-assigned commit time.

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 ConfigureError added in v0.18.0 

type ConfigureError struct{ Reason error }

ConfigureError is ErrConfigure carrying the connector's reason verbatim.

func (*ConfigureError) Error added in v0.18.0 

func (e *ConfigureError) Error() string

Error is the connector's reason, unprefixed.

func (*ConfigureError) Is added in v0.18.0 

func (e *ConfigureError) Is(target error) bool

Is matches ErrConfigure so callers can branch without knowing the type.

func (*ConfigureError) Unwrap added in v0.18.0 

func (e *ConfigureError) Unwrap() error

Unwrap exposes the connector's reason to errors.Is and errors.As.

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 ConnectorRef added in v0.18.0 

type ConnectorRef struct {
	Kind ConnectorKind
	Name string
}

ConnectorRef names one registered connector by kind and name.

type ConnectorSpec 

type ConnectorSpec struct {
	// AliasTarget is the concrete registration selected by a catalog alias.
	// The registry sets it; concrete catalog entries leave it empty.
	AliasTarget    string
	Stream         *StreamCapabilities
	Name           string
	DisplayName    string
	Description    string
	DarkLogoURL    string
	LightLogoURL   string
	Version        string
	APIVersion     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 ContinuousRunStore added in v0.6.0 

type ContinuousRunStore interface {
	DataStore
	StreamRuntimeStore
	ReplicationStreamRunStore
	ClaimStreamAttempt(context.Context, LeaseToken) error
	RetireUnclaimedAttempt(context.Context, LeaseToken) error
	PendingStreamRuns(context.Context, string, int) ([]RunState, error)
	StreamProgress(context.Context, TenantID, RunID) (int64, int64, time.Time, error)
}

ContinuousRunStore adds durable run admission and dispatch to the epoch store. Implementations serialize claims and ownership independently of connector type.

type ContinuousWritePlan added in v0.6.0 

type ContinuousWritePlan struct {
	Policy   WritePolicy
	Ordering Ordering
}

ContinuousWritePlan is the pure compatibility decision shared by admission, API validation, and worker startup. Keys are bound from live schemas later.

func PlanContinuousWrite added in v0.6.0 

func PlanContinuousWrite(source ConnectorSpec, sink SinkSpec, mode WriteMode) (ContinuousWritePlan, error)

PlanContinuousWrite selects a sink's explicit epoch policy. Runtime limits belong here rather than being inferred from connector names or bounded modes.

type Control added in v0.6.0 

type Control = rowmodel.Control

Control carries standalone source boundary evidence.

type ControlKind added in v0.6.0 

type ControlKind = rowmodel.ControlKind

ControlKind distinguishes transaction, progress, and membership markers.

type Counter 

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

Counter is a monotonically increasing metric.

type Coverage added in v0.6.0 

type Coverage struct {
	Positions DomainPositions
	Claims    []InboxClaimRef
}

Coverage holds either candidate domain progress or exact inbox claims, never both.

func NewClaimCoverage added in v0.6.0 

func NewClaimCoverage(claims []InboxClaimRef) (Coverage, error)

NewClaimCoverage validates and copies exact claims; it does not verify live claim ownership or settle inbox rows.

func NewPositionCoverage added in v0.6.0 

func NewPositionCoverage(positions DomainPositions) (Coverage, error)

NewPositionCoverage validates and takes an independent copy of candidate positions. Receiving boundaries must still validate exported fields.

func (Coverage) Canonicalize added in v0.6.0 

func (c Coverage) Canonicalize(r CodecResolver) (Coverage, error)

Canonicalize validates the representation, copies it, and normalizes positions and claim order.

func (Coverage) Clone added in v0.6.0 

func (c Coverage) Clone() Coverage

Clone returns independently owned positions and claims.

func (Coverage) ValidateRepresentation added in v0.6.0 

func (c Coverage) ValidateRepresentation() error

ValidateRepresentation checks representation only; it cannot prove contiguous processing, transaction completion, claim authority or pipeline durability. Empty coverage is rejected until a coordinator defines an explicit idle/no-progress policy.

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
	// ResolveTenant maps a provider's organization id onto filament's tenant
	// id, minting the row on first sight and refreshing the display name.
	ResolveTenant(ctx context.Context, externalID, name string) (TenantID, error)
	// EnsureUser creates the user row for a provider subject within a tenant,
	// or touches it when it exists, and returns filament's id for it.
	EnsureUser(ctx context.Context, tenant TenantID, externalID string) (UserID, 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, tenant TenantID, 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, tenant TenantID, id RunID) error

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

	SaveCheckpoint(ctx context.Context, tenant TenantID, id RunID, cp Checkpoint) error
	LoadCheckpoint(ctx context.Context, tenant TenantID, id RunID, resource string) (Checkpoint, error)
	SaveResourceCheckpoint(ctx context.Context, tenant TenantID, state ResourceCheckpointState) error
	LoadResourceCheckpoint(ctx context.Context, tenant TenantID, key ResourceCheckpointKey) (ResourceCheckpointState, error)
	ListResourceCheckpoints(ctx context.Context, tenant TenantID, route ResourceCheckpointRoute) ([]ResourceCheckpointState, error)
	DeleteResourceCheckpoint(ctx context.Context, tenant TenantID, 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, tenant TenantID, id string) (Connection, error)
	ListConnections(ctx context.Context, f ConnectionFilter) ([]Connection, int, error)
	DeleteConnection(ctx context.Context, tenant TenantID, id string) error

	CreatePipeline(ctx context.Context, p *ingestionv1.Pipeline) (*ingestionv1.Pipeline, error)
	CreatePipelineVersion(ctx context.Context, tenant TenantID, 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, tenant TenantID, id string) (*ingestionv1.Pipeline, error)
	LoadPipelineVersion(ctx context.Context, tenant TenantID, 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, tenant TenantID, id string) error

	// CreateNotifier adds a pipeline notification rule.
	CreateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)
	// LoadNotifier includes deleted rules. Check DeletedAt before using one.
	LoadNotifier(ctx context.Context, tenant TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)
	// ListNotifiers returns matching rules ordered by ID.
	ListNotifiers(ctx context.Context, tenant TenantID, pipelineID string, includeDeleted bool) ([]*ingestionv1.Notifier, error)
	// UpdateNotifier replaces a rule's settings. Its ID, tenant, pipeline, and
	// type cannot change. Clean up replaced secrets only after success.
	UpdateNotifier(ctx context.Context, n *ingestionv1.Notifier) (*ingestionv1.Notifier, error)
	// DeleteNotifier marks a rule deleted and returns it, including secret
	// references for cleanup.
	DeleteNotifier(ctx context.Context, tenant TenantID, pipelineID, id string) (*ingestionv1.Notifier, error)

	Name() string
}

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

type DeliveryGuarantee added in v0.6.0 

type DeliveryGuarantee string

DeliveryGuarantee describes input replay and loss semantics.

type DesiredStateChange added in v0.6.0 

type DesiredStateChange struct {
	StreamStateRequest
	ExpectedRevision int64
	Desired          StreamDesiredState
}

DesiredStateChange requests an intent update guarded by the expected revision.

type DestinationResourceNamer added in v0.6.0 

type DestinationResourceNamer interface {
	DestinationResource(resource string) string
}

DestinationResourceNamer optionally supplies a default output name without schema discovery. Implementations must be deterministic and perform no I/O. An empty result leaves the resource name unchanged. Pipeline labels override it.

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 DomainKey added in v0.6.0 

type DomainKey = rowmodel.DomainKey

DomainKey identifies an ordering domain within a source incarnation.

type DomainPosition added in v0.6.0 

type DomainPosition = rowmodel.DomainPosition

DomainPosition pairs a domain with its position for serialization.

type DomainPositions added in v0.6.0 

type DomainPositions = rowmodel.DomainPositions

DomainPositions maps source ordering domains to their bookmarks.

type EndAttemptRequest added in v0.6.0 

type EndAttemptRequest struct {
	Lease       LeaseToken
	Termination AttemptTermination
	Reason      string
}

EndAttemptRequest conditionally records termination under the owning lease.

type EnumOption 

type EnumOption struct {
	Value string
	Label string
}

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

type EpochCertificate added in v0.6.0 

type EpochCertificate struct {
	FormatVersion     int
	Tenant            TenantID
	Ref               EpochRef
	PipelineVersionID string
	Coverage          Coverage
	Receipts          []EpochReceipt
	Records, Bytes    int64
	MessageSchemas    map[string]schema.MessageLayout `json:",omitempty"`
}

EpochCertificate is immutable content for a fenced store transaction. Structural canonicalization does not authorize a commit; EpochCommit binds new certificates to private completed pipeline evidence.

func (EpochCertificate) CanonicalBytes added in v0.6.0 

func (c EpochCertificate) CanonicalBytes(r CodecResolver) ([]byte, error)

CanonicalBytes sorts semantically unordered domains, claims and receipts; preserves nil versus empty opaque payloads; excludes datastore timestamps. Version 2 also includes accepted message layouts. Position canonicalization is codec-owned. Receipt evidence is already encoded by the destination and is not interpreted as source progress.

func (EpochCertificate) Clone added in v0.6.0 

func (c EpochCertificate) Clone() EpochCertificate

Clone returns independently owned coverage and receipt evidence.

func (EpochCertificate) Digest added in v0.6.0 

func (c EpochCertificate) Digest(r CodecResolver) ([32]byte, error)

Digest hashes the canonical immutable certificate content.

func (EpochCertificate) ValidateMessageSchemas added in v0.23.0 

func (c EpochCertificate) ValidateMessageSchemas() error

ValidateMessageSchemas keeps v1 bytes unchanged and fails closed on missing or unsupported v2 layouts. Every resource written under v2 needs an input layout.

type EpochCommit added in v0.6.0 

type EpochCommit struct {
	Certificate EpochCertificate
	Completion  EpochCompletion
}

EpochCommit binds a certificate to source controls accepted by the pipeline. Coverage remains connector-authoritative; offsets alone cannot prove a prefix. Completion is required for new commits. Historical identical retries may use only Certificate, since their canonical content has already been certified.

func (EpochCommit) ValidateCompletion added in v0.6.0 

func (c EpochCommit) ValidateCompletion(codecs CodecResolver) error

ValidateCompletion checks exact canonical coverage and completed Apply totals. No public constructor can manufacture nonzero pipeline completion evidence.

type EpochCompletion added in v0.6.0 

type EpochCompletion = streamproof.Completion

EpochCompletion is immutable, process-local evidence from a bound and sealed pipeline. It is not proof of sink durability; the native sink supplies receipts.

type EpochKey added in v0.6.0 

type EpochKey struct {
	StreamID          string
	Generation, Epoch int64
}

EpochKey identifies a certificate across attempts within a stream generation.

type EpochLookup added in v0.6.0 

type EpochLookup struct {
	Tenant TenantID
	Key    EpochKey
}

EpochLookup identifies a historical certificate within its tenant scope.

type EpochReceipt added in v0.6.0 

type EpochReceipt struct {
	Resource    string
	Rows, Bytes int64
	WriteCRC    uint32
	Evidence    *ReceiptEvidence
}

EpochReceipt contains stable aggregate evidence; it is not a dedup guarantee. Evidence uses a destination-owned versioned encoding, immutable across retries.

type EpochRef added in v0.6.0 

type EpochRef struct {
	Attempt                   AttemptRef
	Epoch, MembershipRevision int64
}

EpochRef binds a commit cycle and membership revision to an admitted attempt.

func (EpochRef) Binding added in v0.6.0 

func (e EpochRef) Binding() string

Binding returns a deterministic identity for process-local completion evidence.

func (EpochRef) Key added in v0.6.0 

func (e EpochRef) Key() EpochKey

Key returns the attempt-independent certificate identity.

func (EpochRef) Validate added in v0.6.0 

func (e EpochRef) Validate() error

Validate checks epoch identity without verifying live ownership or durability.

func (EpochRef) ValidateApply added in v0.6.0 

func (e EpochRef) ValidateApply(opts ApplyOptions) error

ValidateApply checks explicit session binding, not destination-side fencing.

type EventIdentity added in v0.6.0 

type EventIdentity = rowmodel.EventIdentity

EventIdentity identifies a source event independently of ingestion attempts.

type ExecutionMode added in v0.6.0 

type ExecutionMode string

ExecutionMode selects bounded or continuous execution independently of input semantics. 

const (
	ExecutionBounded    ExecutionMode = "bounded"
	ExecutionContinuous ExecutionMode = "continuous"
)

Execution modes retain bounded behavior as the default.

func (ExecutionMode) Normalize added in v0.6.0 

func (e ExecutionMode) Normalize() ExecutionMode

Normalize maps the compatibility zero value to bounded execution.

func (ExecutionMode) Validate added in v0.6.0 

func (e ExecutionMode) Validate() error

Validate checks the mode, independently of runtime availability.

type ExportJobProgress added in v0.15.0 

type ExportJobProgress struct {
	CorrelationID string        `json:"correlationId"`
	Polls         int           `json:"polls"`
	Elapsed       time.Duration `json:"elapsed"`
	Resumed       bool          `json:"resumed"`
}

ExportJobProgress describes one job execution within an extraction attempt. CorrelationID is opaque and changes on resume; it is never an upstream job ID, parent key, or URL. Polls and Elapsed are local to this execution, including when Resumed is true. Ready means an artifact is available, not sink commit.

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 InboxClaimRef added in v0.6.0 

type InboxClaimRef struct {
	RowID, Owner string
	Token        int64
}

InboxClaimRef identifies an exact claim, never a sequence watermark. The enclosing certificate supplies tenant scope. No inbox infrastructure is implemented by these value types.

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 IncrementalResumePlanner added in v0.15.0 

type IncrementalResumePlanner interface {
	PlanIncrementalResume(ctx context.Context, planned, attempt map[string]Checkpoint) (map[string]Checkpoint, error)
}

IncrementalResumePlanner restores source job state from the current run without promoting it to a cross-run watermark. Opt-in plans carry replay_on_resume metadata; their rows must be replayed after interruption, even for completed jobs, because the previous sink transaction may not have committed.

type IngestionPlan 

type IngestionPlan struct {
	// IngestionTypes contains effective per-resource modes after keyless fallback.
	IngestionTypes map[string]IngestionType
	Ordering       Ordering
	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. Continuous execution negotiates native stream guarantees and epoch policies through the same entry point.

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. An unspecified write mode defaults to upsert.

func IngestionForKeys added in v0.11.0 

func IngestionForKeys(t IngestionType, keys []string) IngestionType

IngestionForKeys resolves bounded upsert to append when a resource has no key. CDC and delete policies retain their key requirements.

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 InputSemantics added in v0.6.0 

type InputSemantics string

InputSemantics describes whether input consists of rows, changes, or messages.

type InspectStatus added in v0.18.0 

type InspectStatus int

InspectStatus is the outcome of inspecting one resource. 

const (
	// InspectOK means PrimaryKey, Columns and ManagedIncremental are populated.
	InspectOK InspectStatus = iota
	// InspectUnsupported means the source cannot list columns; PrimaryKey and
	// ManagedIncremental are still populated where the source reports them.
	InspectUnsupported
	// InspectFailed means the source errored for this resource; Message says why.
	InspectFailed
)

The inspect statuses.

type Inspection added in v0.18.0 

type Inspection struct {
	Name    string
	Status  InspectStatus
	Message string

	PrimaryKey []string
	// Columns are the resource's schema fields with cursor eligibility laid
	// over them when Ranked is true; cursor columns the schema lacks follow.
	// Without a schema they are the cursor columns alone.
	Columns            []CursorColumn
	Ranked             bool
	ManagedIncremental bool
	// Schema is nil when the source provides none.
	Schema *RecordSchema
}

Inspection is what a configured source reports about one resource.

type Label 

type Label struct{ Key, Value string }

Label is a metric label.

type LeaseToken added in v0.6.0 

type LeaseToken struct {
	Tenant  TenantID
	Attempt AttemptRef
}

LeaseToken binds complete attempt ownership to its tenant.

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 ManagedIncrementalSource added in v0.11.0 

type ManagedIncrementalSource interface {
	ManagedIncremental(resource string) bool
}

ManagedIncrementalSource uses source-owned state rather than a row column.

type MessageSchemaRecordSink added in v0.23.0 

type MessageSchemaRecordSink interface {
	StreamRecordSink
	MessageBuilder(context.Context, string, rowmodel.Schema, schema.MessageLayout) (RowWriter, error)
}

MessageSchemaRecordSink admits a message layout before any rows in an epoch. The returned builder replaces the resource's previous builder when needed.

type MessageSchemaSource added in v0.23.0 

type MessageSchemaSource interface {
	StreamSource
	OpenMessageStream(context.Context, StreamOpenOpts) (StreamSession, error)
}

MessageSchemaSource supports batch probing and restoration of accepted layouts. The runner calls this method only when the destination enables schema policy. Implementations must validate the whole batch before emitting any rows.

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 Ordering added in v0.6.0 

type Ordering string

Ordering identifies a source guarantee or destination ordering requirement.

func PlanContinuousRun added in v0.6.0 

func PlanContinuousRun(source Source, sink Sink, spec RunSpec) (map[string]WritePolicy, Ordering, error)

PlanContinuousRun requires an explicit write policy for every resource, either resource-specific or supplied as the default policy. IngestionTypes are not used to infer continuous policies. The sink declaration is authoritative; stale or weakened submitted capabilities are rejected. It is safe to repeat this before opening I/O.

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 {
	Tenant     TenantID
	PipelineID string
	ListOptions
}

PipelineVersionFilter narrows a pipeline version listing.

type Position added in v0.6.0 

type Position = rowmodel.Position

Position stores codec-encoded source progress.

type PositionCodec added in v0.6.0 

type PositionCodec = rowmodel.PositionCodec

PositionCodec defines validation, comparison, and canonicalization for progress.

type PositionOrder added in v0.6.0 

type PositionOrder = rowmodel.PositionOrder

PositionOrder reports codec-defined ordering between compatible positions.

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 ReceiptEvidence added in v0.6.0 

type ReceiptEvidence struct {
	Format  string
	Version int
	Payload []byte
}

ReceiptEvidence is destination-owned evidence, not comparable source progress. Nil evidence means absent. A present value must name its format and have a nonnegative version selected by that destination format.

func (*ReceiptEvidence) Clone added in v0.6.0 

func (e *ReceiptEvidence) Clone() *ReceiptEvidence

Clone copies evidence bytes and preserves absent evidence and nil-versus-empty payloads.

func (*ReceiptEvidence) Validate added in v0.6.0 

func (e *ReceiptEvidence) Validate() error

Validate checks a present evidence format and version; nil means absent.

type ReconcilePage added in v0.6.0 

type ReconcilePage struct {
	States []StreamState
	Next   string
}

ReconcilePage returns stream states and an opaque continuation cursor.

type ReconcileQuery added in v0.6.0 

type ReconcileQuery struct {
	Tenant TenantID
	After  string
	Limit  int
}

ReconcileQuery selects a bounded page of tenant-scoped reconciliation candidates.

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 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 ReplicationFor added in v0.18.0 

func ReplicationFor(spec ConnectorSpec, cfg Config) ReplicationMode

ReplicationFor resolves a connection's replication mode from the source's declared policies and the connection config. CDC requires both: a CDC source policy on the spec and replication=cdc in the config.

type ReplicationStream added in v0.0.27 

type ReplicationStream struct {
	ID                           string
	Tenant                       TenantID
	PipelineID                   string
	Route                        string
	Generation                   int64
	SourceConnectionID           string
	SinkConnectionID             string
	ConsumerName                 string
	ConsumerConfig               map[string]any
	ContinuityFingerprint        string
	Status                       ReplicationStreamStatus
	CreatedFromPipelineVersionID string
	Error                        string
	RetiredAt                    *time.Time
	CreatedAt                    time.Time
	UpdatedAt                    time.Time
}

ReplicationStream is the continuity identity shared by compatible versions of one pipeline route. ConsumerName is connector-specific: a PostgreSQL slot, Kafka consumer group, or NATS durable consumer.

type ReplicationStreamCleaner added in v0.0.27 

type ReplicationStreamCleaner interface {
	CleanupReplicationStream(ctx context.Context, stream ReplicationStream) error
}

ReplicationStreamCleaner releases a retired connector-side consumer after a successor generation has committed successfully. Implementations must treat an already-absent consumer as success.

type ReplicationStreamPlan added in v0.0.27 

type ReplicationStreamPlan struct {
	// Resources is the connector-resolved fixed membership for continuous admission.
	// Bounded CDC planning may leave it empty.
	Resources        []string
	ConsumerName     string
	ConsumerConfig   map[string]any
	ContinuityConfig map[string]any
}

ReplicationStreamPlan describes the connector-specific external consumer and the normalized config fields that determine whether its continuity is reusable.

type ReplicationStreamPlanner added in v0.0.27 

type ReplicationStreamPlanner interface {
	PlanReplicationStream(request ReplicationStreamPlanningRequest) (ReplicationStreamPlan, error)
	BindReplicationStream(config map[string]any, stream ReplicationStream) (map[string]any, error)
}

ReplicationStreamPlanner lets a streaming source own concepts such as a PostgreSQL slot, Kafka consumer group, or NATS durable consumer. Binding applies an already-resolved stream to the source's runtime configuration.

type ReplicationStreamPlanningRequest added in v0.0.27 

type ReplicationStreamPlanningRequest struct {
	// SourceConnectionID is the stable Filament connection identity, independent of runs.
	SourceConnectionID  string
	ReplicationStreamID string
	Config              Config
	// Resources fixes continuous membership. Empty asks the connector for its default.
	Resources []string
}

ReplicationStreamPlanningRequest is the connector-owned input for planning an independently advancing source consumer.

type ReplicationStreamResource added in v0.0.27 

type ReplicationStreamResource struct {
	ID                  string
	ReplicationStreamID string
	Tenant              TenantID
	Resource            string
	Status              ReplicationStreamResourceStatus
	BootstrapMode       string
	BootstrapConfig     map[string]any
	SchemaFingerprint   string
	BootstrapRun        RunID
	BootstrapStartedAt  *time.Time
	ActivatedAt         *time.Time
	RetiredAt           *time.Time
	Error               string
	CreatedAt           time.Time
	UpdatedAt           time.Time
}

ReplicationStreamResource is one resource's membership and bootstrap state within a replication stream.

type ReplicationStreamResourceStatus added in v0.0.27 

type ReplicationStreamResourceStatus int16

ReplicationStreamResourceStatus records whether one table/topic is waiting for bootstrap, participating in the stream, or retired from its retention floor. Checkpoint presence is deliberately separate from membership. 

const (
	// ReplicationStreamResourcePending indicates that the resource is waiting to bootstrap.
	// Its ordinal is persisted; map it explicitly if exposed by protobuf.
	ReplicationStreamResourcePending ReplicationStreamResourceStatus = iota
	// ReplicationStreamResourceBootstrapping indicates that the resource is being bootstrapped.
	ReplicationStreamResourceBootstrapping
	// ReplicationStreamResourceActive indicates that the resource participates in the stream.
	ReplicationStreamResourceActive
	// ReplicationStreamResourceRetired indicates that the resource no longer participates in the stream.
	ReplicationStreamResourceRetired
	// ReplicationStreamResourceError indicates that the resource could not join or advance with the stream.
	ReplicationStreamResourceError
)

type ReplicationStreamRunStore added in v0.0.27 

type ReplicationStreamRunStore interface {
	CreateRunWithReplicationStream(ctx context.Context, state RunState, desired ReplicationStream) error
}

ReplicationStreamRunStore admits a run and resolves its desired replication stream in one transaction. Keeping this separate from CreateRun prevents a compile or a rejected overlapping run from retiring the stream that an existing run still owns.

type ReplicationStreamStatus added in v0.0.27 

type ReplicationStreamStatus int16

ReplicationStreamStatus is the lifecycle of one independently advancing source consumer. An active pipeline route has exactly one active generation. 

const (
	// ReplicationStreamActive indicates that the stream is advancing normally.
	// Its ordinal is persisted; map it explicitly if exposed by protobuf.
	ReplicationStreamActive ReplicationStreamStatus = iota
	// ReplicationStreamRetired indicates that the stream no longer participates in replication.
	ReplicationStreamRetired
	// ReplicationStreamError indicates that the stream cannot currently advance.
	ReplicationStreamError
)

type ReplicationStreamStore added in v0.0.27 

type ReplicationStreamStore interface {
	LoadReplicationStream(ctx context.Context, id string) (ReplicationStream, error)
	ListRetiredReplicationStreams(ctx context.Context, pipelineID, route string) ([]ReplicationStream, error)
	MarkReplicationStreamCleaned(ctx context.Context, id string) error
	ReconcileReplicationStreamResources(ctx context.Context, streamID string, tenant TenantID, resources []string, bootstrapMode string) ([]ReplicationStreamResource, error)
	ListReplicationStreamResources(ctx context.Context, streamID string) ([]ReplicationStreamResource, error)
}

ReplicationStreamStore is the optional durable stream capability used by CDC and event-stream routes. It is separate from DataStore so lightweight stores do not need to model external consumer lifecycles.

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
	// ReplicationStreamID replaces pipeline-version identity for durable stream
	// checkpoints. It is empty for version-scoped incremental checkpoints.
	ReplicationStreamID 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
	// ReplicationStreamID selects the active resources of one durable stream.
	// It is empty for version-scoped incremental checkpoint listings.
	ReplicationStreamID 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 ResourceDomainRef added in v0.6.0 

type ResourceDomainRef = rowmodel.ResourceDomainRef

ResourceDomainRef relates a resource to a shared source ordering domain.

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 scopes the listing. Empty means every tenant: system sweeps
	// like the reaper run unscoped; API paths always set it.
	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 {
	Execution           ExecutionMode `json:",omitempty"`
	EpochBoundary       *Boundary     `json:",omitempty"`
	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 {
	// WritePolicies preserves continuous admission policies and destination-name
	// overrides for bounded runs. Runtime negotiation binds bounded capabilities.
	WritePolicies      map[string]WritePolicy `json:",omitempty"`
	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
	// ReplicationStream identifies CDC/event-stream progress independently of
	// the immutable pipeline version. Generation fences stale route versions.
	ReplicationStream *StreamRef `json:",omitempty"`
	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"`
	// Transform is the route's yaml transform definition, empty when none.
	Transform string `json:",omitempty"`
}

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 {
	// StreamAttempt is authoritative for continuous execution. Existing dispatch
	// fields, when populated, must agree with the admitted attempt.
	StreamAttempt *AttemptRef `json:",omitempty"`
	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
	ReplicationStream  *StreamRef
	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
	// Transform is the yaml transform definition applied between source and
	// sink; empty when the route has no transform node.
	Transform string
}

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.

func (RunSpec) ValidateStreamAttempt added in v0.6.0 

func (s RunSpec) ValidateStreamAttempt() error

ValidateStreamAttempt checks admitted continuous ownership against duplicated run/dispatch identities. It does not grant authority or check runtime support. Bounded specifications do not interpret this optional continuous context.

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.

func (RunState) ExecutionSpec added in v0.6.0 

func (s RunState) ExecutionSpec() RunSpec

ExecutionSpec snapshots persisted execution settings without resolving secrets.

func (RunState) StreamStateRequest added in v0.6.0 

func (s RunState) StreamStateRequest() (StreamStateRequest, error)

StreamStateRequest identifies the admitted stream for this run.

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 RunSubmission added in v0.0.27 

type RunSubmission struct {
	Request                  RunRequest
	DesiredReplicationStream *ReplicationStream `json:"-"`
}

RunSubmission carries the persisted request and any transient state that must be resolved atomically while admitting it.

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,
		tenant TenantID,
		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, tenant TenantID, id ScheduleID) (ScheduleState, error)
	LoadPipelineSchedule(ctx context.Context, tenant TenantID, pipelineID string) (ScheduleState, error)
	ListSchedules(ctx context.Context, f ScheduleFilter) ([]ScheduleState, error)
	DeleteSchedule(ctx context.Context, tenant TenantID, id ScheduleID) error
	ClaimDue(ctx context.Context, now time.Time, limit int) ([]ScheduleState, error) // SELECT … FOR UPDATE SKIP LOCKED
	ReleaseScheduleClaim(ctx context.Context, tenant TenantID, 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 SchemaPolicySink added in v0.23.0 

type SchemaPolicySink interface {
	Schematized
	SchemaPolicy() schema.Policy
}

SchemaPolicySink exposes the sink's configured migration rules after Open.

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. The runner supplies destination resource names consistently to Open, EnsureSchema, and Apply.

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 {
	Stream        *StreamingSinkCapabilities
	Transactional 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
	// PreferredBatchBytes is the sink's preferred approximate unencoded value
	// bytes per Apply, used when the run does not set Options.BatchMaxBytes.
	// 0 leaves batches unbounded by bytes.
	PreferredBatchBytes int64
	// PreferredFlushInterval is used when the pipeline does not explicitly
	// choose a flush interval. Zero uses the engine default.
	PreferredFlushInterval time.Duration
}

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)
	Spec(name string) (SinkSpec, 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 SnapshotMode added in v0.2.0 

type SnapshotMode string

SnapshotMode is whether a CDC resource without a checkpoint is read in full before its change stream starts. 

const (
	SnapshotInitial SnapshotMode = "initial"
	SnapshotNone    SnapshotMode = "none"
)

The snapshot modes.

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
	ExportJob    ExportJobProgress
}

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
	SourceProgressExportJobCreated
	SourceProgressExportJobPolled
	SourceProgressExportJobReady
)

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)
	Spec(name string) (ConnectorSpec, 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 StartAttemptRequest added in v0.6.0 

type StartAttemptRequest struct {
	Tenant                                TenantID
	Stream                                StreamRef
	Run                                   RunID
	ExecutionID, PipelineVersionID, Route string
	ExpectedRevision                      int64
	TTL                                   time.Duration
}

StartAttemptRequest supplies immutable execution identity and expected state for admission.

type StreamActivation added in v0.6.0 

type StreamActivation struct {
	StreamStateRequest
	Spec RunSpec
}

StreamActivation initializes immutable execution for an already admitted run. Compatible execution revisions are a later compiler feature; initialization is idempotent only for the same specification. The spec must not contain attempt identity.

type StreamCapabilities added in v0.6.0 

type StreamCapabilities struct {
	// EmitsOps defaults to inserts for rows/messages; changes must declare operations.
	EmitsOps []Operation
	Input    InputSemantics
	Ordering []Ordering
	Delivery DeliveryGuarantee
}

StreamCapabilities describes source input, ordering, and replay guarantees.

type StreamDesiredState added in v0.6.0 

type StreamDesiredState string

StreamDesiredState records intended execution independently of worker health. 

const (
	StreamEnabled StreamDesiredState = "enabled"
	StreamPaused  StreamDesiredState = "paused"
	StreamStopped StreamDesiredState = "stopped"
)

Desired stream states guide whether a supervisor should admit further attempts.

type StreamMeta added in v0.6.0 

type StreamMeta = rowmodel.StreamMeta

StreamMeta carries optional native event metadata alongside a row.

type StreamOpenOpts added in v0.6.0 

type StreamOpenOpts struct {
	// MessageSchemaPolicy opts into batch-boundary negotiation. Empty retains
	// fixed first-message inference. Layouts come from certified progress only.
	MessageSchemaPolicy schema.Policy
	MessageSchemas      map[string]schema.MessageLayout
	// SourceConnectionID identifies the connection used during stream planning.
	SourceConnectionID string
	// CheckAuthority verifies current admitted ownership before provider acknowledgements,
	// including already-certified redeliveries suppressed during Read. It must honor
	// cancellation and be safe alongside lease renewal. Resume positions alone do
	// not grant acknowledgement authority.
	CheckAuthority func(context.Context) error

	// ReportResourceError persists a resource-local startup failure; nil clears a
	// previous failure after successful resolution. A reporting error is fatal.
	// Sources must fail closed when no reporter is supplied.
	ReportResourceError func(context.Context, string, error) error

	Resources          []string
	CommittedPositions DomainPositions
	Membership         map[string]ReplicationStreamResourceStatus
	Attempt            AttemptRef
}

StreamOpenOpts supplies selected resources, durable resume state, and admitted ownership.

type StreamRecordSink added in v0.6.0 

type StreamRecordSink interface {
	RecordSink
	Control(context.Context, Control) error
}

StreamRecordSink extends row writing with owner-ordered source boundary controls. Control runs on the producer owner, flushes affected builders synchronously, and returns success only after preceding Apply calls and integrity checks complete. It does not commit a sink epoch or certify durable progress.

type StreamRef added in v0.0.27 

type StreamRef struct {
	ID         string
	Generation int64
}

StreamRef identifies one admitted generation of a durable replication stream.

type StreamResourceErrorStore added in v0.6.0 

type StreamResourceErrorStore interface {
	SetStreamResourceError(context.Context, LeaseToken, string, string) error
}

StreamResourceErrorStore persists resource health under current worker authority. An empty message clears an earlier startup failure without resetting progress.

type StreamRuntimeStore added in v0.6.0 

type StreamRuntimeStore interface {
	ActivateStream(context.Context, StreamActivation) (StreamState, error)
	StartAttempt(context.Context, StartAttemptRequest) (Attempt, error)
	RenewLease(context.Context, LeaseToken, time.Duration) error
	EndAttempt(context.Context, EndAttemptRequest) error
	LoadStreamState(context.Context, StreamStateRequest) (StreamState, error)
	ListReconcileCandidates(context.Context, ReconcileQuery) (ReconcilePage, error)
	SetDesiredState(context.Context, DesiredStateChange) error
	CommitEpoch(context.Context, EpochCommit) (CommittedEpoch, error)
	GetEpoch(context.Context, EpochLookup) (CommittedEpoch, error)
}

StreamRuntimeStore is an optional engine extension, not part of DataStore or connector lifecycles. PostgreSQL implements it on Store with explicitly configured codecs. All operations are tenant-scoped. StartAttempt, renewal and certification must serialize authority. Tokens increase across generations. Identical certificate retries return historical success without granting current ack authority. New epoch commits require sealed pipeline completion, in addition to source coverage and destination durability receipts. Historical retries grant no authority.

type StreamSession added in v0.6.0 

type StreamSession interface {
	Read(context.Context, StreamRecordSink, Boundary) (Coverage, error)
	Acknowledge(context.Context, Coverage) error
	Close(context.Context) error
}

StreamSession owns its connection and protocol liveness under backpressure. Read and Acknowledge are serialized. Cancel and join an outstanding call before Close; concurrent Close is not required. Cancellation is never a safe boundary. Read returns candidate complete-source coverage; only the coordinator can certify pipeline completion and durability. Acknowledge receives certified coverage and must still check current provider authority.

type StreamSource added in v0.6.0 

type StreamSource interface {
	// Lookup resolves pure position codecs without configuration or network access.
	// The connector also registers these codecs for runtime persistence at composition.
	CodecResolver
	OpenStream(context.Context, StreamOpenOpts) (StreamSession, error)
}

StreamSource is an optional native factory for reusable source sessions.

type StreamState added in v0.6.0 

type StreamState struct {
	StreamStateRequest
	Desired                  StreamDesiredState
	Revision                 int64
	PipelineVersionID, Route string
	Run                      RunID
	Attempt                  *Attempt
	CommittedPositions       DomainPositions
	LastEpoch                *EpochKey
	MembershipRevision       int64
	Membership               map[string]ReplicationStreamResourceStatus
}

StreamState provides authoritative intent and recovery state for a stream generation.

type StreamStateRequest added in v0.6.0 

type StreamStateRequest struct {
	Tenant TenantID
	Stream StreamRef
}

StreamStateRequest identifies a tenant-scoped stream generation.

type StreamingSink added in v0.6.0 

type StreamingSink interface {
	BeginEpoch(context.Context, EpochRef) error
	CommitEpoch(context.Context, EpochRef) ([]EpochReceipt, error)
	AbortEpoch(context.Context, EpochRef) error
	CloseSession(context.Context) error
}

StreamingSink extends an opened Sink with a native reusable epoch lifecycle. CommitEpoch waits for negotiated durability. AbortEpoch discards only tentative effects. CloseSession succeeds only when no later destination effects remain possible; a timeout or mere client resource release must return an error. Calls are serialized initially. No legacy Commit/Abort adaptation is implied.

type StreamingSinkCapabilities added in v0.6.0 

type StreamingSinkCapabilities struct {
	// WritePolicies explicitly describes epoch writes; bounded policies are never inferred.
	WritePolicies                []WritePolicyCapability
	OwnerFencing, IsolatedEpochs bool
	InFlightBound                *time.Duration
}

StreamingSinkCapabilities are claims requiring connector-specific evidence. OwnerFencing checks authority across generations atomically with effects. IsolatedEpochs additionally requires independent contexts (a later contract). InFlightBound is nil when unknown; when set it must be positive and enforced server-side over all residual effects after a proven end to submissions.

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 UserID added in v0.0.27 

type UserID string

UserID identifies a user row filament owns; providers map their own subject onto it through the tenant-scoped external id.

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 Worker added in v0.18.0 

type Worker interface {
	Describe(ctx context.Context) (Catalog, error)
	SourceSpec(ctx context.Context, name string) (ConnectorSpec, error)
	SinkSpec(ctx context.Context, name string) (SinkSpec, error)

	Validate(ctx context.Context, ref ConnectorRef, cfg Config) error
	TestConnection(ctx context.Context, ref ConnectorRef, cfg Config) error
	Discover(ctx context.Context, source string, cfg Config, opts DiscoverOpts) ([]Resource, error)
	// Inspect reports on the named resources, or on every selectable resource
	// when none are named.
	Inspect(ctx context.Context, source string, cfg Config, resources []string) ([]Inspection, error)
	PlanReplicationStream(ctx context.Context, source string, req ReplicationStreamPlanningRequest) (ReplicationStreamPlan, error)
}

Worker answers every connector question the control services ask: the registered specs, and live operations that configure a connector with a caller-supplied config. Unknown connectors are ErrNotFound, missing optional contracts are ErrUnsupported, and a connector rejecting its config is ErrConfigure; every other error is a host failure.

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 (
	AtomicityRecord   WriteAtomicity = "record"
	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. CDC has no read lever; its pairing with append or merge lives here so no caller re-derives it.

type WritePolicy 

type WritePolicy struct {
	// DestinationResource overrides the sink name without changing source progress identity.
	DestinationResource string
	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 checks keyed writes have non-null identity and rejects operations the policy does not accept. Nullable source IDs remain valid for unkeyed writes.

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
auth/v1
auth/v1/authv1connect
ingestion/v1
ingestion/v1/ingestionv1connect
metrics/v1
metrics/v1/metricsv1connect
worker/v1
worker/v1/workerv1connect
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/internal/connection
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/export
Package export runs finite asynchronous HTTP exports.
 Package export runs finite asynchronous HTTP exports.
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.
hubspot
Package hubspot wires the HubSpot source onto the default registry.
 Package hubspot wires the HubSpot source onto the default registry.
hubspot/source
Package hubspot reads HubSpot CRM data through Filament's native source contracts.
 Package hubspot reads HubSpot CRM data through Filament's native source contracts.
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/encoder
Package encoder defines the streaming encoder contract and object formats.
 Package encoder defines the streaming encoder contract and object formats.
internal/gzip
Package gzipencoder wraps an encoder in one streaming gzip member.
 Package gzipencoder wraps an encoder in one streaming gzip member.
internal/json
Package jsonencoder renders Arrow rows as line-delimited JSON or one JSON array for sinks that write text such as stdout and object storage.
 Package jsonencoder renders Arrow rows as line-delimited JSON or one JSON array for sinks that write text such as stdout and object storage.
internal/parquet
Package parquet incrementally renders Arrow record batches as one Parquet file.
 Package parquet incrementally renders Arrow record batches as one Parquet file.
kafka
Package kafka registers the Kafka source, sink, and offset codec.
 Package kafka registers the Kafka source, sink, and offset codec.
kafka/internal/client
Package client supplies shared Kafka connection configuration.
 Package client supplies shared Kafka connection configuration.
kafka/sink
Package sink publishes acknowledged, append-only Kafka records.
 Package sink publishes acknowledged, append-only Kafka records.
kafka/source
Package source consumes Kafka partitions using Filament-owned checkpoints.
 Package source consumes Kafka partitions using Filament-owned checkpoints.
meilisearch
Package meilisearch registers the Meilisearch sink with the default registry.
 Package meilisearch registers the Meilisearch sink with the default registry.
meilisearch/sink
Package sink implements the filament.Sink interface for Meilisearch.
 Package sink implements the filament.Sink interface for Meilisearch.
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+).
nats
Package nats registers the JetStream source, sink, and position codec.
 Package nats registers the JetStream source, sink, and position codec.
nats/internal/connection
Package connection holds the connection settings, validation, and dialing shared by the NATS JetStream source and sink.
 Package connection holds the connection settings, validation, and dialing shared by the NATS JetStream source and sink.
nats/internal/subject
Package subject implements NATS subject pattern algebra shared by the JetStream source and sink.
 Package subject implements NATS subject pattern algebra shared by the JetStream source and sink.
nats/sink
Package sink publishes append-only JSON records to NATS JetStream.
 Package sink publishes append-only JSON records to NATS JetStream.
nats/source
Package source consumes NATS JetStream messages as continuous resources.
 Package source consumes NATS JetStream messages as continuous resources.
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
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
sqlite
Package sqlite implements filament.DataStore, filament.ScheduleStore, and the pipeline store on SQLite via the CGO-free modernc driver, with queries generated by sqlc from datastore/sqlite/queries/*.sql.
 Package sqlite implements filament.DataStore, filament.ScheduleStore, and the pipeline store on SQLite via the CGO-free modernc driver, with queries generated by sqlc from datastore/sqlite/queries/*.sql.
sqlite/metrics
Package metrics implements filament.MetricsStore over the same SQLite runs table datastore/sqlite.Store persists to, mirroring datastore/postgres/metrics: a separate package sharing the handle rather than growing the core datastore with dashboard-query-specific SQL.
 Package metrics implements filament.MetricsStore over the same SQLite runs table datastore/sqlite.Store persists to, mirroring datastore/postgres/metrics: a separate package sharing the handle rather than growing the core datastore with dashboard-query-specific SQL.
sqlite/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).
Package identity is filament's authentication port.
 Package identity is filament's authentication port.
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.
convert
Package convert maps root types onto the generated API types and back.
 Package convert maps root types onto the generated API types and back.
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/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/notifier
Package notifier delivers pipeline notifications from ingestion events.
 Package notifier delivers pipeline notifications from ingestion events.
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/streamsupervisor
Package streamsupervisor reconciles durable activations independently of bus delivery.
 Package streamsupervisor reconciles durable activations independently of bus delivery.
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.
notifier
Package notifier contains private notification validation and delivery contracts.
 Package notifier contains private notification validation and delivery contracts.
notifier/slack
Package slack delivers notifications to Slack incoming webhooks.
 Package slack delivers notifications to Slack incoming webhooks.
notifier/webhook
Package webhook defines webhook destination configuration and validation.
 Package webhook defines webhook destination configuration and validation.
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.
stream
Package stream provides shared source and sink lifecycle bookkeeping.
 Package stream provides shared source and sink lifecycle bookkeeping.
streamproof
Package streamproof carries process-local pipeline completion evidence.
 Package streamproof carries process-local pipeline completion evidence.
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.
Package schema compares complete schemas and evaluates migration rules.
 Package schema compares complete schemas and evaluates migration rules.
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.
sqlite
Package sqlite implements filament.Secrets over the secrets table in the shared Filament SQLite schema, mirroring secret/postgres.
 Package sqlite implements filament.Secrets over the secrets table in the shared Filament SQLite schema, mirroring secret/postgres.
Package streamkit provides optional public helpers for native stream connectors.
 Package streamkit provides optional public helpers for native stream connectors.
Package transform compiles a yaml definition of column transforms into a plan that runs over Arrow record batches.
 Package transform compiles a yaml definition of column transforms into a plan that runs over Arrow record batches.
kernel
Package kernel implements the whole-column operations a transform plan runs.
 Package kernel implements the whole-column operations a transform plan runs.
Package ui serves the Filament web UI.
 Package ui serves the Filament web UI.
Package worker answers connector questions for the control services.
 Package worker answers connector questions for the control services.

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.