meathook-rs 0.4.0

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 the interval elapsed, the record limit was reached, or someone called flush(). 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, it drains each sink stack before exiting.

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, 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))
                // 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 an outer memory buffer, a durable JSONL spool, TOML configuration, and graceful shutdown.

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.

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.

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. The final flush moves records through the memory and JSONL tiers before it commits them to Hugging Face:

^C INFO meathook::runtime: received ctrl-c; shutting down
   INFO meathook::pipeline: pipeline shutting down; draining sink stack pipeline=air_temperature
   INFO meathook::sink::huggingface: committed window to hugging face pipeline=air_temperature ...
   INFO meathook::runtime: meathook runtime stopped

Successful commits remove their JSONL segment files. Pipeline directories stay in place, so the following command should produce no output:

find ./spool-test -type f

spool-test holds retry and crash-recovery data, not the final dataset. When Hugging Face rejects a window, the segment stays there and is replayed on the next run. The durability section explains what the spool covers and what an outer MemStore does not.

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.

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.