[sqlite-watcher] add unix socket listener and tests

Author: taariq

Cargo.lock

@@ -28,3 +28,4 @@ tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 [dev-dependencies]
 tempfile = "3.8"
 tokio = { version = "1.35", features = ["rt", "macros"] }
+tower = "0.4"

sqlite-watcher/Cargo.toml

@@ -32,11 +32,11 @@ Flag summary:
 - `--log-level`: Tracing filter (also settable via `SQLITE_WATCHER_LOG`).
 - `--poll-interval-ms`: How often to check the WAL file for growth (default 500 ms). Lower values react faster but cost more syscalls.
 - `--min-event-bytes`: Minimum WAL byte growth before emitting an event. Use larger values to avoid spam when very small transactions occur.
-- `--listen` + `--token-file` now control the embedded gRPC server. Clients must send `Authorization: Bearer <token>` metadata when calling the `Watcher` service (see `proto/watcher.proto`). Unix sockets/pipes are placeholders until Ticket D is completed; TCP listens on `127.0.0.1:<port>`.
+- `--listen` + `--token-file` now control the embedded gRPC server. Clients must send `Authorization: Bearer <token>` metadata when calling the `Watcher` service (see `proto/watcher.proto`). TCP (`tcp:50051`) and Unix sockets (`unix:/tmp/sqlite-watcher.sock`) are available today; Windows named pipes currently fall back to TCP until native support lands.
 
 ## Cross-platform notes
 
-- **Linux/macOS**: Default listener is a Unix domain socket at `/tmp/sqlite-watcher.sock`. Ensure the target SQLite database enables WAL journaling.
+- **Linux/macOS**: Default listener is a Unix domain socket at `/tmp/sqlite-watcher.sock`. The watcher cleans up stale socket files on startup; point `--listen unix:/path` elsewhere if needed.
 - **Windows**: Unix sockets are disabled; pass `--listen tcp:50051` or `--listen pipe:SerenWatcher`. Named pipes allow local service accounts without opening TCP ports.
 - All platforms expect the token file to live under `~/.seren/sqlite-watcher/token` by default; create the directory with `0700` permissions so the watcher refuses to start if the secret is world-readable.
 - The current WAL watcher polls the `*.sqlite-wal` file for byte growth. To keep WAL history available, configure your writers with `PRAGMA journal_mode=WAL;` and raise `wal_autocheckpoint` (or disable it) so the SQLite engine does not aggressively truncate the log.

sqlite-watcher/README.md

@@ -10,7 +10,9 @@ use anyhow::{anyhow, bail, Context, Result};
 use clap::Parser;
 use sqlite_watcher::decoder::WalGrowthDecoder;
 use sqlite_watcher::queue::ChangeQueue;
-use sqlite_watcher::server::TcpServerHandle;
+#[cfg(unix)]
+use sqlite_watcher::server::spawn_unix_server;
+use sqlite_watcher::server::{spawn_tcp_server, ServerHandle};
 use sqlite_watcher::wal::{start_wal_watcher, WalWatcherConfig as TailConfig};
 use tracing_subscriber::EnvFilter;
 
@@ -281,21 +283,25 @@ fn start_grpc_server(
     listen: &ListenAddress,
     queue_path: &Path,
     token: &str,
-) -> Result<Option<TcpServerHandle>> {
+) -> Result<Option<ServerHandle>> {
     match listen {
         ListenAddress::Tcp { host, port } => {
             let addr: SocketAddr = format!("{}:{}", host, port)
                 .parse()
                 .with_context(|| format!("invalid tcp listen address {host}:{port}"))?;
-            let handle = TcpServerHandle::spawn(addr, queue_path.to_path_buf(), token.to_string())?;
+            let handle = spawn_tcp_server(addr, queue_path.to_path_buf(), token.to_string())?;
             Ok(Some(handle))
         }
         ListenAddress::Unix(path) => {
-            tracing::warn!(
-                path = %path.display(),
-                "unix socket gRPC transport is not yet implemented"
-            );
-            Ok(None)
+            #[cfg(unix)]
+            {
+                let handle = spawn_unix_server(path, queue_path.to_path_buf(), token.to_string())?;
+                Ok(Some(handle))
+            }
+            #[cfg(not(unix))]
+            {
+                bail!("unix sockets are not supported on this platform")
+            }
         }
         ListenAddress::Pipe(name) => {
             tracing::warn!(pipe = name, "named pipe transport is not yet implemented");

sqlite-watcher/src/main.rs

@@ -1,5 +1,5 @@
 use std::net::SocketAddr;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::thread::{self, JoinHandle};
 
@@ -18,55 +18,115 @@ use crate::watcher_proto::{
     SetStateRequest, SetStateResponse,
 };
 
-pub struct TcpServerHandle {
+#[cfg(unix)]
+use tokio::net::UnixListener;
+#[cfg(unix)]
+use tokio_stream::wrappers::UnixListenerStream;
+
+pub struct ServerHandle {
     shutdown: Option<oneshot::Sender<()>>,
     thread: Option<JoinHandle<Result<()>>>,
+    #[cfg(unix)]
+    unix_path: Option<PathBuf>,
 }
 
-impl Drop for TcpServerHandle {
+impl Drop for ServerHandle {
     fn drop(&mut self) {
         if let Some(tx) = self.shutdown.take() {
             let _ = tx.send(());
         }
         if let Some(handle) = self.thread.take() {
             let _ = handle.join();
         }
+        #[cfg(unix)]
+        if let Some(path) = self.unix_path.take() {
+            let _ = std::fs::remove_file(path);
+        }
     }
 }
 
-impl TcpServerHandle {
-    pub fn spawn(addr: SocketAddr, queue_path: PathBuf, token: String) -> Result<Self> {
-        let (shutdown_tx, shutdown_rx) = oneshot::channel();
-        let thread = thread::spawn(move || -> Result<()> {
-            let runtime = Builder::new_multi_thread()
-                .enable_all()
-                .build()
-                .context("failed to build tokio runtime")?;
-            runtime.block_on(async move {
-                let listener = tokio::net::TcpListener::bind(addr)
-                    .await
-                    .context("failed to bind tcp listener")?;
-                let queue_path = Arc::new(queue_path);
-                let svc = WatcherService::new(queue_path.clone());
-                let interceptor = AuthInterceptor {
-                    token: Arc::new(token),
-                };
-                Server::builder()
-                    .add_service(WatcherServer::with_interceptor(svc, interceptor))
-                    .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async move {
-                        let _ = shutdown_rx.await;
-                    })
-                    .await
-                    .context("grpc server exited with error")?;
-                Ok(())
-            })
-        });
-
-        Ok(Self {
-            shutdown: Some(shutdown_tx),
-            thread: Some(thread),
-        })
+pub fn spawn_tcp_server(
+    addr: SocketAddr,
+    queue_path: PathBuf,
+    token: String,
+) -> Result<ServerHandle> {
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let thread = thread::spawn(move || -> Result<()> {
+        let runtime = Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .context("failed to build tokio runtime")?;
+        runtime.block_on(async move {
+            let listener = tokio::net::TcpListener::bind(addr)
+                .await
+                .context("failed to bind tcp listener")?;
+            let queue_path = Arc::new(queue_path);
+            let svc = WatcherService::new(queue_path);
+            let interceptor = AuthInterceptor {
+                token: Arc::new(token),
+            };
+            Server::builder()
+                .add_service(WatcherServer::with_interceptor(svc, interceptor))
+                .serve_with_incoming_shutdown(TcpListenerStream::new(listener), async move {
+                    let _ = shutdown_rx.await;
+                })
+                .await
+                .context("grpc server exited with error")?;
+            Ok::<(), anyhow::Error>(())
+        })?;
+        Ok(())
+    });
+
+    Ok(ServerHandle {
+        shutdown: Some(shutdown_tx),
+        thread: Some(thread),
+        #[cfg(unix)]
+        unix_path: None,
+    })
+}
+
+#[cfg(unix)]
+pub fn spawn_unix_server(path: &Path, queue_path: PathBuf, token: String) -> Result<ServerHandle> {
+    if path.exists() {
+        std::fs::remove_file(path)
+            .with_context(|| format!("failed to remove stale unix socket {}", path.display()))?;
+    }
+    if let Some(parent) = path.parent() {
+        std::fs::create_dir_all(parent)
+            .with_context(|| format!("failed to create unix socket dir {}", parent.display()))?;
     }
+    let path_buf = path.to_path_buf();
+    let (shutdown_tx, shutdown_rx) = oneshot::channel();
+    let path_for_drop = path_buf.clone();
+    let thread = thread::spawn(move || -> Result<()> {
+        let runtime = Builder::new_multi_thread()
+            .enable_all()
+            .build()
+            .context("failed to build tokio runtime")?;
+        runtime.block_on(async move {
+            let listener = UnixListener::bind(&path_buf).context("failed to bind unix socket")?;
+            let queue_path = Arc::new(queue_path);
+            let svc = WatcherService::new(queue_path);
+            let interceptor = AuthInterceptor {
+                token: Arc::new(token),
+            };
+            Server::builder()
+                .add_service(WatcherServer::with_interceptor(svc, interceptor))
+                .serve_with_incoming_shutdown(UnixListenerStream::new(listener), async move {
+                    let _ = shutdown_rx.await;
+                })
+                .await
+                .context("grpc server exited with error")?;
+            Ok::<(), anyhow::Error>(())
+        })?;
+        Ok(())
+    });
+
+    Ok(ServerHandle {
+        shutdown: Some(shutdown_tx),
+        thread: Some(thread),
+        unix_path: Some(path_for_drop),
+    })
 }
 
 struct WatcherService {
@@ -83,6 +143,30 @@ impl WatcherService {
     }
 }
 
+#[derive(Clone)]
+struct AuthInterceptor {
+    token: Arc<String>,
+}
+
+impl Interceptor for AuthInterceptor {
+    fn call(&mut self, request: Request<()>) -> Result<Request<()>, Status> {
+        let header = request
+            .metadata()
+            .get("authorization")
+            .ok_or_else(|| Status::unauthenticated("missing authorization header"))?;
+        let expected = format!("Bearer {}", self.token.as_ref());
+        if header
+            .to_str()
+            .map(|value| value == expected)
+            .unwrap_or(false)
+        {
+            Ok(request)
+        } else {
+            Err(Status::unauthenticated("invalid authorization header"))
+        }
+    }
+}
+
 #[tonic::async_trait]
 impl Watcher for WatcherService {
     async fn health_check(
@@ -196,27 +280,3 @@ fn change_to_proto(row: crate::queue::ChangeRecord) -> Change {
         cursor: row.cursor.unwrap_or_default(),
     }
 }
-
-#[derive(Clone)]
-struct AuthInterceptor {
-    token: Arc<String>,
-}
-
-impl Interceptor for AuthInterceptor {
-    fn call(&mut self, request: Request<()>) -> Result<Request<()>, Status> {
-        let header = request
-            .metadata()
-            .get("authorization")
-            .ok_or_else(|| Status::unauthenticated("missing authorization header"))?;
-        let expected = format!("Bearer {}", self.token.as_ref());
-        if header
-            .to_str()
-            .map(|value| value == expected)
-            .unwrap_or(false)
-        {
-            Ok(request)
-        } else {
-            Err(Status::unauthenticated("invalid authorization header"))
-        }
-    }
-}

sqlite-watcher/src/server.rs

@@ -1,7 +1,9 @@
 use std::net::SocketAddr;
 use std::time::Duration;
 
-use sqlite_watcher::server::TcpServerHandle;
+use sqlite_watcher::server::spawn_tcp_server;
+#[cfg(unix)]
+use sqlite_watcher::server::spawn_unix_server;
 use sqlite_watcher::watcher_proto::watcher_client::WatcherClient;
 use sqlite_watcher::watcher_proto::HealthCheckRequest;
 use tempfile::tempdir;
@@ -15,7 +17,7 @@ async fn health_check_responds_ok() {
     let addr: SocketAddr = "127.0.0.1:55051".parse().unwrap();
     let token = "secret-token".to_string();
 
-    let _handle = TcpServerHandle::spawn(addr, queue_path, token.clone()).unwrap();
+    let _handle = spawn_tcp_server(addr, queue_path, token.clone()).unwrap();
     sleep(Duration::from_millis(200)).await;
 
     let channel = tonic::transport::Channel::from_shared(format!("http://{}", addr))
@@ -30,3 +32,33 @@ async fn health_check_responds_ok() {
     let resp = client.health_check(req).await.unwrap();
     assert_eq!(resp.into_inner().status, "ok");
 }
+#[cfg(unix)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn health_check_over_unix_socket() {
+    use tokio::net::UnixStream;
+    use tonic::transport::Endpoint;
+    use tower::service_fn;
+
+    let dir = tempdir().unwrap();
+    let queue_path = dir.path().join("queue.db");
+    let socket_path = dir.path().join("watcher.sock");
+    let token = "secret-token".to_string();
+
+    let _handle = spawn_unix_server(&socket_path, queue_path, token.clone()).unwrap();
+    sleep(Duration::from_millis(200)).await;
+
+    let endpoint = Endpoint::try_from("http://[::]:50051").unwrap();
+    let channel = endpoint
+        .connect_with_connector(service_fn(move |_: tonic::transport::Uri| {
+            let path = socket_path.clone();
+            async move { UnixStream::connect(path).await }
+        }))
+        .await
+        .unwrap();
+    let mut client = WatcherClient::new(channel);
+    let mut req = tonic::Request::new(HealthCheckRequest {});
+    let header = MetadataValue::try_from(format!("Bearer {}", token)).unwrap();
+    req.metadata_mut().insert("authorization", header);
+    let resp = client.health_check(req).await.unwrap();
+    assert_eq!(resp.into_inner().status, "ok");
+}