storage: stage frontend listener activation

cmd/unbounded-storage/ARCHITECTURE.md

@@ -178,10 +178,16 @@ shutdown; teardown drains terminal reports and returns a failing process status
 instead of continuing with a missing shard.
 
 Backend and frontend reloads construct replacements before swapping them into
-their shard-local registries. A failed replacement therefore leaves the prior
-resource, binding, and stripe geometry live and retryable. This is a
-per-resource guarantee; route, peer, shard, and disk publication across the
-whole process is not yet one transaction.
+their shard-local registries. Network frontends first validate and bind a
+dormant `SO_REUSEPORT` socket, then enter the listening state only during
+activation. A failed preparation therefore leaves the prior resource, binding,
+and stripe geometry live and retryable. Active frontend entries record the
+binding and stripe geometry they realized, so a failed rebuild is detected and
+retried on the next apply. Replacement uses a staged transaction: the complete
+frontend change set is prepared before activation and swaps into the live
+registry only after every listener activates successfully. This is not yet
+a whole-process guarantee; route, peer, shard, and disk publication across
+the whole process is not yet one transaction.
 
 ### Shutdown
 
@@ -220,10 +226,11 @@ per-shard `!Send` object graph:
    `LiveShardLocalStore` because a `PageRef` resolves through exactly one
    backing's geometry.
 6. **Frontends.** A shard hosts a `FrontendRegistry` of any number of
-   frontends keyed by component name. Each spec binds its listener with `SO_REUSEPORT` and
-   builds an `HttpDriver`/`S3Driver`; the registry can add and remove frontends
-   in place on a live config apply (a removed driver's `Drop` closes its
-   listener fd).
+   frontends keyed by component name. Each network frontend is prepared as a
+   bound but non-listening `SO_REUSEPORT` socket, then activated into an
+   `HttpDriver`/`S3Driver` that owns the listening socket. The registry can add
+   and remove frontends in place on a live config apply; RAII closes removed
+   listeners after their pending accept future is dropped.
 7. **Tick loop.** Register tick hooks (socket-ring `progress()`, the
    control-drain hook that reconciles backends/frontends/routing from applied
    configs, and the frontend registry's `progress()`), report `Up`, and run

cmd/unbounded-storage/src/frontend/http_serve.rs

@@ -32,7 +32,7 @@
 use std::collections::VecDeque;
 use std::future::{Future, poll_fn};
 use std::net::SocketAddr;
-use std::os::fd::RawFd;
+use std::os::fd::{AsRawFd, RawFd};
 use std::pin::Pin;
 use std::rc::Rc;
 use std::task::{Context, Poll, Waker};
@@ -51,9 +51,9 @@ use crate::frontend::range::{
 };
 use crate::frontend::serve_metrics::{ConnGuard, ReqOutcome};
 use crate::http::{
-    FdGuard, HttpRequest, MAX_HEADER_BYTES, Method, ParseError, RECV_CHUNK, bind_listener,
-    connection_header_value, request_allows_keep_alive, send_all, serialize_response_head,
-    split_query,
+    BoundListener, FdGuard, HttpRequest, ListeningSocket, MAX_HEADER_BYTES, Method, ParseError,
+    RECV_CHUNK, bind_socket, connection_header_value, request_allows_keep_alive, send_all,
+    serialize_response_head, split_query,
 };
 use crate::ring::NetHandle;
 use crate::runtime::noop_waker;
@@ -118,14 +118,16 @@ impl HttpFrontend {
         self.max_requests_per_connection
     }
 
-    /// Create, bind, and listen the per-shard accept socket with
-    /// `SO_REUSEPORT` (and `SO_REUSEADDR`) so every shard accepts on the
-    /// same port, returning the listening [`RawFd`]. Linux-only.
-    ///
-    /// The caller owns the returned fd and is responsible for closing
-    /// it; [`HttpDriver`] only reads it for `accept`.
-    pub fn bind_listener(&self) -> Result<RawFd, FrontendError> {
-        bind_listener(self.addr).map_err(|_| FrontendError::BadBind(self.addr.to_string()))
+    /// Bind a dormant per-shard `SO_REUSEPORT` socket.
+    pub fn bind_socket(&self) -> Result<BoundListener, FrontendError> {
+        bind_socket(self.addr).map_err(|_| FrontendError::BadBind(self.addr.to_string()))
+    }
+
+    /// Bind and activate the per-shard listener.
+    pub fn bind_listener(&self) -> Result<ListeningSocket, FrontendError> {
+        self.bind_socket()?
+            .listen()
+            .map_err(|_| FrontendError::BadBind(self.addr.to_string()))
     }
 }
 
@@ -143,7 +145,6 @@ impl HttpFrontend {
 pub struct HttpDriver<P: BufferPool<Req = StripeReq> + 'static> {
     pool: Rc<P>,
     handle: NetHandle,
-    listen_fd: RawFd,
     frontend_id: Rc<str>,
     backend_id: String,
     cache_id: Option<String>,
@@ -153,12 +154,14 @@ pub struct HttpDriver<P: BufferPool<Req = StripeReq> + 'static> {
     bypass: bool,
     max_requests_per_connection: usize,
     accept_fut: Pin<Box<dyn Future<Output = std::io::Result<RawFd>>>>,
+    // Declared after the future so the pending accept is cancelled first.
+    listener: ListeningSocket,
     conns: Vec<Pin<Box<dyn Future<Output = ()>>>>,
     waker: Waker,
 }
 
 impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
-    /// Build a serving engine over a bound `listen_fd`.
+    /// Build a serving engine that owns an active listener.
     ///
     /// `stripe_size` and `page_size` come from the shard's pool
     /// geometry. `fanout` is this shard's view of the stripe-ownership
@@ -168,7 +171,7 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
     pub fn new(
         pool: Rc<P>,
         handle: NetHandle,
-        listen_fd: RawFd,
+        listener: ListeningSocket,
         frontend_id: Rc<str>,
         backend_id: String,
         cache_id: Option<String>,
@@ -178,11 +181,10 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
         bypass: bool,
         max_requests_per_connection: usize,
     ) -> Self {
-        let accept_fut = Box::pin(handle.accept(listen_fd));
+        let accept_fut = Box::pin(handle.accept(listener.as_raw_fd()));
         Self {
             pool,
             handle,
-            listen_fd,
             frontend_id,
             backend_id,
             cache_id,
@@ -192,6 +194,7 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
             bypass,
             max_requests_per_connection,
             accept_fut,
+            listener,
             conns: Vec::new(),
             waker: noop_waker(),
         }
@@ -231,7 +234,8 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
                             unsafe {
                                 libc::close(fd);
                             }
-                            self.accept_fut = Box::pin(self.handle.accept(self.listen_fd));
+                            self.accept_fut =
+                                Box::pin(self.handle.accept(self.listener.as_raw_fd()));
                             continue;
                         }
                         let serve = serve_connection(
@@ -251,7 +255,7 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
                     }
                     // Rearm regardless of accept success so a transient
                     // accept error does not stop the listener.
-                    self.accept_fut = Box::pin(self.handle.accept(self.listen_fd));
+                    self.accept_fut = Box::pin(self.handle.accept(self.listener.as_raw_fd()));
                 }
                 Poll::Pending => break,
             }
@@ -271,24 +275,6 @@ impl<P: BufferPool<Req = StripeReq> + 'static> HttpDriver<P> {
     }
 }
 
-impl<P: BufferPool<Req = StripeReq> + 'static> Drop for HttpDriver<P> {
-    /// Close the listen fd the driver owns. The driver is the sole owner
-    /// of this `SO_REUSEPORT` listener (the embedder hands it the fd from
-    /// `bind_listener` and never touches it again), so dropping the
-    /// driver (on shard shutdown or a live frontend removal) must close
-    /// it or the fd leaks. The accept future borrows only the fd number,
-    /// not ownership, so closing here is sound once the driver is gone.
-    fn drop(&mut self) {
-        if self.listen_fd >= 0 {
-            // SAFETY: `listen_fd` was returned by `bind_listener` and is
-            // owned exclusively by this driver; it is closed exactly once.
-            unsafe {
-                libc::close(self.listen_fd);
-            }
-        }
-    }
-}
-
 ///
 /// Owns everything it needs so the future is `'static` and can live in
 /// the driver's future set across ticks. All paths close `conn_fd` via
@@ -1362,8 +1348,8 @@ mod tests {
             }
         };
         let handle = NetHandle::new(ring);
-        let listen_fd = match bind_listener("127.0.0.1:0".parse().unwrap()) {
-            Ok(fd) => fd,
+        let listener = match bind_socket("127.0.0.1:0".parse().unwrap()).and_then(|b| b.listen()) {
+            Ok(listener) => listener,
             Err(e) => {
                 eprintln!("driver_idle_progress: bind failed: {e}; skipping");
                 return;
@@ -1374,7 +1360,7 @@ mod tests {
                 err: Error::from("mock pool has no data"),
             }),
             handle,
-            listen_fd,
+            listener,
             Rc::from("primary"),
             "primary".to_string(),
             None,

cmd/unbounded-storage/src/frontend/s3_serve.rs

@@ -24,7 +24,7 @@
 
 use std::future::Future;
 use std::net::SocketAddr;
-use std::os::fd::RawFd;
+use std::os::fd::{AsRawFd, RawFd};
 use std::pin::Pin;
 use std::rc::Rc;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -40,8 +40,8 @@ use crate::frontend::range::{ByteRange, RangeError, ResolvedRange, full_object,
 use crate::frontend::s3_xml::{S3ErrorCode, error_xml};
 use crate::frontend::serve_metrics::{ConnGuard, ReqOutcome};
 use crate::http::{
-    FdGuard, HttpRequest, MAX_HEADER_BYTES, Method, ParseError, RECV_CHUNK, bind_listener,
-    send_all, serialize_response_head, split_query,
+    BoundListener, FdGuard, HttpRequest, ListeningSocket, MAX_HEADER_BYTES, Method, ParseError,
+    RECV_CHUNK, bind_socket, send_all, serialize_response_head, split_query,
 };
 use crate::ring::NetHandle;
 use crate::runtime::noop_waker;
@@ -102,14 +102,16 @@ impl S3Frontend {
         self.addr
     }
 
-    /// Create, bind, and listen the per-shard accept socket with
-    /// `SO_REUSEPORT` (and `SO_REUSEADDR`) so every shard accepts on the
-    /// same port, returning the listening [`RawFd`]. Linux-only.
-    ///
-    /// The caller owns the returned fd and is responsible for closing
-    /// it; [`S3Driver`] only reads it for `accept`.
-    pub fn bind_listener(&self) -> Result<RawFd, FrontendError> {
-        bind_listener(self.addr).map_err(|_| FrontendError::BadBind(self.addr.to_string()))
+    /// Bind a dormant per-shard `SO_REUSEPORT` socket.
+    pub fn bind_socket(&self) -> Result<BoundListener, FrontendError> {
+        bind_socket(self.addr).map_err(|_| FrontendError::BadBind(self.addr.to_string()))
+    }
+
+    /// Bind and activate the per-shard listener.
+    pub fn bind_listener(&self) -> Result<ListeningSocket, FrontendError> {
+        self.bind_socket()?
+            .listen()
+            .map_err(|_| FrontendError::BadBind(self.addr.to_string()))
     }
 }
 
@@ -125,47 +127,48 @@ impl S3Frontend {
 pub struct S3Driver<P: BufferPool<Req = StripeReq> + 'static> {
     pool: Rc<P>,
     handle: NetHandle,
-    listen_fd: RawFd,
     frontend_id: Rc<str>,
     backend_id: String,
     cache_id: Option<String>,
     stripe_size: u64,
     page_size: usize,
     bypass: bool,
     accept_fut: Pin<Box<dyn Future<Output = std::io::Result<RawFd>>>>,
+    // Declared after the future so the pending accept is cancelled first.
+    listener: ListeningSocket,
     conns: Vec<Pin<Box<dyn Future<Output = ()>>>>,
     waker: Waker,
 }
 
 impl<P: BufferPool<Req = StripeReq> + 'static> S3Driver<P> {
-    /// Build a serving engine over a bound `listen_fd`.
+    /// Build a serving engine that owns an active listener.
     ///
     /// `stripe_size` and `page_size` come from the shard's pool
     /// geometry. When `bypass` is set, the frontend bridges straight to
     /// its backend, skipping the disk cache and peer routing.
     pub fn new(
         pool: Rc<P>,
         handle: NetHandle,
-        listen_fd: RawFd,
+        listener: ListeningSocket,
         frontend_id: Rc<str>,
         backend_id: String,
         cache_id: Option<String>,
         stripe_size: u64,
         page_size: usize,
         bypass: bool,
     ) -> Self {
-        let accept_fut = Box::pin(handle.accept(listen_fd));
+        let accept_fut = Box::pin(handle.accept(listener.as_raw_fd()));
         Self {
             pool,
             handle,
-            listen_fd,
             frontend_id,
             backend_id,
             cache_id,
             stripe_size,
             page_size,
             bypass,
             accept_fut,
+            listener,
             conns: Vec::new(),
             waker: noop_waker(),
         }
@@ -215,7 +218,7 @@ impl<P: BufferPool<Req = StripeReq> + 'static> S3Driver<P> {
                     }
                     // Rearm regardless of accept success so a transient
                     // accept error does not stop the listener.
-                    self.accept_fut = Box::pin(self.handle.accept(self.listen_fd));
+                    self.accept_fut = Box::pin(self.handle.accept(self.listener.as_raw_fd()));
                 }
                 Poll::Pending => break,
             }
@@ -235,24 +238,6 @@ impl<P: BufferPool<Req = StripeReq> + 'static> S3Driver<P> {
     }
 }
 
-impl<P: BufferPool<Req = StripeReq> + 'static> Drop for S3Driver<P> {
-    /// Close the listen fd the driver owns. The driver is the sole owner
-    /// of this `SO_REUSEPORT` listener (the embedder hands it the fd from
-    /// `bind_listener` and never touches it again), so dropping the
-    /// driver (on shard shutdown or a live frontend removal) must close
-    /// it or the fd leaks. The accept future borrows only the fd number,
-    /// not ownership, so closing here is sound once the driver is gone.
-    fn drop(&mut self) {
-        if self.listen_fd >= 0 {
-            // SAFETY: `listen_fd` was returned by `bind_listener` and is
-            // owned exclusively by this driver; it is closed exactly once.
-            unsafe {
-                libc::close(self.listen_fd);
-            }
-        }
-    }
-}
-
 /// Serve one accepted connection end-to-end, then close its fd.
 ///
 /// Owns everything it needs so the future is `'static` and can live in
@@ -864,8 +849,8 @@ mod tests {
             }
         };
         let handle = NetHandle::new(ring);
-        let listen_fd = match bind_listener("127.0.0.1:0".parse().unwrap()) {
-            Ok(fd) => fd,
+        let listener = match bind_socket("127.0.0.1:0".parse().unwrap()).and_then(|b| b.listen()) {
+            Ok(listener) => listener,
             Err(e) => {
                 eprintln!("driver_idle_progress: bind failed: {e}; skipping");
                 return;
@@ -874,7 +859,7 @@ mod tests {
         let mut driver = S3Driver::new(
             Rc::new(MockPool),
             handle,
-            listen_fd,
+            listener,
             Rc::from("primary"),
             "primary".to_string(),
             None,

cmd/unbounded-storage/src/http/mod.rs

@@ -37,6 +37,8 @@ pub use connection::{
 pub use headers::{Header, ParseError};
 pub use request::{HttpRequest, serialize_request};
 pub use response::{ResponseHead, serialize_response_head};
+#[cfg(target_os = "linux")]
+pub use server::{BoundListener, ListeningSocket};
 
 /// Cross-platform, storage-policy-free server plumbing reused by the
 /// HTTP serving frontends.
@@ -45,7 +47,7 @@ pub(crate) use server::{MAX_HEADER_BYTES, RECV_CHUNK, split_query};
 /// Linux-only server plumbing that depends on `libc` and the io_uring
 /// [`NetHandle`](crate::ring::NetHandle).
 #[cfg(target_os = "linux")]
-pub(crate) use server::{FdGuard, bind_listener, send_all};
+pub(crate) use server::{FdGuard, bind_socket, send_all};
 
 /// Canonical typed `Method`/`StatusCode`, re-exported from the [`http`]
 /// crate so the rest of the crate has a single source for them.