1use std::sync::Arc;
2
3use crate::cursor::{EdgeCursor, IndexCursor, IndexRow, IndexScanPlan, ScanOptions, VertexCursor};
4use crate::macros::{
5 forward_methods, forward_timed_read_methods, forward_timed_write_methods, forward_undo_methods,
6};
7use crate::{
8 StorageAdmin, StorageAuthOps, StorageClient, StorageCommitOps, StorageGcOps,
9 StorageOperationContext, StorageOperationContextOps, StoragePersistenceOps, StorageReader,
10 StorageRecoveryOps, StorageSchemaContextOps, StorageSchemaOps, StorageSnapshotOps,
11 StorageStats, StorageSyncContextOps, StorageTtlPurgeOps, StorageWriter,
12};
13use linkrs_core::metadata::{IndexMetadataManager, SchemaManager};
14use linkrs_core::types::{
15 EdgeTypeInfo, Index, InsertEdgeInfo, InsertVertexInfo, LabelId, PasswordInfo, PropertyDef,
16 SpaceInfo, TagInfo, UpdateInfo, UserAlterInfo, UserInfo, VertexId,
17};
18use linkrs_core::{
19 Edge, EdgeDeleteKey, EdgeDirection, RoleType, StorageError, StorageResult, Value, Vertex,
20};
21use linkrs_sync::SyncManager;
22
23pub struct MetricsStorage<S: StorageClient> {
24 inner: S,
25 stats: Option<Arc<linkrs_metrics::StatsManager>>,
26}
27
28impl<S: StorageClient> MetricsStorage<S> {
29 pub fn new(inner: S) -> Self {
30 Self { inner, stats: None }
31 }
32
33 pub fn with_stats(inner: S, stats: Arc<linkrs_metrics::StatsManager>) -> Self {
34 Self {
35 inner,
36 stats: Some(stats),
37 }
38 }
39
40 pub fn into_inner(self) -> S {
41 self.inner
42 }
43
44 fn record_read(&self, start: std::time::Instant) {
45 if let Some(stats) = &self.stats {
46 stats.record_storage_read(start.elapsed().as_micros() as u64);
47 }
48 }
49
50 fn record_write(&self, start: std::time::Instant) {
51 if let Some(stats) = &self.stats {
52 stats.record_storage_write(start.elapsed().as_micros() as u64);
53 }
54 }
55
56 fn timed_read<T>(
57 &self,
58 run: impl FnOnce() -> Result<T, StorageError>,
59 ) -> Result<T, StorageError> {
60 let start = std::time::Instant::now();
61 let result = run();
62 if result.is_err() {
63 if let Some(stats) = &self.stats {
64 stats.record_storage_error();
65 }
66 }
67 self.record_read(start);
68 result
69 }
70}
71
72impl<S: StorageClient> crate::stats_reader::ColumnStatsReader for MetricsStorage<S> {
73 fn vertex_column_stats(
74 &self,
75 space: &str,
76 tag: &str,
77 column: &str,
78 ) -> Option<Arc<crate::stats_reader::ColumnStatsSnapshot>> {
79 self.inner.vertex_column_stats(space, tag, column)
80 }
81
82 fn edge_column_stats(
83 &self,
84 space: &str,
85 edge_type: &str,
86 column: &str,
87 ) -> Option<Arc<crate::stats_reader::ColumnStatsSnapshot>> {
88 self.inner.edge_column_stats(space, edge_type, column)
89 }
90
91 fn vertex_table_stats(
92 &self,
93 space: &str,
94 tag: &str,
95 ) -> Option<Arc<crate::stats_reader::TableCardinalitySnapshot>> {
96 self.inner.vertex_table_stats(space, tag)
97 }
98
99 fn edge_table_stats(
100 &self,
101 space: &str,
102 edge_type: &str,
103 ) -> Option<Arc<crate::stats_reader::TableCardinalitySnapshot>> {
104 self.inner.edge_table_stats(space, edge_type)
105 }
106
107 fn stats_epoch(&self) -> u64 {
108 self.inner.stats_epoch()
109 }
110}
111
112impl<S: StorageClient + crate::AutoCommitBatchOps> crate::AutoCommitBatchOps for MetricsStorage<S> {
113 fn begin_auto_commit_batch(&self) -> StorageResult<Arc<crate::AutoCommitBatchWindow>> {
114 self.inner.begin_auto_commit_batch()
115 }
116
117 fn bind_auto_commit_statement(
118 &self,
119 window: &Arc<crate::AutoCommitBatchWindow>,
120 ) -> StorageResult<Self> {
121 let inner = self.inner.bind_auto_commit_statement(window)?;
122 Ok(Self {
123 inner,
124 stats: self.stats.clone(),
125 })
126 }
127
128 fn finalize_auto_commit_batch(
129 &self,
130 window: &crate::AutoCommitBatchWindow,
131 ) -> StorageResult<()> {
132 self.inner.finalize_auto_commit_batch(window)
133 }
134}
135
136impl<S: StorageClient + crate::AutoCommitGroupOps> crate::AutoCommitGroupOps for MetricsStorage<S> {
137 fn begin_auto_commit_group(&self) -> StorageResult<Arc<crate::AutoCommitBatchWindow>> {
138 self.inner.begin_auto_commit_group()
139 }
140
141 fn finalize_auto_commit_group(
142 &self,
143 window: &crate::AutoCommitBatchWindow,
144 ) -> StorageResult<()> {
145 self.inner.finalize_auto_commit_group(window)
146 }
147}
148
149impl<S: StorageClient> StorageReader for MetricsStorage<S> {
150 forward_timed_read_methods!(inner;
151 fn get_vertex(&self, space: &str, tag: &str, id: &VertexId) -> Result<Option<Vertex>, StorageError>;
152 fn scan_vertices_by_tag(&self, space: &str, tag: &str) -> Result<Vec<Vertex>, StorageError>;
153 );
154
155 fn get_vertex_projected(
156 &self,
157 space: &str,
158 tag: &str,
159 id: &VertexId,
160 projection: &[std::sync::Arc<str>],
161 ) -> Result<Option<Vertex>, StorageError> {
162 self.timed_read(|| self.inner.get_vertex_projected(space, tag, id, projection))
163 }
164
165 fn scan_vertices(&self, space: &str) -> Result<Vec<Vertex>, StorageError> {
166 self.timed_read(|| self.inner.scan_vertices(space))
167 }
168
169 fn get_edge(
170 &self,
171 space: &str,
172 src: &VertexId,
173 dst: &VertexId,
174 edge_type: &str,
175 rank: i64,
176 ) -> Result<Option<Edge>, StorageError> {
177 self.timed_read(|| self.inner.get_edge(space, src, dst, edge_type, rank))
178 }
179
180 fn get_node_edges(
181 &self,
182 space: &str,
183 node_id: &VertexId,
184 direction: EdgeDirection,
185 edge_types: &[String],
186 ) -> Result<Vec<Edge>, StorageError> {
187 self.timed_read(|| {
188 self.inner
189 .get_node_edges(space, node_id, direction, edge_types)
190 })
191 }
192
193 fn scan_edges_by_type(&self, space: &str, edge_type: &str) -> Result<Vec<Edge>, StorageError> {
194 self.timed_read(|| self.inner.scan_edges_by_type(space, edge_type))
195 }
196
197 fn scan_all_edges(&self, space: &str) -> Result<Vec<Edge>, StorageError> {
198 self.timed_read(|| self.inner.scan_all_edges(space))
199 }
200
201 fn count_vertices_by_tag(&self, space: &str, tag: &str) -> Result<u64, StorageError> {
202 self.timed_read(|| self.inner.count_vertices_by_tag(space, tag))
203 }
204
205 fn count_edges_by_type(&self, space: &str, edge_type: &str) -> Result<u64, StorageError> {
206 self.timed_read(|| self.inner.count_edges_by_type(space, edge_type))
207 }
208
209 fn scan_vertices_by_tag_paginated(
210 &self,
211 space: &str,
212 tag: &str,
213 offset: usize,
214 limit: usize,
215 ) -> Result<Vec<Vertex>, StorageError> {
216 self.timed_read(|| {
217 self.inner
218 .scan_vertices_by_tag_paginated(space, tag, offset, limit)
219 })
220 }
221
222 fn scan_vertices_by_prop(
223 &self,
224 space: &str,
225 tag: &str,
226 prop: &str,
227 value: &Value,
228 ) -> Result<Vec<Vertex>, StorageError> {
229 self.timed_read(|| self.inner.scan_vertices_by_prop(space, tag, prop, value))
230 }
231
232 fn get_edge_projected(
233 &self,
234 space: &str,
235 src: &VertexId,
236 dst: &VertexId,
237 edge_type: &str,
238 rank: i64,
239 projection: &[Arc<str>],
240 ) -> Result<Option<Edge>, StorageError> {
241 self.timed_read(|| {
242 self.inner
243 .get_edge_projected(space, src, dst, edge_type, rank, projection)
244 })
245 }
246
247 fn get_node_edges_projected(
248 &self,
249 space: &str,
250 node_id: &VertexId,
251 direction: EdgeDirection,
252 edge_types: &[String],
253 projection: Option<&[Arc<str>]>,
254 limit: Option<usize>,
255 ) -> Result<Vec<Edge>, StorageError> {
256 self.timed_read(|| {
257 self.inner
258 .get_node_edges_projected(space, node_id, direction, edge_types, projection, limit)
259 })
260 }
261
262 fn get_vertices_batch(
263 &self,
264 space: &str,
265 tag: &str,
266 ids: &[VertexId],
267 ) -> Result<Vec<Option<Vertex>>, StorageError> {
268 self.timed_read(|| self.inner.get_vertices_batch(space, tag, ids))
269 }
270
271 fn get_vertices_projected_batch(
272 &self,
273 space: &str,
274 tag: &str,
275 ids: &[VertexId],
276 projection: &[std::sync::Arc<str>],
277 ) -> Result<Vec<Option<Vertex>>, StorageError> {
278 self.timed_read(|| {
279 self.inner
280 .get_vertices_projected_batch(space, tag, ids, projection)
281 })
282 }
283
284 fn neighbor_dst_ids_batch(
285 &self,
286 space: &str,
287 src_ids: &[VertexId],
288 direction: EdgeDirection,
289 edge_types: &[String],
290 ) -> Result<Vec<Vec<VertexId>>, StorageError> {
291 self.timed_read(|| {
292 self.inner
293 .neighbor_dst_ids_batch(space, src_ids, direction, edge_types)
294 })
295 }
296
297 fn out_degree_batch(
298 &self,
299 space: &str,
300 src_ids: &[VertexId],
301 direction: EdgeDirection,
302 edge_types: &[String],
303 ) -> Result<Vec<usize>, StorageError> {
304 self.timed_read(|| {
305 self.inner
306 .out_degree_batch(space, src_ids, direction, edge_types)
307 })
308 }
309
310 fn scan_edges_by_type_paginated(
311 &self,
312 space: &str,
313 edge_type: &str,
314 offset: usize,
315 limit: usize,
316 ) -> Result<Vec<Edge>, StorageError> {
317 self.timed_read(|| {
318 self.inner
319 .scan_edges_by_type_paginated(space, edge_type, offset, limit)
320 })
321 }
322
323 fn lookup_edges_by_property_range(
324 &self,
325 space: &str,
326 edge_type: &str,
327 prop_name: &str,
328 lower: Option<&Value>,
329 upper: Option<&Value>,
330 include_lower: bool,
331 include_upper: bool,
332 ) -> Result<Vec<Edge>, StorageError> {
333 self.timed_read(|| {
334 self.inner.lookup_edges_by_property_range(
335 space,
336 edge_type,
337 prop_name,
338 lower,
339 upper,
340 include_lower,
341 include_upper,
342 )
343 })
344 }
345
346 fn lookup_index(
347 &self,
348 space: &str,
349 index: &str,
350 value: &Value,
351 ) -> Result<Vec<Value>, StorageError> {
352 self.timed_read(|| self.inner.lookup_index(space, index, value))
353 }
354
355 fn get_vertex_with_schema(
356 &self,
357 space: &str,
358 tag: &str,
359 id: &Value,
360 ) -> Result<Option<(TagInfo, Vec<u8>)>, StorageError> {
361 self.timed_read(|| self.inner.get_vertex_with_schema(space, tag, id))
362 }
363
364 fn get_edge_with_schema(
365 &self,
366 space: &str,
367 edge_type: &str,
368 src: &Value,
369 dst: &Value,
370 ) -> Result<Option<(EdgeTypeInfo, Vec<u8>)>, StorageError> {
371 self.timed_read(|| self.inner.get_edge_with_schema(space, edge_type, src, dst))
372 }
373
374 fn scan_vertices_with_schema(
375 &self,
376 space: &str,
377 tag: &str,
378 ) -> Result<Vec<(TagInfo, Vec<u8>)>, StorageError> {
379 self.timed_read(|| self.inner.scan_vertices_with_schema(space, tag))
380 }
381
382 fn scan_edges_with_schema(
383 &self,
384 space: &str,
385 edge_type: &str,
386 ) -> Result<Vec<(EdgeTypeInfo, Vec<u8>)>, StorageError> {
387 self.timed_read(|| self.inner.scan_edges_with_schema(space, edge_type))
388 }
389
390 fn create_vertex_cursor(
391 &self,
392 space: &str,
393 options: &ScanOptions,
394 ) -> Result<Box<dyn VertexCursor>, StorageError> {
395 self.timed_read(|| self.inner.create_vertex_cursor(space, options))
396 }
397
398 fn create_edge_cursor(
399 &self,
400 space: &str,
401 options: &ScanOptions,
402 ) -> Result<Box<dyn EdgeCursor>, StorageError> {
403 self.timed_read(|| self.inner.create_edge_cursor(space, options))
404 }
405
406 fn create_index_cursor(
407 &self,
408 plan: &IndexScanPlan,
409 ) -> Result<Box<dyn IndexCursor<Row = IndexRow>>, StorageError> {
410 self.timed_read(|| self.inner.create_index_cursor(plan))
411 }
412
413 forward_methods!(inner;
414 fn layout_version(&self) -> u64;
415 fn vertex_id_domain(&self, space: &str) -> Option<std::ops::Range<i64>>;
416 fn enable_edge_property_index(&self, space: &str, edge_type: &str, pool_capacity: u64) -> Result<bool, StorageError>;
417 fn has_edge_property_index(&self, space: &str, edge_type: &str) -> Result<bool, StorageError>;
418 fn disable_edge_property_index(&self, space: &str, edge_type: &str) -> Result<(), StorageError>;
419 fn get_space(&self, space: &str) -> Result<Option<SpaceInfo>, StorageError>;
420 fn get_space_by_id(&self, space_id: u64) -> Result<Option<SpaceInfo>, StorageError>;
421 fn list_spaces(&self) -> Result<Vec<SpaceInfo>, StorageError>;
422 fn get_space_id(&self, space: &str) -> Result<u64, StorageError>;
423 fn space_exists(&self, space: &str) -> bool;
424 fn get_tag(&self, space: &str, tag: &str) -> Result<Option<TagInfo>, StorageError>;
425 fn list_tags(&self, space: &str) -> Result<Vec<TagInfo>, StorageError>;
426 fn get_edge_type(&self, space: &str, edge_type: &str) -> Result<Option<EdgeTypeInfo>, StorageError>;
427 fn list_edge_types(&self, space: &str) -> Result<Vec<EdgeTypeInfo>, StorageError>;
428 fn resolve_edge_type_name(&self, space: &str, hash: u32) -> Result<Option<String>, StorageError>;
429 fn get_tag_index(&self, space: &str, index: &str) -> Result<Option<Index>, StorageError>;
430 fn list_tag_indexes(&self, space: &str) -> Result<Vec<Index>, StorageError>;
431 fn get_edge_index(&self, space: &str, index: &str) -> Result<Option<Index>, StorageError>;
432 fn list_edge_indexes(&self, space: &str) -> Result<Vec<Index>, StorageError>;
433 fn get_vertex_version_history(&self, space: &str, tag: &str) -> Result<Option<crate::LabelVersionHistory>, StorageError>;
434 fn get_edge_version_history(&self, space: &str, edge_type: &str) -> Result<Option<crate::LabelVersionHistory>, StorageError>;
435 fn get_vertex_schema_changes(&self, space: &str, tag: &str, from_version: u64, to_version: u64) -> Result<Vec<crate::PropertyChange>, StorageError>;
436 fn get_edge_schema_changes(&self, space: &str, edge_type: &str, from_version: u64, to_version: u64) -> Result<Vec<crate::PropertyChange>, StorageError>;
437 fn detect_vertex_breaking_changes(&self, space: &str, tag: &str, from_version: u64, to_version: u64) -> Result<Vec<crate::PropertyChange>, StorageError>;
438 fn detect_edge_breaking_changes(&self, space: &str, edge_type: &str, from_version: u64, to_version: u64) -> Result<Vec<crate::PropertyChange>, StorageError>;
439 fn list_migration_history(&self, space: &str, label: &str, is_edge: bool) -> Result<Vec<crate::MigrationHistoryRecord>, StorageError>;
440 fn get_applied_versions(&self, space: &str, label: &str, is_edge: bool) -> Result<Vec<u64>, StorageError>;
441 fn record_migration_history(&self, record: crate::MigrationHistoryRecord) -> Result<(), StorageError>;
442 fn list_all_migration_history(&self) -> Result<Vec<crate::MigrationHistoryRecord>, StorageError>;
443 );
444}
445
446impl<S: StorageClient> StorageWriter for MetricsStorage<S> {
447 forward_timed_write_methods!(inner;
448 fn insert_vertex(&mut self, space: &str, vertex: Vertex) -> Result<VertexId, StorageError>;
449 fn insert_edge(&mut self, space: &str, edge: Edge) -> Result<(), StorageError>;
450 fn update_vertex(&mut self, space: &str, vertex: Vertex) -> Result<(), StorageError>;
451 fn update_vertex_replace(&mut self, space: &str, vertex: Vertex) -> Result<(), StorageError>;
452 fn delete_vertex_with_edges(&mut self, space: &str, tag: &str, id: &VertexId) -> Result<(), StorageError>;
453 fn batch_delete_vertices_with_edges(&mut self, space: &str, tag: &str, ids: &[VertexId]) -> Result<usize, StorageError>;
454 fn batch_insert_vertices(&mut self, space: &str, vertices: Vec<Vertex>) -> Result<Vec<VertexId>, StorageError>;
455 fn update_edge(&mut self, space: &str, edge: Edge) -> Result<(), StorageError>;
456 fn update_edge_replace(&mut self, space: &str, edge: Edge) -> Result<(), StorageError>;
457 fn batch_insert_edges(&mut self, space: &str, edges: Vec<Edge>) -> Result<(), StorageError>;
458 fn batch_delete_edges(&mut self, space: &str, deletes: &[EdgeDeleteKey]) -> Result<usize, StorageError>;
459 fn insert_vertex_data(&mut self, space: &str, info: &InsertVertexInfo) -> Result<bool, StorageError>;
460 fn insert_edge_data(&mut self, space: &str, info: &InsertEdgeInfo) -> Result<bool, StorageError>;
461 fn delete_vertex_data(&mut self, space: &str, tag: &str, vertex_id: &str) -> Result<bool, StorageError>;
462 fn delete_edge_data(&mut self, space: &str, src: &str, dst: &str, rank: i64) -> Result<bool, StorageError>;
463 fn update_data(&mut self, space: &str, space_id: u64, info: &UpdateInfo) -> Result<bool, StorageError>;
464 fn delete_vertex(&mut self, space: &str, tag: &str, id: &VertexId) -> Result<(), StorageError>;
465 fn delete_edge(&mut self, space: &str, src: &VertexId, dst: &VertexId, edge_type: &str, rank: i64) -> Result<(), StorageError>;
466 );
467}
468
469impl<S: StorageClient> StorageSchemaOps for MetricsStorage<S> {
470 forward_methods!(inner;
471 fn create_space(&mut self, space: &mut SpaceInfo) -> Result<bool, StorageError>;
472 fn drop_space(&mut self, space: &str) -> Result<bool, StorageError>;
473 fn clear_space(&mut self, space: &str) -> Result<bool, StorageError>;
474 fn create_tag(&mut self, space: &str, tag: &TagInfo) -> Result<u32, StorageError>;
475 fn alter_tag(&mut self, space: &str, tag: &str, additions: Vec<PropertyDef>, deletions: Vec<String>, new_comment: Option<String>) -> Result<bool, StorageError>;
476 fn rename_vertex_property(&mut self, label: LabelId, old_name: &str, new_name: &str) -> Result<(), StorageError>;
477 fn rename_tag_property(&mut self, space: &str, tag: &str, old_name: &str, new_name: &str) -> Result<bool, StorageError>;
478 fn rename_tag(&mut self, space: &str, old_name: &str, new_name: &str) -> Result<bool, StorageError>;
479 fn drop_tag(&mut self, space: &str, tag: &str) -> Result<bool, StorageError>;
480 fn create_edge_type(&mut self, space: &str, edge: &EdgeTypeInfo) -> Result<u32, StorageError>;
481 fn alter_edge_type(&mut self, space: &str, edge_type: &str, additions: Vec<PropertyDef>, deletions: Vec<String>, new_comment: Option<String>) -> Result<bool, StorageError>;
482 fn rename_edge_type(&mut self, space: &str, old_name: &str, new_name: &str) -> Result<bool, StorageError>;
483 fn rename_edge_property(&mut self, space: &str, edge_type: &str, old_name: &str, new_name: &str) -> Result<bool, StorageError>;
484 fn update_edge_endpoints(&mut self, space: &str, edge_type: &str, src_tag: &str, dst_tag: &str) -> Result<bool, StorageError>;
485 fn drop_edge_type(&mut self, space: &str, edge_type: &str) -> Result<bool, StorageError>;
486 fn create_tag_index(&mut self, space: &str, info: &Index) -> Result<bool, StorageError>;
487 fn drop_tag_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
488 fn rebuild_tag_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
489 fn create_edge_index(&mut self, space: &str, info: &Index) -> Result<bool, StorageError>;
490 fn drop_edge_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
491 fn rebuild_edge_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
492 );
493}
494
495impl<S: StorageClient> StorageAuthOps for MetricsStorage<S> {
496 forward_methods!(inner;
497 fn change_password(&mut self, info: &PasswordInfo) -> Result<bool, StorageError>;
498 fn create_user(&mut self, info: &UserInfo) -> Result<bool, StorageError>;
499 fn alter_user(&mut self, info: &UserAlterInfo) -> Result<bool, StorageError>;
500 fn drop_user(&mut self, username: &str) -> Result<bool, StorageError>;
501 fn grant_role(&mut self, username: &str, space_id: i64, role: RoleType) -> Result<bool, StorageError>;
502 fn revoke_role(&mut self, username: &str, space_id: i64) -> Result<bool, StorageError>;
503 );
504
505 forward_methods!(inner;
506 fn user_exists(&self, username: &str) -> bool;
507 fn list_users(&self) -> Vec<String>;
508 fn get_user(&self, username: &str) -> Option<UserInfo>;
509 fn update_last_login(&self, username: &str) -> Result<bool, StorageError>;
510 fn list_user_roles(&self, username: &str) -> Vec<(i64, RoleType)>;
511 fn list_all_user_roles(&self) -> Vec<(String, Vec<(i64, RoleType)>)>;
512 );
513}
514
515impl<S: StorageClient> StorageAdmin for MetricsStorage<S> {
516 forward_methods!(inner;
517 fn load_from_disk(&mut self) -> Result<(), StorageError>;
518 fn repair_dangling_edges(&mut self, space: &str) -> Result<usize, StorageError>;
519 );
520
521 forward_methods!(inner;
522 fn save_to_disk(&self) -> Result<(), StorageError>;
523 fn get_storage_stats(&self) -> StorageStats;
524 fn get_db_path(&self) -> &str;
525 fn find_dangling_edges(&self, space: &str) -> Result<Vec<Edge>, StorageError>;
526 fn purge_expired(&self, space: &str, target: Option<crate::PurgeTarget<'_>>) -> Result<Vec<crate::TtlPurgeEntry>, StorageError>;
527 );
528}
529
530impl<S: StorageClient> StoragePersistenceOps for MetricsStorage<S> {
531 forward_methods!(inner;
532 fn flush(&self) -> Result<(), StorageError>;
533 fn save_data(&self) -> linkrs_core::StorageResult<()>;
534 fn save_data_to_dir(&self, dir: &std::path::Path) -> linkrs_core::StorageResult<()>;
535 fn create_checkpoint(&self) -> linkrs_core::StorageResult<Option<crate::CheckpointStats>>;
536 fn verify_snapshot(&self, snapshot_id: u64) -> linkrs_core::StorageResult<bool>;
537 fn cleanup_snapshots(&self) -> linkrs_core::StorageResult<usize>;
538 fn snapshot_stats(&self) -> crate::SnapshotStats;
539 fn persistence_diagnostics(&self) -> Option<crate::PersistenceDiagnostics>;
540 fn compact(&self, config: &linkrs_core::types::CompactConfig) -> linkrs_core::StorageResult<()>;
541 fn auto_flush_if_needed(&self) -> linkrs_core::StorageResult<bool>;
542 fn auto_checkpoint_if_needed(&self) -> linkrs_core::StorageResult<Option<crate::CheckpointStats>>;
543 fn should_flush(&self) -> bool;
544 fn should_checkpoint(&self) -> bool;
545 );
546}
547
548impl<S: StorageClient + StorageSchemaContextOps> StorageSchemaContextOps for MetricsStorage<S> {
549 forward_methods!(inner;
550 fn get_schema_manager(&self) -> Option<Arc<SchemaManager>>;
551 fn get_index_metadata_manager(&self) -> Option<Arc<dyn IndexMetadataManager>>;
552 );
553}
554
555impl<S: StorageClient> StorageOperationContextOps for MetricsStorage<S> {
556 fn bind_auto_commit_context(&self) -> StorageResult<Self> {
557 Ok(Self {
558 inner: self.inner.bind_auto_commit_context()?,
559 stats: self.stats.clone(),
560 })
561 }
562
563 fn bind_operation_context(&self, context: StorageOperationContext) -> Self {
564 Self {
565 inner: self.inner.bind_operation_context(context),
566 stats: self.stats.clone(),
567 }
568 }
569
570 fn bind_read_operation_context(&self) -> StorageResult<Self> {
571 Ok(Self {
572 inner: self.inner.bind_read_operation_context()?,
573 stats: self.stats.clone(),
574 })
575 }
576
577 fn operation_context(&self) -> Option<Arc<StorageOperationContext>> {
578 self.inner.operation_context()
579 }
580
581 fn finalize_operation(&self, committed: bool) -> linkrs_core::StorageResult<()> {
582 self.inner.finalize_operation(committed)
583 }
584}
585
586impl<S: StorageClient> StorageCommitOps for MetricsStorage<S> {
587 forward_methods!(inner;
588 fn commit_staged_writes(&self, transaction_id: linkrs_core::types::TransactionId, intents: &[linkrs_core::wal::OutboxIntent]) -> linkrs_core::StorageResult<linkrs_core::types::CommitLsn>;
589 fn abort_staged_writes(&self, transaction_id: linkrs_core::types::TransactionId) -> linkrs_core::StorageResult<()>;
590 fn recover_outbox_projection(&self, sync_manager: &linkrs_sync::SyncManager) -> linkrs_core::StorageResult<usize>;
591 );
592}
593
594impl<S: StorageClient + StorageSyncContextOps> StorageSyncContextOps for MetricsStorage<S> {
595 forward_methods!(inner;
596 fn get_sync_manager(&self) -> Option<Arc<SyncManager>>;
597 );
598}
599
600impl<S: StorageClient> StorageRecoveryOps for MetricsStorage<S> {
601 forward_methods!(inner;
602 fn needs_recovery(&self) -> bool;
603 fn recover_from_wal(&self) -> linkrs_core::StorageResult<linkrs_transaction::wal::recovery::RecoveryStats>;
604 fn recover_from_wal_with_config(
605 &self,
606 config: linkrs_transaction::wal::recovery::RecoveryConfig,
607 ) -> linkrs_core::StorageResult<linkrs_transaction::wal::recovery::RecoveryStats>;
608 fn init_with_recovery(&self) -> linkrs_core::StorageResult<Option<linkrs_transaction::wal::recovery::RecoveryStats>>;
609 );
610}
611
612impl<S: StorageClient> StorageGcOps for MetricsStorage<S> {
613 forward_methods!(inner;
614 fn is_index_gc_running(&self) -> bool;
615 fn start_index_gc(&self) -> Option<crate::thread_pool::BackgroundTaskHandle>;
616 );
617
618 forward_methods!(inner;
619 fn stop_index_gc(&self);
620 );
621}
622
623impl<S: StorageClient + crate::StorageTtlPurgeOps> StorageTtlPurgeOps for MetricsStorage<S> {
624 forward_methods!(inner;
625 fn start_ttl_purge(&self, interval: std::time::Duration);
626 fn stop_ttl_purge(&self);
627 );
628
629 forward_methods!(inner;
630 fn is_ttl_purge_running(&self) -> bool;
631 );
632}
633
634impl<S: StorageClient> std::fmt::Debug for MetricsStorage<S> {
635 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
636 f.debug_struct("MetricsStorage")
637 .field("inner", &self.inner)
638 .finish()
639 }
640}
641
642impl<S: StorageClient> Clone for MetricsStorage<S>
643where
644 S: Clone,
645{
646 fn clone(&self) -> Self {
647 Self {
648 inner: self.inner.clone(),
649 stats: self.stats.clone(),
650 }
651 }
652}
653
654impl<S: crate::client::StorageClient + StorageSnapshotOps + 'static>
655 crate::client::StorageSnapshotOps for MetricsStorage<S>
656{
657 fn get_freeze_stats(&self) -> Option<crate::engine::background_freeze::FreezeStats> {
658 self.inner.get_freeze_stats()
659 }
660
661 fn trigger_background_freeze(&self) -> linkrs_core::StorageResult<()> {
662 self.inner.trigger_background_freeze()
663 }
664}
665
666impl<S: linkrs_transaction::UndoTarget + StorageClient> linkrs_transaction::UndoTarget
667 for MetricsStorage<S>
668{
669 forward_undo_methods!(inner;
670 fn delete_vertex_type(&self, label: linkrs_core::types::LabelId) -> linkrs_transaction::undo_log::UndoLogResult<()>;
671 fn delete_edge_type(&self, edge_key: linkrs_core::types::EdgeKey) -> linkrs_transaction::undo_log::UndoLogResult<()>;
672 fn delete_vertex(&self, vertex: linkrs_core::types::VertexIdentifier, ts: linkrs_transaction::wal::Timestamp) -> linkrs_transaction::undo_log::UndoLogResult<()>;
673 fn delete_edge(&self, edge_ctx: linkrs_core::types::EdgeDeletionContext) -> linkrs_transaction::undo_log::UndoLogResult<()>;
674 fn restore_edge(&self, edge: linkrs_core::types::EdgeIdentifier, properties: Vec<(std::sync::Arc<str>, linkrs_core::Value)>, ts: linkrs_transaction::wal::Timestamp) -> linkrs_transaction::undo_log::UndoLogResult<()>;
675 fn undo_update_edge_property(&self, edge_id: linkrs_core::types::EdgeIdentifier, col_id: linkrs_core::types::ColumnId, value: linkrs_core::Value, ts: linkrs_transaction::wal::Timestamp) -> linkrs_transaction::undo_log::UndoLogResult<()>;
676 fn revert_delete_edge(&self, edge_ctx: linkrs_core::types::EdgeDeletionContext) -> linkrs_transaction::undo_log::UndoLogResult<()>;
677 fn revert_delete_vertex_properties(&self, label_name: &str, prop_names: &[Arc<str>]) -> linkrs_transaction::undo_log::UndoLogResult<()>;
678 fn revert_delete_edge_properties(&self, src_label: &str, dst_label: &str, edge_label: &str, prop_names: &[Arc<str>]) -> linkrs_transaction::undo_log::UndoLogResult<()>;
679 fn revert_delete_vertex_label(&self, label_name: &str) -> linkrs_transaction::undo_log::UndoLogResult<()>;
680 fn revert_delete_edge_label(&self, src_label: &str, dst_label: &str, edge_label: &str) -> linkrs_transaction::undo_log::UndoLogResult<()>;
681 fn revert_rename_vertex_properties(&self, label_name: &str, current_names: &[Arc<str>], original_names: &[Arc<str>]) -> linkrs_transaction::undo_log::UndoLogResult<()>;
682 fn revert_rename_edge_properties(&self, src_label: &str, dst_label: &str, edge_label: &str, current_names: &[Arc<str>], original_names: &[Arc<str>]) -> linkrs_transaction::undo_log::UndoLogResult<()>;
683 fn staged_write_mark(&self, txn_id: linkrs_core::types::TransactionId) -> Option<linkrs_core::types::StagedWriteMark>;
684 fn rollback_staged_writes(&self, txn_id: linkrs_core::types::TransactionId, mark: linkrs_core::types::StagedWriteMark) -> linkrs_transaction::undo_log::UndoLogResult<()>;
685 );
686}
687
688#[cfg(test)]
689mod tests {
690 use crate::{GraphStorage, MetricsStorage, StoragePersistenceOps};
691
692 #[test]
693 fn delegates_admin_checkpoint_operations() {
694 let temp_dir = tempfile::tempdir().expect("Failed to create temp dir");
695 let inner = GraphStorage::new_with_path(temp_dir.path().to_path_buf())
696 .expect("Failed to create GraphStorage");
697 let storage = MetricsStorage::new(inner);
698
699 let checkpoint = storage
700 .create_checkpoint()
701 .expect("checkpoint should succeed");
702
703 assert!(checkpoint.is_some());
704 }
705}