shardline-server 1.0.0

HTTP server boundary, runtime, and operator workflows for Shardline.
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

use shardline_fsck::run_fsck_with_stores;
use shardline_index::{LocalIndexStore, LocalRecordStore, PostgresIndexStore, PostgresRecordStore};

use crate::{
    ServerConfig, ServerError,
    error::{IndexError, ObjectStoreError},
    object_store::object_store_from_config,
    postgres_backend::connect_postgres_metadata_pool,
};

pub use shardline_fsck::{
    FsckError, FsckIssueDetail, FsckIssueKind, FsckReconstructionPlanDetail, LocalFsckIssue,
    LocalFsckIssueKind, LocalFsckReport, ProviderRepositoryStateTimestampField, run_local_fsck,
};

/// Runs integrity checks against the configured metadata backend and local chunk storage.
///
/// # Errors
///
/// Returns [`ServerError`] when the storage root cannot be traversed, metadata cannot be
/// queried, or chunk/record bytes cannot be read.
pub async fn run_fsck(config: ServerConfig) -> Result<LocalFsckReport, ServerError> {
    let object_root = config.root_dir().join("chunks");
    let object_store = object_store_from_config(&config)?;
    if let Some(index_postgres_url) = config.index_postgres_url() {
        let pool = connect_postgres_metadata_pool(index_postgres_url, 4)?;
        let index_store = PostgresIndexStore::new(pool.clone());
        let record_store = PostgresRecordStore::new(pool);
        return run_fsck_with_stores(
            &record_store,
            &index_store,
            &object_root,
            &object_store,
            config.shard_metadata_limits(),
        )
        .await
        .map_err(ServerError::from);
    }

    let index_store = LocalIndexStore::open(config.root_dir().to_path_buf());
    let record_store = LocalRecordStore::open(config.root_dir().to_path_buf());
    run_fsck_with_stores(
        &record_store,
        &index_store,
        &object_root,
        &object_store,
        config.shard_metadata_limits(),
    )
    .await
    .map_err(ServerError::from)
}

impl From<FsckError> for ServerError {
    fn from(value: FsckError) -> Self {
        match value {
            FsckError::Io(e) => Self::Io(e),
            FsckError::Json(e) => Self::Json(e),
            FsckError::NumericConversion(e) => Self::NumericConversion(e),
            FsckError::Overflow => Self::Overflow,
            FsckError::LocalObjectStore(e) => Self::ObjectStore(ObjectStoreError::Local(e)),
            FsckError::S3ObjectStore(e) => Self::ObjectStore(ObjectStoreError::S3(e)),
            FsckError::ObjectStore(e) => match e {
                shardline_server_core::ServerObjectStoreError::NotFound => Self::NotFound,
                shardline_server_core::ServerObjectStoreError::Overflow => Self::Overflow,
                shardline_server_core::ServerObjectStoreError::InvalidContentHash => {
                    Self::InvalidContentHash
                }
                shardline_server_core::ServerObjectStoreError::StoredObjectLengthMismatch => {
                    Self::ObjectStore(ObjectStoreError::StoredLengthMismatch)
                }
                shardline_server_core::ServerObjectStoreError::Local(e) => {
                    Self::ObjectStore(ObjectStoreError::Local(e))
                }
                shardline_server_core::ServerObjectStoreError::S3(e) => {
                    Self::ObjectStore(ObjectStoreError::S3(e))
                }
                shardline_server_core::ServerObjectStoreError::Io(e) => Self::Io(e),
                shardline_server_core::ServerObjectStoreError::NumericConversion(e) => {
                    Self::NumericConversion(e)
                }
            },
            FsckError::XetAdapter(e) => Self::from(e),
            FsckError::LocalIndexStore(e) => Self::Index(IndexError::Local(e)),
            FsckError::MemoryIndexStore(e) => Self::Index(IndexError::MemoryIndex(e)),
            FsckError::MemoryRecordStore(e) => Self::Index(IndexError::MemoryRecord(e)),
            FsckError::PostgresMetadata(e) => Self::Index(IndexError::PostgresMetadata(e)),
            FsckError::StoredFileMetadataTooLarge {
                observed_bytes,
                maximum_bytes,
            } => Self::StoredFileMetadataTooLarge {
                observed_bytes,
                maximum_bytes,
            },
        }
    }
}