loonfs-api 0.2.0

Wire types and durable-format codecs for LoonFS.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
//! Pagination: page-size policy, typed page envelopes, and the opaque
//! cursors each paginated endpoint round-trips.

use crate::capability::{LIMIT_PAGINATION_DEFAULT, LIMIT_PAGINATION_MAX};
use crate::{ChangeSeq, InodeId, NameKey, NamespaceId, RevisionNo};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::num::NonZeroU32;
use thiserror::Error;

/// Default page size for endpoints that can return unbounded result sets.
pub const DEFAULT_PAGE_LIMIT: u32 = 1_000;
/// Default maximum accepted page size.
pub const DEFAULT_MAX_PAGE_LIMIT: u32 = 1_000;

/// Wire cursor format version.
pub const PAGE_CURSOR_VERSION: u8 = 1;

/// A validated page size selected from a caller request and a policy.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub struct EffectiveLimit(NonZeroU32);

impl EffectiveLimit {
    /// Creates an effective limit from a non-zero value.
    pub fn new(value: NonZeroU32) -> Self {
        Self(value)
    }

    /// Returns the numeric page size.
    pub fn get(self) -> u32 {
        self.0.get()
    }

    /// Returns the page size as a `usize` for vector reservations and counters.
    pub fn as_usize(self) -> usize {
        self.0.get() as usize
    }

    /// Returns the number of items an engine should try to read to detect a next page.
    pub fn limit_plus_one(self) -> usize {
        self.as_usize().saturating_add(1)
    }
}

/// Deployment or namespace policy for paginated endpoints.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PaginationPolicy {
    default_limit: NonZeroU32,
    max_limit: NonZeroU32,
}

impl PaginationPolicy {
    /// Creates a policy, requiring the default to be no larger than the max.
    pub fn new(
        default_limit: NonZeroU32,
        max_limit: NonZeroU32,
    ) -> Result<Self, PaginationPolicyError> {
        if default_limit > max_limit {
            return Err(PaginationPolicyError::DefaultExceedsMax {
                default_limit: default_limit.get(),
                max_limit: max_limit.get(),
            });
        }
        Ok(Self {
            default_limit,
            max_limit,
        })
    }

    /// Creates a policy from raw integers.
    pub fn from_values(default_limit: u32, max_limit: u32) -> Result<Self, PaginationPolicyError> {
        let default_limit =
            NonZeroU32::new(default_limit).ok_or(PaginationPolicyError::ZeroDefaultLimit)?;
        let max_limit = NonZeroU32::new(max_limit).ok_or(PaginationPolicyError::ZeroMaxLimit)?;
        Self::new(default_limit, max_limit)
    }

    /// Returns the page size applied when callers omit `limit`.
    pub fn default_limit(self) -> NonZeroU32 {
        self.default_limit
    }

    /// Returns the largest accepted caller-supplied `limit`.
    pub fn max_limit(self) -> NonZeroU32 {
        self.max_limit
    }

    /// Resolves a caller-supplied limit into the enforced page size.
    pub fn resolve_limit(self, requested: Option<u32>) -> Result<EffectiveLimit, LimitError> {
        match requested {
            None => Ok(EffectiveLimit(self.default_limit)),
            Some(0) => Err(LimitError::Zero),
            Some(value) if value > self.max_limit.get() => Err(LimitError::ExceedsMax {
                requested: value,
                max_limit: self.max_limit.get(),
            }),
            Some(value) => NonZeroU32::new(value)
                .map(EffectiveLimit)
                .ok_or(LimitError::Zero),
        }
    }

    /// Returns the advisory capability-document limits for this policy.
    pub fn capability_limits(self) -> BTreeMap<String, u64> {
        BTreeMap::from([
            (
                LIMIT_PAGINATION_DEFAULT.to_owned(),
                u64::from(self.default_limit.get()),
            ),
            (
                LIMIT_PAGINATION_MAX.to_owned(),
                u64::from(self.max_limit.get()),
            ),
        ])
    }
}

impl Default for PaginationPolicy {
    fn default() -> Self {
        let default_limit = const { NonZeroU32::new(DEFAULT_PAGE_LIMIT).unwrap() };
        let max_limit = const { NonZeroU32::new(DEFAULT_MAX_PAGE_LIMIT).unwrap() };
        Self {
            default_limit,
            max_limit,
        }
    }
}

/// Invalid pagination policy configuration.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum PaginationPolicyError {
    /// The configured default limit was zero.
    #[error("pagination default limit must be greater than zero")]
    ZeroDefaultLimit,
    /// The configured max limit was zero.
    #[error("pagination max limit must be greater than zero")]
    ZeroMaxLimit,
    /// The configured default limit exceeded the configured max.
    #[error("pagination default limit `{default_limit}` exceeds max limit `{max_limit}`")]
    DefaultExceedsMax {
        /// Default page size rejected by policy construction.
        default_limit: u32,
        /// Maximum page size the rejected default exceeded.
        max_limit: u32,
    },
}

/// Invalid caller-supplied page size.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum LimitError {
    /// The caller supplied `limit=0`.
    #[error("limit must be greater than zero")]
    Zero,
    /// The caller supplied a limit larger than the active policy allows.
    #[error("limit `{requested}` exceeds max limit `{max_limit}`")]
    ExceedsMax {
        /// Page size supplied by the caller.
        requested: u32,
        /// Largest page size allowed by the active policy.
        max_limit: u32,
    },
}

/// Typed request envelope for internal runtime/core page methods.
///
/// This is not a direct wire response type. HTTP handlers parse public query
/// fields into this shape after validating `limit` and decoding `cursor`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PageRequest<C> {
    /// Enforced page size.
    pub limit: EffectiveLimit,
    /// Optional decoded endpoint cursor.
    pub cursor: Option<C>,
}

/// Typed result envelope for internal runtime/core page methods.
///
/// This is not a direct wire response type. HTTP handlers encode
/// `next_cursor` into the public response envelope.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Page<T, C> {
    /// Returned items.
    pub items: Vec<T>,
    /// Cursor for the next page, if another page is available.
    pub next_cursor: Option<C>,
}

/// Cursor for one directory listing position.
///
/// Directory pagination advances in canonical `name_key` order. The cursor
/// is an ordering resume, not a snapshot pin: any head at or past `head_seq`
/// serves the next page, resuming strictly after `last_name_key` — the same
/// forward-only drift grep cursors tolerate.
///
/// The cursor intentionally contains only the minting head (`head_seq`),
/// listed directory identity (`directory_inode_id`), and resume position
/// (`last_name_key`). HTTP clients must pass the URL namespace and `path` on
/// every page. Runtime/server code resolves that path at the current head
/// and rejects the cursor unless it names `directory_inode_id`.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct DirectoryPageCursor {
    /// Head sequence the issuing page was evaluated at.
    pub head_seq: ChangeSeq,
    /// Directory inode resolved at `head_seq`.
    // The wire field is frozen as `dir_inode_id` in page cursor version 1.
    #[serde(rename = "dir_inode_id")]
    pub directory_inode_id: InodeId,
    /// Last canonical name key returned to the client.
    pub last_name_key: NameKey,
}

impl PageCursor for DirectoryPageCursor {
    const KIND: &'static str = "directory";
}

/// Cursor for one file revision listing position.
///
/// Revision pagination advances in newest-first revision order for one file
/// inode. Like directory and grep cursors, it is an ordering resume that
/// tolerates forward head drift; it includes the minting head plus the last
/// returned row's complete ordering identity so ties stay unambiguous.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct FileRevisionsPageCursor {
    /// Head sequence the issuing page was evaluated at.
    pub head_seq: ChangeSeq,
    /// File inode whose revisions are being listed.
    pub inode_id: InodeId,
    /// Last revision number returned to the client.
    pub last_revision_no: RevisionNo,
    /// Namespace sequence that created the last returned revision.
    pub last_committed_seq: ChangeSeq,
    /// WAL delta index that created the last returned revision.
    pub last_revision_delta_index: u32,
}

impl PageCursor for FileRevisionsPageCursor {
    const KIND: &'static str = "file_revisions";
}

/// Cursor for one trash listing position.
///
/// Trash pagination advances oldest deletion first, in ascending
/// `(deleted_at_seq, root_inode_id)` order — the order the derived
/// active-deletion family is keyed in. Like every cursor, it is an ordering
/// resume tolerating forward head drift: the next page evaluates at whatever
/// head is loaded and continues strictly after the deletion generation named
/// here.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TrashPageCursor {
    /// Head sequence the issuing page was evaluated at.
    pub head_seq: ChangeSeq,
    /// Commit sequence of the deletion the previous page ended on.
    pub last_deleted_at_seq: ChangeSeq,
    /// Deleted root inode the previous page ended on.
    pub last_root_inode_id: InodeId,
}

impl PageCursor for TrashPageCursor {
    const KIND: &'static str = "trash";
}

/// Cursor for one content-search (grep) snapshot.
///
/// Matches advance in ascending `(inode_id, byte_offset)` order: candidate
/// files by durable inode identity, match positions within a file by byte
/// offset. The cursor resumes strictly after the last returned match.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct GrepPageCursor {
    /// Sequence the issuing page was evaluated at.
    pub head_seq: ChangeSeq,
    /// Inode of the last candidate the issuing page finished scanning.
    pub last_inode_id: InodeId,
    /// Byte offset of the last returned match within that file, or
    /// `u64::MAX` when the file was fully scanned (budget stops and
    /// matchless candidates resume at the next inode).
    pub last_byte_offset: u64,
    /// Fingerprint of the request (pattern, flags, scope) that issued the
    /// cursor; a cursor replayed under a different request is rejected
    /// instead of silently skipping results.
    pub fingerprint: u64,
}

impl PageCursor for GrepPageCursor {
    const KIND: &'static str = "grep";
}

/// One paginated endpoint's cursor.
///
/// Cursors are opaque to clients: hex-encoded JSON carrying the endpoint's
/// [`KIND`](Self::KIND) and the format version, so a cursor replayed against
/// the wrong endpoint or an older build is rejected rather than misread.
pub trait PageCursor: Serialize + serde::de::DeserializeOwned {
    /// Frozen endpoint discriminator written into the encoded cursor.
    const KIND: &'static str;
}

#[derive(Serialize, Deserialize)]
struct CursorEnvelope<C> {
    // The wire field is frozen as `v` in page cursor version 1.
    #[serde(rename = "v")]
    version: u8,
    kind: String,
    #[serde(flatten)]
    cursor: C,
}

/// Encodes a cursor as the opaque string clients round-trip.
pub fn encode_cursor<C: PageCursor>(cursor: &C) -> Result<String, PageCursorError> {
    let bytes = serde_json::to_vec(&CursorEnvelope {
        version: PAGE_CURSOR_VERSION,
        kind: C::KIND.to_owned(),
        cursor,
    })
    .map_err(|error| PageCursorError::InvalidJson(error.to_string()))?;
    Ok(crate::hex::hex_encode_bytes(&bytes))
}

/// Version and endpoint, read before the body so a cursor from another
/// endpoint reports `WrongKind` rather than a missing-field decode error.
#[derive(Deserialize)]
struct CursorHeader {
    #[serde(rename = "v")]
    version: u8,
    kind: String,
}

/// Decodes a cursor issued by [`encode_cursor`] for the same endpoint.
pub fn decode_cursor<C: PageCursor>(value: &str) -> Result<C, PageCursorError> {
    let bytes =
        crate::hex::hex_decode_bytes(value).map_err(|_| PageCursorError::InvalidEncoding)?;
    let header: CursorHeader = serde_json::from_slice(&bytes)
        .map_err(|error| PageCursorError::InvalidJson(error.to_string()))?;
    if header.version != PAGE_CURSOR_VERSION {
        return Err(PageCursorError::UnsupportedVersion {
            expected: PAGE_CURSOR_VERSION,
            actual: header.version,
        });
    }
    if header.kind != C::KIND {
        return Err(PageCursorError::WrongKind {
            expected: C::KIND,
            actual: header.kind,
        });
    }
    let envelope: CursorEnvelope<C> = serde_json::from_slice(&bytes)
        .map_err(|error| PageCursorError::InvalidJson(error.to_string()))?;
    Ok(envelope.cursor)
}

/// A cursor that resumes an enumeration of one namespace's own keyspace.
///
/// Maintenance passes walk keys rather than rows, and their cursors are
/// enumeration shortcuts and nothing else: a pass re-reads whatever
/// authorizes the work it does, whatever position it resumed from, so a
/// cursor that is lost or refused costs a repeated walk and never a wrong
/// decision. What the binding buys is that a token minted for another
/// namespace, another job, or another key family is refused instead of
/// quietly skipping the keys between here and wherever it points.
pub trait NamespaceCursor: PageCursor {
    /// Namespace whose keyspace this cursor walks.
    fn namespace_id(&self) -> &NamespaceId;

    /// Key the enumeration stopped at, or `None` at the start.
    fn last_key(&self) -> Option<&str>;

    /// Prefix every key this cursor may name lies under.
    fn key_prefix(&self) -> String;
}

/// Decodes a cursor issued for `expected_namespace_id`'s own keyspace.
pub fn decode_namespace_cursor<C: NamespaceCursor>(
    token: &str,
    expected_namespace_id: &NamespaceId,
) -> Result<C, NamespaceCursorError> {
    let cursor: C = decode_cursor(token)?;
    if cursor.namespace_id() != expected_namespace_id {
        return Err(NamespaceCursorError::ForeignNamespace);
    }
    let prefix = cursor.key_prefix();
    if cursor
        .last_key()
        .is_some_and(|key| !key.starts_with(&prefix))
    {
        return Err(NamespaceCursorError::OutsideKeyspace);
    }
    Ok(cursor)
}

/// Why a namespace-bound cursor cannot resume the enumeration replaying it.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum NamespaceCursorError {
    /// Not a cursor this enumeration issued: unreadable, or from another
    /// endpoint, job, or cursor version.
    #[error(transparent)]
    Malformed(#[from] PageCursorError),
    /// A cursor for a different namespace than the one replaying it.
    #[error("cursor belongs to a different namespace")]
    ForeignNamespace,
    /// A cursor naming a key outside the prefix its enumeration walks.
    #[error("cursor names a key outside the enumeration it resumes")]
    OutsideKeyspace,
}

/// Invalid opaque page cursor.
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum PageCursorError {
    /// The cursor was not hex-encoded JSON.
    #[error("invalid page cursor encoding")]
    InvalidEncoding,
    /// The cursor JSON did not match a supported cursor shape.
    #[error("invalid page cursor JSON: {0}")]
    InvalidJson(String),
    /// The cursor was valid, but for a different paginated endpoint.
    #[error("page cursor kind `{actual}` cannot be used as `{expected}` cursor")]
    WrongKind {
        /// Cursor kind accepted by the endpoint doing the decoding.
        expected: &'static str,
        /// Cursor kind recovered from the caller's opaque token.
        actual: String,
    },
    /// The cursor format version is not supported by this build.
    #[error("unsupported page cursor version `{actual}`; expected `{expected}`")]
    UnsupportedVersion {
        /// Cursor format version this build can decode.
        expected: u8,
        /// Version embedded in the caller's opaque token.
        actual: u8,
    },
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn default_policy_resolves_omitted_limit_to_default() {
        let policy = PaginationPolicy::default();
        let limit = policy.resolve_limit(None).expect("default limit");

        assert_eq!(limit.get(), DEFAULT_PAGE_LIMIT);
        assert_eq!(limit.limit_plus_one(), 1_001);
    }

    #[test]
    fn policy_rejects_invalid_limits() {
        let policy = PaginationPolicy::default();

        assert_eq!(policy.resolve_limit(Some(0)), Err(LimitError::Zero));
        assert_eq!(
            policy.resolve_limit(Some(DEFAULT_MAX_PAGE_LIMIT + 1)),
            Err(LimitError::ExceedsMax {
                requested: DEFAULT_MAX_PAGE_LIMIT + 1,
                max_limit: DEFAULT_MAX_PAGE_LIMIT,
            })
        );
    }

    #[test]
    fn policy_rejects_default_above_max() {
        assert_eq!(
            PaginationPolicy::from_values(10, 5),
            Err(PaginationPolicyError::DefaultExceedsMax {
                default_limit: 10,
                max_limit: 5,
            })
        );
    }

    #[test]
    fn policy_exports_capability_limits() {
        let limits = PaginationPolicy::default().capability_limits();

        assert_eq!(
            limits.get(LIMIT_PAGINATION_DEFAULT),
            Some(&u64::from(DEFAULT_PAGE_LIMIT))
        );
        assert_eq!(
            limits.get(LIMIT_PAGINATION_MAX),
            Some(&u64::from(DEFAULT_MAX_PAGE_LIMIT))
        );
    }

    #[test]
    fn directory_cursor_round_trips() {
        let cursor = DirectoryPageCursor {
            head_seq: ChangeSeq(11),
            directory_inode_id: InodeId(7),
            last_name_key: NameKey::parse("plan.md").expect("name key"),
        };

        let encoded = encode_cursor(&cursor).expect("encode cursor");
        let decoded: DirectoryPageCursor = decode_cursor(&encoded).expect("decode cursor");

        assert_eq!(decoded, cursor);
    }

    #[test]
    fn file_revisions_cursor_round_trips() {
        let cursor = FileRevisionsPageCursor {
            head_seq: ChangeSeq(11),
            inode_id: InodeId(7),
            last_revision_no: RevisionNo(5),
            last_committed_seq: ChangeSeq(10),
            last_revision_delta_index: 3,
        };

        let encoded = encode_cursor(&cursor).expect("encode cursor");
        let decoded: FileRevisionsPageCursor = decode_cursor(&encoded).expect("decode cursor");

        assert_eq!(decoded, cursor);
    }

    #[test]
    fn cursor_kind_must_match_decoder() {
        let cursor = FileRevisionsPageCursor {
            head_seq: ChangeSeq(11),
            inode_id: InodeId(7),
            last_revision_no: RevisionNo(5),
            last_committed_seq: ChangeSeq(10),
            last_revision_delta_index: 3,
        };
        let encoded = encode_cursor(&cursor).expect("encode cursor");

        assert_eq!(
            decode_cursor::<DirectoryPageCursor>(&encoded),
            Err(PageCursorError::WrongKind {
                expected: "directory",
                actual: "file_revisions".to_owned(),
            })
        );
    }

    #[test]
    fn malformed_cursor_is_invalid_encoding() {
        assert_eq!(
            decode_cursor::<DirectoryPageCursor>("not-hex"),
            Err(PageCursorError::InvalidEncoding)
        );
    }

    #[test]
    fn unsupported_cursor_version_is_rejected() {
        let bytes = serde_json::to_vec(&CursorEnvelope {
            version: PAGE_CURSOR_VERSION + 1,
            kind: DirectoryPageCursor::KIND.to_owned(),
            cursor: DirectoryPageCursor {
                head_seq: ChangeSeq(11),
                directory_inode_id: InodeId(7),
                last_name_key: NameKey::parse("plan.md").expect("name key"),
            },
        })
        .expect("encode cursor");
        let encoded = crate::hex::hex_encode_bytes(&bytes);

        assert_eq!(
            decode_cursor::<DirectoryPageCursor>(&encoded),
            Err(PageCursorError::UnsupportedVersion {
                expected: PAGE_CURSOR_VERSION,
                actual: PAGE_CURSOR_VERSION + 1,
            })
        );
    }
}