loonfs_api/v0/
search.rs

1//! Content search (grep) request and response shapes: the `query/v0`
2//! plane's first operation (API spec, "Content search").
3
4use crate::{AbsolutePath, ChangeSeq, CheckpointId, InodeId, NamespaceId, RevisionNo};
5use serde::{Deserialize, Serialize};
6use xxhash_rust::xxh64::xxh64;
7
8/// One content-search request.
9///
10/// Every field but `pattern` is optional, and each one selects results. A
11/// misspelled `case_insensitive` or `path_prefix` would decode to the default
12/// and answer a different search than the caller asked for, with no sign that
13/// anything was dropped, so unknown fields are rejected.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
16#[serde(deny_unknown_fields)]
17pub struct GrepRequest {
18    /// The pattern, in the Rust `regex` crate's dialect (no backreferences
19    /// or lookaround). Patterns that require no literal bytes are rejected
20    /// with `query_unindexable` unless `allow_scan` is set.
21    pub pattern: String,
22    /// Match case-insensitively. Verification is exact; the index remains
23    /// consulted through its case-folded grams.
24    #[serde(default)]
25    pub case_insensitive: bool,
26    /// Restrict matches to files under this complete absolute path, resolved
27    /// to a directory inode before candidates are filtered.
28    #[serde(default, skip_serializing_if = "Option::is_none")]
29    pub path_prefix: Option<AbsolutePath>,
30    /// Resume cursor from a previous page. The cursor resumes strictly
31    /// after the last candidate the issuing page finished scanning and is
32    /// bound to that page's request; each page is evaluated against the
33    /// namespace head at page time.
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub cursor: Option<String>,
36    /// Maximum matches per page.
37    #[serde(default, skip_serializing_if = "Option::is_none")]
38    pub limit: Option<u32>,
39    /// When the unindexed tail exceeds the scan budget, return
40    /// indexed-only results (reported via `tail_scanned: false`) instead
41    /// of failing with `index_lagging`.
42    #[serde(default)]
43    pub allow_stale: bool,
44    /// Permit a capped exhaustive scan when the pattern yields no required
45    /// grams. Refused beyond the server's scan budget.
46    #[serde(default)]
47    pub allow_scan: bool,
48}
49
50impl GrepRequest {
51    /// Fingerprint of the fields that select results, binding cursors to
52    /// the request that issued them. Not a durable format: cursors are
53    /// opaque and short-lived, so this may change between builds.
54    pub fn fingerprint(&self) -> u64 {
55        let mut seed = xxh64(self.pattern.as_bytes(), 0);
56        seed = xxh64(
57            self.path_prefix
58                .as_ref()
59                .map(AbsolutePath::as_str)
60                .unwrap_or("")
61                .as_bytes(),
62            seed,
63        );
64        let flags = [
65            u8::from(self.case_insensitive),
66            u8::from(self.allow_stale),
67            u8::from(self.allow_scan),
68        ];
69        xxh64(&flags, seed)
70    }
71}
72
73/// One line-oriented match.
74#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
75#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
76pub struct GrepMatch {
77    /// The file's absolute path, derived at the snapshot.
78    pub path: AbsolutePath,
79    /// Durable identity of the matched file.
80    #[serde(with = "crate::public_inode_id")]
81    #[cfg_attr(
82        feature = "openapi",
83        schema(schema_with = crate::public_inode_id::schema)
84    )]
85    pub inode_id: InodeId,
86    /// The matched revision (the newest visible one at the snapshot).
87    pub revision_no: RevisionNo,
88    /// One-based line number of the match.
89    pub line_number: u64,
90    /// Byte offset of the match within the file.
91    pub byte_offset: u64,
92    /// The matching line, truncated to the server's line cap.
93    pub line: String,
94    /// True when `line` was truncated.
95    #[serde(default)]
96    pub line_truncated: bool,
97}
98
99/// One content-search page.
100#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
101#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
102pub struct GrepResponse {
103    /// Namespace searched.
104    pub namespace_id: NamespaceId,
105    /// Sequence this page was evaluated at. Pages are evaluated against
106    /// the namespace head at page time; the cursor is an ordering resume,
107    /// not a snapshot pin.
108    pub head_seq: ChangeSeq,
109    /// Commits at or below this sequence were answered from the index.
110    pub built_through_seq: ChangeSeq,
111    /// True when revisions after `built_through_seq` were scanned
112    /// exhaustively; false only when `allow_stale` skipped them.
113    pub tail_scanned: bool,
114    /// Matches in ascending `(inode_id, byte_offset)` order. A page may
115    /// return fewer matches than its limit and still carry a cursor: the
116    /// per-page verified-candidate budget bounds how much content one
117    /// request reads, whatever the plan's false-positive rate.
118    pub matches: Vec<GrepMatch>,
119    /// Present when another page follows.
120    #[serde(default, skip_serializing_if = "Option::is_none")]
121    pub next_cursor: Option<String>,
122}
123
124/// Where a namespace's grep index is in its lifecycle.
125///
126/// Each status contains only the fields valid for that lifecycle state.
127/// `Backfilling` reports its target and current position. `Active` reports
128/// how far the index has been built. Clients should treat a namespace as
129/// searchable only when the index is `Active`.
130#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
131#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
132#[serde(tag = "status", rename_all = "snake_case")]
133pub enum GrepIndexLifecycle {
134    /// No index is maintained for this namespace.
135    Disabled,
136    /// The initial walk over a pinned checkpoint is running. Nothing is
137    /// searchable yet.
138    Backfilling {
139        /// Namespace sequence the pinned checkpoint captured. Reaching it
140        /// is what completes the backfill.
141        target_seq: ChangeSeq,
142        /// Inode the walk resumes strictly after. Absent before the first
143        /// page.
144        #[serde(
145            default,
146            skip_serializing_if = "Option::is_none",
147            with = "crate::public_inode_id::option"
148        )]
149        #[cfg_attr(
150            feature = "openapi",
151            schema(schema_with = crate::public_inode_id::optional_schema)
152        )]
153        cursor_inode_id: Option<InodeId>,
154        /// Checkpoint pinning the state being walked.
155        checkpoint_id: CheckpointId,
156    },
157    /// The index follows the change feed. Commits at or below the watermark
158    /// are searchable.
159    Active {
160        /// Sequence of the commit at the index cursor.
161        built_through_seq: ChangeSeq,
162        /// Offset of the next change event within `built_through_seq`, or
163        /// zero when the whole commit is represented.
164        #[serde(default, skip_serializing_if = "is_zero")]
165        next_event_index: u32,
166    },
167}
168
169impl GrepIndexLifecycle {
170    /// Whether every commit at or below `target_seq` is represented.
171    ///
172    /// A watermark inside a commit (`next_event_index` above zero) has that
173    /// commit only partly indexed, so it counts as reached only for earlier
174    /// sequences.
175    pub fn is_built_through(&self, target_seq: ChangeSeq) -> bool {
176        match self {
177            Self::Disabled | Self::Backfilling { .. } => false,
178            Self::Active {
179                built_through_seq,
180                next_event_index,
181            } => {
182                *built_through_seq > target_seq
183                    || (*built_through_seq == target_seq && *next_event_index == 0)
184            }
185        }
186    }
187}
188
189fn is_zero(value: &u32) -> bool {
190    *value == 0
191}
192
193/// The namespace's grep-index lifecycle and its cheap bookkeeping (admin
194/// plane).
195#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
196#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
197pub struct GrepIndexStatusResponse {
198    /// Namespace the status describes.
199    pub namespace_id: NamespaceId,
200    /// Where the index is in its lifecycle.
201    #[serde(flatten)]
202    pub lifecycle: GrepIndexLifecycle,
203    /// Next logical run ordinal the index will allocate.
204    #[cfg_attr(feature = "openapi", schema(maximum = 9007199254740991_u64))]
205    pub next_run_ordinal: u64,
206    /// True while a partitioned segment reorganization is in progress.
207    pub reorganize_pending: bool,
208}
209
210/// One explicit grep-index garbage-collection pass (admin plane).
211#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
212#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
213#[serde(deny_unknown_fields)]
214pub struct GrepGcRequest {
215    /// Reads this pass may spend before returning with a `next_cursor`.
216    /// Omit to take the same per-pass default the runtime's own collection
217    /// takes.
218    #[serde(default, skip_serializing_if = "Option::is_none")]
219    pub max_objects: Option<u64>,
220    /// Opaque resume token returned as `next_cursor` by an earlier pass
221    /// against the same namespace.
222    #[serde(default, skip_serializing_if = "Option::is_none")]
223    pub cursor: Option<String>,
224}
225
226/// Result of one explicit grep-index garbage-collection pass (admin plane).
227#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
228#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
229pub struct GrepGcResponse {
230    /// Namespace whose grep-owned keyspace was inspected.
231    pub namespace_id: NamespaceId,
232    /// Unreferenced grep segments deleted after the grace window.
233    pub deleted_segments: u64,
234    /// Other unreferenced grep objects deleted after the grace window.
235    pub deleted_other_objects: u64,
236    /// Whether an absent or tombstoned namespace had extension state reaped.
237    pub namespace_reaped: bool,
238    /// Young or concurrently revived candidates retained by the pass.
239    pub retained_candidates: u64,
240    /// Whether unreadable namespace or grep state forced conservative retention.
241    pub namespace_degraded: bool,
242    /// Present when the budget stopped the pass with keys left to examine.
243    #[serde(default, skip_serializing_if = "Option::is_none")]
244    pub next_cursor: Option<String>,
245}
246
247#[cfg(test)]
248mod tests {
249    use super::*;
250
251    #[test]
252    fn grep_paths_keep_the_plain_string_wire_shape() {
253        let request = GrepRequest {
254            pattern: "needle".to_owned(),
255            case_insensitive: false,
256            path_prefix: Some(AbsolutePath::parse("/docs").expect("path prefix")),
257            cursor: None,
258            limit: None,
259            allow_stale: false,
260            allow_scan: false,
261        };
262        assert_eq!(
263            serde_json::to_value(request).expect("serialize grep request"),
264            serde_json::json!({
265                "pattern": "needle",
266                "case_insensitive": false,
267                "path_prefix": "/docs",
268                "allow_stale": false,
269                "allow_scan": false
270            })
271        );
272
273        let found = GrepMatch {
274            path: AbsolutePath::parse("/docs/a.txt").expect("match path"),
275            inode_id: InodeId(2),
276            revision_no: RevisionNo(3),
277            line_number: 4,
278            byte_offset: 5,
279            line: "needle".to_owned(),
280            line_truncated: false,
281        };
282        assert_eq!(
283            serde_json::to_value(found).expect("serialize grep match"),
284            serde_json::json!({
285                "path": "/docs/a.txt",
286                "inode_id": "ino_2",
287                "revision_no": 3,
288                "line_number": 4,
289                "byte_offset": 5,
290                "line": "needle",
291                "line_truncated": false
292            })
293        );
294    }
295
296    #[test]
297    fn lifecycle_statuses_never_share_a_sequence_field() {
298        let backfilling = GrepIndexLifecycle::Backfilling {
299            target_seq: ChangeSeq(9),
300            cursor_inode_id: Some(InodeId(4)),
301            checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000009")
302                .expect("checkpoint id"),
303        };
304        assert_eq!(
305            serde_json::to_value(&backfilling).expect("serialize backfilling"),
306            serde_json::json!({
307                "status": "backfilling",
308                "target_seq": 9,
309                "cursor_inode_id": "ino_4",
310                "checkpoint_id": "chk_00000000000000000000000000000009"
311            }),
312            "a backfill reports its target and its walk, never a watermark"
313        );
314
315        assert_eq!(
316            serde_json::to_value(GrepIndexLifecycle::Active {
317                built_through_seq: ChangeSeq(9),
318                next_event_index: 0,
319            })
320            .expect("serialize active"),
321            serde_json::json!({"status": "active", "built_through_seq": 9}),
322            "an active index reports its watermark and no target"
323        );
324
325        assert_eq!(
326            serde_json::to_value(GrepIndexLifecycle::Disabled).expect("serialize disabled"),
327            serde_json::json!({"status": "disabled"})
328        );
329    }
330
331    #[test]
332    fn only_an_active_index_has_built_through_a_sequence() {
333        let backfilling = GrepIndexLifecycle::Backfilling {
334            target_seq: ChangeSeq(9),
335            cursor_inode_id: None,
336            checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000009")
337                .expect("checkpoint id"),
338        };
339        assert!(
340            !backfilling.is_built_through(ChangeSeq(0)),
341            "a backfill has indexed nothing until it turns active"
342        );
343        assert!(!GrepIndexLifecycle::Disabled.is_built_through(ChangeSeq(0)));
344
345        let active = |built_through_seq, next_event_index| GrepIndexLifecycle::Active {
346            built_through_seq,
347            next_event_index,
348        };
349        assert!(active(ChangeSeq(9), 0).is_built_through(ChangeSeq(9)));
350        assert!(active(ChangeSeq(9), 0).is_built_through(ChangeSeq(8)));
351        assert!(!active(ChangeSeq(9), 0).is_built_through(ChangeSeq(10)));
352        // A watermark inside a commit leaves the rest of that commit
353        // unindexed, so only earlier sequences count as reached.
354        assert!(!active(ChangeSeq(9), 3).is_built_through(ChangeSeq(9)));
355        assert!(active(ChangeSeq(9), 3).is_built_through(ChangeSeq(8)));
356    }
357
358    #[test]
359    fn grep_index_status_flattens_active_lifecycle() {
360        let response = GrepIndexStatusResponse {
361            namespace_id: NamespaceId::parse("demo").expect("namespace id"),
362            lifecycle: GrepIndexLifecycle::Active {
363                built_through_seq: ChangeSeq(12),
364                next_event_index: 0,
365            },
366            next_run_ordinal: 3,
367            reorganize_pending: false,
368        };
369
370        assert_eq!(
371            serde_json::to_string(&response).expect("serialize active status"),
372            r#"{"namespace_id":"demo","status":"active","built_through_seq":12,"next_run_ordinal":3,"reorganize_pending":false}"#
373        );
374    }
375
376    #[test]
377    fn grep_index_status_flattens_backfilling_lifecycle() {
378        let response = GrepIndexStatusResponse {
379            namespace_id: NamespaceId::parse("demo").expect("namespace id"),
380            lifecycle: GrepIndexLifecycle::Backfilling {
381                target_seq: ChangeSeq(12),
382                cursor_inode_id: Some(InodeId(4)),
383                checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000009")
384                    .expect("checkpoint id"),
385            },
386            next_run_ordinal: 1,
387            reorganize_pending: false,
388        };
389
390        assert_eq!(
391            serde_json::to_string(&response).expect("serialize backfilling status"),
392            r#"{"namespace_id":"demo","status":"backfilling","target_seq":12,"cursor_inode_id":"ino_4","checkpoint_id":"chk_00000000000000000000000000000009","next_run_ordinal":1,"reorganize_pending":false}"#
393        );
394    }
395
396    #[test]
397    fn grep_path_prefix_validates_during_deserialization() {
398        let encoded = serde_json::json!({
399            "pattern": "needle",
400            "path_prefix": "relative/path"
401        });
402
403        assert!(serde_json::from_value::<GrepRequest>(encoded).is_err());
404    }
405
406    /// Every optional field on a search body selects results, so a typo would
407    /// answer a different question than the caller asked and say nothing
408    /// about it.
409    #[test]
410    fn search_request_bodies_reject_unknown_fields() {
411        serde_json::from_value::<GrepRequest>(serde_json::json!({
412            "pattern": "needle",
413            "case_insensitive": true,
414            "path_prefix": "/docs",
415            "limit": 10,
416            "allow_stale": true,
417            "allow_scan": true
418        }))
419        .expect("the same body without a typo decodes");
420
421        for body in [
422            serde_json::json!({"pattern": "needle", "case_insensitve": true}),
423            serde_json::json!({"pattern": "needle", "caseInsensitive": true}),
424            serde_json::json!({"pattern": "needle", "pathPrefix": "/docs"}),
425            serde_json::json!({"pattern": "needle", "allow_scans": true}),
426        ] {
427            assert!(
428                serde_json::from_value::<GrepRequest>(body.clone()).is_err(),
429                "an unknown field decoded instead of failing the search: {body}"
430            );
431        }
432
433        serde_json::from_value::<GrepGcRequest>(serde_json::json!({"max_objects": 8}))
434            .expect("the same collection body without a typo decodes");
435        assert!(
436            serde_json::from_value::<GrepGcRequest>(serde_json::json!({"maxObjects": 8})).is_err()
437        );
438    }
439}
loonfs_api/v0/search.rs

loonfs_api/v0/
search.rs
