1use super::*;
2
3const SINGLE_FILE_FORK_GUIDANCE: &str = "store fork is not available directly on embedded \
4single-file stores. Export the .rdb through the operational-directory layout first, then fork \
5that operational store. See docs/engine/operational-storage-profiles.md#forking-an-embedded-single-file-store";
6
7impl RedDBRuntime {
8 pub fn fork_store(&self, name: &str) -> RedDBResult<reddb_file::ForkInfo> {
9 let name = name.trim();
10 if name.is_empty() {
11 return Err(RedDBError::InvalidOperation(
12 "store fork name cannot be empty".to_string(),
13 ));
14 }
15 if self.inner.embedded_single_file {
16 return Err(RedDBError::InvalidOperation(
17 SINGLE_FILE_FORK_GUIDANCE.to_string(),
18 ));
19 }
20 let data_path = self.inner.db.options().data_path.as_ref().ok_or_else(|| {
21 RedDBError::InvalidOperation("store fork requires a data path".into())
22 })?;
23
24 self.flush()?;
25 let fork_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
26 let manifest =
27 crate::storage::operational_manifest::OperationalManifest::for_db_path(data_path);
28 manifest
29 .create_fork(name, fork_lsn)
30 .map_err(|err| RedDBError::InvalidOperation(err.to_string()))?;
31 manifest
32 .list_forks()
33 .map_err(|err| RedDBError::InvalidOperation(err.to_string()))?
34 .into_iter()
35 .find(|fork| fork.name == name)
36 .ok_or_else(|| RedDBError::Internal(format!("created store fork {name} is missing")))
37 }
38
39 pub fn detach_fork_store(&self, name: &str) -> RedDBResult<bool> {
40 let name = name.trim();
41 if name.is_empty() {
42 return Err(RedDBError::InvalidOperation(
43 "store fork name cannot be empty".to_string(),
44 ));
45 }
46 let data_path = self.inner.db.options().data_path.as_ref().ok_or_else(|| {
47 RedDBError::InvalidOperation("store fork requires a data path".into())
48 })?;
49
50 let manifest =
51 crate::storage::operational_manifest::OperationalManifest::for_db_path(data_path);
52 manifest
53 .detach_fork(name)
54 .map(|detached| detached.is_some())
55 .map_err(|err| RedDBError::InvalidOperation(err.to_string()))
56 }
57
58 pub fn replica_relay_manifest_path(&self, replica_id: &str) -> Option<std::path::PathBuf> {
59 let plan = self.primary_replica_file_plan()?;
60 Some(plan.relay_manifest_path(replica_id))
61 }
62
63 pub fn primary_replica_file_plan(&self) -> Option<reddb_file::PrimaryReplicaFilePlan> {
64 self.primary_replica_file_plan_result().ok().flatten()
65 }
66
67 fn primary_replica_root(&self) -> Option<std::path::PathBuf> {
68 let data_path = self.inner.db.options().data_path.as_ref()?;
69 Some(crate::replication::primary::PrimaryReplication::primary_replica_root_for(data_path))
70 }
71
72 pub fn primary_replica_file_plan_result(
73 &self,
74 ) -> RedDBResult<Option<reddb_file::PrimaryReplicaFilePlan>> {
75 let Some(root) = self.primary_replica_root() else {
76 return Ok(None);
77 };
78 let timeline = self.primary_replica_current_timeline(&root)?;
79 Ok(Some(reddb_file::PrimaryReplicaFilePlan::new(
80 root, timeline,
81 )))
82 }
83
84 fn primary_replica_current_timeline(
85 &self,
86 root: &std::path::Path,
87 ) -> RedDBResult<reddb_file::TimelineId> {
88 let path = reddb_file::PrimaryReplicaFilePlan::new(root, reddb_file::TimelineId::initial())
89 .timeline_history_path();
90 match reddb_file::TimelineHistory::read_from_path(&path) {
91 Ok(history) => Ok(history
92 .current()
93 .unwrap_or_else(reddb_file::TimelineId::initial)),
94 Err(reddb_file::RdbFileError::Io(err))
95 if err.kind() == std::io::ErrorKind::NotFound =>
96 {
97 Ok(reddb_file::TimelineId::initial())
98 }
99 Err(err) => Err(RedDBError::Internal(err.to_string())),
100 }
101 }
102
103 pub fn create_primary_replica_basebackup(
104 &self,
105 chunk_bytes: usize,
106 ) -> RedDBResult<Option<reddb_file::PrimaryReplicaBaseBackupManifest>> {
107 let Some(plan) = self.primary_replica_file_plan_result()? else {
108 return Ok(None);
109 };
110 self.flush()?;
111 let checkpoint_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
112 let snapshot = self.inner.db.store().to_binary_dump_bytes();
113 let backup = reddb_file::BaseBackupPlan::new(plan.timeline, 0, checkpoint_lsn);
114 let manifest = plan
115 .write_basebackup_snapshot_parts(backup, &snapshot, chunk_bytes)
116 .map_err(|err| RedDBError::Internal(err.to_string()))?;
117 manifest
118 .write_to_path(plan.basebackup_path(&backup))
119 .map_err(|err| RedDBError::Internal(err.to_string()))?;
120 Ok(Some(manifest))
121 }
122
123 pub fn materialize_primary_replica_basebackup_snapshot(
124 &self,
125 manifest: &reddb_file::PrimaryReplicaBaseBackupManifest,
126 parts_root: impl AsRef<std::path::Path>,
127 destination: impl AsRef<std::path::Path>,
128 ) -> RedDBResult<u64> {
129 let snapshot = manifest
130 .read_snapshot_parts(parts_root)
131 .map_err(|err| RedDBError::Internal(err.to_string()))?;
132 let loaded = crate::storage::unified::UnifiedStore::load_from_bytes_with_config(
133 &snapshot,
134 crate::storage::unified::UnifiedStoreConfig::default(),
135 )
136 .map_err(|err| RedDBError::Internal(format!("validate basebackup snapshot: {err}")))?;
137 loaded.set_config_tree(
138 "red.replication",
139 &crate::json!({
140 "last_applied_lsn": manifest.checkpoint_lsn,
141 "state": "healthy",
142 "last_error": "",
143 }),
144 );
145 let snapshot = loaded.to_binary_dump_bytes();
146 let destination = destination.as_ref();
147 if let Some(parent) = destination.parent() {
148 std::fs::create_dir_all(parent)?;
149 }
150 crate::storage::EmbeddedRdbArtifact::create_with_snapshot(destination, &snapshot)?;
151 Ok(manifest.checkpoint_lsn)
152 }
153
154 pub fn replica_rebootstrap_staging_root(&self) -> Option<std::path::PathBuf> {
155 let data_path = self.inner.db.options().data_path.as_ref()?;
156 Some(reddb_file::layout::rebootstrap_staging_root(data_path))
157 }
158
159 pub fn replica_rebootstrap_pending_path(&self) -> Option<std::path::PathBuf> {
160 let data_path = self.inner.db.options().data_path.as_ref()?;
161 Some(reddb_file::layout::rebootstrap_pending_path(data_path))
162 }
163
164 pub(crate) async fn stage_primary_replica_rebootstrap_from_snapshot(
165 &self,
166 client: &mut crate::grpc::proto::red_db_client::RedDbClient<tonic::transport::Channel>,
167 chunk_bytes: usize,
168 ) -> RedDBResult<Option<u64>> {
169 let Some(parts_root) = self.replica_rebootstrap_staging_root() else {
170 return Ok(None);
171 };
172 let Some(pending_path) = self.replica_rebootstrap_pending_path() else {
173 return Ok(None);
174 };
175 let data_path = self
176 .inner
177 .db
178 .options()
179 .data_path
180 .as_ref()
181 .ok_or_else(|| RedDBError::Internal("replica data path unavailable".into()))?;
182 let intent_log_path = reddb_file::layout::rebootstrap_intent_log_path(data_path);
183 let intent_log = crate::telemetry::admin_intent_log::AdminIntentLog::open(intent_log_path)
184 .map_err(|err| RedDBError::Internal(err.to_string()))?;
185 let replica_id = self.resolve_replica_id();
186 let bootstrapper = crate::replication::replica::ReplicaBootstrapper::new(replica_id);
187 let source_lsn = self.config_u64("red.replication.last_applied_lsn", 0);
188 let _ = std::fs::remove_file(&pending_path);
189
190 let chunk_bytes = chunk_bytes.max(1);
191 let (resume, mut bootstrap_handle) = match bootstrapper.resume(&intent_log) {
192 Some((resume, handle)) => (Some(resume), handle),
193 None => (
194 None,
195 bootstrapper
196 .begin(&intent_log, source_lsn, 0)
197 .map_err(|err| RedDBError::Internal(err.to_string()))?,
198 ),
199 };
200 let mut token: Option<String> = resume
201 .as_ref()
202 .and_then(|resume| resume.snapshot_token.clone());
203 let mut manifest: Option<reddb_file::PrimaryReplicaBaseBackupManifest> = None;
204 let mut written = std::collections::BTreeSet::new();
205 let mut offset = resume
206 .as_ref()
207 .map(|resume| resume.snapshot_offset)
208 .unwrap_or(0);
209
210 loop {
211 let mut request = tonic::Request::new(crate::grpc::proto::Empty {});
212 request.metadata_mut().insert(
213 "x-reddb-snapshot-max-bytes",
214 chunk_bytes
215 .to_string()
216 .parse()
217 .map_err(|err| RedDBError::Internal(format!("snapshot max bytes: {err}")))?,
218 );
219 request.metadata_mut().insert(
220 "x-reddb-snapshot-offset",
221 offset
222 .to_string()
223 .parse()
224 .map_err(|err| RedDBError::Internal(format!("snapshot offset: {err}")))?,
225 );
226 if let Some(token) = &token {
227 request.metadata_mut().insert(
228 "x-reddb-snapshot-token",
229 token.parse().map_err(|err| {
230 RedDBError::Internal(format!("snapshot token metadata: {err}"))
231 })?,
232 );
233 }
234
235 let response = client
236 .replication_snapshot(request)
237 .await
238 .map_err(|err| RedDBError::Internal(format!("replication snapshot: {err}")))?;
239 let payload = reddb_wire::replication::BaseBackupChunk::decode_json(
240 response.into_inner().payload.as_bytes(),
241 )
242 .map_err(|err| RedDBError::Internal(format!("parse replication snapshot: {err}")))?;
243 if token.is_none() {
244 token = payload.snapshot_token.clone();
245 }
246 let staged =
247 crate::replication::replica::stage_basebackup_snapshot_chunk(&payload, &parts_root)
248 .map_err(|err| RedDBError::Internal(err.to_string()))?
249 .ok_or_else(|| {
250 RedDBError::Internal(
251 "replication snapshot did not include basebackup payload".into(),
252 )
253 })?;
254 if let Some(existing) = &manifest {
255 if existing != &staged.manifest {
256 return Err(RedDBError::Internal(
257 "replication snapshot basebackup manifest changed while downloading".into(),
258 ));
259 }
260 } else {
261 manifest = Some(staged.manifest.clone());
262 written.extend(
263 crate::replication::replica::recover_staged_basebackup_chunks(
264 &staged.manifest,
265 &parts_root,
266 )
267 .map_err(|err| RedDBError::Internal(err.to_string()))?,
268 );
269 }
270 if let Some(ordinal) = staged.chunk_ordinal {
271 written.insert(ordinal);
272 }
273 let current = manifest
274 .as_ref()
275 .expect("manifest set after staging basebackup chunk");
276 if current
277 .chunks
278 .iter()
279 .all(|chunk| written.contains(&chunk.ordinal))
280 {
281 current
282 .verify_snapshot_parts(&parts_root)
283 .map_err(|err| RedDBError::Internal(err.to_string()))?;
284 let checkpoint_lsn = self.materialize_primary_replica_basebackup_snapshot(
285 current,
286 &parts_root,
287 &pending_path,
288 )?;
289 self.inner.db.store().set_config_tree(
290 "red.replication",
291 &crate::json!({
292 "state": "rebootstrap_ready",
293 "rebootstrap_pending_path": pending_path.display().to_string(),
294 "rebootstrap_checkpoint_lsn": checkpoint_lsn,
295 "rebootstrap_timeline": current.timeline.0,
296 }),
297 );
298 let data_path =
299 self.inner.db.options().data_path.as_ref().ok_or_else(|| {
300 RedDBError::Internal("replica data path unavailable".into())
301 })?;
302 reddb_file::write_rebootstrap_ready_marker(
303 data_path,
304 &reddb_file::ReplicaRebootstrapReadyMarker {
305 pending_path: pending_path.clone(),
306 checkpoint_lsn,
307 timeline: current.timeline,
308 },
309 )
310 .map_err(|err| RedDBError::Internal(err.to_string()))?;
311 bootstrap_handle
312 .complete(current.chunks.len() as u64, 0)
313 .map_err(|err| RedDBError::Internal(err.to_string()))?;
314 return Ok(Some(checkpoint_lsn));
315 }
316 let next = current
317 .chunks
318 .iter()
319 .find(|chunk| !written.contains(&chunk.ordinal))
320 .ok_or_else(|| RedDBError::Internal("basebackup chunk tracking stalled".into()))?;
321 offset = next.snapshot_offset;
322 if let Some(snapshot_token) = token.as_deref() {
323 bootstrap_handle
324 .checkpoint_snapshot_transfer(
325 snapshot_token,
326 offset,
327 source_lsn,
328 written.len() as u64,
329 )
330 .map_err(|err| RedDBError::Internal(err.to_string()))?;
331 }
332 }
333 }
334
335 pub fn primary_replica_slot_catalog(
336 &self,
337 ) -> RedDBResult<Option<reddb_file::ReplicationSlotCatalog>> {
338 let Some(plan) = self.primary_replica_file_plan_result()? else {
339 return Ok(None);
340 };
341 match reddb_file::ReplicationSlotCatalog::read_from_path(plan.slots_path()) {
342 Ok(catalog) => Ok(Some(catalog)),
343 Err(reddb_file::RdbFileError::Io(err))
344 if err.kind() == std::io::ErrorKind::NotFound =>
345 {
346 Ok(None)
347 }
348 Err(err) => Err(RedDBError::Internal(err.to_string())),
349 }
350 }
351
352 fn primary_replica_fork_lsns(&self) -> RedDBResult<Vec<u64>> {
353 let Some(data_path) = self.inner.db.options().data_path.as_ref() else {
354 return Ok(Vec::new());
355 };
356 crate::storage::operational_manifest::OperationalManifest::for_db_path(data_path)
357 .list_forks()
358 .map(|forks| forks.into_iter().map(|fork| fork.fork_lsn).collect())
359 .map_err(|err| RedDBError::Internal(err.to_string()))
360 }
361
362 pub fn primary_replica_wal_retention_plan(
363 &self,
364 ) -> RedDBResult<Option<reddb_file::WalRetentionPlan>> {
365 let Some(plan) = self.primary_replica_file_plan_result()? else {
366 return Ok(None);
367 };
368 let Some(catalog) = self.primary_replica_slot_catalog()? else {
369 return Ok(None);
370 };
371 let current_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
372 let fork_lsns = self.primary_replica_fork_lsns()?;
373 Ok(Some(
374 plan.plan_wal_retention_with_fork_lsns(&catalog, &fork_lsns, current_lsn)
375 .map_err(|err| RedDBError::Internal(err.to_string()))?,
376 ))
377 }
378
379 pub fn primary_replica_catchup_mode(
380 &self,
381 available_from_lsn: u64,
382 replica_lsn: u64,
383 ) -> RedDBResult<Option<reddb_file::ReplicaCatchupMode>> {
384 let Some(plan) = self.primary_replica_file_plan_result()? else {
385 return Ok(None);
386 };
387 let basebackups = plan
388 .list_basebackups()
389 .map_err(|err| RedDBError::Internal(err.to_string()))?;
390 Ok(Some(plan.catchup_mode_with_basebackups(
391 available_from_lsn,
392 replica_lsn,
393 &basebackups,
394 )))
395 }
396
397 pub fn primary_replica_timeline_history_path(&self) -> Option<std::path::PathBuf> {
398 let root = self.primary_replica_root()?;
399 Some(
400 reddb_file::PrimaryReplicaFilePlan::new(root, reddb_file::TimelineId::initial())
401 .timeline_history_path(),
402 )
403 }
404
405 pub fn primary_replica_rejoin_decision(
406 &self,
407 node_timeline: reddb_file::TimelineId,
408 node_flushed_lsn: u64,
409 available_from_lsn: u64,
410 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
411 let Some(path) = self.primary_replica_timeline_history_path() else {
412 return Ok(None);
413 };
414 let history = match reddb_file::TimelineHistory::read_from_path(&path) {
415 Ok(history) => history,
416 Err(reddb_file::RdbFileError::Io(err))
417 if err.kind() == std::io::ErrorKind::NotFound =>
418 {
419 reddb_file::TimelineHistory::new(crate::utils::now_unix_millis())
420 }
421 Err(err) => return Err(RedDBError::Internal(err.to_string())),
422 };
423 Ok(Some(history.rejoin_decision(
424 node_timeline,
425 node_flushed_lsn,
426 available_from_lsn,
427 )))
428 }
429
430 pub fn persist_primary_replica_rejoin_plan(
431 &self,
432 node_timeline: reddb_file::TimelineId,
433 node_flushed_lsn: u64,
434 available_from_lsn: u64,
435 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
436 let Some(decision) = self.primary_replica_rejoin_decision(
437 node_timeline,
438 node_flushed_lsn,
439 available_from_lsn,
440 )?
441 else {
442 return Ok(None);
443 };
444
445 let (state, target_timeline, rewind_to_lsn, start_lsn) = match decision {
446 reddb_file::RejoinDecision::AlreadyCurrent => {
447 ("timeline_current", node_timeline.0, 0, node_flushed_lsn)
448 }
449 reddb_file::RejoinDecision::FollowNewTimeline {
450 target_timeline,
451 start_lsn,
452 } => ("rejoin_follow_wal", target_timeline.0, 0, start_lsn),
453 reddb_file::RejoinDecision::Rewind {
454 target_timeline,
455 rewind_to_lsn,
456 } => (
457 "rejoin_rewind_required",
458 target_timeline.0,
459 rewind_to_lsn,
460 0,
461 ),
462 reddb_file::RejoinDecision::Reclone => ("reclone_required", 0, 0, 0),
463 };
464 self.inner.db.store().set_config_tree(
465 "red.replication",
466 &crate::json!({
467 "state": state,
468 "rejoin_node_timeline": node_timeline.0,
469 "rejoin_node_flushed_lsn": node_flushed_lsn,
470 "rejoin_available_from_lsn": available_from_lsn,
471 "rejoin_target_timeline": target_timeline,
472 "rejoin_rewind_to_lsn": rewind_to_lsn,
473 "rejoin_start_lsn": start_lsn,
474 "rejoin_rewind_confirmed_timeline": 0,
475 "rejoin_rewind_confirmed_lsn": 0,
476 }),
477 );
478
479 Ok(Some(decision))
480 }
481
482 pub fn prune_primary_replica_wal_segments(
483 &self,
484 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
485 let current_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
486 self.prune_primary_replica_wal_segments_at(current_lsn)
487 }
488
489 fn prune_primary_replica_wal_segments_at(
490 &self,
491 current_lsn: u64,
492 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
493 let Some(plan) = self.primary_replica_file_plan_result()? else {
494 return Ok(None);
495 };
496 let Some(catalog) = self.primary_replica_slot_catalog()? else {
497 return Ok(None);
498 };
499 let fork_lsns = self.primary_replica_fork_lsns()?;
500 Ok(Some(
501 plan.prune_wal_segments_with_fork_lsns(&catalog, &fork_lsns, current_lsn)
502 .map_err(|err| RedDBError::Internal(err.to_string()))?,
503 ))
504 }
505
506 pub fn ack_primary_replica_lsn_and_prune(
507 &self,
508 replica_id: &str,
509 applied_lsn: u64,
510 durable_lsn: u64,
511 apply_errors_total: u64,
512 divergence_total: u64,
513 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
514 let Some(repl) = self.inner.db.replication.as_ref() else {
515 return Ok(None);
516 };
517 repl.ack_replica_lsn_with_observability(
518 replica_id,
519 applied_lsn,
520 durable_lsn,
521 apply_errors_total,
522 divergence_total,
523 );
524 self.refresh_replication_flow_control();
525 let current_lsn = self
526 .primary_logical_head_lsn()
527 .max(self.cdc_current_lsn())
528 .max(applied_lsn)
529 .max(durable_lsn);
530 self.prune_primary_replica_wal_segments_at(current_lsn)
531 }
532
533 pub fn record_failover_timeline_promotion(
534 &self,
535 replica_id: &str,
536 applied_lsn: u64,
537 ) -> RedDBResult<reddb_file::TimelineHistory> {
538 let Some(path) = self.primary_replica_timeline_history_path() else {
539 return Ok(reddb_file::TimelineHistory::new(
540 crate::utils::now_unix_millis(),
541 ));
542 };
543 let now_ms = crate::utils::now_unix_millis();
544 let history = match reddb_file::TimelineHistory::read_from_path(&path) {
545 Ok(history) => history,
546 Err(reddb_file::RdbFileError::Io(err))
547 if err.kind() == std::io::ErrorKind::NotFound =>
548 {
549 reddb_file::TimelineHistory::new(now_ms)
550 }
551 Err(err) => return Err(RedDBError::Internal(err.to_string())),
552 };
553 let parent = history
554 .current()
555 .unwrap_or_else(reddb_file::TimelineId::initial);
556 let candidate = reddb_file::PromotionCandidate {
557 replica_id: replica_id.to_string(),
558 timeline: parent,
559 received_lsn: applied_lsn,
560 flushed_lsn: applied_lsn,
561 applied_lsn,
562 };
563 let promoted = history
564 .promotion_history(&candidate, parent.next(), now_ms)
565 .map_err(|err| RedDBError::Internal(err.to_string()))?;
566 promoted
567 .write_to_path(path)
568 .map_err(|err| RedDBError::Internal(err.to_string()))?;
569 Ok(promoted)
570 }
571
572 pub fn record_replica_relay_batch(
573 &self,
574 replica_id: &str,
575 records: &[(u64, Vec<u8>)],
576 applied_lsn: u64,
577 ) -> RedDBResult<()> {
578 let Some(plan) = self.primary_replica_file_plan_result()? else {
579 return Ok(());
580 };
581 let path = plan.relay_manifest_path(replica_id);
582 let Some((first_lsn, _)) = records.first() else {
583 return Ok(());
584 };
585 let end_lsn = records
586 .iter()
587 .map(|(lsn, _)| *lsn)
588 .max()
589 .unwrap_or(*first_lsn);
590 let mut manifest = match reddb_file::ReplicaRelayLogManifest::read_from_path(&path) {
591 Ok(manifest) => {
592 let relay_dir = path.parent().ok_or_else(|| {
593 RedDBError::Internal("relay manifest path has no parent".into())
594 })?;
595 manifest
596 .validate_segments(relay_dir)
597 .map_err(|err| RedDBError::Internal(err.to_string()))?;
598 manifest
599 }
600 Err(reddb_file::RdbFileError::Io(err))
601 if err.kind() == std::io::ErrorKind::NotFound =>
602 {
603 reddb_file::ReplicaRelayLogManifest::new(
604 replica_id,
605 reddb_file::TimelineId::initial(),
606 )
607 }
608 Err(err) => return Err(RedDBError::Internal(err.to_string())),
609 };
610 if manifest.replica_id != replica_id {
611 return Err(RedDBError::Internal(format!(
612 "relay manifest replica_id {} does not match {}",
613 manifest.replica_id, replica_id
614 )));
615 }
616 if manifest.timeline != reddb_file::TimelineId::initial() {
617 return Err(RedDBError::Internal(format!(
618 "relay manifest timeline {} is not current timeline {}",
619 manifest.timeline.0,
620 reddb_file::TimelineId::initial().0
621 )));
622 }
623
624 if end_lsn > manifest.flushed_lsn {
625 let new_records = records
626 .iter()
627 .filter(|(lsn, _)| *lsn > manifest.flushed_lsn)
628 .map(|(lsn, payload)| reddb_file::ReplicaRelayLogRecord::new(*lsn, payload.clone()))
629 .collect::<Vec<_>>();
630 let segment = reddb_file::ReplicaRelayLogSegment::from_records(
631 reddb_file::TimelineId::initial(),
632 new_records,
633 )
634 .map_err(|err| RedDBError::Internal(err.to_string()))?;
635 let start_lsn = segment.start_lsn;
636 let end_lsn = segment.end_lsn;
637 let relative_path = reddb_file::layout::relay_segment_relative_path(start_lsn, end_lsn);
638 let segment_path = path
639 .parent()
640 .ok_or_else(|| RedDBError::Internal("relay manifest path has no parent".into()))?
641 .join(&relative_path);
642 segment
643 .write_to_path(&segment_path)
644 .map_err(|err| RedDBError::Internal(err.to_string()))?;
645 manifest
646 .push_segment(
647 reddb_file::RelayLogSegmentRef::new(
648 relative_path,
649 start_lsn,
650 end_lsn,
651 segment
652 .checksum()
653 .map_err(|err| RedDBError::Internal(err.to_string()))?,
654 )
655 .map_err(|err| RedDBError::Internal(err.to_string()))?,
656 )
657 .map_err(|err| RedDBError::Internal(err.to_string()))?;
658 }
659 manifest
660 .mark_applied(applied_lsn.min(manifest.flushed_lsn))
661 .map_err(|err| RedDBError::Internal(err.to_string()))?;
662 manifest
663 .write_to_path(path)
664 .map_err(|err| RedDBError::Internal(err.to_string()))
665 }
666}