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