appcore_control_plane/
file.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: file.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/23 23:50:45 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 14:12:17 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11//! Crash-consistent file-backed reference control plane.
12
13use super::memory::{InMemoryControlPlane, InMemoryState};
14use super::*;
15use appcore_core::{Clock, SystemClock};
16use fs2::FileExt;
17use std::fs::{self, File, OpenOptions};
18use std::io::{Read, Write};
19use std::path::{Path, PathBuf};
20use std::sync::atomic::{AtomicU64, Ordering};
21
22const STATE_FORMAT_VERSION: u16 = 1;
23const STATE_FILE: &str = "control-plane-state-v1.json";
24const LOCK_FILE: &str = "control-plane-state.lock";
25const MAX_CONTROL_PLANE_STATE_BYTES: u64 = 16 * 1024 * 1024;
26// appcore-norm: allow(global-state) reason: atomic sequence prevents process-local temporary path collisions
27static TEMP_COUNTER: AtomicU64 = AtomicU64::new(1);
28
29/// Durable control-plane implementation for one shared deployment directory.
30///
31/// Every operation takes an operating-system file lock, reloads validated
32/// state, applies one contract operation, and atomically persists the result.
33/// The deployment directory is the authentication and isolation boundary and
34/// is created with owner-only permissions on Unix.
35#[derive(Clone)]
36pub struct FileControlPlane {
37    root: PathBuf,
38    state_path: PathBuf,
39    lock_path: PathBuf,
40    retention_ms: u64,
41    clock: Arc<dyn Clock>,
42}
43
44impl std::fmt::Debug for FileControlPlane {
45    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        formatter
47            .debug_struct("FileControlPlane")
48            .field("root", &self.root)
49            .field("retention_ms", &self.retention_ms)
50            .finish_non_exhaustive()
51    }
52}
53
54impl FileControlPlane {
55    /// Opens or creates a durable reference control plane.
56    pub fn open(root: impl Into<PathBuf>, retention_ms: u64) -> ControlPlaneResult<Self> {
57        Self::with_clock(root, retention_ms, Arc::new(SystemClock::new()))
58    }
59
60    /// Opens a control plane using an explicit authoritative server clock.
61    pub fn with_clock(
62        root: impl Into<PathBuf>,
63        retention_ms: u64,
64        clock: Arc<dyn Clock>,
65    ) -> ControlPlaneResult<Self> {
66        if retention_ms == 0 {
67            return Err(ControlPlaneError::Rejected(
68                "control-plane retention must be greater than zero".to_string(),
69            ));
70        }
71        let root = root.into();
72        prepare_root(&root)?;
73        let control = Self {
74            state_path: root.join(STATE_FILE),
75            lock_path: root.join(LOCK_FILE),
76            root,
77            retention_ms,
78            clock,
79        };
80        control.initialize()?;
81        Ok(control)
82    }
83
84    /// Returns the durable state path.
85    pub fn state_path(&self) -> &Path {
86        &self.state_path
87    }
88
89    /// Returns the presence retention window in milliseconds.
90    pub fn retention_ms(&self) -> u64 {
91        self.retention_ms
92    }
93
94    /// Creates an integrity-validated point-in-time state backup.
95    pub fn backup_to(&self, destination: impl AsRef<Path>) -> ControlPlaneResult<()> {
96        let _lock = self.lock_exclusive()?;
97        let envelope = self.load_envelope()?;
98        let encoded = encode_envelope(&envelope)?;
99        write_atomic(destination.as_ref(), &encoded)
100    }
101
102    /// Replaces state from a validated backup.
103    pub fn restore_from(&self, source: impl AsRef<Path>) -> ControlPlaneResult<()> {
104        reject_symlink(source.as_ref())?;
105        let encoded = read_bounded(source.as_ref(), "control-plane backup read")?;
106        let _ = decode_envelope(&encoded)?;
107        let _lock = self.lock_exclusive()?;
108        write_atomic(&self.state_path, &encoded)
109    }
110
111    fn initialize(&self) -> ControlPlaneResult<()> {
112        let _lock = self.lock_exclusive()?;
113        if self.state_path.exists() {
114            let _ = self.load_envelope()?;
115            return Ok(());
116        }
117        self.save_control(&InMemoryControlPlane::default())
118    }
119
120    fn load_control(&self) -> ControlPlaneResult<InMemoryControlPlane> {
121        let envelope = self.load_envelope()?;
122        Ok(InMemoryControlPlane::from_state(envelope.state))
123    }
124
125    fn load_envelope(&self) -> ControlPlaneResult<StateEnvelope> {
126        reject_symlink(&self.state_path)?;
127        let encoded = read_bounded(&self.state_path, "control-plane state read")?;
128        decode_envelope(&encoded)
129    }
130
131    fn save_control(&self, control: &InMemoryControlPlane) -> ControlPlaneResult<()> {
132        let envelope = StateEnvelope {
133            format_version: STATE_FORMAT_VERSION,
134            state: control.snapshot()?,
135        };
136        write_atomic(&self.state_path, &encode_envelope(&envelope)?)
137    }
138
139    fn prune(&self, control: &InMemoryControlPlane, now_ms: u64) -> ControlPlaneResult<()> {
140        let cutoff = now_ms.saturating_sub(self.retention_ms);
141        let _ = control.prune_registrations(cutoff)?;
142        Ok(())
143    }
144
145    fn lock_exclusive(&self) -> ControlPlaneResult<File> {
146        reject_symlink(&self.lock_path)?;
147        let mut options = OpenOptions::new();
148        options.create(true).read(true).write(true);
149        #[cfg(unix)]
150        {
151            use std::os::unix::fs::OpenOptionsExt;
152            options.mode(0o600);
153        }
154        let file = options
155            .open(&self.lock_path)
156            .map_err(|error| transport_error("control-plane lock open", error))?;
157        file.lock_exclusive()
158            .map_err(|error| transport_error("control-plane lock acquire", error))?;
159        Ok(file)
160    }
161}
162
163impl ControlPlaneProvider for FileControlPlane {
164    fn register<'a>(
165        &'a self,
166        mut registration: CoreRegistration,
167    ) -> ControlPlaneFuture<'a, CorePresence> {
168        Box::pin(async move {
169            let _lock = self.lock_exclusive()?;
170            let control = self.load_control()?;
171            let now_ms = self.clock.now_ms();
172            registration.registered_at_ms = now_ms;
173            self.prune(&control, now_ms)?;
174            let result = control.register(registration).await?;
175            self.save_control(&control)?;
176            Ok(result)
177        })
178    }
179
180    fn heartbeat<'a>(
181        &'a self,
182        mut request: HeartbeatRequest,
183    ) -> ControlPlaneFuture<'a, HeartbeatResponse> {
184        Box::pin(async move {
185            let _lock = self.lock_exclusive()?;
186            let control = self.load_control()?;
187            let now_ms = self.clock.now_ms();
188            request.sent_at_ms = now_ms;
189            self.prune(&control, now_ms)?;
190            let result = control.heartbeat(request).await?;
191            self.save_control(&control)?;
192            Ok(result)
193        })
194    }
195
196    fn discover_peers<'a>(
197        &'a self,
198        identity: &'a CoreIdentity,
199    ) -> ControlPlaneFuture<'a, PeerDirectory> {
200        Box::pin(async move {
201            let _lock = self.lock_exclusive()?;
202            let control = self.load_control()?;
203            let now_ms = self.clock.now_ms();
204            self.prune(&control, now_ms)?;
205            let mut result = control.discover_peers(identity).await?;
206            result.refreshed_at_ms = now_ms;
207            self.save_control(&control)?;
208            Ok(result)
209        })
210    }
211
212    fn acquire_or_renew_service_lease<'a>(
213        &'a self,
214        identity: &'a CoreIdentity,
215        service_id: &'a ServiceId,
216        ttl_ms: u64,
217        _client_now_ms: u64,
218    ) -> ControlPlaneFuture<'a, ServiceLeaderLease> {
219        Box::pin(async move {
220            let _lock = self.lock_exclusive()?;
221            let control = self.load_control()?;
222            let result = control
223                .acquire_or_renew_service_lease(identity, service_id, ttl_ms, self.clock.now_ms())
224                .await?;
225            self.save_control(&control)?;
226            Ok(result)
227        })
228    }
229
230    fn release_service_lease<'a>(
231        &'a self,
232        lease: ServiceLeaderLease,
233    ) -> ControlPlaneFuture<'a, ()> {
234        Box::pin(async move {
235            let _lock = self.lock_exclusive()?;
236            let control = self.load_control()?;
237            control.release_service_lease(lease).await?;
238            self.save_control(&control)
239        })
240    }
241}
242
243#[derive(Debug, serde::Serialize, serde::Deserialize)]
244#[serde(deny_unknown_fields)]
245struct StateEnvelope {
246    format_version: u16,
247    state: InMemoryState,
248}
249
250fn encode_envelope(envelope: &StateEnvelope) -> ControlPlaneResult<Vec<u8>> {
251    let encoded = serde_json::to_vec(envelope)
252        .map_err(|error| ControlPlaneError::InvalidResponse(error.to_string()))?;
253    if encoded.len() as u64 > MAX_CONTROL_PLANE_STATE_BYTES {
254        return Err(ControlPlaneError::Rejected(
255            "control-plane state exceeds configured limit".to_string(),
256        ));
257    }
258    Ok(encoded)
259}
260
261fn decode_envelope(encoded: &[u8]) -> ControlPlaneResult<StateEnvelope> {
262    let envelope = serde_json::from_slice::<StateEnvelope>(encoded).map_err(|_| {
263        ControlPlaneError::InvalidResponse("NO MORE SUPPORTED PLEASE UPDATE".to_string())
264    })?;
265    if envelope.format_version != STATE_FORMAT_VERSION {
266        return Err(ControlPlaneError::InvalidResponse(
267            "NO MORE SUPPORTED PLEASE UPDATE".to_string(),
268        ));
269    }
270    Ok(envelope)
271}
272
273fn prepare_root(root: &Path) -> ControlPlaneResult<()> {
274    reject_symlink(root)?;
275    fs::create_dir_all(root)
276        .map_err(|error| transport_error("control-plane directory create", error))?;
277    #[cfg(unix)]
278    {
279        use std::os::unix::fs::PermissionsExt;
280        fs::set_permissions(root, fs::Permissions::from_mode(0o700))
281            .map_err(|error| transport_error("control-plane directory permissions", error))?;
282    }
283    Ok(())
284}
285
286fn reject_symlink(path: &Path) -> ControlPlaneResult<()> {
287    match fs::symlink_metadata(path) {
288        Ok(metadata) if metadata.file_type().is_symlink() => Err(ControlPlaneError::Rejected(
289            "control-plane path cannot be a symlink".to_string(),
290        )),
291        Ok(_) => Ok(()),
292        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
293        Err(error) => Err(transport_error("control-plane path inspection", error)),
294    }
295}
296
297fn read_bounded(path: &Path, operation: &str) -> ControlPlaneResult<Vec<u8>> {
298    reject_symlink(path)?;
299    let mut file = File::open(path).map_err(|error| transport_error(operation, error))?;
300    let mut encoded = Vec::new();
301    Read::by_ref(&mut file)
302        .take(MAX_CONTROL_PLANE_STATE_BYTES.saturating_add(1))
303        .read_to_end(&mut encoded)
304        .map_err(|error| transport_error(operation, error))?;
305    if encoded.len() as u64 > MAX_CONTROL_PLANE_STATE_BYTES {
306        return Err(ControlPlaneError::Rejected(
307            "control-plane state exceeds configured limit".to_string(),
308        ));
309    }
310    Ok(encoded)
311}
312
313fn write_atomic(path: &Path, bytes: &[u8]) -> ControlPlaneResult<()> {
314    let parent = path.parent().unwrap_or_else(|| Path::new("."));
315    fs::create_dir_all(parent)
316        .map_err(|error| transport_error("control-plane parent create", error))?;
317    let temp = parent.join(format!(
318        ".control-plane.{}.{}.tmp",
319        std::process::id(),
320        TEMP_COUNTER.fetch_add(1, Ordering::Relaxed)
321    ));
322    let result = write_and_replace(&temp, path, parent, bytes);
323    if result.is_err() {
324        let _ = fs::remove_file(temp);
325    }
326    result
327}
328
329fn write_and_replace(
330    temp: &Path,
331    path: &Path,
332    _parent: &Path,
333    bytes: &[u8],
334) -> ControlPlaneResult<()> {
335    let mut options = OpenOptions::new();
336    options.create_new(true).write(true);
337    #[cfg(unix)]
338    {
339        use std::os::unix::fs::OpenOptionsExt;
340        options.mode(0o600);
341    }
342    let mut file = options
343        .open(temp)
344        .map_err(|error| transport_error("control-plane temp create", error))?;
345    file.write_all(bytes)
346        .and_then(|_| file.sync_all())
347        .map_err(|error| transport_error("control-plane state write", error))?;
348    fs::rename(temp, path)
349        .map_err(|error| transport_error("control-plane state replace", error))?;
350    #[cfg(unix)]
351    File::open(_parent)
352        .and_then(|directory| directory.sync_all())
353        .map_err(|error| transport_error("control-plane directory sync", error))?;
354    Ok(())
355}
356
357fn transport_error(operation: &str, error: std::io::Error) -> ControlPlaneError {
358    ControlPlaneError::Transport(format!("{operation}: {error}"))
359}
appcore_control_plane/file.rs

appcore_control_plane/
file.rs
