1use super::*;
2use crate::api::REDDB_FORMAT_VERSION;
3
4impl PhysicalMetadataFile {
5 pub fn from_state(
6 options: RedDBOptions,
7 catalog: CatalogSnapshot,
8 collection_roots: BTreeMap<String, u64>,
9 indexes: Vec<PhysicalIndexState>,
10 previous: Option<&PhysicalMetadataFile>,
11 ) -> Self {
12 let now = unix_ms_now();
13 let mut manifest = SchemaManifest::now(options, catalog.total_collections);
14 if let Some(previous) = previous {
15 manifest.created_at_unix_ms = previous.manifest.created_at_unix_ms;
16 }
17 manifest.updated_at_unix_ms = now;
18
19 let sequence = previous
20 .map(|previous| previous.superblock.sequence.saturating_add(1))
21 .unwrap_or(1);
22 let mut manifest_events = previous
23 .map(|previous| previous.manifest_events.clone())
24 .unwrap_or_default();
25 manifest_events.extend(build_manifest_events(
26 previous.map(|previous| &previous.superblock.collection_roots),
27 &collection_roots,
28 sequence,
29 ));
30 trim_manifest_history(&mut manifest_events);
31
32 let superblock = SuperblockHeader {
33 format_version: REDDB_FORMAT_VERSION,
34 sequence,
35 copies: DEFAULT_SUPERBLOCK_COPIES,
36 manifest: previous
37 .map(|previous| previous.superblock.manifest.clone())
38 .unwrap_or_default(),
39 free_set: previous
40 .map(|previous| previous.superblock.free_set)
41 .unwrap_or_default(),
42 collection_roots,
43 };
44
45 let mut snapshots = previous
46 .map(|previous| previous.snapshots.clone())
47 .unwrap_or_default();
48 snapshots.push(SnapshotDescriptor {
49 snapshot_id: sequence,
50 created_at_unix_ms: now,
51 superblock_sequence: sequence,
52 collection_count: catalog.total_collections,
53 total_entities: catalog.total_entities,
54 });
55 trim_snapshot_history(&mut snapshots, manifest.options.snapshot_retention);
56
57 Self {
58 protocol_version: PHYSICAL_METADATA_PROTOCOL_VERSION.to_string(),
59 generated_at_unix_ms: now,
60 last_loaded_from: previous.and_then(|previous| previous.last_loaded_from.clone()),
61 last_healed_at_unix_ms: previous.and_then(|previous| previous.last_healed_at_unix_ms),
62 manifest,
63 catalog,
64 manifest_events,
65 indexes,
66 graph_projections: previous
67 .map(|previous| previous.graph_projections.clone())
68 .unwrap_or_default(),
69 analytics_jobs: previous
70 .map(|previous| previous.analytics_jobs.clone())
71 .unwrap_or_default(),
72 tree_definitions: previous
73 .map(|previous| previous.tree_definitions.clone())
74 .unwrap_or_default(),
75 collection_ttl_defaults_ms: previous
76 .map(|previous| previous.collection_ttl_defaults_ms.clone())
77 .unwrap_or_default(),
78 collection_contracts: previous
79 .map(|previous| previous.collection_contracts.clone())
80 .unwrap_or_default(),
81 hypertables: previous
82 .map(|previous| previous.hypertables.clone())
83 .unwrap_or_default(),
84 exports: previous
85 .map(|previous| previous.exports.clone())
86 .unwrap_or_default(),
87 superblock,
88 snapshots,
89 }
90 }
91
92 pub fn metadata_path_for(data_path: &Path) -> PathBuf {
93 reddb_file::layout::physical_metadata_json_path(data_path)
94 }
95
96 pub fn metadata_binary_path_for(data_path: &Path) -> PathBuf {
97 reddb_file::layout::physical_metadata_binary_path(data_path)
98 }
99
100 pub fn metadata_journal_path_for(data_path: &Path, sequence: u64) -> PathBuf {
101 reddb_file::layout::physical_metadata_journal_path(data_path, sequence)
102 }
103
104 pub fn export_data_path_for(data_path: &Path, name: &str) -> PathBuf {
105 reddb_file::layout::physical_export_data_path(data_path, name)
106 }
107
108 pub fn load_for_data_path(data_path: &Path) -> io::Result<Self> {
109 Self::load_for_data_path_with_source(data_path).map(|(metadata, _)| metadata)
110 }
111
112 pub fn load_for_data_path_with_source(
113 data_path: &Path,
114 ) -> io::Result<(Self, PhysicalMetadataSource)> {
115 let binary_path = Self::metadata_binary_path_for(data_path);
116 if binary_path.exists() {
117 match Self::load_from_binary_path(&binary_path) {
118 Ok(metadata) => {
119 return Ok((metadata, PhysicalMetadataSource::Binary));
120 }
121 Err(error) if is_collection_contract_decode_error(&error) => return Err(error),
122 Err(_) => {
123 let mut journal_paths = Self::journal_paths_for_data_path(data_path)?;
124 journal_paths.reverse();
125 for journal_path in journal_paths {
126 match Self::load_from_binary_path(&journal_path) {
127 Ok(metadata) => {
128 let healed =
129 metadata.mark_recovery(PhysicalMetadataSource::BinaryJournal);
130 let _ = healed.heal_primary_metadata_for_data_path(data_path);
131 return Ok((healed, PhysicalMetadataSource::BinaryJournal));
132 }
133 Err(error) if is_collection_contract_decode_error(&error) => {
134 return Err(error)
135 }
136 Err(_) => {}
137 }
138 }
139 }
140 }
141 }
142 Self::load_from_path(&Self::metadata_path_for(data_path)).map(|metadata| {
143 let healed = metadata.mark_recovery(PhysicalMetadataSource::Json);
144 let _ = healed.heal_primary_metadata_for_data_path(data_path);
145 (healed, PhysicalMetadataSource::Json)
146 })
147 }
148
149 pub fn save_for_data_path(&self, data_path: &Path) -> io::Result<PathBuf> {
150 let binary_path = Self::metadata_binary_path_for(data_path);
151 if binary_path.exists() && super::seqn_journal_enabled() {
152 let sequence = Self::load_from_binary_path(&binary_path)
153 .map(|metadata| metadata.superblock.sequence)
154 .unwrap_or(self.superblock.sequence);
155 let _ = reddb_file::copy_physical_metadata_binary_to_journal(
156 data_path,
157 &binary_path,
158 sequence,
159 );
160 }
161 self.save_to_binary_path(&binary_path)?;
162 self.prune_journal_for_data_path(data_path)?;
163 if super::meta_json_sidecar_enabled() {
164 let json_path = Self::metadata_path_for(data_path);
165 self.save_to_path(&json_path)?;
166 }
167 Ok(binary_path)
168 }
169
170 pub fn load_from_path(path: &Path) -> io::Result<Self> {
171 let text = reddb_file::read_physical_metadata_document(path).map_err(|err| {
172 io::Error::new(
173 io::ErrorKind::InvalidData,
174 format!("invalid physical metadata JSON: {err}"),
175 )
176 })?;
177 Self::from_document_json(&text, "invalid physical metadata JSON")
178 }
179
180 pub fn save_to_path(&self, path: &Path) -> io::Result<()> {
181 let text = self.to_document_json(true)?;
182 reddb_file::write_physical_metadata_json_document(path, &text)
183 .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
184 }
185
186 pub fn load_from_binary_path(path: &Path) -> io::Result<Self> {
187 let text = reddb_file::read_physical_metadata_document(path).map_err(|err| {
188 io::Error::new(
189 io::ErrorKind::InvalidData,
190 format!("invalid physical metadata binary: {err}"),
191 )
192 })?;
193 Self::from_document_json(&text, "invalid physical metadata binary")
194 }
195
196 pub fn save_to_binary_path(&self, path: &Path) -> io::Result<()> {
197 let text = self.to_document_json(false)?;
198 reddb_file::write_physical_metadata_binary_document(path, &text)
199 .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
200 }
201
202 pub fn journal_paths_for_data_path(data_path: &Path) -> io::Result<Vec<PathBuf>> {
203 reddb_file::list_physical_metadata_journal_paths(data_path)
204 .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
205 }
206
207 pub fn prune_journal_for_data_path(&self, data_path: &Path) -> io::Result<()> {
208 let retention = super::seqn_journal_retention();
209 reddb_file::prune_physical_metadata_journal_paths(data_path, retention)
210 .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
211 }
212
213 pub fn heal_primary_metadata_for_data_path(&self, data_path: &Path) -> io::Result<()> {
214 let binary_path = Self::metadata_binary_path_for(data_path);
215 self.save_to_binary_path(&binary_path)?;
216 if super::meta_json_sidecar_enabled() {
217 let json_path = Self::metadata_path_for(data_path);
218 self.save_to_path(&json_path)?;
219 }
220 Ok(())
221 }
222
223 pub fn mark_recovery(&self, source: PhysicalMetadataSource) -> Self {
224 let mut metadata = self.clone();
225 metadata.last_loaded_from = Some(source.as_str().to_string());
226 metadata.last_healed_at_unix_ms = Some(unix_ms_now());
227 metadata
228 }
229
230 pub fn to_json_value(&self) -> JsonValue {
231 let json = self
232 .to_document_json(false)
233 .expect("physical metadata must encode as JSON");
234 crate::json::from_str::<JsonValue>(&json)
235 .expect("reddb-file emitted JSON the server can parse")
236 }
237
238 pub fn from_json_value(value: &JsonValue) -> io::Result<Self> {
239 Self::from_document_json(&value.to_string_compact(), "invalid physical metadata JSON")
240 }
241
242 fn to_document_json(&self, pretty: bool) -> io::Result<String> {
243 reddb_file::encode_physical_metadata_document_root_json(
244 &self.to_document_envelope(),
245 pretty,
246 )
247 .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err.to_string()))
248 }
249
250 fn from_document_json(text: &str, context: &'static str) -> io::Result<Self> {
251 let envelope =
252 reddb_file::decode_physical_metadata_document_root_json(text).map_err(|err| {
253 io::Error::new(io::ErrorKind::InvalidData, format!("{context}: {err}"))
254 })?;
255 Self::from_document_envelope(envelope)
256 }
257
258 fn to_document_envelope(&self) -> reddb_file::PhysicalMetadataDocumentEnvelope {
259 reddb_file::PhysicalMetadataDocumentEnvelope {
260 protocol_version: self.protocol_version.clone(),
261 generated_at_unix_ms: self.generated_at_unix_ms,
262 last_loaded_from: self.last_loaded_from.clone(),
263 last_healed_at_unix_ms: self.last_healed_at_unix_ms,
264 manifest_json: manifest_to_json(&self.manifest).to_string_compact(),
265 catalog_json: catalog_to_json(&self.catalog).to_string_compact(),
266 manifest_events_json: self
267 .manifest_events
268 .iter()
269 .map(|event| manifest_event_to_json(event).to_string_compact())
270 .collect(),
271 indexes_json: self
272 .indexes
273 .iter()
274 .map(|index| index_state_to_json(index).to_string_compact())
275 .collect(),
276 graph_projections_json: self
277 .graph_projections
278 .iter()
279 .map(|projection| graph_projection_to_json(projection).to_string_compact())
280 .collect(),
281 analytics_jobs_json: self
282 .analytics_jobs
283 .iter()
284 .map(|job| analytics_job_to_json(job).to_string_compact())
285 .collect(),
286 tree_definitions_json: self
287 .tree_definitions
288 .iter()
289 .map(|definition| tree_definition_to_json(definition).to_string_compact())
290 .collect(),
291 collection_ttl_defaults_ms: self.collection_ttl_defaults_ms.clone(),
292 collection_contracts_json: self
293 .collection_contracts
294 .iter()
295 .map(|contract| collection_contract_to_json(contract).to_string_compact())
296 .collect(),
297 hypertables_json: self
298 .hypertables
299 .iter()
300 .map(|hypertable| hypertable_to_json(hypertable).to_string_compact())
301 .collect(),
302 exports_json: self
303 .exports
304 .iter()
305 .map(|export| export_descriptor_to_json(export).to_string_compact())
306 .collect(),
307 superblock_json: superblock_to_json(&self.superblock).to_string_compact(),
308 snapshots_json: self
309 .snapshots
310 .iter()
311 .map(|snapshot| snapshot_descriptor_to_json(snapshot).to_string_compact())
312 .collect(),
313 }
314 }
315
316 fn from_document_envelope(
317 envelope: reddb_file::PhysicalMetadataDocumentEnvelope,
318 ) -> io::Result<Self> {
319 Ok(Self {
320 protocol_version: envelope.protocol_version,
321 generated_at_unix_ms: envelope.generated_at_unix_ms,
322 last_loaded_from: envelope.last_loaded_from,
323 last_healed_at_unix_ms: envelope.last_healed_at_unix_ms,
324 manifest: manifest_from_json(&parse_document_fragment(
325 &envelope.manifest_json,
326 "manifest",
327 )?)?,
328 catalog: catalog_from_json(&parse_document_fragment(
329 &envelope.catalog_json,
330 "catalog",
331 )?)?,
332 manifest_events: parse_document_fragments(
333 &envelope.manifest_events_json,
334 "manifest_events",
335 manifest_event_from_json,
336 )?,
337 indexes: parse_document_fragments(
338 &envelope.indexes_json,
339 "indexes",
340 index_state_from_json,
341 )?,
342 graph_projections: parse_document_fragments(
343 &envelope.graph_projections_json,
344 "graph_projections",
345 graph_projection_from_json,
346 )?,
347 analytics_jobs: parse_document_fragments(
348 &envelope.analytics_jobs_json,
349 "analytics_jobs",
350 analytics_job_from_json,
351 )?,
352 tree_definitions: parse_document_fragments(
353 &envelope.tree_definitions_json,
354 "tree_definitions",
355 tree_definition_from_json,
356 )?,
357 collection_ttl_defaults_ms: envelope.collection_ttl_defaults_ms,
358 collection_contracts: parse_document_fragments(
359 &envelope.collection_contracts_json,
360 "collection_contracts",
361 collection_contract_from_json,
362 )?,
363 hypertables: parse_document_fragments(
364 &envelope.hypertables_json,
365 "hypertables",
366 hypertable_from_json,
367 )?,
368 exports: parse_document_fragments(
369 &envelope.exports_json,
370 "exports",
371 export_descriptor_from_json,
372 )?,
373 superblock: superblock_from_json(&parse_document_fragment(
374 &envelope.superblock_json,
375 "superblock",
376 )?)?,
377 snapshots: parse_document_fragments(
378 &envelope.snapshots_json,
379 "snapshots",
380 snapshot_descriptor_from_json,
381 )?,
382 })
383 }
384}
385
386fn parse_document_fragment(json: &str, field: &'static str) -> io::Result<JsonValue> {
387 crate::json::from_str::<JsonValue>(json)
388 .map_err(|err| invalid_data(format!("invalid physical metadata field '{field}': {err}")))
389}
390
391fn parse_document_fragments<T>(
392 fragments: &[String],
393 field: &'static str,
394 decode: fn(&JsonValue) -> io::Result<T>,
395) -> io::Result<Vec<T>> {
396 fragments
397 .iter()
398 .map(|fragment| decode(&parse_document_fragment(fragment, field)?))
399 .collect()
400}