linkrs_storage/engine/graph_storage/
ttl.rs

1use crate::{PurgeTarget, TtlPurgeEntry};
2use linkrs_core::types::VertexId;
3use std::collections::HashMap;
4
5use linkrs_core::{StorageResult, Value};
6
7use super::context::GraphStorageContext;
8use super::GraphStorage;
9
10/// Rows deleted between flushes inside a purge. Bounds the working set of the
11/// delete loop instead of collecting every expired id up front, so a large
12/// table cannot spike memory. Per-schema deletions stay sequential.
13const PURGE_DELETE_BATCH: usize = 1024;
14
15/// TTL check over a raw property slice — the shared primitive for read
16/// paths that hold `VertexRecord`/`EdgeRecord` slices instead of a
17/// materialized `Vertex`/`Edge`.
18pub(crate) fn record_expired_from_slice(
19    ttl_duration: Option<i64>,
20    ttl_col: Option<&str>,
21    props: &[(std::sync::Arc<str>, Value)],
22) -> bool {
23    let (Some(duration), Some(col)) = (ttl_duration, ttl_col) else {
24        return false;
25    };
26    if duration <= 0 || col.is_empty() {
27        return false;
28    }
29    let map: HashMap<std::sync::Arc<str>, Value> = props.iter().cloned().collect();
30    linkrs_core::types::ttl::is_record_expired(Some(duration), Some(col), &map)
31}
32
33/// Delete one bounded batch of expired vertices in order, draining the batch.
34fn flush_vertex_batch(
35    ctx: &GraphStorageContext,
36    space: &str,
37    tag: &str,
38    batch: &mut Vec<VertexId>,
39) -> StorageResult<usize> {
40    let mut removed = 0;
41    for vid in batch.drain(..) {
42        super::writer::delete_vertex(ctx, space, tag, &vid)?;
43        removed += 1;
44    }
45    Ok(removed)
46}
47
48/// Delete one bounded batch of expired edges in order, draining the batch.
49fn flush_edge_batch(
50    ctx: &GraphStorageContext,
51    space: &str,
52    edge_type: &str,
53    batch: &mut Vec<(VertexId, VertexId, i64)>,
54) -> StorageResult<usize> {
55    let mut removed = 0;
56    for (src, dst, rank) in batch.drain(..) {
57        super::writer::delete_edge(ctx, space, &src, &dst, edge_type, rank)?;
58        removed += 1;
59    }
60    Ok(removed)
61}
62
63impl GraphStorage {
64    pub fn purge_expired_vertices(&self, space: &str, tag: &str) -> StorageResult<usize> {
65        let ctx: &GraphStorageContext = &self.ctx;
66        let tag_info = ctx.schema_manager().get_tag(space, tag)?.ok_or_else(|| {
67            linkrs_core::StorageError::not_found(format!("Tag {tag} not found in space {space}"))
68        })?;
69        if !linkrs_core::types::ttl::has_valid_ttl(
70            tag_info.ttl_duration,
71            tag_info.ttl_col.as_deref(),
72        ) {
73            return Ok(0);
74        }
75        let ts = ctx.get_read_timestamp();
76        let Some(records) = ctx.scan_vertices(tag_info.tag_id, ts) else {
77            return Ok(0);
78        };
79        let ttl_duration = tag_info.ttl_duration;
80        let ttl_col = tag_info.ttl_col.as_deref();
81        let mut removed = 0;
82        let mut batch: Vec<VertexId> = Vec::with_capacity(PURGE_DELETE_BATCH.min(records.len()));
83        for record in &records {
84            if record_expired_from_slice(ttl_duration, ttl_col, &record.properties) {
85                batch.push(record.vid);
86                if batch.len() >= PURGE_DELETE_BATCH {
87                    removed += flush_vertex_batch(ctx, space, tag, &mut batch)?;
88                }
89            }
90        }
91        removed += flush_vertex_batch(ctx, space, tag, &mut batch)?;
92        Ok(removed)
93    }
94
95    pub fn purge_expired_edges(&self, space: &str, edge_type: &str) -> StorageResult<usize> {
96        let ctx: &GraphStorageContext = &self.ctx;
97        let edge_info = ctx
98            .schema_manager()
99            .get_edge_type(space, edge_type)?
100            .ok_or_else(|| {
101                linkrs_core::StorageError::not_found(format!(
102                    "Edge type {edge_type} not found in space {space}"
103                ))
104            })?;
105        if !linkrs_core::types::ttl::has_valid_ttl(
106            edge_info.ttl_duration,
107            edge_info.ttl_col.as_deref(),
108        ) {
109            return Ok(0);
110        }
111        let mut cursor = super::cursor_impl::create_edge_cursor(
112            std::sync::Arc::new(ctx.clone()),
113            space,
114            &crate::cursor::ScanOptions {
115                edge_type: Some(edge_type.to_string()),
116                ..Default::default()
117            },
118        )?;
119        let mut removed = 0;
120        let mut batch: Vec<(VertexId, VertexId, i64)> = Vec::with_capacity(PURGE_DELETE_BATCH);
121        loop {
122            let chunk = cursor.next_batch(PURGE_DELETE_BATCH)?;
123            if chunk.is_empty() {
124                break;
125            }
126            for edge in &chunk {
127                if linkrs_core::types::ttl::is_record_expired(
128                    edge_info.ttl_duration,
129                    edge_info.ttl_col.as_deref(),
130                    edge.properties(),
131                ) {
132                    batch.push((edge.src, edge.dst, edge.ranking));
133                    if batch.len() >= PURGE_DELETE_BATCH {
134                        removed += flush_edge_batch(ctx, space, edge_type, &mut batch)?;
135                    }
136                }
137            }
138        }
139        removed += flush_edge_batch(ctx, space, edge_type, &mut batch)?;
140        Ok(removed)
141    }
142
143    /// Physical-deletion entry point shared by the `PURGE EXPIRED` statement
144    /// and maintenance callers.
145    ///
146    /// `target = None` scans every TTL-configured tag and edge type in the
147    /// space, returning one [`TtlPurgeEntry`] per such schema. `target =
148    /// Some(..)` scans only that schema, returning its single entry (a schema
149    /// that exists but has no TTL yields `removed = 0`; a schema that does not
150    /// exist is an error).
151    pub fn purge_expired(
152        &self,
153        space: &str,
154        target: Option<PurgeTarget<'_>>,
155    ) -> StorageResult<Vec<TtlPurgeEntry>> {
156        match target {
157            Some(PurgeTarget::Tag(name)) => {
158                let removed = self.purge_expired_vertices(space, name)?;
159                Ok(vec![TtlPurgeEntry {
160                    schema_name: name.to_string(),
161                    is_edge: false,
162                    removed,
163                }])
164            }
165            Some(PurgeTarget::Edge(name)) => {
166                let removed = self.purge_expired_edges(space, name)?;
167                Ok(vec![TtlPurgeEntry {
168                    schema_name: name.to_string(),
169                    is_edge: true,
170                    removed,
171                }])
172            }
173            None => {
174                let ctx: &GraphStorageContext = &self.ctx;
175                let mut entries = Vec::new();
176                for tag in ctx.schema_manager().list_tags(space)? {
177                    if linkrs_core::types::ttl::has_valid_ttl(
178                        tag.ttl_duration,
179                        tag.ttl_col.as_deref(),
180                    ) {
181                        let removed = self.purge_expired_vertices(space, &tag.tag_name)?;
182                        entries.push(TtlPurgeEntry {
183                            schema_name: tag.tag_name,
184                            is_edge: false,
185                            removed,
186                        });
187                    }
188                }
189                for edge in ctx.schema_manager().list_edge_types(space)? {
190                    if linkrs_core::types::ttl::has_valid_ttl(
191                        edge.ttl_duration,
192                        edge.ttl_col.as_deref(),
193                    ) {
194                        let removed = self.purge_expired_edges(space, &edge.edge_type_name)?;
195                        entries.push(TtlPurgeEntry {
196                            schema_name: edge.edge_type_name,
197                            is_edge: true,
198                            removed,
199                        });
200                    }
201                }
202                Ok(entries)
203            }
204        }
205    }
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211
212    #[test]
213    fn raw_helper_matches_record_helper() {
214        let props: Vec<(std::sync::Arc<str>, Value)> =
215            vec![(std::sync::Arc::<str>::from("ts"), Value::BigInt(100))];
216        assert!(record_expired_from_slice(Some(10), Some("ts"), &props));
217        assert!(!record_expired_from_slice(None, None, &props));
218        assert!(!record_expired_from_slice(
219            Some(10),
220            Some("missing"),
221            &props
222        ));
223    }
224
225    #[test]
226    fn edge_types_ttl_active_covers_three_states() {
227        use crate::{StorageReader, StorageSchemaOps};
228        use linkrs_core::types::{EdgeTypeInfo, PropertyDef, SpaceInfo};
229
230        let mut storage = GraphStorage::new().expect("create storage");
231
232        // Space with no TTL edge type at all.
233        let mut nottl = SpaceInfo::new("nottl".to_string());
234        StorageSchemaOps::create_space(&mut storage, &mut nottl).expect("create space");
235        let plain_only =
236            EdgeTypeInfo::new("PLAIN".to_string()).with_properties(vec![PropertyDef::new(
237                "w".into(),
238                linkrs_core::DataType::Int,
239            )]);
240        StorageSchemaOps::create_edge_type(&mut storage, "nottl", &plain_only)
241            .expect("create plain edge");
242        assert!(!storage
243            .edge_types_ttl_active("nottl", &[])
244            .expect("nottl all types"));
245        assert!(!storage
246            .edge_types_ttl_active("nottl", &["PLAIN".to_string()])
247            .expect("nottl specific type"));
248
249        // Space mixing a TTL edge type with a plain one.
250        let mut mixed = SpaceInfo::new("mixed".to_string());
251        StorageSchemaOps::create_space(&mut storage, &mut mixed).expect("create space");
252        StorageSchemaOps::create_edge_type(&mut storage, "mixed", &plain_only)
253            .expect("create plain edge");
254        let ttl_edge = EdgeTypeInfo::new("TTLED".to_string())
255            .with_properties(vec![PropertyDef::new(
256                "ts".into(),
257                linkrs_core::DataType::BigInt,
258            )])
259            .with_ttl(Some(100), Some("ts".to_string()));
260        StorageSchemaOps::create_edge_type(&mut storage, "mixed", &ttl_edge)
261            .expect("create ttl edge");
262
263        // Querying the specific non-TTL type stays inactive.
264        assert!(!storage
265            .edge_types_ttl_active("mixed", &["PLAIN".to_string()])
266            .expect("mixed plain type"));
267        // Partial activation: querying the TTL type is active.
268        assert!(storage
269            .edge_types_ttl_active("mixed", &["TTLED".to_string()])
270            .expect("mixed ttled type"));
271        // Empty means all edge types, which include a TTL one.
272        assert!(storage
273            .edge_types_ttl_active("mixed", &[])
274            .expect("mixed all types"));
275    }
276}
linkrs_storage/engine/graph_storage/ttl.rs

linkrs_storage/engine/graph_storage/
ttl.rs
