car-sync 0.34.0

//! Network sync backend — CAR's client for a **remote** relay + lease register
//! (the Parslee sync service), behind the same [`Relay`] / [`LeaseCoordinator`]
//! traits the local [`FsRelay`] / [`InMemoryLeaseCoordinator`] already satisfy.
//!
//! # Why
//!
//! `FsRelay` converges two devices only if they share a filesystem directory —
//! fine for one machine, useless for "my phone and my Mac". This module makes
//! the relay a *service*: [`NetworkRelay`] and [`NetworkLeaseCoordinator`] speak
//! to it over a pluggable [`SyncTransport`], scoped to the caller's Parslee
//! identity (`scope`, e.g. `user:<id>` or `org:<id>`). The Parslee backend holds
//! only what `FsRelay` holds — op metadata (`op_id`/`hlc`/`seq`) in the clear and
//! op *payloads* as E2E ciphertext (see [`crate::crypto`]) — so it can route and
//! GC without reading your config.
//!
//! # The transport seam
//!
//! [`SyncTransport`] is the wire contract (blocking, because [`Relay`] is sync;
//! an HTTP impl blocks inside these calls, which the daemon already runs off a
//! task). Two impls:
//!
//! - [`LoopbackTransport`] — an **in-process reference server**. Each scope is
//!   backed by a real [`FsRelay`] + [`InMemoryLeaseCoordinator`] over a temp
//!   dir, so `NetworkRelay`-over-loopback is *semantically identical* to
//!   `FsRelay` by construction (no re-implementation of the GC-floor / dedup /
//!   monotone-ack / dominance logic). It is the test double AND a usable
//!   single-host multi-process relay. `Clone` shares one server across N
//!   devices.
//! - `HttpTransport` (a later slice) — reqwest against the real Parslee endpoint.
//!   It must implement the exact same contract; the loopback is the executable
//!   spec.

use std::collections::BTreeMap;
use std::sync::{Arc, Mutex};

use serde::{Deserialize, Serialize};

use crate::checkpoint::Checkpoint;
use crate::lease::{InMemoryLeaseCoordinator, Lease, LeaseCoordinator, LeaseError};
use crate::oplog::{Hlc, OpRecord, WallClock};
use crate::relay::{
    AckOutcome, Frontier, GcReport, PullResult, PushOutcome, Relay, RelayConfig, RelayError,
    RosterEntry,
};

/// A transport-level failure — the *service* was unreachable or misbehaved.
/// Distinct from an in-band verdict (a deduped push, a `Held` lease): those are
/// success returns, not errors.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TransportError {
    /// The service could not be reached / timed out / returned a transport
    /// error (HTTP 5xx, connection reset, …).
    Unavailable(String),
    /// The caller is not authorized for `scope` (bad/expired Parslee token).
    Unauthorized(String),
    /// A reply could not be parsed into the expected shape (protocol drift).
    Protocol(String),
    /// A pull requested a frontier below the relay's GC floor for `device_id`
    /// (retained ops start at `dropped_below`). NOT a failure — the service's
    /// signal that the caller must **cold-bootstrap** from the latest
    /// checkpoint. Maps to [`RelayError::FrontierTruncated`], which
    /// [`crate::session::SyncSession::pump`] already handles by rebasing.
    FrontierTruncated {
        device_id: String,
        dropped_below: u64,
    },
}

impl std::fmt::Display for TransportError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TransportError::Unavailable(m) => write!(f, "sync transport unavailable: {m}"),
            TransportError::Unauthorized(m) => write!(f, "sync transport unauthorized: {m}"),
            TransportError::Protocol(m) => write!(f, "sync transport protocol error: {m}"),
            TransportError::FrontierTruncated {
                device_id,
                dropped_below,
            } => write!(
                f,
                "sync transport frontier truncated: device {device_id} ops below seq \
                 {dropped_below} were GC'd (cold-bootstrap from checkpoint)"
            ),
        }
    }
}
impl std::error::Error for TransportError {}

impl From<TransportError> for RelayError {
    fn from(e: TransportError) -> Self {
        match e {
            // The service's cold-bootstrap signal maps to the relay's own
            // FrontierTruncated, which the pump handles by rebasing.
            TransportError::FrontierTruncated {
                device_id,
                dropped_below,
            } => RelayError::FrontierTruncated {
                device_id,
                dropped_below,
            },
            // Everything else has no relay variant; surface it as an IO-ish
            // failure so the pump's best-effort loop treats it as "try later".
            other => RelayError::Io(std::io::Error::other(other.to_string())),
        }
    }
}

/// The serialized verdict of a lease-register call. Mirrors the coordinator's
/// `Result<_, LeaseError>` in a wire-portable shape; the client reconstructs the
/// real [`LeaseError`] from it (the `agent_id` is known client-side).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum LeaseWire {
    /// `acquire`/`renew` granted this lease.
    Granted(Lease),
    /// `acquire` CAS-failed: a still-valid lease is held.
    Held {
        holder: String,
        epoch: u64,
        expires_at_ms: u64,
    },
    /// `renew`/`release` found the caller is no longer the holder at its epoch.
    Lost {
        claimed_epoch: u64,
        current_epoch: u64,
    },
    /// `release` succeeded.
    Released,
    /// `current` read (holder or unheld).
    Current(Option<Lease>),
}

/// The wire contract for a remote sync service, scoped per Parslee identity.
///
/// Every method takes a `scope` (`user:<id>` / `org:<id>`) so one service backs
/// many tenants. Blocking by design. Impls MUST be faithful to [`FsRelay`] /
/// [`InMemoryLeaseCoordinator`] semantics — [`LoopbackTransport`] is the spec.
pub trait SyncTransport: Send + Sync {
    fn enroll(&self, scope: &str, device_id: &str) -> Result<RosterEntry, TransportError>;
    fn push(
        &self,
        scope: &str,
        device_id: &str,
        ops: &[OpRecord],
    ) -> Result<PushOutcome, TransportError>;
    fn pull(
        &self,
        scope: &str,
        device_id: &str,
        since: &Frontier,
    ) -> Result<PullResult, TransportError>;
    fn ack(
        &self,
        scope: &str,
        device_id: &str,
        frontier: Hlc,
    ) -> Result<AckOutcome, TransportError>;
    fn checkpoint_put(
        &self,
        scope: &str,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, TransportError>;
    fn checkpoint_get(&self, scope: &str) -> Result<Option<Checkpoint>, TransportError>;
    fn roster(&self, scope: &str) -> Result<Vec<RosterEntry>, TransportError>;
    fn stable_frontier(&self, scope: &str) -> Result<Option<Hlc>, TransportError>;
    fn gc(&self, scope: &str) -> Result<GcReport, TransportError>;

    // --- execution lease register (B5, distributed) ---
    fn lease_acquire(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        ttl_ms: u64,
    ) -> Result<LeaseWire, TransportError>;
    fn lease_renew(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<LeaseWire, TransportError>;
    fn lease_release(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
    ) -> Result<LeaseWire, TransportError>;
    fn lease_current(&self, scope: &str, agent_id: &str) -> Result<LeaseWire, TransportError>;
}

/// A [`Relay`] backed by a remote [`SyncTransport`], scoped to one Parslee
/// identity. One per device; share the transport (`Arc`) across devices.
pub struct NetworkRelay {
    transport: Arc<dyn SyncTransport>,
    scope: String,
}

impl NetworkRelay {
    pub fn new(transport: Arc<dyn SyncTransport>, scope: impl Into<String>) -> Self {
        Self {
            transport,
            scope: scope.into(),
        }
    }
}

impl Relay for NetworkRelay {
    fn register(&mut self, device_id: &str) -> Result<RosterEntry, RelayError> {
        Ok(self.transport.enroll(&self.scope, device_id)?)
    }
    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
        Ok(self.transport.push(&self.scope, device_id, ops)?)
    }
    fn pull(&mut self, device_id: &str, since: &Frontier) -> Result<PullResult, RelayError> {
        Ok(self.transport.pull(&self.scope, device_id, since)?)
    }
    fn ack(&mut self, device_id: &str, frontier: Hlc) -> Result<AckOutcome, RelayError> {
        Ok(self.transport.ack(&self.scope, device_id, frontier)?)
    }
    fn checkpoint_put(
        &mut self,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, RelayError> {
        Ok(self
            .transport
            .checkpoint_put(&self.scope, device_id, checkpoint)?)
    }
    fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError> {
        Ok(self.transport.checkpoint_get(&self.scope)?)
    }
    fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError> {
        Ok(self.transport.roster(&self.scope)?)
    }
    fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError> {
        Ok(self.transport.stable_frontier(&self.scope)?)
    }
    fn gc(&mut self) -> Result<GcReport, RelayError> {
        Ok(self.transport.gc(&self.scope)?)
    }
}

/// A [`LeaseCoordinator`] backed by a remote [`SyncTransport`] — the
/// **distributed** execution lease that makes "phone and Mac" mutually
/// exclusive on a given agent. Reconstructs the real [`LeaseError`] from the
/// transport's [`LeaseWire`] verdict.
pub struct NetworkLeaseCoordinator {
    transport: Arc<dyn SyncTransport>,
    scope: String,
}

impl NetworkLeaseCoordinator {
    pub fn new(transport: Arc<dyn SyncTransport>, scope: impl Into<String>) -> Self {
        Self {
            transport,
            scope: scope.into(),
        }
    }
}

fn wire_to_lease(agent_id: &str, wire: LeaseWire, ctx: &'static str) -> Result<Lease, LeaseError> {
    match wire {
        LeaseWire::Granted(l) => Ok(l),
        LeaseWire::Held {
            holder,
            epoch,
            expires_at_ms,
        } => Err(LeaseError::Held {
            agent_id: agent_id.to_string(),
            holder,
            epoch,
            expires_at_ms,
        }),
        LeaseWire::Lost {
            claimed_epoch,
            current_epoch,
        } => Err(LeaseError::Lost {
            agent_id: agent_id.to_string(),
            claimed_epoch,
            current_epoch,
        }),
        other => Err(LeaseError::Backend(format!(
            "{ctx}: unexpected lease verdict {other:?}"
        ))),
    }
}

impl LeaseCoordinator for NetworkLeaseCoordinator {
    fn acquire(
        &mut self,
        agent_id: &str,
        device_id: &str,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError> {
        let wire = self
            .transport
            .lease_acquire(&self.scope, agent_id, device_id, ttl_ms)
            .map_err(|e| LeaseError::Backend(e.to_string()))?;
        wire_to_lease(agent_id, wire, "acquire")
    }

    fn renew(
        &mut self,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<Lease, LeaseError> {
        let wire = self
            .transport
            .lease_renew(&self.scope, agent_id, device_id, epoch, ttl_ms)
            .map_err(|e| LeaseError::Backend(e.to_string()))?;
        wire_to_lease(agent_id, wire, "renew")
    }

    fn release(&mut self, agent_id: &str, device_id: &str, epoch: u64) -> Result<(), LeaseError> {
        let wire = self
            .transport
            .lease_release(&self.scope, agent_id, device_id, epoch)
            .map_err(|e| LeaseError::Backend(e.to_string()))?;
        match wire {
            LeaseWire::Released => Ok(()),
            LeaseWire::Lost {
                claimed_epoch,
                current_epoch,
            } => Err(LeaseError::Lost {
                agent_id: agent_id.to_string(),
                claimed_epoch,
                current_epoch,
            }),
            other => Err(LeaseError::Backend(format!(
                "release: unexpected lease verdict {other:?}"
            ))),
        }
    }

    fn current(&mut self, agent_id: &str) -> Result<Option<Lease>, LeaseError> {
        let wire = self
            .transport
            .lease_current(&self.scope, agent_id)
            .map_err(|e| LeaseError::Backend(e.to_string()))?;
        match wire {
            LeaseWire::Current(l) => Ok(l),
            other => Err(LeaseError::Backend(format!(
                "current: unexpected lease verdict {other:?}"
            ))),
        }
    }
}

// ---------------------------------------------------------------------------
// LoopbackTransport — the in-process reference server.
// ---------------------------------------------------------------------------

struct ScopeBackend {
    relay: crate::relay::FsRelay,
    lease: InMemoryLeaseCoordinator,
}

struct LoopbackInner {
    dir: tempfile::TempDir,
    wall: WallClock,
    config: RelayConfig,
    scopes: BTreeMap<String, ScopeBackend>,
}

/// An in-process [`SyncTransport`]: each scope is a real [`FsRelay`] +
/// [`InMemoryLeaseCoordinator`] over a temp dir, so it reproduces the canonical
/// relay/lease semantics exactly. `Clone` shares one server across devices.
#[derive(Clone)]
pub struct LoopbackTransport {
    inner: Arc<Mutex<LoopbackInner>>,
}

impl LoopbackTransport {
    /// A fresh reference server (system wall clock, no eviction horizon).
    pub fn new() -> std::io::Result<Self> {
        Self::with_config(RelayConfig::default(), crate::oplog::system_clock())
    }

    pub fn with_config(config: RelayConfig, wall: WallClock) -> std::io::Result<Self> {
        Ok(Self {
            inner: Arc::new(Mutex::new(LoopbackInner {
                dir: tempfile::TempDir::new()?,
                wall,
                config,
                scopes: BTreeMap::new(),
            })),
        })
    }

    /// Run `f` against the (lazily-created) backend for `scope`.
    fn with_scope<R>(
        &self,
        scope: &str,
        f: impl FnOnce(&mut ScopeBackend) -> Result<R, TransportError>,
    ) -> Result<R, TransportError> {
        let mut inner = self
            .inner
            .lock()
            .map_err(|_| TransportError::Unavailable("loopback lock poisoned".into()))?;
        if !inner.scopes.contains_key(scope) {
            // FsRelay dirs are per-scope so one server keeps tenants isolated.
            let dir = inner.dir.path().join(sanitize_scope(scope));
            std::fs::create_dir_all(&dir)
                .map_err(|e| TransportError::Unavailable(format!("loopback mkdir: {e}")))?;
            let relay = crate::relay::FsRelay::open(&dir, inner.config.clone(), inner.wall.clone())
                .map_err(|e| TransportError::Unavailable(format!("loopback FsRelay: {e}")))?;
            let lease = InMemoryLeaseCoordinator::new(inner.wall.clone());
            inner
                .scopes
                .insert(scope.to_string(), ScopeBackend { relay, lease });
        }
        let backend = inner.scopes.get_mut(scope).expect("just inserted");
        f(backend)
    }
}

/// Map a scope string to a filesystem-safe dir name (loopback only).
fn sanitize_scope(scope: &str) -> String {
    scope
        .chars()
        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
        .collect()
}

fn relay_err(e: RelayError) -> TransportError {
    TransportError::Unavailable(e.to_string())
}

impl SyncTransport for LoopbackTransport {
    fn enroll(&self, scope: &str, device_id: &str) -> Result<RosterEntry, TransportError> {
        self.with_scope(scope, |b| b.relay.register(device_id).map_err(relay_err))
    }
    fn push(
        &self,
        scope: &str,
        device_id: &str,
        ops: &[OpRecord],
    ) -> Result<PushOutcome, TransportError> {
        self.with_scope(scope, |b| b.relay.push(device_id, ops).map_err(relay_err))
    }
    fn pull(
        &self,
        scope: &str,
        device_id: &str,
        since: &Frontier,
    ) -> Result<PullResult, TransportError> {
        self.with_scope(scope, |b| b.relay.pull(device_id, since).map_err(relay_err))
    }
    fn ack(
        &self,
        scope: &str,
        device_id: &str,
        frontier: Hlc,
    ) -> Result<AckOutcome, TransportError> {
        self.with_scope(scope, |b| {
            b.relay.ack(device_id, frontier).map_err(relay_err)
        })
    }
    fn checkpoint_put(
        &self,
        scope: &str,
        device_id: &str,
        checkpoint: &Checkpoint,
    ) -> Result<bool, TransportError> {
        self.with_scope(scope, |b| {
            b.relay
                .checkpoint_put(device_id, checkpoint)
                .map_err(relay_err)
        })
    }
    fn checkpoint_get(&self, scope: &str) -> Result<Option<Checkpoint>, TransportError> {
        self.with_scope(scope, |b| b.relay.checkpoint_get().map_err(relay_err))
    }
    fn roster(&self, scope: &str) -> Result<Vec<RosterEntry>, TransportError> {
        self.with_scope(scope, |b| b.relay.roster().map_err(relay_err))
    }
    fn stable_frontier(&self, scope: &str) -> Result<Option<Hlc>, TransportError> {
        self.with_scope(scope, |b| b.relay.stable_frontier().map_err(relay_err))
    }
    fn gc(&self, scope: &str) -> Result<GcReport, TransportError> {
        self.with_scope(scope, |b| b.relay.gc().map_err(relay_err))
    }

    fn lease_acquire(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        ttl_ms: u64,
    ) -> Result<LeaseWire, TransportError> {
        self.with_scope(scope, |b| {
            Ok(match b.lease.acquire(agent_id, device_id, ttl_ms) {
                Ok(l) => LeaseWire::Granted(l),
                Err(LeaseError::Held {
                    holder,
                    epoch,
                    expires_at_ms,
                    ..
                }) => LeaseWire::Held {
                    holder,
                    epoch,
                    expires_at_ms,
                },
                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
            })
        })
    }
    fn lease_renew(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
        ttl_ms: u64,
    ) -> Result<LeaseWire, TransportError> {
        self.with_scope(scope, |b| {
            Ok(match b.lease.renew(agent_id, device_id, epoch, ttl_ms) {
                Ok(l) => LeaseWire::Granted(l),
                Err(LeaseError::Lost {
                    claimed_epoch,
                    current_epoch,
                    ..
                }) => LeaseWire::Lost {
                    claimed_epoch,
                    current_epoch,
                },
                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
            })
        })
    }
    fn lease_release(
        &self,
        scope: &str,
        agent_id: &str,
        device_id: &str,
        epoch: u64,
    ) -> Result<LeaseWire, TransportError> {
        self.with_scope(scope, |b| {
            Ok(match b.lease.release(agent_id, device_id, epoch) {
                Ok(()) => LeaseWire::Released,
                Err(LeaseError::Lost {
                    claimed_epoch,
                    current_epoch,
                    ..
                }) => LeaseWire::Lost {
                    claimed_epoch,
                    current_epoch,
                },
                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
            })
        })
    }
    fn lease_current(&self, scope: &str, agent_id: &str) -> Result<LeaseWire, TransportError> {
        self.with_scope(scope, |b| {
            b.lease
                .current(agent_id)
                .map(LeaseWire::Current)
                .map_err(|e| TransportError::Unavailable(e.to_string()))
        })
    }
}

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

    fn transport() -> Arc<dyn SyncTransport> {
        let clock: WallClock = Arc::new(|| 1000);
        Arc::new(LoopbackTransport::with_config(RelayConfig::default(), clock).unwrap())
    }

    #[test]
    fn two_devices_converge_through_a_shared_service() {
        // Mac and phone push through ONE remote service (shared Arc). Each pulls
        // the other's ops — the "phone + Mac" convergence, but over the network
        // relay rather than a shared folder.
        let svc = transport();
        let scope = "user:matt";
        let mut mac = NetworkRelay::new(svc.clone(), scope);
        let mut phone = NetworkRelay::new(svc.clone(), scope);

        mac.register("mac").unwrap();
        phone.register("phone").unwrap();

        // Build a real, content-addressed op (the relay verifies op_id == the
        // content hash, so it must go through OpRecord::new).
        let op = crate::oplog::OpRecord::new(
            Hlc {
                wall_ms: 1000,
                counter: 0,
                device_id: "mac".into(),
            },
            0,
            None,
            crate::oplog::Scope::Personal,
            crate::oplog::Surface::Routing,
            serde_json::json!({"hello": "phone"}),
        );
        let out = mac.push("mac", std::slice::from_ref(&op)).unwrap();
        assert_eq!(out.accepted, 1);

        // The phone pulls from empty and sees the Mac's op.
        let pulled = phone.pull("phone", &Frontier::new()).unwrap();
        assert_eq!(pulled.ops.len(), 1);
        assert_eq!(pulled.ops[0].op_id, op.op_id);

        // Tenant isolation: a different scope is a different, empty relay.
        let mut other = NetworkRelay::new(svc.clone(), "user:someone_else");
        assert!(other.pull("d", &Frontier::new()).unwrap().ops.is_empty());
    }

    #[test]
    fn frontier_truncation_maps_to_the_relay_cold_bootstrap_signal() {
        // The service's distinct truncation signal must become
        // RelayError::FrontierTruncated so the pump rebases (cold bootstrap),
        // not a generic IO error that would just retry forever.
        let te = TransportError::FrontierTruncated {
            device_id: "mac".into(),
            dropped_below: 7,
        };
        match RelayError::from(te) {
            RelayError::FrontierTruncated {
                device_id,
                dropped_below,
            } => {
                assert_eq!(device_id, "mac");
                assert_eq!(dropped_below, 7);
            }
            other => panic!("expected FrontierTruncated, got {other:?}"),
        }
    }

    #[test]
    fn distributed_lease_is_mutually_exclusive_across_devices() {
        // Mac and phone contend for the same agent's execution lease through the
        // one service. Only one holds it; the loser sees `Held`; on release the
        // other acquires with the next (monotone) epoch — the fencing token that
        // stops both devices double-running the agent.
        let svc = transport();
        let scope = "user:matt";
        let mut mac = NetworkLeaseCoordinator::new(svc.clone(), scope);
        let mut phone = NetworkLeaseCoordinator::new(svc.clone(), scope);

        let l1 = mac.acquire("milo", "mac", 10_000).unwrap();
        assert_eq!(l1.epoch, 1);
        assert_eq!(l1.holder, "mac");

        // Phone loses the CAS while the Mac's lease is valid.
        match phone.acquire("milo", "phone", 10_000) {
            Err(LeaseError::Held { holder, epoch, .. }) => {
                assert_eq!(holder, "mac");
                assert_eq!(epoch, 1);
            }
            other => panic!("expected Held, got {other:?}"),
        }

        // Both devices read the same holder.
        assert_eq!(phone.current("milo").unwrap().unwrap().holder, "mac");

        // Mac releases; phone now acquires with epoch 2 (never reused).
        mac.release("milo", "mac", 1).unwrap();
        let l2 = phone.acquire("milo", "phone", 10_000).unwrap();
        assert_eq!(l2.epoch, 2);
        assert_eq!(l2.holder, "phone");

        // The Mac (a zombie holding the stale epoch) learns it lost.
        match mac.renew("milo", "mac", 1, 10_000) {
            Err(LeaseError::Lost { current_epoch, .. }) => assert_eq!(current_epoch, 2),
            other => panic!("expected Lost, got {other:?}"),
        }
    }
}