appcore_provider/
coordination.rs

1// =============================================================================
2//        #######
3//     ###       ###     F: coordination.rs
4//    ##   ## ##   ##    P: AppCore-Runtime
5//         ## ##
6//                       C: 2026/07/23 10:59:21 by dnettoRaw
7//    ##   ## ##   ##    U: 2026/08/02 14:12:17 by dnettoRaw
8//      ###########      S: 1.0.1-rc.8
9// =============================================================================
10
11use crate::{ProviderError, ProviderResult};
12use std::fs::{self, OpenOptions};
13use std::io::Write;
14use std::path::{Path, PathBuf};
15use std::sync::atomic::{AtomicBool, Ordering};
16use std::sync::Mutex;
17
18/// Latest coordination schema understood by this Runtime release.
19pub const COORDINATION_SCHEMA_VERSION: u64 = 2;
20
21/// Runtime-owned coordination tables allowed by schema version 2.
22pub const COORDINATION_TABLES: &[&str] = &[
23    "audit",
24    "capabilities",
25    "jobs",
26    "leases",
27    "runtime_instances",
28    "runtime_versions",
29    "schema_migrations",
30    "tenants",
31];
32
33const STORE_FORMAT: &str = "appcore.coordination-store.v1";
34const METADATA_FILE: &str = "coordination-schema.meta";
35// appcore-norm: allow(global-state) reason: atomic sequence prevents process-local temporary path collisions
36static TEMP_COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
37
38/// Backend contract for the Runtime-owned coordination schema.
39///
40/// Runtime nodes normally reach this store through a control-plane provider.
41/// This contract exists for control-plane implementations and deliberately does
42/// not expose generic business-data reads or writes.
43pub trait CoordinationStoreProvider: Send + Sync {
44    /// Returns the latest migration version applied by the provider.
45    fn schema_version(&self) -> ProviderResult<u64>;
46
47    /// Verifies backend connectivity and schema access.
48    fn health(&self) -> ProviderResult<()>;
49
50    /// Rejects stores that do not implement the schema required by this Runtime.
51    fn ensure_compatible(&self) -> ProviderResult<()> {
52        self.health()?;
53        let actual = self.schema_version()?;
54        if actual < COORDINATION_SCHEMA_VERSION {
55            return Err(ProviderError::InvalidConfiguration(format!(
56                "coordination schema {actual} is older than required schema {COORDINATION_SCHEMA_VERSION}"
57            )));
58        }
59        Ok(())
60    }
61}
62
63/// Deterministic coordination store for embedded control planes and tests.
64#[derive(Debug)]
65pub struct InMemoryCoordinationStore {
66    schema_version: u64,
67    healthy: AtomicBool,
68}
69
70impl Default for InMemoryCoordinationStore {
71    fn default() -> Self {
72        Self {
73            schema_version: COORDINATION_SCHEMA_VERSION,
74            healthy: AtomicBool::new(true),
75        }
76    }
77}
78
79impl InMemoryCoordinationStore {
80    /// Creates a store at an explicit schema version for migration tests.
81    pub fn with_schema_version(schema_version: u64) -> Self {
82        Self {
83            schema_version,
84            healthy: AtomicBool::new(true),
85        }
86    }
87
88    /// Changes the deterministic health result.
89    pub fn set_healthy(&self, healthy: bool) {
90        self.healthy.store(healthy, Ordering::SeqCst);
91    }
92}
93
94impl CoordinationStoreProvider for InMemoryCoordinationStore {
95    fn schema_version(&self) -> ProviderResult<u64> {
96        Ok(self.schema_version)
97    }
98
99    fn health(&self) -> ProviderResult<()> {
100        if self.healthy.load(Ordering::SeqCst) {
101            Ok(())
102        } else {
103            Err(ProviderError::Initialization(
104                "in-memory coordination store is unhealthy".to_string(),
105            ))
106        }
107    }
108}
109
110/// Durable single-control-plane coordination schema store.
111///
112/// The file provider certifies schema ownership and crash-consistent metadata
113/// for deployments in which one control-plane service owns the data directory.
114/// Runtime nodes never receive database credentials through this contract.
115#[derive(Debug)]
116pub struct FileCoordinationStore {
117    root: PathBuf,
118    metadata_path: PathBuf,
119    lock: Mutex<()>,
120}
121
122impl FileCoordinationStore {
123    /// Opens a store, creating or transactionally migrating metadata to V2.
124    pub fn open(root: impl Into<PathBuf>) -> ProviderResult<Self> {
125        let root = root.into();
126        prepare_root(&root)?;
127        let store = Self {
128            metadata_path: root.join(METADATA_FILE),
129            root,
130            lock: Mutex::new(()),
131        };
132        store.migrate()?;
133        store.ensure_compatible()?;
134        Ok(store)
135    }
136
137    /// Returns the deployment-owned data directory.
138    pub fn root(&self) -> &Path {
139        &self.root
140    }
141
142    /// Writes a validated metadata backup using atomic replacement.
143    pub fn backup_to(&self, destination: impl AsRef<Path>) -> ProviderResult<()> {
144        let _guard = self
145            .lock
146            .lock()
147            .map_err(|_| initialization("coordination store lock poisoned"))?;
148        let metadata = self.read_metadata()?;
149        validate_metadata(&metadata)?;
150        write_atomic(destination.as_ref(), render_metadata(&metadata).as_bytes())
151    }
152
153    /// Restores validated metadata and reruns forward-only migrations.
154    pub fn restore_from(&self, source: impl AsRef<Path>) -> ProviderResult<()> {
155        let contents = fs::read_to_string(source.as_ref())
156            .map_err(|error| initialization(format!("coordination backup read failed: {error}")))?;
157        let metadata = parse_metadata(&contents)?;
158        validate_metadata(&metadata)?;
159        let _guard = self
160            .lock
161            .lock()
162            .map_err(|_| initialization("coordination store lock poisoned"))?;
163        write_atomic(&self.metadata_path, render_metadata(&metadata).as_bytes())?;
164        drop(_guard);
165        self.migrate()
166    }
167
168    fn migrate(&self) -> ProviderResult<()> {
169        let _guard = self
170            .lock
171            .lock()
172            .map_err(|_| initialization("coordination store lock poisoned"))?;
173        let metadata = if self.metadata_path.exists() {
174            let current = self.read_metadata()?;
175            migrate_metadata(current)?
176        } else {
177            CoordinationMetadata::latest()
178        };
179        write_atomic(&self.metadata_path, render_metadata(&metadata).as_bytes())
180    }
181
182    fn read_metadata(&self) -> ProviderResult<CoordinationMetadata> {
183        reject_symlink(&self.metadata_path)?;
184        let contents = fs::read_to_string(&self.metadata_path).map_err(|error| {
185            initialization(format!("coordination metadata read failed: {error}"))
186        })?;
187        parse_metadata(&contents)
188    }
189}
190
191impl CoordinationStoreProvider for FileCoordinationStore {
192    fn schema_version(&self) -> ProviderResult<u64> {
193        let _guard = self
194            .lock
195            .lock()
196            .map_err(|_| initialization("coordination store lock poisoned"))?;
197        Ok(self.read_metadata()?.version)
198    }
199
200    fn health(&self) -> ProviderResult<()> {
201        reject_symlink(&self.root)?;
202        let metadata = self.read_metadata()?;
203        validate_metadata(&metadata)
204    }
205}
206
207#[derive(Debug, Clone, PartialEq, Eq)]
208struct CoordinationMetadata {
209    version: u64,
210    tables: Vec<String>,
211}
212
213impl CoordinationMetadata {
214    fn latest() -> Self {
215        Self {
216            version: COORDINATION_SCHEMA_VERSION,
217            tables: COORDINATION_TABLES
218                .iter()
219                .map(|table| (*table).to_string())
220                .collect(),
221        }
222    }
223}
224
225fn migrate_metadata(mut metadata: CoordinationMetadata) -> ProviderResult<CoordinationMetadata> {
226    if metadata.version > COORDINATION_SCHEMA_VERSION {
227        return Err(ProviderError::InvalidConfiguration(format!(
228            "coordination schema {} is newer than supported schema {}",
229            metadata.version, COORDINATION_SCHEMA_VERSION
230        )));
231    }
232    if metadata.version == 0 {
233        return Err(ProviderError::InvalidConfiguration(
234            "coordination schema version must be positive".to_string(),
235        ));
236    }
237    metadata.version = COORDINATION_SCHEMA_VERSION;
238    metadata.tables = COORDINATION_TABLES
239        .iter()
240        .map(|table| (*table).to_string())
241        .collect();
242    Ok(metadata)
243}
244
245fn parse_metadata(contents: &str) -> ProviderResult<CoordinationMetadata> {
246    let mut format = None;
247    let mut version = None;
248    let mut tables = None;
249    for line in contents.lines().filter(|line| !line.trim().is_empty()) {
250        let (key, value) = line.split_once('=').ok_or_else(|| {
251            ProviderError::InvalidConfiguration("invalid coordination metadata".to_string())
252        })?;
253        match key {
254            "format" => format = Some(value),
255            "version" => {
256                version = Some(value.parse::<u64>().map_err(|_| {
257                    ProviderError::InvalidConfiguration(
258                        "invalid coordination schema version".to_string(),
259                    )
260                })?)
261            }
262            "tables" => {
263                tables = Some(value.split(',').map(str::to_string).collect::<Vec<_>>());
264            }
265            _ => {
266                return Err(ProviderError::InvalidConfiguration(format!(
267                    "unknown coordination metadata field: {key}"
268                )))
269            }
270        }
271    }
272    if format != Some(STORE_FORMAT) {
273        return Err(ProviderError::InvalidConfiguration(
274            "unsupported coordination metadata format".to_string(),
275        ));
276    }
277    Ok(CoordinationMetadata {
278        version: version.ok_or_else(|| {
279            ProviderError::InvalidConfiguration("missing coordination schema version".to_string())
280        })?,
281        tables: tables.ok_or_else(|| {
282            ProviderError::InvalidConfiguration("missing coordination table allowlist".to_string())
283        })?,
284    })
285}
286
287fn render_metadata(metadata: &CoordinationMetadata) -> String {
288    format!(
289        "format={STORE_FORMAT}\nversion={}\ntables={}\n",
290        metadata.version,
291        metadata.tables.join(",")
292    )
293}
294
295fn validate_metadata(metadata: &CoordinationMetadata) -> ProviderResult<()> {
296    if metadata.version > COORDINATION_SCHEMA_VERSION {
297        return Err(ProviderError::InvalidConfiguration(
298            "coordination schema is newer than this Runtime".to_string(),
299        ));
300    }
301    let expected = COORDINATION_TABLES
302        .iter()
303        .map(|table| (*table).to_string())
304        .collect::<Vec<_>>();
305    if metadata.tables != expected {
306        return Err(ProviderError::InvalidConfiguration(
307            "coordination table allowlist does not match schema V2".to_string(),
308        ));
309    }
310    Ok(())
311}
312
313fn prepare_root(root: &Path) -> ProviderResult<()> {
314    reject_symlink(root)?;
315    fs::create_dir_all(root)
316        .map_err(|error| initialization(format!("coordination directory failed: {error}")))?;
317    #[cfg(unix)]
318    {
319        use std::os::unix::fs::PermissionsExt;
320        fs::set_permissions(root, fs::Permissions::from_mode(0o700)).map_err(|error| {
321            initialization(format!(
322                "coordination directory permissions failed: {error}"
323            ))
324        })?;
325    }
326    Ok(())
327}
328
329fn reject_symlink(path: &Path) -> ProviderResult<()> {
330    match fs::symlink_metadata(path) {
331        Ok(metadata) if metadata.file_type().is_symlink() => {
332            Err(ProviderError::InvalidConfiguration(
333                "coordination path cannot be a symlink".to_string(),
334            ))
335        }
336        Ok(_) => Ok(()),
337        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
338        Err(error) => Err(initialization(format!(
339            "coordination path inspection failed: {error}"
340        ))),
341    }
342}
343
344fn write_atomic(path: &Path, bytes: &[u8]) -> ProviderResult<()> {
345    let parent = path.parent().unwrap_or_else(|| Path::new("."));
346    fs::create_dir_all(parent)
347        .map_err(|error| initialization(format!("coordination parent failed: {error}")))?;
348    let temp = parent.join(format!(
349        ".coordination.{}.{}.tmp",
350        std::process::id(),
351        TEMP_COUNTER.fetch_add(1, Ordering::Relaxed)
352    ));
353    let result = write_and_replace(&temp, path, parent, bytes);
354    if result.is_err() {
355        let _ = fs::remove_file(temp);
356    }
357    result
358}
359
360fn write_and_replace(temp: &Path, path: &Path, _parent: &Path, bytes: &[u8]) -> ProviderResult<()> {
361    let mut options = OpenOptions::new();
362    options.create_new(true).write(true);
363    #[cfg(unix)]
364    {
365        use std::os::unix::fs::OpenOptionsExt;
366        options.mode(0o600);
367    }
368    let mut file = options
369        .open(temp)
370        .map_err(|error| initialization(format!("coordination temp create failed: {error}")))?;
371    file.write_all(bytes)
372        .and_then(|_| file.sync_all())
373        .map_err(|error| initialization(format!("coordination metadata write failed: {error}")))?;
374    fs::rename(temp, path).map_err(|error| {
375        initialization(format!("coordination metadata replace failed: {error}"))
376    })?;
377    #[cfg(unix)]
378    fs::File::open(_parent)
379        .and_then(|directory| directory.sync_all())
380        .map_err(|error| initialization(format!("coordination directory sync failed: {error}")))?;
381    Ok(())
382}
383
384fn initialization(message: impl Into<String>) -> ProviderError {
385    ProviderError::Initialization(message.into())
386}
appcore_provider/coordination.rs

appcore_provider/
coordination.rs
