nodedb_lite/engine/array/ops/
compact.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! `ArrayOp::Compact` handler for NodeDB-Lite.
4//!
5//! Delegates to `crate::engine::array::retention::run_retention` to merge
6//! out-of-horizon tile-versions per segment and rewrite the manifest.
7//! When `audit_retain_ms` is `None` the array has no retention policy and
8//! compact is a no-op (returns `rows_affected = 0`).
9
10use std::sync::Arc;
11
12use nodedb_types::result::QueryResult;
13
14use crate::engine::array::engine::ArrayEngineState;
15use crate::error::LiteError;
16use crate::runtime::now_millis_i64;
17use crate::storage::engine::StorageEngine;
18
19/// Execute `ArrayOp::Compact` for the Lite engine.
20///
21/// If `audit_retain_ms` is `Some`, runs retention merge across every segment
22/// in the manifest and updates the manifest. `rows_affected` is set to the
23/// number of segments rewritten.
24///
25/// If `audit_retain_ms` is `None`, no merge is needed and the call returns
26/// immediately with `rows_affected = 0`.
27pub async fn compact<S: StorageEngine>(
28    array_state: &Arc<tokio::sync::Mutex<ArrayEngineState>>,
29    storage: &Arc<S>,
30    name: &str,
31    audit_retain_ms: Option<i64>,
32) -> Result<QueryResult, LiteError> {
33    let retain_ms = match audit_retain_ms {
34        Some(r) => r,
35        None => {
36            return Ok(QueryResult {
37                columns: vec!["segments_rewritten".to_string()],
38                rows: vec![vec![nodedb_types::value::Value::Integer(0)]],
39                rows_affected: 0,
40                command: None,
41            });
42        }
43    };
44
45    let now_ms = now_millis_i64();
46
47    let rewritten = {
48        let mut state = array_state.lock().await;
49        let arr = state
50            .arrays
51            .get_mut(name)
52            .ok_or_else(|| LiteError::BadRequest {
53                detail: format!("array '{name}' not found"),
54            })?;
55        let schema = arr.schema.clone();
56        let schema_hash = arr.schema_hash;
57        // Drop the lock before the async call.
58        let (schema, schema_hash, manifest_snapshot) = (schema, schema_hash, arr.manifest.clone());
59        drop(state);
60        let mut manifest = manifest_snapshot;
61        let n = crate::engine::array::retention::run_retention(
62            storage,
63            name,
64            &mut manifest,
65            &schema,
66            schema_hash,
67            retain_ms,
68            now_ms,
69        )
70        .await?;
71        // Write the updated manifest back.
72        let mut state = array_state.lock().await;
73        if let Some(arr) = state.arrays.get_mut(name) {
74            arr.manifest = manifest;
75        }
76        n
77    };
78
79    Ok(QueryResult {
80        columns: vec!["segments_rewritten".to_string()],
81        rows: vec![vec![nodedb_types::value::Value::Integer(rewritten as i64)]],
82        rows_affected: rewritten as u64,
83        command: None,
84    })
85}
nodedb_lite/engine/array/ops/compact.rs

nodedb_lite/engine/array/ops/
compact.rs
