vstorage/sync/plan/
mod.rs

1// Copyright 2023-2026 Hugo Osvaldo Barrera
2//
3// SPDX-License-Identifier: EUPL-1.2
4
5//! Plan for a synchronisation.
6//!
7//! Handles the generation of synchronisation operations.
8
9mod collection;
10pub mod incremental;
11pub(crate) mod items;
12mod mapping;
13
14use std::{pin::Pin, sync::Arc, task::Context, task::Poll};
15
16use futures_util::Stream;
17use tokio::sync::mpsc;
18
19use crate::{
20    Href,
21    sync::{
22        Side,
23        declare::StoragePair,
24        mapping::ResolvedMapping,
25        operation::Operation,
26        ordering::CompletionDroppedError,
27        plan::{
28            collection::{GenerateError, generate_collection_operations},
29            mapping::create_mappings_for_pair,
30        },
31        status::{FindStaleMappingsError, MappingUid, StatusDatabase, StatusError},
32    },
33};
34
35pub use incremental::IncrementalPlan;
36
37/// Error that occurs when planning synchronisation operations.
38#[derive(thiserror::Error, Debug)]
39pub enum PlanError {
40    /// Conflicting mapping haves been defined.
41    ///
42    /// Two (or more) collections on one side would be synchronised to the same collection on the
43    /// other side. The `Side` and `Href` parameters refer to the collection that has multiple
44    /// counterparts.
45    #[error("Conflicting mappings on side {0} for href {1}.")]
46    ConflictingMappings(Side, Href),
47
48    /// Discovering collections on storage A failed.
49    #[error("Discovery failed for storage A: {0}")]
50    DiscoveryFailedA(#[source] crate::Error),
51
52    /// Discovering collections on storage B failed.
53    #[error("Discovery failed for storage B: {0}")]
54    DiscoveryFailedB(#[source] crate::Error),
55
56    /// Error occurred interacting with a storage.
57    #[error("Interacting with underlying storage: {0}")]
58    Storage(#[from] crate::Error),
59
60    /// Error occurred reading the status database.
61    #[error("Querying status database: {0}")]
62    StatusDb(#[from] StatusError),
63
64    /// Error querying status database for stale mappings.
65    #[error("Finding stale mappings: {0}")]
66    FindStaleMappings(#[from] FindStaleMappingsError),
67
68    /// Completion handle dropped before signaling.
69    #[error("Completion handle dropped before signaling: {0}")]
70    CompletionDropped(#[from] CompletionDroppedError),
71}
72
73/// Stream generating actions that would synchronise a pair of storages.
74///
75/// In order to inspect the stream or export it into a human-readable format, collect it into
76/// memory first.
77///
78/// Operations are streamed in order:
79///
80/// 1. Collection operations (create/update).
81/// 2. Item operation.
82/// 3. Property operations.
83/// 4. Collection deletions (after items cleared).
84///
85/// It is safe to run operations concurrently; operations contain handles to ensure dependant
86/// operations wait for others before running.
87pub struct Plan(mpsc::Receiver<Result<Operation, PlanError>>);
88
89impl Plan {
90    /// Create a new plan for a given storage pair.
91    ///
92    /// # Errors
93    ///
94    /// Returns an error if collection discovery or mapping resolution fails.
95    pub async fn new(
96        pair: StoragePair,
97        status: Option<Arc<StatusDatabase>>,
98    ) -> Result<Plan, PlanError> {
99        let mappings = create_mappings_for_pair(&pair).await?;
100        let rx = Plan::receiver(pair, status, mappings)?;
101        Ok(Plan(rx))
102    }
103
104    pub(self) fn receiver(
105        pair: StoragePair,
106        status: Option<Arc<StatusDatabase>>,
107        mappings: Vec<Arc<ResolvedMapping>>,
108    ) -> Result<mpsc::Receiver<Result<Operation, PlanError>>, PlanError> {
109        let mut stale_mappings = None;
110        if let Some(ref status) = status {
111            let mut active_uids = Vec::new();
112            for mapping in &mappings {
113                if let Ok(Some(uid)) =
114                    status.get_mapping_uid(mapping.a().href(), mapping.b().href())
115                {
116                    active_uids.push(uid);
117                }
118            }
119
120            let stale = status.find_stale_mappings(active_uids.into_iter())?;
121            if !stale.is_empty() {
122                stale_mappings = Some(stale);
123            }
124        }
125
126        let (tx, rx) = mpsc::channel(4);
127
128        tokio::spawn(run_plan_generator(
129            tx,
130            pair,
131            status,
132            mappings,
133            stale_mappings,
134        ));
135
136        Ok(rx)
137    }
138}
139
140/// Background task which yields operations via a channel.
141async fn run_plan_generator(
142    tx: mpsc::Sender<Result<Operation, PlanError>>,
143    pair: StoragePair,
144    status: Option<Arc<StatusDatabase>>,
145    mappings: Vec<Arc<ResolvedMapping>>,
146    stale_mappings: Option<Vec<MappingUid>>,
147) {
148    if let Some(stale_uids) = stale_mappings {
149        let op = Operation::FlushStaleMappings { stale_uids };
150        if tx.send(Ok(op)).await.is_err() {
151            return;
152        }
153    }
154
155    for mapping in mappings {
156        match generate_collection_operations(&tx, &pair, mapping, status.as_deref()).await {
157            Ok(()) => {} // Continue.
158            Err(GenerateError::Plan(e)) => {
159                if tx.send(Err(e)).await.is_err() {
160                    return;
161                }
162            }
163            Err(GenerateError::ChannelClosed) => return,
164        }
165    }
166}
167
168impl Stream for Plan {
169    type Item = Result<Operation, PlanError>;
170
171    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
172        self.0.poll_recv(cx)
173    }
174}
175
176#[cfg(test)]
177mod test {
178    use std::{str::FromStr, sync::Arc};
179
180    use tempfile::Builder;
181
182    use crate::{
183        CollectionId, ItemKind,
184        base::Storage,
185        sync::declare::{CollectionDescription, StoragePair, SyncedCollection},
186        vdir::VdirStorage,
187    };
188
189    use super::{PlanError, create_mappings_for_pair};
190
191    #[tokio::test]
192    async fn test_plan_duplicate_mapping() {
193        let dir_a = Builder::new().prefix("vstorage").tempdir().unwrap();
194        let dir_b = Builder::new().prefix("vstorage").tempdir().unwrap();
195
196        let storage_a = Arc::new(
197            VdirStorage::builder(dir_a.path().to_path_buf().try_into().unwrap())
198                .unwrap()
199                .build(ItemKind::Calendar),
200        );
201        let storage_b = Arc::from(
202            VdirStorage::builder(dir_b.path().to_path_buf().try_into().unwrap())
203                .unwrap()
204                .build(ItemKind::Calendar),
205        );
206
207        // Duplicate mapping
208        let collection = CollectionId::from_str("test").unwrap();
209        let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
210            .with_mapping(SyncedCollection::direct(collection.clone()))
211            .with_mapping(SyncedCollection::direct(collection));
212
213        let err = create_mappings_for_pair(&pair).await.unwrap_err();
214        assert!(matches!(err, PlanError::ConflictingMappings(..)));
215    }
216
217    #[tokio::test]
218    async fn test_plan_conflicting_mapping() {
219        let dir_a = Builder::new().prefix("vstorage").tempdir().unwrap();
220        let dir_b = Builder::new().prefix("vstorage").tempdir().unwrap();
221
222        let storage_a = Arc::new(
223            VdirStorage::builder(dir_a.path().to_path_buf().try_into().unwrap())
224                .unwrap()
225                .build(ItemKind::Calendar),
226        );
227        let storage_b = Arc::from(
228            VdirStorage::builder(dir_b.path().to_path_buf().try_into().unwrap())
229                .unwrap()
230                .build(ItemKind::Calendar),
231        );
232        // This sync has duplicate items.
233        let collection = CollectionId::from_str("test").unwrap();
234        let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
235            .with_mapping(SyncedCollection::direct(collection.clone()))
236            .with_mapping(SyncedCollection::mapped(
237                "test".to_string(),
238                CollectionDescription::Id { id: collection },
239                CollectionDescription::Id {
240                    id: CollectionId::from_str("test_2").unwrap(),
241                },
242            ));
243
244        let err = create_mappings_for_pair(&pair).await.unwrap_err();
245        assert!(matches!(err, PlanError::ConflictingMappings(..)));
246    }
247
248    #[tokio::test]
249    async fn test_plan_same_from_both_sides() {
250        let dir_a = Builder::new().prefix("vstorage").tempdir().unwrap();
251        let dir_b = Builder::new().prefix("vstorage").tempdir().unwrap();
252
253        let storage_a = Arc::new(
254            VdirStorage::builder(dir_a.path().to_path_buf().try_into().unwrap())
255                .unwrap()
256                .build(ItemKind::Calendar),
257        );
258        let storage_b = Arc::from(
259            VdirStorage::builder(dir_b.path().to_path_buf().try_into().unwrap())
260                .unwrap()
261                .build(ItemKind::Calendar),
262        );
263        // `from_a` and `from_b` with collection existing on both sides.
264        // This particular scenario is special-cased.
265        std::fs::create_dir(dir_a.path().join("one")).unwrap();
266        std::fs::create_dir(dir_b.path().join("one")).unwrap();
267
268        let disco = storage_a.discover_collections().await.unwrap();
269        assert_eq!(disco.collections().len(), 1);
270
271        let pair = StoragePair::new(storage_a.clone(), storage_b.clone())
272            .with_all_from_a()
273            .with_all_from_b();
274
275        let mappings = create_mappings_for_pair(&pair).await.unwrap();
276        // When the same collection exists on both sides and we discover from both,
277        // it should result in one mapping (not duplicated).
278        assert_eq!(mappings.len(), 1);
279    }
280}
vstorage/sync/plan/mod.rs

vstorage/sync/plan/
mod.rs
