shardline_server_core/
lib.rs

1#![deny(unsafe_code)]
2#![allow(
3    clippy::missing_errors_doc,
4    clippy::missing_panics_doc,
5    clippy::missing_const_for_fn,
6    clippy::must_use_candidate
7)]
8#![cfg_attr(
9    test,
10    allow(
11        clippy::unwrap_used,
12        clippy::expect_used,
13        clippy::indexing_slicing,
14        clippy::arithmetic_side_effects,
15        clippy::shadow_unrelated,
16        clippy::let_underscore_must_use,
17        clippy::format_push_string
18    )
19)]
20
21//! Shared core types for the Shardline server ecosystem.
22//!
23//! This crate contains pure data structures and constants that are shared
24//! between the server crate and potential future crate extractions.
25
26use std::{
27    io::{Error as IoError, Read},
28    num::{NonZeroUsize, TryFromIntError},
29    path::{Path, PathBuf},
30};
31
32use shardline_index::{LocalRecordStore, PostgresRecordStore, RecordStore, RecordTraversal};
33use shardline_protocol::{
34    ByteRange, RepositoryProvider, ShardlineHash, TokenClaims, TokenCodecError, TokenScope,
35};
36use shardline_storage::{
37    DeleteOutcome, LocalObjectStore, LocalObjectStoreError, ObjectBody, ObjectIntegrity, ObjectKey,
38    ObjectKeyError, ObjectMetadata, ObjectPrefix, ObjectStore, PutOutcome, S3ObjectStore,
39    S3ObjectStoreConfig, S3ObjectStoreError,
40};
41use thiserror::Error;
42
43pub mod auth;
44pub mod protocol_support;
45pub mod server_frontend;
46
47/// Provider-agnostic authentication trait.
48///
49/// Implementations verify and mint scoped bearer tokens for the Shardline API.
50/// The server selects a concrete provider at startup based on configuration.
51pub trait AuthProvider: Send + Sync {
52    /// Verifies an opaque bearer token and returns the decoded claims.
53    ///
54    /// # Errors
55    ///
56    /// Returns [`AuthError`] when the token is invalid, expired, or otherwise
57    /// unverifiable.
58    fn verify_token(&self, token: &str) -> Result<TokenClaims, AuthError>;
59
60    /// Mints a signed bearer token from the provided claims.
61    ///
62    /// # Errors
63    ///
64    /// Returns [`AuthError`] when the provider does not support token minting
65    /// or when signing fails.
66    fn mint_token(&self, claims: &TokenClaims) -> Result<String, AuthError>;
67}
68
69/// Verified request authorization context.
70#[derive(Debug, Clone)]
71pub struct AuthContext {
72    /// The decoded token claims.
73    pub claims: TokenClaims,
74}
75
76impl AuthContext {
77    /// Creates an authorization context from verified token claims.
78    #[must_use]
79    pub const fn new(claims: TokenClaims) -> Self {
80        Self { claims }
81    }
82
83    /// Returns the verified claims.
84    #[must_use]
85    pub const fn claims(&self) -> &TokenClaims {
86        &self.claims
87    }
88
89    /// Returns the authenticated subject.
90    #[must_use]
91    pub fn subject(&self) -> &str {
92        self.claims.subject()
93    }
94
95    /// Returns the granted scope.
96    #[must_use]
97    pub const fn scope(&self) -> TokenScope {
98        self.claims.scope()
99    }
100}
101
102/// Authentication provider failure.
103#[derive(Debug, Error)]
104pub enum AuthError {
105    /// The token format was invalid.
106    #[error("invalid token")]
107    InvalidToken,
108    /// The token has expired.
109    #[error("expired token")]
110    ExpiredToken,
111    /// The token does not grant the required scope.
112    #[error("insufficient scope")]
113    InsufficientScope,
114    /// The provider encountered an internal error.
115    #[error("provider error: {0}")]
116    ProviderError(String),
117}
118
119impl From<TokenCodecError> for AuthError {
120    fn from(error: TokenCodecError) -> Self {
121        match error {
122            TokenCodecError::Expired => Self::ExpiredToken,
123            TokenCodecError::InvalidSignature
124            | TokenCodecError::InvalidFormat
125            | TokenCodecError::InvalidHex(_)
126            | TokenCodecError::Claims(_) => Self::InvalidToken,
127            TokenCodecError::EmptySigningKey
128            | TokenCodecError::SigningKeyTooShort
129            | TokenCodecError::Json(_) => Self::ProviderError(error.to_string()),
130        }
131    }
132}
133
134/// Validates that a content hash is exactly 64 lowercase hex characters.
135///
136/// # Errors
137///
138/// Returns an error with the given `error_fn` when the hash is malformed.
139pub fn validate_content_hash_with<E>(value: &str, error_fn: fn() -> E) -> Result<(), E> {
140    if value.len() != 64
141        || !value
142            .bytes()
143            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
144    {
145        return Err(error_fn());
146    }
147    Ok(())
148}
149
150/// Returns the chunk object key for a hex-encoded content hash.
151///
152/// # Errors
153///
154/// Returns [`ServerObjectStoreError`] when the hash is malformed or the key cannot be created.
155pub fn chunk_object_key(hash_hex: &str) -> Result<ObjectKey, ServerObjectStoreError> {
156    validate_content_hash_with(hash_hex, || ServerObjectStoreError::Overflow)?;
157    let prefix = hash_hex.get(..2).ok_or(ServerObjectStoreError::Overflow)?;
158    let key = format!("{prefix}/{hash_hex}");
159    ObjectKey::parse(&key).map_err(map_object_key_error)
160}
161
162/// Extracts the chunk hash from a chunk object key if the key matches the expected layout.
163///
164/// Returns `Some(hash_hex)` if the key is in the format `<2-char-prefix>/<64-char-hash>`,
165/// `None` otherwise.
166///
167/// # Errors
168///
169/// Returns [`ServerObjectStoreError::InvalidContentHash`] if the extracted hash fails validation.
170pub fn chunk_hash_from_chunk_object_key_if_present(
171    key: &ObjectKey,
172) -> Result<Option<&str>, ServerObjectStoreError> {
173    let mut segments = key.as_str().split('/');
174    let Some(prefix) = segments.next() else {
175        return Ok(None);
176    };
177    let Some(candidate_hash_hex) = segments.next() else {
178        return Ok(None);
179    };
180    if segments.next().is_some() {
181        return Ok(None);
182    }
183    if prefix.len() != 2 || !prefix.bytes().all(|byte| byte.is_ascii_hexdigit()) {
184        return Ok(None);
185    }
186    if !candidate_hash_hex.starts_with(prefix) {
187        return Ok(None);
188    }
189    validate_content_hash_with(candidate_hash_hex, || {
190        ServerObjectStoreError::InvalidContentHash
191    })?;
192    Ok(Some(candidate_hash_hex))
193}
194
195/// Computes a blake3 content hash for the given bytes.
196#[must_use]
197pub fn chunk_hash(bytes: &[u8]) -> ShardlineHash {
198    let digest = blake3::hash(bytes);
199    ShardlineHash::from_bytes(*digest.as_bytes())
200}
201
202/// Computes a blake3 content hash for a file record's chunk layout.
203#[must_use]
204pub fn content_hash(
205    total_bytes: u64,
206    chunk_size: u64,
207    chunks: &[shardline_index::FileChunkRecord],
208) -> String {
209    let mut hasher = blake3::Hasher::new();
210    hasher.update(&total_bytes.to_le_bytes());
211    hasher.update(&chunk_size.to_le_bytes());
212    for chunk in chunks {
213        hasher.update(chunk.hash.as_bytes());
214        hasher.update(&chunk.offset.to_le_bytes());
215        hasher.update(&chunk.length.to_le_bytes());
216    }
217    hasher.finalize().to_hex().to_string()
218}
219
220const fn map_object_key_error(error: ObjectKeyError) -> ServerObjectStoreError {
221    match error {
222        ObjectKeyError::Empty
223        | ObjectKeyError::UnsafePath
224        | ObjectKeyError::ControlCharacter
225        | ObjectKeyError::TooLong => ServerObjectStoreError::Overflow,
226    }
227}
228
229/// Reads the full contents of an object from the store.
230///
231/// # Errors
232///
233/// Returns [`ServerObjectStoreError`] on storage backend failures, length
234/// mismatches, or arithmetic overflows.
235pub fn read_full_object(
236    store: &ServerObjectStore,
237    object_key: &ObjectKey,
238    length: u64,
239) -> Result<Vec<u8>, ServerObjectStoreError> {
240    store.read_full_object(object_key, length)
241}
242
243/// Lifecycle metadata consistency failure.
244#[derive(Debug, Clone, Error, PartialEq, Eq)]
245pub enum InvalidLifecycleMetadataError {
246    /// A quarantine candidate cannot be deleted before it was first observed.
247    #[error(
248        "quarantine candidate for {object_key} had delete-after {delete_after_unix_seconds} before first-seen {first_seen_unreachable_at_unix_seconds}"
249    )]
250    QuarantineCandidateDeleteBeforeFirstSeen {
251        /// Quarantined object key.
252        object_key: String,
253        /// Candidate deletion timestamp.
254        delete_after_unix_seconds: u64,
255        /// First observed unreachable timestamp.
256        first_seen_unreachable_at_unix_seconds: u64,
257    },
258    /// A quarantine candidate referenced an object that is no longer present.
259    #[error("quarantine candidate referenced missing object {object_key}")]
260    QuarantineCandidateMissingObject {
261        /// Quarantined object key.
262        object_key: String,
263    },
264    /// A quarantine candidate recorded a length that differs from object-store metadata.
265    #[error(
266        "quarantine candidate for {object_key} expected length {expected_length}, got {observed_length}"
267    )]
268    QuarantineCandidateLengthMismatch {
269        /// Quarantined object key.
270        object_key: String,
271        /// Length recorded in quarantine metadata.
272        expected_length: u64,
273        /// Length observed in object-store metadata.
274        observed_length: u64,
275    },
276    /// A retention hold cannot be released before it was created.
277    #[error(
278        "retention hold for {object_key} had release-after {release_after_unix_seconds} before held-at {held_at_unix_seconds}"
279    )]
280    RetentionHoldReleaseBeforeHeld {
281        /// Held object key.
282        object_key: String,
283        /// Hold release timestamp.
284        release_after_unix_seconds: u64,
285        /// Hold creation timestamp.
286        held_at_unix_seconds: u64,
287    },
288    /// An active retention hold referenced an object that is no longer present.
289    #[error("active retention hold referenced missing object {object_key}")]
290    ActiveRetentionHoldMissingObject {
291        /// Held object key.
292        object_key: String,
293    },
294    /// An active retention hold coexisted with quarantine metadata for the same object.
295    #[error("active retention hold for {object_key} coexisted with quarantine state")]
296    ActiveRetentionHoldQuarantined {
297        /// Held object key.
298        object_key: String,
299    },
300}
301
302/// Serialized shard validation failure.
303#[derive(Debug, Clone, Copy, Error, PartialEq, Eq)]
304pub enum InvalidSerializedShardError {
305    /// The external shard parser rejected the bytes.
306    #[error("shard parser rejected metadata")]
307    ParserRejectedMetadata,
308    /// A native Xet term used an empty or inverted chunk range.
309    #[error("native xet term had an empty or inverted chunk range")]
310    NativeXetTermEmptyOrInvertedChunkRange,
311    /// A native Xet term referenced chunks past the end of its xorb.
312    #[error("native xet term range exceeded xorb chunk count")]
313    NativeXetTermRangeExceededXorbChunkCount,
314    /// A shard file term used an empty or inverted chunk range.
315    #[error("shard file term had an empty or inverted chunk range")]
316    ShardFileTermEmptyOrInvertedChunkRange,
317    /// The transient xorb metadata cache could not return a just-inserted entry.
318    #[error("xorb metadata cache insertion failed")]
319    XorbMetadataCacheInsertionFailed,
320    /// A shard term started past the referenced xorb chunk list.
321    #[error("shard term chunk range started past the xorb chunk list")]
322    ShardTermRangeStartedPastXorbChunkList,
323    /// A shard term ended past the referenced xorb chunk list.
324    #[error("shard term chunk range ended past the xorb chunk list")]
325    ShardTermRangeEndedPastXorbChunkList,
326    /// The retained shard chunk hash list was not strictly ordered.
327    #[error("retained shard chunk hashes were not strictly ordered")]
328    RetainedShardChunkHashesNotStrictlyOrdered,
329}
330
331/// Reconstruction response shape failure.
332#[derive(Debug, Clone, Copy, Error, PartialEq, Eq)]
333pub enum InvalidReconstructionResponseError {
334    /// A guarded test record store detected a forbidden global latest-record walk.
335    #[error("global latest-record walk attempted")]
336    RecordStoreGlobalLatestWalkAttempted,
337    /// A guarded test record store could not find the requested record.
338    #[error("record not found")]
339    RecordStoreRecordNotFound,
340    /// V1 response emitted more terms than the source record has chunks.
341    #[error("response term count exceeded record chunk count")]
342    TermCountExceededRecordChunkCount,
343    /// A response term had no bytes.
344    #[error("response term had zero unpacked length")]
345    TermHadZeroUnpackedLength,
346    /// A response term contained an empty chunk range.
347    #[error("response term had an empty chunk range")]
348    TermHadEmptyChunkRange,
349    /// A response term did not have matching fetch metadata.
350    #[error("response term did not have matching fetch info")]
351    TermMissingFetchInfo,
352    /// A fetch-info entry had no fetches.
353    #[error("response fetch info contained an empty fetch list")]
354    EmptyFetchList,
355    /// A fetch URL did not point to the xorb hash that owns it.
356    #[error("response fetch URL did not match its xorb hash")]
357    FetchUrlHashMismatch,
358    /// A fetch entry had an empty chunk range.
359    #[error("response fetch entry had an empty chunk range")]
360    FetchEntryEmptyChunkRange,
361    /// A fetch entry had an inverted byte range.
362    #[error("response fetch entry had an inverted byte range")]
363    FetchEntryInvertedByteRange,
364    /// A fetch entry did not correspond to any response term.
365    #[error("response fetch entry did not have a matching term")]
366    FetchEntryMissingTerm,
367    /// V2 conversion changed `offset_into_first_range`.
368    #[error("v2 response changed offset_into_first_range")]
369    V2ChangedOffsetIntoFirstRange,
370    /// V2 conversion changed the reconstruction terms.
371    #[error("v2 response changed reconstruction terms")]
372    V2ChangedTerms,
373    /// V2 conversion changed the xorb fetch-info cardinality.
374    #[error("v2 response changed xorb fetch-info cardinality")]
375    V2ChangedXorbFetchInfoCardinality,
376    /// V2 conversion emitted a hash absent from V1 fetch-info.
377    #[error("v2 response emitted a fetch hash absent from v1")]
378    V2FetchHashAbsentFromV1,
379    /// V2 conversion emitted an empty fetch list.
380    #[error("v2 response emitted an empty fetch list")]
381    V2EmptyFetchList,
382    /// V2 conversion emitted a fetch entry without ranges.
383    #[error("v2 response emitted a fetch entry without ranges")]
384    V2FetchEntryWithoutRanges,
385    /// V2 conversion emitted an empty chunk range.
386    #[error("v2 response emitted an empty chunk range")]
387    V2EmptyChunkRange,
388    /// V2 conversion emitted an inverted byte range.
389    #[error("v2 response emitted an inverted byte range")]
390    V2InvertedByteRange,
391    /// V2 fetch count did not match V1.
392    #[error("v2 response fetch count disagreed with v1")]
393    V2FetchCountDisagreedWithV1,
394    /// V2 range count did not match V1.
395    #[error("v2 response range count disagreed with v1")]
396    V2RangeCountDisagreedWithV1,
397}
398
399/// Default bounded-parser limits for native Xet shard metadata.
400pub const DEFAULT_MAX_SHARD_FILES: NonZeroUsize = match NonZeroUsize::new(16_384) {
401    Some(value) => value,
402    None => NonZeroUsize::MIN,
403};
404
405/// Default maximum shard xorb sections.
406pub const DEFAULT_MAX_SHARD_XORBS: NonZeroUsize = match NonZeroUsize::new(16_384) {
407    Some(value) => value,
408    None => NonZeroUsize::MIN,
409};
410
411/// Default maximum shard reconstruction terms.
412pub const DEFAULT_MAX_SHARD_RECONSTRUCTION_TERMS: NonZeroUsize = match NonZeroUsize::new(65_536) {
413    Some(value) => value,
414    None => NonZeroUsize::MIN,
415};
416
417/// Default maximum shard xorb chunk records.
418pub const DEFAULT_MAX_SHARD_XORB_CHUNKS: NonZeroUsize = match NonZeroUsize::new(65_536) {
419    Some(value) => value,
420    None => NonZeroUsize::MIN,
421};
422
423/// Default bounded-parser limits for native Xet shard metadata.
424pub const DEFAULT_SHARD_METADATA_LIMITS: ShardMetadataLimits = ShardMetadataLimits::new(
425    DEFAULT_MAX_SHARD_FILES,
426    DEFAULT_MAX_SHARD_XORBS,
427    DEFAULT_MAX_SHARD_RECONSTRUCTION_TERMS,
428    DEFAULT_MAX_SHARD_XORB_CHUNKS,
429);
430
431/// Bounded-parser limits for native Xet shard metadata.
432#[derive(Debug, Clone, Copy, PartialEq, Eq)]
433pub struct ShardMetadataLimits {
434    max_files: NonZeroUsize,
435    max_xorbs: NonZeroUsize,
436    max_reconstruction_terms: NonZeroUsize,
437    max_xorb_chunks: NonZeroUsize,
438}
439
440impl ShardMetadataLimits {
441    /// Creates native Xet shard metadata limits.
442    #[must_use]
443    pub const fn new(
444        max_files: NonZeroUsize,
445        max_xorbs: NonZeroUsize,
446        max_reconstruction_terms: NonZeroUsize,
447        max_xorb_chunks: NonZeroUsize,
448    ) -> Self {
449        Self {
450            max_files,
451            max_xorbs,
452            max_reconstruction_terms,
453            max_xorb_chunks,
454        }
455    }
456
457    /// Returns the maximum file sections accepted in one uploaded shard.
458    #[must_use]
459    pub const fn max_files(self) -> NonZeroUsize {
460        self.max_files
461    }
462
463    /// Returns the maximum xorb sections accepted in one uploaded shard.
464    #[must_use]
465    pub const fn max_xorbs(self) -> NonZeroUsize {
466        self.max_xorbs
467    }
468
469    /// Returns the maximum file reconstruction terms accepted in one uploaded shard.
470    #[must_use]
471    pub const fn max_reconstruction_terms(self) -> NonZeroUsize {
472        self.max_reconstruction_terms
473    }
474
475    /// Returns the maximum xorb chunk records accepted in one uploaded shard.
476    #[must_use]
477    pub const fn max_xorb_chunks(self) -> NonZeroUsize {
478        self.max_xorb_chunks
479    }
480}
481
482impl Default for ShardMetadataLimits {
483    fn default() -> Self {
484        DEFAULT_SHARD_METADATA_LIMITS
485    }
486}
487
488/// Object-store backend error.
489#[derive(Debug, Error)]
490pub enum ServerObjectStoreError {
491    /// Requested content was not found.
492    #[error("content not found")]
493    NotFound,
494    /// Arithmetic overflowed a checked bound.
495    #[error("arithmetic overflow")]
496    Overflow,
497    /// A content hash was malformed.
498    #[error("content hash must be 64 hexadecimal characters")]
499    InvalidContentHash,
500    /// Stored object metadata disagreed with the expected transfer length.
501    #[error("stored object length did not match indexed metadata")]
502    StoredObjectLengthMismatch,
503    /// Local storage IO failed.
504    #[error("local storage operation failed")]
505    Local(#[from] LocalObjectStoreError),
506    /// S3-compatible object-storage access failed.
507    #[error("s3 object storage operation failed")]
508    S3(#[from] S3ObjectStoreError),
509    /// A local filesystem I/O error occurred.
510    #[error("local storage io failed")]
511    Io(#[from] IoError),
512    /// Numeric conversion exceeded supported bounds.
513    #[error("numeric conversion exceeded supported bounds")]
514    NumericConversion(#[from] TryFromIntError),
515}
516
517/// Unified object-store backend that delegates to local, S3, or blackhole storage.
518#[derive(Debug, Clone)]
519pub enum ServerObjectStore {
520    /// Local filesystem object store.
521    Local(LocalObjectStore),
522    /// S3-compatible object store.
523    S3(S3ObjectStore),
524    /// Blackhole object store that discards all writes.
525    Blackhole,
526}
527
528impl ObjectStore for ServerObjectStore {
529    type Error = ServerObjectStoreError;
530
531    fn put_if_absent(
532        &self,
533        key: &ObjectKey,
534        body: ObjectBody<'_>,
535        integrity: &ObjectIntegrity,
536    ) -> Result<PutOutcome, Self::Error> {
537        match self {
538            Self::Local(store) => Ok(store.put_if_absent(key, body, integrity)?),
539            Self::S3(store) => Ok(store.put_if_absent(key, body, integrity)?),
540            Self::Blackhole => Ok(PutOutcome::Inserted),
541        }
542    }
543
544    fn read_range(&self, key: &ObjectKey, range: ByteRange) -> Result<Vec<u8>, Self::Error> {
545        match self {
546            Self::Local(store) => Ok(store.read_range(key, range)?),
547            Self::S3(store) => Ok(store.read_range(key, range)?),
548            Self::Blackhole => Err(ServerObjectStoreError::NotFound),
549        }
550    }
551
552    fn contains(&self, key: &ObjectKey) -> Result<bool, Self::Error> {
553        match self {
554            Self::Local(store) => Ok(store.contains(key)?),
555            Self::S3(store) => Ok(store.contains(key)?),
556            Self::Blackhole => Ok(false),
557        }
558    }
559
560    fn metadata(&self, key: &ObjectKey) -> Result<Option<ObjectMetadata>, Self::Error> {
561        match self {
562            Self::Local(store) => Ok(store.metadata(key)?),
563            Self::S3(store) => Ok(store.metadata(key)?),
564            Self::Blackhole => Ok(None),
565        }
566    }
567
568    fn list_prefix(&self, prefix: &ObjectPrefix) -> Result<Vec<ObjectMetadata>, Self::Error> {
569        match self {
570            Self::Local(store) => Ok(store.list_prefix(prefix)?),
571            Self::S3(store) => Ok(store.list_prefix(prefix)?),
572            Self::Blackhole => Ok(Vec::new()),
573        }
574    }
575
576    fn delete_if_present(&self, key: &ObjectKey) -> Result<DeleteOutcome, Self::Error> {
577        match self {
578            Self::Local(store) => Ok(store.delete_if_present(key)?),
579            Self::S3(store) => Ok(store.delete_if_present(key)?),
580            Self::Blackhole => Ok(DeleteOutcome::NotFound),
581        }
582    }
583}
584
585impl ServerObjectStore {
586    /// Creates a local filesystem object store rooted at the given path.
587    ///
588    /// # Errors
589    ///
590    /// Returns [`ServerObjectStoreError::Local`] if the local store cannot be created.
591    pub fn local(root: impl Into<PathBuf>) -> Result<Self, ServerObjectStoreError> {
592        Ok(Self::Local(LocalObjectStore::new(root.into())?))
593    }
594
595    /// Creates an S3-compatible object store from the provided configuration.
596    ///
597    /// # Errors
598    ///
599    /// Returns [`ServerObjectStoreError::S3`] if the S3 store cannot be created.
600    pub fn s3(config: S3ObjectStoreConfig) -> Result<Self, ServerObjectStoreError> {
601        Ok(Self::S3(S3ObjectStore::new(config)?))
602    }
603
604    /// Creates a blackhole object store that discards all writes.
605    #[must_use]
606    pub const fn blackhole() -> Self {
607        Self::Blackhole
608    }
609
610    /// Stores an object, overwriting any existing object at the given key.
611    ///
612    /// # Errors
613    ///
614    /// Returns [`ServerObjectStoreError`] on storage backend failures.
615    pub fn put_overwrite(
616        &self,
617        key: &ObjectKey,
618        body: ObjectBody<'_>,
619        integrity: &ObjectIntegrity,
620    ) -> Result<(), ServerObjectStoreError> {
621        match self {
622            Self::Local(store) => store
623                .put_overwrite(key, body, integrity)
624                .map_err(Into::into),
625            Self::S3(store) => store
626                .put_overwrite(key, body, integrity)
627                .map_err(Into::into),
628            Self::Blackhole => Ok(()),
629        }
630    }
631
632    /// Visits all objects under the given prefix, invoking the visitor for each.
633    ///
634    /// # Errors
635    ///
636    /// Returns any error produced by the visitor or the underlying storage backend.
637    pub fn visit_prefix<F, E>(&self, prefix: &ObjectPrefix, mut visitor: F) -> Result<(), E>
638    where
639        F: FnMut(ObjectMetadata) -> Result<(), E>,
640        E: From<LocalObjectStoreError> + From<S3ObjectStoreError>,
641    {
642        match self {
643            Self::Local(store) => store.visit_prefix(prefix, &mut visitor),
644            Self::S3(store) => store.visit_prefix(prefix, &mut visitor),
645            Self::Blackhole => Ok(()),
646        }
647    }
648
649    /// Lists objects under the given prefix with pagination.
650    ///
651    /// # Errors
652    ///
653    /// Returns [`ServerObjectStoreError`] on storage backend failures.
654    pub fn list_flat_namespace_page(
655        &self,
656        prefix: &ObjectPrefix,
657        start_after: Option<&ObjectKey>,
658        limit: usize,
659    ) -> Result<Vec<ObjectMetadata>, ServerObjectStoreError> {
660        match self {
661            Self::Local(store) => store
662                .list_flat_namespace_page(prefix, start_after, limit)
663                .map_err(Into::into),
664            Self::S3(store) => store
665                .list_flat_namespace_page(prefix, start_after, limit)
666                .map_err(Into::into),
667            Self::Blackhole => Ok(Vec::new()),
668        }
669    }
670
671    /// Returns the local filesystem path for an object key, if backed by local storage.
672    #[must_use]
673    pub fn local_path_for_key(&self, key: &ObjectKey) -> Option<PathBuf> {
674        match self {
675            Self::Local(store) => Some(store.path_for_key(key)),
676            Self::S3(_store) => None,
677            Self::Blackhole => None,
678        }
679    }
680
681    /// Copies an object from source to destination if no object exists at the destination.
682    ///
683    /// # Errors
684    ///
685    /// Returns [`ServerObjectStoreError::NotFound`] for blackhole stores or
686    /// storage backend errors.
687    pub fn copy_if_absent(
688        &self,
689        source: &ObjectKey,
690        destination: &ObjectKey,
691    ) -> Result<PutOutcome, ServerObjectStoreError> {
692        match self {
693            Self::Local(store) => store
694                .copy_object_if_absent(source, destination)
695                .map_err(Into::into),
696            Self::S3(store) => store
697                .copy_object_if_absent(source, destination)
698                .map_err(Into::into),
699            Self::Blackhole => Err(ServerObjectStoreError::NotFound),
700        }
701    }
702
703    /// Stores a content-addressed file from the local filesystem.
704    ///
705    /// # Errors
706    ///
707    /// Returns [`ServerObjectStoreError`] on storage backend failures.
708    pub fn put_content_addressed_file(
709        &self,
710        key: &ObjectKey,
711        path: &Path,
712        integrity: &ObjectIntegrity,
713    ) -> Result<PutOutcome, ServerObjectStoreError> {
714        match self {
715            Self::Local(store) => store
716                .put_temporary_file_if_absent(key, path, integrity)
717                .map_err(Into::into),
718            Self::S3(store) => store
719                .put_content_addressed_file(key, path, integrity)
720                .map_err(Into::into),
721            Self::Blackhole => Ok(PutOutcome::Inserted),
722        }
723    }
724
725    /// Returns the local filesystem root, if backed by local storage.
726    #[must_use]
727    pub fn local_root(&self) -> Option<&Path> {
728        match self {
729            Self::Local(store) => Some(store.root()),
730            Self::S3(_store) => None,
731            Self::Blackhole => None,
732        }
733    }
734
735    /// Returns the backend name for this object store.
736    #[must_use]
737    pub const fn backend_name(&self) -> &'static str {
738        match self {
739            Self::Local(_store) => "local",
740            Self::S3(_store) => "s3",
741            Self::Blackhole => "blackhole",
742        }
743    }
744
745    /// Reads the full contents of an object from the store.
746    ///
747    /// # Errors
748    ///
749    /// Returns [`ServerObjectStoreError`] on storage backend failures, length
750    /// mismatches, or arithmetic overflows.
751    pub fn read_full_object(
752        &self,
753        object_key: &ObjectKey,
754        length: u64,
755    ) -> Result<Vec<u8>, ServerObjectStoreError> {
756        if length == 0 {
757            return Ok(Vec::new());
758        }
759
760        if let Self::Local(store) = self {
761            let file = store.open_object_file(object_key)?;
762            let actual_length = file.metadata()?.len();
763            if actual_length != length {
764                return Err(ServerObjectStoreError::StoredObjectLengthMismatch);
765            }
766            let capacity = usize::try_from(length)?;
767            let mut output = Vec::with_capacity(capacity);
768            let mut limited = file.take(length);
769            Read::read_to_end(&mut limited, &mut output)?;
770            if output.len() != capacity {
771                return Err(ServerObjectStoreError::StoredObjectLengthMismatch);
772            }
773            return Ok(output);
774        }
775
776        let end = length
777            .checked_sub(1)
778            .ok_or(ServerObjectStoreError::Overflow)?;
779        let range = ByteRange::new(0, end).map_err(|_error| ServerObjectStoreError::Overflow)?;
780        self.read_range(object_key, range)
781    }
782}
783
784/// Operation-time record-store classification.
785#[derive(Debug, Clone, Copy, PartialEq, Eq)]
786pub enum OpsRecordKind {
787    /// Latest record.
788    Latest,
789    /// Version record.
790    Version,
791}
792
793/// Extra locator metadata needed by operator tooling.
794pub trait OpsRecordStore: RecordStore {
795    /// Renders a stable operator-facing location for one record locator.
796    fn locator_display(&self, locator: &<Self as RecordTraversal>::Locator) -> String;
797
798    /// Extracts the file identifier implied by a locator.
799    fn locator_file_id(
800        &self,
801        locator: &<Self as RecordTraversal>::Locator,
802        kind: OpsRecordKind,
803    ) -> Option<String>;
804
805    /// Extracts the immutable content hash implied by a version locator.
806    fn locator_content_hash(
807        &self,
808        locator: &<Self as RecordTraversal>::Locator,
809        kind: OpsRecordKind,
810    ) -> Option<String>;
811}
812
813impl OpsRecordStore for LocalRecordStore {
814    fn locator_display(&self, locator: &<Self as RecordTraversal>::Locator) -> String {
815        locator.record_key().to_owned()
816    }
817
818    fn locator_file_id(
819        &self,
820        locator: &<Self as RecordTraversal>::Locator,
821        _kind: OpsRecordKind,
822    ) -> Option<String> {
823        Some(locator.file_id().to_owned())
824    }
825
826    fn locator_content_hash(
827        &self,
828        locator: &<Self as RecordTraversal>::Locator,
829        kind: OpsRecordKind,
830    ) -> Option<String> {
831        if kind != OpsRecordKind::Version {
832            return None;
833        }
834
835        locator.content_hash().map(ToOwned::to_owned)
836    }
837}
838
839impl OpsRecordStore for PostgresRecordStore {
840    fn locator_display(&self, locator: &<Self as RecordTraversal>::Locator) -> String {
841        locator.record_key().to_owned()
842    }
843
844    fn locator_file_id(
845        &self,
846        locator: &<Self as RecordTraversal>::Locator,
847        _kind: OpsRecordKind,
848    ) -> Option<String> {
849        Some(locator.file_id().to_owned())
850    }
851
852    fn locator_content_hash(
853        &self,
854        locator: &<Self as RecordTraversal>::Locator,
855        kind: OpsRecordKind,
856    ) -> Option<String> {
857        if kind != OpsRecordKind::Version {
858            return None;
859        }
860
861        locator.content_hash().map(ToOwned::to_owned)
862    }
863}
864
865/// Maximum allowed stored file record metadata size in bytes.
866pub const MAX_LOCAL_RECORD_METADATA_BYTES: u64 = 1_073_741_824;
867
868/// Parses stored file record bytes, rejecting oversized metadata before JSON parsing.
869///
870/// # Errors
871///
872/// Returns an error if the metadata exceeds [`MAX_LOCAL_RECORD_METADATA_BYTES`] or
873/// if JSON deserialization fails.
874pub fn parse_stored_file_record_bytes(
875    bytes: &[u8],
876) -> Result<shardline_index::FileRecord, ParseStoredFileRecordError> {
877    let observed_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
878    if observed_bytes > MAX_LOCAL_RECORD_METADATA_BYTES {
879        return Err(ParseStoredFileRecordError::StoredFileMetadataTooLarge {
880            observed_bytes,
881            maximum_bytes: MAX_LOCAL_RECORD_METADATA_BYTES,
882        });
883    }
884
885    Ok(serde_json::from_slice(bytes)?)
886}
887
888/// Stored file record parsing failure.
889#[derive(Debug, Error)]
890pub enum ParseStoredFileRecordError {
891    /// Stored file metadata exceeded the bounded parser ceiling.
892    #[error("stored file metadata exceeded the bounded parser ceiling")]
893    StoredFileMetadataTooLarge {
894        /// Observed file length in bytes.
895        observed_bytes: u64,
896        /// Maximum accepted file length in bytes.
897        maximum_bytes: u64,
898    },
899    /// JSON deserialization failed.
900    #[error("json operation failed")]
901    Json(#[from] serde_json::Error),
902}
903
904/// Returns the provider directory string for the given repository provider.
905#[must_use]
906pub const fn provider_directory(provider: RepositoryProvider) -> &'static str {
907    provider.as_str()
908}
909
910/// Maximum byte length for a validated file identifier.
911const MAX_IDENTIFIER_BYTES: usize = 1024;
912
913/// Validates that a file identifier is safe for use as a single path component.
914///
915/// # Errors
916///
917/// Returns [`ValidateIdentifierError`] if the identifier is empty, contains
918/// path separators, traversal sequences, control characters, or exceeds the
919/// maximum byte length.
920pub fn validate_identifier(value: &str) -> Result<(), ValidateIdentifierError> {
921    if value.trim().is_empty()
922        || value == "."
923        || value.len() > MAX_IDENTIFIER_BYTES
924        || value.starts_with('/')
925        || value.contains("..")
926        || value.contains('\\')
927        || value.contains('/')
928        || value.chars().any(char::is_control)
929    {
930        return Err(ValidateIdentifierError);
931    }
932
933    Ok(())
934}
935
936/// File identifier validation failure.
937#[derive(Debug, Clone, Copy, Error)]
938#[error("file identifier must be relative and must not contain traversal or control characters")]
939pub struct ValidateIdentifierError;
940
941/// Validates that a content hash is exactly 64 lowercase hex characters.
942///
943/// # Errors
944///
945/// Returns [`ValidateContentHashError`] if the hash is malformed.
946pub fn validate_content_hash(value: &str) -> Result<(), ValidateContentHashError> {
947    if value.len() != 64
948        || !value
949            .bytes()
950            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
951    {
952        return Err(ValidateContentHashError);
953    }
954
955    Ok(())
956}
957
958/// Content hash validation failure.
959#[derive(Debug, Clone, Copy, Error)]
960#[error("content hash must be 64 hexadecimal characters")]
961pub struct ValidateContentHashError;
962
963/// Checked addition returning an error on overflow.
964///
965/// # Errors
966///
967/// Returns [`RebuildOverflowError`] when the addition overflows.
968pub const fn checked_add(left: u64, right: u64) -> Result<u64, RebuildOverflowError> {
969    match left.checked_add(right) {
970        Some(value) => Ok(value),
971        None => Err(RebuildOverflowError),
972    }
973}
974
975/// Checked increment returning an error on overflow.
976///
977/// # Errors
978///
979/// Returns [`RebuildOverflowError`] when the increment overflows.
980pub const fn checked_increment(value: u64) -> Result<u64, RebuildOverflowError> {
981    checked_add(value, 1)
982}
983
984/// Arithmetic overflow during rebuild operations.
985#[derive(Debug, Clone, Copy, Error)]
986#[error("arithmetic overflow")]
987pub struct RebuildOverflowError;
988
989/// Returns the current Unix time in seconds, or an error if the system clock
990/// is before the Unix epoch.
991///
992/// # Errors
993///
994/// Returns [`RebuildOverflowError`] when the system time is before the Unix
995/// epoch.
996pub fn unix_now_seconds_checked() -> Result<u64, RebuildOverflowError> {
997    std::time::SystemTime::now()
998        .duration_since(std::time::UNIX_EPOCH)
999        .map(|duration| duration.as_secs())
1000        .map_err(|_e| RebuildOverflowError)
1001}
1002
1003/// Default retention window for new local quarantine candidates.
1004pub const DEFAULT_LOCAL_GC_RETENTION_SECONDS: u64 = 86_400;
1005
1006#[cfg(test)]
1007mod tests {
1008    use super::*;
1009
1010    use proptest::prelude::*;
1011
1012    #[test]
1013    fn validate_identifier_accepts_simple_name() {
1014        assert!(validate_identifier("hello.txt").is_ok());
1015    }
1016
1017    #[test]
1018    fn validate_identifier_accepts_dotted_name() {
1019        assert!(validate_identifier("file.name.txt").is_ok());
1020    }
1021
1022    #[test]
1023    fn validate_identifier_rejects_empty() {
1024        assert!(validate_identifier("").is_err());
1025    }
1026
1027    #[test]
1028    fn validate_identifier_rejects_whitespace_only() {
1029        assert!(validate_identifier("   ").is_err());
1030    }
1031
1032    #[test]
1033    fn validate_identifier_rejects_dot() {
1034        assert!(validate_identifier(".").is_err());
1035    }
1036
1037    #[test]
1038    fn validate_identifier_rejects_leading_slash() {
1039        assert!(validate_identifier("/etc/passwd").is_err());
1040    }
1041
1042    #[test]
1043    fn validate_identifier_rejects_traversal() {
1044        assert!(validate_identifier("foo/../bar").is_err());
1045    }
1046
1047    #[test]
1048    fn validate_identifier_rejects_backslash() {
1049        assert!(validate_identifier("foo\\bar").is_err());
1050    }
1051
1052    #[test]
1053    fn validate_identifier_rejects_control_char() {
1054        assert!(validate_identifier("foo\tbar").is_err());
1055    }
1056
1057    #[test]
1058    fn validate_content_hash_accepts_valid_hash() {
1059        let hash = "a".repeat(64);
1060        assert!(validate_content_hash(&hash).is_ok());
1061    }
1062
1063    #[test]
1064    fn validate_content_hash_rejects_too_short() {
1065        assert!(validate_content_hash("abc123").is_err());
1066    }
1067
1068    #[test]
1069    fn validate_content_hash_rejects_too_long() {
1070        let hash = "a".repeat(65);
1071        assert!(validate_content_hash(&hash).is_err());
1072    }
1073
1074    #[test]
1075    fn validate_content_hash_rejects_uppercase() {
1076        let hash = "A".repeat(64);
1077        assert!(validate_content_hash(&hash).is_err());
1078    }
1079
1080    #[test]
1081    fn validate_content_hash_rejects_non_hex() {
1082        let mut hash = "a".repeat(64);
1083        hash.push('g');
1084        hash.remove(0);
1085        assert!(validate_content_hash(&hash).is_err());
1086    }
1087
1088    #[test]
1089    fn checked_add_normal() {
1090        assert_eq!(checked_add(1, 2).unwrap(), 3);
1091    }
1092
1093    #[test]
1094    fn checked_add_zero() {
1095        assert_eq!(checked_add(0, 0).unwrap(), 0);
1096    }
1097
1098    #[test]
1099    fn checked_add_overflow() {
1100        assert!(checked_add(u64::MAX, 1).is_err());
1101    }
1102
1103    #[test]
1104    fn checked_increment_normal() {
1105        assert_eq!(checked_increment(0).unwrap(), 1);
1106    }
1107
1108    #[test]
1109    fn checked_increment_overflow() {
1110        assert!(checked_increment(u64::MAX).is_err());
1111    }
1112
1113    #[test]
1114    fn chunk_object_key_valid() {
1115        let hash = "a".repeat(64);
1116        let key = chunk_object_key(&hash).unwrap();
1117        assert!(key.as_str().starts_with("aa/"));
1118        assert!(key.as_str().ends_with(&hash));
1119    }
1120
1121    #[test]
1122    fn chunk_object_key_invalid_hash() {
1123        assert!(chunk_object_key("short").is_err());
1124    }
1125
1126    #[test]
1127    fn parse_stored_file_record_bytes_valid() {
1128        let json = r#"{"file_id":"test.txt","content_hash":"aabb","total_bytes":100,"chunk_size":10,"chunks":[]}"#;
1129        assert!(parse_stored_file_record_bytes(json.as_bytes()).is_ok());
1130    }
1131
1132    #[test]
1133    fn parse_stored_file_record_bytes_invalid_json() {
1134        assert!(parse_stored_file_record_bytes(b"not json").is_err());
1135    }
1136
1137    #[test]
1138    fn parse_stored_file_record_bytes_oversized() {
1139        let valid =
1140            r#"{"file_id":"test","content_hash":"aa","total_bytes":0,"chunk_size":0,"chunks":[]}"#;
1141        assert!(parse_stored_file_record_bytes(valid.as_bytes()).is_ok());
1142
1143        let oversized = vec![0u8; (MAX_LOCAL_RECORD_METADATA_BYTES + 1) as usize];
1144        assert!(parse_stored_file_record_bytes(&oversized).is_err());
1145    }
1146
1147    proptest::proptest! {
1148        #[test]
1149        fn proptest_validate_identifier_rejects_leading_slash(s in "[a-z]{1,100}") {
1150            let input = format!("/{s}");
1151            prop_assert!(validate_identifier(&input).is_err(), "leading slash should be rejected: {input:?}");
1152        }
1153
1154        #[test]
1155        fn proptest_validate_identifier_rejects_traversal(s in "[a-z]{1,50}") {
1156            let input = format!("{s}/../{s}");
1157            prop_assert!(validate_identifier(&input).is_err(), "traversal should be rejected: {input:?}");
1158        }
1159
1160        #[test]
1161        fn proptest_validate_identifier_rejects_backslash(s in "[a-z]{1,50}") {
1162            let input = format!("{s}\\{s}");
1163            prop_assert!(validate_identifier(&input).is_err(), "backslash should be rejected: {input:?}");
1164        }
1165
1166        #[test]
1167        fn proptest_validate_identifier_accepts_valid_names(segs in prop::collection::vec("[a-z]{1,20}", 1..3usize)) {
1168            let input = segs.join(".");
1169            let result = validate_identifier(&input);
1170            prop_assert!(result.is_ok(), "valid identifier should be accepted: {input:?}");
1171        }
1172
1173        #[test]
1174        fn proptest_validate_identifier_rejects_control_characters(segs in prop::collection::vec("[a-z]{1,20}", 1..3usize)) {
1175            let mut input = segs.join(".");
1176            input.push('\t');
1177            let result = validate_identifier(&input);
1178            prop_assert!(result.is_err(), "control characters should be rejected: {input:?}");
1179        }
1180    }
1181}
shardline_server_core/lib.rs

shardline_server_core/
lib.rs
