mongreldb_server/
storage_runtime.rs

1//! Authoritative server storage runtime (P0.2 / ADR-0001).
2//!
3//! The daemon holds exactly one of:
4//! - [`ServerStorageRuntime::Standalone`] — ordinary single-node data plane
5//!   over a local [`mongreldb_core::Database`]
6//! - [`ServerStorageRuntime::Cluster`] — consensus/tablet data plane via
7//!   [`crate::cluster_runtime::ClusterRuntimeHandle`]; ordinary public writes
8//!   must not use a standalone WAL bypass
9//!
10//! Dual-root ownership (standalone user database + live cluster runtime as
11//! peer data planes) is refused.
12
13use std::path::{Path, PathBuf};
14use std::sync::Arc;
15
16use mongreldb_core::Database;
17use mongreldb_query::ai_retrieval::RemoteAiEndpoint;
18use mongreldb_query::distributed::RemoteFragmentEndpoint;
19
20use crate::cluster_runtime::ClusterRuntimeHandle;
21
22/// Error when code asks the standalone data plane for a cluster process (or
23/// the reverse).
24#[derive(Debug, Clone, PartialEq, Eq)]
25pub struct StorageRuntimeError {
26    message: String,
27}
28
29impl StorageRuntimeError {
30    /// Build a fail-closed error.
31    pub fn new(message: impl Into<String>) -> Self {
32        Self {
33            message: message.into(),
34        }
35    }
36
37    /// Stable message for the cluster standalone-bypass refusal.
38    pub fn cluster_refuses_standalone_bypass() -> Self {
39        Self::new(
40            "cluster mode refuses standalone AppState.db data-plane access; \
41             public data operations are owned by consensus/tablet state",
42        )
43    }
44}
45
46impl std::fmt::Display for StorageRuntimeError {
47    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
48        f.write_str(&self.message)
49    }
50}
51
52impl std::error::Error for StorageRuntimeError {}
53
54impl From<StorageRuntimeError> for mongreldb_core::MongrelError {
55    fn from(error: StorageRuntimeError) -> Self {
56        mongreldb_core::MongrelError::Other(error.message)
57    }
58}
59
60/// Standalone (single-node server) data plane.
61#[derive(Clone)]
62pub struct StandaloneRuntime {
63    db: Arc<Database>,
64}
65
66impl StandaloneRuntime {
67    /// Wrap an already-opened local database.
68    pub fn new(db: Arc<Database>) -> Self {
69        Self { db }
70    }
71
72    /// Local engine handle.
73    pub fn db(&self) -> &Arc<Database> {
74        &self.db
75    }
76
77    /// Durable root path for this standalone core.
78    pub fn root(&self) -> &Path {
79        self.db.root()
80    }
81}
82
83/// Cluster gateway / node data plane.
84///
85/// Holds the live [`ClusterRuntimeHandle`] and optional fragment/AI worker
86/// endpoints installed on the authenticated internal RPC transport. Public
87/// SQL/Kit/native user writes do not open a peer standalone WAL here.
88#[derive(Clone)]
89pub struct ClusterGatewayRuntime {
90    handle: ClusterRuntimeHandle,
91    fragment_endpoint: Option<Arc<RemoteFragmentEndpoint>>,
92    ai_endpoint: Option<Arc<RemoteAiEndpoint>>,
93}
94
95impl ClusterGatewayRuntime {
96    /// Wrap a started node runtime (workers may be installed later).
97    pub fn new(handle: ClusterRuntimeHandle) -> Self {
98        Self {
99            handle,
100            fragment_endpoint: None,
101            ai_endpoint: None,
102        }
103    }
104
105    /// Wrap a started node runtime that already has workers installed.
106    pub fn with_workers(
107        handle: ClusterRuntimeHandle,
108        fragment_endpoint: Arc<RemoteFragmentEndpoint>,
109        ai_endpoint: Arc<RemoteAiEndpoint>,
110    ) -> Self {
111        Self {
112            handle,
113            fragment_endpoint: Some(fragment_endpoint),
114            ai_endpoint: Some(ai_endpoint),
115        }
116    }
117
118    /// Live node runtime handle.
119    pub fn handle(&self) -> &ClusterRuntimeHandle {
120        &self.handle
121    }
122
123    /// Node data directory (identity, cluster-meta, tablet roots).
124    pub fn node_data(&self) -> &Path {
125        self.handle.node_data()
126    }
127
128    /// Installed fragment worker endpoint, when production start completed.
129    pub fn fragment_endpoint(&self) -> Option<&Arc<RemoteFragmentEndpoint>> {
130        self.fragment_endpoint.as_ref()
131    }
132
133    /// Installed AI worker endpoint, when production start completed.
134    pub fn ai_endpoint(&self) -> Option<&Arc<RemoteAiEndpoint>> {
135        self.ai_endpoint.as_ref()
136    }
137
138    /// Record worker endpoints after install (tests / staged startup).
139    pub fn set_workers(
140        &mut self,
141        fragment_endpoint: Arc<RemoteFragmentEndpoint>,
142        ai_endpoint: Arc<RemoteAiEndpoint>,
143    ) {
144        self.fragment_endpoint = Some(fragment_endpoint);
145        self.ai_endpoint = Some(ai_endpoint);
146    }
147
148    /// Whether both production workers are installed.
149    pub fn workers_installed(&self) -> bool {
150        self.fragment_endpoint.is_some() && self.ai_endpoint.is_some()
151    }
152}
153
154/// Single authoritative storage runtime for the HTTP/native daemon (P0.2).
155#[derive(Clone)]
156pub enum ServerStorageRuntime {
157    /// Single-node server-owned standalone core.
158    Standalone(StandaloneRuntime),
159    /// Cluster node: public data owned by consensus/tablet state.
160    Cluster(ClusterGatewayRuntime),
161}
162
163impl ServerStorageRuntime {
164    /// Construct the standalone variant.
165    pub fn standalone(db: Arc<Database>) -> Self {
166        Self::Standalone(StandaloneRuntime::new(db))
167    }
168
169    /// Construct the cluster variant (workers optional until installed).
170    pub fn cluster(handle: ClusterRuntimeHandle) -> Self {
171        Self::Cluster(ClusterGatewayRuntime::new(handle))
172    }
173
174    /// Construct the cluster variant with workers already installed.
175    pub fn cluster_with_workers(
176        handle: ClusterRuntimeHandle,
177        fragment_endpoint: Arc<RemoteFragmentEndpoint>,
178        ai_endpoint: Arc<RemoteAiEndpoint>,
179    ) -> Self {
180        Self::Cluster(ClusterGatewayRuntime::with_workers(
181            handle,
182            fragment_endpoint,
183            ai_endpoint,
184        ))
185    }
186
187    /// `true` when this process is a cluster node data plane.
188    pub fn is_cluster(&self) -> bool {
189        matches!(self, Self::Cluster(_))
190    }
191
192    /// `true` when this process is standalone.
193    pub fn is_standalone(&self) -> bool {
194        matches!(self, Self::Standalone(_))
195    }
196
197    /// Local standalone engine, when this is standalone mode.
198    pub fn standalone_db(&self) -> Option<&Arc<Database>> {
199        match self {
200            Self::Standalone(rt) => Some(rt.db()),
201            Self::Cluster(_) => None,
202        }
203    }
204
205    /// Require the standalone engine; fail closed in cluster mode so ordinary
206    /// public writes cannot bypass Raft through `AppState.db`.
207    pub fn require_standalone_db(&self) -> Result<&Arc<Database>, StorageRuntimeError> {
208        self.standalone_db()
209            .ok_or_else(StorageRuntimeError::cluster_refuses_standalone_bypass)
210    }
211
212    /// Live cluster handle, when this is cluster mode.
213    pub fn cluster_handle(&self) -> Option<&ClusterRuntimeHandle> {
214        match self {
215            Self::Cluster(rt) => Some(rt.handle()),
216            Self::Standalone(_) => None,
217        }
218    }
219
220    /// Cluster gateway pieces, when this is cluster mode.
221    pub fn cluster_gateway(&self) -> Option<&ClusterGatewayRuntime> {
222        match self {
223            Self::Cluster(rt) => Some(rt),
224            Self::Standalone(_) => None,
225        }
226    }
227
228    /// Mutable cluster gateway (worker install bookkeeping).
229    pub fn cluster_gateway_mut(&mut self) -> Option<&mut ClusterGatewayRuntime> {
230        match self {
231            Self::Cluster(rt) => Some(rt),
232            Self::Standalone(_) => None,
233        }
234    }
235
236    /// Durable path used for process-local server state (idempotency, pid
237    /// hints). Standalone uses the database root; cluster uses node data.
238    pub fn durable_path(&self) -> PathBuf {
239        match self {
240            Self::Standalone(rt) => rt.root().to_path_buf(),
241            Self::Cluster(rt) => rt.node_data().to_path_buf(),
242        }
243    }
244
245    /// Mode name for health / capabilities surfaces.
246    pub fn mode_name(&self) -> &'static str {
247        match self {
248            Self::Standalone(_) => "standalone",
249            Self::Cluster(_) => "cluster",
250        }
251    }
252}
253
254#[cfg(test)]
255mod tests {
256    use super::*;
257    use tempfile::tempdir;
258
259    #[test]
260    fn standalone_exposes_db_cluster_refuses() {
261        let dir = tempdir().unwrap();
262        let db = Arc::new(Database::create(dir.path()).unwrap());
263        let standalone = ServerStorageRuntime::standalone(Arc::clone(&db));
264        assert!(standalone.is_standalone());
265        assert!(!standalone.is_cluster());
266        assert!(Arc::ptr_eq(
267            standalone.require_standalone_db().unwrap(),
268            &db
269        ));
270        assert_eq!(standalone.mode_name(), "standalone");
271    }
272
273    #[test]
274    fn cluster_runtime_enum_exists_and_has_no_standalone_db() {
275        // Structural: Cluster variant holds gateway pieces without a peer
276        // standalone Database field. Construction with a live runtime is
277        // covered by cluster_runtime / storage integration tests.
278        let dir = tempdir().unwrap();
279        let _ = dir;
280        let err = StorageRuntimeError::cluster_refuses_standalone_bypass();
281        assert!(err.to_string().contains("cluster mode refuses"));
282        let converted: mongreldb_core::MongrelError = err.into();
283        assert!(converted.to_string().contains("cluster mode refuses"));
284    }
285}
mongreldb_server/storage_runtime.rs

mongreldb_server/
storage_runtime.rs
