nodedb_lite/engine/array/
retention.rs

1//! Retention compaction for the Lite array engine.
2//!
3//! Delegates to `nodedb_array::query::retention::merge_for_retention` for the
4//! per-`hilbert_prefix` merge logic. Because Lite has no background TPC, this
5//! runs inline at the end of `array_put_cell` when `audit_retain_ms` is set.
6
7use std::collections::HashMap;
8use std::sync::Arc;
9
10use nodedb_array::query::retention::merge_for_retention;
11use nodedb_array::schema::ArraySchema;
12use nodedb_array::segment::TileEntry;
13use nodedb_array::{SegmentReader, SparseTile, TilePayload};
14
15use crate::engine::array::manifest::{ArrayManifest, SegmentRef, save_manifest};
16use crate::engine::array::segments::{delete_segment, write_segment};
17use crate::error::LiteError;
18use crate::storage::engine::StorageEngine;
19
20/// Run idle retention compaction for one array.
21///
22/// For each segment, groups tile entries by `hilbert_prefix`, calls
23/// `merge_for_retention`, and rewrites the segment when any tile-versions
24/// were merged or dropped. The manifest is updated atomically after all
25/// rewrites succeed.
26///
27/// Returns the number of segments rewritten.
28pub async fn run_retention<S: StorageEngine>(
29    storage: &Arc<S>,
30    name: &str,
31    manifest: &mut ArrayManifest,
32    schema: &ArraySchema,
33    schema_hash: u64,
34    audit_retain_ms: i64,
35    now_ms: i64,
36) -> Result<usize, LiteError> {
37    let horizon_ms = now_ms.saturating_sub(audit_retain_ms);
38    let mut rewritten = 0;
39
40    let mut new_segments: Vec<SegmentRef> = Vec::new();
41
42    let old_segs: Vec<SegmentRef> = manifest.segments.clone();
43
44    for seg_ref in &old_segs {
45        let bytes = crate::engine::array::segments::load_segment(storage, name, seg_ref.id).await?;
46        let reader = SegmentReader::open(&bytes).map_err(|e| LiteError::Storage {
47            detail: format!("open reader seg {}: {e}", seg_ref.id),
48        })?;
49
50        let tiles = reader.tiles();
51        if tiles.is_empty() {
52            new_segments.push(seg_ref.clone());
53            continue;
54        }
55
56        // Check whether any tile is outside the horizon.
57        let needs_merge = tiles.iter().any(|e| e.tile_id.system_from_ms < horizon_ms);
58
59        if !needs_merge {
60            new_segments.push(seg_ref.clone());
61            continue;
62        }
63
64        // Group tile entries by hilbert_prefix.
65        let mut by_prefix: HashMap<u64, Vec<TileEntry>> = HashMap::new();
66        for entry in tiles {
67            by_prefix
68                .entry(entry.tile_id.hilbert_prefix)
69                .or_default()
70                .push(entry.clone());
71        }
72
73        // Merge per-prefix and collect surviving tiles.
74        let mut keep_tiles: Vec<(nodedb_array::types::TileId, SparseTile)> = Vec::new();
75
76        for (&prefix, entries) in &by_prefix {
77            let result =
78                merge_for_retention(entries, &reader, schema, horizon_ms).map_err(|e| {
79                    LiteError::Storage {
80                        detail: format!("merge_for_retention: {e}"),
81                    }
82                })?;
83
84            // Ceiling tile: synthesised live state AS OF horizon for out-of-horizon cells.
85            if let Some(ceiling) = result.ceiling_tile {
86                // Place the ceiling at `horizon_ms` as a new tile-version.
87                let ceil_id = nodedb_array::types::TileId::new(prefix, horizon_ms);
88                keep_tiles.push((ceil_id, ceiling));
89            }
90
91            // Keep in-horizon tile versions unchanged.
92            for tile_id in &result.keep_inhorizon {
93                let payload = reader
94                    .read_tile_as_of(tile_id.hilbert_prefix, tile_id.system_from_ms, None)
95                    .map_err(|e| LiteError::Storage {
96                        detail: format!("read_tile_as_of: {e}"),
97                    })?;
98                if let Some(TilePayload::Sparse(tile)) = payload {
99                    keep_tiles.push((*tile_id, tile));
100                }
101            }
102        }
103
104        // Sort tiles by TileId (required by SegmentWriter: strictly ascending).
105        keep_tiles.sort_by_key(|(id, _)| *id);
106        keep_tiles.dedup_by_key(|(id, _)| *id);
107
108        // Delete old segment.
109        delete_segment(storage, name, seg_ref.id).await?;
110
111        if keep_tiles.is_empty() {
112            // Nothing survived — segment fully purged.
113            rewritten += 1;
114            continue;
115        }
116
117        let new_id = manifest.next_id;
118        manifest.next_id += 1;
119
120        let refs: Vec<_> = keep_tiles.iter().map(|(id, tile)| (*id, tile)).collect();
121        let new_bytes = write_segment(storage, name, new_id, schema_hash, &refs).await?;
122        new_segments.push(SegmentRef {
123            id: new_id,
124            byte_len: new_bytes.len() as u64,
125        });
126        rewritten += 1;
127    }
128
129    manifest.segments = new_segments;
130    save_manifest(storage, name, manifest).await?;
131
132    Ok(rewritten)
133}
134
135#[cfg(test)]
136mod tests {
137    use super::*;
138    use crate::engine::array::manifest::{ArrayManifest, save_manifest};
139    use crate::engine::array::segments::write_segment;
140    use crate::storage::pagedb_storage::PagedbStorageMem;
141    use nodedb_array::schema::ArraySchemaBuilder;
142    use nodedb_array::schema::attr_spec::{AttrSpec, AttrType};
143    use nodedb_array::schema::dim_spec::{DimSpec, DimType};
144    use nodedb_array::tile::sparse_tile::{SparseRow, SparseTileBuilder};
145    use nodedb_array::types::TileId;
146    use nodedb_array::types::cell_value::value::CellValue;
147    use nodedb_array::types::coord::value::CoordValue;
148    use nodedb_array::types::domain::{Domain, DomainBound};
149    use nodedb_types::{OPEN_UPPER, Surrogate};
150    use std::sync::Arc;
151
152    fn schema() -> ArraySchema {
153        ArraySchemaBuilder::new("r")
154            .dim(DimSpec::new(
155                "x",
156                DimType::Int64,
157                Domain::new(DomainBound::Int64(0), DomainBound::Int64(15)),
158            ))
159            .attr(AttrSpec::new("v", AttrType::Int64, true))
160            .tile_extents(vec![4])
161            .build()
162            .unwrap()
163    }
164
165    fn make_tile(s: &ArraySchema, x: i64, v: i64) -> SparseTile {
166        let mut b = SparseTileBuilder::new(s);
167        b.push_row(SparseRow::live(
168            &[CoordValue::Int64(x)],
169            &[CellValue::Int64(v)],
170            Surrogate::new(x as u32 + 1),
171            0,
172            OPEN_UPPER,
173        ))
174        .unwrap();
175        b.build()
176    }
177
178    #[tokio::test]
179    async fn empty_segment_passes_through() {
180        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
181        let s = schema();
182        let hash = crate::engine::array::catalog::hash_schema(&s).unwrap();
183        let mut manifest = ArrayManifest::new();
184
185        // Empty segment.
186        let writer = nodedb_array::SegmentWriter::new(hash);
187        let seg_bytes = writer.finish(None).unwrap();
188        let seg_id = manifest.push_segment(seg_bytes.len() as u64);
189        storage
190            .put(
191                nodedb_types::Namespace::Array,
192                &crate::engine::array::manifest::segment_key("r", seg_id),
193                &seg_bytes,
194            )
195            .await
196            .unwrap();
197        save_manifest(&storage, "r", &manifest).await.unwrap();
198
199        let n = run_retention(&storage, "r", &mut manifest, &s, hash, 60_000, 100_000)
200            .await
201            .unwrap();
202        assert_eq!(n, 0);
203        assert_eq!(manifest.segments.len(), 1);
204    }
205
206    #[tokio::test]
207    async fn in_horizon_segment_not_rewritten() {
208        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
209        let s = schema();
210        let hash = crate::engine::array::catalog::hash_schema(&s).unwrap();
211        let mut manifest = ArrayManifest::new();
212
213        // Tile with system_from_ms = 90_000 (inside horizon: now=100_000, retain=60_000 → horizon=40_000).
214        let tile = make_tile(&s, 1, 10);
215        let seg_id = manifest.push_segment(0);
216        write_segment(
217            &storage,
218            "r",
219            seg_id,
220            hash,
221            &[(TileId::new(0, 90_000), &tile)],
222        )
223        .await
224        .unwrap();
225        save_manifest(&storage, "r", &manifest).await.unwrap();
226
227        let n = run_retention(&storage, "r", &mut manifest, &s, hash, 60_000, 100_000)
228            .await
229            .unwrap();
230        assert_eq!(n, 0);
231    }
232
233    #[tokio::test]
234    async fn out_of_horizon_segment_gets_rewritten() {
235        let storage = Arc::new(PagedbStorageMem::open_in_memory().await.unwrap());
236        let s = schema();
237        let hash = crate::engine::array::catalog::hash_schema(&s).unwrap();
238        let mut manifest = ArrayManifest::new();
239
240        // Two tiles for same coord: out-of-horizon (sys=10) and in-horizon (sys=90_000).
241        // horizon = 100_000 - 60_000 = 40_000. sys=10 < 40_000 → outside.
242        let tile1 = make_tile(&s, 1, 10);
243        let tile2 = make_tile(&s, 1, 20);
244        let seg_id = manifest.push_segment(0);
245        write_segment(
246            &storage,
247            "r",
248            seg_id,
249            hash,
250            &[
251                (TileId::new(0, 10), &tile1),
252                (TileId::new(0, 90_000), &tile2),
253            ],
254        )
255        .await
256        .unwrap();
257        save_manifest(&storage, "r", &manifest).await.unwrap();
258
259        let n = run_retention(&storage, "r", &mut manifest, &s, hash, 60_000, 100_000)
260            .await
261            .unwrap();
262        assert_eq!(n, 1);
263    }
264}
nodedb_lite/engine/array/retention.rs

nodedb_lite/engine/array/
retention.rs
