feat(wal): add independently deployable replica service

.github/workflows/rust-test.yml

@@ -101,6 +101,10 @@ jobs:
         run: |
           cargo clippy -p lance-context-wal --features peer-transport --all-targets -- -D warnings
           cargo test -p lance-context-wal --features peer-transport --lib --test replication_process
+      - name: Standalone WAL service and automatic maintenance
+        run: |
+          cargo clippy -p lance-context-wal --features service --all-targets -- -D warnings
+          cargo test -p lance-context-wal --features service --test node_service
       - name: WAL resource observation in a container cgroup namespace
         run: |
           cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl"

crates/lance-context-wal/Cargo.toml

@@ -7,6 +7,7 @@ publish = false
 description = "Independent Rust WAL experiments owned by lance-context"
 
 [features]
+service = ["peer-transport", "dep:hyper", "dep:hyper-util", "dep:http-body-util", "dep:clap", "tokio/signal", "tokio/rt-multi-thread"]
 peer-transport = ["runtime", "vectored-io", "dep:tokio-rustls", "tokio/net", "tokio/time", "tokio/macros"]
 runtime = ["replication", "dep:tokio", "tokio/sync", "tokio/rt"]
 # Real Lance materialization/recovery qualification; normal WAL replicas do not link Lance.
@@ -25,6 +26,9 @@ ipc = ["dep:anyhow", "dep:arrow-array", "dep:arrow-buffer", "dep:arrow-ipc", "de
 lab = ["ipc", "dep:clap", "dep:hdrhistogram", "dep:raft-engine", "raft-engine/internals", "dep:raft-proto", "dep:serde_json"]
 
 [dependencies]
+hyper = { version = "1", features = ["server", "http1"], optional = true }
+hyper-util = { version = "0.1", features = ["tokio"], optional = true }
+http-body-util = { version = "0.1", optional = true }
 tokio-rustls = { version = "0.26.4", default-features = false, features = ["ring", "tls12"], optional = true }
 lance = { version = "13.0.0", optional = true }
 lance-table = { version = "13.0.0", optional = true }
@@ -64,6 +68,11 @@ bincode = "1.3"
 arrow-ipc = { version = "58", features = ["zstd"] }
 tempfile = "3"
 
+[[bin]]
+name = "context-wal-node"
+path = "src/bin/node.rs"
+required-features = ["service"]
+
 [[bin]]
 name = "context-wal-runner"
 path = "src/bin/runner.rs"
@@ -78,3 +87,7 @@ required-features = ["runner"]
 name = "context-wal-lab"
 path = "src/bin/lab.rs"
 required-features = ["lab"]
+
+[[test]]
+name = "node_service"
+required-features = ["service"]

crates/lance-context-wal/src/bin/node.rs

@@ -0,0 +1,49 @@
+use anyhow::Result;
+use clap::Parser;
+use lance_context_wal::service::{Config, Service};
+use std::path::PathBuf;
+
+#[derive(Parser)]
+#[command(
+    about = "Independent durable WAL node; configuration and signing identities persist across restarts"
+)]
+struct Args {
+    #[arg(long)]
+    config: PathBuf,
+    #[arg(long,default_value_t=2,value_parser=clap::value_parser!(u16).range(1..=64))]
+    io_threads: u16,
+    #[arg(long,default_value_t=4,value_parser=clap::value_parser!(u16).range(1..=64))]
+    blocking_threads: u16,
+}
+async fn signal() -> Result<()> {
+    #[cfg(unix)]
+    {
+        let mut terminate =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+        tokio::select! { result=tokio::signal::ctrl_c()=>result?, _=terminate.recv()=>{} }
+    }
+    #[cfg(not(unix))]
+    tokio::signal::ctrl_c().await?;
+    Ok(())
+}
+fn main() -> Result<()> {
+    let args = Args::parse();
+    let config = Config::read(&args.config)?;
+    tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .worker_threads(usize::from(args.io_threads))
+        .max_blocking_threads(usize::from(args.blocking_threads))
+        .build()?
+        .block_on(async {
+            let mut service = Service::start(config).await?;
+            tokio::select! {
+                result=signal()=>{
+                    // Even signal-handler setup failure must join the running
+                    // service before dropping this Tokio runtime.
+                    let stopped=service.shutdown().await;
+                    result.and(stopped)
+                },
+                result=service.wait()=>result,
+            }
+        })
+}

crates/lance-context-wal/src/lib.rs

@@ -14,6 +14,9 @@ pub mod raft_log;
 #[cfg(feature = "replication")]
 pub mod replication;
 
+#[cfg(feature = "service")]
+pub mod service;
+
 #[cfg(feature = "admission")]
 pub mod admission;
 

crates/lance-context-wal/src/raft_log.rs

@@ -348,6 +348,8 @@ pub struct PhysicalLog {
 /// Local physical-file tuning, independent of the replicated layout and policy.
 /// Thresholds trigger explicit maintenance; they are not disk-admission quotas.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct LogOptions {
     pub target_file_bytes: u64,
     /// Native file-count size estimate; oversized batches can exceed it.

crates/lance-context-wal/src/raft_log/budget.rs

@@ -9,6 +9,8 @@ use crate::admission::Snapshot;
 /// accepted Raft/control traffic and physical rewrites. Peer writes must still
 /// persist and can consume this window. This is not a hard filesystem quota.
 #[derive(Clone, Copy, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(deny_unknown_fields))]
 pub struct DiskOptions {
     pub max_bytes: usize,
     pub data_bytes: usize,

crates/lance-context-wal/src/replication.rs

@@ -92,7 +92,7 @@ impl Default for Limits {
     }
 }
 impl Limits {
-    fn identity(&self) -> Result<Vec<u8>> {
+    pub(crate) fn identity(&self) -> Result<Vec<u8>> {
         ensure!(
             (1..=16_384).contains(&self.max_tables)
                 && self.catalog_key.is_none_or(|key| key != [0; 32])
@@ -525,6 +525,18 @@ impl Node {
         })
     }
 
+    pub(crate) fn checkpoint_index(&self, group: u64) -> Result<u64> {
+        self.healthy()?;
+        Ok(self
+            .groups
+            .get(&group)
+            .context("unhosted group")?
+            .raw
+            .store()
+            .first_index()?
+            .saturating_sub(1))
+    }
+
     fn request_pending(&self, request: u64) -> bool {
         self.groups
             .values()

crates/lance-context-wal/src/replication/runtime.rs

@@ -38,13 +38,16 @@ pub mod peer;
 /// Counts do not include caller allocations or an oversized allocation pinned
 /// by a small Bytes slice. The transport must account for those owners too.
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     pub data_requests: usize,
     pub data_bytes: usize,
     pub control_requests: usize,
     pub control_bytes: usize,
     pub peer_requests: usize,
     pub peer_bytes: usize,
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub tick_interval: Duration,
     /// Maximum commands from each lane before polling durable Raft progress.
     pub batch_requests: usize,
@@ -64,7 +67,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self) -> Result<()> {
+    pub(crate) fn validate(&self) -> Result<()> {
         for (count, bytes) in [
             (self.data_requests, self.data_bytes),
             (self.control_requests, self.control_bytes),
@@ -245,6 +248,8 @@ struct Shared {
 #[derive(Clone, Debug)]
 pub struct Observation {
     pub status: Status,
+    /// Persisted image boundary; local scheduling data, not a quorum result.
+    pub checkpoint_index: u64,
     pub observed_at: Instant,
     pub polling: bool,
     pub stopped: bool,
@@ -477,7 +482,14 @@ impl Runtime {
             layout: node.layout_hash,
             policy: node.policy,
             id: node.id,
-            voters: Arc::new(node.layout.groups.clone()),
+            voters: Arc::new(
+                node.layout
+                    .groups
+                    .iter()
+                    .filter(|(_, voters)| voters.contains(&node.id))
+                    .map(|(&group, voters)| (group, voters.clone()))
+                    .collect(),
+            ),
             max_record: node.limits.max_record_bytes as usize,
         };
         // Validate every initial view before starting any thread. A failure
@@ -488,14 +500,16 @@ impl Runtime {
             .map(|node| {
                 let group = *node.groups.keys().next().context("empty worker group")?;
                 let status = node.status(group)?;
-                Ok((node, group, status))
+                let checkpoint_index = node.checkpoint_index(group)?;
+                Ok((node, group, status, checkpoint_index))
             })
             .collect::<Result<Vec<_>>>()?;
         let mut routes = BTreeMap::new();
         let mut workers: Vec<JoinHandle<Result<()>>> = Vec::new();
-        for (node, group, status) in nodes {
+        for (node, group, status, checkpoint_index) in nodes {
             let observation = Arc::new(Mutex::new(Observation {
                 status,
+                checkpoint_index,
                 observed_at: Instant::now(),
                 polling: false,
                 stopped: false,
@@ -753,8 +767,10 @@ impl Worker {
                     completion();
                 }
                 let status = self.node.status(self.group)?;
+                let checkpoint_index = self.node.checkpoint_index(self.group)?;
                 let mut observation = self.observation.lock().unwrap();
                 observation.status = status;
+                observation.checkpoint_index = checkpoint_index;
                 observation.observed_at = Instant::now();
                 observation.polling = false;
             }

crates/lance-context-wal/src/replication/runtime/client.rs

@@ -49,6 +49,8 @@ pub struct Endpoint {
     pub certificate: [u8; 32],
 }
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     /// Logical live request/response bytes, independent for the two lanes.
     /// Caller allocations and allocator/decoded metadata overhead are separate.
@@ -59,6 +61,7 @@ pub struct Options {
     /// Per server listener, shared across all independently running clients.
     pub server_connections: usize,
     /// Entire exchange, including pool/admission waits and durable completion.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub timeout: Duration,
 }
 impl Default for Options {
@@ -73,7 +76,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self, max_record: usize) -> Result<()> {
+    pub(crate) fn validate(&self, max_record: usize) -> Result<()> {
         ensure!(
             (1..=32).contains(&self.connections)
                 && (1..=256).contains(&self.server_connections)
@@ -259,6 +262,9 @@ pub struct Server {
     tasks: Vec<JoinHandle<Result<()>>>,
 }
 impl Server {
+    pub fn is_finished(&self) -> bool {
+        self.tasks.iter().any(|task| task.is_finished())
+    }
     /// Trusted service certificates may access this node's configured groups.
     /// End-user authorization and table binding stay at the application ingress.
     pub fn spawn(

crates/lance-context-wal/src/replication/runtime/peer.rs

@@ -50,6 +50,8 @@ pub struct Peer {
 }
 
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     pub receive_bytes: usize,
     pub connections: usize,
@@ -58,7 +60,9 @@ pub struct Options {
     /// follower cannot retain the entire runtime outbox's byte allowance.
     pub peer_bytes: usize,
     /// Deadline for a complete handshake or frame, including bounded admission.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub io_timeout: Duration,
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub reconnect_delay: Duration,
 }
 impl Default for Options {
@@ -74,7 +78,7 @@ impl Default for Options {
     }
 }
 impl Options {
-    fn validate(&self) -> Result<()> {
+    pub(crate) fn validate(&self) -> Result<()> {
         ensure!(
             (MAX_HEADER + MAX_BODY..=256 << 20).contains(&self.receive_bytes)
                 && (1..=128).contains(&self.connections)
@@ -223,6 +227,10 @@ impl Transport {
     pub fn statistics(&self) -> &Arc<Statistics> {
         &self.statistics
     }
+    /// A listener or peer supervisor ended; the embedding service must drain.
+    pub fn is_finished(&self) -> bool {
+        self.tasks.iter().any(|task| task.is_finished())
+    }
     /// Close network tasks before shutting down/reopening the worker runtime.
     pub async fn shutdown(mut self) -> Result<()> {
         let _ = self.stop.send(true);

crates/lance-context-wal/src/replication/runtime/peer/snapshot.rs

@@ -43,6 +43,8 @@ pub trait Admission: Send + Sync + 'static {
 }
 
 #[derive(Clone, Debug)]
+#[cfg_attr(feature = "service", derive(serde::Serialize, serde::Deserialize))]
+#[cfg_attr(feature = "service", serde(default, deny_unknown_fields))]
 pub struct Options {
     /// Total queued and active outgoing transfers across all destinations.
     pub outbound_transfers: usize,
@@ -51,8 +53,10 @@ pub struct Options {
     pub max_image_bytes: u64,
     /// Whole transfer deadline, not reset on every chunk. A filesystem call
     /// already in progress is still joined, not cancelled, at shutdown.
+    #[cfg_attr(feature = "service", serde(with = "crate::service::duration_millis"))]
     pub transfer_timeout: Duration,
     #[cfg(test)]
+    #[cfg_attr(feature = "service", serde(skip))]
     pub(super) pause_after_payload_chunk: Option<std::path::PathBuf>,
 }
 impl Default for Options {
@@ -148,6 +152,11 @@ pub struct Transport {
     statistics: Arc<Statistics>,
 }
 impl Transport {
+    pub fn is_finished(&self) -> bool {
+        self.supervisor
+            .as_ref()
+            .is_none_or(|task| task.is_finished())
+    }
     pub fn spawn(handle: Handle, listener: TcpListener, settings: Settings) -> Result<Self> {
         ensure!(
             handle.disk.is_some(),