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/// Tests for the `From<RebuildError> for ServerError` conversion.
108///
109/// These are pure unit tests that can live in the main module without
110/// the integration-test infrastructure required by `rebuild/tests.rs`.
111#[cfg(test)]
112mod from_impl_tests {
113    use super::*;
114    use crate::error::{IndexError, ObjectStoreError};
115
116    #[test]
117    fn from_rebuild_error_io() {
118        let err = RebuildError::Io(std::io::Error::other("io"));
119        let server_err: ServerError = err.into();
120        assert!(matches!(server_err, ServerError::Io(_)));
121    }
122
123    #[test]
124    fn from_rebuild_error_json() {
125        let json_err = serde_json::from_str::<()>("").unwrap_err();
126        let err = RebuildError::Json(json_err);
127        let server_err: ServerError = err.into();
128        assert!(matches!(server_err, ServerError::Json(_)));
129    }
130
131    #[test]
132    fn from_rebuild_error_numeric_conversion() {
133        let num_err = u64::try_from(-1i32).unwrap_err();
134        let err = RebuildError::NumericConversion(num_err);
135        let server_err: ServerError = err.into();
136        assert!(matches!(server_err, ServerError::NumericConversion(_)));
137    }
138
139    #[test]
140    fn from_rebuild_error_invalid_content_hash() {
141        let err = RebuildError::InvalidContentHash;
142        let server_err: ServerError = err.into();
143        assert!(matches!(server_err, ServerError::InvalidContentHash));
144    }
145
146    #[test]
147    fn from_rebuild_error_invalid_file_id() {
148        let err = RebuildError::InvalidFileId;
149        let server_err: ServerError = err.into();
150        assert!(matches!(server_err, ServerError::InvalidFileId));
151    }
152
153    #[test]
154    fn from_rebuild_error_overflow() {
155        let err = RebuildError::Overflow;
156        let server_err: ServerError = err.into();
157        assert!(matches!(server_err, ServerError::Overflow));
158    }
159
160    #[test]
161    fn from_rebuild_error_object_prefix() {
162        let prefix_err = shardline_storage::ObjectPrefixError::UnsafePath;
163        let err = RebuildError::ObjectPrefix(prefix_err);
164        let server_err: ServerError = err.into();
165        assert!(matches!(
166            server_err,
167            ServerError::ObjectStore(ObjectStoreError::Prefix(_))
168        ));
169    }
170
171    #[test]
172    fn from_rebuild_error_local_object_store() {
173        let local_err =
174            shardline_storage::LocalObjectStoreError::Io(std::io::Error::other("local"));
175        let err = RebuildError::LocalObjectStore(local_err);
176        let server_err: ServerError = err.into();
177        assert!(matches!(
178            server_err,
179            ServerError::ObjectStore(ObjectStoreError::Local(_))
180        ));
181    }
182
183    #[test]
184    fn from_rebuild_error_s3_object_store() {
185        let s3_err = shardline_storage::S3ObjectStoreError::IncompleteCredentials;
186        let err = RebuildError::S3ObjectStore(s3_err);
187        let server_err: ServerError = err.into();
188        assert!(matches!(
189            server_err,
190            ServerError::ObjectStore(ObjectStoreError::S3(_))
191        ));
192    }
193
194    #[test]
195    fn from_rebuild_error_xet_adapter() {
196        let xet_err = shardline_xet_adapter::XetAdapterError::Overflow;
197        let err = RebuildError::XetAdapter(xet_err);
198        let server_err: ServerError = err.into();
199        assert!(matches!(server_err, ServerError::Overflow));
200    }
201
202    #[test]
203    fn from_rebuild_error_index_store() {
204        let io_err = std::io::Error::other("index io");
205        let index_err = shardline_index::LocalIndexStoreError::Io(io_err);
206        let err = RebuildError::IndexStore(index_err);
207        let server_err: ServerError = err.into();
208        assert!(matches!(
209            server_err,
210            ServerError::Index(IndexError::Local(_))
211        ));
212    }
213
214    #[test]
215    fn from_rebuild_error_memory_index_store() {
216        let mem_err = shardline_index::MemoryIndexStoreError::LockPoisoned("test".to_owned());
217        let err = RebuildError::MemoryIndexStore(mem_err);
218        let server_err: ServerError = err.into();
219        assert!(matches!(
220            server_err,
221            ServerError::Index(IndexError::MemoryIndex(_))
222        ));
223    }
224
225    #[test]
226    fn from_rebuild_error_memory_record_store() {
227        let mem_err = shardline_index::MemoryRecordStoreError::LockPoisoned("test".to_owned());
228        let err = RebuildError::MemoryRecordStore(mem_err);
229        let server_err: ServerError = err.into();
230        assert!(matches!(
231            server_err,
232            ServerError::Index(IndexError::MemoryRecord(_))
233        ));
234    }
235
236    #[test]
237    fn from_rebuild_error_postgres_metadata() {
238        let pg_err = shardline_index::PostgresMetadataStoreError::HashParse(
239            shardline_protocol::HashParseError::InvalidLength,
240        );
241        let err = RebuildError::PostgresMetadata(pg_err);
242        let server_err: ServerError = err.into();
243        assert!(matches!(
244            server_err,
245            ServerError::Index(IndexError::PostgresMetadata(_))
246        ));
247    }
248
249    #[test]
250    fn from_rebuild_error_hash_parse() {
251        let hash_err = shardline_protocol::HashParseError::InvalidLength;
252        let err = RebuildError::HashParse(hash_err);
253        let server_err: ServerError = err.into();
254        assert!(matches!(server_err, ServerError::HashParse(_)));
255    }
256
257    #[test]
258    fn from_rebuild_error_stored_file_metadata_too_large() {
259        let err = RebuildError::StoredFileMetadataTooLarge {
260            observed_bytes: 5000,
261            maximum_bytes: 1000,
262        };
263        let server_err: ServerError = err.into();
264        assert!(matches!(
265            server_err,
266            ServerError::StoredFileMetadataTooLarge {
267                observed_bytes: 5000,
268                maximum_bytes: 1000,
269            }
270        ));
271    }
272}
273
274#[cfg(test)]
275mod tests;
shardline_server/rebuild.rs

shardline_server/
rebuild.rs
