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