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

loonfs_api/v0/
search.rs
