Configuration file format

A faucet config is a YAML or JSON document with this top-level shape:

version: 1                 # required, must be 1
requires_faucet: ">=1.13"  # optional; oldest faucet this config needs
name: my_pipeline          # optional; used in state keys and metrics
vars: {}                   # optional; reusable values referenced as ${vars.X}
auth: {}                   # optional; named shared auth providers (see below)
schedule: {}               # optional; cron schedule for faucet schedule (see below)
pipeline:                  # required
  source: { type: …, config: { … } }
  transforms: []           # optional list
  sink:   { type: …, config: { … } }
  state:  { type: …, config: { … } }   # optional
  dlq:    { … }            # optional dead-letter queue (see below)
matrix: []                 # optional per-row overrides / DAG
execution:                 # optional
  max_concurrent: 4
  on_error: continue       # continue | stop
selection:                 # optional; row-selection policy (see Row selection)
  include_parents: off     # off | eligible | all

Unknown keys are rejected

The structural blocks (pipeline, each source/sink/transform/state spec, matrix rows, execution) reject unrecognized fields, so a typo like transorms: or parnet: is a load-time error rather than a silently-ignored field. A connector’s own config: { … } object is checked against the keys that connector declares (faucet schema source <type> / faucet schema sink <type> lists them), so a misspelled connector key is refused too.

Repeated keys are rejected

A mapping that names the same key twice (two sink: blocks, a doubled write_mode:) fails to load with the key and its line, in YAML and in JSON, in configs, templates, overlays and included files alike. Earlier releases kept the last value silently (#953).

requires_faucet

An optional semver requirement (Cargo syntax: ">=1.13", "^1.13.2", ">=1.13, <2") naming the faucet versions that can run this config. validate, run, schedule, a serve submit and template register refuse a config whose requirement the running binary does not meet, with this pipeline requires faucet >=1.13; this binary is 1.12.0. Hub source and sink templates take the same key. faucet init writes it for the scaffolding binary’s minor version. A binary that predates the key rejects it as an unknown field. See Pinning the faucet version.

pipeline

source and sink each take a type (the connector name) and a config object whose fields are that connector’s schema — see faucet schema source <name>. transforms is an ordered list applied to every record. state attaches a state store; dlq attaches a dead-letter queue.

Transforms layering

Transforms can be declared at three layers and are resolved additively per matrix row in lifecycle order:

final = T_pipeline ++ T_source ++ T_row

• pipeline.transforms — cross-cutting policy, runs first on every row.
• pipeline.sources.<name>.transforms — bound to a source template; runs for every row that resolves to this source.
• matrix[i].transforms — row-specific extras, runs last.

Each declaring layer (source template, matrix row) carries an inherit_transforms: bool (default true); setting it false drops every upstream layer for that scope.

A sink may carry free-form attributes: (string pairs such as residency: eu, environment: prod) that a data-flow policy reasons about; a matrix row’s sink.attributes adds or overrides the template’s.

Sinks reject both transforms: and inherit_transforms: at expand time — destination shaping belongs at the pipeline or row layer. See the transforms cookbook for the full model and worked examples.

Available transforms

The full catalogue (with shapes and worked examples) lives in the transforms cookbook; faucet list prints the same set, and faucet schema transform <name> returns the JSON schema for each. Highlights:

• filter — keep records where a JSONPath predicate is true. See the cookbook for the operator set and path syntax.
• explode — expand an array field into one record per element. See the cookbook for the merge rule and on_missing semantics.
• zip_columns — zip a positional report into one object per row: one header list (columns_path) or several header + cell-array groups (from, header, header_label?, value? — a runReport response’s dimensions and metrics). A width mismatch or a column two groups both name fails the page.

Source-side incremental replication (replication_method, replication_key, replication_bind) is configured per connector: rest binds the bookmark into a query param / header / body field / path, graphql into a GraphQL variable — see state & incremental.

Config composition

Three top-level mechanisms let a config be assembled from reusable pieces. They are resolved when the file is read, before any ${...} interpolation.

Mechanism	Form	Effect
extends:	extends: ./base.yaml or a list	Inherit one or more base files; the child deep-merges on top.
profiles:	profiles: { dev: {…}, prod: {…} }	Named overlays, selected at run time with --profile NAME / FAUCET_PROFILE.
!include	key: !include ./frag.yaml	Substitute a YAML fragment at any node (YAML only).

# app.yaml — inherits a base and pulls in a transform fragment.
extends: ./base.yaml          # single path, or a list (merged left-to-right)
pipeline:
  transforms: !include ./transforms.yaml

# base.yaml — shared connection + sink, with named per-environment overlays.
version: 1
name: composed-pipeline
pipeline:
  source: { type: file, config: { path: ./data/input.csv } }
  sink:   { type: file, config: { path: ./out/dev.jsonl } }
profiles:
  dev:  { pipeline: { sink: { config: { path: ./out/dev.jsonl } } } }
  prod: { pipeline: { sink: { config: { path: ./out/prod.jsonl } } } }

• extends — relative paths resolve against the directory of the file that declares them. A list of bases merges left-to-right; the child document overrides them all. Bases may themselves extends: further files (depth-capped, cycle-detected).
• profiles — nothing is applied unless a profile is selected. Select with --profile prod or FAUCET_PROFILE=prod; the flag overrides the env var. An undeclared name is a load-time error.
• !include — a YAML tag (no JSON equivalent) that replaces the tagged node with the parsed contents of another YAML file (sequence, mapping, or scalar). Paths resolve against the including file’s directory.

Merge rule and precedence. Everything composes with the same deep-merge used by matrix rows (objects merge recursively, arrays replace wholesale, scalars replace). Lowest-to-highest priority (last wins):

extended base(s)  →  child document  →  selected profile  →  matrix row

Load-time ordering. Composition runs first, then interpolation:

1. Composition — extends / !include stitched, then the selected profile overlaid; the extends: / profiles: metadata keys are stripped.
2. ${env:…} / ${file:…} / ${secret:…}, then ${vars.X} and ${sources.X} / ${sinks.X} (see Interpolation).
3. Secrets-manager directives (${vault:…} etc.).
4. matrix expansion.

Inspect the result with faucet validate --show-composed — it prints the fully composed document (bases merged, profile applied, fragments substituted, metadata stripped) before interpolation.

Composition is file-loads-only

extends / profiles / !include apply to configs faucet reads from disk (run, validate, preview, doctor, schedule). They are not honored for configs submitted to faucet serve over HTTP — a submitted body is a single self-contained document with no filesystem access. See the config-composition cookbook.

Interpolation

Three stages resolve placeholders:

• Load time: ${env:VAR}, ${file:PATH}, ${secret:VAR} are resolved when the file is read. ${vars.X} resolves against the top-level vars: block; ${sources.NAME.PATH} / ${sinks.NAME.PATH} resolve against named templates. A substituted value is data: it can never combine with the text around it (or another substituted value) into a new ${…} directive — such a ${ stays literal. A directive written whole inside a vars: value still resolves where the value is used. Secret-manager directives (see below) run as the final load-time stage.
• Trigger time: ${param.NAME} resolves against the top-level params: block, bound from --param / an HTTP params object before the config is parsed. ${tenant.id} / ${tenant.name} / ${tenant.labels.KEY} bind in a run faucet serve starts for a tenant (see below).
• Runtime: ${row_id.dotted.path} tokens are resolved per parent record in DAG runs. ${now.*} tokens are resolved per invocation at run time (see below). A value lands in the field shaped for it: an xml source’s body / soap.body_inner gets it XML-escaped, a rest / webhook / xml path and every URL field (base_url, the http sink’s url, a connection URL) get an embedded token percent-encoded into one segment, a file or object path (a file sink’s path, a SQLite database_url, a prefix) refuses a value holding /, \, NUL or .. (a field that is exactly one token takes the value whole as a relative path — it may not be absolute, start with ~ or climb out with ..), and a SQL source’s query gets it as a bind parameter. A field whose value is code the connector runs (a GraphQL query, a DynamoDB filter_expression / key_condition_expression / condition_expression, a sql transform’s query; marked x-faucet-code in the schema) may not hold a row token at all: the config fails to load. Pass the value as data instead (GraphQL variables, DynamoDB expression_attribute_values). ${now.*}, ${backfill.*}, ${partition.*}, ${param.*} and ${vars.*} stay allowed there.

Reference cycles surface as a clear InterpolationCycle error.

Write $${ for a literal ${. The escape holds through every stage — a $${param.x}, $${now.date} or $${vault:…} is never resolved — and becomes ${ only in the value the connector or transform finally receives — in a matrix row, a topology node, a vars: value, or a block such as masking:, notifications: or observability: alike.

${now.*} — run-clock interpolation

${now.*} tokens inject the current wall time into source and sink config values — and into a set transform’s values, so a run can stamp a column with the run date (set: { values: { run_date: "${now.date}" } }, #568). Each invocation evaluates them once at run time:

Token	Example output	Notes
${now.date}	2026-03-08	YYYY-MM-DD
${now.datetime}	2026-03-08T14:05:09+00:00	RFC 3339; alias: ${now.iso}
${now.iso}	2026-03-08T14:05:09+00:00	Alias for ${now.datetime}
${now.year}	2026	Zero-padded 4-digit year
${now.month}	03	Zero-padded month (01–12)
${now.day}	08	Zero-padded day (01–31)
${now.hour}	14	Zero-padded hour (00–23)
${now.minute}	05	Zero-padded minute (00–59)
${now.second}	09	Zero-padded second (00–59)
${now.unix}	1741442709	Unix epoch seconds
${now.strftime.<fmt>}	2026/03/08/14	Arbitrary chrono strftime — e.g. ${now.strftime.%Y/%m/%d/%H}

An unknown token (e.g. ${now.foo}) is a config error at run time. An invalid strftime format produces a clean config error rather than a panic.

Clock source:

• faucet run — the process start time in UTC. Override with --clock <value> for backfills: an RFC 3339 timestamp (2026-03-01T00:00:00Z) or a bare date (2026-03-01, treated as midnight UTC). See the run command reference.
• faucet schedule — the tick’s scheduled time, rendered in the schedule’s timezone. ${now.date} therefore reflects the date in the timezone the cron fires in (e.g. America/Los_Angeles), not UTC. Queued runs use their original scheduled time; --once uses the current wall clock.

Scope: ${now.*} tokens (and ${row_id.path} parent-record references) are resolved only in source and sink config values. Using one in a state:, dlq:, or transforms: config is a config error at validate/expand time — it is rejected rather than silently passed to the connector as a literal ${…} string. (${env:…} / ${vars.X} / ${sources.X} still resolve everywhere.)

Reserved id: now is a reserved matrix row id — a matrix row cannot be named now.

SQL queries: in the query of a SQL source (postgres, mysql, sqlite, duckdb, redshift, mssql, clickhouse, spanner, snowflake, bigquery, oracle; sql for databricks), a ${row_id.path} token is bound as a query parameter, never spliced into the SQL text, because its value comes from another system. A token may stand alone (WHERE id = ${p.id}) or be a whole quoted literal (WHERE name = '${p.name}'); a token embedded in a longer literal ('x-${p.id}'), inside a quoted identifier or inside a comment cannot be bound and is refused when the config loads. The value binds with the type its text had in the SQL: a quoted literal binds as text (a number becomes its digits), and a bare token holding an integer or a decimal as a string ("42", the usual shape of an API id, or "12.50") binds as that number — a decimal only up to 15 significant digits, which a double holds exactly; a longer one, like anything else, keeps its JSON type, so compare it with a numeric column through a CAST. A token that fills a table, table_name, table_id, collection, index, schema or dataset field must resolve to a plain identifier (letters, digits, _, ., -, $). ${now.*} is the run clock, not upstream data, and still substitutes as text.

${tenant.*} — tenant values

In a run started for a tenant (POST /v1/tenants/{tenant}/runs, a tenant template trigger, a fan-out, or a trigger with tenants:), faucet serve binds these anywhere in the document before it is loaded:

Token	Value
${tenant.id}	The tenant id
${tenant.name}	The tenant’s display name (its id when it has none)
${tenant.labels.<key>}	One of the tenant’s labels; a missing label is an error

Anywhere else — faucet run, a plain POST /v1/runs — a ${tenant.*} token is refused with an error naming why, never passed to a connector as literal text. tenant is a reserved matrix row id.

Secrets-manager directives

Four additional load-time schemes pull values from external secrets managers. Each requires the matching build feature (--features secrets-vault, etc.; --features secrets enables all four). Values are fetched concurrently and de-duplicated; they are never written to disk.

Directive	Backend	Auth
${vault:<path>[#field]}	HashiCorp Vault KV v2	VAULT_ADDR + VAULT_TOKEN (+ optional VAULT_NAMESPACE)
${aws-sm:<name-or-ARN>[#field]}	AWS Secrets Manager	aws-config default chain (env / profile / instance / web-identity)
${gcp-sm:projects/<p>/secrets/<s>/versions/<v>}	GCP Secret Manager (versions/latest ok)	Application Default Credentials
${azure-kv:<vault>/<secret>[/<version>]}	Azure Key Vault	AZURE_* env / managed identity / az login

The #field selector (Vault and AWS only) parses the secret body as a JSON object and extracts a single key. Use faucet schema secrets for the machine-readable grammar reference and faucet validate --no-secrets to check grammar offline.

See the secrets cookbook for full examples, the redaction guarantee, and the known limitation around the auth: catalog.

params

Declares the config’s trigger-time override surface: the values that change per run, each typed. Referenced anywhere in the config as ${param.NAME} and bound before the config is parsed, so a param can never alter the document’s structure and never reaches a connector unresolved.

params:
  tenant_id: { type: string, required: true, description: "Tenant to sync" }
  since:     { default: "1970-01-01" }
  page_size: { type: int, default: 500 }
  api_token: { required: true, secret: true }
  region:    { default: us, values: [us, eu, apac] }

Field	Default	Meaning
type	string	string · int · float · bool · list (see List params below)
required	false	The caller must supply a value. Mutually exclusive with default.
default	—	Value when none is supplied. An ordinary config scalar, so default: "${env:SINCE}" resolves. On a typed param it may be text that resolves to the type (type: int, default: "${env:PORT}").
secret	false	Registered for redaction the instant it is bound — never reaches a log, error, API response, audit record, or the template registry.
description	—	Surfaced by faucet template list/show, GET /v1/templates, and the MCP get_template tool.
computed	—	A derived value (see below). Mutually exclusive with required, default, and secret; excluded from the trigger surface.
values	—	Closed set of acceptable values. Anything else is rejected at bind time, naming the allowed set. Mutually exclusive with computed; a default must be one of them.
items	—	(list) The spec every element is held to: a scalar type (with optional values and secret), or type: object with properties.
properties	—	(object items) Each field’s spec: a scalar type, required, default, values, secret. An element may carry no other field.
min_items	1	(list) The fewest elements. An empty list is refused by default rather than running with no rows.
max_items	—	(list) The most elements.
trusted	false	On faucet serve, a caller’s value of this param reaches path, URL and command fields without the caller reference policy (#989). Without it such a value is checked like a submitted config — see template params a caller supplies. Not allowed on computed params or on a list’s items / fields.

List params. type: list (#952) takes a JSON array over HTTP ("tables": ["orders", "customers"]), --param tables='["orders","customers"]' on the CLI, or — for a list of strings — comma-separated text (--param tables=orders,customers). Every element is coerced against items before the config binds; a wrong-typed element, one outside values:, an object missing a required field or carrying an unknown one, and a count outside min_items..max_items are refused naming the element (param 'shards[1]': expected an integer, got "abc"). int elements stay exact past 2^53; float elements are read as exact decimals. A list of lists and an object param outside items are refused. A list is used whole (columns: ${param.cols} lands as an array) or by a source template’s for_each stream; it cannot be spliced into a longer string. secret: true (on the list or its items) masks the whole list as ***; a secret field of an object item masks that field in each element.

Closed value sets. values: turns a param into an enumerable axis. Without it a typo’d value — region: ue — binds happily and surfaces as a 404 mid-run; with it the bind fails up front naming the three allowed values. It also makes the axis machine-enumerable, which is what a template test suite’s auto.enum_coverage sweeps. Each listed value must match the declared type, and a default must be one of them.

Computed params & ${map:…}. A param can be derived from other params instead of supplied, via computed: — resolved after the ordinary params bind and excluded from the trigger surface (the console form, --param, and the HTTP trigger body); supplying a value for one is an error. A computed expression may reference other params (${param.NAME}, including other computed params) and the ${map:NAME|case=value|*=default} lookup — a small, non-Turing-complete switch on another param’s value. Cycles and an unmatched map with no * default are load-time errors. A computed param that references a secret: true param (directly or through another computed param) is secret too: its value is redacted from logs and shown as *** in trigger responses.

params:
  region:          { default: com }
  accounts_domain: { computed: "${map:region|ca=examplecloud|*=example}" }
# region=ca → accounts.examplecloud.ca ; anything else → accounts.example.<region>

${map:NAME|…} also works inline in a source/sink config value, not only in a computed param.

Supplying values. faucet run --param name=value (repeatable), faucet template run <id> --param …, or POST /v1/templates/{id}/runs with a params object. --param-env NAME[=VALUE] overrides an environment variable for that run’s ${env:VAR} resolution only, without mutating the process environment.

Typing. When ${param.NAME} is a scalar’s entire text the declared type survives (an int param lands as a JSON number); embedded in a longer string it is stringified, like every other namespace. Values are accepted as JSON (500) or as strings ("500") and coerced to the declared type, so CLI and HTTP behave identically. A type mismatch, a missing required param, an undeclared --param, or an undeclared ${param.x} reference is an error naming the param.

Validation. faucet validate with no --param binds required params to type-shaped placeholders (the first entry of a values: set, and a ${map:} switching on a placeholder takes its first case), so a parameterized config validates in CI without inventing values; passing any --param switches to strict binding. faucet schema params prints the JSON Schema for one entry.

Persisting a parameterized config for register-once / trigger-by-id use is the pipeline template registry.

matrix

Each row is deep-merged onto pipeline (scalars replace, objects merge, arrays replace). A row with parent: runs once per parent record. See the matrix DAG tutorial. For DRY configs with many rows, define named templates under pipeline.sources / pipeline.sinks and select them per row with ref:.

A row id may contain only letters, digits, _ and - (it is part of the row’s state key and of ${id.field} tokens). A row may reference another row’s record with ${id.field} only when id is its parent: (or, for a for_each: row, one of its dimensions) — any other row’s token would reach the connector unresolved, so it is refused at load time.

depends_on — completion ordering between rows

A row with depends_on: [row_id, …] starts only after every listed row’s invocations finish successfully. Unlike parent:, no records are consumed and there is no per-record fan-out — it is pure run ordering (“load dimensions, then facts”), typically paired with a downstream row whose source reads what the upstream row’s sink wrote.

matrix:
  - id: dims
    source: { config: { query: "SELECT * FROM src_dims" } }
    sink:   { config: { table_name: dims } }
  - id: facts
    depends_on: [dims]        # starts only after `dims` succeeds
    source: { config: { query: "SELECT * FROM src_facts" } }
    sink:   { config: { table_name: facts } }

Semantics:

• Rows whose dependencies are all satisfied run concurrently under the usual execution.max_concurrent budget.
• A failed or skipped dependency skips the dependent row (and its own children and dependents in turn); the run’s exit code reflects the original failure.
• Waiting on a row waits for that row’s own invocations only. To also wait for its per-record children, list them explicitly.
• parent: and depends_on: compose on the same row (the parent edge is an implicit dependency).
• Unknown ids, self-dependencies, and cycles through any mix of parent: / depends_on: edges are rejected at load time by faucet validate.
• Ordering works identically under faucet run, schedule, and serve — they all execute the same expanded plan.

legacy_state_ids — carry state across a row-id change

A row’s id is part of its state key, so renaming a row would start it from scratch. legacy_state_ids: [old_id, …] lists ids the row’s state was stored under before; while the row has no state of its own, its first run moves the bookmark from the first of them that holds one (a partition chunk looks under old_id::partition::<chunk>). faucet discover and runtime fan-out write it when a dataset’s id changed (see Discover).

fan_out: / for_each: — discovery-driven fan-out (#501)

Note

Spelled discover: before #654; the old key is still accepted. It was renamed because discover already meant enumerating a connection’s datasets (faucet discover), and this block enumerates a fan-out axis — one row per value, not one row per dataset.

A fan_out: row enumerates a value-set at run time from a live endpoint, and a for_each: row fans a stream out over the cartesian product of those value-sets — “sync this report once per {subsidiary} × {custom-field} returned by a discovery call”. This is the first-class version of the “enumerate then fan out” shape common to report-style APIs.

pipeline:
  sources:
    api:                      # one complete template both roles reference
      type: rest
      config: { method: GET, base_url: "https://api.example.com", path: "/", auth: { type: none },
                query_params: {}, pagination: { type: None }, replication_method: { type: FullTable } }
  sink: { type: file, config: { path: "./out/${subs.subsidiary_id}-${flds.field_id}.jsonl" } }

matrix:
  - id: subs                  # discovery dimension
    fan_out:
      source: { ref: api, config: { path: "/subsidiaries", records_path: "$.subsidiaries[*]" } }
      select: "$.id"          # JSONPath projecting the value from each record
      as: subsidiary_id       # exposed as ${subs.subsidiary_id}
  - id: flds
    fan_out:
      source: { ref: api, config: { path: "/fields", records_path: "$.fields[*]" } }
      select: "$.id"
      as: field_id
  - id: report                # runs once per (subsidiary × field) tuple
    for_each: [subs, flds]
    source:
      ref: api
      config:
        path: "/reports/trial-balance"
        records_path: "$.rows[*]"
        query_params: { subsidiary_id: "${subs.subsidiary_id}", field_id: "${flds.field_id}" }

Semantics:

• A fan_out: row runs its source once, projects select (a JSONPath such as $.items[*].id; a bare path is read as $.<path>, a dot path whose keys JSONPath shorthand cannot spell — account-id, meta.2fa_id, Account Name — looks each key up verbatim, and $ = whole record) from each record, and dedups (first-seen order; null / missing values skipped). It has no sink and writes nothing. It runs once per pipeline run, cached across all dependents.
• The discovery source is either a { ref: <name> } to a pipeline.sources template (recommended — reuses a complete connector config) or a standalone { type, config }. It is not merged over the default template.
• A for_each: [dims] row runs once per tuple of the cartesian product of the listed dimensions, with ${<dim>.<alias>} substituted into its source and sink config. Time windows (${now.*}, faucet backfill) compose per-invocation without multiplying the matrix.
• Each dimension is folded into the row’s depends_on, so readiness, the skip cascade, and cycle detection reuse the ordering machinery. Per-tuple state keys ({name}::{row}::alias:value/…; a value outside [A-Za-z0-9_.-] or longer than 64 characters is replaced by a readable prefix plus a hash) let every cell resume independently.
• Guards (all at load time via faucet validate): for_each must name fan_out: rows; a fan_out: row can’t carry a sink or parent:, and no row can name one in parent: (use for_each:); for_each can’t combine with parent: (v1). The product is bounded by MAX_MATRIX_PRODUCT (10 000) — a larger cross-product fails rather than spawning an unbounded fleet.

See cli/examples/discovery_matrix.yaml.

Chained / two-level discovery + collected lists (#531)

Some APIs need a second discovery step per discovered value, then the whole result injected into one request — the classic case being “discover the object types, then discover each type’s fields, then read each type asking for all its fields” (CRM custom objects, object-describe endpoints, spreadsheet-style bases, …). Two additions cover it:

• Chained discovery — a fan_out: row may itself carry for_each: [dims], so it runs once per upstream tuple with ${<dim>.<alias>} resolved in its own source config. It must set collect: true.
• Collected (list-valued) dimensions — collect: true publishes the whole deduped value-set as one list per upstream tuple (not a cartesian axis). A consuming row references it as ${<id>.<alias>}, which renders the list comma-joined (e.g. ?properties=a,b,c). Referencing a collected row adds it to the consuming row’s depends_on automatically.

matrix:
  - id: types                      # 1. object types
    fan_out: { source: { ref: type_schemas }, select: "$.name", as: name }
  - id: props                      # 2. per type, collect its property names
    for_each: [types]
    fan_out:
      source: { ref: type_properties, config: { path: "/crm/v3/properties/${types.name}" } }
      select: "$.name"
      as: name
      collect: true
  - id: records                    # 3. per type, read with the whole list at once
    for_each: [types]
    source:
      ref: type_objects
      config:
        path: "/crm/v3/objects/${types.name}"
        query_params: { properties: "${props.name}" }

Guards: a chained fan_out: row (for_each: present) must set collect: true; collect: true requires for_each:; chained-discovery cycles are rejected at load time. See cli/examples/discovered_custom_objects.yaml.

pipeline.nodes / pipeline.edges (topology mode)

An alternative to matrix: for pipelines that need fan-out, fan-in, or joins: declare an explicit graph of typed nodes (source / transform / tee / merge / join / sink) under pipeline.nodes, connected by pipeline.edges ({ from, to, as? }). Topology mode is mutually exclusive with matrix: — both non-empty is a load-time error. faucet run / validate / preview all understand it. See Topology mode for the full grammar, the join: node, state semantics, and runnable examples. A node entry accepts only its kind’s keys — a misspelt key (confg:) is a load-time error — and ${vars.*} / ${sources.*} / ${sinks.*} resolve inside node overrides and transform nodes as they do elsewhere.

What applies in topology mode. Each block is scoped to the node where it makes sense. The per-page governance passes — pipeline.masking, pipeline.quality, pipeline.contract, schema: — are enforced per sink node, and resilience: applies to its writes. sla: keeps per-sink-node history under {pipeline}::{node_id} (staleness and volume checks, like profiling:, need a durable state: block, as in matrix mode); notifications: reports per sink node; lineage: emits one job per sink node ({pipeline}.{node_id}) whose inputs are every source that reaches it; catalog: records a dataset per source and per sink plus an edge for each pair the graph connects. budget:, metadata_columns: and reconcile: apply per sink node too. So do --dry-run / --limit, ${now.*} / --clock, state:, and dlq: (a --dry-run counts would-be dead letters without writing them, and a preview runs at-least-once). verify: and rollback: are refused (a graph sink has no single source to verify against, and runs are undone per matrix row), and usage: is accepted but not applied. See the applies-per-node table for the detail.

faucet run, faucet schedule and faucet serve run a topology config as its graph; commands that work on matrix rows (doctor, plan, backfill, verify, rollback, mirror, change requests) refuse it with an explicit error.

Column lineage is the one thing deliberately withheld: with several inputs feeding a sink, the per-column derivation is not knowable from the graph, so the facet is omitted for a multi-input sink rather than guessed. Single-input sinks emit it as usual.

delivery: exactly_once is supported, with five requirements checked at load time: exactly one source node, a replayable source, every sink idempotent, a durable state:, and no dlq:. Each sink node carries its own commit watermark and a restart resumes from the position furthest behind; every sink ahead of it skips the replayed pages it already committed, so no sink is resumed past its own progress or writes a page twice. See Exactly-once delivery.

At-least-once resume is deliberately conservative. Each sink node owns a bookmark under {pipeline}::{node_id}. Without exactly-once the source resumes from a stored position only when the graph has exactly one source node and every sink’s bookmark is identical; otherwise it replays in full (with a warning). Bookmarks are compared for equality, never ordered — a resume position is often structured (a CDC LSN map, a Kafka offset map), and an ordered “minimum” over those can sit ahead of the true minimum and skip the lagging sink’s records. Replaying costs duplicates on a non-idempotent sink; skipping would lose data, so the trade is made in that direction. Use write_mode: upsert on the sinks — or delivery: exactly_once — when a graph is resumed routinely.

Row selection

Register many rows, run a few. Selection resolves after expansion and never changes a row’s state key ({name}::{row_id}), so bookmarks are identical across a full run and any subset. It runs on run, validate, and preview via the flags in the CLI reference. Four axes compose through one formula:

1. eligible  = status gate ({mandatory, active} ∪ --status)
2. narrowed  = (eligible ∩ --tag) ∪ (--select / --only by id)
3. parents   = apply include_parents policy to narrowed
4. run set   = parents − (--skip)

status: (readiness ladder) — a field on the row’s source (template or source: override; deep-merges as a scalar). Default active, so existing configs are unchanged. mandatory always runs; active runs by default; available/draft/ archived run only when their tier is added with --status. mandatory is removable only by an explicit --skip <id>.

tags: — free-form ^[a-z0-9][a-z0-9_-]*$ labels on a row, union-merged with the source template’s tags: (the one deliberate exception to array-replace). Tags narrow within the eligible set (--tag); they never resurrect a parked row — raise --status for that.

selection.include_parents: — the single policy deciding whether a selected row’s parent: / depends_on: ancestor (not independently selected) is pulled in. off (default, strict) errors naming each missing pair; eligible auto-includes status-eligible ancestors (errors on a parked one); all includes any (warns on parked). --select <id> by name always satisfies a dependency. Overridable with --include-parents; env FAUCET_INCLUDE_PARENTS; precedence flag > env > config > default.

matrix:
  - id: people
    source: { ref: hibob, status: active,    config: { path: /v1/people } }
    tags: [core, daily]
  - id: payroll
    source: { ref: hibob, status: mandatory, config: { path: /v1/payroll } }
    tags: [finance]
  - id: audit
    source: { ref: hibob, status: available, config: { path: /v1/audit } }
    tags: [finance]

selection:
  include_parents: off   # off (default) | eligible | all

Command	Runs
faucet run cfg.yaml	people, payroll
faucet run cfg.yaml --tag finance	payroll (audit is finance but available)
faucet run cfg.yaml --status available --tag finance	payroll, audit
faucet run cfg.yaml --select audit	audit (opt-in, forced by name)

Unknown tokens, an empty run set, and a missing required ancestor are hard, fail-fast errors (no partial run).

auth

A map of named auth providers, each { type, config } (type ∈ static / oauth2 / oauth2_refresh / token_endpoint / google_service_account / flow / oauth1). A connector references one with auth: { ref: <name> } instead of inline auth; faucet builds each provider once and shares it across every connector that references it (one token, single-flight refresh). See the authentication cookbook.

auth:
  api:
    type: oauth2_refresh
    config:
      token_url: ${env:API_TOKEN_URL}
      client_id: ${secret:API_CLIENT_ID}
      client_secret: ${secret:API_CLIENT_SECRET}
      refresh_token: ${secret:API_REFRESH_TOKEN}
      persist: { path: "./state/auth" }   # persist a rotated refresh_token across runs

oauth2_refresh accepts persist: { path } to durably store a rotated refresh token so later runs survive rotation. token_endpoint accepts encoding: form (urlencoded body) and apply_as: { header, template } (place the token in a custom header, e.g. a session cookie). See the authentication cookbook.

google_service_account authenticates with a Google service-account key (RFC 7523 JWT-bearer): key_json (the key file’s contents, e.g. ${secret:GOOGLE_SA_KEY}) or key_file (a path), scopes (required), optional subject (domain-wide delegation), token_uri (defaults to the key’s) and expiry_ratio. A malformed key fails at load time.

auth:
  google:
    type: google_service_account
    config:
      key_json: ${secret:GOOGLE_SA_KEY}
      scopes: ["https://www.googleapis.com/auth/analytics.readonly"]

delivery

Controls the delivery guarantee for every pipeline row.

delivery: at_least_once   # default — no behaviour change
# or:
delivery: exactly_once

Value	Behaviour
at_least_once	Default. A crash between the sink write and the bookmark persist causes the page to be re-delivered on the next run. Downstream must tolerate duplicates.
exactly_once	Require at least effectively-once. Two mechanisms qualify: the atomic watermark (the sink durably records a per-page commit token — which embeds the page’s resume bookmark — atomically with the data; on resume the pipeline recovers the exact stream position from the sink’s watermark, or skips already-committed pages for legacy tokens), and keyed upsert (write_mode: upsert + key on an upsert-capable sink, any source). faucet validate prints which mechanism each row derives.

Per-row override: set delivery: directly on a matrix row to override the top-level value for that row.

delivery: at_least_once    # top-level default

matrix:
  - id: critical_row
    delivery: exactly_once  # this row uses effectively-once
  - id: best_effort_row
    # inherits top-level at_least_once

Requirements for exactly_once

The config is accepted when either effectively-once mechanism is achievable and rejected otherwise, at config-load time (faucet validate and faucet run). A violation is a hard config error naming the limiting side — no run is started.

Keyed-upsert path (any source): the sink must be upsert-capable (postgres, sqlite, mysql, mssql, mongodb, elasticsearch, bigquery) and configured with write_mode: upsert (or delete) and a non-empty key. No other requirement — no watermark is used.

Atomic-watermark path, all four conditions:

1. Positional-replay source — the source must be one of: postgres-cdc, mysql-cdc, mongodb-cdc, kafka. These emit a complete resume position on every page over an immutable log. Query-based sources are rejected because different data on replay would cause the pipeline to silently skip records it never wrote.
2. Idempotent sink — the sink must be one of: sqlite, postgres, mysql, mssql, iceberg, bigquery, kafka, snowflake, redis, mongodb (MongoDB requires a replica set at run time). These sinks atomically commit both the data and a watermark token inside the same transaction or snapshot.
3. Durable state store — a state: block is required, and it must be a durable backend (file, redis, or postgres) — memory is rejected. The pipeline stores the per-page sequence number alongside the bookmark; the watermark must survive a restart, so an in-memory store (lost on process exit) would silently re-deliver an already-committed page on resume.
4. No DLQ — a dlq: block is incompatible with the atomic-watermark path in this version. (The keyed-upsert path permits a DLQ.)

See the Effectively-once delivery cookbook for a worked example and the full rationale.

schema

Optional pipeline-level block (a sibling of source / sink / transforms / state inside pipeline:) that declares one uniform policy for schema drift — when an incoming page’s top-level shape diverges from the sink’s live destination schema. Fully opt-in: with no block, sinks keep their existing per-connector behaviour. See the Schema drift cookbook for the full model, sink-support matrix, and per-sink nuances.

pipeline:
  schema:
    on_drift: warn                     # warn | evolve | ignore | quarantine | fail
    allow_type_widening: true          # default true; only consulted by `evolve`
    on_incompatible: fail              # fail | quarantine — `evolve` only (default fail)
    relax_nullability_on_missing: false # default false; `evolve` only
  source: { ... }
  sink: { ... }

Field	Default	Purpose
on_drift	warn	Policy applied when drift is detected: warn (metric + log, write unchanged), ignore (drop unknown fields), fail (abort with a SchemaDrift error), quarantine (route drift-exhibiting rows to the DLQ, write the rest), evolve (apply additive/widening DDL, then write).
allow_type_widening	true	Whether a lossless widening (integer → number, gaining nullability) counts as evolvable rather than incompatible. Only consulted by evolve.
on_incompatible	fail	evolve only — action for an incompatible residue (narrowing / type swap): fail aborts, quarantine routes the offending rows to the DLQ.
relax_nullability_on_missing	false	evolve only — whether a NOT NULL destination column absent from a page may have its NOT NULL constraint dropped. Default false: an omitted column is not evidence of optionality, so the constraint is left untouched (a genuinely-missing required value then fails at write time). Set true only to deliberately let omission relax nullability. Relaxation from an observed null in a present column (a widening) is unaffected.

Detection is top-level only — a nested object is one column, so changes inside it are invisible.

Gates (validated at config-load time)

A violation is a hard config error naming the offending row; no run is started.

1. evolve needs an evolution-capable sink — one of postgres, mysql, mssql, sqlite, bigquery, elasticsearch. iceberg supports detection but not evolve (blocked on upstream iceberg-rust, #255); schemaless sinks have nothing to evolve. Both are rejected for on_drift: evolve.
2. quarantine needs a dlq: block — on_drift: quarantine, or evolve with on_incompatible: quarantine.
3. quarantine is incompatible with delivery: exactly_once (effectively-once forbids a DLQ). evolve / ignore / fail / warn all compose with effectively-once and with write_mode: upsert.

Against a schemaless sink (jsonl, csv, stdout, mongodb, redis, http, kafka, s3, gcs, snowflake, parquet) any non-evolve policy is inert — the sink reports no schema to diverge from.

contract

Optional pipeline-level block (a sibling of source / sink / transforms inside pipeline:; no matrix-row override in v1) declaring a data contract: a versioned promise about the pipeline’s output shape, enforced per page after transforms and quality checks and before the sink write. Requires the contract Cargo feature (in the default build). See the Data contracts cookbook for the full model and faucet schema contract for the block’s JSON Schema.

pipeline:
  contract:
    version: "1.0.0"            # required, non-empty
    description: Orders feed.   # optional metadata
    owner: data-platform        # optional metadata
    on_breach: fail             # fail (default) | quarantine | warn
    allow_extra_fields: true    # default true
    fields:                     # required, non-empty; names unique
      - name: order_id
        type: string            # string | integer | number | boolean | object | array
        required: true          # default true
        nullable: false         # default false
        min_length: 1           # string-only (with max_length)
      - name: status
        type: string
        enum: [open, shipped, cancelled]
      - name: amount
        type: number
        min: 0                  # numeric-only (with max)

Field	Default	Purpose
version	—	Carried into breach errors, DLQ envelopes, and exports. Semver recommended (major = breaking, minor = additive).
on_breach	fail	fail aborts on the first breach (nothing from the page is written); quarantine routes breaching records to the DLQ and writes the rest (requires a dlq: block — validated at load time); warn logs + counts but writes everything.
allow_extra_fields	true	When false, an undeclared top-level key is a breach (extra_field).
fields[]	—	Per-field type + constraints: required, nullable, enum, pattern (string), min/max (numeric, inclusive), min_length/max_length (string, inclusive), description.

A malformed contract (empty version, duplicate fields, invalid regex, empty or type-mismatched enum, constraints on the wrong type, min > max) is a config-load error — faucet validate catches it. fail/warn compose with delivery: exactly_once; quarantine does not (effectively-once forbids a DLQ). Inspect or export the contract with faucet contract.

masking

Optional pipeline-level block (a sibling of source / sink / transforms inside pipeline:) declaring a PII detection + column-masking policy. The masking pass runs first — before the quality, contract, and schema-drift passes and before every sink write, the DLQ, and lineage sampling — so PII never reaches a sink (including the DLQ) or an OpenLineage facet unmasked. Masking is value-only and key-preserving: it never fails a run or quarantines (no dlq: required). Requires the masking Cargo feature (in the default build). See the masking cookbook for the full model and faucet schema masking for the block’s JSON Schema.

pipeline:
  masking:
    description: Mask customer PII.       # optional metadata
    key: ${vault:secret/faucet#mask_key}  # optional — keyed HMAC-SHA256 for hash/tokenize
    rules:                                # required, non-empty; first match per field wins
      - name: emails                      # optional label (logs + metric); default rule_<n>
        match:                            # at least one of the three must be set
          value_detector: email          # email | credit_card | ssn | phone | phone_loose | ipv4
        action: { type: redact }          # replace with `mask` (default "***")
      - match: { field_pattern: '(?i)^ssn$' }   # regex over the field dot-path
        action: { type: hash }            # HMAC-SHA256 (keyed) / SHA-256 (unkeyed) hex
      - match: { fields: [card] }         # explicit dot-paths
        action: { type: partial, keep_last: 4 }   # reveal only the last N chars
        applies_to: [warehouse]           # scope to sink template name(s) / connector kind(s)

Field	Default	Purpose
description	—	Documentation metadata.
key	—	Secret for keyed HMAC-SHA256 hash/tokenize (deterministic + irreversible). Absent → unkeyed SHA-256 (deterministic but recomputable). Resolved after secrets, so ${vault:...} etc. work.
rules[]	—	Required, non-empty. Each rule = name (optional label) + match + action + optional applies_to. Evaluated in order; the first rule that matches a field wins.
rules[].match	—	At least one of field_pattern (regex over the dot-path), value_detector (email/credit_card/ssn/phone/phone_loose/ipv4, run over string values), fields (explicit dot-paths). A match on a container masks the whole subtree (redact replaces it; hash/tokenize/partial rewrite every non-null scalar inside it).
rules[].action	—	Tagged by type: redact (mask, default "***"; mask: null nulls the field), hash, tokenize (prefix), partial (keep_last default 4, mask_char default *; keep_last >= len masks everything).
rules[].applies_to	[] (all sinks)	Scope the rule to specific sinks by template name (under pipeline.sinks:) or connector kind (e.g. bigquery).

Detectors are conservative (fully anchored; credit_card requires a valid Luhn checksum; ssn excludes never-issued ranges) so false positives stay rare. A credit_card or phone_loose rule can match a JSON number, so with one in the policy every number is written as a string, keeping each column one type across pages (see the masking cookbook). hash/tokenize are deterministic → masked values stay joinable across pipelines that share a key. A malformed policy (empty rules, an empty match, an invalid regex, an empty tokenize prefix) is a config-load error — faucet validate and faucet masking catch it.

• faucet_masking_fields_total{pipeline,row,rule,action,detector} — one increment per masked field (detector empty for name-based matches).

policy

Optional top-level block declaring a data-flow policy: classifications that label columns and rules about which sinks a labelled column may reach. Decided before any data moves (validate / policy / plan / doctor report; run and the serve submit path refuse) and backed at run time by value detectors on the real records. A --policy <file> merges on top of this block; a deployment overlay’s policy: lands here. Requires the policy Cargo feature (in the default build; implies masking). faucet schema policy prints the JSON Schema.

policy:
  version: 1
  description: PII stays in the EU.
  classifications:
    - label: pii
      fields: [email, phone]              # full dot-path or leaf key
      field_pattern: "(?i)_?name$"        # regex over the dot-path
      value_detector: email               # email | credit_card | ssn | phone | phone_loose | ipv4
  rules:
    - name: pii-eu
      when: { label: pii }                # + optional sink_kind: [..], sink: { attr: [..] }
      require: { residency: [eu] }        # sink attribute must be one of these
      mask: [hash, tokenize]              # or the column reaches the sink masked
      on_runtime: fail                    # fail | quarantine (needs a dlq: block)
    - name: finance-no-prod-files
      when: { label: finance, sink_kind: [jsonl, csv], sink: { environment: [prod] } }
      deny: true

Field	Default	Meaning
version	1	Policy document version.
classifications[].label	—	The label a matching column earns. Several classifications may share one.
classifications[].fields	[]	Column names — the full dot-path or its leaf key (ssn covers user.ssn).
classifications[].field_pattern	—	Regex over the column’s dot-path.
classifications[].value_detector	—	A masking detector run over string values at run time.
rules[].when.label	—	The label the rule governs.
rules[].when.sink_kind	[] (all)	Only sinks of these connector kinds.
rules[].when.sink	{}	Only sinks whose attribute has one of the listed values (a sink without the attribute is out of scope).
rules[].require	{}	Sink attributes that must be present with a listed value.
rules[].mask	[]	Masking actions (redact / hash / tokenize / partial) that satisfy the rule when the column reaches the sink masked.
rules[].deny	false	Never satisfied.
rules[].on_runtime	fail	What the runtime backstop does: fail the page, or quarantine the rows to the DLQ.

Every classification needs at least one of fields / field_pattern / value_detector; every rule needs a when.label and one of require / mask / deny; two rules with one name are refused. Metric: faucet_policy_violations_total{pipeline,row,rule,phase,action}.

execution

• max_concurrent — one shared concurrency budget across roots and child fan-outs.
• on_error — continue (siblings finish; failed subtree skipped) or stop (abort pending and in-flight work on first failure).
• schedule — the order ready sibling rows queue for a permit: declared (default, matrix order) or lpt.

Dispatch order (schedule: lpt)

With rows of very different sizes, total wall-clock depends on the order you happened to list them in: a large object listed late becomes an idle tail while the other slots sit empty. schedule: lpt (longest-processing-time-first) starts the heaviest rows first, which is provably within 4/3 of the optimal makespan — whereas largest-last, which declaration order can produce by accident, is the worst case.

execution:
  max_concurrent: 8
  schedule: lpt
matrix:
  - id: contact
    weight: 15500000      # optional; `faucet discover` fills this in

Rows are ranked by weight, highest first. Ties and rows without a weight keep declaration order, so dispatch stays deterministic and predictable from the config alone; unweighted rows sort after weighted ones.

faucet discover writes a weight per generated row automatically, estimated as rows × row width from each dataset’s row estimate and column types — bytes moved, not row count, because a 970k-row × 3-column table is lighter than a 122k-row × 100-column one. Set weight by hand when you know better.

The win depends on duration correlating with weight. For sources whose runtime is dominated by a server-side queue (a bulk export API, say), the estimate is a proxy rather than a prediction — LPT still cannot do worse than declared order, but it may not help as much.

Only the enqueue order changes: the same permit budget, the same on_error behaviour, and children / depends_on rows keep their completion-gated ordering.

Adaptive batch sizing

The optional adaptive_batch_size: sub-block enables the AIMD controller that auto-tunes the effective write batch size from observed sink latency and error rate. Default enabled: false (opt-in).

execution:
  adaptive_batch_size:
    enabled: true          # master switch
    controller: aimd       # only "aimd" is supported in v1
    min: 100               # lower bound (rows)
    max: 50000             # upper bound; inert above the source page size
    increase_step: 250     # additive growth per clean batch
    decrease_factor: 0.5   # multiplicative shrink on error/high latency  (0, 1)
    cooldown_batches: 5    # batches to skip after a shrink
    target_latency_ms: null  # optional write-latency target (ms)
    latency_window: 10     # rolling window size for p50 latency
    error_threshold: 0.01  # per-batch error rate that triggers a shrink
    respect_source_max: true  # cap at source page size (see Caveats)
    log_every: 50          # tracing::info every N adjustments

Key caveats:

• Error-driven shrink requires a dlq: block. Without one the controller sees no per-row errors; only target_latency_ms can drive shrinks.
• Effective ceiling = source page size. In v1 the controller reslices pages in-memory — it cannot buffer across pages. Setting max higher than the source batch_size is harmless but inert. Raise the source batch_size to allow bigger write batches.
• No-op for per-record sinks. jsonl, csv, and stdout write one record at a time; the controller adjusts normally but the write granularity is unchanged.

See the Adaptive batching cookbook for a full worked example, the AIMD trajectory, and the four Prometheus metrics (faucet_pipeline_adaptive_batch_*).

resilience

Optional top-level block giving the pipeline one declarative place to configure retry, a circuit breaker, and per-row poison-pill handling. Fully opt-in: with no resilience: block, sink writes are not retried and source connectors keep their built-in retry defaults. See the Resilience cookbook for the full model, composition notes, and metrics.

resilience:
  retry:
    max_attempts: 5            # total tries including the first (1 = no retry)
    backoff: exponential       # none | fixed | exponential
    base_ms: 200
    max_ms: 30000              # per-sleep cap, before jitter
    jitter: true
  retry_on: [http_5xx, rate_limited, connection, timeout]
  circuit_breaker:
    consecutive_failures: 5
    cooldown_secs: 60
  poison:
    max_row_attempts: 3
    action: dlq                # dlq | drop | fail

• retry — max_attempts (default 5; 1 disables retry), backoff (none / fixed / exponential, default exponential), base_ms (default 200), max_ms (per-sleep cap, default 30000), jitter (default true, applies [0.5, 1.5) decorrelated jitter).
• retry_on — the transient error classes that are retried: http_5xx (HTTP 5xx), rate_limited (HTTP 429 / rate-limit signals), connection (DNS / refused / reset), timeout (request timeouts). Omit for all four; an empty list is rejected at config load.
• circuit_breaker — consecutive_failures consecutive fully-failed pages open the breaker and fail the run with a CircuitOpen error; cooldown_secs is advisory for faucet schedule (delays the next cron tick).
• poison — per-row DLQ-path handling: max_row_attempts re-submits a still-failing retriable row before the terminal action — dlq (requires a dlq: block), drop, or fail.

The rest source’s legacy max_retries / retry_backoff fields win when set explicitly; otherwise the injected policy’s max_attempts + base apply (its retry_on / max / jitter are inert on REST, honored on xml / graphql and on every sink-side write).

For APIs that signal a rate limit with a 4xx other than 429 plus a code in the body, inside a successful response, or with a reset time in a header, the rest and graphql sources’ retry_on_response list turns matching responses into throttling (counted in faucet_source_throttled_total and retried up to max_retries times in a row):

Field	Default	Meaning
status	[]	Statuses the rule applies to (empty = any non-2xx, or any status with match_success).
match_success	false	Also match 2xx responses (GraphQL THROTTLED errors, x-ratelimit-remaining: 0). Requires body_path or header.
body_path	—	JSONPath into the JSON body; any match equal to one of values satisfies the rule.
values	[]	Matching values (numbers and strings compare by text). Required with body_path.
header	—	A header that must be present (compared with values when there is no body_path).
backoff_from	—	Where the response states the wait: header (name, unit), header_json (name, path, unit), body (path, unit) or cost_bucket (requested, available, restore_rate paths). unit: seconds, ms, minutes, epoch_s, epoch_ms, rfc3339; absolute instants are measured from the response Date header. Missing or unreadable → the next source below.
backoff_secs	—	Fixed wait; otherwise Retry-After, otherwise the exponential retry_backoff.
max_wait_secs	3600	Longest wait backoff_from may ask for; a longer one fails the run with the reset named.

On rest, rules are checked before tolerated_http_errors and apply to data pages, async_job requests and discovery requests; on graphql they are checked before the response’s errors[], and a match retries the whole request. See the resilience cookbook.

Source fields: incremental keys, body targets, CSV nulls

A few connector-level fields change how records are selected and decoded. They live in the source’s config, not at the top level; faucet schema source rest prints the full list.

rest — incremental replication

Field	Default	Meaning
replication_key	—	A top-level field (updated_at), a dot path (fields.updated, items.0.date; a literal top-level field of that name wins), or a JSON Pointer (/fields/updated) for names containing dots. With async_job it is injected into the submit query, so a pointer is rejected there.
on_missing_key	keep	A record whose key is missing or null: keep (counted in faucet_source_replication_key_missing_total, warned once per run), drop (counted and warned), or fail.
replication_bind.path	—	into: body only: a JSON Pointer into body (/filterGroups/0/filters/0/value) instead of a top-level name. Set exactly one of name / path; the pointer must resolve to an existing value or a new key of an existing object.
replication_bind.value_type	string	number writes the rendered bookmark as a JSON number.
window.lower / window.upper .path, .value_type	—	Same as the two above, for window bounds (e.g. /dateRanges/0/startDate).
window.*.template placeholders	${window}	${window} is the bind’s own boundary; ${window.start} / ${window.end} render the window’s start and granularity-adjusted end in any bind, so one bind can carry both (a report query’s segments.date BETWEEN '${window.start}' AND '${window.end}').
window.upper	—	Optional when lower.template contains ${window.end}; only lower is then applied. Omitting it otherwise is rejected (unbounded above).

rest — pagination and CSV

Field	Default	Meaning
pagination.body_cursor_field (CursorInBody)	—	A top-level body field, or a JSON Pointer such as /variables/after.
pagination.offset_field / limit_field (OffsetInBody)	—	Likewise a field name or a JSON Pointer.
csv_null_values	[]	CSV fields equal to a listed string decode as null ([""] for bulk-export results that write null as an empty field). Applies to response_format: csv and parse: { format: csv } decode steps.

With LinkHeader / NextLinkInBody pagination, next-page links are resolved against the request URL (relative links work), and query binds are sent on the first request only — the link already carries them.

csv source

Field	Default	Meaning
null_values	[]	Fields equal to a listed string decode as null. Headers are never mapped.

dlq

Optional dead-letter queue under pipeline: (or per matrix row). See the DLQ cookbook.

Field	Type	Default	Description
sink	connector	—	Where failed and quarantined records go (usually jsonl).
on_batch_error	propagate | dlq_all	propagate	When a whole sink write fails: fail the run, or send every row of it to the DLQ and continue. dlq_all is refused on a sink whose failed write can land part of a batch — see batch atomicity.
allow_duplicates_on_dlq_all	bool	false	Accept dlq_all on such a sink anyway: rows that already landed also go to the DLQ, so a replay writes them twice.
max_failures_per_page	int	—	Fail the run once one page sends more records than this to the DLQ.
max_failures_total	int	—	Fail the run once more records than this have gone to the DLQ in total.
include_original_payload	bool	true	Keep each failed record’s payload in its envelope.

sla

Optional top-level block declaring a freshness/volume SLA for the pipeline (evaluated after every root invocation by faucet run / schedule / serve / mirror). Fully opt-in and never fails a run: violations emit the faucet_pipeline_sla_violations_total{pipeline,row,kind} counter and a structured warning, and faucet doctor reports staleness / baseline health. See the SLA monitoring cookbook.

sla:
  max_staleness_secs: 7200     # stale when a run ends 2h+ after the last success
  min_rows_per_run: 1          # a successful run writing fewer records violates
  volume_anomaly:              # learned-baseline anomaly detection
    method: zscore             # zscore | iqr
    sensitivity: 3.0           # zscore default 3.0; iqr default 1.5
    min_history: 5             # successful runs before detection starts
    window: 20                 # rolling baseline size
  max_lag_bytes: 1073741824    # the source may be at most 1 GiB behind its head

Field	Type	Default	Description
max_staleness_secs	int	—	Maximum seconds since the last successful run. Evaluated when a run ends (a failure, or a success arriving later than this after the previous one) and by faucet doctor. A pipeline that stops running emits nothing — alert on faucet_pipeline_sla_last_success_unix_seconds. Requires a state: block.
min_rows_per_run	int	—	Static volume floor for a successful run (catches a source silently returning nothing). Stateless — works without a state: block.
volume_anomaly.method	zscore | iqr	zscore	How a successful run’s volume is compared against the rolling baseline of recent successful runs.
volume_anomaly.sensitivity	float	3.0 / 1.5	zscore: max |x − mean| / std. iqr: Tukey fence multiplier. Defaults per method.
volume_anomaly.min_history	int	5	Cold-start guard: successful runs of history required before detection fires (min 2).
volume_anomaly.window	int	20	Rolling window of successful-run volumes kept as the baseline (≥ min_history).
max_lag_bytes	int	—	Maximum source lag in bytes — unread Postgres WAL or MySQL binlog (#733).
max_lag_events	int	—	Maximum source lag in events — unconsumed Kafka messages, SQL Server change transactions.
max_lag_seconds	int	—	Maximum age of the oldest change the source has not delivered — MongoDB, SQL Server, Oracle, Kinesis, DynamoDB Streams.

The max_lag_* thresholds are checked against the lag the source reports when each run ends (kind="lag" on the violations counter), by faucet status (--probe queries the source now) and by faucet doctor’s lag probe. A unit the source does not report is not checked, and a source with no notion of a head (a query source) never breaches them. They need no state: block.

At least one check must be set. max_staleness_secs / volume_anomaly require a state: block (enforced at config load); the history is persisted next to the pipeline’s bookmarks under {name}::{row}::__sla__. With a memory state store the history only persists within a single faucet schedule / serve process. Schema: faucet schema sla.

Per-row override (#679). A matrix row may carry its own sla:, which replaces the top-level block for that row’s invocations (the state-store gate applies to it the same way). This is also where a deployment overlay’s per-stream sla lands.

sla: { min_rows_per_run: 1 }        # every row …
matrix:
  - id: invoices
    sla: { max_staleness_secs: 3600 }   # … except this one
  - id: customers

profiling

Optional top-level block declaring learned column profiles with drift detection (#708): every root invocation profiles the records it wrote (null rate, type mix, distinct estimate, numeric / string summaries, top values) into a bounded-memory sketch, compares the result with a rolling baseline of earlier runs kept in the state: store, and reports a statistically significant change per column — no thresholds to write. See the column profiling cookbook.

profiling:
  min_history: 5               # baseline runs before detection starts
  window: 20                   # rolling baseline size
  method: zscore               # zscore | iqr (numeric metrics)
  new_value_min_share: 0.05    # share a new value / type must reach to be reported
  psi_threshold: 0.2           # value-distribution shift threshold
  categorical_max_distinct: 100
  exclude: ["_faucet_*"]
  on_drift: warn               # warn | notify | fail

Field	Type	Default	Description
columns	list	all	Only these top-level columns.
exclude	list	["_faucet_*"]	Exact names or prefix* globs to skip.
max_columns	int	200	Cap on profiled columns per run.
top_values	int	10	Frequent values kept per categorical column; 0 disables.
categorical_max_distinct	int	100	Distinct estimate above which a column publishes no example values.
window	int	20	Rolling baseline size (≥ min_history).
min_history	int	5	Baseline runs before detection starts (≥ 2).
method	zscore | iqr	zscore	Test for the numeric metrics (null rate, distinct, mean, min, max, string length).
sensitivity	float	3.0 / 1.5	z-score threshold, or the IQR fence multiplier.
new_value_min_share	float	0.05	Share a previously unseen value or JSON type must reach.
psi_threshold	float	0.2	Population stability index above which the value distribution counts as drifted.
on_drift	warn | notify | fail	warn	Log + metric; plus a profile_drift notification; plus the run is reported failed (data already written).

Requires a state: block (the baseline lives under {name}::{row}::__profiling__); a memory store only baselines within one process. Profiling runs after transforms and masking, for real root invocations only. A matrix row’s own profiling: replaces the top-level block for that row. Inspect and re-baseline with faucet profiling. Schema: faucet schema profiling.

reconcile

Opt-in completeness reconciliation (#502): after a successful root run, fetch an authoritative row count and fail the run on a shortfall beyond tolerance — a guard against silent truncation quietly replacing good data with less (especially under write_mode: overwrite).

reconcile:
  count:                       # a source that returns the authoritative count
    type: postgres
    config:
      connection_url: ${secret:PG_URL}
      query: "SELECT count(*) AS n FROM orders WHERE updated_at >= '${now.date}'"
    count_field: n             # required when the probe row has more than one numeric field
  tolerance_pct: 0.0           # allow this % shortfall before failing (default 0)
  timeout_secs: 300            # time the probe may take to return its count (default 300)

The count probe is any faucet source (a SQL count(*), an OData $count endpoint via rest, …); its first returned record supplies the count. A probe that has not returned within timeout_secs fails the run, so raise it for a count that is slow to compute. The run fails when rows_written < authoritative × (1 − tolerance_pct/100). Compares rows written to the destination, so it is most meaningful for straight loads / full-refreshes. The count describes one dataset, so reconcile: is refused on a config with more than one root row or a partitioned row (each invocation would be compared with the whole dataset’s count). Schema: faucet schema (the reconcile block).

verify

Opt-in content verification (#701): after every successful root run, compare the destination to the source by key — key ranges by digest (server-side where both backends share an algorithm, so matching ranges ship no rows), differing ranges bisected down to the keys — and fail the run on a mismatch. faucet verify runs the same comparison on demand.

verify:
  key: [id]                 # default: the sink's upsert key (required otherwise)
  columns: [name, amount]   # default: every column on either side
  exclude: ["_faucet_*"]    # names or `prefix*` globs (default)
  destination:              # default: the sink's own read-back (SQL sinks)
    type: postgres
    config: { connection_url: ${secret:PG_URL}, query: "SELECT * FROM orders" }
  ranges: 16                # first-pass key ranges
  leaf_rows: 1000           # bisect a differing range down to this many rows
  max_differences: 1000     # report cap (the count keeps going)
  max_rows_scanned: 5000000 # stop (truncated report) after this many rows
  normalize: { float_tolerance: 0.0, timestamps: true, numeric_strings: false }
  after_run: true           # verify after every successful root run
  fail_on_difference: true  # a mismatch fails the run
  repair: false             # re-sync differing keys through the sink first
  allow_delete: false       # let a repair delete destination-only rows

The source side runs through the row’s transforms and masking, so a deterministic mask matches on both sides. See the verification cookbook. Schema: faucet schema verify.

rollback

Opt-in undoable runs (#706): stamp the _faucet_run_id column, journal the before-image of every key an upsert/delete run touches (in the write’s own transaction), keep the table an overwrite replaces as <table>__faucet_prev, and record the pre-run bookmark, so faucet rollback --run <id> can undo the run and rewind the sync.

rollback:
  enabled: true
  journal: true         # before-images for upsert / delete runs
  keep_previous: true   # keep the replaced table of an overwrite
  retain: 10            # undoable runs kept per row

Requires a durable state: block (not memory) and a rollback-capable sink (postgres / sqlite / mysql in column mode); refused at load time otherwise, or when metadata_columns.enabled is false. faucet run prints each row’s run id. See the rollback cookbook. Schema: faucet schema rollback.

usage

Cost & usage accounting (#704) — the pricing table estimates are computed from. Accounting itself is always on: every invocation reports records, estimated bytes, round trips and connector-reported cost signals, priced against these rates. Every rate has a shipped default (public list prices, USD).

usage:
  pricing_file: ./pricing.yaml       # optional; merged under the inline table; relative to the config file
  pricing:
    currency: USD
    egress_per_gb: 0                 # bytes read when source + sink are not both local files
    object_storage: { read_per_1k_requests: 0.0004, write_per_1k_requests: 0.005 }
    warehouse:
      bigquery_per_tib_scanned: 6.25
      bigquery_streaming_per_gib: 0.05
      snowflake_per_credit: 3.0
    hosted_elt_per_million_rows: 15  # the `hosted_equivalent` comparison rate

Recording across runs (for faucet usage / GET /v1/usage) rides the catalog store or a faucet serve history backend. A config submitted to faucet serve may not set pricing_file. See the usage cookbook. Schema: faucet schema usage.

budget

Run budgets (#703): hard ceilings on what one invocation may move.

budget:
  max_records: 1000000        # the page that would cross it is refused whole
  max_bytes: 5368709120       # estimated bytes written, same rule
  max_duration_secs: 1800     # cancels cooperatively at the next page boundary
  allowed_sinks: [warehouse]  # sink template names / connector kinds; checked before anything runs

A refused page never lands and the bookmark stays put; the run fails with budget_exceeded. faucet run --max-records / --max-bytes / --max-duration-secs / --allowed-sink merge with the block (the stricter of each ceiling, the intersection of the sink lists). Applies to run, schedule and serve; a backfill is bounded by its window instead. See the usage cookbook. Schema: faucet schema budget.

pool

The execution pool faucet serve runs the config in (#957):

pool: private   # lowercase letters, digits, `-`, `_`; at most 63 characters

Only a replica started with --pool private claims the run, its shards and its re-claims. A request’s own pool (POST /v1/runs, a template run, a trigger’s run.pool, …) wins; with neither, the run goes to default. A source template may carry pool: too, and the composed config inherits it. A pool no live replica serves is refused at submit time with 422. Ignored by faucet run, schedule, mirror and backfill. See Execution pools.

pool_override says whether a run of the config as a registered template may ask for another pool: false (the default) refuses one, true allows any, a list allows those pools. It limits template runs a caller or a tenant starts; the server’s own triggers, schedules and fan-outs are not limited by it. A source template may carry pool_override: too.

pool: private
pool_override: [private, gpu]

notifications

(requires the notify build feature)

A list of rules that fan pipeline lifecycle / health events out to Slack, PagerDuty, or a signed webhook. Events: run_failure, run_success, sla_breach, circuit_open, contract_abort, dlq_threshold, scheduler_stuck, profile_drift, change_requested, budget_exceeded. Fires from every runtime; delivery never fails a run.

notifications:
  - name: oncall
    on: [run_failure, circuit_open, contract_abort]
    dedupe_window_secs: 300     # optional leading-edge coalesce
    min_severity: error         # optional floor: info|warning|error|critical
    channel:
      type: pagerduty           # slack | pagerduty | webhook — {type, config}
      config:
        routing_key: "${env:PAGERDUTY_ROUTING_KEY}"

Per-rule fields: name (unique), on (event kinds; empty = all), min_severity, dedupe_window_secs, dlq_threshold (min DLQ rows for the dlq_threshold event), and channel ({ type, config }). Channel secrets should come from ${env:...} / ${secret:...} so they are log-redacted. See the Notifications cookbook for channel details, metrics, and faucet notify test. Schema: faucet schema notifications.

The webhook channel additionally takes headers, hmac_secret / signature_header, and extra_fields (static values merged into the emitted body; a key colliding with a faucet-emitted field is rejected at load time). Its payload carries run_id / invocation_id / started_at / finished_at / duration_secs alongside the event, so it can drive an external job-status callback — under faucet serve, run_id is the id returned by POST /v1/runs. See the payload table.

partition

Split a row into N independent invocations over a chunked range (#479), each scoped by ${partition.*} tokens substituted into the connector configs. Set at the top level (applies to every root row) or on an individual matrix row, which overrides it.

partition:
  kind: integer            # integer | timestamp | offset
  from: 0
  to: 1000000              # or a probe: { from_source: {…}, value_path: "$.max_id" }
  chunk_size: 10000
  bounds: inclusive        # integer only — REQUIRED, no default
  to_unbounded: false      # defaults ON when `to` is discovered; the last chunk's `end` is then the largest integer

Kind	Fields	Tokens
integer	from, to, chunk_size, bounds, to_unbounded	start, end, index, id
timestamp	from, to, chunk_size, timezone	start, end, start_date, end_date, start_unix, end_unix, index, id
offset	total, chunk_size	offset, limit, index, id

bounds has no default — inclusive and half-open differ by one at every boundary, and picking wrong silently duplicates or drops a record per chunk. total (a count) exists only on offset and to (a key) only on integer, so the two cannot be confused.

Chunks are ordinary sibling rows: they share execution.max_concurrent, get per-chunk state keys, and obey execution.on_error. A partitioned row cannot be referenced by another row’s parent: or depends_on:. Schema: faucet schema partition. See Parallel range partitioning.

mirror

(Formerly replication: — still accepted, with a deprecation warning; renamed in #670 so replication means only bookmark-based incremental reads.)

Present only when you run faucet mirror. It turns the main pipeline (whose source is a CDC connector) into a snapshot→CDC mirror by adding a one-time bulk-read snapshot source. faucet run ignores this block, the same way it ignores schedule:.

mirror:
  mode: snapshot_then_cdc          # REQUIRED. Only mode in v1.
  continuous: true                 # After the snapshot, keep streaming CDC until SIGTERM. Default true.
  snapshot:                        # REQUIRED. The one-time bulk-read source.
    source:
      type: postgres               # A non-CDC query reader of the same upstream DB.
      config:
        connection_url: ${env:SOURCE_PG_URL}
        query: "SELECT * FROM public.orders"

Field	Type	Default	Description
mode	snapshot_then_cdc	required	Replication strategy. Only snapshot_then_cdc exists in v1: capture the CDC position, bulk-snapshot the table, then stream CDC from that position.
snapshot.source	connector	required	A non-CDC bulk-read source (e.g. postgres / mysql / mongodb running a query) pointing at the same upstream database. Back-fills the destination through pipeline.sink before CDC starts.
continuous	bool	true	When true, keep streaming CDC after the snapshot completes until Ctrl-C / SIGTERM; a transient CDC-phase failure is logged, backed off (capped, reset on success), and resumed from the persisted bookmark rather than crash-exiting. When false, drain CDC once and exit (surfacing a transient error as a non-zero exit).

Requirements (enforced at config-load time, also reported by faucet validate):

• pipeline.source must be a CDC connector — postgres-cdc, mysql-cdc, or mongodb-cdc (the capture-capable set).
• pipeline.sink should use write_mode: upsert with a key for a true mirror; an append sink validates with a warning (boundary duplicates are possible).
• A durable state: backend is required (file / redis / postgres) — memory is rejected, since the snapshot→CDC handoff and resume depend on the persisted phase marker and bookmark.
• No matrix: — a mirror is one pipeline; mirror several tables with tables: (below).
• For postgres-cdc, a permanent replication slot (slot_type: permanent, the default) is required so WAL is retained across the snapshot.

See the replication cookbook for the correctness model (capture-before-snapshot + upsert idempotency), the resume behaviour, and the per-database log-retention caveats.

Multi-table mirror (mirror.tables)

With a tables: block the mirror replicates a set of tables over one change stream (#731). pipeline.sink becomes a template (the table is filled in per table) and snapshot.source the discovery connection.

Field	Type	Default	Description
tables.include	list of globs	["*"]	Discovered table names to mirror (* any run, ? one character).
tables.exclude	list of globs	[]	Removed from include.
tables.new_tables	follow / ignore	follow	Snapshot and stream a matching table created after the mirror started.
tables.discover_interval_secs	int	300	How often discovery re-runs to notice created / dropped tables (0 = only when the stream shows an unknown table).
tables.without_primary_key	refuse / append	refuse	A table with no primary key is refused (shown in status) or mirrored append-only.
tables.destination	mapping	per sink kind	Sink-config patch per table; strings may use {table}, {table_name}, {schema}. Required for sinks without a single table field (file sinks).
tables.on_table_error	pause / fail	pause	A table whose pipeline keeps failing is paused (the rest continue) or fails the mirror.
tables.max_table_failures	int	3	Consecutive failed cycles before a table is paused.
tables.retry_paused_secs	int	300	Seconds before a paused table is re-snapshotted (0 = stay paused until restart). Every start of faucet mirror — including each one-shot (continuous: false) run — retries paused tables.
tables.lag_warning_secs	int	300	Flag an active table as lagging past this many seconds without a confirmed position.
per_table.<table>.key	list	discovered PK	Upsert key.
per_table.<table>.write_mode	string	template’s, else upsert	Destination write mode.
per_table.<table>.sink	mapping	—	Extra sink-config patch (after destination).
per_table.<table>.snapshot	mapping	—	Snapshot-source config patch (after the discovered selection).
per_table.<table>.schema_drift	schema: block	top-level schema:	Drift policy for this table.
snapshot.concurrency	int	4	Tables snapshotted in parallel.
snapshot.shards	int	0	Primary-key ranges per table snapshot (single-column integer key, a source with shard: support).

Requirements: a CDC source that routes records to tables (postgres-cdc, mysql-cdc, mongodb-cdc with a database scope, mssql-cdc, oracle-cdc, dynamodb in mode: streams), a snapshot source that supports discovery, per-transaction CDC pages (no batch_size: 0), and a durable state:. State: the marker lives at {name}::__replication__, each table’s position at {name}::{table}. See the cookbook.

backfill

Optional defaults for faucet backfill — the range itself always comes from the command line. faucet run ignores this block, the same way it ignores schedule: / mirror:. Whenever the block is present, faucet validate also checks that at least one root source references a ${backfill.*} / ${now.*} scoping token (an unscoped source would replay identical data into every window).

backfill:
  window: 1d                  # default --window: 45s / 30m / 6h / 1d / 1w
  concurrency: 4              # default --concurrency (max units in flight); default 1
  timezone: America/New_York  # default --timezone (IANA); default UTC

Field	Type	Default	Description
window	string	— (whole range as one unit)	Chunk duration for the requested range.
concurrency	int ≥ 1	1	Max concurrently-running window units.
timezone	string	UTC	IANA zone for date boundaries and ${now.*} rendering.

faucet schema backfill prints the JSON Schema. See the backfill cookbook for the token table, resume semantics, and the HTTP endpoint.

schedule

Present only when you run faucet schedule. Absent configs are rejected by that command with a hint to use faucet run instead. All fields except cron are optional.

schedule:
  cron: "0 2 * * *"               # REQUIRED. Standard 5-field cron, or 6-field with leading seconds.
  timezone: "UTC"                 # IANA timezone name. Default UTC.
  overlap_policy: skip            # skip | queue | forbid. Default skip.
  max_runs: null                  # null = run forever; N = exit 0 after N successful runs.
  max_consecutive_failures: null  # null = never exit on failure; N = exit non-zero after N straight failures.
  on_failure: continue            # continue | stop. Default continue.
  start_immediately: false        # Run once on startup before waiting for the first tick. Default false.
  run_timeout_secs: null          # Per-run wall-clock kill switch (seconds). Timed-out runs count as failed.
  shutdown_grace_secs: 30         # SIGTERM: wait this long for the in-flight run before aborting. Default 30.

Field	Type	Default	Description
cron	string	required	5-field standard Unix cron (MIN HOUR DOM MON DOW) or 6-field with a leading seconds field (SEC MIN HOUR DOM MON DOW). Validated at load time.
timezone	string	"UTC"	IANA timezone name (e.g. "America/Los_Angeles", "Europe/Berlin"). Affects how the cron expression is interpreted.
overlap_policy	skip | queue | forbid	skip	What to do when a tick fires while a run is already in flight. skip drops the tick; queue buffers one missed tick (in-memory only, lost on restart); forbid exits non-zero.
max_runs	integer | null	null	Stop the scheduler cleanly (exit 0) after this many successful runs. null means run forever. 0 is rejected as a config error.
max_consecutive_failures	integer | null	null	Exit non-zero after this many consecutive failed runs without a success in between. A successful run resets the counter. null means never exit on failures alone.
on_failure	continue | stop	continue	stop exits non-zero immediately after the first failed run. continue keeps scheduling; use max_consecutive_failures to bound sustained outages.
start_immediately	bool	false	When true, the first run fires right on startup before the cron clock reaches its first tick.
run_timeout_secs	integer | null	null	Per-run time limit in seconds. A run that exceeds this is cancelled at its next page boundary (flushing its sinks, within shutdown_grace_secs) and counts as a failure. null means no timeout.
shutdown_grace_secs	integer	30	On SIGTERM/SIGINT, wait this many seconds for the in-flight run to finish before forcibly aborting it.

Validation: faucet validate pipeline.yaml checks the schedule: block at parse time — bad cron syntax, unknown timezone names, max_runs: 0, and a cron expression that can never fire all produce a clear config error: schedule: … message before any run starts.

See the scheduling cookbook for worked examples, the DST/timezone details, the overlap-policy decision tree, and the full Prometheus metric set.

lineage

Optional. When present, every pipeline run emits OpenLineage RunEvents describing the job, its input/output datasets, inferred schemas, and column-level lineage. Emission never fails a run — transport errors are logged and counted but do not propagate.

lineage:
  namespace: prod.warehouse      # REQUIRED. Logical namespace for all jobs and datasets.
  transport:                     # REQUIRED. Where to send events.
    type: http                   # http | file | kafka (kafka requires lineage-kafka feature)
    config:
      url: ${env:MARQUEZ_URL}
  job_name: ${name}::${row_id}   # Default. Resolved per matrix row at run time.
  include_schema_facet: false    # Emit DatasetFacets.schema (inferred from a sample).
  include_column_lineage: false  # Emit column-level lineage where statically derivable.
  include_source_code_facet: false  # Emit resolved config as a sourceCode job facet (warns; may expose secrets).
  emit_on:
    start: true
    running: false               # RUNNING heartbeats; see heartbeat_interval.
    complete: true
    fail: true
    abort: true
  sample_records: 100            # Max records sampled for schema/column facets.
  heartbeat_interval: 30         # Seconds between RUNNING heartbeats (when emit_on.running is true).

See the Lineage cookbook for the full field reference, the three transports (HTTP, file, Kafka), the column-lineage support matrix, schema-facet behavior, and the Prometheus metrics (faucet_lineage_events_total, etc.).

metadata_columns

Optional (#510). Stamp _faucet_* run/lineage metadata columns onto every row via a connector-agnostic sink decorator (works for any sink). Off unless present.

metadata_columns:
  prefix: "_faucet"                       # column-name prefix (default)
  columns: [extracted_at, loaded_at, run_id, source]   # or add `sequence`

Field	Default	Description
enabled	true	Master switch (toggle off without removing the block).
prefix	_faucet	Column-name prefix ({prefix}_{column}).
columns	[extracted_at, loaded_at, run_id, source]	Subset of extracted_at / loaded_at (write time), run_id, source (connector kind), sequence (monotonic per-run ordinal).

extracted_at is captured at sink-decorator time (≈ loaded_at in this model). Non-object records pass through unchanged.

Before the first write, a column-mapped destination that already exists is checked for the metadata columns: missing ones are added (TEXT; sequence an integer) through the sink’s schema evolution, and a sink that cannot add columns fails the run rather than dropping them. A source field with a metadata column’s name is replaced, with a one-time warning.

catalog

Optional. When present, faucet run / schedule / mirror record every successful root invocation into the Data Movement Catalog — the persistent, cross-run store of datasets, schema timelines, volume/freshness stats, and lineage edges. Recording never fails a run. faucet serve ignores this block: it records into its --history backend automatically. Requires a build with the catalog feature (in --features full).

catalog:
  url: sqlite:./faucet-catalog.db   # REQUIRED. sqlite:<path> | postgres://… | memory
  sample_records: 100               # Records sampled per side for schema inference.
  datasets:                         # Owners + declared consumers (#707), optional.
    - dataset: file://./out/records.jsonl   # canonical URI (as `faucet catalog datasets` prints it) or the id
      owners: [data-platform]
      consumers:
        - name: records-dashboard
          kind: dashboard           # free-form
          contact: "#analytics"
          columns: [id, name]       # what it reads; empty = every column

datasets[] annotations are merged into the catalog after every run that touches the dataset (owners replace, consumers upsert by name) and are what faucet plan --impact names; the same annotation can be made with faucet catalog annotate or POST /v1/catalog/datasets/{id}/consumers.

SQL stores additionally require the matching serve-history-sqlite / serve-history-postgres build feature. Browse the store with faucet catalog, the /v1/catalog/* HTTP endpoints, or the web console’s Datasets / Lineage views. Schema: faucet schema catalog.

local_outputs

Optional. Retention policy for the local files this pipeline’s sinks write — jsonl, csv, and parquet paths on the local filesystem. Requires a build with the catalog feature (in --features full).

local_outputs:
  retention_days: 3   # override the runtime default (7) for this pipeline's outputs
  track: true         # record them at all (default true)

Without the block a pipeline still records its local outputs — so they can be listed in the console and reclaimed — and inherits the runtime’s default window (--local-output-retention-days / FAUCET_LOCAL_SINK_OUTPUT_RETENTION_DAYS, itself 7 days). retention_days: 0 keeps this pipeline’s outputs forever; the window is measured from the last write, so an output a local run keeps refreshing never expires underneath it.

track: false opts the pipeline out of the ledger entirely. Because the GC only ever deletes recorded paths, that also means its outputs are never automatically deleted — an opt-out of the bookkeeping, not just of the listing.

Recording needs somewhere to record to: the catalog: block’s store for faucet run / schedule / mirror, or the --history backend under faucet serve. With neither, tracking is inert and logs one line rather than failing the run.

Reclaim outputs with faucet cleanup, the console’s Datasets page, or the /v1/local-outputs* endpoints; the sweeper that runs them automatically is described under Local output retention. It deletes only files faucet recorded as its own sink outputs — never a glob, never a directory, and never a file faucet merely appended to — and never touches run history, catalog entries, or lineage. Schema: faucet schema local-outputs.

observability

Optional top-level block that enables runtime observability backends. All sub-blocks are independently optional; omitting the entire observability: key leaves the defaults (no Prometheus server, no OTLP export).

otel:

Pushes traces and metrics to any OTLP-compatible collector. Requires building the CLI with --features otel (included in full).

observability:
  otel:
    endpoint: "http://localhost:4317"
    protocol: grpc
    headers: {}
    sample_ratio: 1.0
    export: [traces, metrics]
    service_name: faucet
    timeout_secs: 10
    metric_interval_secs: 60

Field	Type	Default	Description
endpoint	string	http://localhost:4317 (grpc) / http://localhost:4318 (http)	OTLP collector URL. For http, if the URL does not already contain a per-signal path (/v1/traces, /v1/metrics), faucet appends it automatically.
protocol	grpc | http	grpc	Transport protocol. grpc uses tonic; http uses HTTP/Protobuf. The faucet CLI always runs inside a tokio runtime, so both work without extra setup.
headers	map<string, string>	{}	Extra headers sent on every export request — auth tokens, team keys, etc. Values are secret-interpolated the same as any config value (e.g. "${env:HONEYCOMB_KEY}").
sample_ratio	float	1.0	Head-based trace sampling probability, 0.0–1.0. 1.0 exports every trace; 0.1 keeps ~10%. Does not affect metric export.
export	list	[traces, metrics]	Which signals to push. Each element is traces, metrics or logs. logs ships every run-log line from a durable local buffer (see logs: below). Omit a signal to disable it entirely.
service_name	string	faucet	Value of the OpenTelemetry resource attribute service.name attached to every span and metric point.
timeout_secs	integer	10	Per-export timeout in seconds. Timed-out exports are counted in faucet_otel_export_failures_total but do not fail the run.
metric_interval_secs	integer	60	How often (in seconds) accumulated metric points are pushed to the collector.

Coexistence: observability.otel: and observability.prometheus: are fully independent; both can be active at the same time and metrics fan out to both exporters. Export failures are never propagated to the pipeline — they increment faucet_otel_export_failures_total{signal} and are logged.

logs:

Durable log shipping (#806), on when otel.export lists logs. Every run-log line is written to a local spool first and shipped over OTLP; the per-run watermark advances only when the collector acknowledges a batch. See Shipping logs.

observability:
  otel: { endpoint: http://collector:4317, export: [traces, logs] }
  logs:
    spool_dir: /var/lib/faucet/logs
    retention_secs: 86400

Field	Type	Default	Description
spool_dir	path	$XDG_STATE_HOME/faucet/logs (~/.local/state/faucet/logs)	Where faucet run / faucet schedule spool lines before shipping (one <run_id>.jsonl + cursor + meta + lock per run).
retention_secs	integer	86400	How long delivered lines stay in the spool.
buffer_max_age_secs	integer	604800	Undelivered lines older than this are dropped (counted; the run reports partially_dropped). Must be > 0.
buffer_max_bytes	integer	1073741824	Spool size bound; past it the oldest lines go first. Must be > 0.
flush_timeout_secs	integer	10	How long faucet run waits at exit for its lines to ship. The rest stays for the next process or faucet logs ship.
max_lines_per_run	integer	100000	Per-run line cap; lines past it are dropped and counted (max_lines).
notify_after_secs	integer	300	Emit log_export_failed once a run’s export has been failing this long.
link_template	string	—	A link to the run’s logs in the log service. Placeholders {run_id} {pipeline} {row} {tenant} {started_at} {ended_at}, URL-encoded.

Discovery & env files

run / validate / preview / schedule auto-discover faucet.yaml → .yml → .json in the current directory, and load a sibling .env unless --no-env-file is given (--env-file PATH points elsewhere).

Note

The authoritative, exhaustive grammar — including every matrix and template edge case — is in cli/README.md.