sui_dockerfile_node_cache_daemon/
lib.rs

1//! Phase 3b of `supa-charge-akeyless-ci`: a node-local L0 disk cache
2//! daemon.
3//!
4//! One instance runs per Kubernetes node (as a `DaemonSet`, see the
5//! `sui-dockerfile-node-cache-daemon` Helm chart). It listens on a
6//! Unix domain socket and answers `Get`/`Put`/`Warm` requests (see
7//! [`protocol`]) against a node-local disk cache ([`store`]), falling
8//! through to the existing remote
9//! [`sui_cache::storage::StorageBackend`] (Postgres L2 / Redis L1,
10//! unmodified — see [`server`]) on a local miss, persisting the result
11//! locally so the next lookup on the same node is a zero-network-hop
12//! local hit.
13//!
14//! ## Why a Unix domain socket, not TCP-localhost
15//!
16//! A UDS avoids the kernel's TCP/IP stack entirely (no port allocation,
17//! no `SO_REUSEADDR` races across pod restarts, no risk of binding a
18//! port another node-local process expects), and its filesystem-path
19//! identity is exactly the shape a `DaemonSet`'s `hostPath` mount and a
20//! runner pod's matching mount need to share to discover each other
21//! (see the Helm chart's `values.yaml` for the mount-path contract).
22//! Standard UNIX file permissions on the socket's parent directory are
23//! also a real (if coarse) access boundary that a TCP port lacks. A
24//! TCP-localhost fallback was judged unnecessary complexity for a
25//! same-host-only protocol — this crate does not implement one.
26//!
27//! ## Reuse, not reinvention
28//!
29//! This crate does **not** invent a new remote cache abstraction — the
30//! "remote tier" side of [`server::NodeCacheDaemon`] is exactly
31//! [`sui_cache::storage::StorageBackend`], the same trait
32//! `sui-dockerfile-wrapper` (Phase 2) and `sui cache serve` already
33//! consume. Only the local L0 tier ([`store::LocalCacheStore`]) and the
34//! wire protocol ([`protocol`]) are new.
35
36#![warn(clippy::pedantic)]
37#![allow(clippy::module_name_repetitions)]
38
39pub mod protocol;
40pub mod server;
41pub mod store;
42
43use std::path::{Path, PathBuf};
44use std::sync::Arc;
45
46use tokio::net::{UnixListener, UnixStream};
47use tokio::sync::watch;
48
49pub use protocol::{CachedArtifact, DaemonRequest, DaemonResponse, WarmStatus};
50pub use server::NodeCacheDaemon;
51pub use store::{LocalCacheStore, MockLocalCacheStore, RealLocalCacheStore};
52
53/// Errors this crate's server/store/protocol layers can produce.
54/// Always typed, never a panic on a fallible path.
55#[derive(Debug, thiserror::Error)]
56pub enum DaemonError {
57    #[error("io: {0}")]
58    Io(#[from] std::io::Error),
59    #[error("protocol: {0}")]
60    Protocol(String),
61    #[error("remote cache backend: {0}")]
62    Remote(#[from] sui_cache::CacheError),
63}
64
65/// Bind a `UnixListener` at `socket_path`, removing any stale socket
66/// file left behind by a prior (crashed) instance first — a fresh bind
67/// on an existing path otherwise fails with `AddrInUse`.
68///
69/// # Errors
70///
71/// Propagates any I/O failure removing the stale path or binding.
72pub fn bind_unix_listener(socket_path: &Path) -> Result<UnixListener, DaemonError> {
73    if socket_path.exists() {
74        std::fs::remove_file(socket_path)?;
75    }
76    if let Some(parent) = socket_path.parent() {
77        std::fs::create_dir_all(parent)?;
78    }
79    Ok(UnixListener::bind(socket_path)?)
80}
81
82/// Serve requests on `listener` until `shutdown` reports `true`.
83/// Each accepted connection handles exactly one request-response pair
84/// then closes — this daemon's protocol is not multiplexed, mirroring
85/// its "quick node-local lookup" workload rather than a long-lived
86/// session.
87///
88/// Graceful shutdown: `tokio::select!` races `listener.accept()`
89/// against the shutdown watch, so an in-flight connection being
90/// accepted never races a mid-shutdown drop; already-spawned
91/// connection tasks are allowed to finish on their own (fire-and-forget
92/// `Warm` background fetches are, by design, best-effort and may be
93/// dropped on shutdown — never a correctness requirement).
94pub async fn serve<L: LocalCacheStore + 'static>(
95    listener: UnixListener,
96    daemon: Arc<NodeCacheDaemon<L>>,
97    mut shutdown: watch::Receiver<bool>,
98) {
99    loop {
100        tokio::select! {
101            accepted = listener.accept() => {
102                match accepted {
103                    Ok((stream, _addr)) => {
104                        let daemon = Arc::clone(&daemon);
105                        tokio::spawn(async move {
106                            if let Err(e) = handle_connection(stream, &daemon).await {
107                                tracing::warn!(
108                                    target: "sui-dockerfile-node-cache-daemon",
109                                    error = %e,
110                                    "connection handling failed"
111                                );
112                            }
113                        });
114                    }
115                    Err(e) => {
116                        tracing::warn!(target: "sui-dockerfile-node-cache-daemon", error = %e, "accept failed");
117                    }
118                }
119            }
120            _ = shutdown.changed() => {
121                if *shutdown.borrow() {
122                    tracing::info!(target: "sui-dockerfile-node-cache-daemon", "graceful shutdown requested");
123                    break;
124                }
125            }
126        }
127    }
128}
129
130async fn handle_connection<L: LocalCacheStore + 'static>(
131    mut stream: UnixStream,
132    daemon: &NodeCacheDaemon<L>,
133) -> Result<(), DaemonError> {
134    let request: DaemonRequest = protocol::read_message(&mut stream).await?;
135    let response = daemon.handle_request(request).await;
136    protocol::write_message(&mut stream, &response).await
137}
138
139/// Default socket path — the well-known integration-contract path a
140/// `DaemonSet` pod and a same-node GitHub Actions runner pod must both
141/// mount (as a shared `hostPath`, see the Helm chart) to discover each
142/// other. Also the default the `sui-dockerfile-wrapper`
143/// `DaemonAwareCacheClient` probes when a caller doesn't override it.
144#[must_use]
145pub fn default_socket_path() -> PathBuf {
146    PathBuf::from("/var/run/sui-dockerfile-cache/node-cache.sock")
147}
148
149#[cfg(test)]
150mod tests {
151    use super::*;
152    use async_trait::async_trait;
153    use sui_cache::storage::StorageBackend;
154    use sui_cache::CacheError;
155    use tokio::net::UnixStream as ClientStream;
156
157    struct EmptyRemote;
158    #[async_trait]
159    impl StorageBackend for EmptyRemote {
160        async fn get_narinfo(&self, _hash: &str) -> Result<Option<String>, CacheError> {
161            Ok(None)
162        }
163        async fn put_narinfo(&self, _hash: &str, _content: &str) -> Result<(), CacheError> {
164            Ok(())
165        }
166        async fn get_nar(&self, _path: &str) -> Result<Option<Vec<u8>>, CacheError> {
167            Ok(None)
168        }
169        async fn put_nar(&self, _path: &str, _data: &[u8]) -> Result<(), CacheError> {
170            Ok(())
171        }
172        async fn delete(&self, _hash: &str) -> Result<(), CacheError> {
173            Ok(())
174        }
175        async fn list_narinfos(&self) -> Result<Vec<String>, CacheError> {
176            Ok(Vec::new())
177        }
178    }
179
180    #[tokio::test]
181    async fn end_to_end_over_a_real_unix_socket() {
182        let dir = tempfile::tempdir().unwrap();
183        let socket_path = dir.path().join("test.sock");
184
185        let local = Arc::new(
186            MockLocalCacheStore::new().with_entry("hit", CachedArtifact { image_ref: "img:e2e".to_string() }),
187        );
188        let remote: Arc<dyn StorageBackend> = Arc::new(EmptyRemote);
189        let daemon = Arc::new(NodeCacheDaemon::new(local, remote));
190
191        let listener = bind_unix_listener(&socket_path).unwrap();
192        let (shutdown_tx, shutdown_rx) = watch::channel(false);
193        let server_task = tokio::spawn(serve(listener, daemon, shutdown_rx));
194
195        let mut client = ClientStream::connect(&socket_path).await.unwrap();
196        protocol::write_message(&mut client, &DaemonRequest::Get { content_hash: "hit".to_string() })
197            .await
198            .unwrap();
199        let resp: DaemonResponse = protocol::read_message(&mut client).await.unwrap();
200        match resp {
201            DaemonResponse::Get { artifact: Some(a) } => assert_eq!(a.image_ref, "img:e2e"),
202            other => panic!("expected Get hit, got {other:?}"),
203        }
204
205        shutdown_tx.send(true).unwrap();
206        server_task.await.unwrap();
207    }
208
209    #[test]
210    fn default_socket_path_matches_the_helm_chart_mount_contract() {
211        assert_eq!(default_socket_path(), PathBuf::from("/var/run/sui-dockerfile-cache/node-cache.sock"));
212    }
213}
sui_dockerfile_node_cache_daemon/lib.rs

sui_dockerfile_node_cache_daemon/
lib.rs
