Configuration file format

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

version: 1                 # required, must be 1
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
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 still passed through verbatim to that connector.

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.

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.

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.

# 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: csv,   config: { path: ./data/input.csv } }
  sink:   { type: jsonl, 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. 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.
• 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).

Reference cycles surface as a clear InterpolationCycle error.

${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:

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 caveat: ${now.*} substitutes as plain text into config values — the same semantics as ${row_id.path} tokens. For SQL sources that interpolate ${now.*} into a query string, prefer the connector’s bind-parameter path (substitute_context_bind_params) over raw text substitution to avoid injection risk.

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.

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] }

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.

params:
  region:          { default: com }
  accounts_domain: { computed: "${map:region|ca=zohocloud|*=zoho}" }
# region=ca → accounts.zohocloud.ca ; anything else → accounts.zoho.<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, 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:.

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.

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

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: jsonl, 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 dot-path, optionally $-prefixed; $ = 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&…) 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:; 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” (HubSpot custom objects, Salesforce describe, Airtable, …). 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: hs_schemas }, select: "$.name", as: name }
  - id: props                      # 2. per type, collect its property names
    for_each: [types]
    fan_out:
      source: { ref: hs_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: hs_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/hubspot_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.

What applies in topology mode. Every top-level block does, each 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}; 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. So do --dry-run / --limit, ${now.*} / --clock, state:, and dlq:. See the applies-per-node table for the detail.

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 lowest committed sequence, so no sink is resumed past its own progress. 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

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). 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.

delivery

Controls the delivery guarantee for every pipeline row.

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

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: { ... }

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)

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 | 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)

Detectors are conservative (fully anchored; credit_card requires a valid Luhn checksum; ssn excludes never-issued ranges) so false positives stay rare. 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).

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 (Salesforce Bulk, 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).

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 no successful run within 2h
  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

At least one of the three checks 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

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             # optional; defaults to the first numeric field
  tolerance_pct: 0.0           # allow this % shortfall before failing (default 0)

The count probe is any faucet source (a SQL count(*), an OData $count endpoint via rest, …); its first returned record supplies the count. 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. 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.

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. 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

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"

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: — replication is a single pipeline in v1.
• 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.

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

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.

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`

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

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.

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

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.

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).

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