acyclic_fs/
distributed_fs.rs

1//! Small composition root for one distributed-filesystem deployment.
2
3use crate::demand::DemandSource;
4use crate::model::VolumeConfig;
5use crate::{
6    AsyncAuthorityStore, AsyncObjectStore, Fs, GitCompatRepository, LazyWorkspace,
7    LazyWorkspaceError, LazyWorkspaceStore, MultiRootPublicationAuthorizer,
8    MultiRootPublicationCoordinator, MultiRootPublicationStore, MultiRootPublisher,
9    OperationWindowCoordinator, Workspace, WorkspaceContextRegistry, WorkspaceContextStore,
10    WorkspaceGraph, WorkspaceId, WorkspaceLineageError, WorkspaceLineageStore, WorkspaceResolver,
11};
12use std::sync::Arc;
13
14/// Shared constructors for the semantic handles of one filesystem deployment.
15pub struct DistributedFs<A, O, S> {
16    fs: Fs<A, O>,
17    store: S,
18}
19
20impl<A, O, S: Clone> Clone for DistributedFs<A, O, S> {
21    fn clone(&self) -> Self {
22        Self::new(self.fs.clone(), self.store.clone())
23    }
24}
25
26impl<A, O, S> DistributedFs<A, O, S> {
27    /// Composes an existing filesystem and durable semantic-state store.
28    #[must_use]
29    pub const fn new(fs: Fs<A, O>, store: S) -> Self {
30        Self { fs, store }
31    }
32
33    /// Resolves one workspace by authenticated durable identity.
34    pub async fn workspace(
35        &self,
36        workspace_id: WorkspaceId,
37    ) -> Result<Workspace<A, O>, WorkspaceLineageError<S::Error>>
38    where
39        A: AsyncAuthorityStore,
40        O: AsyncObjectStore,
41        S: WorkspaceLineageStore + Clone,
42    {
43        let record = self.lineage().resolve(workspace_id).await?;
44        let workspace = self.fs.open_workspace(&record.workspace_name).await?;
45        if workspace.id() != workspace_id {
46            return Err(WorkspaceLineageError::IncompatibleState);
47        }
48        Ok(workspace)
49    }
50
51    /// Constructs the durable recursive context handle.
52    pub fn contexts(&self) -> WorkspaceContextRegistry<S>
53    where
54        S: WorkspaceContextStore + Clone,
55    {
56        WorkspaceContextRegistry::new(self.store.clone())
57    }
58
59    /// Constructs the authenticated direct-workspace lineage handle.
60    pub fn lineage(&self) -> WorkspaceGraph<S>
61    where
62        S: WorkspaceLineageStore + Clone,
63    {
64        WorkspaceGraph::new(self.store.clone())
65    }
66
67    /// Constructs one root's Git-shaped compatibility view.
68    pub fn git(&self, workspace_id: WorkspaceId) -> GitCompatRepository<S>
69    where
70        S: Clone,
71    {
72        GitCompatRepository::new(workspace_id, self.store.clone())
73    }
74
75    /// Attaches an unresolved source without enumerating or reading its tree.
76    pub async fn attach_lazy<D: DemandSource + 'static>(
77        &self,
78        name: impl AsRef<str>,
79        source: Arc<D>,
80    ) -> Result<LazyWorkspace<A, O, D, S>, LazyWorkspaceError>
81    where
82        A: AsyncAuthorityStore,
83        O: AsyncObjectStore,
84        S: LazyWorkspaceStore + Clone,
85    {
86        LazyWorkspace::attach(&self.fs, name, source, self.store.clone()).await
87    }
88
89    /// Attaches an unresolved source to a workspace with exact filesystem semantics.
90    pub async fn attach_lazy_with_config<D: DemandSource + 'static>(
91        &self,
92        name: impl AsRef<str>,
93        source: Arc<D>,
94        config: VolumeConfig,
95    ) -> Result<LazyWorkspace<A, O, D, S>, LazyWorkspaceError>
96    where
97        A: AsyncAuthorityStore,
98        O: AsyncObjectStore,
99        S: LazyWorkspaceStore + Clone,
100    {
101        LazyWorkspace::attach_with_config(&self.fs, name, source, self.store.clone(), config).await
102    }
103
104    /// Reopens an existing workspace against its unresolved source.
105    pub async fn open_lazy<D: DemandSource + 'static>(
106        &self,
107        workspace: Workspace<A, O>,
108        source: Arc<D>,
109    ) -> Result<LazyWorkspace<A, O, D, S>, LazyWorkspaceError>
110    where
111        A: AsyncAuthorityStore,
112        O: AsyncObjectStore,
113        S: LazyWorkspaceStore + Clone,
114    {
115        LazyWorkspace::open(workspace, source, self.store.clone()).await
116    }
117
118    /// Composes the SDK-owned all-roots publication state machine.
119    pub fn publications<P: MultiRootPublisher, R: MultiRootPublicationAuthorizer>(
120        &self,
121        publisher: P,
122        authorizer: R,
123    ) -> MultiRootPublicationCoordinator<S, P, R>
124    where
125        S: MultiRootPublicationStore + Clone,
126    {
127        MultiRootPublicationCoordinator::new(self.store.clone(), publisher, authorizer)
128    }
129}
130
131impl<P, O, S> DistributedFs<crate::StreamAuthorityStore<P>, O, S> {
132    /// Constructs operation windows on the same authority stream that fences
133    /// generation publication.
134    ///
135    /// Keeping the lease gate and workspace authority on one provider makes it
136    /// impossible to construct a coordinator whose leases cannot authorize the
137    /// publications they protect.
138    pub fn operations(&self) -> OperationWindowCoordinator<crate::StreamOperationWindowStore<P>> {
139        OperationWindowCoordinator::new(self.fs.operation_window_store())
140    }
141}
142
143impl<A, O, S> WorkspaceResolver<A, O> for DistributedFs<A, O, S>
144where
145    A: AsyncAuthorityStore + Send + Sync,
146    O: AsyncObjectStore + Send + Sync,
147    S: WorkspaceLineageStore + Clone,
148{
149    type Error = WorkspaceLineageError<S::Error>;
150
151    async fn resolve(&self, workspace_id: WorkspaceId) -> Result<Workspace<A, O>, Self::Error> {
152        self.workspace(workspace_id).await
153    }
154}
acyclic_fs/distributed_fs.rs

acyclic_fs/
distributed_fs.rs
