relay_knowledge/storage/sqlite/code/
software_projection_store.rs

1//! SQLite adapter for software-global projection refresh and reads.
2
3use crate::{
4    domain::{
5        CodeIndexPublicationFence, CodeSoftwareProjectionPhase, SoftwareGlobalProjection,
6        SoftwareGlobalRequest,
7    },
8    storage::{SoftwareProjectionStore, StorageError, StorageFuture},
9};
10
11use super::{ensure_queryable_code_scope, lifecycle};
12use crate::storage::sqlite::{SqliteGraphStore, software};
13
14impl SoftwareProjectionStore for SqliteGraphStore {
15    fn refresh_software_global_projection(
16        &self,
17        source_scope: String,
18    ) -> StorageFuture<'_, SoftwareGlobalProjection> {
19        self.run(move |connection| {
20            ensure_queryable_code_scope(connection, &source_scope)?;
21            software::refresh_projection(connection, &source_scope)
22        })
23    }
24
25    fn refresh_software_global_projection_with_fence(
26        &self,
27        source_scope: String,
28        fence: CodeIndexPublicationFence,
29    ) -> StorageFuture<'_, SoftwareGlobalProjection> {
30        let this = self.clone();
31        Box::pin(async move {
32            let mut completed_phases = 0usize;
33            loop {
34                if completed_phases >= CodeSoftwareProjectionPhase::COUNT {
35                    return Err(StorageError::Invariant(format!(
36                        "software projection for scope '{source_scope}' exceeded its durable phase bound"
37                    )));
38                }
39                let step_scope = source_scope.clone();
40                let step_fence = fence.clone();
41                let authority_path = this.publication_authority_path.clone();
42                let advance = this
43                    .run(move |connection| {
44                        let guard = lifecycle::publication_fence::prepare_guard(
45                            connection,
46                            step_fence,
47                            authority_path.as_deref(),
48                        )?;
49                        software::advance_fenced_projection(connection, &step_scope, &guard)
50                    })
51                    .await?;
52                match advance {
53                    software::FencedProjectionAdvance::Complete => break,
54                    software::FencedProjectionAdvance::Pending { checkpoint_state } => {
55                        completed_phases += 1;
56                        tracing::debug!(
57                            source_scope,
58                            checkpoint_state,
59                            "durable software projection phase committed"
60                        );
61                        // Keep the projection future pending between writer quanta so the
62                        // application lease heartbeat can acquire the released writer.
63                        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
64                    }
65                }
66            }
67            this.run_read_snapshot(move |connection| {
68                software::refreshed_fenced_projection(connection, &source_scope)
69            })
70            .await
71        })
72    }
73
74    fn software_global_projection(
75        &self,
76        request: SoftwareGlobalRequest,
77    ) -> StorageFuture<'_, SoftwareGlobalProjection> {
78        self.run_read_snapshot(move |connection| software::projection(connection, request))
79    }
80
81    fn software_global_projection_for_scope(
82        &self,
83        source_scope: String,
84        request: SoftwareGlobalRequest,
85    ) -> StorageFuture<'_, SoftwareGlobalProjection> {
86        self.run_read_snapshot(move |connection| {
87            ensure_queryable_code_scope(connection, &source_scope)?;
88            software::projection_for_scope(connection, &source_scope, request)
89        })
90    }
91}
relay_knowledge/storage/sqlite/code/software_projection_store.rs

relay_knowledge/storage/sqlite/code/
software_projection_store.rs
