faucet-source-redis

A Redis source for the faucet-stream ecosystem. Reads records from a Redis list, stream, or a set of keys matched by a glob pattern, yielding each record as a serde_json::Value.

Built on a lazily-opened, reused MultiplexedConnection and each mode's native paging primitive (LRANGE, XRANGE, SCAN + MGET), so it streams page-by-page without buffering the whole dataset in memory. Reach for it to drain a work queue, replay a stream's history, or load a keyspace into any faucet-stream sink.

Feature highlights

• Three source modes — List (FIFO list elements), Stream (entries with IDs + fields), and Keys (glob-matched keys with their values).
• Native streaming in all three modes — stream_pages walks the list, stream, or keyspace in pages so the sink writes as data arrives.
• Server-side cursoring — Keys drives the SCAN cursor manually and MGETs a page at a time, so a million-key namespace never materializes in memory at once.
• Consumer-group reads — Stream + group/consumer uses XREADGROUP on the fetch_all path, draining all currently-pending new messages (not just one batch).
• JSON-aware values — list elements, stream field values, and key values are parsed as JSON when they parse, and returned as strings otherwise.
• Connection reuse — one multiplexed connection is opened on first use and cheaply cloned across every call (no per-call TCP/AUTH handshake).
• Secrets-safe — the connection URL is masked in Debug output and credentials are stripped from the lineage dataset URI.

Installation

# As a library:
cargo add faucet-source-redis
cargo add tokio --features full

# Or via the umbrella crate:
cargo add faucet-stream --features source-redis

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-redis

The Redis source is opt-in — it is not in the CLI's default feature set.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: redis
    config:
      url: redis://127.0.0.1:6379
      source_type:
        type: List
        key: jobs:pending
  sink:
    type: jsonl
    config:
      path: ./jobs.jsonl

faucet run pipeline.yaml

Configuration reference

RedisSourceConfig

source_type — RedisSourceType

A tagged enum (type discriminator). Pick exactly one variant.

List

Reads list elements via LRANGE. Each element is parsed as JSON, falling back to a JSON string.

Stream

Reads stream entries via XRANGE (streaming) or XREAD / XREADGROUP (the fetch_all consumer-group convenience path).

Consumer groups apply to fetch_all only. The streaming path (stream_pages, used by faucet run) always uses XRANGE and re-reads the whole stream every run — consumer-group acknowledgement can't compose with bookmark-checkpoint draining. If you set group/consumer and the pipeline streams, the source logs a one-shot warning; use the fetch_all path for XREADGROUP consume-once semantics, or drop group/consumer to silence it.

Each entry is returned as:

{ "id": "1234567890-0", "fields": { "field1": "value1", "field2": "value2" } }

Keys

Scans keys matching a glob pattern via SCAN, then MGETs each batch.

Each key-value pair is returned as:

{ "key": "user:123", "value": { "name": "Alice", "email": "alice@example.com" } }

Stream field values and key values are parsed as JSON when possible; otherwise returned as strings.

Examples

Drain a list into SQLite

version: 1
name: redis_to_sqlite
pipeline:
  source:
    type: redis
    config:
      url: redis://localhost:6379
      source_type:
        type: List
        key: jobs:pending
      max_records: 10000
  sink:
    type: sqlite
    config:
      database_url: sqlite:./cache.db
      table_name: jobs
      column_mapping:
        type: auto_map
      batch_size: 500

Replay a stream's history

version: 1
pipeline:
  source:
    type: redis
    config:
      url: redis://localhost:6379
      source_type:
        type: Stream
        key: order-events
      batch_size: 2000
  sink:
    type: jsonl
    config:
      path: ./order-events.jsonl

Consumer-group read

version: 1
pipeline:
  source:
    type: redis
    config:
      url: redis://localhost:6379
      source_type:
        type: Stream
        key: events
        group: analytics-group
        consumer: worker-1
        count: 200
  sink:
    type: stdout
    config:
      format: jsonl

Consumer-group fields (group / consumer) apply only to the XREADGROUP path used by fetch_all. Streaming via stream_pages always uses XRANGE.

Scan a keyspace by pattern

version: 1
pipeline:
  source:
    type: redis
    config:
      url: redis://localhost:6379
      source_type:
        type: Keys
        pattern: "session:*"
      max_records: 500
  sink:
    type: jsonl
    config:
      path: ./sessions.jsonl

Streaming & batching

RedisSource overrides Source::stream_pages so the pipeline writes pages to the sink as they arrive instead of buffering the full result set. Each mode maps batch_size onto its native paging primitive:

batch_size = 0 is the "no batching" sentinel — every mode drains its primitive into a single page: LRANGE 0 -1 for List, XRANGE - + for Stream, and the full SCAN cursor followed by one MGET for Keys. Useful for small lookup tables or for sinks (SQL COPY, BigQuery load jobs) that prefer one large request.

The trait-level batch_size argument passed to stream_pages is ignored in favour of RedisSourceConfig::batch_size — the config is the authoritative, user-facing knob, so a pipeline-supplied hint can never silently override an explicit config value.

Every emitted page carries bookmark: None — the Redis source has no incremental-replication / resume mode today, so it is not resumable and does not support effectively-once delivery.

Config loading

use faucet_core::config::{load_json, load_env_file};
use faucet_source_redis::RedisSourceConfig;

let config: RedisSourceConfig = load_json("config.json")?;
let config: RedisSourceConfig = load_env_file(".env", "REDIS_SOURCE")?;

# .env
REDIS_SOURCE_URL=redis://127.0.0.1:6379
REDIS_SOURCE_MAX_RECORDS=1000

Schema introspection

faucet schema source redis

use faucet_core::Source;
let source = faucet_source_redis::RedisSource::new(config)?;
println!("{}", serde_json::to_string_pretty(&source.config_schema())?);

Library usage

use faucet_source_redis::{RedisSource, RedisSourceConfig, RedisSourceType};
use faucet_core::Source;

# async fn example() -> Result<(), Box<dyn std::error::Error>> {
// List mode, capped at 100 records.
let config = RedisSourceConfig::new(
    "redis://127.0.0.1:6379",
    RedisSourceType::List { key: "notifications".into() },
)
.max_records(100)
.with_batch_size(500);

let source = RedisSource::new(config)?;
let records = source.fetch_all().await?;
for record in &records {
    println!("{record}");
}

// Stream mode with a consumer group.
let config = RedisSourceConfig::new(
    "redis://127.0.0.1:6379",
    RedisSourceType::Stream {
        key: "order-events".into(),
        group: Some("analytics-group".into()),
        consumer: Some("worker-1".into()),
        count: Some(200),
    },
);
let source = RedisSource::new(config)?;
let events = source.fetch_all().await?;
for event in &events {
    println!("event {}: {:?}", event["id"], event["fields"]);
}
# Ok(()) }

RedisSource::new is synchronous and does no I/O — the connection is opened lazily on first use. It returns Err(FaucetError::Config) only on an out-of-range batch_size. To drive it through the streaming pipeline, hand the source to faucet_core::run_stream or Pipeline.

How it works

The multiplexed connection is opened once on first use (OnceCell) and cloned per call — MultiplexedConnection shares a single underlying socket, so there is no repeated TCP/AUTH handshake. List and stream pages advance a server-side cursor (a sliding LRANGE window; the immediate-successor stream ID via next_stream_id) so no entry is re-emitted across page boundaries. Keys drives the SCAN cursor manually and MGETs a page as soon as batch_size keys accumulate, rather than materializing the whole matched keyset first; keys deleted between SCAN and MGET are dropped.

List paging consistency caveat: index-based LRANGE paging is only stable if the list is not mutated mid-scan. A concurrent LPUSH/LPOP shifts every element's index, so a writer pushing/popping while the source drains can skip or duplicate elements across page boundaries. For a queue being consumed concurrently, prefer a Redis Stream (XRANGE / consumer groups) over a list.

Lineage dataset URI

redis://<host>:<port>/<db>?key=<key> (List / Keys) or ?stream=<stream> (Stream), with credentials stripped — e.g. redis://localhost:6379/0?key=jobs:pending.

Feature flags

This crate has no optional features of its own. Enable it in the CLI / umbrella crate via the source-redis feature.

Troubleshooting / FAQ

See also

• Connector reference · faucet-sink-redis · faucet-source-kafka

License

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