linkrs_storage/engine/sync_wrapper/
operation.rs

1use super::SyncWrapper;
2use crate::macros::forward_methods;
3use crate::{
4    StorageClient, StorageCommitOps, StorageGcOps, StorageOperationContext,
5    StorageOperationContextOps, StorageRecoveryOps, StorageSchemaContextOps, StorageSyncContextOps,
6    StorageTtlPurgeOps,
7};
8use linkrs_core::metadata::{IndexMetadataManager, SchemaManager};
9use linkrs_core::StorageResult;
10use std::sync::Arc;
11
12impl<S: StorageClient + 'static> crate::stats_reader::ColumnStatsReader for SyncWrapper<S> {
13    fn vertex_column_stats(
14        &self,
15        space: &str,
16        tag: &str,
17        column: &str,
18    ) -> Option<std::sync::Arc<crate::stats_reader::ColumnStatsSnapshot>> {
19        self.inner.vertex_column_stats(space, tag, column)
20    }
21
22    fn edge_column_stats(
23        &self,
24        space: &str,
25        edge_type: &str,
26        column: &str,
27    ) -> Option<std::sync::Arc<crate::stats_reader::ColumnStatsSnapshot>> {
28        self.inner.edge_column_stats(space, edge_type, column)
29    }
30
31    fn vertex_table_stats(
32        &self,
33        space: &str,
34        tag: &str,
35    ) -> Option<std::sync::Arc<crate::stats_reader::TableCardinalitySnapshot>> {
36        self.inner.vertex_table_stats(space, tag)
37    }
38
39    fn edge_table_stats(
40        &self,
41        space: &str,
42        edge_type: &str,
43    ) -> Option<std::sync::Arc<crate::stats_reader::TableCardinalitySnapshot>> {
44        self.inner.edge_table_stats(space, edge_type)
45    }
46
47    fn stats_epoch(&self) -> u64 {
48        self.inner.stats_epoch()
49    }
50}
51
52impl<S: StorageClient + StorageSchemaContextOps + 'static> StorageSchemaContextOps
53    for SyncWrapper<S>
54{
55    forward_methods!(inner;
56        fn get_schema_manager(&self) -> Option<Arc<SchemaManager>>;
57        fn get_index_metadata_manager(&self) -> Option<Arc<dyn IndexMetadataManager>>;
58    );
59}
60
61impl<S: StorageClient + 'static> StorageOperationContextOps for SyncWrapper<S> {
62    fn bind_auto_commit_context(&self) -> StorageResult<Self> {
63        let inner = self.inner.bind_auto_commit_context()?;
64        let inner = match inner.operation_context() {
65            Some(context) => inner.bind_operation_context((*context).clone()),
66            None => inner,
67        };
68        Ok(Self {
69            inner,
70            sync_manager: self.sync_manager.clone(),
71            enabled: self.enabled,
72            auto_commit_owner: true,
73        })
74    }
75
76    fn bind_operation_context(&self, context: StorageOperationContext) -> Self {
77        Self {
78            inner: self.inner.bind_operation_context(context),
79            sync_manager: self.sync_manager.clone(),
80            enabled: self.enabled,
81            auto_commit_owner: false,
82        }
83    }
84
85    fn bind_read_operation_context(&self) -> StorageResult<Self> {
86        let inner = self.inner.bind_read_operation_context()?;
87        let inner = match inner.operation_context() {
88            Some(context) => inner.bind_operation_context((*context).clone()),
89            None => inner,
90        };
91        Ok(Self {
92            inner,
93            sync_manager: self.sync_manager.clone(),
94            enabled: self.enabled,
95            auto_commit_owner: true,
96        })
97    }
98
99    fn operation_context(&self) -> Option<Arc<StorageOperationContext>> {
100        self.inner.operation_context()
101    }
102
103    fn finalize_operation(&self, committed: bool) -> StorageResult<()> {
104        self.inner.finalize_operation(committed)
105    }
106}
107
108impl<S: StorageClient + 'static> StorageCommitOps for SyncWrapper<S> {
109    fn commit_staged_writes(
110        &self,
111        transaction_id: linkrs_core::types::TransactionId,
112        intents: &[linkrs_core::wal::OutboxIntent],
113    ) -> StorageResult<linkrs_core::types::CommitLsn> {
114        self.inner.commit_staged_writes(transaction_id, intents)
115    }
116
117    fn abort_staged_writes(
118        &self,
119        transaction_id: linkrs_core::types::TransactionId,
120    ) -> StorageResult<()> {
121        self.inner.abort_staged_writes(transaction_id)
122    }
123
124    fn commit_staged_writes_with_durability(
125        &self,
126        transaction_id: linkrs_core::types::TransactionId,
127        intents: &[linkrs_core::wal::OutboxIntent],
128        durability: linkrs_core::types::DurabilityLevel,
129    ) -> StorageResult<linkrs_core::types::CommitLsn> {
130        self.inner
131            .commit_staged_writes_with_durability(transaction_id, intents, durability)
132    }
133
134    fn recover_outbox_projection(
135        &self,
136        sync_manager: &linkrs_sync::SyncManager,
137    ) -> StorageResult<usize> {
138        self.inner.recover_outbox_projection(sync_manager)
139    }
140}
141
142impl<S: StorageClient + 'static> StorageSyncContextOps for SyncWrapper<S> {
143    fn get_sync_manager(&self) -> Option<Arc<linkrs_sync::SyncManager>> {
144        self.sync_manager.clone()
145    }
146}
147
148impl<S: StorageClient + 'static> StorageRecoveryOps for SyncWrapper<S> {
149    forward_methods!(inner;
150        fn needs_recovery(&self) -> bool;
151        fn recover_from_wal(&self) -> linkrs_core::StorageResult<linkrs_transaction::wal::recovery::RecoveryStats>;
152        fn recover_from_wal_with_config(
153            &self,
154            config: linkrs_transaction::wal::recovery::RecoveryConfig,
155        ) -> linkrs_core::StorageResult<linkrs_transaction::wal::recovery::RecoveryStats>;
156        fn init_with_recovery(&self) -> linkrs_core::StorageResult<Option<linkrs_transaction::wal::recovery::RecoveryStats>>;
157    );
158}
159
160impl<S: StorageClient + 'static> StorageGcOps for SyncWrapper<S> {
161    forward_methods!(inner;
162        fn is_index_gc_running(&self) -> bool;
163        fn start_index_gc(&self) -> Option<crate::thread_pool::BackgroundTaskHandle>;
164    );
165
166    forward_methods!(inner;
167        fn stop_index_gc(&self);
168    );
169}
170
171impl<S: StorageClient + crate::StorageTtlPurgeOps + 'static> StorageTtlPurgeOps for SyncWrapper<S> {
172    forward_methods!(inner;
173        fn start_ttl_purge(&self, interval: std::time::Duration);
174        fn stop_ttl_purge(&self);
175    );
176
177    forward_methods!(inner;
178        fn is_ttl_purge_running(&self) -> bool;
179    );
180}
181
182impl<S: crate::client::StorageClient + crate::client::StorageSnapshotOps + 'static>
183    crate::client::StorageSnapshotOps for SyncWrapper<S>
184{
185    forward_methods!(inner;
186        fn get_freeze_stats(&self) -> Option<crate::engine::background_freeze::FreezeStats>;
187    );
188
189    forward_methods!(inner;
190        fn trigger_background_freeze(&self) -> linkrs_core::StorageResult<()>;
191    );
192}
linkrs_storage/engine/sync_wrapper/operation.rs

linkrs_storage/engine/sync_wrapper/
operation.rs
