car_sync/
net_relay.rs

1//! Network sync backend — CAR's client for a **remote** relay + lease register
2//! (the Parslee sync service), behind the same [`Relay`] / [`LeaseCoordinator`]
3//! traits the local [`FsRelay`] / [`InMemoryLeaseCoordinator`] already satisfy.
4//!
5//! # Why
6//!
7//! `FsRelay` converges two devices only if they share a filesystem directory —
8//! fine for one machine, useless for "my phone and my Mac". This module makes
9//! the relay a *service*: [`NetworkRelay`] and [`NetworkLeaseCoordinator`] speak
10//! to it over a pluggable [`SyncTransport`], scoped to the caller's Parslee
11//! identity (`scope`, e.g. `user:<id>` or `org:<id>`). The Parslee backend holds
12//! only what `FsRelay` holds — op metadata (`op_id`/`hlc`/`seq`) in the clear and
13//! op *payloads* as E2E ciphertext (see [`crate::crypto`]) — so it can route and
14//! GC without reading your config.
15//!
16//! # The transport seam
17//!
18//! [`SyncTransport`] is the wire contract (blocking, because [`Relay`] is sync;
19//! an HTTP impl blocks inside these calls, which the daemon already runs off a
20//! task). Two impls:
21//!
22//! - [`LoopbackTransport`] — an **in-process reference server**. Each scope is
23//!   backed by a real [`FsRelay`] + [`InMemoryLeaseCoordinator`] over a temp
24//!   dir, so `NetworkRelay`-over-loopback is *semantically identical* to
25//!   `FsRelay` by construction (no re-implementation of the GC-floor / dedup /
26//!   monotone-ack / dominance logic). It is the test double AND a usable
27//!   single-host multi-process relay. `Clone` shares one server across N
28//!   devices.
29//! - `HttpTransport` (a later slice) — reqwest against the real Parslee endpoint.
30//!   It must implement the exact same contract; the loopback is the executable
31//!   spec.
32
33use std::collections::BTreeMap;
34use std::sync::{Arc, Mutex};
35
36use serde::{Deserialize, Serialize};
37
38use crate::checkpoint::Checkpoint;
39use crate::lease::{InMemoryLeaseCoordinator, Lease, LeaseCoordinator, LeaseError};
40use crate::oplog::{Hlc, OpRecord, WallClock};
41use crate::relay::{
42    AckOutcome, Frontier, GcReport, PullResult, PushOutcome, Relay, RelayConfig, RelayError,
43    RosterEntry,
44};
45
46/// A transport-level failure — the *service* was unreachable or misbehaved.
47/// Distinct from an in-band verdict (a deduped push, a `Held` lease): those are
48/// success returns, not errors.
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub enum TransportError {
51    /// The service could not be reached / timed out / returned a transport
52    /// error (HTTP 5xx, connection reset, …).
53    Unavailable(String),
54    /// The caller is not authorized for `scope` (bad/expired Parslee token).
55    Unauthorized(String),
56    /// A reply could not be parsed into the expected shape (protocol drift).
57    Protocol(String),
58    /// A pull requested a frontier below the relay's GC floor for `device_id`
59    /// (retained ops start at `dropped_below`). NOT a failure — the service's
60    /// signal that the caller must **cold-bootstrap** from the latest
61    /// checkpoint. Maps to [`RelayError::FrontierTruncated`], which
62    /// [`crate::session::SyncSession::pump`] already handles by rebasing.
63    FrontierTruncated {
64        device_id: String,
65        dropped_below: u64,
66    },
67}
68
69impl std::fmt::Display for TransportError {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        match self {
72            TransportError::Unavailable(m) => write!(f, "sync transport unavailable: {m}"),
73            TransportError::Unauthorized(m) => write!(f, "sync transport unauthorized: {m}"),
74            TransportError::Protocol(m) => write!(f, "sync transport protocol error: {m}"),
75            TransportError::FrontierTruncated {
76                device_id,
77                dropped_below,
78            } => write!(
79                f,
80                "sync transport frontier truncated: device {device_id} ops below seq \
81                 {dropped_below} were GC'd (cold-bootstrap from checkpoint)"
82            ),
83        }
84    }
85}
86impl std::error::Error for TransportError {}
87
88impl From<TransportError> for RelayError {
89    fn from(e: TransportError) -> Self {
90        match e {
91            // The service's cold-bootstrap signal maps to the relay's own
92            // FrontierTruncated, which the pump handles by rebasing.
93            TransportError::FrontierTruncated {
94                device_id,
95                dropped_below,
96            } => RelayError::FrontierTruncated {
97                device_id,
98                dropped_below,
99            },
100            // Everything else has no relay variant; surface it as an IO-ish
101            // failure so the pump's best-effort loop treats it as "try later".
102            other => RelayError::Io(std::io::Error::other(other.to_string())),
103        }
104    }
105}
106
107/// The serialized verdict of a lease-register call. Mirrors the coordinator's
108/// `Result<_, LeaseError>` in a wire-portable shape; the client reconstructs the
109/// real [`LeaseError`] from it (the `agent_id` is known client-side).
110#[derive(Debug, Clone, Serialize, Deserialize)]
111pub enum LeaseWire {
112    /// `acquire`/`renew` granted this lease.
113    Granted(Lease),
114    /// `acquire` CAS-failed: a still-valid lease is held.
115    Held {
116        holder: String,
117        epoch: u64,
118        expires_at_ms: u64,
119    },
120    /// `renew`/`release` found the caller is no longer the holder at its epoch.
121    Lost {
122        claimed_epoch: u64,
123        current_epoch: u64,
124    },
125    /// `release` succeeded.
126    Released,
127    /// `current` read (holder or unheld).
128    Current(Option<Lease>),
129}
130
131/// The wire contract for a remote sync service, scoped per Parslee identity.
132///
133/// Every method takes a `scope` (`user:<id>` / `org:<id>`) so one service backs
134/// many tenants. Blocking by design. Impls MUST be faithful to [`FsRelay`] /
135/// [`InMemoryLeaseCoordinator`] semantics — [`LoopbackTransport`] is the spec.
136pub trait SyncTransport: Send + Sync {
137    fn enroll(&self, scope: &str, device_id: &str) -> Result<RosterEntry, TransportError>;
138    fn push(
139        &self,
140        scope: &str,
141        device_id: &str,
142        ops: &[OpRecord],
143    ) -> Result<PushOutcome, TransportError>;
144    fn pull(
145        &self,
146        scope: &str,
147        device_id: &str,
148        since: &Frontier,
149    ) -> Result<PullResult, TransportError>;
150    fn ack(
151        &self,
152        scope: &str,
153        device_id: &str,
154        frontier: Hlc,
155    ) -> Result<AckOutcome, TransportError>;
156    fn checkpoint_put(
157        &self,
158        scope: &str,
159        device_id: &str,
160        checkpoint: &Checkpoint,
161    ) -> Result<bool, TransportError>;
162    fn checkpoint_get(&self, scope: &str) -> Result<Option<Checkpoint>, TransportError>;
163    fn roster(&self, scope: &str) -> Result<Vec<RosterEntry>, TransportError>;
164    fn stable_frontier(&self, scope: &str) -> Result<Option<Hlc>, TransportError>;
165    fn gc(&self, scope: &str) -> Result<GcReport, TransportError>;
166
167    // --- execution lease register (B5, distributed) ---
168    fn lease_acquire(
169        &self,
170        scope: &str,
171        agent_id: &str,
172        device_id: &str,
173        ttl_ms: u64,
174    ) -> Result<LeaseWire, TransportError>;
175    fn lease_renew(
176        &self,
177        scope: &str,
178        agent_id: &str,
179        device_id: &str,
180        epoch: u64,
181        ttl_ms: u64,
182    ) -> Result<LeaseWire, TransportError>;
183    fn lease_release(
184        &self,
185        scope: &str,
186        agent_id: &str,
187        device_id: &str,
188        epoch: u64,
189    ) -> Result<LeaseWire, TransportError>;
190    fn lease_current(&self, scope: &str, agent_id: &str) -> Result<LeaseWire, TransportError>;
191}
192
193/// A [`Relay`] backed by a remote [`SyncTransport`], scoped to one Parslee
194/// identity. One per device; share the transport (`Arc`) across devices.
195pub struct NetworkRelay {
196    transport: Arc<dyn SyncTransport>,
197    scope: String,
198}
199
200impl NetworkRelay {
201    pub fn new(transport: Arc<dyn SyncTransport>, scope: impl Into<String>) -> Self {
202        Self {
203            transport,
204            scope: scope.into(),
205        }
206    }
207}
208
209impl Relay for NetworkRelay {
210    fn register(&mut self, device_id: &str) -> Result<RosterEntry, RelayError> {
211        Ok(self.transport.enroll(&self.scope, device_id)?)
212    }
213    fn push(&mut self, device_id: &str, ops: &[OpRecord]) -> Result<PushOutcome, RelayError> {
214        Ok(self.transport.push(&self.scope, device_id, ops)?)
215    }
216    fn pull(&mut self, device_id: &str, since: &Frontier) -> Result<PullResult, RelayError> {
217        Ok(self.transport.pull(&self.scope, device_id, since)?)
218    }
219    fn ack(&mut self, device_id: &str, frontier: Hlc) -> Result<AckOutcome, RelayError> {
220        Ok(self.transport.ack(&self.scope, device_id, frontier)?)
221    }
222    fn checkpoint_put(
223        &mut self,
224        device_id: &str,
225        checkpoint: &Checkpoint,
226    ) -> Result<bool, RelayError> {
227        Ok(self
228            .transport
229            .checkpoint_put(&self.scope, device_id, checkpoint)?)
230    }
231    fn checkpoint_get(&mut self) -> Result<Option<Checkpoint>, RelayError> {
232        Ok(self.transport.checkpoint_get(&self.scope)?)
233    }
234    fn roster(&mut self) -> Result<Vec<RosterEntry>, RelayError> {
235        Ok(self.transport.roster(&self.scope)?)
236    }
237    fn stable_frontier(&mut self) -> Result<Option<Hlc>, RelayError> {
238        Ok(self.transport.stable_frontier(&self.scope)?)
239    }
240    fn gc(&mut self) -> Result<GcReport, RelayError> {
241        Ok(self.transport.gc(&self.scope)?)
242    }
243}
244
245/// A [`LeaseCoordinator`] backed by a remote [`SyncTransport`] — the
246/// **distributed** execution lease that makes "phone and Mac" mutually
247/// exclusive on a given agent. Reconstructs the real [`LeaseError`] from the
248/// transport's [`LeaseWire`] verdict.
249pub struct NetworkLeaseCoordinator {
250    transport: Arc<dyn SyncTransport>,
251    scope: String,
252}
253
254impl NetworkLeaseCoordinator {
255    pub fn new(transport: Arc<dyn SyncTransport>, scope: impl Into<String>) -> Self {
256        Self {
257            transport,
258            scope: scope.into(),
259        }
260    }
261}
262
263fn wire_to_lease(agent_id: &str, wire: LeaseWire, ctx: &'static str) -> Result<Lease, LeaseError> {
264    match wire {
265        LeaseWire::Granted(l) => Ok(l),
266        LeaseWire::Held {
267            holder,
268            epoch,
269            expires_at_ms,
270        } => Err(LeaseError::Held {
271            agent_id: agent_id.to_string(),
272            holder,
273            epoch,
274            expires_at_ms,
275        }),
276        LeaseWire::Lost {
277            claimed_epoch,
278            current_epoch,
279        } => Err(LeaseError::Lost {
280            agent_id: agent_id.to_string(),
281            claimed_epoch,
282            current_epoch,
283        }),
284        other => Err(LeaseError::Backend(format!(
285            "{ctx}: unexpected lease verdict {other:?}"
286        ))),
287    }
288}
289
290impl LeaseCoordinator for NetworkLeaseCoordinator {
291    fn acquire(
292        &mut self,
293        agent_id: &str,
294        device_id: &str,
295        ttl_ms: u64,
296    ) -> Result<Lease, LeaseError> {
297        let wire = self
298            .transport
299            .lease_acquire(&self.scope, agent_id, device_id, ttl_ms)
300            .map_err(|e| LeaseError::Backend(e.to_string()))?;
301        wire_to_lease(agent_id, wire, "acquire")
302    }
303
304    fn renew(
305        &mut self,
306        agent_id: &str,
307        device_id: &str,
308        epoch: u64,
309        ttl_ms: u64,
310    ) -> Result<Lease, LeaseError> {
311        let wire = self
312            .transport
313            .lease_renew(&self.scope, agent_id, device_id, epoch, ttl_ms)
314            .map_err(|e| LeaseError::Backend(e.to_string()))?;
315        wire_to_lease(agent_id, wire, "renew")
316    }
317
318    fn release(&mut self, agent_id: &str, device_id: &str, epoch: u64) -> Result<(), LeaseError> {
319        let wire = self
320            .transport
321            .lease_release(&self.scope, agent_id, device_id, epoch)
322            .map_err(|e| LeaseError::Backend(e.to_string()))?;
323        match wire {
324            LeaseWire::Released => Ok(()),
325            LeaseWire::Lost {
326                claimed_epoch,
327                current_epoch,
328            } => Err(LeaseError::Lost {
329                agent_id: agent_id.to_string(),
330                claimed_epoch,
331                current_epoch,
332            }),
333            other => Err(LeaseError::Backend(format!(
334                "release: unexpected lease verdict {other:?}"
335            ))),
336        }
337    }
338
339    fn current(&mut self, agent_id: &str) -> Result<Option<Lease>, LeaseError> {
340        let wire = self
341            .transport
342            .lease_current(&self.scope, agent_id)
343            .map_err(|e| LeaseError::Backend(e.to_string()))?;
344        match wire {
345            LeaseWire::Current(l) => Ok(l),
346            other => Err(LeaseError::Backend(format!(
347                "current: unexpected lease verdict {other:?}"
348            ))),
349        }
350    }
351}
352
353// ---------------------------------------------------------------------------
354// LoopbackTransport — the in-process reference server.
355// ---------------------------------------------------------------------------
356
357struct ScopeBackend {
358    relay: crate::relay::FsRelay,
359    lease: InMemoryLeaseCoordinator,
360}
361
362struct LoopbackInner {
363    dir: tempfile::TempDir,
364    wall: WallClock,
365    config: RelayConfig,
366    scopes: BTreeMap<String, ScopeBackend>,
367}
368
369/// An in-process [`SyncTransport`]: each scope is a real [`FsRelay`] +
370/// [`InMemoryLeaseCoordinator`] over a temp dir, so it reproduces the canonical
371/// relay/lease semantics exactly. `Clone` shares one server across devices.
372#[derive(Clone)]
373pub struct LoopbackTransport {
374    inner: Arc<Mutex<LoopbackInner>>,
375}
376
377impl LoopbackTransport {
378    /// A fresh reference server (system wall clock, no eviction horizon).
379    pub fn new() -> std::io::Result<Self> {
380        Self::with_config(RelayConfig::default(), crate::oplog::system_clock())
381    }
382
383    pub fn with_config(config: RelayConfig, wall: WallClock) -> std::io::Result<Self> {
384        Ok(Self {
385            inner: Arc::new(Mutex::new(LoopbackInner {
386                dir: tempfile::TempDir::new()?,
387                wall,
388                config,
389                scopes: BTreeMap::new(),
390            })),
391        })
392    }
393
394    /// Run `f` against the (lazily-created) backend for `scope`.
395    fn with_scope<R>(
396        &self,
397        scope: &str,
398        f: impl FnOnce(&mut ScopeBackend) -> Result<R, TransportError>,
399    ) -> Result<R, TransportError> {
400        let mut inner = self
401            .inner
402            .lock()
403            .map_err(|_| TransportError::Unavailable("loopback lock poisoned".into()))?;
404        if !inner.scopes.contains_key(scope) {
405            // FsRelay dirs are per-scope so one server keeps tenants isolated.
406            let dir = inner.dir.path().join(sanitize_scope(scope));
407            std::fs::create_dir_all(&dir)
408                .map_err(|e| TransportError::Unavailable(format!("loopback mkdir: {e}")))?;
409            let relay = crate::relay::FsRelay::open(&dir, inner.config.clone(), inner.wall.clone())
410                .map_err(|e| TransportError::Unavailable(format!("loopback FsRelay: {e}")))?;
411            let lease = InMemoryLeaseCoordinator::new(inner.wall.clone());
412            inner
413                .scopes
414                .insert(scope.to_string(), ScopeBackend { relay, lease });
415        }
416        let backend = inner.scopes.get_mut(scope).expect("just inserted");
417        f(backend)
418    }
419}
420
421/// Map a scope string to a filesystem-safe dir name (loopback only).
422fn sanitize_scope(scope: &str) -> String {
423    scope
424        .chars()
425        .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' })
426        .collect()
427}
428
429fn relay_err(e: RelayError) -> TransportError {
430    TransportError::Unavailable(e.to_string())
431}
432
433impl SyncTransport for LoopbackTransport {
434    fn enroll(&self, scope: &str, device_id: &str) -> Result<RosterEntry, TransportError> {
435        self.with_scope(scope, |b| b.relay.register(device_id).map_err(relay_err))
436    }
437    fn push(
438        &self,
439        scope: &str,
440        device_id: &str,
441        ops: &[OpRecord],
442    ) -> Result<PushOutcome, TransportError> {
443        self.with_scope(scope, |b| b.relay.push(device_id, ops).map_err(relay_err))
444    }
445    fn pull(
446        &self,
447        scope: &str,
448        device_id: &str,
449        since: &Frontier,
450    ) -> Result<PullResult, TransportError> {
451        self.with_scope(scope, |b| b.relay.pull(device_id, since).map_err(relay_err))
452    }
453    fn ack(
454        &self,
455        scope: &str,
456        device_id: &str,
457        frontier: Hlc,
458    ) -> Result<AckOutcome, TransportError> {
459        self.with_scope(scope, |b| {
460            b.relay.ack(device_id, frontier).map_err(relay_err)
461        })
462    }
463    fn checkpoint_put(
464        &self,
465        scope: &str,
466        device_id: &str,
467        checkpoint: &Checkpoint,
468    ) -> Result<bool, TransportError> {
469        self.with_scope(scope, |b| {
470            b.relay
471                .checkpoint_put(device_id, checkpoint)
472                .map_err(relay_err)
473        })
474    }
475    fn checkpoint_get(&self, scope: &str) -> Result<Option<Checkpoint>, TransportError> {
476        self.with_scope(scope, |b| b.relay.checkpoint_get().map_err(relay_err))
477    }
478    fn roster(&self, scope: &str) -> Result<Vec<RosterEntry>, TransportError> {
479        self.with_scope(scope, |b| b.relay.roster().map_err(relay_err))
480    }
481    fn stable_frontier(&self, scope: &str) -> Result<Option<Hlc>, TransportError> {
482        self.with_scope(scope, |b| b.relay.stable_frontier().map_err(relay_err))
483    }
484    fn gc(&self, scope: &str) -> Result<GcReport, TransportError> {
485        self.with_scope(scope, |b| b.relay.gc().map_err(relay_err))
486    }
487
488    fn lease_acquire(
489        &self,
490        scope: &str,
491        agent_id: &str,
492        device_id: &str,
493        ttl_ms: u64,
494    ) -> Result<LeaseWire, TransportError> {
495        self.with_scope(scope, |b| {
496            Ok(match b.lease.acquire(agent_id, device_id, ttl_ms) {
497                Ok(l) => LeaseWire::Granted(l),
498                Err(LeaseError::Held {
499                    holder,
500                    epoch,
501                    expires_at_ms,
502                    ..
503                }) => LeaseWire::Held {
504                    holder,
505                    epoch,
506                    expires_at_ms,
507                },
508                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
509            })
510        })
511    }
512    fn lease_renew(
513        &self,
514        scope: &str,
515        agent_id: &str,
516        device_id: &str,
517        epoch: u64,
518        ttl_ms: u64,
519    ) -> Result<LeaseWire, TransportError> {
520        self.with_scope(scope, |b| {
521            Ok(match b.lease.renew(agent_id, device_id, epoch, ttl_ms) {
522                Ok(l) => LeaseWire::Granted(l),
523                Err(LeaseError::Lost {
524                    claimed_epoch,
525                    current_epoch,
526                    ..
527                }) => LeaseWire::Lost {
528                    claimed_epoch,
529                    current_epoch,
530                },
531                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
532            })
533        })
534    }
535    fn lease_release(
536        &self,
537        scope: &str,
538        agent_id: &str,
539        device_id: &str,
540        epoch: u64,
541    ) -> Result<LeaseWire, TransportError> {
542        self.with_scope(scope, |b| {
543            Ok(match b.lease.release(agent_id, device_id, epoch) {
544                Ok(()) => LeaseWire::Released,
545                Err(LeaseError::Lost {
546                    claimed_epoch,
547                    current_epoch,
548                    ..
549                }) => LeaseWire::Lost {
550                    claimed_epoch,
551                    current_epoch,
552                },
553                Err(e) => return Err(TransportError::Unavailable(e.to_string())),
554            })
555        })
556    }
557    fn lease_current(&self, scope: &str, agent_id: &str) -> Result<LeaseWire, TransportError> {
558        self.with_scope(scope, |b| {
559            b.lease
560                .current(agent_id)
561                .map(LeaseWire::Current)
562                .map_err(|e| TransportError::Unavailable(e.to_string()))
563        })
564    }
565}
566
567#[cfg(test)]
568mod tests {
569    use super::*;
570    use crate::relay::Relay;
571
572    fn transport() -> Arc<dyn SyncTransport> {
573        let clock: WallClock = Arc::new(|| 1000);
574        Arc::new(LoopbackTransport::with_config(RelayConfig::default(), clock).unwrap())
575    }
576
577    #[test]
578    fn two_devices_converge_through_a_shared_service() {
579        // Mac and phone push through ONE remote service (shared Arc). Each pulls
580        // the other's ops — the "phone + Mac" convergence, but over the network
581        // relay rather than a shared folder.
582        let svc = transport();
583        let scope = "user:matt";
584        let mut mac = NetworkRelay::new(svc.clone(), scope);
585        let mut phone = NetworkRelay::new(svc.clone(), scope);
586
587        mac.register("mac").unwrap();
588        phone.register("phone").unwrap();
589
590        // Build a real, content-addressed op (the relay verifies op_id == the
591        // content hash, so it must go through OpRecord::new).
592        let op = crate::oplog::OpRecord::new(
593            Hlc {
594                wall_ms: 1000,
595                counter: 0,
596                device_id: "mac".into(),
597            },
598            0,
599            None,
600            crate::oplog::Scope::Personal,
601            crate::oplog::Surface::Routing,
602            serde_json::json!({"hello": "phone"}),
603        );
604        let out = mac.push("mac", std::slice::from_ref(&op)).unwrap();
605        assert_eq!(out.accepted, 1);
606
607        // The phone pulls from empty and sees the Mac's op.
608        let pulled = phone.pull("phone", &Frontier::new()).unwrap();
609        assert_eq!(pulled.ops.len(), 1);
610        assert_eq!(pulled.ops[0].op_id, op.op_id);
611
612        // Tenant isolation: a different scope is a different, empty relay.
613        let mut other = NetworkRelay::new(svc.clone(), "user:someone_else");
614        assert!(other.pull("d", &Frontier::new()).unwrap().ops.is_empty());
615    }
616
617    #[test]
618    fn frontier_truncation_maps_to_the_relay_cold_bootstrap_signal() {
619        // The service's distinct truncation signal must become
620        // RelayError::FrontierTruncated so the pump rebases (cold bootstrap),
621        // not a generic IO error that would just retry forever.
622        let te = TransportError::FrontierTruncated {
623            device_id: "mac".into(),
624            dropped_below: 7,
625        };
626        match RelayError::from(te) {
627            RelayError::FrontierTruncated {
628                device_id,
629                dropped_below,
630            } => {
631                assert_eq!(device_id, "mac");
632                assert_eq!(dropped_below, 7);
633            }
634            other => panic!("expected FrontierTruncated, got {other:?}"),
635        }
636    }
637
638    #[test]
639    fn distributed_lease_is_mutually_exclusive_across_devices() {
640        // Mac and phone contend for the same agent's execution lease through the
641        // one service. Only one holds it; the loser sees `Held`; on release the
642        // other acquires with the next (monotone) epoch — the fencing token that
643        // stops both devices double-running the agent.
644        let svc = transport();
645        let scope = "user:matt";
646        let mut mac = NetworkLeaseCoordinator::new(svc.clone(), scope);
647        let mut phone = NetworkLeaseCoordinator::new(svc.clone(), scope);
648
649        let l1 = mac.acquire("milo", "mac", 10_000).unwrap();
650        assert_eq!(l1.epoch, 1);
651        assert_eq!(l1.holder, "mac");
652
653        // Phone loses the CAS while the Mac's lease is valid.
654        match phone.acquire("milo", "phone", 10_000) {
655            Err(LeaseError::Held { holder, epoch, .. }) => {
656                assert_eq!(holder, "mac");
657                assert_eq!(epoch, 1);
658            }
659            other => panic!("expected Held, got {other:?}"),
660        }
661
662        // Both devices read the same holder.
663        assert_eq!(phone.current("milo").unwrap().unwrap().holder, "mac");
664
665        // Mac releases; phone now acquires with epoch 2 (never reused).
666        mac.release("milo", "mac", 1).unwrap();
667        let l2 = phone.acquire("milo", "phone", 10_000).unwrap();
668        assert_eq!(l2.epoch, 2);
669        assert_eq!(l2.holder, "phone");
670
671        // The Mac (a zombie holding the stale epoch) learns it lost.
672        match mac.renew("milo", "mac", 1, 10_000) {
673            Err(LeaseError::Lost { current_epoch, .. }) => assert_eq!(current_epoch, 2),
674            other => panic!("expected Lost, got {other:?}"),
675        }
676    }
677}
car_sync/net_relay.rs

car_sync/
net_relay.rs
