prospero_core/
ownership.rs

1//! Which process is the single writer for a given stream.
2//!
3//! Standalone uses [`SelfOwnsAll`]: one process owns every stream, so the lease
4//! is a no-op. The clustered `LeasedOwnership` (a Postgres lease row + reaper)
5//! drops in behind the same trait in a later phase — see the topology design
6//! spec §3.3. The `epoch` on [`Lease`] exists now so control-fencing can be
7//! added later without a wire change.
8
9use async_trait::async_trait;
10
11use crate::error::Result;
12
13/// A claim on a stream's single-writer role. `epoch` is a monotonic fencing
14/// token (always 0 under [`SelfOwnsAll`]).
15#[derive(Debug, Clone)]
16pub struct Lease {
17    /// The owned stream key.
18    pub stream_key: String,
19    /// Monotonic fencing epoch for the claim.
20    pub epoch: u64,
21}
22
23/// Single-writer ownership of streams.
24#[async_trait]
25pub trait Ownership: Send + Sync {
26    /// Claim `stream_key` if it is free, expired, or already held by THIS
27    /// process (idempotent — re-acquiring your own live lease returns it and
28    /// does not change its epoch). Returns `None` if another live replica owns
29    /// it.
30    async fn try_acquire(&self, stream_key: &str) -> Option<Lease>;
31
32    /// Extend a held lease. `Err` if the lease was lost (stolen/expired) — which
33    /// is how a replica learns it is no longer the owner.
34    async fn renew(&self, lease: &Lease) -> Result<()>;
35
36    /// Release a held stream so a peer may claim it immediately (graceful
37    /// hand-off), rather than waiting for TTL expiry.
38    async fn release(&self, stream_key: &str);
39
40    /// Whether this process currently owns `stream_key`. Cheap/in-memory: it is
41    /// consulted on the poll loop's hot path.
42    fn owns(&self, stream_key: &str) -> bool;
43}
44
45/// Standalone ownership: this process owns every stream unconditionally.
46pub struct SelfOwnsAll;
47
48#[async_trait]
49impl Ownership for SelfOwnsAll {
50    async fn try_acquire(&self, stream_key: &str) -> Option<Lease> {
51        Some(Lease {
52            stream_key: stream_key.to_string(),
53            epoch: 0,
54        })
55    }
56    async fn renew(&self, _lease: &Lease) -> Result<()> {
57        Ok(())
58    }
59    async fn release(&self, _stream_key: &str) {}
60    fn owns(&self, _stream_key: &str) -> bool {
61        true
62    }
63}
64
65#[cfg(test)]
66mod tests {
67    use super::*;
68
69    #[tokio::test]
70    async fn self_owns_all_always_acquires_and_owns() {
71        let o = SelfOwnsAll;
72        let lease = o
73            .try_acquire("a1")
74            .await
75            .expect("standalone always acquires");
76        assert_eq!(lease.stream_key, "a1");
77        assert_eq!(lease.epoch, 0);
78        assert!(o.owns("a1"));
79        assert!(o.owns("anything-else"));
80        o.renew(&lease).await.unwrap();
81        o.release("a1").await; // no-op, must not panic
82    }
83}
prospero_core/ownership.rs

prospero_core/
ownership.rs
