Skip to main content

typst_pack/opendal/
write.rs

1//! OpenDAL exact-key writing vocabulary and crate-private execution.
2
3use std::{collections::BTreeMap, future::Future};
4
5use futures_util::StreamExt;
6use opendal::ErrorKind;
7
8use super::location::validate_decoded_artifact_key_path;
9use super::{
10    BoxError, Location, LocationError, LocationRoleError, OperatorBinding, OperatorResolver,
11};
12use crate::redacted_error::RedactedError;
13use crate::{
14    CanonicalIdentity, CommitCertainty, CompilationResult, CompilationStatus, PackArchiveBytes,
15};
16pub use crate::{
17    CompilationArtifactWriteEntry, CompilationArtifactWriteProgress,
18    CompilationArtifactWriteReceipt, PackExtractionWriteEntry, PackExtractionWriteProgress,
19    PackExtractionWriteReceipt, WriteKeyOutcome,
20};
21
22/// The exact-key conflict policy for an OpenDAL write operation.
23#[derive(Clone, Copy, Debug, Eq, PartialEq)]
24#[non_exhaustive]
25pub enum WritePolicy {
26    /// Create absent objects and accept existing objects only when their bytes match.
27    CreateOrVerify,
28    /// Write every exact key without inspecting its existing value.
29    OverwriteExactKeys,
30}
31
32/// The OpenDAL adapter phase reached by a write attempt.
33#[derive(Clone, Copy, Debug, Eq, PartialEq)]
34#[non_exhaustive]
35pub enum OpenDalWritePhase {
36    ResultValidation,
37    DestinationValidation,
38    ResolveOperator,
39    CapabilityAppraisal,
40    PreflightRead,
41    ConditionalCreate,
42    RaceVerification,
43    DirectWrite,
44    Complete,
45}
46
47/// A validated request to write one exact Pack Archive object.
48#[derive(Clone, Debug)]
49pub struct PackArchiveWriteRequest {
50    destination: Location,
51    policy: WritePolicy,
52}
53
54impl PackArchiveWriteRequest {
55    /// Validates an exact-object destination and retains the explicit policy.
56    pub fn new(
57        destination: Location,
58        policy: WritePolicy,
59    ) -> Result<Self, PackArchiveWriteRequestError> {
60        destination.require_object().map_err(|source| {
61            PackArchiveWriteRequestError::InvalidDestinationRole {
62                location: destination.clone(),
63                source,
64            }
65        })?;
66
67        Ok(Self {
68            destination,
69            policy,
70        })
71    }
72
73    pub fn destination(&self) -> &Location {
74        &self.destination
75    }
76
77    pub const fn policy(&self) -> WritePolicy {
78        self.policy
79    }
80}
81
82/// A reason a Pack Archive write request cannot be accepted.
83#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
84#[non_exhaustive]
85pub enum PackArchiveWriteRequestError {
86    #[error("Pack Archive destination {location} is not an exact object: {source}")]
87    InvalidDestinationRole {
88        location: Location,
89        #[source]
90        source: LocationRoleError,
91    },
92}
93
94/// Writes exact borrowed Pack Archive bytes to one normalized object.
95///
96/// Dropping the returned future yields no receipt, and already-issued storage
97/// work may have occurred. The caller retains `archive`; full replay with the
98/// same exact bytes is the recovery contract.
99///
100/// ```no_run
101/// use typst_pack::{Pack, PackArchiveBytes};
102/// use typst_pack::opendal::{Location, OperatorBindings};
103/// use typst_pack::opendal::pack_archive::{
104///     PackArchiveReadRequest, read_pack_archive,
105/// };
106/// use typst_pack::opendal::write::{
107///     PackArchiveWriteRequest, WritePolicy, write_pack_archive,
108/// };
109/// use typst_pack::pack_archive::{ReadLimits, DecodeError, DecodeLimits, decode};
110///
111/// enum WriteThenReadOutcome {
112///     Matching {
113///         read: PackArchiveBytes,
114///         decoded: Result<Pack, DecodeError>,
115///     },
116///     DestinationChanged {
117///         read: PackArchiveBytes,
118///     },
119/// }
120///
121/// async fn write_replay_and_read(
122///     bindings: &OperatorBindings,
123///     destination: Location,
124///     archive: &PackArchiveBytes,
125/// ) -> Result<WriteThenReadOutcome, Box<dyn std::error::Error>> {
126///     let overwrite = PackArchiveWriteRequest::new(
127///         destination.clone(),
128///         WritePolicy::OverwriteExactKeys,
129///     )?;
130///     write_pack_archive(bindings, &overwrite, archive).await?;
131///
132///     let replay = PackArchiveWriteRequest::new(
133///         destination.clone(),
134///         WritePolicy::CreateOrVerify,
135///     )?;
136///     write_pack_archive(bindings, &replay, archive).await?;
137///     write_pack_archive(bindings, &replay, archive).await?;
138///
139///     let read = PackArchiveReadRequest::new(
140///         destination,
141///         ReadLimits::reference_v1(),
142///     )?;
143///     let read = read_pack_archive(bindings, &read).await?;
144///
145///     // The caller still owns `archive`; preserve the independently read
146///     // bytes and do not decode when the mutable destination changed.
147///     if archive.as_slice() != read.as_slice() {
148///         return Ok(WriteThenReadOutcome::DestinationChanged { read });
149///     }
150///
151///     let decoded = decode(&read, DecodeLimits::reference_v1());
152///     Ok(WriteThenReadOutcome::Matching { read, decoded })
153/// }
154/// ```
155#[allow(clippy::result_large_err)]
156pub async fn write_pack_archive<R: OperatorResolver + ?Sized>(
157    resolver: &R,
158    request: &PackArchiveWriteRequest,
159    archive: &PackArchiveBytes,
160) -> Result<PackArchiveWriteReceipt, PackArchiveWriteError> {
161    let mut progress = PackArchiveWriteProgress::new();
162    let destination_path = request.destination().operation_path();
163    let keys = [ExactKey::new(destination_path, archive.as_slice())];
164    {
165        let mut operation = PackArchiveWriteOperation {
166            request,
167            progress: &mut progress,
168        };
169        write_exact_keys(
170            resolver,
171            request.destination().binding(),
172            request.policy(),
173            &keys,
174            &mut operation,
175        )
176        .await?;
177    }
178
179    Ok(PackArchiveWriteReceipt {
180        destination: request.destination().clone(),
181        policy: request.policy(),
182        progress,
183    })
184}
185
186/// A failure while writing exact Pack Archive bytes through OpenDAL.
187///
188/// This error's own `Display` and `Debug` output omit native resolver and
189/// OpenDAL messages. Rendering its source chain may disclose backend context.
190#[derive(Debug, thiserror::Error)]
191#[error(
192    "Pack Archive write failed for binding {} at exact-object operation path {:?} during {phase:?}: {cause}",
193    .destination.binding(),
194    .destination.operation_path(),
195)]
196pub struct PackArchiveWriteError {
197    destination: Location,
198    policy: WritePolicy,
199    failed_path: Option<String>,
200    phase: OpenDalWritePhase,
201    progress: PackArchiveWriteProgress,
202    commit_certainty: CommitCertainty,
203    #[source]
204    cause: RedactedError<PackArchiveWriteErrorCause>,
205}
206
207impl PackArchiveWriteError {
208    pub fn destination(&self) -> &Location {
209        &self.destination
210    }
211
212    pub const fn policy(&self) -> WritePolicy {
213        self.policy
214    }
215
216    pub fn failed_path(&self) -> Option<&str> {
217        self.failed_path.as_deref()
218    }
219
220    pub const fn phase(&self) -> OpenDalWritePhase {
221        self.phase
222    }
223
224    pub fn progress(&self) -> &PackArchiveWriteProgress {
225        &self.progress
226    }
227
228    pub const fn commit_certainty(&self) -> CommitCertainty {
229        self.commit_certainty
230    }
231
232    pub fn cause(&self) -> &PackArchiveWriteErrorCause {
233        self.cause.inner()
234    }
235}
236
237/// The typed cause of an OpenDAL Pack Archive write failure.
238#[derive(Debug, thiserror::Error)]
239#[non_exhaustive]
240pub enum PackArchiveWriteErrorCause {
241    #[error("operator resolution failed")]
242    ResolveOperator(#[source] BoxError),
243    #[error("the write policy is unsupported")]
244    UnsupportedPolicy { policy: WritePolicy },
245    #[error("the archive exceeds the advertised object size")]
246    UnsupportedObjectSize { byte_length: u64 },
247    #[error("a preflight read failed")]
248    PreflightRead(#[source] ::opendal::Error),
249    #[error("destination bytes conflict")]
250    ByteConflict {
251        expected_byte_length: u64,
252        observed_byte_length_at_least: u64,
253    },
254    #[error("a conditional create failed")]
255    ConditionalCreate(#[source] ::opendal::Error),
256    #[error("race verification failed")]
257    RaceVerification(#[source] ::opendal::Error),
258    #[error("a direct write failed")]
259    DirectWrite(#[source] ::opendal::Error),
260}
261
262/// A validated request to write caller-supplied bytes to one package-cache object.
263///
264/// This request fixes [`WritePolicy::CreateOrVerify`]. It does not offer a
265/// replacement mode and does not represent Package Archive Expansion or Package
266/// Catalog insertion.
267#[derive(Clone, Debug)]
268pub struct PackageCacheArchiveWriteRequest {
269    destination: Location,
270}
271
272impl PackageCacheArchiveWriteRequest {
273    /// Validates and retains a normalized exact-object cache destination.
274    pub fn new(destination: Location) -> Result<Self, PackageCacheArchiveWriteRequestError> {
275        destination.require_object().map_err(|source| {
276            PackageCacheArchiveWriteRequestError::InvalidDestinationRole {
277                location: destination.clone(),
278                source,
279            }
280        })?;
281
282        Ok(Self { destination })
283    }
284
285    pub fn destination(&self) -> &Location {
286        &self.destination
287    }
288
289    pub const fn policy(&self) -> WritePolicy {
290        WritePolicy::CreateOrVerify
291    }
292}
293
294/// A reason a package-cache archive write request cannot be accepted.
295#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
296#[non_exhaustive]
297pub enum PackageCacheArchiveWriteRequestError {
298    #[error("package-cache archive destination {location} is not an exact object: {source}")]
299    InvalidDestinationRole {
300        location: Location,
301        #[source]
302        source: LocationRoleError,
303    },
304}
305
306/// Writes caller-supplied exact archive bytes to one package-cache object.
307///
308/// This low-level operation does not expand the archive, validate a Package
309/// Tree, or insert it into a Package Catalog. Direct use with unvalidated bytes
310/// can poison a cache because a present malformed cache candidate is terminal.
311/// Callers should write registry bytes only after successful expansion,
312/// validation, and insertion.
313///
314/// Dropping the returned future yields no receipt, and already-issued storage
315/// work may have occurred. The caller retains `archive`; full replay with the
316/// same exact bytes is the recovery contract.
317///
318/// ```no_run
319/// # #[cfg(feature = "package-reading")]
320/// # mod example {
321/// use std::error::Error;
322/// use typst_pack::{
323///     PackageReadFailures, PackageCatalog, PackageDisposition,
324///     PackageExpansionLimits,
325/// };
326/// use typst_pack::opendal::OperatorBindings;
327/// use typst_pack::opendal::pack_assembly::{
328///     PackageRead, RegistryArchiveResidue, insert_read_package,
329/// };
330/// use typst_pack::opendal::write::{
331///     PackageCacheArchiveWriteRequest, write_package_cache_archive,
332/// };
333///
334/// async fn insert_then_write_registry_archive(
335///     bindings: &OperatorBindings,
336///     catalog: &mut PackageCatalog,
337///     failures: &mut PackageReadFailures,
338///     read: PackageRead,
339/// ) -> Result<Option<RegistryArchiveResidue>, Box<dyn Error>> {
340///     let Some(residue) = insert_read_package(
341///         catalog,
342///         failures,
343///         read,
344///         PackageDisposition::Embedded,
345///         PackageExpansionLimits::reference_v1(),
346///     )? else {
347///         return Ok(None);
348///     };
349///
350///     let request = PackageCacheArchiveWriteRequest::new(
351///         residue.destination().clone(),
352///     )?;
353///     if let Err(cache_failure) =
354///         write_package_cache_archive(bindings, &request, residue.bytes()).await
355///     {
356///         // Insertion remains successful. The residue retains the exact bytes
357///         // and destination so the caller can report and replay independently.
358///         drop(cache_failure);
359///     }
360///
361///     Ok(Some(residue))
362/// }
363/// # }
364/// ```
365#[allow(clippy::result_large_err)]
366pub async fn write_package_cache_archive<R: OperatorResolver + ?Sized>(
367    resolver: &R,
368    request: &PackageCacheArchiveWriteRequest,
369    archive: &[u8],
370) -> Result<PackageCacheArchiveWriteReceipt, PackageCacheArchiveWriteError> {
371    let mut progress = PackageCacheArchiveWriteProgress::new();
372    let destination_path = request.destination().operation_path();
373    let keys = [ExactKey::new(destination_path, archive)];
374    {
375        let mut operation = PackageCacheArchiveWriteOperation {
376            request,
377            progress: &mut progress,
378        };
379        write_create_or_verify_exact_keys(
380            resolver,
381            request.destination().binding(),
382            &keys,
383            &mut operation,
384        )
385        .await?;
386    }
387
388    Ok(PackageCacheArchiveWriteReceipt {
389        destination: request.destination().clone(),
390        policy: request.policy(),
391        progress,
392    })
393}
394
395/// A failure while writing caller-supplied package-cache archive bytes.
396///
397/// This error's own `Display` and `Debug` output omit native resolver and
398/// OpenDAL messages. Rendering its source chain may disclose backend context.
399#[derive(Debug, thiserror::Error)]
400#[error(
401    "package-cache archive write failed for binding {} at exact-object operation path {:?} during {phase:?}: {cause}",
402    .destination.binding(),
403    .destination.operation_path(),
404)]
405pub struct PackageCacheArchiveWriteError {
406    destination: Location,
407    policy: WritePolicy,
408    failed_path: Option<String>,
409    phase: OpenDalWritePhase,
410    progress: PackageCacheArchiveWriteProgress,
411    commit_certainty: CommitCertainty,
412    #[source]
413    cause: RedactedError<PackageCacheArchiveWriteErrorCause>,
414}
415
416impl PackageCacheArchiveWriteError {
417    pub fn destination(&self) -> &Location {
418        &self.destination
419    }
420
421    pub const fn policy(&self) -> WritePolicy {
422        self.policy
423    }
424
425    pub fn failed_path(&self) -> Option<&str> {
426        self.failed_path.as_deref()
427    }
428
429    pub const fn phase(&self) -> OpenDalWritePhase {
430        self.phase
431    }
432
433    pub fn progress(&self) -> &PackageCacheArchiveWriteProgress {
434        &self.progress
435    }
436
437    pub const fn commit_certainty(&self) -> CommitCertainty {
438        self.commit_certainty
439    }
440
441    pub fn cause(&self) -> &PackageCacheArchiveWriteErrorCause {
442        self.cause.inner()
443    }
444}
445
446/// The typed cause of an OpenDAL package-cache archive write failure.
447#[derive(Debug, thiserror::Error)]
448#[non_exhaustive]
449pub enum PackageCacheArchiveWriteErrorCause {
450    #[error("operator resolution failed")]
451    ResolveOperator(#[source] BoxError),
452    #[error("the write policy is unsupported")]
453    UnsupportedPolicy { policy: WritePolicy },
454    #[error("the archive exceeds the advertised object size")]
455    UnsupportedObjectSize { byte_length: u64 },
456    #[error("a preflight read failed")]
457    PreflightRead(#[source] ::opendal::Error),
458    #[error("destination bytes conflict")]
459    ByteConflict {
460        expected_byte_length: u64,
461        observed_byte_length_at_least: u64,
462    },
463    #[error("a conditional create failed")]
464    ConditionalCreate(#[source] ::opendal::Error),
465    #[error("race verification failed")]
466    RaceVerification(#[source] ::opendal::Error),
467}
468
469/// A validated request to write one Pack Extraction Plan beneath a prefix.
470#[derive(Clone, Debug)]
471pub struct PackExtractionWriteRequest {
472    destination: Location,
473    policy: WritePolicy,
474}
475
476impl PackExtractionWriteRequest {
477    /// Validates a normalized prefix destination and retains the explicit policy.
478    pub fn new(
479        destination: Location,
480        policy: WritePolicy,
481    ) -> Result<Self, PackExtractionWriteRequestError> {
482        destination.require_prefix().map_err(|source| {
483            PackExtractionWriteRequestError::InvalidDestinationRole {
484                location: destination.clone(),
485                source,
486            }
487        })?;
488
489        Ok(Self {
490            destination,
491            policy,
492        })
493    }
494
495    pub fn destination(&self) -> &Location {
496        &self.destination
497    }
498
499    pub const fn policy(&self) -> WritePolicy {
500        self.policy
501    }
502}
503
504/// A reason a Pack Extraction write request cannot be accepted.
505#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
506#[non_exhaustive]
507pub enum PackExtractionWriteRequestError {
508    #[error("Pack Extraction destination {location} is not a prefix: {source}")]
509    InvalidDestinationRole {
510        location: Location,
511        #[source]
512        source: LocationRoleError,
513    },
514}
515
516/// A validated request to write every artifact in one succeeded Compilation Result.
517#[derive(Clone, Debug)]
518pub struct CompilationArtifactWriteRequest {
519    compilation_result_identity: CanonicalIdentity,
520    destination: Location,
521    artifact_keys: Vec<String>,
522    policy: WritePolicy,
523}
524
525impl CompilationArtifactWriteRequest {
526    /// Validates a prefix destination and one decoded relative key per canonical artifact.
527    pub fn new(
528        result: &CompilationResult,
529        destination: Location,
530        artifact_keys: impl IntoIterator<Item = impl Into<String>>,
531        policy: WritePolicy,
532    ) -> Result<Self, CompilationArtifactWriteRequestRejection> {
533        let compilation_result_identity = result.result_identity();
534        let artifact_keys = artifact_keys
535            .into_iter()
536            .map(Into::into)
537            .collect::<Vec<_>>();
538        let mut issues = Vec::new();
539
540        if result.status() != CompilationStatus::Succeeded {
541            issues.push(CompilationArtifactWriteRequestIssue::ResultNotSucceeded);
542        }
543        if let Err(source) = destination.require_prefix() {
544            issues.push(
545                CompilationArtifactWriteRequestIssue::InvalidDestinationRole {
546                    location: destination.clone(),
547                    source,
548                },
549            );
550        }
551        if result.artifacts().len() != artifact_keys.len() {
552            issues.push(
553                CompilationArtifactWriteRequestIssue::ArtifactKeyCountMismatch {
554                    expected: result.artifacts().len(),
555                    actual: artifact_keys.len(),
556                },
557            );
558        }
559        let mut first_indices = BTreeMap::new();
560        for (artifact_index, key) in artifact_keys.iter().enumerate() {
561            if let Err(reason) = validate_artifact_key(key) {
562                issues.push(CompilationArtifactWriteRequestIssue::InvalidArtifactKey {
563                    artifact_index,
564                    key: key.clone(),
565                    reason,
566                });
567            }
568            if let Some(&first_artifact_index) = first_indices.get(key) {
569                issues.push(CompilationArtifactWriteRequestIssue::DuplicateArtifactKey {
570                    key: key.clone(),
571                    first_artifact_index,
572                    duplicate_artifact_index: artifact_index,
573                });
574            } else {
575                first_indices.insert(key.clone(), artifact_index);
576            }
577        }
578
579        if !issues.is_empty() {
580            return Err(CompilationArtifactWriteRequestRejection {
581                compilation_result_identity,
582                destination,
583                issues: issues.into_boxed_slice(),
584            });
585        }
586
587        Ok(Self {
588            compilation_result_identity,
589            destination,
590            artifact_keys,
591            policy,
592        })
593    }
594
595    pub const fn compilation_result_identity(&self) -> CanonicalIdentity {
596        self.compilation_result_identity
597    }
598
599    pub const fn destination(&self) -> &Location {
600        &self.destination
601    }
602
603    pub fn artifact_keys(&self) -> &[String] {
604        &self.artifact_keys
605    }
606
607    pub const fn policy(&self) -> WritePolicy {
608        self.policy
609    }
610}
611
612/// Complete deterministic rejection of a Compilation Output Artifact write request.
613#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
614#[error(
615    "Compilation Output Artifact write request rejected for binding {} beneath prefix operation path {:?} with {} issue(s)",
616    .destination.binding(),
617    .destination.operation_path(),
618    .issues.len(),
619)]
620pub struct CompilationArtifactWriteRequestRejection {
621    compilation_result_identity: CanonicalIdentity,
622    destination: Location,
623    issues: Box<[CompilationArtifactWriteRequestIssue]>,
624}
625
626impl CompilationArtifactWriteRequestRejection {
627    pub const fn compilation_result_identity(&self) -> CanonicalIdentity {
628        self.compilation_result_identity
629    }
630
631    pub fn issues(&self) -> &[CompilationArtifactWriteRequestIssue] {
632        &self.issues
633    }
634}
635
636/// One independently detectable issue in a Compilation Output Artifact write request.
637#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
638#[non_exhaustive]
639pub enum CompilationArtifactWriteRequestIssue {
640    #[error("a rejected Compilation Result cannot be written")]
641    ResultNotSucceeded,
642    #[error("Compilation Output Artifact destination {location} is not a prefix: {source}")]
643    InvalidDestinationRole {
644        location: Location,
645        #[source]
646        source: LocationRoleError,
647    },
648    #[error("expected {expected} artifact key(s), but received {actual}")]
649    ArtifactKeyCountMismatch { expected: usize, actual: usize },
650    #[error("artifact key {key:?} at index {artifact_index} is invalid: {reason}")]
651    InvalidArtifactKey {
652        artifact_index: usize,
653        key: String,
654        reason: CompilationArtifactKeyIssue,
655    },
656    #[error(
657        "artifact key {key:?} at index {duplicate_artifact_index} duplicates index {first_artifact_index}"
658    )]
659    DuplicateArtifactKey {
660        key: String,
661        first_artifact_index: usize,
662        duplicate_artifact_index: usize,
663    },
664}
665
666/// A reason a decoded relative Compilation Output Artifact key is unsafe or ambiguous.
667#[derive(Clone, Debug, Eq, PartialEq, thiserror::Error)]
668#[non_exhaustive]
669pub enum CompilationArtifactKeyIssue {
670    #[error("an artifact key cannot be empty")]
671    Empty,
672    #[error("an artifact key cannot start with a slash")]
673    LeadingSlash,
674    #[error("an artifact key cannot end with a slash")]
675    TrailingSlash,
676    #[error("an artifact key cannot contain a repeated separator")]
677    RepeatedSeparator,
678    #[error("an artifact key cannot contain a dot segment")]
679    DotSegment,
680    #[error("an artifact key cannot contain a backslash")]
681    Backslash,
682    #[error("an artifact key cannot contain a control character")]
683    ControlCharacter,
684    #[error("an artifact key aliases another operation path at byte {index}")]
685    NormalizationAlias { index: usize },
686}
687
688fn validate_artifact_key(key: &str) -> Result<(), CompilationArtifactKeyIssue> {
689    if key.is_empty() {
690        return Err(CompilationArtifactKeyIssue::Empty);
691    }
692    if key.starts_with('/') {
693        return Err(CompilationArtifactKeyIssue::LeadingSlash);
694    }
695    if key.ends_with('/') {
696        return Err(CompilationArtifactKeyIssue::TrailingSlash);
697    }
698    validate_decoded_artifact_key_path(key).map_err(|error| match error {
699        LocationError::RepeatedSeparator { .. } => CompilationArtifactKeyIssue::RepeatedSeparator,
700        LocationError::DotSegment { .. } => CompilationArtifactKeyIssue::DotSegment,
701        LocationError::Backslash { .. } => CompilationArtifactKeyIssue::Backslash,
702        LocationError::ControlCharacter { .. } => CompilationArtifactKeyIssue::ControlCharacter,
703        LocationError::NormalizationAlias { index } => {
704            CompilationArtifactKeyIssue::NormalizationAlias { index }
705        }
706        _ => unreachable!("decoded operation-path validation returned an unrelated error"),
707    })
708}
709
710/// Writes every entry in one Pack Extraction Plan beneath the request's prefix.
711///
712/// The caller-owned progress is cleared synchronously before the returned future
713/// can be polled or dropped. Replaying the same plan with `CreateOrVerify`
714/// accepts objects whose bytes already match.
715///
716/// ```no_run
717/// use typst_pack::PackExtractionPlan;
718/// use typst_pack::opendal::OperatorBindings;
719/// use typst_pack::opendal::write::{
720///     PackExtractionWriteProgress, PackExtractionWriteRequest,
721///     WritePolicy, write_pack_extraction_plan,
722/// };
723///
724/// async fn write_and_replay_partial_attempt(
725///     bindings: &OperatorBindings,
726///     plan: &PackExtractionPlan,
727/// ) -> Result<(), Box<dyn std::error::Error>> {
728///     let request = PackExtractionWriteRequest::new(
729///         "project:/extracted/".parse()?,
730///         WritePolicy::CreateOrVerify,
731///     )?;
732///     let mut progress = PackExtractionWriteProgress::new();
733///
734///     if let Err(error) =
735///         write_pack_extraction_plan(bindings, &request, plan, &mut progress).await
736///     {
737///         // The caller retains the exact completed prefix after a partial attempt.
738///         assert_eq!(error.progress(), &progress);
739///         write_pack_extraction_plan(bindings, &request, plan, &mut progress).await?;
740///     }
741///
742///     Ok(())
743/// }
744/// ```
745#[allow(clippy::result_large_err)]
746pub fn write_pack_extraction_plan<'a, R: OperatorResolver + ?Sized>(
747    resolver: &'a R,
748    request: &'a PackExtractionWriteRequest,
749    plan: &'a crate::PackExtractionPlan,
750    progress: &'a mut PackExtractionWriteProgress,
751) -> impl Future<Output = Result<PackExtractionWriteReceipt, PackExtractionWriteError>> + 'a {
752    progress.clear();
753    async move {
754        let mut destinations = Vec::with_capacity(plan.entries().len());
755        for entry in plan.entries() {
756            let destination = request
757                .destination()
758                .compose(entry.relative_path())
759                .map_err(|_| {
760                    pack_extraction_write_error(
761                        request,
762                        Some(entry.relative_path().to_owned()),
763                        None,
764                        OpenDalWritePhase::DestinationValidation,
765                        progress,
766                        CommitCertainty::NotCommitted,
767                        PackExtractionWriteErrorCause::InvalidDestinationPath {
768                            relative_path: entry.relative_path().to_owned(),
769                        },
770                    )
771                })?;
772            destinations.push(destination);
773        }
774
775        let keys = destinations
776            .iter()
777            .zip(plan.entries())
778            .map(|(destination, entry)| ExactKey::new(destination.operation_path(), entry.bytes()))
779            .collect::<Vec<_>>();
780        {
781            let mut operation = PackExtractionWriteOperation {
782                request,
783                plan,
784                progress,
785            };
786            write_exact_keys(
787                resolver,
788                request.destination().binding(),
789                request.policy(),
790                &keys,
791                &mut operation,
792            )
793            .await?;
794        }
795
796        Ok(PackExtractionWriteReceipt::new(
797            *plan.pack_identity(),
798            progress.clone(),
799        ))
800    }
801}
802
803/// A failure while writing a Pack Extraction Plan through OpenDAL.
804///
805/// This error's own `Display` and `Debug` output omit native resolver and
806/// OpenDAL messages. Rendering its source chain may disclose backend context.
807#[derive(Debug, thiserror::Error)]
808#[error(
809    "Pack Extraction write failed for binding {} beneath prefix operation path {:?} during {phase:?}: {cause}",
810    .destination.binding(),
811    .destination.operation_path(),
812)]
813pub struct PackExtractionWriteError {
814    destination: Location,
815    policy: WritePolicy,
816    failed_relative_path: Option<String>,
817    failed_destination_path: Option<String>,
818    phase: OpenDalWritePhase,
819    progress: PackExtractionWriteProgress,
820    commit_certainty: CommitCertainty,
821    #[source]
822    cause: RedactedError<PackExtractionWriteErrorCause>,
823}
824
825impl PackExtractionWriteError {
826    pub fn destination(&self) -> &Location {
827        &self.destination
828    }
829
830    pub const fn policy(&self) -> WritePolicy {
831        self.policy
832    }
833
834    pub fn failed_relative_path(&self) -> Option<&str> {
835        self.failed_relative_path.as_deref()
836    }
837
838    pub fn failed_destination_path(&self) -> Option<&str> {
839        self.failed_destination_path.as_deref()
840    }
841
842    pub const fn phase(&self) -> OpenDalWritePhase {
843        self.phase
844    }
845
846    pub fn progress(&self) -> &PackExtractionWriteProgress {
847        &self.progress
848    }
849
850    pub const fn commit_certainty(&self) -> CommitCertainty {
851        self.commit_certainty
852    }
853
854    pub fn cause(&self) -> &PackExtractionWriteErrorCause {
855        self.cause.inner()
856    }
857}
858
859/// The typed cause of an OpenDAL Pack Extraction write failure.
860#[derive(Debug, thiserror::Error)]
861#[non_exhaustive]
862pub enum PackExtractionWriteErrorCause {
863    #[error("a composed destination path was invalid")]
864    InvalidDestinationPath { relative_path: String },
865    #[error("operator resolution failed")]
866    ResolveOperator(#[source] BoxError),
867    #[error("the write policy is unsupported")]
868    UnsupportedPolicy { policy: WritePolicy },
869    #[error("an entry exceeds the advertised object size")]
870    UnsupportedObjectSize { byte_length: u64 },
871    #[error("a preflight read failed")]
872    PreflightRead(#[source] ::opendal::Error),
873    #[error("destination bytes conflict")]
874    ByteConflict {
875        expected_byte_length: u64,
876        observed_byte_length_at_least: u64,
877    },
878    #[error("a conditional create failed")]
879    ConditionalCreate(#[source] ::opendal::Error),
880    #[error("race verification failed")]
881    RaceVerification(#[source] ::opendal::Error),
882    #[error("a direct write failed")]
883    DirectWrite(#[source] ::opendal::Error),
884}
885
886fn pack_extraction_write_error(
887    request: &PackExtractionWriteRequest,
888    failed_relative_path: Option<String>,
889    failed_destination_path: Option<String>,
890    phase: OpenDalWritePhase,
891    progress: &PackExtractionWriteProgress,
892    commit_certainty: CommitCertainty,
893    cause: PackExtractionWriteErrorCause,
894) -> PackExtractionWriteError {
895    PackExtractionWriteError {
896        destination: request.destination().clone(),
897        policy: request.policy(),
898        failed_relative_path,
899        failed_destination_path,
900        phase,
901        progress: progress.clone(),
902        commit_certainty,
903        cause: RedactedError::new(cause),
904    }
905}
906
907/// Writes every canonical artifact beneath the request's normalized prefix.
908///
909/// The caller-owned progress is cleared synchronously before the returned future
910/// can be polled or dropped. Replaying the same result with `CreateOrVerify`
911/// accepts objects whose bytes already match.
912///
913/// ```no_run
914/// use typst_pack::CompilationResult;
915/// use typst_pack::opendal::OperatorBindings;
916/// use typst_pack::opendal::write::{
917///     CompilationArtifactWriteProgress, CompilationArtifactWriteRequest,
918///     WritePolicy, write_compilation_artifacts,
919/// };
920///
921/// async fn write_and_replay(
922///     bindings: &OperatorBindings,
923///     document_result: &CompilationResult,
924///     page_result: &CompilationResult,
925/// ) -> Result<(), Box<dyn std::error::Error>> {
926///     let document_request = CompilationArtifactWriteRequest::new(
927///         document_result,
928///         "artifacts:/document/".parse()?,
929///         ["document.pdf"],
930///         WritePolicy::CreateOrVerify,
931///     )?;
932///     let page_keys = page_result
933///         .artifacts()
934///         .iter()
935///         .map(|artifact| format!("page-{}.svg", artifact.source_page_number().unwrap()))
936///         .collect::<Vec<_>>();
937///     let page_request = CompilationArtifactWriteRequest::new(
938///         page_result,
939///         "artifacts:/pages/".parse()?,
940///         page_keys,
941///         WritePolicy::CreateOrVerify,
942///     )?;
943///
944///     let mut document_progress = CompilationArtifactWriteProgress::new();
945///     write_compilation_artifacts(
946///         bindings,
947///         &document_request,
948///         document_result,
949///         &mut document_progress,
950///     )
951///     .await?;
952///     write_compilation_artifacts(
953///         bindings,
954///         &document_request,
955///         document_result,
956///         &mut document_progress,
957///     )
958///     .await?;
959///
960///     let mut page_progress = CompilationArtifactWriteProgress::new();
961///     write_compilation_artifacts(bindings, &page_request, page_result, &mut page_progress)
962///         .await?;
963///     write_compilation_artifacts(bindings, &page_request, page_result, &mut page_progress)
964///         .await?;
965///     Ok(())
966/// }
967/// ```
968#[allow(clippy::result_large_err)]
969pub fn write_compilation_artifacts<'a, R: OperatorResolver + ?Sized>(
970    resolver: &'a R,
971    request: &'a CompilationArtifactWriteRequest,
972    result: &'a CompilationResult,
973    progress: &'a mut CompilationArtifactWriteProgress,
974) -> impl Future<Output = Result<CompilationArtifactWriteReceipt, CompilationArtifactWriteError>> + 'a
975{
976    progress.clear();
977    async move {
978        if request.compilation_result_identity() != result.result_identity() {
979            return Err(compilation_artifact_write_error(
980                request,
981                None,
982                None,
983                OpenDalWritePhase::ResultValidation,
984                progress,
985                CommitCertainty::NotCommitted,
986                CompilationArtifactWriteErrorCause::CompilationResultMismatch {
987                    expected: request.compilation_result_identity(),
988                    actual: result.result_identity(),
989                },
990            ));
991        }
992
993        let mut destinations = Vec::with_capacity(request.artifact_keys().len());
994        for (artifact_index, key) in request.artifact_keys().iter().enumerate() {
995            let destination = request.destination().compose(key).map_err(|_| {
996                compilation_artifact_write_error(
997                    request,
998                    Some(artifact_index),
999                    None,
1000                    OpenDalWritePhase::DestinationValidation,
1001                    progress,
1002                    CommitCertainty::NotCommitted,
1003                    CompilationArtifactWriteErrorCause::InvalidDestinationPath {
1004                        artifact_index,
1005                        key: key.clone(),
1006                    },
1007                )
1008            })?;
1009            destinations.push(destination);
1010        }
1011
1012        let keys = destinations
1013            .iter()
1014            .zip(result.artifacts())
1015            .map(|(destination, artifact)| {
1016                ExactKey::new(destination.operation_path(), artifact.bytes())
1017            })
1018            .collect::<Vec<_>>();
1019        {
1020            let mut operation = CompilationArtifactWriteOperation { request, progress };
1021            write_exact_keys(
1022                resolver,
1023                request.destination().binding(),
1024                request.policy(),
1025                &keys,
1026                &mut operation,
1027            )
1028            .await?;
1029        }
1030
1031        Ok(CompilationArtifactWriteReceipt::new(
1032            request.compilation_result_identity(),
1033            progress.clone(),
1034        ))
1035    }
1036}
1037
1038/// A failure while writing a Compilation Result's exact artifacts through OpenDAL.
1039///
1040/// This error's own `Display` and `Debug` output omit native resolver and
1041/// OpenDAL messages. Rendering its source chain may disclose backend context.
1042#[derive(Debug, thiserror::Error)]
1043#[error(
1044    "Compilation Output Artifact write failed for binding {} beneath prefix operation path {:?} during {phase:?}: {cause}",
1045    .destination.binding(),
1046    .destination.operation_path(),
1047)]
1048pub struct CompilationArtifactWriteError {
1049    compilation_result_identity: CanonicalIdentity,
1050    destination: Location,
1051    policy: WritePolicy,
1052    failed_artifact_index: Option<usize>,
1053    failed_key: Option<String>,
1054    failed_destination_path: Option<String>,
1055    phase: OpenDalWritePhase,
1056    progress: CompilationArtifactWriteProgress,
1057    commit_certainty: CommitCertainty,
1058    #[source]
1059    cause: RedactedError<CompilationArtifactWriteErrorCause>,
1060}
1061
1062impl CompilationArtifactWriteError {
1063    pub const fn compilation_result_identity(&self) -> CanonicalIdentity {
1064        self.compilation_result_identity
1065    }
1066
1067    pub const fn destination(&self) -> &Location {
1068        &self.destination
1069    }
1070
1071    pub const fn policy(&self) -> WritePolicy {
1072        self.policy
1073    }
1074
1075    pub const fn failed_artifact_index(&self) -> Option<usize> {
1076        self.failed_artifact_index
1077    }
1078
1079    pub fn failed_key(&self) -> Option<&str> {
1080        self.failed_key.as_deref()
1081    }
1082
1083    pub fn failed_destination_path(&self) -> Option<&str> {
1084        self.failed_destination_path.as_deref()
1085    }
1086
1087    pub const fn phase(&self) -> OpenDalWritePhase {
1088        self.phase
1089    }
1090
1091    pub const fn progress(&self) -> &CompilationArtifactWriteProgress {
1092        &self.progress
1093    }
1094
1095    pub const fn commit_certainty(&self) -> CommitCertainty {
1096        self.commit_certainty
1097    }
1098
1099    pub const fn cause(&self) -> &CompilationArtifactWriteErrorCause {
1100        self.cause.inner()
1101    }
1102}
1103
1104/// The typed cause of an OpenDAL Compilation Output Artifact write failure.
1105#[derive(Debug, thiserror::Error)]
1106#[non_exhaustive]
1107pub enum CompilationArtifactWriteErrorCause {
1108    #[error("the Compilation Result identity mismatched")]
1109    CompilationResultMismatch {
1110        expected: CanonicalIdentity,
1111        actual: CanonicalIdentity,
1112    },
1113    #[error("a composed destination path was invalid")]
1114    InvalidDestinationPath { artifact_index: usize, key: String },
1115    #[error("operator resolution failed")]
1116    ResolveOperator(#[source] BoxError),
1117    #[error("the write policy is unsupported")]
1118    UnsupportedPolicy { policy: WritePolicy },
1119    #[error("an artifact exceeds the advertised object size")]
1120    UnsupportedObjectSize {
1121        artifact_index: usize,
1122        byte_length: u64,
1123    },
1124    #[error("a preflight read failed")]
1125    PreflightRead(#[source] ::opendal::Error),
1126    #[error("destination bytes conflict")]
1127    ByteConflict {
1128        expected_byte_length: u64,
1129        observed_byte_length_at_least: u64,
1130    },
1131    #[error("a conditional create failed")]
1132    ConditionalCreate(#[source] ::opendal::Error),
1133    #[error("race verification failed")]
1134    RaceVerification(#[source] ::opendal::Error),
1135    #[error("a direct write failed")]
1136    DirectWrite(#[source] ::opendal::Error),
1137}
1138
1139fn compilation_artifact_write_error(
1140    request: &CompilationArtifactWriteRequest,
1141    failed_artifact_index: Option<usize>,
1142    failed_destination_path: Option<String>,
1143    phase: OpenDalWritePhase,
1144    progress: &CompilationArtifactWriteProgress,
1145    commit_certainty: CommitCertainty,
1146    cause: CompilationArtifactWriteErrorCause,
1147) -> CompilationArtifactWriteError {
1148    let failed_key = failed_artifact_index.map(|index| request.artifact_keys()[index].clone());
1149    CompilationArtifactWriteError {
1150        compilation_result_identity: request.compilation_result_identity(),
1151        destination: request.destination().clone(),
1152        policy: request.policy(),
1153        failed_artifact_index,
1154        failed_key,
1155        failed_destination_path,
1156        phase,
1157        progress: progress.clone(),
1158        commit_certainty,
1159        cause: RedactedError::new(cause),
1160    }
1161}
1162
1163macro_rules! workflow_evidence {
1164    (
1165        $entry:ident, $progress:ident, $receipt:ident,
1166        entry { $($entry_field:ident: $entry_type:ty),* $(,)? },
1167        entry_accessors { $($entry_accessors:item)* },
1168        progress_accessors { $($progress_accessors:item)* },
1169        receipt { $($receipt_field:ident: $receipt_type:ty),* $(,)? },
1170        receipt_accessors { $($receipt_accessors:item)* }
1171    ) => {
1172        #[derive(Clone, Debug, Eq, PartialEq)]
1173        pub struct $entry {
1174            $($entry_field: $entry_type,)*
1175            outcome: WriteKeyOutcome,
1176        }
1177
1178        impl $entry {
1179            $($entry_accessors)*
1180
1181            pub const fn outcome(&self) -> WriteKeyOutcome {
1182                self.outcome
1183            }
1184
1185        }
1186
1187        #[derive(Clone, Debug, Default, Eq, PartialEq)]
1188        pub struct $progress {
1189            completed: Vec<$entry>,
1190        }
1191
1192        impl $progress {
1193            pub const fn new() -> Self {
1194                Self { completed: Vec::new() }
1195            }
1196
1197            $($progress_accessors)*
1198
1199            pub(crate) fn clear(&mut self) {
1200                self.completed.clear();
1201            }
1202
1203            pub(crate) fn push(&mut self, entry: $entry) {
1204                self.completed.push(entry);
1205            }
1206        }
1207
1208        #[derive(Clone, Debug, Eq, PartialEq)]
1209        pub struct $receipt {
1210            $($receipt_field: $receipt_type,)*
1211            progress: $progress,
1212        }
1213
1214        impl $receipt {
1215            $($receipt_accessors)*
1216
1217            pub const fn progress(&self) -> &$progress {
1218                &self.progress
1219            }
1220        }
1221    };
1222}
1223
1224workflow_evidence!(
1225    PackArchiveWriteEntry,
1226    PackArchiveWriteProgress,
1227    PackArchiveWriteReceipt,
1228    entry { destination_path: String },
1229    entry_accessors {
1230        pub fn destination_path(&self) -> &str { &self.destination_path }
1231    },
1232    progress_accessors {
1233        pub fn completed(&self) -> Option<&PackArchiveWriteEntry> { self.completed.first() }
1234        pub fn outcome(&self) -> Option<WriteKeyOutcome> {
1235            self.completed().map(PackArchiveWriteEntry::outcome)
1236        }
1237    },
1238    receipt { destination: Location, policy: WritePolicy },
1239    receipt_accessors {
1240        pub fn destination(&self) -> &Location { &self.destination }
1241        pub const fn policy(&self) -> WritePolicy { self.policy }
1242        pub fn completed(&self) -> &PackArchiveWriteEntry {
1243            self.progress.completed().expect("a Pack Archive receipt has one completed entry")
1244        }
1245        pub const fn outcome(&self) -> WriteKeyOutcome {
1246            match self.progress.completed.as_slice() {
1247                [entry, ..] => entry.outcome,
1248                [] => panic!("a Pack Archive receipt has one completed entry"),
1249            }
1250        }
1251    }
1252);
1253
1254workflow_evidence!(
1255    PackageCacheArchiveWriteEntry,
1256    PackageCacheArchiveWriteProgress,
1257    PackageCacheArchiveWriteReceipt,
1258    entry { destination_path: String },
1259    entry_accessors {
1260        pub fn destination_path(&self) -> &str { &self.destination_path }
1261    },
1262    progress_accessors {
1263        pub fn completed(&self) -> Option<&PackageCacheArchiveWriteEntry> { self.completed.first() }
1264        pub fn outcome(&self) -> Option<WriteKeyOutcome> {
1265            self.completed().map(PackageCacheArchiveWriteEntry::outcome)
1266        }
1267    },
1268    receipt { destination: Location, policy: WritePolicy },
1269    receipt_accessors {
1270        pub fn destination(&self) -> &Location { &self.destination }
1271        pub const fn policy(&self) -> WritePolicy { self.policy }
1272        pub fn completed(&self) -> &PackageCacheArchiveWriteEntry {
1273            self.progress.completed().expect("a package-cache archive receipt has one completed entry")
1274        }
1275        pub const fn outcome(&self) -> WriteKeyOutcome {
1276            match self.progress.completed.as_slice() {
1277                [entry, ..] => entry.outcome,
1278                [] => panic!("a package-cache archive receipt has one completed entry"),
1279            }
1280        }
1281    }
1282);
1283
1284pub(crate) struct ExactKey<'a> {
1285    path: &'a str,
1286    bytes: &'a [u8],
1287}
1288
1289impl<'a> ExactKey<'a> {
1290    pub(crate) const fn new(path: &'a str, bytes: &'a [u8]) -> Self {
1291        Self { path, bytes }
1292    }
1293}
1294
1295#[derive(Debug)]
1296pub(crate) struct ExactKeyWriteReceipt {
1297    completed: Vec<ExactKeyWriteEntry>,
1298}
1299
1300impl ExactKeyWriteReceipt {
1301    #[cfg(test)]
1302    fn completed(&self) -> &[ExactKeyWriteEntry] {
1303        &self.completed
1304    }
1305}
1306
1307#[derive(Clone, Debug, Eq, PartialEq)]
1308pub(crate) struct ExactKeyWriteEntry {
1309    pub(crate) index: usize,
1310    pub(crate) outcome: WriteKeyOutcome,
1311}
1312
1313struct ExactKeyWriteFailure {
1314    phase: OpenDalWritePhase,
1315    failed_index: Option<usize>,
1316    failed_path: Option<String>,
1317    commit_certainty: CommitCertainty,
1318}
1319
1320impl ExactKeyWriteFailure {
1321    fn operation(phase: OpenDalWritePhase) -> Self {
1322        Self {
1323            phase,
1324            failed_index: None,
1325            failed_path: None,
1326            commit_certainty: CommitCertainty::NotCommitted,
1327        }
1328    }
1329
1330    fn key(
1331        phase: OpenDalWritePhase,
1332        index: usize,
1333        key: &ExactKey<'_>,
1334        commit_certainty: CommitCertainty,
1335    ) -> Self {
1336        Self {
1337            phase,
1338            failed_index: Some(index),
1339            failed_path: Some(key.path.to_owned()),
1340            commit_certainty,
1341        }
1342    }
1343}
1344
1345trait ExactKeyWriteCause: Sized {
1346    fn resolve_operator(source: BoxError) -> Self;
1347    fn unsupported_policy(policy: WritePolicy) -> Self;
1348    fn unsupported_object_size(index: usize, byte_length: u64) -> Self;
1349    fn preflight_read(source: opendal::Error) -> Self;
1350    fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self;
1351    fn conditional_create(source: opendal::Error) -> Self;
1352    fn race_verification(source: opendal::Error) -> Self;
1353}
1354
1355trait ExactKeyOverwriteCause: ExactKeyWriteCause {
1356    fn direct_write(source: opendal::Error) -> Self;
1357}
1358
1359trait ExactKeyWriteOperation {
1360    type Error;
1361    type Cause: ExactKeyWriteCause;
1362
1363    fn completed_entry(&mut self, entry: ExactKeyWriteEntry);
1364    fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error;
1365}
1366
1367struct PackArchiveWriteOperation<'a> {
1368    request: &'a PackArchiveWriteRequest,
1369    progress: &'a mut PackArchiveWriteProgress,
1370}
1371
1372impl ExactKeyWriteOperation for PackArchiveWriteOperation<'_> {
1373    type Error = PackArchiveWriteError;
1374    type Cause = PackArchiveWriteErrorCause;
1375
1376    fn completed_entry(&mut self, entry: ExactKeyWriteEntry) {
1377        self.progress.push(PackArchiveWriteEntry {
1378            destination_path: self.request.destination().operation_path().to_owned(),
1379            outcome: entry.outcome,
1380        });
1381    }
1382
1383    fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error {
1384        PackArchiveWriteError {
1385            destination: self.request.destination().clone(),
1386            policy: self.request.policy(),
1387            failed_path: failure.failed_path,
1388            phase: failure.phase,
1389            progress: self.progress.clone(),
1390            commit_certainty: failure.commit_certainty,
1391            cause: RedactedError::new(cause),
1392        }
1393    }
1394}
1395
1396impl ExactKeyWriteCause for PackArchiveWriteErrorCause {
1397    fn resolve_operator(source: BoxError) -> Self {
1398        Self::ResolveOperator(source)
1399    }
1400
1401    fn unsupported_policy(policy: WritePolicy) -> Self {
1402        Self::UnsupportedPolicy { policy }
1403    }
1404
1405    fn unsupported_object_size(_: usize, byte_length: u64) -> Self {
1406        Self::UnsupportedObjectSize { byte_length }
1407    }
1408
1409    fn preflight_read(source: opendal::Error) -> Self {
1410        Self::PreflightRead(source)
1411    }
1412
1413    fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self {
1414        Self::ByteConflict {
1415            expected_byte_length,
1416            observed_byte_length_at_least,
1417        }
1418    }
1419
1420    fn conditional_create(source: opendal::Error) -> Self {
1421        Self::ConditionalCreate(source)
1422    }
1423
1424    fn race_verification(source: opendal::Error) -> Self {
1425        Self::RaceVerification(source)
1426    }
1427}
1428
1429impl ExactKeyOverwriteCause for PackArchiveWriteErrorCause {
1430    fn direct_write(source: opendal::Error) -> Self {
1431        Self::DirectWrite(source)
1432    }
1433}
1434
1435struct PackageCacheArchiveWriteOperation<'a> {
1436    request: &'a PackageCacheArchiveWriteRequest,
1437    progress: &'a mut PackageCacheArchiveWriteProgress,
1438}
1439
1440impl ExactKeyWriteOperation for PackageCacheArchiveWriteOperation<'_> {
1441    type Error = PackageCacheArchiveWriteError;
1442    type Cause = PackageCacheArchiveWriteErrorCause;
1443
1444    fn completed_entry(&mut self, entry: ExactKeyWriteEntry) {
1445        self.progress.push(PackageCacheArchiveWriteEntry {
1446            destination_path: self.request.destination().operation_path().to_owned(),
1447            outcome: entry.outcome,
1448        });
1449    }
1450
1451    fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error {
1452        PackageCacheArchiveWriteError {
1453            destination: self.request.destination().clone(),
1454            policy: self.request.policy(),
1455            failed_path: failure.failed_path,
1456            phase: failure.phase,
1457            progress: self.progress.clone(),
1458            commit_certainty: failure.commit_certainty,
1459            cause: RedactedError::new(cause),
1460        }
1461    }
1462}
1463
1464impl ExactKeyWriteCause for PackageCacheArchiveWriteErrorCause {
1465    fn resolve_operator(source: BoxError) -> Self {
1466        Self::ResolveOperator(source)
1467    }
1468
1469    fn unsupported_policy(policy: WritePolicy) -> Self {
1470        Self::UnsupportedPolicy { policy }
1471    }
1472
1473    fn unsupported_object_size(_: usize, byte_length: u64) -> Self {
1474        Self::UnsupportedObjectSize { byte_length }
1475    }
1476
1477    fn preflight_read(source: opendal::Error) -> Self {
1478        Self::PreflightRead(source)
1479    }
1480
1481    fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self {
1482        Self::ByteConflict {
1483            expected_byte_length,
1484            observed_byte_length_at_least,
1485        }
1486    }
1487
1488    fn conditional_create(source: opendal::Error) -> Self {
1489        Self::ConditionalCreate(source)
1490    }
1491
1492    fn race_verification(source: opendal::Error) -> Self {
1493        Self::RaceVerification(source)
1494    }
1495}
1496
1497struct PackExtractionWriteOperation<'a> {
1498    request: &'a PackExtractionWriteRequest,
1499    plan: &'a crate::PackExtractionPlan,
1500    progress: &'a mut PackExtractionWriteProgress,
1501}
1502
1503impl ExactKeyWriteOperation for PackExtractionWriteOperation<'_> {
1504    type Error = PackExtractionWriteError;
1505    type Cause = PackExtractionWriteErrorCause;
1506
1507    fn completed_entry(&mut self, entry: ExactKeyWriteEntry) {
1508        let index = entry.index;
1509        self.progress.push(PackExtractionWriteEntry::new(
1510            self.plan.entries()[index].relative_path().to_owned(),
1511            entry.outcome,
1512        ));
1513    }
1514
1515    fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error {
1516        let failed_relative_path = failure
1517            .failed_index
1518            .map(|index| self.plan.entries()[index].relative_path().to_owned());
1519        pack_extraction_write_error(
1520            self.request,
1521            failed_relative_path,
1522            failure.failed_path,
1523            failure.phase,
1524            self.progress,
1525            failure.commit_certainty,
1526            cause,
1527        )
1528    }
1529}
1530
1531impl ExactKeyWriteCause for PackExtractionWriteErrorCause {
1532    fn resolve_operator(source: BoxError) -> Self {
1533        Self::ResolveOperator(source)
1534    }
1535
1536    fn unsupported_policy(policy: WritePolicy) -> Self {
1537        Self::UnsupportedPolicy { policy }
1538    }
1539
1540    fn unsupported_object_size(_: usize, byte_length: u64) -> Self {
1541        Self::UnsupportedObjectSize { byte_length }
1542    }
1543
1544    fn preflight_read(source: opendal::Error) -> Self {
1545        Self::PreflightRead(source)
1546    }
1547
1548    fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self {
1549        Self::ByteConflict {
1550            expected_byte_length,
1551            observed_byte_length_at_least,
1552        }
1553    }
1554
1555    fn conditional_create(source: opendal::Error) -> Self {
1556        Self::ConditionalCreate(source)
1557    }
1558
1559    fn race_verification(source: opendal::Error) -> Self {
1560        Self::RaceVerification(source)
1561    }
1562}
1563
1564impl ExactKeyOverwriteCause for PackExtractionWriteErrorCause {
1565    fn direct_write(source: opendal::Error) -> Self {
1566        Self::DirectWrite(source)
1567    }
1568}
1569
1570struct CompilationArtifactWriteOperation<'a> {
1571    request: &'a CompilationArtifactWriteRequest,
1572    progress: &'a mut CompilationArtifactWriteProgress,
1573}
1574
1575impl ExactKeyWriteOperation for CompilationArtifactWriteOperation<'_> {
1576    type Error = CompilationArtifactWriteError;
1577    type Cause = CompilationArtifactWriteErrorCause;
1578
1579    fn completed_entry(&mut self, entry: ExactKeyWriteEntry) {
1580        let artifact_index = entry.index;
1581        self.progress.push(CompilationArtifactWriteEntry::new(
1582            artifact_index,
1583            entry.outcome,
1584        ));
1585    }
1586
1587    fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error {
1588        compilation_artifact_write_error(
1589            self.request,
1590            failure.failed_index,
1591            failure.failed_path,
1592            failure.phase,
1593            self.progress,
1594            failure.commit_certainty,
1595            cause,
1596        )
1597    }
1598}
1599
1600impl ExactKeyWriteCause for CompilationArtifactWriteErrorCause {
1601    fn resolve_operator(source: BoxError) -> Self {
1602        Self::ResolveOperator(source)
1603    }
1604
1605    fn unsupported_policy(policy: WritePolicy) -> Self {
1606        Self::UnsupportedPolicy { policy }
1607    }
1608
1609    fn unsupported_object_size(artifact_index: usize, byte_length: u64) -> Self {
1610        Self::UnsupportedObjectSize {
1611            artifact_index,
1612            byte_length,
1613        }
1614    }
1615
1616    fn preflight_read(source: opendal::Error) -> Self {
1617        Self::PreflightRead(source)
1618    }
1619
1620    fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self {
1621        Self::ByteConflict {
1622            expected_byte_length,
1623            observed_byte_length_at_least,
1624        }
1625    }
1626
1627    fn conditional_create(source: opendal::Error) -> Self {
1628        Self::ConditionalCreate(source)
1629    }
1630
1631    fn race_verification(source: opendal::Error) -> Self {
1632        Self::RaceVerification(source)
1633    }
1634}
1635
1636impl ExactKeyOverwriteCause for CompilationArtifactWriteErrorCause {
1637    fn direct_write(source: opendal::Error) -> Self {
1638        Self::DirectWrite(source)
1639    }
1640}
1641
1642async fn write_exact_keys<R, O>(
1643    resolver: &R,
1644    binding: &OperatorBinding,
1645    policy: WritePolicy,
1646    keys: &[ExactKey<'_>],
1647    operation: &mut O,
1648) -> Result<ExactKeyWriteReceipt, O::Error>
1649where
1650    R: OperatorResolver + ?Sized,
1651    O: ExactKeyWriteOperation,
1652    O::Cause: ExactKeyOverwriteCause,
1653{
1654    if keys.is_empty() {
1655        return Ok(ExactKeyWriteReceipt {
1656            completed: Vec::new(),
1657        });
1658    }
1659
1660    let operator = resolver.resolve(binding).map_err(|source| {
1661        operation.error(
1662            ExactKeyWriteFailure::operation(OpenDalWritePhase::ResolveOperator),
1663            O::Cause::resolve_operator(Box::new(source)),
1664        )
1665    })?;
1666    appraise_capabilities(&operator, policy, keys, operation)?;
1667
1668    let mut completed = Vec::with_capacity(keys.len());
1669    match policy {
1670        WritePolicy::OverwriteExactKeys => {
1671            for (index, key) in keys.iter().enumerate() {
1672                operator
1673                    .write(key.path, key.bytes.to_vec())
1674                    .await
1675                    .map_err(|source| {
1676                        operation.error(
1677                            ExactKeyWriteFailure::key(
1678                                OpenDalWritePhase::DirectWrite,
1679                                index,
1680                                key,
1681                                CommitCertainty::Indeterminate,
1682                            ),
1683                            O::Cause::direct_write(source),
1684                        )
1685                    })?;
1686                let entry = ExactKeyWriteEntry {
1687                    index,
1688                    outcome: WriteKeyOutcome::Written,
1689                };
1690                operation.completed_entry(entry.clone());
1691                completed.push(entry);
1692            }
1693        }
1694        WritePolicy::CreateOrVerify => {
1695            write_create_or_verify(&operator, keys, &mut completed, operation).await?;
1696        }
1697    }
1698
1699    Ok(ExactKeyWriteReceipt { completed })
1700}
1701
1702async fn write_create_or_verify_exact_keys<R, O>(
1703    resolver: &R,
1704    binding: &OperatorBinding,
1705    keys: &[ExactKey<'_>],
1706    operation: &mut O,
1707) -> Result<ExactKeyWriteReceipt, O::Error>
1708where
1709    R: OperatorResolver + ?Sized,
1710    O: ExactKeyWriteOperation,
1711{
1712    if keys.is_empty() {
1713        return Ok(ExactKeyWriteReceipt {
1714            completed: Vec::new(),
1715        });
1716    }
1717
1718    let operator = resolver.resolve(binding).map_err(|source| {
1719        operation.error(
1720            ExactKeyWriteFailure::operation(OpenDalWritePhase::ResolveOperator),
1721            O::Cause::resolve_operator(Box::new(source)),
1722        )
1723    })?;
1724    appraise_capabilities(&operator, WritePolicy::CreateOrVerify, keys, operation)?;
1725
1726    let mut completed = Vec::with_capacity(keys.len());
1727    write_create_or_verify(&operator, keys, &mut completed, operation).await?;
1728    Ok(ExactKeyWriteReceipt { completed })
1729}
1730
1731fn appraise_capabilities<O: ExactKeyWriteOperation>(
1732    operator: &opendal::Operator,
1733    policy: WritePolicy,
1734    keys: &[ExactKey<'_>],
1735    operation: &O,
1736) -> Result<(), O::Error> {
1737    let capability = operator.info().capability();
1738    let policy_supported = capability.write
1739        && (!keys.iter().any(|key| key.bytes.is_empty()) || capability.write_can_empty)
1740        && (policy != WritePolicy::CreateOrVerify
1741            || (capability.read && capability.write_with_if_not_exists));
1742    if !policy_supported {
1743        return Err(operation.error(
1744            ExactKeyWriteFailure::operation(OpenDalWritePhase::CapabilityAppraisal),
1745            O::Cause::unsupported_policy(policy),
1746        ));
1747    }
1748    if let Some(maximum) = capability.write_total_max_size {
1749        for (index, key) in keys.iter().enumerate() {
1750            if key.bytes.len() > maximum {
1751                return Err(operation.error(
1752                    ExactKeyWriteFailure::key(
1753                        OpenDalWritePhase::CapabilityAppraisal,
1754                        index,
1755                        key,
1756                        CommitCertainty::NotCommitted,
1757                    ),
1758                    O::Cause::unsupported_object_size(index, byte_length(key.bytes)),
1759                ));
1760            }
1761        }
1762    }
1763    Ok(())
1764}
1765
1766async fn write_create_or_verify<O: ExactKeyWriteOperation>(
1767    operator: &opendal::Operator,
1768    keys: &[ExactKey<'_>],
1769    completed: &mut Vec<ExactKeyWriteEntry>,
1770    operation: &mut O,
1771) -> Result<(), O::Error> {
1772    let mut observations = Vec::with_capacity(keys.len());
1773    for (index, key) in keys.iter().enumerate() {
1774        let observation = match compare_object(operator, key.path, key.bytes).await {
1775            Ok(observation) => observation,
1776            Err(CompareError::Read {
1777                source,
1778                observed_byte_length: 0,
1779            }) if source.kind() == ErrorKind::NotFound => ExistingObject::Absent,
1780            Err(CompareError::Read { source, .. }) => {
1781                return Err(operation.error(
1782                    ExactKeyWriteFailure::key(
1783                        OpenDalWritePhase::PreflightRead,
1784                        index,
1785                        key,
1786                        CommitCertainty::NotCommitted,
1787                    ),
1788                    O::Cause::preflight_read(source),
1789                ));
1790            }
1791            Err(CompareError::Conflict {
1792                observed_byte_length_at_least,
1793            }) => {
1794                return Err(byte_conflict_error(
1795                    operation,
1796                    OpenDalWritePhase::PreflightRead,
1797                    index,
1798                    key,
1799                    observed_byte_length_at_least,
1800                ));
1801            }
1802        };
1803        if observation == ExistingObject::Matching && completed.len() == index {
1804            let entry = ExactKeyWriteEntry {
1805                index,
1806                outcome: WriteKeyOutcome::AlreadyMatching,
1807            };
1808            operation.completed_entry(entry.clone());
1809            completed.push(entry);
1810        }
1811        observations.push(observation);
1812    }
1813
1814    for (index, (key, observation)) in keys.iter().zip(observations).enumerate() {
1815        if index < completed.len() {
1816            debug_assert_eq!(observation, ExistingObject::Matching);
1817            continue;
1818        }
1819        let outcome = match observation {
1820            ExistingObject::Matching => WriteKeyOutcome::AlreadyMatching,
1821            ExistingObject::Absent => {
1822                match operator
1823                    .write_with(key.path, key.bytes.to_vec())
1824                    .if_not_exists(true)
1825                    .await
1826                {
1827                    Ok(_) => WriteKeyOutcome::Created,
1828                    Err(source)
1829                        if matches!(
1830                            source.kind(),
1831                            ErrorKind::AlreadyExists | ErrorKind::ConditionNotMatch
1832                        ) =>
1833                    {
1834                        match compare_object(operator, key.path, key.bytes).await {
1835                            Ok(ExistingObject::Matching) => WriteKeyOutcome::AlreadyMatching,
1836                            Ok(ExistingObject::Absent) => {
1837                                unreachable!("a successful comparison never reports absence")
1838                            }
1839                            Err(CompareError::Read { source, .. }) => {
1840                                return Err(operation.error(
1841                                    ExactKeyWriteFailure::key(
1842                                        OpenDalWritePhase::RaceVerification,
1843                                        index,
1844                                        key,
1845                                        CommitCertainty::NotCommitted,
1846                                    ),
1847                                    O::Cause::race_verification(source),
1848                                ));
1849                            }
1850                            Err(CompareError::Conflict {
1851                                observed_byte_length_at_least,
1852                            }) => {
1853                                return Err(byte_conflict_error(
1854                                    operation,
1855                                    OpenDalWritePhase::RaceVerification,
1856                                    index,
1857                                    key,
1858                                    observed_byte_length_at_least,
1859                                ));
1860                            }
1861                        }
1862                    }
1863                    Err(source) => {
1864                        return Err(operation.error(
1865                            ExactKeyWriteFailure::key(
1866                                OpenDalWritePhase::ConditionalCreate,
1867                                index,
1868                                key,
1869                                CommitCertainty::Indeterminate,
1870                            ),
1871                            O::Cause::conditional_create(source),
1872                        ));
1873                    }
1874                }
1875            }
1876        };
1877        let entry = ExactKeyWriteEntry { index, outcome };
1878        operation.completed_entry(entry.clone());
1879        completed.push(entry);
1880    }
1881    Ok(())
1882}
1883
1884fn byte_conflict_error<O: ExactKeyWriteOperation>(
1885    operation: &O,
1886    phase: OpenDalWritePhase,
1887    index: usize,
1888    key: &ExactKey<'_>,
1889    observed_byte_length_at_least: u64,
1890) -> O::Error {
1891    operation.error(
1892        ExactKeyWriteFailure::key(phase, index, key, CommitCertainty::NotCommitted),
1893        O::Cause::byte_conflict(byte_length(key.bytes), observed_byte_length_at_least),
1894    )
1895}
1896
1897#[derive(Clone, Copy, Debug, Eq, PartialEq)]
1898enum ExistingObject {
1899    Absent,
1900    Matching,
1901}
1902
1903enum CompareError {
1904    Read {
1905        source: opendal::Error,
1906        observed_byte_length: u64,
1907    },
1908    Conflict {
1909        observed_byte_length_at_least: u64,
1910    },
1911}
1912
1913async fn compare_object(
1914    operator: &opendal::Operator,
1915    path: &str,
1916    expected: &[u8],
1917) -> Result<ExistingObject, CompareError> {
1918    let expected_byte_length = byte_length(expected);
1919    let reader = operator
1920        .reader(path)
1921        .await
1922        .map_err(|source| CompareError::Read {
1923            source,
1924            observed_byte_length: 0,
1925        })?;
1926    let mut stream = reader
1927        .into_stream(..)
1928        .await
1929        .map_err(|source| CompareError::Read {
1930            source,
1931            observed_byte_length: 0,
1932        })?;
1933    let mut observed = 0u64;
1934
1935    while let Some(buffer) = stream.next().await {
1936        let buffer = buffer.map_err(|source| CompareError::Read {
1937            source,
1938            observed_byte_length: observed,
1939        })?;
1940        for chunk in buffer {
1941            for byte in chunk {
1942                if observed == expected_byte_length {
1943                    return Err(CompareError::Conflict {
1944                        observed_byte_length_at_least: expected_byte_length
1945                            .checked_add(1)
1946                            .expect("an addressable slice is shorter than u64::MAX bytes"),
1947                    });
1948                }
1949                let index = usize::try_from(observed)
1950                    .expect("observed bytes fit usize while comparing an addressable slice");
1951                observed = observed
1952                    .checked_add(1)
1953                    .expect("an addressable slice is shorter than u64::MAX bytes");
1954                if expected[index] != byte {
1955                    return Err(CompareError::Conflict {
1956                        observed_byte_length_at_least: observed,
1957                    });
1958                }
1959            }
1960        }
1961    }
1962
1963    if observed != expected_byte_length {
1964        return Err(CompareError::Conflict {
1965            observed_byte_length_at_least: observed,
1966        });
1967    }
1968    Ok(ExistingObject::Matching)
1969}
1970
1971fn byte_length(bytes: &[u8]) -> u64 {
1972    u64::try_from(bytes.len()).expect("OpenDAL write supports no 128-bit target")
1973}
1974
1975#[cfg(test)]
1976mod tests {
1977    use std::convert::Infallible;
1978    use std::future::Future;
1979    use std::pin::pin;
1980    use std::task::{Context, Poll, Waker};
1981
1982    use opendal::ErrorKind;
1983
1984    use crate::opendal::scripted_service::{
1985        DestinationMutation, PendingPoint, WriteCapabilities, WriteCondition,
1986        WriteDroppedOperation, WriteOperationLogEntry, WriteReadScript, WriteReadStep, WriteScript,
1987        WriteService, WriteStep,
1988    };
1989    use crate::opendal::{OperatorBinding, OperatorResolver};
1990    use crate::pack_archive::CommitCertainty;
1991    use crate::{
1992        CompilationLimits, CompilationOutputSpecification, Pack, PackCompilationRequest,
1993        SvgOutputSpecification, compile_with_limits,
1994    };
1995
1996    use super::{
1997        CompilationArtifactWriteErrorCause, CompilationArtifactWriteProgress,
1998        CompilationArtifactWriteRequest, ExactKey, ExactKeyOverwriteCause, ExactKeyWriteCause,
1999        ExactKeyWriteEntry, ExactKeyWriteFailure, ExactKeyWriteOperation, OpenDalWritePhase,
2000        PackArchiveWriteEntry, PackArchiveWriteProgress, WriteKeyOutcome, WritePolicy,
2001        write_compilation_artifacts, write_exact_keys,
2002    };
2003
2004    #[test]
2005    fn empty_write_succeeds_without_resolving_an_operator() {
2006        let resolver = RejectingResolver;
2007        let binding = binding();
2008        let mut completed = Vec::new();
2009        let receipt = {
2010            let mut operation = TestWriteOperation::new(&mut completed);
2011            let mut write = pin!(write_exact_keys(
2012                &resolver,
2013                &binding,
2014                WritePolicy::OverwriteExactKeys,
2015                &[],
2016                &mut operation,
2017            ));
2018            expect_ready(write.as_mut()).unwrap()
2019        };
2020
2021        assert!(receipt.completed().is_empty());
2022        assert!(completed.is_empty());
2023    }
2024
2025    #[test]
2026    fn invalid_composed_artifact_destination_fails_before_resolution() {
2027        let result = two_artifact_result();
2028        let request = CompilationArtifactWriteRequest {
2029            compilation_result_identity: result.result_identity(),
2030            destination: "destination:/prefix/".parse().unwrap(),
2031            artifact_keys: vec!["valid.svg".to_owned(), "../alias.svg".to_owned()],
2032            policy: WritePolicy::OverwriteExactKeys,
2033        };
2034        let mut progress = CompilationArtifactWriteProgress::new();
2035
2036        let error = expect_ready(pin!(write_compilation_artifacts(
2037            &RejectingResolver,
2038            &request,
2039            &result,
2040            &mut progress,
2041        )))
2042        .unwrap_err();
2043
2044        assert_eq!(error.phase(), OpenDalWritePhase::DestinationValidation);
2045        assert_eq!(error.failed_artifact_index(), Some(1));
2046        assert_eq!(error.failed_key(), Some("../alias.svg"));
2047        assert_eq!(error.failed_destination_path(), None);
2048        assert_eq!(error.commit_certainty(), CommitCertainty::NotCommitted);
2049        assert!(error.progress().completed().is_empty());
2050        assert!(matches!(
2051            error.cause(),
2052            CompilationArtifactWriteErrorCause::InvalidDestinationPath {
2053                artifact_index: 1,
2054                key,
2055            } if key == "../alias.svg"
2056        ));
2057    }
2058
2059    #[test]
2060    fn overwrite_writes_each_key_once_in_order_without_reading() {
2061        let service = WriteService::new(
2062            WriteCapabilities::all(),
2063            [],
2064            [],
2065            [
2066                WriteScript::new("first.bin", WriteCondition::Direct, []),
2067                WriteScript::new("second.bin", WriteCondition::Direct, []),
2068            ],
2069            16,
2070        );
2071        let resolver = ServiceResolver(service.operator());
2072        let keys = [
2073            ExactKey::new("first.bin", b"first"),
2074            ExactKey::new("second.bin", b"second"),
2075        ];
2076        let mut completed = Vec::new();
2077
2078        let receipt = expect_ready(pin!(write_exact_keys(
2079            &resolver,
2080            &binding(),
2081            WritePolicy::OverwriteExactKeys,
2082            &keys,
2083            &mut TestWriteOperation::new(&mut completed),
2084        )))
2085        .unwrap();
2086
2087        assert_eq!(
2088            service.destination().object("first.bin"),
2089            Some(b"first".as_slice())
2090        );
2091        assert_eq!(
2092            service.destination().object("second.bin"),
2093            Some(b"second".as_slice())
2094        );
2095        assert_eq!(completed, receipt.completed());
2096        assert_eq!(
2097            completed
2098                .iter()
2099                .map(|entry| (entry.index, entry.outcome))
2100                .collect::<Vec<_>>(),
2101            [(0, WriteKeyOutcome::Written), (1, WriteKeyOutcome::Written),]
2102        );
2103        assert!(
2104            service
2105                .log()
2106                .entries()
2107                .iter()
2108                .all(|entry| !matches!(entry, WriteOperationLogEntry::ReadInvoked { .. }))
2109        );
2110    }
2111
2112    #[test]
2113    fn create_or_verify_compares_every_key_before_mutation() {
2114        let service = WriteService::new(
2115            WriteCapabilities::all(),
2116            [("conflict.bin".to_owned(), b"wrong".to_vec())],
2117            [
2118                WriteReadScript::new(
2119                    "absent.bin",
2120                    0,
2121                    [WriteReadStep::failure(ErrorKind::NotFound)],
2122                )
2123                .unwrap(),
2124                WriteReadScript::new("conflict.bin", 1, [WriteReadStep::chunk(0..5)]).unwrap(),
2125            ],
2126            [WriteScript::new(
2127                "absent.bin",
2128                WriteCondition::IfNotExists,
2129                [],
2130            )],
2131            32,
2132        );
2133        let resolver = ServiceResolver(service.operator());
2134        let keys = [
2135            ExactKey::new("absent.bin", b"new"),
2136            ExactKey::new("conflict.bin", b"right"),
2137        ];
2138        let mut completed = Vec::new();
2139
2140        let error = expect_ready(pin!(write_exact_keys(
2141            &resolver,
2142            &binding(),
2143            WritePolicy::CreateOrVerify,
2144            &keys,
2145            &mut TestWriteOperation::new(&mut completed),
2146        )))
2147        .unwrap_err();
2148
2149        assert_eq!(error.phase, OpenDalWritePhase::PreflightRead);
2150        assert_eq!(error.failed_index, Some(1));
2151        assert_eq!(error.commit_certainty, CommitCertainty::NotCommitted);
2152        assert!(matches!(
2153            error.cause,
2154            TestWriteErrorCause::ByteConflict {
2155                expected_byte_length: 5,
2156                observed_byte_length_at_least: 1,
2157            }
2158        ));
2159        assert!(completed.is_empty());
2160        assert!(service.destination().object("absent.bin").is_none());
2161        assert!(
2162            service
2163                .log()
2164                .entries()
2165                .iter()
2166                .all(|entry| !matches!(entry, WriteOperationLogEntry::WriteInvoked { .. }))
2167        );
2168    }
2169
2170    #[test]
2171    fn later_preflight_conflict_retains_the_leading_matching_prefix() {
2172        let service = WriteService::new(
2173            WriteCapabilities::all(),
2174            [
2175                ("matching.bin".to_owned(), b"matching".to_vec()),
2176                ("conflict.bin".to_owned(), b"wrong".to_vec()),
2177            ],
2178            [
2179                WriteReadScript::new("matching.bin", 1, [WriteReadStep::chunk(0..8)]).unwrap(),
2180                WriteReadScript::new("conflict.bin", 1, [WriteReadStep::chunk(0..5)]).unwrap(),
2181            ],
2182            [],
2183            16,
2184        );
2185        let resolver = ServiceResolver(service.operator());
2186        let keys = [
2187            ExactKey::new("matching.bin", b"matching"),
2188            ExactKey::new("conflict.bin", b"right"),
2189        ];
2190        let mut completed = Vec::new();
2191
2192        let error = expect_ready(pin!(write_exact_keys(
2193            &resolver,
2194            &binding(),
2195            WritePolicy::CreateOrVerify,
2196            &keys,
2197            &mut TestWriteOperation::new(&mut completed),
2198        )))
2199        .unwrap_err();
2200
2201        assert_eq!(error.phase, OpenDalWritePhase::PreflightRead);
2202        assert_eq!(completed.len(), 1);
2203        assert_eq!(completed[0].index, 0);
2204        assert_eq!(completed[0].outcome, WriteKeyOutcome::AlreadyMatching);
2205    }
2206
2207    #[test]
2208    fn mutable_matching_stream_is_read_only_evidence_without_commit_certainty() {
2209        let service = WriteService::new(
2210            WriteCapabilities::all(),
2211            [("mutable.bin".to_owned(), b"abcdef".to_vec())],
2212            [WriteReadScript::new(
2213                "mutable.bin",
2214                2,
2215                [
2216                    WriteReadStep::chunk(0..3),
2217                    WriteReadStep::mutate(DestinationMutation::set("mutable.bin", b"abcXYZ")),
2218                    WriteReadStep::chunk(3..6),
2219                ],
2220            )
2221            .unwrap()],
2222            [],
2223            16,
2224        );
2225        let resolver = ServiceResolver(service.operator());
2226        let keys = [ExactKey::new("mutable.bin", b"abcXYZ")];
2227        let mut completed = Vec::new();
2228
2229        let receipt = expect_ready(pin!(write_exact_keys(
2230            &resolver,
2231            &binding(),
2232            WritePolicy::CreateOrVerify,
2233            &keys,
2234            &mut TestWriteOperation::new(&mut completed),
2235        )))
2236        .unwrap();
2237
2238        assert_eq!(
2239            receipt.completed()[0].outcome,
2240            WriteKeyOutcome::AlreadyMatching
2241        );
2242        assert!(
2243            service
2244                .log()
2245                .entries()
2246                .iter()
2247                .all(|entry| !matches!(entry, WriteOperationLogEntry::WriteInvoked { .. }))
2248        );
2249    }
2250
2251    #[test]
2252    fn disappearance_after_a_partial_stream_is_not_treated_as_absence() {
2253        let service = WriteService::new(
2254            WriteCapabilities::all(),
2255            [("unstable.bin".to_owned(), b"planned".to_vec())],
2256            [WriteReadScript::new(
2257                "unstable.bin",
2258                1,
2259                [
2260                    WriteReadStep::chunk(0..3),
2261                    WriteReadStep::failure(ErrorKind::NotFound),
2262                ],
2263            )
2264            .unwrap()],
2265            [WriteScript::new(
2266                "unstable.bin",
2267                WriteCondition::IfNotExists,
2268                [],
2269            )],
2270            16,
2271        );
2272        let resolver = ServiceResolver(service.operator());
2273        let keys = [ExactKey::new("unstable.bin", b"planned")];
2274        let mut completed = Vec::new();
2275
2276        let error = expect_ready(pin!(write_exact_keys(
2277            &resolver,
2278            &binding(),
2279            WritePolicy::CreateOrVerify,
2280            &keys,
2281            &mut TestWriteOperation::new(&mut completed),
2282        )))
2283        .unwrap_err();
2284
2285        assert_eq!(error.phase, OpenDalWritePhase::PreflightRead);
2286        assert!(matches!(
2287            error.cause,
2288            TestWriteErrorCause::PreflightRead(ref source)
2289                if source.kind() == ErrorKind::NotFound
2290        ));
2291        assert!(completed.is_empty());
2292        assert!(
2293            service
2294                .log()
2295                .entries()
2296                .iter()
2297                .all(|entry| !matches!(entry, WriteOperationLogEntry::WriteInvoked { .. }))
2298        );
2299    }
2300
2301    #[test]
2302    fn conditional_conflict_performs_one_bounded_verification() {
2303        let pending = PendingPoint::new();
2304        let service = WriteService::new(
2305            WriteCapabilities::all(),
2306            [],
2307            [
2308                WriteReadScript::new("race.bin", 0, [WriteReadStep::failure(ErrorKind::NotFound)])
2309                    .unwrap(),
2310                WriteReadScript::new("race.bin", 1, [WriteReadStep::chunk(0..7)]).unwrap(),
2311            ],
2312            [WriteScript::new(
2313                "race.bin",
2314                WriteCondition::IfNotExists,
2315                [WriteStep::pending(pending.clone()), WriteStep::commit()],
2316            )],
2317            32,
2318        );
2319        let resolver = ServiceResolver(service.operator());
2320        let keys = [ExactKey::new("race.bin", b"planned")];
2321        let mut completed = Vec::new();
2322        let binding = binding();
2323        let receipt = {
2324            let mut operation = TestWriteOperation::new(&mut completed);
2325            let mut write = pin!(write_exact_keys(
2326                &resolver,
2327                &binding,
2328                WritePolicy::CreateOrVerify,
2329                &keys,
2330                &mut operation,
2331            ));
2332
2333            assert!(matches!(poll_once(write.as_mut()), Poll::Pending));
2334            service.mutate(DestinationMutation::set("race.bin", b"planned"));
2335            pending.release();
2336            expect_ready(write.as_mut()).unwrap()
2337        };
2338
2339        assert_eq!(
2340            receipt.completed()[0].outcome,
2341            WriteKeyOutcome::AlreadyMatching
2342        );
2343        assert_eq!(
2344            service
2345                .log()
2346                .entries()
2347                .iter()
2348                .filter(|entry| matches!(entry, WriteOperationLogEntry::ReadInvoked { .. }))
2349                .count(),
2350            2
2351        );
2352        assert_eq!(completed.len(), 1);
2353    }
2354
2355    #[test]
2356    fn appraisal_rejects_capabilities_and_sizes_before_effects() {
2357        let cases = [
2358            WriteCapabilities {
2359                write: false,
2360                write_can_empty: true,
2361                write_with_if_not_exists: true,
2362                read: true,
2363                write_total_max_size: None,
2364            },
2365            WriteCapabilities {
2366                write: true,
2367                write_can_empty: false,
2368                write_with_if_not_exists: true,
2369                read: true,
2370                write_total_max_size: None,
2371            },
2372            WriteCapabilities {
2373                write: true,
2374                write_can_empty: true,
2375                write_with_if_not_exists: false,
2376                read: true,
2377                write_total_max_size: None,
2378            },
2379            WriteCapabilities {
2380                write: true,
2381                write_can_empty: true,
2382                write_with_if_not_exists: true,
2383                read: false,
2384                write_total_max_size: None,
2385            },
2386        ];
2387        for capabilities in cases {
2388            let service = WriteService::new(capabilities, [], [], [], 4);
2389            let resolver = ServiceResolver(service.operator());
2390            let keys = [ExactKey::new("empty.bin", b"")];
2391            let mut completed = Vec::new();
2392
2393            let error = expect_ready(pin!(write_exact_keys(
2394                &resolver,
2395                &binding(),
2396                WritePolicy::CreateOrVerify,
2397                &keys,
2398                &mut TestWriteOperation::new(&mut completed),
2399            )))
2400            .unwrap_err();
2401
2402            assert_eq!(error.phase, OpenDalWritePhase::CapabilityAppraisal);
2403            assert!(matches!(
2404                error.cause,
2405                TestWriteErrorCause::UnsupportedPolicy { .. }
2406            ));
2407            assert!(service.log().entries().is_empty());
2408        }
2409
2410        let service = WriteService::new(
2411            WriteCapabilities {
2412                write_total_max_size: Some(3),
2413                ..WriteCapabilities::all()
2414            },
2415            [],
2416            [],
2417            [],
2418            4,
2419        );
2420        let resolver = ServiceResolver(service.operator());
2421        let keys = [ExactKey::new("large.bin", b"four")];
2422        let mut completed = Vec::new();
2423        let error = expect_ready(pin!(write_exact_keys(
2424            &resolver,
2425            &binding(),
2426            WritePolicy::OverwriteExactKeys,
2427            &keys,
2428            &mut TestWriteOperation::new(&mut completed),
2429        )))
2430        .unwrap_err();
2431
2432        assert!(matches!(
2433            error.cause,
2434            TestWriteErrorCause::UnsupportedObjectSize { byte_length: 4 }
2435        ));
2436        assert!(service.log().entries().is_empty());
2437    }
2438
2439    #[test]
2440    fn issued_write_failure_is_indeterminate_and_retains_the_completed_prefix() {
2441        let service = WriteService::new(
2442            WriteCapabilities::all(),
2443            [],
2444            [],
2445            [
2446                WriteScript::new("first.bin", WriteCondition::Direct, []),
2447                WriteScript::write_failure(
2448                    "second.bin",
2449                    WriteCondition::Direct,
2450                    ErrorKind::Unexpected,
2451                ),
2452            ],
2453            16,
2454        );
2455        let resolver = ServiceResolver(service.operator());
2456        let keys = [
2457            ExactKey::new("first.bin", b"first"),
2458            ExactKey::new("second.bin", b"second"),
2459        ];
2460        let mut completed = Vec::new();
2461
2462        let error = expect_ready(pin!(write_exact_keys(
2463            &resolver,
2464            &binding(),
2465            WritePolicy::OverwriteExactKeys,
2466            &keys,
2467            &mut TestWriteOperation::new(&mut completed),
2468        )))
2469        .unwrap_err();
2470
2471        assert_eq!(error.phase, OpenDalWritePhase::DirectWrite);
2472        assert_eq!(error.failed_index, Some(1));
2473        assert_eq!(error.failed_path.as_deref(), Some("second.bin"));
2474        assert_eq!(error.commit_certainty, CommitCertainty::Indeterminate);
2475        assert_eq!(completed.len(), 1);
2476        assert_eq!(completed[0].index, 0);
2477    }
2478
2479    #[test]
2480    fn dropping_a_pending_write_leaves_the_completed_prefix_with_the_caller() {
2481        let pending = PendingPoint::new();
2482        let service = WriteService::new(
2483            WriteCapabilities::all(),
2484            [],
2485            [],
2486            [
2487                WriteScript::new("first.bin", WriteCondition::Direct, []),
2488                WriteScript::new(
2489                    "second.bin",
2490                    WriteCondition::Direct,
2491                    [WriteStep::pending(pending.clone())],
2492                ),
2493            ],
2494            16,
2495        );
2496        let resolver = ServiceResolver(service.operator());
2497        let binding = binding();
2498        let keys = [
2499            ExactKey::new("first.bin", b"first"),
2500            ExactKey::new("second.bin", b"second"),
2501        ];
2502        let mut completed = Vec::new();
2503        {
2504            let mut operation = TestWriteOperation::new(&mut completed);
2505            let mut write = pin!(write_exact_keys(
2506                &resolver,
2507                &binding,
2508                WritePolicy::OverwriteExactKeys,
2509                &keys,
2510                &mut operation,
2511            ));
2512            assert!(matches!(poll_once(write.as_mut()), Poll::Pending));
2513            assert!(pending.was_observed());
2514        }
2515
2516        assert_eq!(completed.len(), 1);
2517        assert_eq!(completed[0].index, 0);
2518        assert_eq!(
2519            service.cancellations(),
2520            [WriteDroppedOperation::Write {
2521                id: 1,
2522                path: "second.bin".to_owned(),
2523                length: 6,
2524                condition: WriteCondition::Direct,
2525                issued: true,
2526            }]
2527        );
2528    }
2529
2530    #[test]
2531    fn dropping_a_later_preflight_read_retains_the_leading_matching_prefix() {
2532        let pending = PendingPoint::new();
2533        let service = WriteService::new(
2534            WriteCapabilities::all(),
2535            [("matching.bin".to_owned(), b"matching".to_vec())],
2536            [
2537                WriteReadScript::new("matching.bin", 1, [WriteReadStep::chunk(0..8)]).unwrap(),
2538                WriteReadScript::new("pending.bin", 0, [WriteReadStep::pending(pending.clone())])
2539                    .unwrap(),
2540            ],
2541            [],
2542            16,
2543        );
2544        let resolver = ServiceResolver(service.operator());
2545        let binding = binding();
2546        let keys = [
2547            ExactKey::new("matching.bin", b"matching"),
2548            ExactKey::new("pending.bin", b"pending"),
2549        ];
2550        let mut completed = Vec::new();
2551        {
2552            let mut operation = TestWriteOperation::new(&mut completed);
2553            let mut write = pin!(write_exact_keys(
2554                &resolver,
2555                &binding,
2556                WritePolicy::CreateOrVerify,
2557                &keys,
2558                &mut operation,
2559            ));
2560            assert!(matches!(poll_once(write.as_mut()), Poll::Pending));
2561            assert!(pending.was_observed());
2562        }
2563
2564        assert_eq!(completed.len(), 1);
2565        assert_eq!(completed[0].index, 0);
2566        assert_eq!(completed[0].outcome, WriteKeyOutcome::AlreadyMatching);
2567    }
2568
2569    #[test]
2570    fn workflow_evidence_retains_observed_outcomes() {
2571        let mut progress = PackArchiveWriteProgress::new();
2572        progress.push(PackArchiveWriteEntry {
2573            destination_path: "archive.typk".to_owned(),
2574            outcome: WriteKeyOutcome::AlreadyMatching,
2575        });
2576
2577        assert_eq!(progress.outcome(), Some(WriteKeyOutcome::AlreadyMatching));
2578
2579        progress.clear();
2580        progress.push(PackArchiveWriteEntry {
2581            destination_path: "archive.typk".to_owned(),
2582            outcome: WriteKeyOutcome::Created,
2583        });
2584        assert_eq!(progress.outcome(), Some(WriteKeyOutcome::Created));
2585    }
2586
2587    #[derive(Debug)]
2588    struct TestWriteError {
2589        phase: OpenDalWritePhase,
2590        failed_index: Option<usize>,
2591        failed_path: Option<String>,
2592        commit_certainty: CommitCertainty,
2593        cause: TestWriteErrorCause,
2594    }
2595
2596    #[derive(Debug)]
2597    enum TestWriteErrorCause {
2598        ResolveOperator(crate::opendal::BoxError),
2599        UnsupportedPolicy {
2600            policy: WritePolicy,
2601        },
2602        UnsupportedObjectSize {
2603            byte_length: u64,
2604        },
2605        PreflightRead(opendal::Error),
2606        ByteConflict {
2607            expected_byte_length: u64,
2608            observed_byte_length_at_least: u64,
2609        },
2610        ConditionalCreate(opendal::Error),
2611        RaceVerification(opendal::Error),
2612        DirectWrite(opendal::Error),
2613    }
2614
2615    impl ExactKeyWriteCause for TestWriteErrorCause {
2616        fn resolve_operator(source: crate::opendal::BoxError) -> Self {
2617            Self::ResolveOperator(source)
2618        }
2619
2620        fn unsupported_policy(policy: WritePolicy) -> Self {
2621            Self::UnsupportedPolicy { policy }
2622        }
2623
2624        fn unsupported_object_size(_: usize, byte_length: u64) -> Self {
2625            Self::UnsupportedObjectSize { byte_length }
2626        }
2627
2628        fn preflight_read(source: opendal::Error) -> Self {
2629            Self::PreflightRead(source)
2630        }
2631
2632        fn byte_conflict(expected_byte_length: u64, observed_byte_length_at_least: u64) -> Self {
2633            Self::ByteConflict {
2634                expected_byte_length,
2635                observed_byte_length_at_least,
2636            }
2637        }
2638
2639        fn conditional_create(source: opendal::Error) -> Self {
2640            Self::ConditionalCreate(source)
2641        }
2642
2643        fn race_verification(source: opendal::Error) -> Self {
2644            Self::RaceVerification(source)
2645        }
2646    }
2647
2648    impl ExactKeyOverwriteCause for TestWriteErrorCause {
2649        fn direct_write(source: opendal::Error) -> Self {
2650            Self::DirectWrite(source)
2651        }
2652    }
2653
2654    struct TestWriteOperation<'a> {
2655        completed: &'a mut Vec<ExactKeyWriteEntry>,
2656    }
2657
2658    impl<'a> TestWriteOperation<'a> {
2659        fn new(completed: &'a mut Vec<ExactKeyWriteEntry>) -> Self {
2660            Self { completed }
2661        }
2662    }
2663
2664    impl ExactKeyWriteOperation for TestWriteOperation<'_> {
2665        type Error = TestWriteError;
2666        type Cause = TestWriteErrorCause;
2667
2668        fn completed_entry(&mut self, entry: ExactKeyWriteEntry) {
2669            self.completed.push(entry);
2670        }
2671
2672        fn error(&self, failure: ExactKeyWriteFailure, cause: Self::Cause) -> Self::Error {
2673            TestWriteError {
2674                phase: failure.phase,
2675                failed_index: failure.failed_index,
2676                failed_path: failure.failed_path,
2677                commit_certainty: failure.commit_certainty,
2678                cause,
2679            }
2680        }
2681    }
2682
2683    fn expect_ready<F: Future>(future: std::pin::Pin<&mut F>) -> F::Output {
2684        match poll_once(future) {
2685            Poll::Ready(output) => output,
2686            Poll::Pending => panic!("future unexpectedly pending"),
2687        }
2688    }
2689
2690    fn poll_once<F: Future>(future: std::pin::Pin<&mut F>) -> Poll<F::Output> {
2691        future.poll(&mut Context::from_waker(Waker::noop()))
2692    }
2693
2694    fn binding() -> OperatorBinding {
2695        OperatorBinding::new("destination").unwrap()
2696    }
2697
2698    fn two_artifact_result() -> crate::CompilationResult {
2699        let pack = Pack::builder("main.typ")
2700            .file(
2701                "main.typ",
2702                b"composition validation\n#pagebreak()\nsecond page".to_vec(),
2703            )
2704            .unwrap()
2705            .build()
2706            .unwrap();
2707        compile_with_limits(
2708            PackCompilationRequest::new(
2709                pack,
2710                CompilationOutputSpecification::Svg(SvgOutputSpecification::default()),
2711            ),
2712            CompilationLimits::reference_v1(),
2713        )
2714        .unwrap()
2715        .result()
2716        .unwrap()
2717        .clone()
2718    }
2719
2720    struct ServiceResolver(opendal::Operator);
2721
2722    impl OperatorResolver for ServiceResolver {
2723        type Error = Infallible;
2724
2725        fn resolve(&self, _: &OperatorBinding) -> Result<opendal::Operator, Self::Error> {
2726            Ok(self.0.clone())
2727        }
2728    }
2729
2730    struct RejectingResolver;
2731
2732    impl OperatorResolver for RejectingResolver {
2733        type Error = Infallible;
2734
2735        fn resolve(&self, _: &OperatorBinding) -> Result<opendal::Operator, Self::Error> {
2736            panic!("an empty write must not resolve an operator")
2737        }
2738    }
2739}