typst_pack/pack_archive/
write.rs

1use std::io::{self, Write};
2
3#[cfg(feature = "fs")]
4use std::path::{Path, PathBuf};
5#[cfg(feature = "fs")]
6use std::sync::atomic::{AtomicU64, Ordering};
7
8use super::{EncodeError, EncodeLimits, encode_with_limits};
9use crate::{CommitCertainty, Pack, PackArchiveBytes};
10
11/// The stream write phase reached by an attempt.
12#[derive(Debug, Clone, Copy, Eq, PartialEq)]
13pub enum StreamWritePhase {
14    Write,
15    Flush,
16    Complete,
17}
18
19/// Evidence from successful exact stream write.
20#[derive(Debug, Clone, Copy, Eq, PartialEq)]
21pub struct StreamWriteReceipt {
22    visible_prefix: u64,
23}
24
25impl StreamWriteReceipt {
26    pub const fn visible_prefix(&self) -> u64 {
27        self.visible_prefix
28    }
29}
30
31/// Evidence and the concrete cause from failed exact stream write.
32#[derive(Debug, thiserror::Error)]
33#[error(
34    "Pack Archive stream write failed during {phase:?} after a {visible_prefix}-byte visible prefix: {source}"
35)]
36pub struct StreamWriteError {
37    phase: StreamWritePhase,
38    visible_prefix: u64,
39    commit_certainty: CommitCertainty,
40    #[source]
41    source: io::Error,
42}
43
44impl StreamWriteError {
45    pub const fn phase(&self) -> StreamWritePhase {
46        self.phase
47    }
48
49    pub const fn visible_prefix(&self) -> u64 {
50        self.visible_prefix
51    }
52
53    pub const fn commit_certainty(&self) -> CommitCertainty {
54        self.commit_certainty
55    }
56
57    pub const fn io_error(&self) -> &io::Error {
58        &self.source
59    }
60}
61
62/// Writes exact Pack Archive bytes to a stream and flushes the writer.
63pub fn write(
64    mut writer: impl Write,
65    archive: &PackArchiveBytes,
66) -> Result<StreamWriteReceipt, StreamWriteError> {
67    let mut visible_prefix = 0usize;
68    while visible_prefix < archive.as_slice().len() {
69        match writer.write(&archive.as_slice()[visible_prefix..]) {
70            Ok(0) => {
71                return Err(StreamWriteError {
72                    phase: StreamWritePhase::Write,
73                    visible_prefix: visible_prefix as u64,
74                    commit_certainty: CommitCertainty::NotCommitted,
75                    source: io::Error::new(
76                        io::ErrorKind::WriteZero,
77                        "failed to write the complete Pack Archive",
78                    ),
79                });
80            }
81            Ok(written) => visible_prefix += written,
82            Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
83            Err(source) => {
84                return Err(StreamWriteError {
85                    phase: StreamWritePhase::Write,
86                    visible_prefix: visible_prefix as u64,
87                    commit_certainty: CommitCertainty::NotCommitted,
88                    source,
89                });
90            }
91        }
92    }
93    if let Err(source) = writer.flush() {
94        return Err(StreamWriteError {
95            phase: StreamWritePhase::Flush,
96            visible_prefix: visible_prefix as u64,
97            commit_certainty: CommitCertainty::Indeterminate,
98            source,
99        });
100    }
101    Ok(StreamWriteReceipt {
102        visible_prefix: visible_prefix as u64,
103    })
104}
105
106/// A failure in Pack Archive Encoding followed by stream write.
107#[derive(Debug, thiserror::Error)]
108#[non_exhaustive]
109pub enum WritePackError {
110    #[error(transparent)]
111    Encode(#[from] EncodeError),
112    #[error("encoded Pack Archive could not be written: {source}")]
113    Write {
114        archive: PackArchiveBytes,
115        #[source]
116        source: StreamWriteError,
117    },
118}
119
120/// Encodes and writes one Pack while preserving exact bytes on write failure.
121pub fn write_pack(writer: impl Write, pack: &Pack) -> Result<StreamWriteReceipt, WritePackError> {
122    write_pack_with_limits(writer, pack, EncodeLimits::reference_v1())
123}
124
125/// Encodes under explicit resource ceilings and writes one Pack.
126pub fn write_pack_with_limits(
127    writer: impl Write,
128    pack: &Pack,
129    encode_limits: EncodeLimits,
130) -> Result<StreamWriteReceipt, WritePackError> {
131    let archive = encode_with_limits(pack, encode_limits)?;
132    match write(writer, &archive) {
133        Ok(receipt) => Ok(receipt),
134        Err(source) => Err(WritePackError::Write { archive, source }),
135    }
136}
137
138/// The strict atomic policy requested for filesystem write.
139#[cfg(feature = "fs")]
140#[derive(Debug, Clone, Copy, Eq, PartialEq)]
141pub enum FileWritePolicy {
142    CreateNew,
143    ReplaceExisting,
144}
145
146/// The filesystem write phase reached by an attempt.
147#[cfg(feature = "fs")]
148#[derive(Debug, Clone, Copy, Eq, PartialEq)]
149pub enum FileWritePhase {
150    Policy,
151    StagingCreate,
152    StagingWrite,
153    StagingFlush,
154    Commit,
155    StagingCleanup,
156    Complete,
157}
158
159/// The observed state of same-directory staging when an attempt returns.
160#[cfg(feature = "fs")]
161#[derive(Debug, Clone, Copy, Eq, PartialEq)]
162pub enum StagingResidueStatus {
163    Absent,
164    Present,
165    Indeterminate,
166}
167
168/// The concrete cause of a filesystem write failure.
169#[cfg(feature = "fs")]
170#[derive(Debug, thiserror::Error)]
171#[non_exhaustive]
172pub enum FileWriteErrorCause {
173    #[error("the platform cannot guarantee the requested {0:?} write policy")]
174    UnsupportedPolicy(FileWritePolicy),
175    #[error(transparent)]
176    Io(#[from] io::Error),
177}
178
179/// Evidence from successful atomic filesystem write.
180#[cfg(feature = "fs")]
181#[derive(Debug, Clone, Eq, PartialEq)]
182pub struct FileWriteReceipt {
183    destination: PathBuf,
184    policy: FileWritePolicy,
185    byte_length: u64,
186}
187
188#[cfg(feature = "fs")]
189impl FileWriteReceipt {
190    pub fn destination(&self) -> &Path {
191        &self.destination
192    }
193
194    pub const fn policy(&self) -> FileWritePolicy {
195        self.policy
196    }
197
198    pub const fn byte_length(&self) -> u64 {
199        self.byte_length
200    }
201}
202
203/// Evidence and the concrete cause from failed atomic filesystem write.
204#[cfg(feature = "fs")]
205#[derive(Debug, thiserror::Error)]
206#[error(
207    "Pack Archive write to {destination:?} failed during {phase:?} with {commit_certainty:?} certainty: {source}"
208)]
209pub struct FileWriteError {
210    destination: PathBuf,
211    policy: FileWritePolicy,
212    byte_length: u64,
213    phase: FileWritePhase,
214    staging_residue: Option<PathBuf>,
215    staging_residue_status: StagingResidueStatus,
216    commit_certainty: CommitCertainty,
217    #[source]
218    source: FileWriteErrorCause,
219}
220
221#[cfg(feature = "fs")]
222impl FileWriteError {
223    pub fn destination(&self) -> &Path {
224        &self.destination
225    }
226
227    pub const fn policy(&self) -> FileWritePolicy {
228        self.policy
229    }
230
231    pub const fn byte_length(&self) -> u64 {
232        self.byte_length
233    }
234
235    pub const fn phase(&self) -> FileWritePhase {
236        self.phase
237    }
238
239    pub fn staging_residue(&self) -> Option<&Path> {
240        self.staging_residue.as_deref()
241    }
242
243    pub const fn staging_residue_status(&self) -> StagingResidueStatus {
244        self.staging_residue_status
245    }
246
247    pub const fn commit_certainty(&self) -> CommitCertainty {
248        self.commit_certainty
249    }
250
251    pub const fn cause(&self) -> &FileWriteErrorCause {
252        &self.source
253    }
254}
255
256#[cfg(feature = "fs")]
257fn write_error(
258    destination: &Path,
259    policy: FileWritePolicy,
260    byte_length: u64,
261    phase: FileWritePhase,
262    staging_residue: Option<PathBuf>,
263    commit_certainty: CommitCertainty,
264    source: impl Into<FileWriteErrorCause>,
265) -> FileWriteError {
266    let staging_residue_status = staging_residue
267        .as_deref()
268        .map(observe_staging_residue)
269        .unwrap_or(StagingResidueStatus::Absent);
270    let staging_residue = if staging_residue_status == StagingResidueStatus::Absent {
271        None
272    } else {
273        staging_residue
274    };
275    FileWriteError {
276        destination: destination.to_owned(),
277        policy,
278        byte_length,
279        phase,
280        staging_residue,
281        staging_residue_status,
282        commit_certainty,
283        source: source.into(),
284    }
285}
286
287/// Atomically writes exact Pack Archive bytes from same-directory staging.
288#[cfg(feature = "fs")]
289pub fn write_file(
290    destination: impl AsRef<Path>,
291    archive: &PackArchiveBytes,
292    policy: FileWritePolicy,
293) -> Result<FileWriteReceipt, FileWriteError> {
294    write_file_before_commit(destination.as_ref(), archive, policy, |_| {})
295}
296
297#[cfg(feature = "fs")]
298fn write_file_before_commit(
299    destination: &Path,
300    archive: &PackArchiveBytes,
301    policy: FileWritePolicy,
302    before_commit: impl FnOnce(&Path),
303) -> Result<FileWriteReceipt, FileWriteError> {
304    let byte_length = archive.len();
305    if policy == FileWritePolicy::ReplaceExisting && !replace_existing_supported() {
306        return Err(write_error(
307            destination,
308            policy,
309            byte_length,
310            FileWritePhase::Policy,
311            None,
312            CommitCertainty::NotCommitted,
313            FileWriteErrorCause::UnsupportedPolicy(policy),
314        ));
315    }
316
317    let (staging, mut file) = create_staging(destination).map_err(|source| {
318        write_error(
319            destination,
320            policy,
321            byte_length,
322            FileWritePhase::StagingCreate,
323            None,
324            CommitCertainty::NotCommitted,
325            source,
326        )
327    })?;
328    if let Err((phase, source)) = write_staging(&mut file, archive.as_slice()) {
329        drop(file);
330        return Err(write_error(
331            destination,
332            policy,
333            byte_length,
334            phase,
335            Some(staging),
336            CommitCertainty::NotCommitted,
337            source,
338        ));
339    }
340    drop(file);
341    before_commit(&staging);
342
343    if let Err(source) = commit_staging(&staging, destination, policy) {
344        return Err(write_error(
345            destination,
346            policy,
347            byte_length,
348            FileWritePhase::Commit,
349            Some(staging),
350            CommitCertainty::NotCommitted,
351            source,
352        ));
353    }
354    if let Err(source) = std::fs::remove_file(&staging)
355        && source.kind() != io::ErrorKind::NotFound
356    {
357        return Err(write_error(
358            destination,
359            policy,
360            byte_length,
361            FileWritePhase::StagingCleanup,
362            Some(staging),
363            CommitCertainty::Committed,
364            source,
365        ));
366    }
367
368    Ok(FileWriteReceipt {
369        destination: destination.to_owned(),
370        policy,
371        byte_length,
372    })
373}
374
375#[cfg(feature = "fs")]
376fn create_staging(destination: &Path) -> io::Result<(PathBuf, std::fs::File)> {
377    static NEXT_STAGE: AtomicU64 = AtomicU64::new(0);
378    const ATTEMPTS: usize = 128;
379
380    let parent = destination.parent().unwrap_or_else(|| Path::new("."));
381    let file_name = destination.file_name().ok_or_else(|| {
382        io::Error::new(io::ErrorKind::InvalidInput, "destination has no file name")
383    })?;
384    for _ in 0..ATTEMPTS {
385        let sequence = NEXT_STAGE.fetch_add(1, Ordering::Relaxed);
386        let mut staging_name = file_name.to_os_string();
387        staging_name.push(format!(
388            ".typst-pack-stage-{}-{sequence}",
389            std::process::id()
390        ));
391        let staging = parent.join(staging_name);
392        match std::fs::OpenOptions::new()
393            .write(true)
394            .create_new(true)
395            .open(&staging)
396        {
397            Ok(file) => return Ok((staging, file)),
398            Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {}
399            Err(error) => return Err(error),
400        }
401    }
402    Err(io::Error::new(
403        io::ErrorKind::AlreadyExists,
404        "could not allocate a unique same-directory staging file",
405    ))
406}
407
408#[cfg(feature = "fs")]
409fn observe_staging_residue(staging: &Path) -> StagingResidueStatus {
410    match std::fs::symlink_metadata(staging) {
411        Ok(_) => StagingResidueStatus::Present,
412        Err(error) if error.kind() == io::ErrorKind::NotFound => StagingResidueStatus::Absent,
413        Err(_) => StagingResidueStatus::Indeterminate,
414    }
415}
416
417#[cfg(feature = "fs")]
418fn write_staging(writer: &mut impl Write, bytes: &[u8]) -> Result<(), (FileWritePhase, io::Error)> {
419    let mut written = 0;
420    while written < bytes.len() {
421        match writer.write(&bytes[written..]) {
422            Ok(0) => {
423                return Err((
424                    FileWritePhase::StagingWrite,
425                    io::Error::new(
426                        io::ErrorKind::WriteZero,
427                        "failed to write complete staging file",
428                    ),
429                ));
430            }
431            Ok(count) => written += count,
432            Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
433            Err(error) => return Err((FileWritePhase::StagingWrite, error)),
434        }
435    }
436    writer
437        .flush()
438        .map_err(|error| (FileWritePhase::StagingFlush, error))
439}
440
441#[cfg(feature = "fs")]
442fn commit_staging(staging: &Path, destination: &Path, policy: FileWritePolicy) -> io::Result<()> {
443    match policy {
444        FileWritePolicy::CreateNew => std::fs::hard_link(staging, destination),
445        FileWritePolicy::ReplaceExisting => replace_existing(staging, destination),
446    }
447}
448
449#[cfg(feature = "fs")]
450const fn replace_existing_supported() -> bool {
451    cfg!(any(
452        target_os = "linux",
453        target_os = "android",
454        target_os = "macos",
455        target_os = "ios"
456    ))
457}
458
459#[cfg(all(feature = "fs", any(target_os = "linux", target_os = "android")))]
460fn replace_existing(staging: &Path, destination: &Path) -> io::Result<()> {
461    use std::ffi::CString;
462    use std::os::unix::ffi::OsStrExt;
463
464    let staging = CString::new(staging.as_os_str().as_bytes())?;
465    let destination = CString::new(destination.as_os_str().as_bytes())?;
466    // RENAME_EXCHANGE atomically requires both paths and leaves the old target at staging.
467    let result = unsafe {
468        libc::syscall(
469            libc::SYS_renameat2,
470            libc::AT_FDCWD,
471            staging.as_ptr(),
472            libc::AT_FDCWD,
473            destination.as_ptr(),
474            libc::RENAME_EXCHANGE,
475        )
476    };
477    if result == 0 {
478        Ok(())
479    } else {
480        Err(io::Error::last_os_error())
481    }
482}
483
484#[cfg(all(feature = "fs", any(target_os = "macos", target_os = "ios")))]
485fn replace_existing(staging: &Path, destination: &Path) -> io::Result<()> {
486    use std::ffi::CString;
487    use std::os::unix::ffi::OsStrExt;
488
489    let staging = CString::new(staging.as_os_str().as_bytes())?;
490    let destination = CString::new(destination.as_os_str().as_bytes())?;
491    // RENAME_SWAP atomically requires both paths and leaves the old target at staging.
492    let result =
493        unsafe { libc::renamex_np(staging.as_ptr(), destination.as_ptr(), libc::RENAME_SWAP) };
494    if result == 0 {
495        Ok(())
496    } else {
497        Err(io::Error::last_os_error())
498    }
499}
500
501#[cfg(all(
502    feature = "fs",
503    not(any(
504        target_os = "linux",
505        target_os = "android",
506        target_os = "macos",
507        target_os = "ios"
508    ))
509))]
510fn replace_existing(_staging: &Path, _destination: &Path) -> io::Result<()> {
511    Err(io::Error::new(
512        io::ErrorKind::Unsupported,
513        "strict replace-existing write is unsupported",
514    ))
515}
516
517/// A failure in Pack Archive Encoding followed by atomic file write.
518#[cfg(feature = "fs")]
519#[derive(Debug, thiserror::Error)]
520#[non_exhaustive]
521pub enum SavePackError {
522    #[error(transparent)]
523    Encode(#[from] EncodeError),
524    #[error("encoded Pack Archive could not be written: {source}")]
525    Write {
526        archive: PackArchiveBytes,
527        #[source]
528        source: FileWriteError,
529    },
530}
531
532/// Encodes and atomically writes one Pack while preserving bytes on failure.
533#[cfg(feature = "fs")]
534pub fn save_pack(
535    destination: impl AsRef<Path>,
536    pack: &Pack,
537    policy: FileWritePolicy,
538) -> Result<FileWriteReceipt, SavePackError> {
539    save_pack_with_limits(destination, pack, EncodeLimits::reference_v1(), policy)
540}
541
542/// Encodes under explicit resource ceilings and atomically writes one Pack.
543#[cfg(feature = "fs")]
544pub fn save_pack_with_limits(
545    destination: impl AsRef<Path>,
546    pack: &Pack,
547    encode_limits: EncodeLimits,
548    policy: FileWritePolicy,
549) -> Result<FileWriteReceipt, SavePackError> {
550    let archive = encode_with_limits(pack, encode_limits)?;
551    match write_file(destination, &archive, policy) {
552        Ok(receipt) => Ok(receipt),
553        Err(source) => Err(SavePackError::Write { archive, source }),
554    }
555}
556
557#[cfg(all(test, feature = "fs"))]
558mod tests {
559    use super::*;
560
561    #[test]
562    fn create_new_destination_race_does_not_replace_the_racing_file() {
563        let directory = tempfile::tempdir().unwrap();
564        let destination = directory.path().join("archive.typk");
565        let archive = PackArchiveBytes::from_vec(b"new archive".to_vec());
566
567        let error =
568            write_file_before_commit(&destination, &archive, FileWritePolicy::CreateNew, |_| {
569                std::fs::write(&destination, b"racing archive").unwrap()
570            })
571            .unwrap_err();
572
573        assert_eq!(error.phase(), FileWritePhase::Commit);
574        assert_eq!(error.commit_certainty(), CommitCertainty::NotCommitted);
575        assert_eq!(std::fs::read(&destination).unwrap(), b"racing archive");
576        assert_eq!(
577            std::fs::read(error.staging_residue().unwrap()).unwrap(),
578            archive.as_slice()
579        );
580    }
581
582    #[test]
583    fn replace_existing_destination_race_does_not_create_a_new_file() {
584        if !replace_existing_supported() {
585            return;
586        }
587        let directory = tempfile::tempdir().unwrap();
588        let destination = directory.path().join("archive.typk");
589        std::fs::write(&destination, b"old archive").unwrap();
590        let archive = PackArchiveBytes::from_vec(b"new archive".to_vec());
591
592        let error = write_file_before_commit(
593            &destination,
594            &archive,
595            FileWritePolicy::ReplaceExisting,
596            |_| std::fs::remove_file(&destination).unwrap(),
597        )
598        .unwrap_err();
599
600        assert_eq!(error.phase(), FileWritePhase::Commit);
601        assert_eq!(error.commit_certainty(), CommitCertainty::NotCommitted);
602        assert!(!destination.exists());
603        assert_eq!(
604            std::fs::read(error.staging_residue().unwrap()).unwrap(),
605            archive.as_slice()
606        );
607    }
608
609    #[test]
610    fn commit_failure_reports_staging_removed_by_a_race_as_absent() {
611        let directory = tempfile::tempdir().unwrap();
612        let destination = directory.path().join("archive.typk");
613        let archive = PackArchiveBytes::from_vec(b"new archive".to_vec());
614
615        let error = write_file_before_commit(
616            &destination,
617            &archive,
618            FileWritePolicy::CreateNew,
619            |staging| std::fs::remove_file(staging).unwrap(),
620        )
621        .unwrap_err();
622
623        assert_eq!(error.phase(), FileWritePhase::Commit);
624        assert_eq!(error.staging_residue_status(), StagingResidueStatus::Absent);
625        assert_eq!(error.staging_residue(), None);
626        assert!(!destination.exists());
627    }
628
629    #[test]
630    fn staging_write_handles_short_writes() {
631        let mut writer = FaultWriter::new(2, None, false);
632
633        write_staging(&mut writer, b"complete staging bytes").unwrap();
634
635        assert_eq!(writer.bytes, b"complete staging bytes");
636        assert!(writer.flush_called);
637    }
638
639    #[test]
640    fn staging_write_and_flush_faults_keep_their_phase() {
641        let mut write_failure = FaultWriter::new(3, Some(6), false);
642        let (phase, _) = write_staging(&mut write_failure, b"complete staging bytes").unwrap_err();
643        assert_eq!(phase, FileWritePhase::StagingWrite);
644        assert_eq!(write_failure.bytes, b"comple");
645
646        let mut flush_failure = FaultWriter::new(usize::MAX, None, true);
647        let (phase, _) = write_staging(&mut flush_failure, b"complete staging bytes").unwrap_err();
648        assert_eq!(phase, FileWritePhase::StagingFlush);
649        assert_eq!(flush_failure.bytes, b"complete staging bytes");
650    }
651
652    struct FaultWriter {
653        bytes: Vec<u8>,
654        maximum_write: usize,
655        fail_after: Option<usize>,
656        fail_flush: bool,
657        flush_called: bool,
658    }
659
660    impl FaultWriter {
661        fn new(maximum_write: usize, fail_after: Option<usize>, fail_flush: bool) -> Self {
662            Self {
663                bytes: Vec::new(),
664                maximum_write,
665                fail_after,
666                fail_flush,
667                flush_called: false,
668            }
669        }
670    }
671
672    impl Write for FaultWriter {
673        fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
674            if self.fail_after == Some(self.bytes.len()) {
675                return Err(io::Error::other("scripted staging write failure"));
676            }
677            let before_failure = self
678                .fail_after
679                .map_or(usize::MAX, |limit| limit - self.bytes.len());
680            let written = buffer.len().min(self.maximum_write).min(before_failure);
681            self.bytes.extend_from_slice(&buffer[..written]);
682            Ok(written)
683        }
684
685        fn flush(&mut self) -> io::Result<()> {
686            self.flush_called = true;
687            if self.fail_flush {
688                return Err(io::Error::other("scripted staging flush failure"));
689            }
690            Ok(())
691        }
692    }
693}
typst_pack/pack_archive/write.rs

typst_pack/pack_archive/
write.rs
