meathook-rs 0.1.2

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.:
                                                                 │
                          Buffered(mem, flush: 5m | 10k records) │  tier 1
                                       DiskSpool(flush: 1h)      │  tier 2
                                  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. 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, 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())
                .spooled("/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

DiskSpool 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 (data/{pipeline}/{YYYY-MM-DD}/{HH}.parquet), so replays are idempotent.

HuggingFace sink

HfSink commits one parquet 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 Hive-style partitioning keeps the HF dataset viewer happy.

Retry/backoff is deliberately not in the sink: the upstream DiskSpool or Buffered tier retains records when the sink errors and retries at its next firing.

Feature flags

With --no-default-features you get the core traits, sink combinators, write-ahead spool, and supervised runtime — 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 DiskSpool 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:

• Apache License, Version 2.0 (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

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.