faucet-state-postgres

PostgreSQL-backed StateStore for the faucet-stream ecosystem. Persists incremental-replication bookmarks in a single JSONB-backed table so sources can resume exactly where they left off after a crash, restart, or scheduled invocation.

Reach for it when you run faucet pipelines across multiple machines (or in Kubernetes) and want durable, shared resume state in a database you already operate — instead of a local file store that lives on one node's disk.

Why durable state?

faucet-stream sources that support incremental replication (CDC sources like postgres-cdc / mysql-cdc / mongodb-cdc, and query sources with a watermark column) emit a bookmark after each committed batch — the replication position they reached. A StateStore persists that bookmark; on the next run the pipeline reads it back and resumes from there, so a restart never re-emits already-delivered rows or skips data.

The built-in memory store forgets everything on exit, and the built-in file store ties state to one node's filesystem. A Postgres-backed store gives you:

• Durability across restarts and crashes — state survives the process and the host.
• A shared source of truth — every replica of a faucet serve cluster, every scheduled tick, and every operator running faucet replicate reads and writes the same row, keyed by pipeline + matrix row.
• One less moving part — if you already run Postgres for the pipeline's source or sink, state lives in the same database you back up and monitor.

Who should use this crate

• End users reference it indirectly: add the CLI's state-postgres feature and point a pipeline's state: block at type: postgres (see below). You never import this crate directly.
• Library authors building their own pipeline runner construct [PostgresStateStore] and hand it to Pipeline::with_state_store / RunStreamOptions.

It implements the [faucet_core::state::StateStore] trait — nothing more — so it slots in anywhere a StateStore is accepted.

Installation

# As a library (you also need faucet-core for the StateStore trait + Pipeline):
cargo add faucet-core faucet-state-postgres

# In the CLI (opt-in state backend):
cargo install faucet-cli --features state-postgres

The postgres state backend is not in the CLI default build — enable the state-postgres feature explicitly (it is included in the full aggregate). The built-in memory and file backends ship in faucet-core and need no feature.

CLI configuration

Point a pipeline's top-level state: block at the postgres backend:

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: postgres-cdc
    config:
      connection_url: ${env:SOURCE_PG_URL}
      slot_name: faucet_slot
      publication: faucet_pub
      tables: [public.orders]
  sink:
    type: jsonl
    config:
      path: ./orders.jsonl
  state:
    type: postgres
    config:
      url: ${env:STATE_PG_URL}      # e.g. postgres://user:pass@localhost:5432/faucet
      table: faucet_state
      ensure_table: true

faucet run pipeline.yaml

state.config fields

The CLI uses a fixed pool size of 5 connections for the state store. Library callers can size the pool themselves via [PostgresStateStore::connect_with].

Tip: the state database can be the same Postgres instance as your source or sink — just give it its own url/table. It does not have to be a separate server.

Library usage

Build it and hand it to a Pipeline

use std::sync::Arc;
use faucet_core::{Pipeline, state::StateStore};
use faucet_state_postgres::PostgresStateStore;

# async fn run(source: impl faucet_core::Source, sink: impl faucet_core::Sink)
#     -> Result<(), faucet_core::FaucetError> {
// Default: 5 connections, `faucet_state` table.
let store = PostgresStateStore::connect("postgres://user:pass@localhost/faucet").await?;
store.ensure_table().await?;            // optional — skip if you manage schema yourself
let store: Arc<dyn StateStore> = Arc::new(store);

Pipeline::new(&source, &sink)
    .with_state_store(store)
    .run()
    .await?;
# Ok(())
# }

Custom pool size, table name, or a shared pool

use faucet_state_postgres::PostgresStateStore;
use sqlx::PgPool;

# async fn build() -> Result<(), faucet_core::FaucetError> {
// Explicit pool size + table name.
let store = PostgresStateStore::connect_with(
    "postgres://user:pass@localhost/faucet",
    10,                 // max_connections
    "pipeline_state",   // table
).await?;

// Or reuse a PgPool you already manage.
# let existing_pool: PgPool = unimplemented!();
let store = PostgresStateStore::from_pool(existing_pool, "faucet_state")?;
# let _ = store;
# Ok(())
# }

Use the store directly

PostgresStateStore implements StateStore, so you can read and write bookmarks yourself:

use faucet_core::state::StateStore;
use faucet_core::serde_json::json;
use faucet_state_postgres::PostgresStateStore;

# async fn demo() -> Result<(), faucet_core::FaucetError> {
let store = PostgresStateStore::connect("postgres://user:pass@localhost/faucet").await?;
store.put("postgres-cdc:faucet_slot", &json!({ "lsn": "0/16B3748" })).await?;
let bookmark = store.get("postgres-cdc:faucet_slot").await?;   // Some(Value) or None
store.delete("postgres-cdc:faucet_slot").await?;               // missing key is not an error
# let _ = bookmark;
# Ok(())
# }

How it works

Storage layout

Every entry is one row in a single table:

CREATE TABLE faucet_state (
    key        TEXT        PRIMARY KEY,
    value      JSONB       NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

ensure_table() runs the CREATE TABLE IF NOT EXISTS form (the identifier is double-quoted via faucet_core::util::quote_ident). If you manage schema with migrations, skip it.

Operations

The pipeline reads the bookmark before fetching and writes it only after the sink confirms the batch, so a crash mid-write never advances state past delivered data.

Performance & safety

• Pooled connections — the sqlx::PgPool is built once in connect/connect_with and reused for every operation. State writes are cheap and infrequent (one per committed batch), so the default pool of 5 is intentionally small.
• Single-row upsert — put is one round-trip with no read-modify-write race; concurrent writers to the same key resolve via ON CONFLICT.
• Validated identifiers — table names are checked against Postgres identifier rules (non-empty, ≤ 63 chars, [A-Za-z0-9_]) at construction and double-quoted in every statement, so a misconfigured table name fails fast instead of risking SQL injection.
• Validated keys — state keys are checked by [faucet_core::state::validate_state_key] (ASCII alphanumerics plus _ - : ., ≤ 256 chars).

Health check

check() runs a real upsert → select → delete round-trip on a sentinel key (__faucet_doctor__), so faucet doctor verifies connectivity, credentials, table existence, and read/write permissions through the actual code path — and leaves no residue.

Feature flags

This crate has no optional features of its own. Enable the backend in the CLI / umbrella via the state-postgres feature.

Troubleshooting / FAQ

See also

• State & resumability cookbook — how bookmarks, resume, and exactly-once work end to end.
• Configuration reference — the full state: block grammar.
• faucet-core — the StateStore trait, Pipeline, and the built-in MemoryStateStore / FileStateStore.
• faucet-state-redis — the Redis-backed alternative for the same role.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.