coven_storage/
memory_upload.rs

1//! Atomic conditional publication in the deterministic provider.
2use super::*;
3use crate::session::{part_span, Confirmation, PlanStart, StartSent};
4
5/// A planned session's start as it goes out: past any hold, the publication
6/// gate is checked again, as S3 checks it as the request goes out, and a start
7/// it lets through has gone out (Appendix E.10).
8pub(super) struct MemoryStart<'a> {
9    publication: &'a PublicationRequest,
10    sent: &'a StartSent,
11}
12
13impl MemoryStart<'_> {
14    pub(super) fn go(&self) -> Result<(), StorageError> {
15        self.publication.check()?;
16        self.sent.going();
17        Ok(())
18    }
19}
20
21pub(super) struct Pending {
22    device: coven_foundation::id_source::DeviceId,
23    /// When the session began, by the provider's clock.
24    begun: coven_format::value::StorageTime,
25    target: UploadTarget,
26    total: u64,
27    pub(super) bytes: Vec<u8>,
28    /// An independent session's parts, by index from zero: each kept once,
29    /// a part sent again taking its own place.
30    parts: BTreeMap<u64, Vec<u8>>,
31    part_size: usize,
32}
33impl Pending {
34    fn check(&self, session: &UploadSession) -> Result<(), StorageError> {
35        if self.target != session.target
36            || self.total != session.total
37            || self.part_size != session.part_size
38        {
39            return Err(StorageFailure::SessionMismatch.into());
40        }
41        let lost = match &session.state {
42            SessionState::MemoryParts { parts, .. } => {
43                parts.iter().any(|index| !self.parts.contains_key(index))
44            }
45            _ => {
46                (self.bytes.len() as u64) < session.confirmed
47                    || self.bytes.len() as u64 > self.total
48            }
49        };
50        if lost {
51            return Err(
52                StorageFailure::Protocol.with_source("memory provider lost confirmed progress")
53            );
54        }
55        Ok(())
56    }
57    /// The bytes publishing hands over: an independent session's parts in
58    /// order, which must be every one; another's bytes as they arrived.
59    fn assembled(self) -> Result<Vec<u8>, StorageError> {
60        if self.parts.is_empty() {
61            return Ok(self.bytes);
62        }
63        let count = self.total.div_ceil(self.part_size as u64);
64        if self.parts.len() as u64 != count {
65            return Err(StorageFailure::InvalidPart.into());
66        }
67        Ok(self.parts.into_values().flatten().collect())
68    }
69}
70fn confirm_published(
71    state: &MemoryState,
72    session: &mut UploadSession,
73    id: u64,
74) -> Result<(), StorageError> {
75    let object = state.objects.get(session.path()).ok_or(StorageFailure::SessionExpired)?;
76    if object.upload_id != Some(id) || object.bytes.len() as u64 != session.total {
77        return Err(StorageFailure::AlreadyExists.into());
78    }
79    session.confirmed = session.total;
80    session.state = SessionState::Memory { id, complete: true };
81    Ok(())
82}
83
84/// Whether every session at a planned one's name is its own: a name its
85/// device alone uses, a path in its own file or snapshot folders, as on S3
86/// (Appendix E.10). A shared slot or floor path proves nothing.
87fn owns_name(session: &UploadSession) -> bool {
88    session.path().device() == Some(session.device)
89}
90
91/// End every session at a planned one's name it owns: one its first part
92/// started, whose start's reply was lost or never recorded.
93fn end_planned(state: &mut MemoryState, session: &UploadSession) {
94    state.uploads.retain(|_, pending| pending.target.path() != session.path());
95}
96
97fn check_target(state: &MemoryState, target: &UploadTarget) -> Result<(), StorageError> {
98    match target {
99        UploadTarget::Create(path) => {
100            if state.objects.contains_key(path) && !state.faults.overwrite_create {
101                return Err(StorageFailure::AlreadyExists.into());
102            }
103        }
104        UploadTarget::Replace(previous) => {
105            if state
106                .objects
107                .get(&previous.path)
108                .is_none_or(|object| object.revision.to_string() != previous.replace_token)
109            {
110                return Err(StorageFailure::Conflict.into());
111            }
112        }
113    }
114    Ok(())
115}
116
117impl MemoryStorage {
118    /// A four-byte upload to `path` the provider holds unpublished: begun,
119    /// then started by its first and only part, as an S3 multipart upload
120    /// is (Appendix E.10).
121    pub async fn started_upload(&self, path: &ObjectPath) -> Result<UploadSession, StorageError> {
122        let mut session = self.begin_upload(path, 4).await?;
123        self.upload_part(&mut session, b"part").await?;
124        Ok(session)
125    }
126    /// Whether the provider holds the unfinished session `session` names; a
127    /// plan names none until its first part starts it.
128    pub async fn holds(&self, session: &UploadSession) -> bool {
129        match session.state {
130            SessionState::Memory { id, .. } | SessionState::MemoryParts { id, .. } => {
131                self.provider.state.lock().await.uploads.contains_key(&id)
132            }
133            _ => false,
134        }
135    }
136    /// Each session still unfinished, in the order it began: the device
137    /// whose connection began it, and its destination.
138    pub async fn unfinished_uploads(
139        &self,
140    ) -> Vec<(coven_foundation::id_source::DeviceId, ObjectPath)> {
141        let state = self.provider.state.lock().await;
142        state
143            .uploads
144            .values()
145            .map(|pending| (pending.device, pending.target.path().clone()))
146            .collect()
147    }
148}
149
150impl MemoryProvider {
151    /// The sessions `device`'s connections began in `store`.
152    pub(super) async fn memory_device_uploads(
153        &self,
154        store: coven_foundation::id_source::StoreId,
155        device: coven_foundation::id_source::DeviceId,
156    ) -> Result<Vec<LeftUpload>, StorageError> {
157        self.before(MemoryRequest::Other("device_uploads")).await?;
158        let state = self.state.lock().await;
159        Ok(state
160            .uploads
161            .iter()
162            .filter(|(_, pending)| {
163                pending.device == device && pending.target.path().store() == store
164            })
165            .map(|(id, _)| LeftUpload(crate::left_upload::Left::Memory(*id)))
166            .collect())
167    }
168    /// The sessions at slots 1 to `through` of `store`, whoever began them:
169    /// the deterministic provider holds a claim's session at its slot, as
170    /// conditional completion on S3 does.
171    pub(super) async fn memory_settled_claim_uploads(
172        &self,
173        store: coven_foundation::id_source::StoreId,
174        through: u64,
175    ) -> Result<Vec<LeftUpload>, StorageError> {
176        self.before(MemoryRequest::Other("settled_claim_uploads")).await?;
177        let state = self.state.lock().await;
178        Ok(state
179            .uploads
180            .iter()
181            .filter(|(_, pending)| {
182                let path = pending.target.path();
183                path.store() == store
184                    && path.slot_number().is_some_and(|slot| slot.get() <= through)
185            })
186            .map(|(id, _)| LeftUpload(crate::left_upload::Left::Memory(*id)))
187            .collect())
188    }
189    /// The sessions at `floor`'s path begun in an earlier second than its
190    /// publication: the deterministic provider dates each session's start,
191    /// as a certified S3 profile does.
192    pub(super) async fn memory_superseded_floor_uploads(
193        &self,
194        floor: &StoredObject,
195    ) -> Result<Vec<LeftUpload>, StorageError> {
196        self.before(MemoryRequest::Other("superseded_floor_uploads")).await?;
197        let state = self.state.lock().await;
198        Ok(state
199            .uploads
200            .iter()
201            .filter(|(_, pending)| {
202                pending.target.path() == &floor.path
203                    && pending.begun.seconds() < floor.stored_at.seconds()
204            })
205            .map(|(id, _)| LeftUpload(crate::left_upload::Left::Memory(*id)))
206            .collect())
207    }
208    /// End one session another device left; one already ended succeeds.
209    pub(super) async fn memory_end_left_upload(
210        &self,
211        upload: &LeftUpload,
212    ) -> Result<(), StorageError> {
213        let crate::left_upload::Left::Memory(id) = upload.0 else {
214            return Err(StorageFailure::SessionMismatch.into());
215        };
216        self.before(MemoryRequest::Other("end_left_upload")).await?;
217        self.state.lock().await.uploads.remove(&id);
218        Ok(())
219    }
220    pub(super) async fn memory_begin_upload(
221        &self,
222        publication: &PublicationRequest,
223        target: &UploadTarget,
224        total: u64,
225        device: coven_foundation::id_source::DeviceId,
226    ) -> Result<UploadSession, StorageError> {
227        publication.check()?;
228        self.before(MemoryRequest::Other("begin_upload")).await?;
229        self.refuse_file_upload(target.path()).await?;
230        check_target(&*self.state.lock().await, target)?;
231        // The caller records the plan; the first part starts the session.
232        Ok(UploadSession {
233            device,
234            location: self.config(),
235            target: target.clone(),
236            total,
237            confirmed: 0,
238            part_size: self.part_size,
239            state: SessionState::MemoryPlanned { started: PlanStart::Unstarted },
240        })
241    }
242    pub(super) async fn memory_resume_upload(
243        &self,
244        publication: &PublicationRequest,
245        session: &mut UploadSession,
246    ) -> Result<(), StorageError> {
247        if let SessionState::MemoryPlanned { started } = session.state {
248            // Nothing is at the name of a plan whose first part was never
249            // sent, and nothing tells a stray at a shared name from another
250            // device's session (Appendix E.10).
251            if started != PlanStart::MayHaveStarted {
252                return Ok(());
253            }
254            if !owns_name(session) {
255                return Err(StorageFailure::SessionIdentityUnknown.into());
256            }
257            publication.check()?;
258            self.before(MemoryRequest::Other("resume_upload")).await?;
259            end_planned(&mut *self.state.lock().await, session);
260            session.state = SessionState::MemoryPlanned { started: PlanStart::Unstarted };
261            return Ok(());
262        }
263        publication.check()?;
264        self.before(MemoryRequest::Other("resume_upload")).await?;
265        let id = self.session_id(session)?;
266        let state = self.state.lock().await;
267        match state.uploads.get(&id) {
268            Some(pending) => {
269                pending.check(session)?;
270                match &mut session.state {
271                    // Every part the provider holds, whichever arrived first.
272                    SessionState::MemoryParts { parts, .. } => {
273                        *parts = pending.parts.keys().copied().collect();
274                        session.confirmed = pending
275                            .parts
276                            .keys()
277                            .filter_map(|index| part_span(pending.total, pending.part_size, *index))
278                            .map(|span| span.end - span.start)
279                            .sum();
280                    }
281                    _ => session.confirmed = pending.bytes.len() as u64,
282                }
283            }
284            None => confirm_published(&state, session, id)?,
285        }
286        Ok(())
287    }
288    pub(super) async fn memory_upload_part(
289        &self,
290        publication: &PublicationRequest,
291        session: &mut UploadSession,
292        bytes: &[u8],
293        end: u64,
294        sent: &ByteProgress,
295    ) -> Result<(), StorageError> {
296        if let SessionState::MemoryPlanned { started } = &mut session.state {
297            // A restored marked plan resumes first, and one whose start this
298            // process sent may have started a session already.
299            if *started == PlanStart::MayHaveStarted {
300                return Err(StorageFailure::SessionIdentityUnknown.into());
301            }
302            // Until it settles, the start may go out at any moment.
303            *started = PlanStart::MayHaveStarted;
304            let start = StartSent::default();
305            if let Err(error) = self.initiate(publication, session, &start).await {
306                session.start_failed(&start);
307                return Err(error);
308            }
309        }
310        if matches!(session.state, SessionState::MemoryParts { .. }) {
311            let offset = end - bytes.len() as u64;
312            let part = self.memory_send_part(publication, session, offset, bytes, sent).await?;
313            return session.confirm(part);
314        }
315        publication.check()?;
316        // The whole part is taken as the call starts, as for a create.
317        sent.moved(bytes.len() as u64);
318        self.before(MemoryRequest::Other("upload_part")).await?;
319        let gate = self.state.lock().await.part_gate.clone();
320        if let Some(gate) = gate {
321            gate.acquire()
322                .await
323                .map_err(|closed| StorageFailure::Protocol.with_source(closed))?
324                .forget();
325        }
326        let id = self.session_id(session)?;
327        let mut state = self.state.lock().await;
328        let pending = state.uploads.get(&id).ok_or(StorageFailure::SessionExpired)?;
329        pending.check(session)?;
330        if pending.bytes.len() as u64 != session.confirmed {
331            return Err(StorageFailure::InvalidPart.into());
332        }
333        if std::mem::replace(&mut state.faults.drop_part, false) {
334            return Err(StorageFailure::Network.into());
335        }
336        state
337            .uploads
338            .get_mut(&id)
339            .ok_or(StorageFailure::SessionExpired)?
340            .bytes
341            .extend_from_slice(bytes);
342        if std::mem::replace(&mut state.faults.lose_part_reply, false) {
343            return Err(StorageFailure::Network.into());
344        }
345        state.sent_bytes += bytes.len() as u64;
346        state.largest_part = state.largest_part.max(bytes.len());
347        session.confirmed = end;
348        Ok(())
349    }
350    /// Send part `offset..` of an independent session, as S3 takes a
351    /// multipart upload's part: kept by its index, replacing one sent
352    /// before, and confirmed for the caller to add to its session.
353    pub(super) async fn memory_send_part(
354        &self,
355        publication: &PublicationRequest,
356        session: &UploadSession,
357        offset: u64,
358        bytes: &[u8],
359        sent: &ByteProgress,
360    ) -> Result<crate::ConfirmedPart, StorageError> {
361        let SessionState::MemoryParts { id, .. } = session.state else {
362            return Err(StorageFailure::SessionMismatch.into());
363        };
364        publication.check()?;
365        // The whole part is taken as the call starts, as for a create.
366        sent.moved(bytes.len() as u64);
367        self.before(MemoryRequest::Other("upload_part")).await?;
368        let gate = self.state.lock().await.part_gate.clone();
369        if let Some(gate) = gate {
370            gate.acquire()
371                .await
372                .map_err(|closed| StorageFailure::Protocol.with_source(closed))?
373                .forget();
374        }
375        let index = offset / session.part_size as u64;
376        let mut state = self.state.lock().await;
377        let pending = state.uploads.get(&id).ok_or(StorageFailure::SessionExpired)?;
378        pending.check(session)?;
379        if !offset.is_multiple_of(session.part_size as u64)
380            || part_span(session.total, session.part_size, index)
381                .is_none_or(|span| span.end - span.start != bytes.len() as u64)
382        {
383            return Err(StorageFailure::InvalidPart.into());
384        }
385        if std::mem::replace(&mut state.faults.drop_part, false) {
386            return Err(StorageFailure::Network.into());
387        }
388        state
389            .uploads
390            .get_mut(&id)
391            .ok_or(StorageFailure::SessionExpired)?
392            .parts
393            .insert(index, bytes.to_vec());
394        if std::mem::replace(&mut state.faults.lose_part_reply, false) {
395            return Err(StorageFailure::Network.into());
396        }
397        state.sent_bytes += bytes.len() as u64;
398        state.largest_part = state.largest_part.max(bytes.len());
399        Ok(crate::ConfirmedPart(Confirmation::Memory { index }))
400    }
401    /// Start a planned session's upload, as its first part does on S3,
402    /// telling `sent` once the start goes out.
403    async fn initiate(
404        &self,
405        publication: &PublicationRequest,
406        session: &mut UploadSession,
407        sent: &StartSent,
408    ) -> Result<(), StorageError> {
409        let start = MemoryStart { publication, sent };
410        self.before_going(MemoryRequest::Other("initiate_upload"), Some(start)).await?;
411        let mut state = self.state.lock().await;
412        let id = state.next_id()?;
413        state.uploads.insert(
414            id,
415            Pending {
416                device: session.device,
417                begun: self.storage_time()?,
418                target: session.target.clone(),
419                total: session.total,
420                bytes: Vec::new(),
421                parts: BTreeMap::new(),
422                part_size: session.part_size,
423            },
424        );
425        if std::mem::replace(&mut state.faults.lose_initiation_reply, false) {
426            return Err(StorageFailure::Network.into());
427        }
428        session.state = if self.independent_parts {
429            SessionState::MemoryParts { id, parts: Vec::new(), complete: false }
430        } else {
431            SessionState::Memory { id, complete: false }
432        };
433        Ok(())
434    }
435    pub(super) async fn memory_finish_upload(
436        &self,
437        publication: &PublicationRequest,
438        session: &mut UploadSession,
439    ) -> Result<(), StorageError> {
440        publication.check()?;
441        self.before(MemoryRequest::Other("finish_upload")).await?;
442        if matches!(session.state, SessionState::MemoryPlanned { .. }) {
443            // Nothing is sent before the first part starts the session.
444            return Err(StorageFailure::InvalidPart.into());
445        }
446        let id = self.session_id(session)?;
447        let mut state = self.state.lock().await;
448        let Some(pending) = state.uploads.get(&id) else {
449            return confirm_published(&state, session, id);
450        };
451        pending.check(session)?;
452        let whole = match &session.state {
453            SessionState::MemoryParts { .. } => {
454                pending.parts.len() as u64 == session.total.div_ceil(session.part_size as u64)
455            }
456            _ => pending.bytes.len() as u64 == session.total,
457        };
458        if !whole || session.confirmed != session.total {
459            return Err(StorageFailure::InvalidPart.into());
460        }
461        check_target(&state, session.target())?;
462        if matches!(session.target(), UploadTarget::Replace(_))
463            && state.faults.refuse_replacements > 0
464        {
465            state.faults.refuse_replacements -= 1;
466            return Err(StorageFailure::Conflict.into());
467        }
468        let revision = state.next_id()?;
469        let pending = state.uploads.remove(&id).ok_or(StorageFailure::SessionExpired)?;
470        let path = pending.target.path().clone();
471        state.objects.insert(
472            path,
473            Object {
474                revision,
475                bytes: pending.assembled()?.into(),
476                stored_at: self.storage_time()?,
477                upload_id: Some(id),
478            },
479        );
480        if std::mem::replace(&mut state.faults.lose_completion_reply, false) {
481            return Err(StorageFailure::Network.into());
482        }
483        session.state = SessionState::Memory { id, complete: true };
484        Ok(())
485    }
486    pub(super) async fn memory_abort_upload(
487        &self,
488        publication: &PublicationRequest,
489        session: &UploadSession,
490    ) -> Result<(), StorageError> {
491        if let SessionState::MemoryPlanned { started } = session.state {
492            // A plan whose first part was never sent has nothing to end
493            // (Appendix E.10).
494            if started != PlanStart::MayHaveStarted {
495                return Ok(());
496            }
497            if !owns_name(session) {
498                return Err(StorageFailure::SessionIdentityUnknown.into());
499            }
500        }
501        publication.check()?;
502        self.before(MemoryRequest::Other("abort_upload")).await?;
503        let id = match session.state {
504            SessionState::MemoryPlanned { .. } => None,
505            _ => Some(self.session_id(session)?),
506        };
507        let mut state = self.state.lock().await;
508        if state.faults.abort_failures > 0 {
509            state.faults.abort_failures -= 1;
510            return Err(StorageFailure::Network.into());
511        }
512        let Some(id) = id else {
513            end_planned(&mut state, session);
514            return Ok(());
515        };
516        if let Some(pending) = state.uploads.get(&id) {
517            pending.check(session)?;
518        }
519        state.uploads.remove(&id);
520        Ok(())
521    }
522}
523
524#[cfg(test)]
525#[path = "memory_upload_tests.rs"]
526mod tests;
coven_storage/memory_upload.rs

coven_storage/
memory_upload.rs
