meathook-rs 0.2.0

A polling runtime with composable, durable sinks
Documentation

How it works

There is no YAML plugin system: you wire collectors and sinks in code by implementing two traits, and sinks compose like tower layers.

tick(interval) ──► Collector::collect() ──► Vec<Record> ──► Sink::ingest(...)
                                                                 │
                              the "sink" is a composed stack of layers, e.g.:
                                                                 │
               Tier(MemStore, flush: 5m | 10k records) │  tier 1 (any number of
                        Tier(JsonlStore, flush: 1h)    │  tier 2  tiers, or none)
                     HfSink(parquet encode+commit)     ▼  terminal
  • Collector produces a batch of row-shaped records per tick. SatayCollector adapts any satay-generated client (e.g. nea-rs) into a collector.
  • Sink receives records. Each layer owns its records until its FlushPolicy fires (interval elapsed / max records / explicit flush()), then pushes downstream. Buffering is one generic layer, Tier, over a pluggable Store backend — in-memory MemStore, write-ahead JsonlStore, or bring your own (SQLite, object storage, ...). Fan-out is just another combinator (Tee).
  • Meathook runs one tokio task per pipeline, respawns panicked pipelines from their factory with exponential backoff, and drains every sink stack on SIGTERM/ctrl-c before exiting.

Errors are concrete thiserror enums end to end — traits carry an associated Error type, nothing is boxed.

Quick start

use std::time::Duration;

use meathook::{FlushPolicy, HfSink, JsonlStore, Meathook, Pipeline, SatayCollector, SinkExt as _};
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 is re-invoked after a panic, rebuilding the whole
            // stack (and re-running spool crash recovery).
            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 API response -> Vec<MyRecord>
            );

            // Durable stack: every tick is fsynced to disk before ingest
            // returns; hourly windows land on HF as parquet.
            let sink = HfSink::new(client.clone(), "you/your-dataset", token.clone())
                .tier(
                    JsonlStore::new("/var/lib/meathook/spool/air_temperature"),
                    FlushPolicy::hourly(),
                );

            Pipeline::new(collector, sink, Duration::from_secs(60))
                // "latest reading" APIs repeat across polls; dedupe by key
                .with_key_fn(|r: &MyRecord| (r.station_id.clone(), r.timestamp.clone()))
        })
        .run()
        .await
}

Records are plain structs — #[derive(Serialize, Deserialize)] is all the parquet encoder needs (the arrow schema is derived from the type via serde_arrow).

See examples/nea_weather.rs for the full reference consumer: three NEA pipelines, TOML config, graceful shutdown.

Durability

A Tier backed by JsonlStore is a write-ahead spool: ingest appends records as JSON lines to an fsynced segment file before returning, and segments are deleted only after the downstream sink accepted them. Storage paths are deterministic per window and content (data/{pipeline}/{YYYY-MM-DD}/{HH}-{MM}-{SS}-{hash}.parquet): a replayed window re-encodes to the same bytes and overwrites its own file, while distinct payloads — sub-hourly windows, or one window drained in chunks by the max_records valve — never collide.

Failure What happens Data lost
SIGKILL / OOM-kill leftover segments replayed on next start ≤ 1 torn record
Task panic supervisor respawns the pipeline from its factory none
Sink outage (HF 5xx) segments accumulate on disk, retried each firing none
Graceful SIGTERM runtime drains every sink stack before exit none
Disk lost spool gone everything unflushed — use a PVC on k8s

HuggingFace sink

HfSink commits one file per window via a hand-written, sans-IO CommitAction implementing satay_runtime::Action (NDJSON commit payload, base64-inlined file), sent through satay_reqwest — the same transport path as the collectors. The wire format is pluggable via HfSink::encoder: ParquetEncoder is the default, JsonEncoder is always available, and CsvEncoder sits behind the csv feature. The Hive-style partitioning keeps the HF dataset viewer happy.

Transient failures (transport errors, 429, 5xx) retry a few times in-sink with backoff; beyond that, an upstream Tier retains the records and retries at its next firing. Pipelines committing to the same repo should share a CommitGate (HfSink::gate) so concurrent flushes — e.g. every pipeline's final window on shutdown — queue client-side instead of piling into HF's per-repo commit concurrency queue; give gated clients a request timeout so a stalled upload cannot hold the gate indefinitely.

Feature flags

Feature Default Implies What it enables
parquet ✓ — Encoder trait + ParquetEncoder (arrow + parquet + serde_arrow)
csv — — CsvEncoder: CSV window encoding
satay — — SatayCollector: any satay-generated API client as a Collector
huggingface ✓ parquet, satay HfSink + sans-IO CommitAction

With --no-default-features you get the core traits, sink combinators, write-ahead spool, supervised runtime, and the always-on Encoder trait + JsonEncoder — no HTTP/IO stack pulled in (transitively satay-free) — bring your own collector and terminal sink.

Configuration

The library API takes typed values (Duration, PathBuf, FlushPolicy) via builders; config parsing lives in your binary. The example parses a small TOML file (examples/meathook.toml):

spool_dir = "/var/lib/meathook/spool"   # PVC mount on k8s

[flush]                  # FlushPolicy for each pipeline's spool tier
every = "1h"
max_records = 50_000

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

[collectors.air_temperature]
interval = "1m"

Secrets come from the environment only: HF_TOKEN (required by the HF sink), X_API_KEY (optional for NEA).

Running the example

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

Testing

cargo test                       # unit tests: paused-clock timing, spool recovery,
                                 # parquet round-trip, HF payload shape (sans-IO)

HF_TOKEN=hf_... MEATHOOK_TEST_REPO=you/meathook-test \
    cargo test --test hf_integration -- --ignored   # real commit to a scratch repo

Security

Please see SECURITY.md for vulnerability reporting.

License

Licensed under either of:

at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.