linkrs_storage/engine/graph_storage/
commit_api.rs

1use linkrs_core::{StorageError, StorageResult};
2
3use super::GraphStorage;
4
5impl crate::StorageCommitOps for GraphStorage {
6    fn commit_staged_writes(
7        &self,
8        transaction_id: linkrs_core::types::TransactionId,
9        intents: &[linkrs_core::wal::OutboxIntent],
10    ) -> StorageResult<linkrs_core::types::CommitLsn> {
11        self.ctx.commit_staged_writes(transaction_id, intents)
12    }
13
14    fn abort_staged_writes(
15        &self,
16        transaction_id: linkrs_core::types::TransactionId,
17    ) -> StorageResult<()> {
18        self.ctx.abort_staged_writes(transaction_id);
19        Ok(())
20    }
21
22    fn commit_staged_writes_with_durability(
23        &self,
24        transaction_id: linkrs_core::types::TransactionId,
25        intents: &[linkrs_core::wal::OutboxIntent],
26        durability: linkrs_core::types::DurabilityLevel,
27    ) -> StorageResult<linkrs_core::types::CommitLsn> {
28        self.ctx
29            .commit_staged_writes_with_durability(transaction_id, intents, durability)
30    }
31
32    fn recover_outbox_projection(
33        &self,
34        sync_manager: &linkrs_sync::SyncManager,
35    ) -> StorageResult<usize> {
36        use linkrs_transaction::wal::{collect_committed_transactions, LocalWalParser, WalParser};
37        let Some(paths) = self.ctx.storage_paths() else {
38            return Ok(0);
39        };
40        if !paths.wal_dir().exists() {
41            return Ok(0);
42        }
43
44        let snapshot_lsn = sync_manager.outbox_materialized_lsn().map_err(|error| {
45            StorageError::db_error(format!(
46                "Failed to read outbox materialization frontier: {}",
47                error
48            ))
49        })?;
50
51        let mut parser = LocalWalParser::new();
52        parser
53            .open(&paths.wal_dir().to_string_lossy())
54            .map_err(|error| {
55                StorageError::wal_error(format!(
56                    "Failed to parse WAL for outbox recovery: {}",
57                    error
58                ))
59            })?;
60        let transactions =
61            collect_committed_transactions(&parser.parse_all_entries()).map_err(|error| {
62                StorageError::wal_error(format!(
63                    "Failed to validate WAL for outbox recovery: {}",
64                    error
65                ))
66            })?;
67
68        let mut recovered = 0usize;
69        for transaction in transactions {
70            if transaction.intents.is_empty() {
71                continue;
72            }
73            if let Some(snapshot_lsn) = snapshot_lsn {
74                if transaction.commit_lsn <= snapshot_lsn {
75                    continue;
76                }
77            }
78            sync_manager
79                .materialize_committed_transaction(
80                    transaction.transaction_id,
81                    transaction.commit_lsn,
82                    &transaction.intents,
83                )
84                .map_err(|error| {
85                    StorageError::db_error(format!(
86                        "Failed to recover outbox transaction {}: {}",
87                        transaction.transaction_id, error
88                    ))
89                })?;
90            recovered = recovered.saturating_add(transaction.intents.len());
91        }
92
93        log::info!(
94            "Outbox projection recovery complete: {} intents replayed (snapshot_lsn={:?})",
95            recovered,
96            snapshot_lsn
97        );
98
99        Ok(recovered)
100    }
101}
linkrs_storage/engine/graph_storage/commit_api.rs

linkrs_storage/engine/graph_storage/
commit_api.rs
