spate-coordination 0.1.0

Distributed work coordination backend for the Spate pipeline framework: a leader-elected planner enumerates weighted splits into a shared low-latency store (NATS JetStream KV) and publishes a desired assignment per instance; workers lease and heartbeat what they are assigned, cooperatively drain what they are not, and commit progress through epoch-fenced compare-and-swap. Applications should depend on the `spate` facade crate with the `coordination` feature (in-memory store) or `coordination-nats` (the NATS backend).
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

//! Error construction and classification helpers.

use crate::store::StoreError;
use spate_core::coordination::{CoordinationError, CoordinationErrorKind};

/// A [`Fatal`](CoordinationErrorKind::Fatal) coordination error.
pub(crate) fn fatal(reason: impl Into<String>) -> CoordinationError {
    CoordinationError::new(CoordinationErrorKind::Fatal, reason)
}

/// A [`Retryable`](CoordinationErrorKind::Retryable) coordination error.
pub(crate) fn retryable(reason: impl Into<String>) -> CoordinationError {
    CoordinationError::new(CoordinationErrorKind::Retryable, reason)
}

/// Map a store failure into the seam taxonomy, with context.
pub(crate) fn store_error(context: &str, e: &StoreError) -> CoordinationError {
    match e {
        StoreError::Retryable(reason) => retryable(format!("{context}: {reason}")),
        StoreError::Fatal(reason) => fatal(format!("{context}: {reason}")),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn store_errors_keep_their_class_and_gain_context() {
        let r = store_error("renewing lease", &StoreError::Retryable("timeout".into()));
        assert_eq!(r.kind, CoordinationErrorKind::Retryable);
        assert!(r.to_string().contains("renewing lease"));
        assert!(r.to_string().contains("timeout"));
        let f = store_error("probing store", &StoreError::Fatal("no CAS".into()));
        assert_eq!(f.kind, CoordinationErrorKind::Fatal);
    }
}