dyniak 1.1.0

Riak-compatible protocol surface (HTTP + PBC) and storage bridge for the Dynomite Rust port
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

//! MapReduce ("pipes") for the Riak compatibility layer.
//!
//! Riak ships a JavaScript / Erlang MapReduce engine. This crate
//! takes a different bet: a fixed registry of named built-in phase
//! functions written in Rust. Operators name a function in a
//! [`Phase`] (for example `"map_object_value"`), and the framework
//! dispatches to the matching Rust impl. The rationale is recorded
//! in `docs/journal/2026-05-24-dyniak-mapreduce.md`.
//!
//! # Architecture
//!
//! A [`MapReduceJob`] has two pieces:
//!
//! * [`Inputs`] -- the seed values for the pipeline. Either an
//!   explicit list of `(bucket, key)` pairs, an inline list of
//!   `KeyDatum` objects (with values inline), or a bucket name (all
//!   keys in the bucket, enumerated by the executor through a
//!   datastore's `list_keys_stream`).
//! * [`Phase`] list -- an ordered pipeline of [`Phase::Map`],
//!   [`Phase::Reduce`], [`Phase::Link`] and [`Phase::WasmModule`]
//!   phases.
//!
//! Execution flows through the [`executor::run_job`] entry point.
//! Each phase is wired between two [`tokio::sync::mpsc`] channels;
//! the executor keeps one task per phase, the previous phase's
//! outbound is the next phase's inbound, and the final phase
//! outbound is collected into the response envelope. The pipe
//! shape follows Riak's "pipe of phases" pattern.
//!
//! # Determinism
//!
//! Built-in phase functions are pure Rust and side-effect-free.
//! `tokio::sync::mpsc` preserves FIFO. Each phase processes its
//! inbound queue serially. The result of running the same job
//! against the same input list is therefore byte-identical across
//! runs.
//!
//! # Examples
//!
//! ```
//! use dyniak::mapreduce::{
//!     builtins, executor::run_job, registry::PhaseRegistry, Inputs,
//!     KeyDatum, MapReduceJob, Phase,
//! };
//! use std::sync::Arc;
//!
//! # tokio::runtime::Builder::new_current_thread()
//! #     .enable_all().build().unwrap().block_on(async {
//! let registry = Arc::new(builtins::default_registry());
//! let job = MapReduceJob {
//!     inputs: Inputs::KeyData(vec![
//!         KeyDatum::with_value("b", "k1", serde_json::json!(1)),
//!         KeyDatum::with_value("b", "k2", serde_json::json!(2)),
//!         KeyDatum::with_value("b", "k3", serde_json::json!(3)),
//!     ]),
//!     phases: vec![
//!         Phase::Map { fn_name: "map_object_value".into(), arg: None, keep: false },
//!         Phase::Reduce { fn_name: "reduce_sum".into(), arg: None, keep: true },
//!     ],
//!     timeout_ms: None,
//! };
//! let out = run_job(job, registry).await.unwrap();
//! assert_eq!(out.len(), 1);
//! assert_eq!(out[0].value, serde_json::json!(6));
//! # });
//! ```

pub mod builtins;
pub mod executor;
pub mod job;
pub mod phase;
pub mod registry;

#[cfg(feature = "wasm")]
pub mod wasm;

pub use crate::mapreduce::executor::{
    run_job, run_job_full, run_job_with_wasm, MrError, PhaseOutput, WasmHook,
};
pub use crate::mapreduce::job::{Inputs, KeyDatum, MapReduceJob};
pub use crate::mapreduce::phase::Phase;
pub use crate::mapreduce::registry::{MapFn, PhaseRegistry, ReduceFn};

// Streaming MapReduce entry points (per-phase output batches),
// added by the streaming HTTP `/mapred` slice. The buffered
// `run_job` entry stays unchanged for callers (PBC `RpbMapRedResp`)
// that still emit a single response.
pub use crate::mapreduce::executor::{
    run_job_streaming, run_job_streaming_full, run_job_streaming_with_wasm, PhaseBatch,
};

#[cfg(feature = "wasm")]
pub use crate::mapreduce::wasm::{
    load_modules_from_config, WasmLimits, WasmModuleStore, WasmStoreError,
};