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.set_config_tree(
108 "red.replication",
109 &crate::json!({
110 "last_applied_lsn": manifest.checkpoint_lsn,
111 "state": "healthy",
112 "last_error": "",
113 }),
114 );
115 let snapshot = loaded.to_binary_dump_bytes();
116 let destination = destination.as_ref();
117 if let Some(parent) = destination.parent() {
118 std::fs::create_dir_all(parent)?;
119 }
120 crate::storage::EmbeddedRdbArtifact::create_with_snapshot(destination, &snapshot)?;
121 Ok(manifest.checkpoint_lsn)
122 }
123
124 pub fn replica_rebootstrap_staging_root(&self) -> Option<std::path::PathBuf> {
125 let data_path = self.inner.db.options().data_path.as_ref()?;
126 Some(reddb_file::layout::rebootstrap_staging_root(data_path))
127 }
128
129 pub fn replica_rebootstrap_pending_path(&self) -> Option<std::path::PathBuf> {
130 let data_path = self.inner.db.options().data_path.as_ref()?;
131 Some(reddb_file::layout::rebootstrap_pending_path(data_path))
132 }
133
134 pub(crate) async fn stage_primary_replica_rebootstrap_from_snapshot(
135 &self,
136 client: &mut crate::grpc::proto::red_db_client::RedDbClient<tonic::transport::Channel>,
137 chunk_bytes: usize,
138 ) -> RedDBResult<Option<u64>> {
139 let Some(parts_root) = self.replica_rebootstrap_staging_root() else {
140 return Ok(None);
141 };
142 let Some(pending_path) = self.replica_rebootstrap_pending_path() else {
143 return Ok(None);
144 };
145 let data_path = self
146 .inner
147 .db
148 .options()
149 .data_path
150 .as_ref()
151 .ok_or_else(|| RedDBError::Internal("replica data path unavailable".into()))?;
152 let intent_log_path = reddb_file::layout::rebootstrap_intent_log_path(data_path);
153 let intent_log = crate::telemetry::admin_intent_log::AdminIntentLog::open(intent_log_path)
154 .map_err(|err| RedDBError::Internal(err.to_string()))?;
155 let replica_id = self.resolve_replica_id();
156 let bootstrapper =
157 crate::replication::replica::ReplicaBootstrapper::new(replica_id.clone());
158 let source_lsn = self.config_u64("red.replication.last_applied_lsn", 0);
159 let _ = std::fs::remove_file(&pending_path);
160
161 let chunk_bytes = chunk_bytes.max(1);
162 let (resume, mut bootstrap_handle) = match bootstrapper.resume(&intent_log) {
163 Some((resume, handle)) => (Some(resume), handle),
164 None => (
165 None,
166 bootstrapper
167 .begin(&intent_log, source_lsn, 0)
168 .map_err(|err| RedDBError::Internal(err.to_string()))?,
169 ),
170 };
171 let mut token: Option<String> = resume
172 .as_ref()
173 .and_then(|resume| resume.snapshot_token.clone());
174 let mut manifest: Option<reddb_file::PrimaryReplicaBaseBackupManifest> = None;
175 let mut written = std::collections::BTreeSet::new();
176 let mut offset = resume
177 .as_ref()
178 .map(|resume| resume.snapshot_offset)
179 .unwrap_or(0);
180
181 let request_auth = crate::replication::request_auth::ReplicaRequestAuth::from_env()
182 .map_err(|message| RedDBError::InvalidOperation(message.into()))?;
183 loop {
184 let mut request = tonic::Request::new(crate::grpc::proto::Empty {});
185 request_auth.authorize(&mut request);
186 request.metadata_mut().insert(
187 "x-reddb-replica-id",
188 replica_id
189 .parse()
190 .map_err(|err| RedDBError::Internal(format!("replica id metadata: {err}")))?,
191 );
192 request.metadata_mut().insert(
193 "x-reddb-snapshot-max-bytes",
194 chunk_bytes
195 .to_string()
196 .parse()
197 .map_err(|err| RedDBError::Internal(format!("snapshot max bytes: {err}")))?,
198 );
199 request.metadata_mut().insert(
200 "x-reddb-snapshot-offset",
201 offset
202 .to_string()
203 .parse()
204 .map_err(|err| RedDBError::Internal(format!("snapshot offset: {err}")))?,
205 );
206 if let Some(token) = &token {
207 request.metadata_mut().insert(
208 "x-reddb-snapshot-token",
209 token.parse().map_err(|err| {
210 RedDBError::Internal(format!("snapshot token metadata: {err}"))
211 })?,
212 );
213 }
214
215 let response = client
216 .replication_snapshot(request)
217 .await
218 .map_err(|err| RedDBError::Internal(format!("replication snapshot: {err}")))?;
219 let payload = reddb_wire::replication::BaseBackupChunk::decode_json(
220 response.into_inner().payload.as_bytes(),
221 )
222 .map_err(|err| RedDBError::Internal(format!("parse replication snapshot: {err}")))?;
223 if token.is_none() {
224 token = payload.snapshot_token.clone();
225 }
226 let staged =
227 crate::replication::replica::stage_basebackup_snapshot_chunk(&payload, &parts_root)
228 .map_err(|err| RedDBError::Internal(err.to_string()))?
229 .ok_or_else(|| {
230 RedDBError::Internal(
231 "replication snapshot did not include basebackup payload".into(),
232 )
233 })?;
234 if let Some(existing) = &manifest {
235 if existing != &staged.manifest {
236 return Err(RedDBError::Internal(
237 "replication snapshot basebackup manifest changed while downloading".into(),
238 ));
239 }
240 } else {
241 manifest = Some(staged.manifest.clone());
242 written.extend(
243 crate::replication::replica::recover_staged_basebackup_chunks(
244 &staged.manifest,
245 &parts_root,
246 )
247 .map_err(|err| RedDBError::Internal(err.to_string()))?,
248 );
249 }
250 if let Some(ordinal) = staged.chunk_ordinal {
251 written.insert(ordinal);
252 }
253 let current = manifest
254 .as_ref()
255 .expect("manifest set after staging basebackup chunk");
256 if current
257 .chunks
258 .iter()
259 .all(|chunk| written.contains(&chunk.ordinal))
260 {
261 current
262 .verify_snapshot_parts(&parts_root)
263 .map_err(|err| RedDBError::Internal(err.to_string()))?;
264 let checkpoint_lsn = self.materialize_primary_replica_basebackup_snapshot(
265 current,
266 &parts_root,
267 &pending_path,
268 )?;
269 self.inner.db.store().set_config_tree(
270 "red.replication",
271 &crate::json!({
272 "state": "rebootstrap_ready",
273 "rebootstrap_pending_path": pending_path.display().to_string(),
274 "rebootstrap_checkpoint_lsn": checkpoint_lsn,
275 "rebootstrap_timeline": current.timeline.0,
276 }),
277 );
278 let data_path =
279 self.inner.db.options().data_path.as_ref().ok_or_else(|| {
280 RedDBError::Internal("replica data path unavailable".into())
281 })?;
282 reddb_file::write_rebootstrap_ready_marker(
283 data_path,
284 &reddb_file::ReplicaRebootstrapReadyMarker {
285 pending_path: pending_path.clone(),
286 checkpoint_lsn,
287 timeline: current.timeline,
288 },
289 )
290 .map_err(|err| RedDBError::Internal(err.to_string()))?;
291 bootstrap_handle
292 .complete(current.chunks.len() as u64, 0)
293 .map_err(|err| RedDBError::Internal(err.to_string()))?;
294 return Ok(Some(checkpoint_lsn));
295 }
296 let next = current
297 .chunks
298 .iter()
299 .find(|chunk| !written.contains(&chunk.ordinal))
300 .ok_or_else(|| RedDBError::Internal("basebackup chunk tracking stalled".into()))?;
301 offset = next.snapshot_offset;
302 if let Some(snapshot_token) = token.as_deref() {
303 bootstrap_handle
304 .checkpoint_snapshot_transfer(
305 snapshot_token,
306 offset,
307 source_lsn,
308 written.len() as u64,
309 )
310 .map_err(|err| RedDBError::Internal(err.to_string()))?;
311 }
312 }
313 }
314
315 pub fn primary_replica_slot_catalog(
316 &self,
317 ) -> RedDBResult<Option<reddb_file::ReplicationSlotCatalog>> {
318 let store = self.inner.db.store();
319 let _publication = store.vault_publication_lock.lock();
320 if let Some(primary) = self.inner.db.replication.as_ref() {
321 primary
322 .ensure_slot_metadata_durable()
323 .map_err(|err| RedDBError::Internal(err.to_string()))?;
324 }
325 if let Some(bytes) = crate::replication::metadata::read(
326 &self.inner.db.store(),
327 crate::replication::metadata::SLOTS,
328 )
329 .map_err(|err| RedDBError::Internal(err.to_string()))?
330 {
331 return reddb_file::ReplicationSlotCatalog::decode(&bytes)
332 .map(Some)
333 .map_err(|err| RedDBError::Internal(err.to_string()));
334 }
335 let Some(plan) = self.primary_replica_file_plan_result()? else {
336 return Ok(None);
337 };
338 match reddb_file::ReplicationSlotCatalog::read_from_path(plan.slots_path()) {
339 Ok(catalog) => Ok(Some(catalog)),
340 Err(reddb_file::RdbFileError::Io(err))
341 if err.kind() == std::io::ErrorKind::NotFound =>
342 {
343 Ok(None)
344 }
345 Err(err) => Err(RedDBError::Internal(err.to_string())),
346 }
347 }
348
349 fn primary_replica_fork_lsns(&self) -> RedDBResult<Vec<u64>> {
350 let Some(data_path) = self.inner.db.options().data_path.as_ref() else {
351 return Ok(Vec::new());
352 };
353 self.inner
354 .db
355 .store()
356 .operational_manifest_for_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
372 .primary_logical_head_lsn()
373 .max(self.replication_commit_lsn());
374 let fork_lsns = self.primary_replica_fork_lsns()?;
375 Ok(Some(
376 plan.plan_wal_retention_with_fork_lsns(&catalog, &fork_lsns, current_lsn)
377 .map_err(|err| RedDBError::Internal(err.to_string()))?,
378 ))
379 }
380
381 pub fn primary_replica_catchup_mode(
382 &self,
383 available_from_lsn: u64,
384 replica_lsn: u64,
385 ) -> RedDBResult<Option<reddb_file::ReplicaCatchupMode>> {
386 let Some(plan) = self.primary_replica_file_plan_result()? else {
387 return Ok(None);
388 };
389 let basebackups = plan
390 .list_basebackups()
391 .map_err(|err| RedDBError::Internal(err.to_string()))?;
392 Ok(Some(plan.catchup_mode_with_basebackups(
393 available_from_lsn,
394 replica_lsn,
395 &basebackups,
396 )))
397 }
398
399 pub fn primary_replica_timeline_history_path(&self) -> Option<std::path::PathBuf> {
400 let root = self.primary_replica_root()?;
401 Some(
402 reddb_file::PrimaryReplicaFilePlan::new(root, reddb_file::TimelineId::initial())
403 .timeline_history_path(),
404 )
405 }
406
407 pub fn primary_replica_timeline_history(
409 &self,
410 ) -> RedDBResult<Option<reddb_file::TimelineHistory>> {
411 let Some(path) = self.primary_replica_timeline_history_path() else {
412 return Ok(None);
413 };
414 crate::replication::metadata::timeline(&self.inner.db.store(), &path)
415 .map(Some)
416 .map_err(|err| RedDBError::Internal(err.to_string()))
417 }
418
419 pub fn primary_replica_rejoin_decision(
420 &self,
421 node_timeline: reddb_file::TimelineId,
422 node_flushed_lsn: u64,
423 available_from_lsn: u64,
424 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
425 let Some(history) = self.primary_replica_timeline_history()? else {
426 return Ok(None);
427 };
428 Ok(Some(history.rejoin_decision(
429 node_timeline,
430 node_flushed_lsn,
431 available_from_lsn,
432 )))
433 }
434
435 pub fn persist_primary_replica_rejoin_plan(
436 &self,
437 node_timeline: reddb_file::TimelineId,
438 node_flushed_lsn: u64,
439 available_from_lsn: u64,
440 ) -> RedDBResult<Option<reddb_file::RejoinDecision>> {
441 let Some(decision) = self.primary_replica_rejoin_decision(
442 node_timeline,
443 node_flushed_lsn,
444 available_from_lsn,
445 )?
446 else {
447 return Ok(None);
448 };
449
450 let (state, target_timeline, rewind_to_lsn, start_lsn) = match decision {
451 reddb_file::RejoinDecision::AlreadyCurrent => {
452 ("timeline_current", node_timeline.0, 0, node_flushed_lsn)
453 }
454 reddb_file::RejoinDecision::FollowNewTimeline {
455 target_timeline,
456 start_lsn,
457 } => ("rejoin_follow_wal", target_timeline.0, 0, start_lsn),
458 reddb_file::RejoinDecision::Rewind {
459 target_timeline,
460 rewind_to_lsn,
461 } => (
462 "rejoin_rewind_required",
463 target_timeline.0,
464 rewind_to_lsn,
465 0,
466 ),
467 reddb_file::RejoinDecision::Reclone => ("reclone_required", 0, 0, 0),
468 };
469 self.inner.db.store().set_config_tree(
470 "red.replication",
471 &crate::json!({
472 "state": state,
473 "rejoin_node_timeline": node_timeline.0,
474 "rejoin_node_flushed_lsn": node_flushed_lsn,
475 "rejoin_available_from_lsn": available_from_lsn,
476 "rejoin_target_timeline": target_timeline,
477 "rejoin_rewind_to_lsn": rewind_to_lsn,
478 "rejoin_start_lsn": start_lsn,
479 "rejoin_rewind_confirmed_timeline": 0,
480 "rejoin_rewind_confirmed_lsn": 0,
481 }),
482 );
483
484 Ok(Some(decision))
485 }
486
487 pub fn prune_primary_replica_wal_segments(
488 &self,
489 ) -> RedDBResult<Option<reddb_file::WalPruneResult>> {
490 let current_lsn = self
491 .primary_logical_head_lsn()
492 .max(self.replication_commit_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.replication_commit_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 store = self.inner.db.store();
546 let _publication = store.vault_publication_lock.lock();
547 let Some(history) = self.primary_replica_timeline_history()? else {
548 return Ok(reddb_file::TimelineHistory::new(
549 crate::utils::now_unix_millis(),
550 ));
551 };
552 let now_ms = crate::utils::now_unix_millis();
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 crate::replication::metadata::write(
567 &self.inner.db.store(),
568 crate::replication::metadata::TIMELINE,
569 &promoted.encode(),
570 )
571 .map_err(|err| RedDBError::Internal(err.to_string()))?;
572 if let Some(path) = self.primary_replica_timeline_history_path() {
573 crate::replication::metadata::retire_legacy(&path)
574 .map_err(|err| RedDBError::Internal(err.to_string()))?;
575 }
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 store = self.inner.db.store();
586 let _publication = store.vault_publication_lock.lock();
587 let Some(plan) = self.primary_replica_file_plan_result()? else {
588 return Ok(());
589 };
590 let path = plan.relay_manifest_path(replica_id);
591 let Some((first_lsn, _)) = records.first() else {
592 return Ok(());
593 };
594 let end_lsn = records
595 .iter()
596 .map(|(lsn, _)| *lsn)
597 .max()
598 .unwrap_or(*first_lsn);
599 let mut manifest = match self.replica_relay_manifest(replica_id)? {
600 Some(manifest) => {
601 let relay_dir = path.parent().ok_or_else(|| {
602 RedDBError::Internal("relay manifest path has no parent".into())
603 })?;
604 manifest
605 .validate_segments(relay_dir)
606 .map_err(|err| RedDBError::Internal(err.to_string()))?;
607 manifest
608 }
609 None => reddb_file::ReplicaRelayLogManifest::new(
610 replica_id,
611 reddb_file::TimelineId::initial(),
612 ),
613 };
614 if manifest.replica_id != replica_id {
615 return Err(RedDBError::Internal(format!(
616 "relay manifest replica_id {} does not match {}",
617 manifest.replica_id, replica_id
618 )));
619 }
620 if manifest.timeline != reddb_file::TimelineId::initial() {
621 return Err(RedDBError::Internal(format!(
622 "relay manifest timeline {} is not current timeline {}",
623 manifest.timeline.0,
624 reddb_file::TimelineId::initial().0
625 )));
626 }
627
628 if end_lsn > manifest.flushed_lsn {
629 let new_records = records
630 .iter()
631 .filter(|(lsn, _)| *lsn > manifest.flushed_lsn)
632 .map(|(lsn, payload)| {
633 reddb_file::ReplicaRelayLogRecord::new(*lsn, payload.as_ref().to_vec())
634 })
635 .collect::<Vec<_>>();
636 let segment = reddb_file::ReplicaRelayLogSegment::from_records(
637 reddb_file::TimelineId::initial(),
638 new_records,
639 )
640 .map_err(|err| RedDBError::Internal(err.to_string()))?;
641 let start_lsn = segment.start_lsn;
642 let end_lsn = segment.end_lsn;
643 let relative_path = reddb_file::layout::relay_segment_relative_path(start_lsn, end_lsn);
644 let segment_path = path
645 .parent()
646 .ok_or_else(|| RedDBError::Internal("relay manifest path has no parent".into()))?
647 .join(&relative_path);
648 segment
649 .write_to_path(&segment_path)
650 .map_err(|err| RedDBError::Internal(err.to_string()))?;
651 manifest
652 .push_segment(
653 reddb_file::RelayLogSegmentRef::new(
654 relative_path,
655 start_lsn,
656 end_lsn,
657 segment
658 .checksum()
659 .map_err(|err| RedDBError::Internal(err.to_string()))?,
660 )
661 .map_err(|err| RedDBError::Internal(err.to_string()))?,
662 )
663 .map_err(|err| RedDBError::Internal(err.to_string()))?;
664 }
665 manifest
666 .mark_applied(applied_lsn.min(manifest.flushed_lsn))
667 .map_err(|err| RedDBError::Internal(err.to_string()))?;
668 let bytes = manifest
669 .encode()
670 .map_err(|err| RedDBError::Internal(err.to_string()))?;
671 crate::replication::metadata::write(
672 &self.inner.db.store(),
673 &format!("replication.relay.{}", hex::encode(replica_id)),
674 &bytes,
675 )
676 .map_err(|err| RedDBError::Internal(err.to_string()))?;
677 crate::replication::metadata::retire_legacy(&path)
678 .map_err(|err| RedDBError::Internal(err.to_string()))
679 }
680}