faucet-state-redis

Redis-backed StateStore for the faucet-stream ecosystem. Persists incremental-replication bookmarks in Redis so a pipeline can resume exactly where it left off after a crash, restart, or scheduled invocation.

Why durable state

A faucet-stream pipeline records a bookmark — the source's resume position (a CDC LSN, a WHERE id > ? watermark, a Kafka offset) — every time the sink confirms a batch. Where that bookmark lives decides what happens after a restart:

• memory (built into faucet-core) — fast, but the bookmark dies with the process. Every run starts from scratch.
• file (built into faucet-core) — survives restarts on a single host, but isn't shared and doesn't fit ephemeral or multi-replica deployments.
• redis (this crate) — a network-accessible, shared store. Ideal when the runner is containerized, scheduled, or horizontally placed, and you want resume state to outlive any one process or host.

The bookmark is read before each fetch and persisted only after the sink confirms the batch — so a crash mid-write replays the last batch rather than losing it (at-least-once). This crate is the storage layer behind that guarantee; you wire it in once and the pipeline handles the rest.

Who depends on this

You only depend on faucet-state-redis directly if you are building a pipeline in Rust and want Redis-backed resume state. If you drive faucet-stream through the faucet CLI, you don't add this crate by hand — it's compiled in via the state-redis feature and selected with a state: block in your YAML config (see below).

Installation

# Library — add alongside faucet-core:
cargo add faucet-core
cargo add faucet-state-redis

# CLI — enable the Redis state backend (opt-in feature):
cargo install faucet-cli --features state-redis

The state-redis feature is not in the CLI's default build — enable it explicitly (or use --features full).

Configuration (CLI state: block)

In a faucet YAML/JSON config, select this backend with type: redis:

version: 1
pipeline:
  source:
    type: postgres-cdc
    config:
      # ... your CDC source ...
  sink:
    type: postgres
    config:
      # ... your sink ...
  state:
    type: redis
    config:
      url: redis://127.0.0.1:6379
      namespace: faucet

Point url at a secret directive to keep credentials out of the file, e.g. url: ${env:REDIS_URL} or url: ${secret:redis-url}.

Library usage

use std::sync::Arc;
use faucet_core::{Pipeline, state::StateStore};
use faucet_state_redis::RedisStateStore;

# async fn run(source: impl faucet_core::Source, sink: impl faucet_core::Sink)
#     -> Result<(), faucet_core::FaucetError> {
let store: Arc<dyn StateStore> = Arc::new(
    RedisStateStore::connect("redis://127.0.0.1:6379", "faucet").await?,
);

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

To share one Redis connection across several stores (e.g. one per pipeline, each with its own namespace), open the connection yourself and use from_connection:

use faucet_state_redis::RedisStateStore;

# async fn run() -> Result<(), faucet_core::FaucetError> {
let client = redis::Client::open("redis://127.0.0.1:6379")
    .map_err(|e| faucet_core::FaucetError::Config(e.to_string()))?;
let conn = client.get_multiplexed_async_connection().await
    .map_err(|e| faucet_core::FaucetError::State(e.to_string()))?;

let prod = RedisStateStore::from_connection(conn.clone(), "prod")?;
let staging = RedisStateStore::from_connection(conn, "staging")?;
# let _ = (prod, staging);
# Ok(())
# }

How it works

Storage layout

Each bookmark is a single Redis string keyed by namespace plus state key:

The namespace is fixed once in connect() / from_connection() and prefixed onto every key. The state_key is supplied by the pipeline (e.g. postgres-cdc:<slot> for a CDC source, or the CLI's {name}::{row_id} composite for matrix runs), so multiple sources sharing one namespace never collide.

Operations

The three StateStore methods map directly onto Redis commands on a namespaced key:

• get(key) → GET {namespace}:{key}, then parse the JSON; a missing key returns None.
• put(key, value) → SET {namespace}:{key} <serialized JSON>.
• delete(key) → DEL {namespace}:{key}; deleting a missing key is not an error.

Connections use redis::aio::MultiplexedConnection, which is cheaply cloneable and safe to share across concurrent tasks — the store clones it per call, so no locking or pooling is needed on the caller's side. The connection is opened once in connect() and reused for the lifetime of the store.

Validation & preflight

• State keys are validated by [faucet_core::state::validate_state_key] — ASCII alphanumerics plus _ - : ., max 256 characters, no leading dot.
• Namespaces are validated at construction: non-empty, and the same character set minus : (since the namespace itself becomes a key prefix). An invalid namespace or URL fails fast as FaucetError::Config.
• faucet doctor exercises a real put → get → delete round-trip on a sentinel key, so the preflight validates connectivity, auth, 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 it in the CLI or the faucet-stream umbrella crate via the state-redis feature.

Troubleshooting / FAQ

See also

• State & resume cookbook — bookmarks, resume semantics, and choosing a backend.
• faucet-core — the StateStore trait this crate implements, plus the built-in memory and file backends.
• faucet-state-postgres — the PostgreSQL-backed sibling backend.
• faucet-stream on GitHub — the full ecosystem.

License

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