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 =
187 crate::replication::replica::ReplicaBootstrapper::new(replica_id.clone());
188 let source_lsn = self.config_u64("red.replication.last_applied_lsn", 0);
189 let _ = std::fs::remove_file(&pending_path);
190
191 let chunk_bytes = chunk_bytes.max(1);
192 let (resume, mut bootstrap_handle) = match bootstrapper.resume(&intent_log) {
193 Some((resume, handle)) => (Some(resume), handle),
194 None => (
195 None,
196 bootstrapper
197 .begin(&intent_log, source_lsn, 0)
198 .map_err(|err| RedDBError::Internal(err.to_string()))?,
199 ),
200 };
201 let mut token: Option<String> = resume
202 .as_ref()
203 .and_then(|resume| resume.snapshot_token.clone());
204 let mut manifest: Option<reddb_file::PrimaryReplicaBaseBackupManifest> = None;
205 let mut written = std::collections::BTreeSet::new();
206 let mut offset = resume
207 .as_ref()
208 .map(|resume| resume.snapshot_offset)
209 .unwrap_or(0);
210
211 loop {
212 let mut request = tonic::Request::new(crate::grpc::proto::Empty {});
213 request.metadata_mut().insert(
214 "x-reddb-replica-id",
215 replica_id
216 .parse()
217 .map_err(|err| RedDBError::Internal(format!("replica id metadata: {err}")))?,
218 );
219 request.metadata_mut().insert(
220 "x-reddb-snapshot-max-bytes",
221 chunk_bytes
222 .to_string()
223 .parse()
224 .map_err(|err| RedDBError::Internal(format!("snapshot max bytes: {err}")))?,
225 );
226 request.metadata_mut().insert(
227 "x-reddb-snapshot-offset",
228 offset
229 .to_string()
230 .parse()
231 .map_err(|err| RedDBError::Internal(format!("snapshot offset: {err}")))?,
232 );
233 if let Some(token) = &token {
234 request.metadata_mut().insert(
235 "x-reddb-snapshot-token",
236 token.parse().map_err(|err| {
237 RedDBError::Internal(format!("snapshot token metadata: {err}"))
238 })?,
239 );
240 }
241
242 let response = client
243 .replication_snapshot(request)
244 .await
245 .map_err(|err| RedDBError::Internal(format!("replication snapshot: {err}")))?;
246 let payload = reddb_wire::replication::BaseBackupChunk::decode_json(
247 response.into_inner().payload.as_bytes(),
248 )
249 .map_err(|err| RedDBError::Internal(format!("parse replication snapshot: {err}")))?;
250 if token.is_none() {
251 token = payload.snapshot_token.clone();
252 }
253 let staged =
254 crate::replication::replica::stage_basebackup_snapshot_chunk(&payload, &parts_root)
255 .map_err(|err| RedDBError::Internal(err.to_string()))?
256 .ok_or_else(|| {
257 RedDBError::Internal(
258 "replication snapshot did not include basebackup payload".into(),
259 )
260 })?;
261 if let Some(existing) = &manifest {
262 if existing != &staged.manifest {
263 return Err(RedDBError::Internal(
264 "replication snapshot basebackup manifest changed while downloading".into(),
265 ));
266 }
267 } else {
268 manifest = Some(staged.manifest.clone());
269 written.extend(
270 crate::replication::replica::recover_staged_basebackup_chunks(
271 &staged.manifest,
272 &parts_root,
273 )
274 .map_err(|err| RedDBError::Internal(err.to_string()))?,
275 );
276 }
277 if let Some(ordinal) = staged.chunk_ordinal {
278 written.insert(ordinal);
279 }
280 let current = manifest
281 .as_ref()
282 .expect("manifest set after staging basebackup chunk");
283 if current
284 .chunks
285 .iter()
286 .all(|chunk| written.contains(&chunk.ordinal))
287 {
288 current
289 .verify_snapshot_parts(&parts_root)
290 .map_err(|err| RedDBError::Internal(err.to_string()))?;
291 let checkpoint_lsn = self.materialize_primary_replica_basebackup_snapshot(
292 current,
293 &parts_root,
294 &pending_path,
295 )?;
296 self.inner.db.store().set_config_tree(
297 "red.replication",
298 &crate::json!({
299 "state": "rebootstrap_ready",
300 "rebootstrap_pending_path": pending_path.display().to_string(),
301 "rebootstrap_checkpoint_lsn": checkpoint_lsn,
302 "rebootstrap_timeline": current.timeline.0,
303 }),
304 );
305 let data_path =
306 self.inner.db.options().data_path.as_ref().ok_or_else(|| {
307 RedDBError::Internal("replica data path unavailable".into())
308 })?;
309 reddb_file::write_rebootstrap_ready_marker(
310 data_path,
311 &reddb_file::ReplicaRebootstrapReadyMarker {
312 pending_path: pending_path.clone(),
313 checkpoint_lsn,
314 timeline: current.timeline,
315 },
316 )
317 .map_err(|err| RedDBError::Internal(err.to_string()))?;
318 bootstrap_handle
319 .complete(current.chunks.len() as u64, 0)
320 .map_err(|err| RedDBError::Internal(err.to_string()))?;
321 return Ok(Some(checkpoint_lsn));
322 }
323 let next = current
324 .chunks
325 .iter()
326 .find(|chunk| !written.contains(&chunk.ordinal))
327 .ok_or_else(|| RedDBError::Internal("basebackup chunk tracking stalled".into()))?;
328 offset = next.snapshot_offset;
329 if let Some(snapshot_token) = token.as_deref() {
330 bootstrap_handle
331 .checkpoint_snapshot_transfer(
332 snapshot_token,
333 offset,
334 source_lsn,
335 written.len() as u64,
336 )
337 .map_err(|err| RedDBError::Internal(err.to_string()))?;
338 }
339 }
340 }
341
342 pub fn primary_replica_slot_catalog(
343 &self,
344 ) -> RedDBResult<Option<reddb_file::ReplicationSlotCatalog>> {
345 let Some(plan) = self.primary_replica_file_plan_result()? else {
346 return Ok(None);
347 };
348 match reddb_file::ReplicationSlotCatalog::read_from_path(plan.slots_path()) {
349 Ok(catalog) => Ok(Some(catalog)),
350 Err(reddb_file::RdbFileError::Io(err))
351 if err.kind() == std::io::ErrorKind::NotFound =>
352 {
353 Ok(None)
354 }
355 Err(err) => Err(RedDBError::Internal(err.to_string())),
356 }
357 }
358
359 fn primary_replica_fork_lsns(&self) -> RedDBResult<Vec<u64>> {
360 let Some(data_path) = self.inner.db.options().data_path.as_ref() else {
361 return Ok(Vec::new());
362 };
363 crate::storage::operational_manifest::OperationalManifest::for_db_path(data_path)
364 .list_forks()
365 .map(|forks| forks.into_iter().map(|fork| fork.fork_lsn).collect())
366 .map_err(|err| RedDBError::Internal(err.to_string()))
367 }
368
369 pub fn primary_replica_wal_retention_plan(
370 &self,
371 ) -> RedDBResult<Option<reddb_file::WalRetentionPlan>> {
372 let Some(plan) = self.primary_replica_file_plan_result()? else {
373 return Ok(None);
374 };
375 let Some(catalog) = self.primary_replica_slot_catalog()? else {
376 return Ok(None);
377 };
378 let current_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
379 let fork_lsns = self.primary_replica_fork_lsns()?;
380 Ok(Some(
381 plan.plan_wal_retention_with_fork_lsns(&catalog, &fork_lsns, current_lsn)
382 .map_err(|err| RedDBError::Internal(err.to_string()))?,
383 ))
384 }
385
386 pub fn primary_replica_catchup_mode(
387 &self,
388 available_from_lsn: u64,
389 replica_lsn: u64,
390 ) -> RedDBResult<Option<reddb_file::ReplicaCatchupMode>> {
391 let Some(plan) = self.primary_replica_file_plan_result()? else {
392 return Ok(None);
393 };
394 let basebackups = plan
395 .list_basebackups()
396 .map_err(|err| RedDBError::Internal(err.to_string()))?;
397 Ok(Some(plan.catchup_mode_with_basebackups(
398 available_from_lsn,
399 replica_lsn,
400 &basebackups,
401 )))
402 }
403
404 pub fn primary_replica_timeline_history_path(&self) -> Option<std::path::PathBuf> {
405 let root = self.primary_replica_root()?;
406 Some(
407 reddb_file::PrimaryReplicaFilePlan::new(root, reddb_file::TimelineId::initial())
408 .timeline_history_path(),
409 )
410 }
411
412 pub fn primary_replica_rejoin_decision(
413 &self,
414 node_timeline: reddb_file::TimelineId,
415 node_flushed_lsn: u64,
416 available_from_lsn: u64,
417 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
418 let Some(path) = self.primary_replica_timeline_history_path() else {
419 return Ok(None);
420 };
421 let history = match reddb_file::TimelineHistory::read_from_path(&path) {
422 Ok(history) => history,
423 Err(reddb_file::RdbFileError::Io(err))
424 if err.kind() == std::io::ErrorKind::NotFound =>
425 {
426 reddb_file::TimelineHistory::new(crate::utils::now_unix_millis())
427 }
428 Err(err) => return Err(RedDBError::Internal(err.to_string())),
429 };
430 Ok(Some(history.rejoin_decision(
431 node_timeline,
432 node_flushed_lsn,
433 available_from_lsn,
434 )))
435 }
436
437 pub fn persist_primary_replica_rejoin_plan(
438 &self,
439 node_timeline: reddb_file::TimelineId,
440 node_flushed_lsn: u64,
441 available_from_lsn: u64,
442 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
443 let Some(decision) = self.primary_replica_rejoin_decision(
444 node_timeline,
445 node_flushed_lsn,
446 available_from_lsn,
447 )?
448 else {
449 return Ok(None);
450 };
451
452 let (state, target_timeline, rewind_to_lsn, start_lsn) = match decision {
453 reddb_file::RejoinDecision::AlreadyCurrent => {
454 ("timeline_current", node_timeline.0, 0, node_flushed_lsn)
455 }
456 reddb_file::RejoinDecision::FollowNewTimeline {
457 target_timeline,
458 start_lsn,
459 } => ("rejoin_follow_wal", target_timeline.0, 0, start_lsn),
460 reddb_file::RejoinDecision::Rewind {
461 target_timeline,
462 rewind_to_lsn,
463 } => (
464 "rejoin_rewind_required",
465 target_timeline.0,
466 rewind_to_lsn,
467 0,
468 ),
469 reddb_file::RejoinDecision::Reclone => ("reclone_required", 0, 0, 0),
470 };
471 self.inner.db.store().set_config_tree(
472 "red.replication",
473 &crate::json!({
474 "state": state,
475 "rejoin_node_timeline": node_timeline.0,
476 "rejoin_node_flushed_lsn": node_flushed_lsn,
477 "rejoin_available_from_lsn": available_from_lsn,
478 "rejoin_target_timeline": target_timeline,
479 "rejoin_rewind_to_lsn": rewind_to_lsn,
480 "rejoin_start_lsn": start_lsn,
481 "rejoin_rewind_confirmed_timeline": 0,
482 "rejoin_rewind_confirmed_lsn": 0,
483 }),
484 );
485
486 Ok(Some(decision))
487 }
488
489 pub fn prune_primary_replica_wal_segments(
490 &self,
491 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
492 let current_lsn = self.primary_logical_head_lsn().max(self.cdc_current_lsn());
493 self.prune_primary_replica_wal_segments_at(current_lsn)
494 }
495
496 fn prune_primary_replica_wal_segments_at(
497 &self,
498 current_lsn: u64,
499 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
500 let Some(plan) = self.primary_replica_file_plan_result()? else {
501 return Ok(None);
502 };
503 let Some(catalog) = self.primary_replica_slot_catalog()? else {
504 return Ok(None);
505 };
506 let fork_lsns = self.primary_replica_fork_lsns()?;
507 Ok(Some(
508 plan.prune_wal_segments_with_fork_lsns(&catalog, &fork_lsns, current_lsn)
509 .map_err(|err| RedDBError::Internal(err.to_string()))?,
510 ))
511 }
512
513 pub fn ack_primary_replica_lsn_and_prune(
514 &self,
515 replica_id: &str,
516 applied_lsn: u64,
517 durable_lsn: u64,
518 apply_errors_total: u64,
519 divergence_total: u64,
520 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
521 let Some(repl) = self.inner.db.replication.as_ref() else {
522 return Ok(None);
523 };
524 repl.ack_replica_lsn_with_observability(
525 replica_id,
526 applied_lsn,
527 durable_lsn,
528 apply_errors_total,
529 divergence_total,
530 );
531 self.refresh_replication_flow_control();
532 let current_lsn = self
533 .primary_logical_head_lsn()
534 .max(self.cdc_current_lsn())
535 .max(applied_lsn)
536 .max(durable_lsn);
537 self.prune_primary_replica_wal_segments_at(current_lsn)
538 }
539
540 pub fn record_failover_timeline_promotion(
541 &self,
542 replica_id: &str,
543 applied_lsn: u64,
544 ) -> RedDBResult<reddb_file::TimelineHistory> {
545 let Some(path) = self.primary_replica_timeline_history_path() else {
546 return Ok(reddb_file::TimelineHistory::new(
547 crate::utils::now_unix_millis(),
548 ));
549 };
550 let now_ms = crate::utils::now_unix_millis();
551 let history = match reddb_file::TimelineHistory::read_from_path(&path) {
552 Ok(history) => history,
553 Err(reddb_file::RdbFileError::Io(err))
554 if err.kind() == std::io::ErrorKind::NotFound =>
555 {
556 reddb_file::TimelineHistory::new(now_ms)
557 }
558 Err(err) => return Err(RedDBError::Internal(err.to_string())),
559 };
560 let parent = history
561 .current()
562 .unwrap_or_else(reddb_file::TimelineId::initial);
563 let candidate = reddb_file::PromotionCandidate {
564 replica_id: replica_id.to_string(),
565 timeline: parent,
566 received_lsn: applied_lsn,
567 flushed_lsn: applied_lsn,
568 applied_lsn,
569 };
570 let promoted = history
571 .promotion_history(&candidate, parent.next(), now_ms)
572 .map_err(|err| RedDBError::Internal(err.to_string()))?;
573 promoted
574 .write_to_path(path)
575 .map_err(|err| RedDBError::Internal(err.to_string()))?;
576 Ok(promoted)
577 }
578
579 pub fn record_replica_relay_batch<T: AsRef<[u8]>>(
580 &self,
581 replica_id: &str,
582 records: &[(u64, T)],
583 applied_lsn: u64,
584 ) -> RedDBResult<()> {
585 let Some(plan) = self.primary_replica_file_plan_result()? else {
586 return Ok(());
587 };
588 let path = plan.relay_manifest_path(replica_id);
589 let Some((first_lsn, _)) = records.first() else {
590 return Ok(());
591 };
592 let end_lsn = records
593 .iter()
594 .map(|(lsn, _)| *lsn)
595 .max()
596 .unwrap_or(*first_lsn);
597 let mut manifest = match reddb_file::ReplicaRelayLogManifest::read_from_path(&path) {
598 Ok(manifest) => {
599 let relay_dir = path.parent().ok_or_else(|| {
600 RedDBError::Internal("relay manifest path has no parent".into())
601 })?;
602 manifest
603 .validate_segments(relay_dir)
604 .map_err(|err| RedDBError::Internal(err.to_string()))?;
605 manifest
606 }
607 Err(reddb_file::RdbFileError::Io(err))
608 if err.kind() == std::io::ErrorKind::NotFound =>
609 {
610 reddb_file::ReplicaRelayLogManifest::new(
611 replica_id,
612 reddb_file::TimelineId::initial(),
613 )
614 }
615 Err(err) => return Err(RedDBError::Internal(err.to_string())),
616 };
617 if manifest.replica_id != replica_id {
618 return Err(RedDBError::Internal(format!(
619 "relay manifest replica_id {} does not match {}",
620 manifest.replica_id, replica_id
621 )));
622 }
623 if manifest.timeline != reddb_file::TimelineId::initial() {
624 return Err(RedDBError::Internal(format!(
625 "relay manifest timeline {} is not current timeline {}",
626 manifest.timeline.0,
627 reddb_file::TimelineId::initial().0
628 )));
629 }
630
631 if end_lsn > manifest.flushed_lsn {
632 let new_records = records
633 .iter()
634 .filter(|(lsn, _)| *lsn > manifest.flushed_lsn)
635 .map(|(lsn, payload)| {
636 reddb_file::ReplicaRelayLogRecord::new(*lsn, payload.as_ref().to_vec())
637 })
638 .collect::<Vec<_>>();
639 let segment = reddb_file::ReplicaRelayLogSegment::from_records(
640 reddb_file::TimelineId::initial(),
641 new_records,
642 )
643 .map_err(|err| RedDBError::Internal(err.to_string()))?;
644 let start_lsn = segment.start_lsn;
645 let end_lsn = segment.end_lsn;
646 let relative_path = reddb_file::layout::relay_segment_relative_path(start_lsn, end_lsn);
647 let segment_path = path
648 .parent()
649 .ok_or_else(|| RedDBError::Internal("relay manifest path has no parent".into()))?
650 .join(&relative_path);
651 segment
652 .write_to_path(&segment_path)
653 .map_err(|err| RedDBError::Internal(err.to_string()))?;
654 manifest
655 .push_segment(
656 reddb_file::RelayLogSegmentRef::new(
657 relative_path,
658 start_lsn,
659 end_lsn,
660 segment
661 .checksum()
662 .map_err(|err| RedDBError::Internal(err.to_string()))?,
663 )
664 .map_err(|err| RedDBError::Internal(err.to_string()))?,
665 )
666 .map_err(|err| RedDBError::Internal(err.to_string()))?;
667 }
668 manifest
669 .mark_applied(applied_lsn.min(manifest.flushed_lsn))
670 .map_err(|err| RedDBError::Internal(err.to_string()))?;
671 manifest
672 .write_to_path(path)
673 .map_err(|err| RedDBError::Internal(err.to_string()))
674 }
675}