shardline_server/
rebuild.rs

1use std::path::PathBuf;
2
3use shardline_index::{LocalIndexStore, PostgresIndexStore, PostgresRecordStore};
4pub use shardline_rebuild::{
5    IndexRebuildIssueDetail, IndexRebuildReconstructionPlanDetail, IndexRebuildReport,
6    LocalIndexRebuildIssue, LocalIndexRebuildIssueKind, LocalIndexRebuildReport, RebuildError,
7    run_index_rebuild_with_stores,
8};
9use shardline_server_core::DEFAULT_SHARD_METADATA_LIMITS;
10
11use crate::{
12    ServerConfig, ServerError,
13    object_store::{ServerObjectStore, object_store_from_config},
14    postgres_backend::connect_postgres_metadata_pool,
15    record_store::LocalRecordStore,
16};
17
18impl From<RebuildError> for ServerError {
19    fn from(value: RebuildError) -> Self {
20        use crate::error::{IndexError, ObjectStoreError};
21        match value {
22            RebuildError::Io(e) => Self::Io(e),
23            RebuildError::Json(e) => Self::Json(e),
24            RebuildError::NumericConversion(e) => Self::NumericConversion(e),
25            RebuildError::InvalidContentHash => Self::InvalidContentHash,
26            RebuildError::InvalidFileId => Self::InvalidFileId,
27            RebuildError::Overflow => Self::Overflow,
28            RebuildError::ObjectPrefix(e) => Self::ObjectStore(ObjectStoreError::Prefix(e)),
29            RebuildError::LocalObjectStore(e) => Self::ObjectStore(ObjectStoreError::Local(e)),
30            RebuildError::S3ObjectStore(e) => Self::ObjectStore(ObjectStoreError::S3(e)),
31            RebuildError::XetAdapter(e) => Self::from(e),
32            RebuildError::IndexStore(e) => Self::Index(IndexError::Local(e)),
33            RebuildError::MemoryIndexStore(e) => Self::Index(IndexError::MemoryIndex(e)),
34            RebuildError::MemoryRecordStore(e) => Self::Index(IndexError::MemoryRecord(e)),
35            RebuildError::PostgresMetadata(e) => Self::Index(IndexError::PostgresMetadata(e)),
36            RebuildError::HashParse(e) => Self::HashParse(e),
37            RebuildError::StoredFileMetadataTooLarge {
38                observed_bytes,
39                maximum_bytes,
40            } => Self::StoredFileMetadataTooLarge {
41                observed_bytes,
42                maximum_bytes,
43            },
44        }
45    }
46}
47
48/// Rebuilds latest-record state against the configured metadata backend.
49///
50/// # Errors
51///
52/// Returns [`ServerError`] when version records cannot be scanned or latest records
53/// cannot be written or removed.
54pub async fn run_index_rebuild(config: ServerConfig) -> Result<IndexRebuildReport, ServerError> {
55    let object_store = object_store_from_config(&config)?;
56    if let Some(index_postgres_url) = config.index_postgres_url() {
57        let pool = connect_postgres_metadata_pool(index_postgres_url, 4)?;
58        let index_store = PostgresIndexStore::new(pool.clone());
59        let record_store = PostgresRecordStore::new(pool);
60        return run_index_rebuild_with_stores(
61            &record_store,
62            &index_store,
63            &object_store,
64            config.shard_metadata_limits(),
65        )
66        .await
67        .map_err(Into::into);
68    }
69
70    let index_store = LocalIndexStore::open(config.root_dir().to_path_buf());
71    let record_store = LocalRecordStore::open(config.root_dir().to_path_buf());
72    run_index_rebuild_with_stores(
73        &record_store,
74        &index_store,
75        &object_store,
76        config.shard_metadata_limits(),
77    )
78    .await
79    .map_err(Into::into)
80}
81
82/// Rebuilds local latest-record state from immutable version records.
83///
84/// The local metadata backend stores immutable version rows and derives visible latest
85/// rows in the same record store.
86///
87/// # Errors
88///
89/// Returns [`ServerError`] when the storage root cannot be traversed or when latest
90/// records cannot be written or removed.
91pub async fn run_local_index_rebuild(
92    root: PathBuf,
93) -> Result<LocalIndexRebuildReport, ServerError> {
94    let object_store = ServerObjectStore::local(root.join("chunks"))?;
95    let index_store = LocalIndexStore::open(root.clone());
96    let record_store = LocalRecordStore::open(root);
97    run_index_rebuild_with_stores(
98        &record_store,
99        &index_store,
100        &object_store,
101        DEFAULT_SHARD_METADATA_LIMITS,
102    )
103    .await
104    .map_err(Into::into)
105}
106
107#[cfg(test)]
108mod tests;
shardline_server/rebuild.rs

shardline_server/
rebuild.rs
