meathook-rs 0.4.2

How it works

Collectors and sinks are wired together in Rust. There is no plugin format or YAML layer to learn: implement two traits, then assemble the pipeline in code.

tick(interval) ──► Collector::collect() ──► Vec<Record> ──► Sink::ingest(...)
                                                                 │
                                        a sink can contain several tiers:
                                                                 │
               Tier(MemStore, flush: 5m | 10k records) │  tier 1 (optional)
                        Tier(JsonlStore, flush: 1h)    │  tier 2 (optional)
                     HfSink(parquet encode+commit)     ▼  terminal sink

Build a SinkStack in the same order that records travel through it. Add the outer tier first, followed by any inner tiers, and finish with the terminal sink:

let sink = SinkStack::new()
    .tier(
        MemStore::new(),
        FlushPolicy::new(Duration::from_secs(300), 10_000),
    )
    .tier(
        JsonlStore::new("/var/lib/meathook/spool/air_temperature"),
        FlushPolicy::hourly(),
    )
    .terminal(hf_sink);
// Records pass through MemStore, JsonlStore, then HfSink.

Collector returns a batch of row-shaped records on each tick. SatayCollector turns a client generated by satay-rs, such as nea-rs, into a collector.

Sink accepts records. A Tier owns its records until its FlushPolicy fires because its UTC wall-clock window closed, the record limit was reached, or someone called flush(). Sink::advance() closes elapsed windows without forcing the current partial one. Each tier uses a Store. The crate includes MemStore and the write-ahead JsonlStore; custom stores can use SQLite, object storage, or another backend. SinkExt::tee sends every batch to two sinks.

Meathook runs each pipeline in its own Tokio task. If a task panics, the supervisor rebuilds it from its factory and retries with exponential backoff. On SIGTERM or Ctrl+C, each pipeline applies its configured ShutdownPolicy.

All traits use associated error types. The built-in errors are concrete thiserror enums, with no boxed errors in the pipeline.

Quick start

use std::time::Duration;

use meathook::{
    FlushPolicy, HfSink, JsonlStore, Meathook, Pipeline, SatayCollector,
    ShutdownPolicy, SinkStack,
};
use satay_reqwest::ReqwestActionExt as _;

#[tokio::main]
async fn main() -> Result<(), meathook::runtime::RuntimeError> {
    let client = reqwest::Client::new();
    let token = std::env::var("HF_TOKEN").expect("HF_TOKEN must be set");

    Meathook::builder()
        .pipeline(move || {
            // The factory runs again after a panic. This rebuilds the stack
            // and lets the spool recover any unfinished segments.
            let api = nea_rs::Api::new();
            let collector = SatayCollector::new(
                "air_temperature",
                client.clone(),
                move |client| {
                    let api = api.clone();
                    async move { api.air_temperature().send_with(&client).await }
                },
                |response| flatten(response), // Typed response to Vec<MyRecord>.
            );

            // JsonlStore fsyncs every collected batch before ingest returns.
            // HfSink uploads each hourly window as Parquet.
            let terminal =
                HfSink::new(client.clone(), "you/your-dataset", token.clone());
            let sink = SinkStack::new()
                .tier(
                    JsonlStore::new("/var/lib/meathook/spool/air_temperature"),
                    FlushPolicy::hourly(),
                )
                .terminal(terminal);

            Pipeline::new(collector, sink, Duration::from_secs(60))
                // Keep the active UTC hour in the durable JSONL segment on
                // graceful restart instead of uploading a partial chunk.
                .with_shutdown_policy(ShutdownPolicy::PreserveActiveWindow)
                // APIs that return the latest reading repeat data across polls.
                .with_key_fn(|r: &MyRecord| (r.station_id.clone(), r.timestamp.clone()))
        })
        .run()
        .await
}

Records are ordinary structs. The Parquet encoder only needs #[derive(Serialize, Deserialize)]; serde_arrow derives the Arrow schema from the record type.

examples/nea_weather.rs is a complete example. It runs three NEA pipelines with a durable JSONL spool, TOML configuration, and active-window preservation across graceful restarts. Read the examples guide for poll timing, UTC windowing, restart behavior, and shutdown policy details.

Durability

When a batch reaches a Tier backed by JsonlStore, the store writes each record as a JSON line and fsyncs the segment before ingest returns. It deletes the segment only after the next sink accepts the batch.

Output paths depend on the pipeline, window, and content:

data/{pipeline}/{YYYY-MM-DD}/{HH}-{MM}-{SS}-{hash}.parquet

Replaying a window produces the same bytes and overwrites the same file. Different payloads do not collide, including sub-hourly windows and windows split into chunks by max_records or an explicit force-flush. One logical wall window can therefore contain multiple physical files.

Wall-clock windows and graceful shutdown

Windows are aligned to UTC Unix-epoch boundaries. An hourly policy maps every record from 12:00 through 12:59 UTC to the same 12:00 segment. Pipeline startup and every poll call Sink::advance() before collection, so the first heartbeat at or after 13:00 delivers the closed 12:00 segment even if collection fails or returns no records.

ShutdownPolicy::FlushAll is the default and immediately delivers the active partial window. Choose ShutdownPolicy::PreserveActiveWindow for a durable JsonlStore tier to retain that segment locally across a graceful restart:

Pipeline::new(collector, sink, Duration::from_secs(60))
    .with_shutdown_policy(ShutdownPolicy::PreserveActiveWindow)

For example, stopping at 12:35 and restarting at 12:40 continues appending to the same 12:00 JSONL segment; it is delivered after 13:00. If the process never restarts, that partial data remains locally durable but unavailable downstream until an explicit flush(). This policy cannot preserve records held only in MemStore, so use FlushAll for stacks with active volatile custody or when immediate remote availability matters.

The compression setting is therefore part of replay identity. Before changing an existing deployment from uncompressed parquet to zstd (or changing zstd levels), drain its JSONL spool if duplicate rows are unacceptable. A segment uploaded before a crash but replayed afterward with different compression has the same logical records but different bytes, so it lands at a new content-hash path. Existing uncompressed files remain valid and can coexist with new zstd files.

The position of JsonlStore matters. Its protection starts when a batch reaches that tier and ends when the downstream sink accepts it. An outer MemStore reduces fsync traffic but leaves its current window in memory. A downstream MemStore may accept a batch, allowing the JSONL segment to be deleted before the terminal sink receives it. To keep write-ahead custody all the way to delivery, place JsonlStore directly before the durable or terminal sink, with no volatile tier on either side.

The table distinguishes batches already held by JSONL from records that are still in a volatile tier:

Hugging Face sink

HfSink commits one file for each window. Its hand-written CommitAction implements satay_runtime::Action without performing I/O itself. The action builds an NDJSON commit body with the file encoded inline as base64, then satay_reqwest sends it through the same transport used by collectors.

The encoder is replaceable through HfSink::encoder. ParquetEncoder is the default, JsonEncoder is always available, and the csv feature adds CsvEncoder. Files use Hive-style partitions so the Hugging Face dataset viewer can read them.

Parquet stays uncompressed by default for compatibility. Compression is part of the encoder type: use Zstd for level 1, or select a level from 1 through 22 with Zstd<LEVEL>. Values outside that range do not implement the parquet compression policy and fail to compile:

use meathook::{HfSink, ParquetEncoder, Zstd};

type DatasetParquet = ParquetEncoder<Zstd<3>>;

let sink = HfSink::<MyRecord>::new(client, repo, token)
    .encoder(DatasetParquet::new());

let uncompressed = ParquetEncoder::default();
let zstd_level_1 = ParquetEncoder::<Zstd>::new();

Compression is recorded per parquet column and is transparent to the Hugging Face dataset viewer and parquet readers.

The sink retries transport errors, HTTP 429 responses, and 5xx responses with backoff. If those retries run out, the upstream tier keeps the records and tries again when it next flushes.

Pipelines that write to the same repository should share a CommitGate via HfSink::gate. The gate queues simultaneous flushes in the client instead of sending them all to Hugging Face's per-repository commit queue. This commonly happens during shutdown, when every pipeline flushes its last window at once. Set a request timeout on gated clients so a stalled upload cannot hold the gate forever.

Hugging Face bucket sink

HfBucketSink (feature hf-bucket) writes windows to a Hugging Face storage bucket — hf://buckets/{namespace}/{bucket} — instead of a dataset repo. Buckets have no git layer: no commit queue, no file-count limits, and Xet chunk deduplication across the whole bucket.

let terminal = HfBucketSink::new(client, "you/your-bucket", token)
    .expect("xet session")?;
let sink = SinkStack::new()
    .tier(JsonlStore::new("/var/lib/meathook/spool/air_temperature"), FlushPolicy::hourly())
    .terminal(terminal);

Each window still lands at the same deterministic path (data/{pipeline}/{date}/{time}-{fingerprint}.parquet), uploaded through Xet and then registered with a single sans-IO BatchAction (POST /api/buckets/{bucket}/batch). The upload leg uses the official hf-xet crate — the same client huggingface_hub uses — so replays of identical bytes transfer almost nothing. There is no CommitGate equivalent: buckets accept writes without a per-repository queue.

The bucket must exist before the first write; create it with CreateBucketAction (409 decodes to AlreadyExists) or hf buckets create you/your-bucket --private.

examples/nea_weather_bucket.rs runs the same three NEA pipelines as the dataset example, sharing collectors, records, and config plumbing through examples/common/mod.rs — only the terminal sink and its config differ. Build it with cargo run --features hf-bucket --example nea_weather_bucket -- examples/meathook_bucket.toml.

Feature flags

With --no-default-features, the crate still provides the core traits, sink combinators, JSONL spool, supervisor, Encoder, and JsonEncoder. The Satay collector, Hugging Face sink, and Parquet encoder are disabled, so applications must supply their own collector and terminal sink.

Configuration

The library accepts typed values such as Duration, PathBuf, and FlushPolicy. Parsing configuration is the application's job. The example loads a small TOML file from examples/meathook.toml:

spool_dir = "/var/lib/meathook/spool"   # Persistent volume mount on Kubernetes.

[flush]                  # Policy for each pipeline's durable JSONL tier.
every = "1h"
max_records = 50_000

[sink.huggingface]
repo = "zeon256/sg-weather"
branch = "main"

[collectors.air_temperature]
interval = "1m"

Secrets are read from the environment. The Hugging Face sink requires HF_TOKEN; NEA accepts an optional X_API_KEY.

Running the example

HF_TOKEN=hf_... cargo run --example nea_weather -- examples/meathook.toml

Ctrl+C starts a graceful shutdown. These example pipelines use PreserveActiveWindow: closed windows are delivered, while each active UTC window remains in its JSONL segment for continuation after restart:

^C INFO meathook::runtime: received ctrl-c; shutting down
   INFO meathook::pipeline: pipeline shutting down; preserving active wall-clock window pipeline=air_temperature
   INFO meathook::runtime: meathook runtime stopped

Successful commits remove closed JSONL segments. The current segment remains, so this command shows the locally preserved active windows:

find ./spool-test -type f

spool-test holds active-window, retry, and crash-recovery data, not the final dataset. On restart, collection resumes in the current segment and closed or rejected segments replay deterministically. The durability section explains the custody guarantees.

Testing

cargo test                       # Timing, spool recovery, Parquet round trips,
                                 # and Hugging Face payload tests without I/O.

HF_TOKEN=hf_... MEATHOOK_TEST_REPO=you/meathook-test \
    cargo test --test hf_integration -- --ignored   # Commit to a scratch repository.

HF_TOKEN=hf_... MEATHOOK_TEST_BUCKET=you/meathook-test \
    cargo test --features hf-bucket --test hf_bucket_integration -- --ignored
                                 # Write to a scratch storage bucket (created
                                 # if missing; delete with `hf buckets delete`).

Security

Report vulnerabilities as described in SECURITY.md.

License

Choose either license:

• Apache License 2.0
• MIT License

Contributing

Unless you state otherwise, any contribution you intentionally submit for inclusion in this project is dual licensed under Apache-2.0 and MIT, with no additional terms or conditions.