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