coven_storage/
session.rs

1use crate::{CloudProvider, ObjectPath, StorageConfig, StorageError, StorageFailure, StoredObject};
2use coven_crypto::SecretBytes;
3use coven_crypto::SecretText;
4use coven_foundation::id_source::DeviceId;
5use serde::{Deserialize, Serialize};
6use std::ops::Range;
7
8pub(crate) const CLOUDKIT_PART_SIZE: usize = 8 * 1024 * 1024;
9pub(crate) const DROPBOX_PART_SIZE: usize = 8 * 1024 * 1024;
10pub(crate) const ONEDRIVE_PART_SIZE: usize = 24 * 320 * 1024;
11/// The most parts one S3 multipart upload holds.
12pub(crate) const S3_MAX_PARTS: usize = 10_000;
13/// The largest body one S3 request carries: a whole object's PUT, one part,
14/// or the object one `CopyObject` copies.
15pub(crate) const S3_REQUEST_LIMIT: u64 = 5 * 1024 * 1024 * 1024;
16/// S3 part sizes are whole multiples of this.
17const S3_PART_UNIT: u64 = 8 * 1024 * 1024;
18
19/// The one condition under which all uploaded bytes may become visible.
20#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
21pub enum UploadTarget {
22    /// Publish only while the path is absent.
23    Create(ObjectPath),
24    /// Publish only while this floor revision still occupies the path.
25    Replace(StoredObject),
26}
27impl UploadTarget {
28    /// The encrypted object's destination.
29    pub fn path(&self) -> &ObjectPath {
30        match self {
31            Self::Create(path) => path,
32            Self::Replace(previous) => &previous.path,
33        }
34    }
35    pub(crate) fn check(&self) -> Result<(), StorageError> {
36        if let Self::Replace(previous) = self {
37            if !previous.path.is_replaceable() {
38                return Err(StorageFailure::InvalidPath.into());
39            }
40            previous.check(&previous.path)?;
41        }
42        Ok(())
43    }
44    pub(crate) fn error(&self, error: StorageError) -> StorageError {
45        match self {
46            Self::Create(_) => error,
47            Self::Replace(_) => error.replacement(),
48        }
49    }
50}
51
52/// A provider upload recorded by the caller's operation (§17.9, §11).
53/// The destination retains its original absent/revision condition across retries.
54///
55/// Record it before sending parts and after each call, including failed calls:
56/// a publication conflict invalidates the native session even if abort fails.
57/// Record it, too, after admitting a part that marks its plan
58/// ([`crate::PartPermit::marked`]), before sending that part. After a crash,
59/// call `resume_upload` to discover bytes accepted before a reply was lost. The
60/// bytes of this value include an upload capability: keep them secret. Dropping
61/// a session never aborts it; explicit abort abandons its pending bytes.
62/// On copy profiles the recorded token names the temporary object as well as
63/// identifying publication, and the session is recorded before the native
64/// multipart upload exists: the first part initiates it. Abort deletes it even after native multipart completion; failed
65/// copy or cleanup keeps the recording incomplete so it can be retried.
66/// OneDrive creation returns its planned temporary UUID before any provider request.
67/// After recording it, resume and record the returned native capability before
68/// sending parts. The UUID stays through receipt, copying, verification
69/// and deletion; native transfer completion alone is not publication.
70/// Providers may retain closed session data until their expiry, without publishing it.
71/// When recovery requires complete-byte comparison with a published object,
72/// resumption resets the confirmed offset to zero. Feed the encrypted
73/// bytes through `upload_part` again: they are compared to the published object,
74/// without uploading them. Only a complete match confirms publication.
75#[derive(Clone, Serialize, Deserialize)]
76#[serde(try_from = "RecordedUploadSession")]
77pub struct UploadSession {
78    /// The device whose connection began it: its temporaries and provider
79    /// session sit in that device's reserved folder (D10).
80    pub(crate) device: DeviceId,
81    pub(crate) location: StorageConfig,
82    pub(crate) target: UploadTarget,
83    pub(crate) total: u64,
84    pub(crate) confirmed: u64,
85    pub(crate) part_size: usize,
86    pub(crate) state: SessionState,
87}
88
89// The wire boundary must construct an unchecked recording before validation.
90// Keeping this private makes both Serde and `decode` return validated sessions.
91#[derive(Deserialize)]
92#[serde(deny_unknown_fields)]
93struct RecordedUploadSession {
94    device: DeviceId,
95    location: StorageConfig,
96    target: UploadTarget,
97    total: u64,
98    confirmed: u64,
99    part_size: usize,
100    state: SessionState,
101}
102
103impl TryFrom<RecordedUploadSession> for UploadSession {
104    type Error = StorageError;
105    fn try_from(recorded: RecordedUploadSession) -> Result<Self, Self::Error> {
106        let session = Self {
107            device: recorded.device,
108            location: recorded.location,
109            target: recorded.target,
110            total: recorded.total,
111            confirmed: recorded.confirmed,
112            part_size: recorded.part_size,
113            state: recorded.state,
114        };
115        session.validate()?;
116        Ok(session)
117    }
118}
119
120#[derive(Clone, Serialize, Deserialize)]
121#[serde(deny_unknown_fields)]
122pub(crate) enum SessionState {
123    S3 {
124        id: SecretText,
125        token: SecretText,
126        parts: Vec<S3Part>,
127    },
128    S3Restart {
129        id: SecretText,
130        token: SecretText,
131    },
132    /// A session whose upload name is recorded before the native multipart
133    /// upload is initiated, so no upload exists that its recording does not
134    /// name (Appendix E.10).
135    S3Planned {
136        token: SecretText,
137        started: PlanStart,
138    },
139    Dropbox {
140        id: SecretText,
141    },
142    OneDrive {
143        temporary: uuid::Uuid,
144        upload: OneDriveUpload,
145    },
146    OneDriveReplacement {
147        url: SecretText,
148    },
149    CloudKit {
150        id: SecretText,
151    },
152    VerifyPublished,
153    Complete,
154    /// The deterministic provider's session named before its first part
155    /// starts it, as an S3 multipart upload is (Appendix E.10).
156    #[cfg(any(test, feature = "test-utils"))]
157    MemoryPlanned {
158        started: PlanStart,
159    },
160    #[cfg(any(test, feature = "test-utils"))]
161    Memory {
162        id: u64,
163        complete: bool,
164    },
165    /// The deterministic provider's session whose parts are independent, as
166    /// an S3 multipart upload's are: the indexes, from zero, of the parts it
167    /// confirmed, in order.
168    #[cfg(any(test, feature = "test-utils"))]
169    MemoryParts {
170        id: u64,
171        parts: Vec<u64>,
172        complete: bool,
173    },
174}
175
176/// Whether a planned session's first part may have started its upload
177/// (§17.9, Appendix E.10). The recording keeps one fact, whether the plan is
178/// marked; what this process knows besides lives in memory only.
179#[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
180#[serde(from = "bool", into = "bool")]
181pub(crate) enum PlanStart {
182    /// Unmarked: no request that may start the upload was sent, so nothing
183    /// is at the plan's name.
184    Unstarted,
185    /// Marked by the connection's admission of the first part, which has not
186    /// been sent: this process knows nothing is at the name yet.
187    Starting,
188    /// Marked, and the first part may have started an upload at the name: it
189    /// is going out here, or went out, or the recording was restored.
190    MayHaveStarted,
191}
192
193impl From<bool> for PlanStart {
194    fn from(marked: bool) -> Self {
195        if marked {
196            Self::MayHaveStarted
197        } else {
198            Self::Unstarted
199        }
200    }
201}
202
203impl From<PlanStart> for bool {
204    fn from(start: PlanStart) -> Self {
205        start != PlanStart::Unstarted
206    }
207}
208
209/// Whether a planned session's start went out: set where the provider hands
210/// the start's request to the network, past every check made as it goes,
211/// the publication gate's and a provider cooldown's (§17.9, Appendix E.10).
212/// The start's attempts share it, so one that went out before a retry was
213/// refused still counts.
214#[derive(Clone, Default)]
215pub(crate) struct StartSent(std::sync::Arc<std::sync::atomic::AtomicBool>);
216
217impl StartSent {
218    /// The start's request goes out now.
219    pub(crate) fn going(&self) {
220        self.0.store(true, std::sync::atomic::Ordering::SeqCst);
221    }
222    fn went(&self) -> bool {
223        self.0.load(std::sync::atomic::Ordering::SeqCst)
224    }
225}
226
227#[derive(Clone, Serialize, Deserialize)]
228#[serde(deny_unknown_fields)]
229pub(crate) struct S3Part {
230    pub(crate) number: i32,
231    pub(crate) size: u64,
232    pub(crate) etag: String,
233}
234
235/// Native transfer and publication are separate; every phase retains the
236/// temporary's identity until its deletion has succeeded.
237#[derive(Clone, Serialize, Deserialize)]
238#[serde(deny_unknown_fields)]
239pub(crate) enum OneDriveUpload {
240    /// The temporary identity exists before any native request (§20.6).
241    Planned {
242        /// Only this process can know initiation has never been attempted.
243        #[serde(skip)]
244        unstarted: bool,
245    },
246    Receiving {
247        url: SecretText,
248    },
249    Uploaded {
250        item: String,
251    },
252    Copying {
253        monitor: SecretText,
254    },
255    // The copy definitively failed; cleanup may already have removed its source.
256    Refused,
257    VerifyPublished,
258    Published,
259}
260
261impl S3Part {
262    /// Whether this is one of the parts of an object of `total` bytes in
263    /// parts of `part_size`: its number names one, its size is that part's,
264    /// and the provider tagged it.
265    pub(crate) fn fits(&self, total: u64, part_size: usize) -> bool {
266        u64::try_from(self.number)
267            .ok()
268            .and_then(|number| number.checked_sub(1))
269            .and_then(|index| part_span(total, part_size, index))
270            .is_some_and(|span| span.end - span.start == self.size)
271            && !self.etag.is_empty()
272    }
273}
274
275/// The bytes of part `index`, counted from zero, of an object of `total`
276/// bytes in parts of `part_size`; none past its last part.
277pub(crate) fn part_span(total: u64, part_size: usize, index: u64) -> Option<Range<u64>> {
278    let start = index.checked_mul(part_size as u64)?;
279    (start < total).then(|| start..start.saturating_add(part_size as u64).min(total))
280}
281
282/// One part a provider whose parts are independent confirmed, for the
283/// caller to add to its session ([`UploadSession::confirm`]).
284pub struct ConfirmedPart(pub(crate) Confirmation);
285
286impl std::fmt::Debug for ConfirmedPart {
287    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
288        let index = match &self.0 {
289            Confirmation::S3(part) => i64::from(part.number) - 1,
290            #[cfg(any(test, feature = "test-utils"))]
291            Confirmation::Memory { index } => *index as i64,
292        };
293        f.debug_struct("ConfirmedPart").field("index", &index).finish()
294    }
295}
296
297pub(crate) enum Confirmation {
298    S3(S3Part),
299    /// Part `index`, from zero, of a deterministic provider's session.
300    #[cfg(any(test, feature = "test-utils"))]
301    Memory {
302        index: u64,
303    },
304}
305
306impl UploadSession {
307    /// Mark a planned session's plan as its first part is admitted, before
308    /// the part is sent: whether that changed the recording, which the caller
309    /// then saves before sending (§17.9, Appendix E.10).
310    pub(crate) fn mark_start(&mut self) -> bool {
311        let started = match &mut self.state {
312            SessionState::S3Planned { started, .. } => started,
313            #[cfg(any(test, feature = "test-utils"))]
314            SessionState::MemoryPlanned { started } => started,
315            _ => return false,
316        };
317        if *started != PlanStart::Unstarted {
318            return false;
319        }
320        *started = PlanStart::Starting;
321        true
322    }
323    /// Settle a planned session's plan after its start failed. One whose
324    /// request went out may have started an upload, and stays marked; one
325    /// refused before it went out began nothing, so its plan is unmarked
326    /// again, for the caller to save before the failure is reported
327    /// (Appendix E.10).
328    pub(crate) fn start_failed(&mut self, start: &StartSent) {
329        let started = match &mut self.state {
330            SessionState::S3Planned { started, .. } => started,
331            #[cfg(any(test, feature = "test-utils"))]
332            SessionState::MemoryPlanned { started } => started,
333            _ => return,
334        };
335        if !start.went() {
336            *started = PlanStart::Unstarted;
337        }
338    }
339    /// Whether native initiation must follow recording this identity, before
340    /// any part is sent. Resume it and save the returned session first (§20.6).
341    pub fn needs_initiation(&self) -> bool {
342        matches!(self.state, SessionState::OneDrive { upload: OneDriveUpload::Planned { .. }, .. })
343    }
344    /// The device whose connection began this upload.
345    pub fn device(&self) -> DeviceId {
346        self.device
347    }
348    /// The destination encrypted-object path.
349    pub fn path(&self) -> &ObjectPath {
350        self.target.path()
351    }
352    /// The original publication condition, retained by every resumed session.
353    pub fn target(&self) -> &UploadTarget {
354        &self.target
355    }
356    /// Bytes confirmed by the provider; only these may advance the operation.
357    pub fn confirmed_bytes(&self) -> u64 {
358        self.confirmed
359    }
360    /// Total encrypted bytes promised when the session began.
361    pub fn total_bytes(&self) -> u64 {
362        self.total
363    }
364    /// Maximum part size. After resuming a partial part, send only the bytes
365    /// remaining to the next part boundary (or the end of the object).
366    pub fn part_size(&self) -> usize {
367        self.part_size
368    }
369    /// Whether publication and provider upload cleanup have succeeded.
370    pub fn is_complete(&self) -> bool {
371        match self.state {
372            SessionState::Complete => true,
373            #[cfg(any(test, feature = "test-utils"))]
374            SessionState::Memory { complete, .. } | SessionState::MemoryParts { complete, .. } => {
375                complete
376            }
377            _ => false,
378        }
379    }
380    /// Whether this session's parts are independent: the provider keeps
381    /// each by its index, whatever order they arrive in and however often
382    /// one is sent, so they may go together (§17.9, Appendix E.10). Only an
383    /// S3 multipart upload's are, once its first part has started it.
384    pub fn parts_independent(&self) -> bool {
385        match self.state {
386            SessionState::S3 { .. } => true,
387            #[cfg(any(test, feature = "test-utils"))]
388            SessionState::MemoryParts { complete, .. } => !complete,
389            _ => false,
390        }
391    }
392    /// The bytes of each part still to send, in order: where parts are
393    /// independent, every part the provider hasn't confirmed; elsewhere only
394    /// the next, which goes before any after it. After resuming a partial
395    /// part, the next runs only to its part's end.
396    pub fn parts_to_send(&self) -> Vec<Range<u64>> {
397        if self.is_complete() || self.confirmed >= self.total {
398            return Vec::new();
399        }
400        if let Some(confirmed) = self.confirmed_parts() {
401            return (0..self.total.div_ceil(self.part_size as u64))
402                .filter(|index| confirmed.binary_search(index).is_err())
403                .filter_map(|index| part_span(self.total, self.part_size, index))
404                .collect();
405        }
406        let next = (self.confirmed / self.part_size as u64 + 1) * self.part_size as u64;
407        std::iter::once(self.confirmed..next.min(self.total)).collect()
408    }
409    /// The indexes, from zero, of an independent session's confirmed parts,
410    /// in order.
411    fn confirmed_parts(&self) -> Option<Vec<u64>> {
412        match &self.state {
413            SessionState::S3 { parts, .. } => Some(
414                parts
415                    .iter()
416                    .filter_map(|part| u64::try_from(part.number).ok()?.checked_sub(1))
417                    .collect(),
418            ),
419            #[cfg(any(test, feature = "test-utils"))]
420            SessionState::MemoryParts { parts, complete: false, .. } => Some(parts.clone()),
421            _ => None,
422        }
423    }
424    /// Add a part the provider confirmed to an independent session. A part
425    /// that isn't one of the session's, or one it holds already, is refused
426    /// and changes nothing.
427    pub fn confirm(&mut self, part: ConfirmedPart) -> Result<(), StorageError> {
428        let (total, part_size) = (self.total, self.part_size);
429        let size = match (&mut self.state, part.0) {
430            (SessionState::S3 { parts, .. }, Confirmation::S3(part)) => {
431                if !part.fits(total, part_size) {
432                    return Err(StorageFailure::Protocol
433                        .with_source("S3 confirmed a part the session doesn't have"));
434                }
435                let at = parts
436                    .binary_search_by_key(&part.number, |held| held.number)
437                    .err()
438                    .ok_or(StorageFailure::InvalidPart)?;
439                let size = part.size;
440                parts.insert(at, part);
441                size
442            }
443            #[cfg(any(test, feature = "test-utils"))]
444            (
445                SessionState::MemoryParts { parts, complete: false, .. },
446                Confirmation::Memory { index },
447            ) => {
448                let span = part_span(total, part_size, index).ok_or(StorageFailure::Protocol)?;
449                let at = parts.binary_search(&index).err().ok_or(StorageFailure::InvalidPart)?;
450                parts.insert(at, index);
451                span.end - span.start
452            }
453            _ => return Err(StorageFailure::SessionMismatch.into()),
454        };
455        self.confirmed += size;
456        Ok(())
457    }
458    /// Encode the complete recorded state without printing upload credentials.
459    pub fn encode(&self) -> Result<SecretBytes, StorageError> {
460        self.validate()?;
461        crate::secret_json::encode(self)
462    }
463    /// Read a recording only if its provider, destination, state and progress agree.
464    /// Validation covers the location, state variant, nonempty identifiers, part
465    /// counts and sizes, and confirmed byte offset; replacement requires a floor.
466    /// The adapter also checks that it names the adapter's location before any request.
467    pub fn decode(bytes: &[u8]) -> Result<Self, StorageError> {
468        let value: RecordedUploadSession = crate::secret_json::decode(bytes)?;
469        value.try_into()
470    }
471    fn validate(&self) -> Result<(), StorageError> {
472        self.location.validate()?;
473        self.target.check()?;
474        if self.part_size == 0
475            || self.total == 0
476            || self.confirmed > self.total
477            || (self.is_complete() && self.confirmed != self.total)
478        {
479            return Err(StorageFailure::InvalidPart.into());
480        }
481        let provider = self.location.provider();
482        match &self.state {
483            SessionState::S3 { id, token, parts } if provider == CloudProvider::S3 => {
484                nonempty(id)?;
485                nonempty(token)?;
486                if parts.len() > S3_MAX_PARTS {
487                    return Err(StorageFailure::InvalidPart.into());
488                }
489                // Each confirmed part once, in number order, whichever
490                // arrived first.
491                let mut confirmed = 0u64;
492                let mut previous = 0;
493                for part in parts {
494                    if part.number <= previous || !part.fits(self.total, self.part_size) {
495                        return Err(StorageFailure::InvalidPart.into());
496                    }
497                    previous = part.number;
498                    confirmed += part.size;
499                }
500                if confirmed != self.confirmed {
501                    return Err(StorageFailure::InvalidPart.into());
502                }
503            }
504            SessionState::S3Restart { id, token } if provider == CloudProvider::S3 => {
505                nonempty(id)?;
506                nonempty(token)?;
507                if self.confirmed != 0 {
508                    return Err(StorageFailure::InvalidPart.into());
509                }
510            }
511            SessionState::S3Planned { token, .. } if provider == CloudProvider::S3 => {
512                nonempty(token)?;
513                if self.confirmed != 0 {
514                    return Err(StorageFailure::InvalidPart.into());
515                }
516            }
517            SessionState::Dropbox { id } if provider == CloudProvider::Dropbox => nonempty(id)?,
518            SessionState::OneDrive { upload, .. }
519                if provider == CloudProvider::OneDrive
520                    && matches!(self.target, UploadTarget::Create(_)) =>
521            {
522                match upload {
523                    OneDriveUpload::Planned { .. } => {
524                        if self.confirmed != 0 {
525                            return Err(StorageFailure::InvalidPart.into());
526                        }
527                    }
528                    OneDriveUpload::Receiving { url } => {
529                        transfer_url(url)?;
530                        if self.confirmed == self.total {
531                            return Err(StorageFailure::InvalidPart.into());
532                        }
533                    }
534                    OneDriveUpload::Uploaded { item } => {
535                        if item.is_empty() || self.confirmed != self.total {
536                            return Err(StorageFailure::InvalidPart.into());
537                        }
538                    }
539                    OneDriveUpload::Copying { monitor } => {
540                        transfer_url(monitor)?;
541                        if self.confirmed != self.total {
542                            return Err(StorageFailure::InvalidPart.into());
543                        }
544                    }
545                    OneDriveUpload::Published | OneDriveUpload::Refused
546                        if self.confirmed != self.total =>
547                    {
548                        return Err(StorageFailure::InvalidPart.into());
549                    }
550                    OneDriveUpload::VerifyPublished if self.confirmed == self.total => {
551                        return Err(StorageFailure::InvalidPart.into());
552                    }
553                    OneDriveUpload::VerifyPublished
554                    | OneDriveUpload::Published
555                    | OneDriveUpload::Refused => {}
556                }
557            }
558            SessionState::OneDriveReplacement { url }
559                if provider == CloudProvider::OneDrive
560                    && matches!(self.target, UploadTarget::Replace(_)) =>
561            {
562                transfer_url(url)?;
563                if self.confirmed == self.total {
564                    return Err(StorageFailure::InvalidPart.into());
565                }
566            }
567            SessionState::CloudKit { id } if provider == CloudProvider::CloudKit => {
568                check_cloudkit_id(id.as_str())?
569            }
570            SessionState::VerifyPublished
571                if provider == CloudProvider::Dropbox && !self.path().is_replaceable() =>
572            {
573                if self.confirmed == self.total {
574                    return Err(StorageFailure::InvalidPart.into());
575                }
576            }
577            SessionState::Complete => {}
578            #[cfg(any(test, feature = "test-utils"))]
579            SessionState::MemoryPlanned { .. } => {
580                return if self.confirmed == 0 {
581                    Ok(())
582                } else {
583                    Err(StorageFailure::InvalidPart.into())
584                };
585            }
586            #[cfg(any(test, feature = "test-utils"))]
587            SessionState::Memory { id, .. } => {
588                return if *id != 0 { Ok(()) } else { Err(StorageFailure::InvalidPart.into()) };
589            }
590            #[cfg(any(test, feature = "test-utils"))]
591            SessionState::MemoryParts { id, parts, complete } => {
592                let mut confirmed = 0u64;
593                for (at, index) in parts.iter().enumerate() {
594                    let span = part_span(self.total, self.part_size, *index)
595                        .ok_or(StorageFailure::InvalidPart)?;
596                    if at > 0 && parts[at - 1] >= *index {
597                        return Err(StorageFailure::InvalidPart.into());
598                    }
599                    confirmed += span.end - span.start;
600                }
601                let held = if *complete { self.total } else { confirmed };
602                return if *id != 0 && held == self.confirmed {
603                    Ok(())
604                } else {
605                    Err(StorageFailure::InvalidPart.into())
606                };
607            }
608            _ => return Err(StorageFailure::SessionMismatch.into()),
609        }
610        let valid_size = match provider {
611            CloudProvider::S3 => self.part_size == s3_part_size(self.total)?,
612            CloudProvider::Dropbox => self.part_size == DROPBOX_PART_SIZE,
613            CloudProvider::OneDrive => self.part_size == ONEDRIVE_PART_SIZE,
614            CloudProvider::CloudKit => self.part_size == CLOUDKIT_PART_SIZE,
615        };
616        if !valid_size {
617            return Err(StorageFailure::InvalidPart.into());
618        }
619        Ok(())
620    }
621    pub(crate) fn check(&self, location: &StorageConfig) -> Result<(), StorageError> {
622        self.validate()?;
623        if !self.location.same_location(location) {
624            return Err(StorageFailure::SessionMismatch.into());
625        }
626        Ok(())
627    }
628    /// Where a part of `len` bytes starting `offset` bytes in ends, if it is
629    /// one this session may send now: where parts are independent, a whole
630    /// part the provider hasn't confirmed; elsewhere the next, from the
631    /// confirmed bytes to at most its part's end.
632    pub(crate) fn part_end(&self, offset: u64, len: usize) -> Result<u64, StorageError> {
633        let end = offset.checked_add(len as u64).ok_or(StorageFailure::InvalidPart)?;
634        if self.is_complete() || len == 0 {
635            return Err(StorageFailure::InvalidPart.into());
636        }
637        if let Some(confirmed) = self.confirmed_parts() {
638            let index = offset / self.part_size as u64;
639            let whole = offset.is_multiple_of(self.part_size as u64)
640                && part_span(self.total, self.part_size, index).is_some_and(|span| span.end == end);
641            if !whole || confirmed.binary_search(&index).is_ok() {
642                return Err(StorageFailure::InvalidPart.into());
643            }
644            return Ok(end);
645        }
646        if offset != self.confirmed
647            || len > self.part_size
648            || end > self.total
649            || (end != self.total && end % self.part_size as u64 != 0)
650        {
651            return Err(StorageFailure::InvalidPart.into());
652        }
653        Ok(end)
654    }
655    /// The bytes `len` bytes sent as the session's next part cover, if they
656    /// fit it: the first part still to send.
657    pub(crate) fn next_part(&self, len: usize) -> Result<Range<u64>, StorageError> {
658        let offset = self
659            .parts_to_send()
660            .first()
661            .map(|part| part.start)
662            .ok_or(StorageFailure::InvalidPart)?;
663        Ok(offset..self.part_end(offset, len)?)
664    }
665}
666
667pub(crate) fn s3_part_size(total: u64) -> Result<usize, StorageError> {
668    let parts = S3_MAX_PARTS as u64;
669    if total == 0 || total > parts * S3_REQUEST_LIMIT {
670        return Err(StorageFailure::InvalidPart.into());
671    }
672    usize::try_from(total.div_ceil(parts).div_ceil(S3_PART_UNIT) * S3_PART_UNIT)
673        .map_err(|error| StorageFailure::InvalidPart.with_source(error))
674}
675pub(crate) fn check_cloudkit_id(value: &str) -> Result<(), StorageError> {
676    coven_foundation::id_source::canonical_uuid(value)
677        .map_err(|error| StorageFailure::SessionMismatch.with_source(error))?;
678    Ok(())
679}
680fn nonempty(id: &SecretText) -> Result<(), StorageError> {
681    if id.as_str().is_empty() {
682        return Err(StorageFailure::SessionMismatch.into());
683    }
684    Ok(())
685}
686fn transfer_url(value: &SecretText) -> Result<(), StorageError> {
687    let url = url::Url::parse(value.as_str())
688        .map_err(|error| StorageFailure::SessionMismatch.with_source(error))?;
689    if !crate::web_url::is_web_url(&url) || url.fragment().is_some() {
690        return Err(StorageFailure::SessionMismatch.into());
691    }
692    Ok(())
693}
694
695#[cfg(test)]
696#[path = "session_tests.rs"]
697mod tests;
coven_storage/session.rs

coven_storage/
session.rs
