fv_streams_runtime/

placement.rs

//! DETERMINISTIC PLACEMENT (D8 / D9): every worker computes the SAME task→worker map from the
//! topology and the sorted worker set, so no worker is told its placement and there is no control
//! plane that owns it (D9 — "zero control plane; the leader is a worker"). Two properties make that
//! sound:
//!
//! - **Deterministic and contiguous.** A stage's parallel instances are handed out in contiguous
//!   blocks ([`owner_of`], instance `i` of `n` → worker `⌊i·w/n⌋`), mirroring [`crate::dataflow::
//!   vnode_ranges`]. A worker therefore owns a contiguous span of a stage's vnode ranges — fewer
//!   cross-worker edges — and the same `(topology, sorted members)` reproduces the same map on every
//!   worker and across a restart.
//! - **Part of the fingerprint.** The member set folds into the checkpoint fingerprint
//!   ([`Members::fingerprint_clause`]), so a member-set change is a topology change that refuses to
//!   restore over the old checkpoint unless `STREAM_RESET_STATE=1` (D8 — restart-based rescale
//!   first; 4c does the authorised re-assignment). A **solo** run (no `--join`) contributes an empty
//!   clause, so its fingerprint — and thus every existing checkpoint — is byte-identical to the
//!   pre-Phase-4 runtime. No regression.
//!
//! The single-process runtime is unchanged: [`Members::solo`] places every task on the one worker,
//! and the exchange wiring (the remote edge wiring) then has no remote edges at all.
use std::collections::HashMap;

/// A worker's index within the sorted member set.
pub type WorkerId = u32;

/// The sorted, de-duplicated set of a run's workers, plus this process's own index. Sorting makes
/// the assignment independent of the order workers happened to join in — the invariant [`owner_of`]
/// relies on.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Members {
    /// Canonical worker addresses, sorted and de-duplicated. A solo run holds one empty entry.
    workers: Vec<String>,
    /// This process's index into `workers`.
    me: WorkerId,
}

impl Members {
    /// A single-worker run — the default when there is no `--join`. Byte-for-byte the pre-Phase-4
    /// behaviour: one worker owns every task, and the fingerprint clause is empty.
    pub fn solo() -> Self {
        Members {
            workers: vec![String::new()],
            me: 0,
        }
    }

    /// From a membership list (in any order) and this process's own address. The list is sorted and
    /// de-duplicated; `me` is this address's index in the sorted set. Errors if `my_addr` is not in
    /// the set (a worker must be a member of the run it joins).
    pub fn new(workers: Vec<String>, my_addr: &str) -> Result<Self, String> {
        let mut workers: Vec<String> = workers;
        workers.sort();
        workers.dedup();
        if workers.is_empty() {
            return Err("empty membership set".into());
        }
        let me = workers
            .iter()
            .position(|w| w == my_addr)
            .ok_or_else(|| format!("this worker ({my_addr}) is not in the membership set"))?
            as WorkerId;
        Ok(Members { workers, me })
    }

    /// The number of workers in the run.
    pub fn len(&self) -> usize {
        self.workers.len()
    }
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// This process's worker id.
    pub fn me(&self) -> WorkerId {
        self.me
    }

    /// A run with exactly one worker: every edge is local and there is nothing to coordinate across
    /// processes.
    pub fn is_solo(&self) -> bool {
        self.workers.len() == 1
    }

    /// The address of worker `w` (for opening exchange links to peers).
    pub fn addr(&self, w: WorkerId) -> Option<&str> {
        self.workers.get(w as usize).map(|s| s.as_str())
    }

    /// Every worker's id except this one — the peers this worker opens links to.
    pub fn peers(&self) -> impl Iterator<Item = WorkerId> + '_ {
        (0..self.workers.len() as WorkerId).filter(move |&w| w != self.me)
    }

    /// The fingerprint clause for this member set. **Empty for a solo run**, so a single-worker
    /// fingerprint — and every checkpoint written before Phase 4 — is unchanged. For a cluster, the
    /// worker *count*: placement ([`owner_of`]) depends only on the number of workers, so a change to
    /// it changes every task's owner and must refuse to restore over the old checkpoint (the D8
    /// fence), while a restart of the same-sized cluster — on whatever addresses/ports — restores
    /// cleanly.
    pub fn fingerprint_clause(&self) -> String {
        if self.is_solo() {
            String::new()
        } else {
            format!("workers={};", self.workers.len())
        }
    }
}

/// The owner of parallel instance `i` of an `n`-way stage across `w` workers. Contiguous blocks,
/// mirroring [`crate::dataflow::vnode_ranges`] (instance `i` → worker `⌊i·w/n⌋`), so a worker owns a
/// contiguous vnode span and adjacent instances tend to share a worker (fewer remote edges). Stable
/// under restart with the same `w`. `w` and `n` are clamped to at least 1; `i ≥ n` is treated as the
/// last instance (defensive — callers pass `i < n`).
pub fn owner_of(i: usize, n: usize, w: usize) -> WorkerId {
    let w = w.max(1);
    let n = n.max(1);
    ((i.min(n - 1) * w) / n) as WorkerId
}

/// Place a whole graph: task id → worker, deriving each task's stage from the builder's `roles` and
/// `stage_of` (both indexed by task id, in creation = id order). A stage's parallel instances are
/// the tasks that share a `(role, stage)` key, in id order; each is spread by [`owner_of`]. Because
/// `owner_of` depends only on `(instance, n, workers)`, a 1:1 forward chain (upstream and downstream
/// stages of equal width, paired by instance) lands each pair on the same worker — a forward edge
/// stays in-process — while a keyBy/shuffle stage of a different width spreads across workers. A
/// solo run (`members.len() == 1`) places everything on worker 0, so `start` and a solo
/// `start_worker` are identical.
pub fn assign(roles: &[&str], stage_of: &[usize], members: &Members) -> Vec<WorkerId> {
    assert_eq!(
        roles.len(),
        stage_of.len(),
        "roles and stage_of are both indexed by task id"
    );
    let w = members.len();
    let mut width: HashMap<(usize, &str), usize> = HashMap::new();
    for (r, s) in roles.iter().zip(stage_of) {
        *width.entry((*s, *r)).or_default() += 1;
    }
    let mut pos: HashMap<(usize, &str), usize> = HashMap::new();
    roles
        .iter()
        .zip(stage_of)
        .map(|(r, s)| {
            let key = (*s, *r);
            let i = pos.entry(key).or_insert(0);
            let owner = owner_of(*i, width[&key], w);
            *i += 1;
            owner
        })
        .collect()
}

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

    #[test]
    fn solo_places_everything_on_the_one_worker_and_adds_no_fingerprint() {
        let m = Members::solo();
        assert!(m.is_solo());
        assert_eq!(m.len(), 1);
        assert_eq!(m.me(), 0);
        // every instance of every stage is local (worker 0).
        for n in 1..=8 {
            for i in 0..n {
                assert_eq!(owner_of(i, n, m.len()), m.me(), "solo owns instance {i}/{n}");
            }
        }
        assert_eq!(m.peers().count(), 0, "a solo run has no peers");
        assert_eq!(
            m.fingerprint_clause(),
            "",
            "solo contributes nothing to the fingerprint"
        );
    }

    #[test]
    fn owner_is_contiguous_and_covers_every_worker() {
        // 8 instances across 3 workers → contiguous blocks [0,0,0][1,1,1][2,2] (⌊i·3/8⌋).
        let got: Vec<WorkerId> = (0..8).map(|i| owner_of(i, 8, 3)).collect();
        assert_eq!(got, vec![0, 0, 0, 1, 1, 1, 2, 2]);
        // blocks are contiguous (never 0,1,0) and every worker owns at least one when n ≥ w.
        let mut seen = [false; 3];
        for w in &got {
            seen[*w as usize] = true;
        }
        assert!(seen.iter().all(|&s| s), "every worker owns a block");
        // monotonic non-decreasing → contiguous.
        assert!(got.windows(2).all(|p| p[0] <= p[1]), "assignment is contiguous");
    }

    #[test]
    fn owner_matches_vnode_ranges_so_a_worker_owns_a_contiguous_vnode_span() {
        // The parallel instances that vnode_ranges splits a stage into are the same instances
        // owner_of assigns; a worker's instances must form a contiguous vnode span (no holes).
        let (vnodes, n, w) = (64u32, 8usize, 3usize);
        let ranges = crate::dataflow::vnode_ranges(vnodes, n as u32);
        for me in 0..w as WorkerId {
            let mine: Vec<usize> = (0..n).filter(|&i| owner_of(i, n, w) == me).collect();
            if mine.is_empty() {
                continue;
            }
            // contiguous instance ids → contiguous, gap-free vnode span.
            assert!(
                mine.windows(2).all(|p| p[1] == p[0] + 1),
                "worker {me}'s instances are contiguous"
            );
            let span_start = ranges[mine[0]].start;
            let span_end = ranges[*mine.last().unwrap()].end;
            let covered: u32 = mine.iter().map(|&i| ranges[i].end - ranges[i].start).sum();
            assert_eq!(span_end - span_start, covered, "worker {me}'s vnode span has no holes");
        }
    }

    #[test]
    fn new_sorts_and_dedups_so_assignment_is_join_order_independent() {
        // Two workers that joined in different orders must agree on who is who.
        let a = Members::new(vec!["10.0.0.2:7000".into(), "10.0.0.1:7000".into()], "10.0.0.1:7000").unwrap();
        let b = Members::new(vec!["10.0.0.1:7000".into(), "10.0.0.2:7000".into()], "10.0.0.2:7000").unwrap();
        assert_eq!(a.len(), 2);
        assert_eq!(
            a.me(),
            0,
            "the lexicographically-smaller address is worker 0 on every node"
        );
        assert_eq!(b.me(), 1);
        assert_eq!(
            a.fingerprint_clause(),
            b.fingerprint_clause(),
            "both derive the same fingerprint"
        );
        // a duplicate address collapses.
        let dup = Members::new(vec!["x".into(), "x".into(), "y".into()], "y").unwrap();
        assert_eq!(dup.len(), 2);
    }

    #[test]
    fn new_rejects_a_worker_that_is_not_a_member() {
        assert!(Members::new(vec!["a".into(), "b".into()], "c").is_err());
        assert!(Members::new(vec![], "a").is_err());
    }

    #[test]
    fn assign_co_locates_forward_chains_and_spreads_shuffle_stages() {
        // A pipeline: 4 sources -> 4 forward operators -> a 2-way keyBy stage -> 1 sink.
        let roles = [
            "sources",
            "sources",
            "sources",
            "sources", // stage 0
            "operators",
            "operators",
            "operators",
            "operators", // stage 1 (forward 1:1)
            "operators",
            "operators", // stage 2 (keyBy, width 2)
            "sinks",     // final
        ];
        let stage_of = [0, 0, 0, 0, 1, 1, 1, 1, 2, 2, 2];
        let two = Members::new(vec!["a".into(), "b".into()], "a").unwrap();
        let of = assign(&roles, &stage_of, &two);
        // sources spread [0,0,1,1]; the forward operators land on the SAME workers (co-located).
        assert_eq!(&of[0..4], &[0, 0, 1, 1], "sources spread across workers");
        assert_eq!(&of[4..8], &[0, 0, 1, 1], "the forward chain co-locates with its source");
        assert_eq!(&of[8..10], &[0, 1], "the keyBy stage spreads over both workers");
        assert_eq!(of[10], 0, "the single sink lands on one worker");

        // A solo run places everything on worker 0 — identical to `start`.
        let solo = assign(&roles, &stage_of, &Members::solo());
        assert!(solo.iter().all(|&w| w == 0), "a solo run is all local");
    }

    #[test]
    fn a_member_set_change_changes_the_fingerprint_but_solo_never_does() {
        let two = Members::new(vec!["a".into(), "b".into()], "a").unwrap();
        let three = Members::new(vec!["a".into(), "b".into(), "c".into()], "a").unwrap();
        assert_ne!(
            two.fingerprint_clause(),
            three.fingerprint_clause(),
            "adding a worker is a topology change (the D8 fence)"
        );
        // the SAME set from a different member is the same clause (co-derivable).
        let two_from_b = Members::new(vec!["b".into(), "a".into()], "b").unwrap();
        assert_eq!(two.fingerprint_clause(), two_from_b.fingerprint_clause());
        assert_eq!(
            Members::solo().fingerprint_clause(),
            "",
            "a solo run is fingerprint-stable forever"
        );
    }
}