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