car-sync 0.33.0

//! The executor **dispatch fence** (slice B6 of
//! `docs/proposals/multi-device-sync.md`, §"Layer 2 (safety): the idempotency
//! / fencing gradient" + §"Deep dive: the execution-lease / fencing
//! protocol").
//!
//! B5 shipped **deterministic ledger convergence + a durable idempotency
//! oracle** and was explicit that this is **NOT** exactly-once execution:
//!
//! > **The exactly-once EXECUTION gate is B6's dispatch fence** — a
//! > dispatch-time linearizable "am I still epoch N?" read plus the durable
//! > non-fenced oracle read **before** the external side effect. That is
//! > necessary, not an optimization; the fold gates the *ledger*, not the
//! > effect.
//!
//! This module is that gate. [`check_dispatch`] runs the two checks the
//! proposal requires at the **point of effect** (immediately before the
//! executor performs a side-effecting tool call for a leased run):
//!
//! 1. **The durable non-fenced idempotency read** —
//!    [`crate::fold::SyncState::committed_run`]. This is checked **first** and
//!    is **fence-independent**: a run that has ever committed (per the keep-all
//!    oracle carried in the checkpoint) must never re-execute, whatever the
//!    current epoch. This closes the failover-double-run: a new holder that
//!    legitimately steals the lease still sees the old holder's committed
//!    record and declines.
//! 2. **The linearizable epoch read** — [`LeaseCoordinator::current`]. Only a
//!    caller that is *still* the current holder at the epoch it claims may
//!    proceed. A stale-epoch zombie (paused past its TTL while another site
//!    stole the lease) is rejected here — the Kleppmann fence: a holder that
//!    *knows* it is stale never acts.
//!
//! The residual the proposal is honest about is unchanged: the fence cannot
//! stop a holder that pauses *after* passing this check and before its
//! external write lands (the external resource does not honor CAR's token).
//! That window is bounded by TTL + the write-ahead intent ledger +
//! reversibility, not eliminated — see the proposal's tier-3 discussion. This
//! module implements the part that IS closeable.

use crate::fold::SyncState;
use crate::lease::{LeaseCoordinator, LeaseError};
use serde::{Deserialize, Serialize};

/// The fence verdict for one dispatch attempt.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "decision", rename_all = "snake_case")]
pub enum FenceDecision {
    /// The oracle shows no prior commit AND the caller still holds the lease
    /// at the claimed epoch — the side effect may run. This is the ONLY
    /// verdict that authorizes an external write.
    Proceed,
    /// The durable idempotency oracle already holds a committed record for
    /// `(agent_id, run_id)` — the run executed (here or on another site) and
    /// must not run again. Fence-independent: returned whatever the epoch.
    AlreadyCommitted {
        /// The `op_id` of the committed oracle record (for logging / dedup).
        committed_op_id: String,
    },
    /// The caller is no longer the current holder at the claimed epoch — a
    /// stale zombie. Do not execute; re-acquire or stand by.
    StaleEpoch {
        /// The epoch the caller believed it held.
        claimed_epoch: u64,
        /// The coordinator's current epoch (`0` if unheld).
        current_epoch: u64,
        /// The device that currently holds the lease.
        current_holder: Option<String>,
    },
    /// No lease is held for this agent at all (never acquired, or released).
    /// Not authorized to execute a leased side effect.
    NotHeld { claimed_epoch: u64 },
}

impl FenceDecision {
    /// Is this the one verdict that authorizes the external side effect?
    pub fn may_dispatch(&self) -> bool {
        matches!(self, FenceDecision::Proceed)
    }
}

/// Run the dispatch fence for one leased run at the point of effect.
///
/// `state` is the caller's folded [`SyncState`] (checkpoint base + journal
/// tail — e.g. [`crate::session::SyncSession::state`]); it supplies the
/// fence-independent committed-run oracle. `coordinator` is the linearizable
/// lease register; [`LeaseCoordinator::current`] is the "am I still epoch N?"
/// read. The oracle read comes **first** so an already-committed run is
/// declined even by a caller that legitimately holds the current lease
/// (idempotency dominates liveness — the whole point of a keep-all oracle).
///
/// Returns [`FenceDecision::Proceed`] **iff** no prior commit exists AND the
/// caller `device_id` holds the current lease at `epoch`. Any other outcome
/// means *do not perform the side effect*.
pub fn check_dispatch(
    coordinator: &mut dyn LeaseCoordinator,
    state: &SyncState,
    agent_id: &str,
    run_id: &str,
    device_id: &str,
    epoch: u64,
) -> Result<FenceDecision, LeaseError> {
    // 1. Durable, fence-independent idempotency read FIRST. A committed run
    //    never re-executes regardless of who now holds the lease.
    if let Some(committed) = state.committed_run(agent_id, run_id) {
        return Ok(FenceDecision::AlreadyCommitted {
            committed_op_id: committed.op_id.clone(),
        });
    }

    // 2. Linearizable "am I still epoch N?" read.
    match coordinator.current(agent_id)? {
        None => Ok(FenceDecision::NotHeld { claimed_epoch: epoch }),
        Some(lease) => {
            if lease.holder == device_id && lease.epoch == epoch {
                Ok(FenceDecision::Proceed)
            } else {
                Ok(FenceDecision::StaleEpoch {
                    claimed_epoch: epoch,
                    current_epoch: lease.epoch,
                    current_holder: Some(lease.holder),
                })
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::fold::fold;
    use crate::lease::{InMemoryLeaseCoordinator, Intent, IntentStatus};
    use crate::oplog::{logical_clock, DeviceLog, Scope, Surface};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::sync::Arc;

    fn coord() -> (Arc<AtomicU64>, InMemoryLeaseCoordinator) {
        let t = Arc::new(AtomicU64::new(0));
        let reader = t.clone();
        (t, InMemoryLeaseCoordinator::new(Arc::new(move || reader.load(Ordering::SeqCst))))
    }

    /// A committed intent op for `run` at `epoch`, folded into a state.
    fn state_with_commit(run: &str, epoch: u64) -> SyncState {
        let mut dev = DeviceLog::new("dev-a");
        dev.set_wall_clock(logical_clock());
        let op = dev.append(
            Scope::Personal,
            Surface::Intent,
            Intent::new("milo", run, epoch, IntentStatus::Committed).payload(),
        );
        fold(&[op])
    }

    #[test]
    fn holder_at_current_epoch_may_dispatch() {
        let (_t, mut c) = coord();
        let lease = c.acquire("milo", "dev-a", 100).unwrap();
        let empty = SyncState::default();
        let decision =
            check_dispatch(&mut c, &empty, "milo", "run-1", "dev-a", lease.epoch).unwrap();
        assert_eq!(decision, FenceDecision::Proceed);
        assert!(decision.may_dispatch());
    }

    #[test]
    fn stale_epoch_holder_is_refused() {
        // dev-a holds epoch 1, pauses past TTL; dev-b steals epoch 2. dev-a's
        // dispatch at epoch 1 must be refused — it is no longer the holder.
        let (t, mut c) = coord();
        c.acquire("milo", "dev-a", 100).unwrap();
        t.store(200, Ordering::SeqCst);
        let stolen = c.acquire("milo", "dev-b", 100).unwrap();
        assert_eq!(stolen.epoch, 2);

        let empty = SyncState::default();
        let decision = check_dispatch(&mut c, &empty, "milo", "run-1", "dev-a", 1).unwrap();
        assert_eq!(
            decision,
            FenceDecision::StaleEpoch {
                claimed_epoch: 1,
                current_epoch: 2,
                current_holder: Some("dev-b".to_string()),
            }
        );
        assert!(!decision.may_dispatch(), "a stale-epoch holder must NOT dispatch");
    }

    #[test]
    fn already_committed_run_is_not_re_executed_even_by_the_current_holder() {
        // The idempotency oracle dominates: even the legitimate current holder
        // declines a run that already committed (the failover double-run kill).
        let (_t, mut c) = coord();
        let lease = c.acquire("milo", "dev-a", 100).unwrap();
        let state = state_with_commit("run-nightly", 1);
        let decision =
            check_dispatch(&mut c, &state, "milo", "run-nightly", "dev-a", lease.epoch).unwrap();
        match decision {
            FenceDecision::AlreadyCommitted { ref committed_op_id } => {
                assert_eq!(
                    *committed_op_id,
                    state.committed_run("milo", "run-nightly").unwrap().op_id
                );
            }
            other => panic!("expected AlreadyCommitted, got {other:?}"),
        }
        assert!(!decision.may_dispatch());
    }

    #[test]
    fn committed_oracle_beats_a_stale_epoch_check() {
        // Both hazards present: the run committed AND the caller is stale. The
        // oracle read runs first, so the verdict is AlreadyCommitted (the more
        // specific "it already ran" — either way, do not dispatch).
        let (t, mut c) = coord();
        c.acquire("milo", "dev-a", 100).unwrap();
        t.store(200, Ordering::SeqCst);
        c.acquire("milo", "dev-b", 100).unwrap();
        let state = state_with_commit("run-x", 2);
        let decision = check_dispatch(&mut c, &state, "milo", "run-x", "dev-a", 1).unwrap();
        assert!(matches!(decision, FenceDecision::AlreadyCommitted { .. }));
    }

    #[test]
    fn unheld_agent_is_not_authorized() {
        let (_t, mut c) = coord();
        let empty = SyncState::default();
        let decision = check_dispatch(&mut c, &empty, "milo", "run-1", "dev-a", 1).unwrap();
        assert_eq!(decision, FenceDecision::NotHeld { claimed_epoch: 1 });
        assert!(!decision.may_dispatch());
    }
}