feat(primitive): add barrier and queue primitives

build.rs

@@ -11,6 +11,8 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             "proto/maintenance.proto",
             "proto/lock.proto",
             "proto/election.proto",
+            "proto/barrier.proto",
+            "proto/queue.proto",
         ],
         &["proto/"],
     )?;

proto/barrier.proto

@@ -0,0 +1,60 @@
+syntax = "proto3";
+
+package aether;
+
+import "common.proto";
+
+// AetherBarrier provides distributed barrier operations.
+// A barrier is a synchronization primitive: while the barrier exists,
+// other callers that try to enter it will block until the barrier
+// is released or its lease expires.
+service AetherBarrier {
+    // Create creates and holds a barrier with the given name.
+    // Other callers trying to enter will block until this barrier
+    // is released or its lease expires.
+    // Returns a barrier key that can be used to release or query.
+    rpc Create(BarrierCreateRequest) returns (BarrierCreateResponse);
+    // Release releases the barrier, unblocking all waiters.
+    rpc Release(BarrierReleaseRequest) returns (BarrierReleaseResponse);
+    // Query returns whether the barrier is currently held.
+    rpc Query(BarrierQueryRequest) returns (BarrierQueryResponse);
+}
+
+message BarrierCreateRequest {
+    // name is the barrier name.
+    bytes name = 1;
+    // lease_id is the lease to associate with the barrier.
+    // If 0, the barrier will not be associated with a lease
+    // and must be explicitly released.
+    int64 lease_id = 2;
+}
+
+message BarrierCreateResponse {
+    ResponseHeader header = 1;
+    // key is the key created for the barrier.
+    bytes key = 2;
+}
+
+message BarrierReleaseRequest {
+    // name is the barrier name to release.
+    bytes name = 1;
+}
+
+message BarrierReleaseResponse {
+    ResponseHeader header = 1;
+}
+
+message BarrierQueryRequest {
+    // name is the barrier name to query.
+    bytes name = 1;
+}
+
+message BarrierQueryResponse {
+    ResponseHeader header = 1;
+    // held is true if the barrier is currently held.
+    bool held = 2;
+    // key is the barrier key if held, empty otherwise.
+    bytes key = 3;
+    // lease_id is the lease associated with the barrier, 0 if none.
+    int64 lease_id = 4;
+}

proto/queue.proto

@@ -0,0 +1,58 @@
+syntax = "proto3";
+
+package aether;
+
+import "common.proto";
+
+// AetherQueue provides distributed FIFO queue operations.
+// Items are stored with sequential keys under a named queue prefix.
+// Enqueue appends an item; Dequeue pops the front item atomically.
+service AetherQueue {
+    // Enqueue adds an item to the named queue.
+    rpc Enqueue(QueueEnqueueRequest) returns (QueueEnqueueResponse);
+    // Dequeue removes and returns the front item from the named queue.
+    // Returns NOT_FOUND if the queue is empty.
+    rpc Dequeue(QueueDequeueRequest) returns (QueueDequeueResponse);
+    // Peek returns the front item without removing it.
+    // Returns NOT_FOUND if the queue is empty.
+    rpc Peek(QueuePeekRequest) returns (QueuePeekResponse);
+}
+
+message QueueEnqueueRequest {
+    // name is the queue name.
+    bytes name = 1;
+    // value is the item to enqueue.
+    bytes value = 2;
+}
+
+message QueueEnqueueResponse {
+    ResponseHeader header = 1;
+    // key is the key assigned to the enqueued item.
+    bytes key = 2;
+}
+
+message QueueDequeueRequest {
+    // name is the queue name.
+    bytes name = 1;
+}
+
+message QueueDequeueResponse {
+    ResponseHeader header = 1;
+    // key is the key of the dequeued item.
+    bytes key = 2;
+    // value is the value of the dequeued item.
+    bytes value = 3;
+}
+
+message QueuePeekRequest {
+    // name is the queue name.
+    bytes name = 1;
+}
+
+message QueuePeekResponse {
+    ResponseHeader header = 1;
+    // key is the key of the front item.
+    bytes key = 2;
+    // value is the value of the front item.
+    bytes value = 3;
+}

src/api/barrier.rs

@@ -0,0 +1,241 @@
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::time::Duration;
+
+use tonic::{Request, Response, Status};
+
+use crate::barrier::{BarrierManager, barrier_key, validate_barrier_name};
+use crate::proto::aether_barrier_server::AetherBarrier;
+use crate::proto::{
+    BarrierCreateRequest, BarrierCreateResponse, BarrierQueryRequest, BarrierQueryResponse,
+    BarrierReleaseRequest, BarrierReleaseResponse, ResponseHeader,
+};
+use crate::raft::{self, RaftHandle, WatchEventType, require_leader};
+use crate::watch::WatchManager;
+
+/// Maximum time to wait for barrier acquisition (30 seconds).
+const BARRIER_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub struct BarrierService {
+    raft: Arc<dyn RaftHandle>,
+    node_id: u64,
+    barrier_manager: Arc<Mutex<BarrierManager>>,
+    watch_manager: Arc<WatchManager>,
+}
+
+impl BarrierService {
+    pub fn new(
+        raft: Arc<dyn RaftHandle>,
+        node_id: u64,
+        barrier_manager: Arc<Mutex<BarrierManager>>,
+        watch_manager: Arc<WatchManager>,
+    ) -> Self {
+        Self {
+            raft,
+            node_id,
+            barrier_manager,
+            watch_manager,
+        }
+    }
+
+    fn header(&self) -> ResponseHeader {
+        ResponseHeader {
+            cluster_id: 0,
+            member_id: self.node_id,
+            revision: 0,
+            raft_term: self.raft.term(),
+        }
+    }
+
+    async fn propose(&self, request: raft::RaftRequest) -> Result<raft::RaftResponse, Status> {
+        require_leader(self.raft.as_ref(), self.node_id)?;
+        self.raft
+            .propose(request)
+            .await
+            .map_err(|e| Status::internal(format!("raft write failed: {e}")))
+    }
+
+    /// Try to create a barrier, blocking if it's already held.
+    /// Creates a watch BEFORE proposing to avoid missing release events.
+    async fn create_with_wait(&self, name: Vec<u8>, lease_id: i64) -> Result<Vec<u8>, Status> {
+        // Create a watch on the barrier key BEFORE proposing
+        // This ensures we don't miss any Delete events
+        let key = barrier_key(&name);
+        let (watch_id, mut watch_rx) = self
+            .watch_manager
+            .create(
+                key,
+                Vec::new(), // exact key match
+                vec![WatchEventType::Delete],
+                false,
+            )
+            .await;
+
+        // Helper to cancel watch
+        let cancel_watch = || {
+            let wm = self.watch_manager.clone();
+            async move {
+                wm.cancel(watch_id, "barrier create completed".to_string())
+                    .await;
+            }
+        };
+
+        loop {
+            // Propose barrier create to Raft
+            let resp = match self
+                .propose(raft::RaftRequest::BarrierCreate {
+                    name: name.clone(),
+                    lease_id,
+                })
+                .await
+            {
+                Ok(resp) => resp,
+                Err(e) => {
+                    cancel_watch().await;
+                    return Err(e);
+                }
+            };
+
+            match resp {
+                raft::RaftResponse::BarrierCreate { key } => {
+                    cancel_watch().await;
+                    return Ok(key);
+                }
+                raft::RaftResponse::BarrierAlreadyHeld { .. } => {
+                    // Wait for the barrier key to be deleted
+                    tracing::debug!(
+                        barrier = %String::from_utf8_lossy(&name),
+                        "barrier is held, waiting for release"
+                    );
+
+                    // Wait for the key to be deleted or timeout
+                    let result = tokio::time::timeout(BARRIER_TIMEOUT, async {
+                        while let Some(resp) = watch_rx.recv().await {
+                            if resp.canceled {
+                                break;
+                            }
+                            for event in &resp.events {
+                                if event.event_type == WatchEventType::Delete {
+                                    return true;
+                                }
+                            }
+                        }
+                        false
+                    })
+                    .await;
+
+                    match result {
+                        Ok(true) => {
+                            // Barrier was released, retry create
+                            tracing::debug!(
+                                barrier = %String::from_utf8_lossy(&name),
+                                "barrier released, retrying create"
+                            );
+                            continue;
+                        }
+                        Ok(false) => {
+                            // Watch was canceled unexpectedly
+                            cancel_watch().await;
+                            return Err(Status::internal("watch canceled unexpectedly"));
+                        }
+                        Err(_) => {
+                            // Timeout
+                            cancel_watch().await;
+                            return Err(Status::deadline_exceeded(format!(
+                                "barrier create timed out after {} seconds: barrier is held",
+                                BARRIER_TIMEOUT.as_secs()
+                            )));
+                        }
+                    }
+                }
+                raft::RaftResponse::Error { message } => {
+                    cancel_watch().await;
+                    return Err(Status::internal(message));
+                }
+                _ => {
+                    cancel_watch().await;
+                    return Err(Status::internal("unexpected response type"));
+                }
+            }
+        }
+    }
+}
+
+#[tonic::async_trait]
+impl AetherBarrier for BarrierService {
+    async fn create(
+        &self,
+        request: Request<BarrierCreateRequest>,
+    ) -> Result<Response<BarrierCreateResponse>, Status> {
+        let req = request.into_inner();
+
+        validate_barrier_name(&req.name).map_err(Status::invalid_argument)?;
+
+        if req.lease_id < 0 {
+            return Err(Status::invalid_argument("lease_id must be non-negative"));
+        }
+
+        // Use create_with_wait to block if barrier is already held
+        let key = self.create_with_wait(req.name, req.lease_id).await?;
+
+        Ok(Response::new(BarrierCreateResponse {
+            header: Some(self.header()),
+            key,
+        }))
+    }
+
+    async fn release(
+        &self,
+        request: Request<BarrierReleaseRequest>,
+    ) -> Result<Response<BarrierReleaseResponse>, Status> {
+        let req = request.into_inner();
+
+        validate_barrier_name(&req.name).map_err(Status::invalid_argument)?;
+
+        let resp = self
+            .propose(raft::RaftRequest::BarrierRelease { name: req.name })
+            .await?;
+
+        match resp {
+            raft::RaftResponse::BarrierRelease {} => Ok(Response::new(BarrierReleaseResponse {
+                header: Some(self.header()),
+            })),
+            raft::RaftResponse::Error { message } => Err(Status::internal(message)),
+            _ => Err(Status::internal("unexpected response type")),
+        }
+    }
+
+    async fn query(
+        &self,
+        request: Request<BarrierQueryRequest>,
+    ) -> Result<Response<BarrierQueryResponse>, Status> {
+        let req = request.into_inner();
+
+        validate_barrier_name(&req.name).map_err(Status::invalid_argument)?;
+
+        require_leader(self.raft.as_ref(), self.node_id)?;
+
+        let mgr = self
+            .barrier_manager
+            .lock()
+            .map_err(|e| Status::internal(format!("barrier manager lock poisoned: {e}")))?;
+
+        let key = mgr
+            .get_key(&req.name)
+            .map(|k| k.to_vec())
+            .unwrap_or_default();
+        let held = !key.is_empty();
+        let lease_id = if held {
+            mgr.get_lease_id(&key).unwrap_or(0)
+        } else {
+            0
+        };
+
+        Ok(Response::new(BarrierQueryResponse {
+            header: Some(self.header()),
+            held,
+            key,
+            lease_id,
+        }))
+    }
+}

src/api/mod.rs

@@ -1,4 +1,5 @@
 mod auth;
+mod barrier;
 mod cluster;
 mod election;
 pub mod health;
@@ -7,15 +8,18 @@ mod lease;
 mod lock;
 pub mod maintenance;
 pub mod metrics;
+mod queue;
 mod shard;
 mod watch;
 
 pub use self::auth::AuthService;
+pub use self::barrier::BarrierService;
 pub use self::cluster::ClusterService;
 pub use self::election::ElectionService;
 pub use self::kv::KvService;
 pub use self::lease::LeaseService;
 pub use self::lock::LockService;
 pub use self::maintenance::MaintenanceService;
+pub use self::queue::QueueService;
 pub use self::shard::ShardService;
 pub use self::watch::WatchService;