Skip to main content

tzap_core/
writer.rs

1use std::cell::RefCell;
2use std::collections::{BTreeMap, BTreeSet};
3use std::io::{Cursor, Read};
4use std::rc::Rc;
5use std::time::{Duration, Instant};
6
7use sha2::{Digest, Sha256};
8use uuid::Uuid;
9
10use crate::compression::{
11    compress_zstd_frame_with_dictionary_and_jobs, compress_zstd_frame_with_jobs,
12};
13use crate::crypto::{
14    aead_encrypt, build_aad, compute_integrity_tag, derive_nonce, HmacDomain, KdfParams, MasterKey,
15    Subkeys,
16};
17use crate::fec::encode_parity_gf16;
18use crate::format::{
19    AeadAlgo, ArchiveWriteError, BlockKind, CompressionAlgo, FecAlgo, FormatError, KdfAlgo,
20    BLOCK_RECORD_FRAMING_LEN, BOOTSTRAP_SIDECAR_HEADER_LEN, CRITICAL_RECOVERY_LOCATOR_LEN,
21    CRYPTO_EXTENSION_HEADER_LEN, CRYPTO_HEADER_FIXED_LEN, CRYPTO_HEADER_HMAC_LEN, FORMAT_VERSION,
22    MANIFEST_FOOTER_LEN, READER_MAX_CMRA_PARITY_PCT, READER_MAX_INDEX_ROOT_FEC_CLASS_SHARDS,
23    READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN, READER_MAX_ROOT_AUTH_FOOTER_LEN,
24    READER_MAX_ROOT_AUTH_SIGNER_IDENTITY_LEN, VOLUME_FORMAT_REV, VOLUME_HEADER_LEN,
25    VOLUME_TRAILER_LEN,
26};
27use crate::metadata::{
28    hash_prefix, normalize_lookup_file_path, validate_file_path_bytes, DirectoryHintEntry,
29    DirectoryHintShardEntry, DirectoryHintTableHeader, EnvelopeEntry, FileEntry, FrameEntry,
30    IndexRoot, IndexRootHeader, IndexShardHeader, ShardEntry, DIRECTORY_HINT_ENTRY_LEN,
31    DIRECTORY_HINT_TABLE_LEN, ENVELOPE_ENTRY_LEN, FILE_ENTRY_LEN, FRAME_ENTRY_LEN,
32    INDEX_SHARD_HEADER_LEN,
33};
34use crate::padding::suffix_pad_for_aead;
35use crate::root_auth::{
36    archive_root, critical_metadata_digest, data_block_merkle_leaf_hash,
37    data_block_merkle_root_from_leaf_hashes, fec_layout_digest, index_digest,
38    root_auth_descriptor_digest, signer_identity_digest, ArchiveRootInputs,
39    CriticalMetadataDigestInputs, FecLayoutObjectRow,
40};
41use crate::wire::{
42    BlockRecord, BootstrapSidecarHeader, CriticalMetadataImage, CriticalMetadataRecoveryHeader,
43    CriticalMetadataRecoveryShard, CriticalRecoveryLocator, CryptoHeader, CryptoHeaderFixed,
44    ManifestFooter, RootAuthFooterV1, SerializedRegion, VolumeHeader, VolumeTrailer,
45};
46
47const TAR_BLOCK_LEN: usize = 512;
48const MAX_REED_SOLOMON_GF16_SHARDS: u64 = 65_535;
49const MIN_BLOCK_SIZE: u32 = 4096;
50const DEFAULT_BLOCK_SIZE: u32 = 64 * 1024;
51const DEFAULT_CHUNK_SIZE: u32 = 256 * 1024;
52const DEFAULT_ENVELOPE_TARGET_SIZE: u32 = 1024 * 1024;
53const DEFAULT_FEC_DATA_SHARDS: u16 = 224;
54const DEFAULT_FEC_PARITY_SHARDS: u16 = 1;
55const DEFAULT_INDEX_FEC_DATA_SHARDS: u16 = 16;
56const DEFAULT_INDEX_FEC_PARITY_SHARDS: u16 = 1;
57const MIN_INDEX_ROOT_FEC_DATA_SHARDS: u16 = 16;
58const DEFAULT_INDEX_ROOT_FEC_DATA_SHARDS: u16 = MIN_INDEX_ROOT_FEC_DATA_SHARDS;
59const DEFAULT_INDEX_ROOT_FEC_PARITY_SHARDS: u16 = 1;
60const DEFAULT_STRIPE_WIDTH: u32 = 8;
61const DEFAULT_VOLUME_LOSS_TOLERANCE: u8 = 1;
62const DEFAULT_BIT_ROT_BUFFER_PCT: u8 = 5;
63const DEFAULT_FILES_PER_INDEX_SHARD: usize = 10_000;
64const DIRECTORY_HINT_REQUIRED_FILE_COUNT: usize = 100_000;
65const MAX_FILES_PER_INDEX_SHARD: usize = 1_000_000;
66const MAX_HASH_PREFIX_RUN_FILES: usize = 50_000;
67const DEFAULT_DIRECTORY_HINT_ENTRIES_PER_SHARD: usize = 10_000;
68const CMRA_SHARD_SIZE: usize = 512;
69
70fn default_jobs() -> usize {
71    std::thread::available_parallelism()
72        .map(|jobs| jobs.get())
73        .unwrap_or(1)
74}
75
76fn should_emit_directory_hints(file_count: usize) -> bool {
77    file_count > DIRECTORY_HINT_REQUIRED_FILE_COUNT
78}
79
80#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub struct WriterOptions {
82    pub block_size: u32,
83    pub chunk_size: u32,
84    pub envelope_target_size: u32,
85    pub stripe_width: u32,
86    pub volume_loss_tolerance: u8,
87    pub bit_rot_buffer_pct: u8,
88    pub zstd_level: i32,
89    pub jobs: usize,
90    pub aead_algo: AeadAlgo,
91    pub fec_data_shards: u16,
92    pub fec_parity_shards: u16,
93    pub index_fec_data_shards: u16,
94    pub index_fec_parity_shards: u16,
95    pub index_root_fec_data_shards: u16,
96    pub index_root_fec_parity_shards: u16,
97    pub max_path_length: u32,
98    pub target_volume_size: Option<u64>,
99    pub archive_uuid: Option<[u8; 16]>,
100    pub session_id: Option<[u8; 16]>,
101    pub closed_at_ns: i64,
102}
103
104impl Default for WriterOptions {
105    fn default() -> Self {
106        Self {
107            block_size: DEFAULT_BLOCK_SIZE,
108            chunk_size: DEFAULT_CHUNK_SIZE,
109            envelope_target_size: DEFAULT_ENVELOPE_TARGET_SIZE,
110            stripe_width: DEFAULT_STRIPE_WIDTH,
111            volume_loss_tolerance: DEFAULT_VOLUME_LOSS_TOLERANCE,
112            bit_rot_buffer_pct: DEFAULT_BIT_ROT_BUFFER_PCT,
113            zstd_level: 3,
114            jobs: default_jobs(),
115            aead_algo: AeadAlgo::AesGcmSiv256,
116            fec_data_shards: DEFAULT_FEC_DATA_SHARDS,
117            fec_parity_shards: DEFAULT_FEC_PARITY_SHARDS,
118            index_fec_data_shards: DEFAULT_INDEX_FEC_DATA_SHARDS,
119            index_fec_parity_shards: DEFAULT_INDEX_FEC_PARITY_SHARDS,
120            index_root_fec_data_shards: DEFAULT_INDEX_ROOT_FEC_DATA_SHARDS,
121            index_root_fec_parity_shards: DEFAULT_INDEX_ROOT_FEC_PARITY_SHARDS,
122            max_path_length: 4096,
123            target_volume_size: None,
124            archive_uuid: None,
125            session_id: None,
126            closed_at_ns: 0,
127        }
128    }
129}
130
131#[derive(Debug, Clone, Copy, PartialEq, Eq)]
132pub struct RootAuthWriterConfig<'a> {
133    pub authenticator_id: u16,
134    pub signer_identity_type: u16,
135    pub signer_identity: &'a [u8],
136    pub authenticator_value_length: u32,
137}
138
139#[derive(Debug, Clone, Copy, PartialEq, Eq)]
140pub struct RootAuthSigningRequest {
141    pub archive_uuid: [u8; 16],
142    pub session_id: [u8; 16],
143    pub archive_root: [u8; 32],
144}
145
146pub type RootAuthAuthenticator<'a> =
147    dyn FnMut(&RootAuthSigningRequest) -> Result<Vec<u8>, FormatError> + 'a;
148
149#[derive(Debug, Clone, Copy, PartialEq, Eq)]
150pub struct RegularFile<'a> {
151    pub path: &'a str,
152    pub contents: &'a [u8],
153    pub mode: u32,
154    pub mtime: u64,
155}
156
157impl<'a> RegularFile<'a> {
158    pub fn new(path: &'a str, contents: &'a [u8]) -> Self {
159        Self {
160            path,
161            contents,
162            mode: 0o644,
163            mtime: 0,
164        }
165    }
166}
167
168/// Re-openable source for one regular file written into an archive.
169///
170/// The writer may replan when options such as target volume sizing need another
171/// pass, so implementations must return a fresh reader from each `open` call.
172pub trait RegularFileSource {
173    fn archive_path(&self) -> &str;
174    fn file_data_size(&self) -> u64;
175    fn mode(&self) -> u32;
176    fn mtime(&self) -> u64;
177    fn open(&self) -> Result<Box<dyn Read + '_>, ArchiveWriteError>;
178}
179
180impl RegularFileSource for RegularFile<'_> {
181    fn archive_path(&self) -> &str {
182        self.path
183    }
184
185    fn file_data_size(&self) -> u64 {
186        self.contents.len() as u64
187    }
188
189    fn mode(&self) -> u32 {
190        self.mode
191    }
192
193    fn mtime(&self) -> u64 {
194        self.mtime
195    }
196
197    fn open(&self) -> Result<Box<dyn Read + '_>, ArchiveWriteError> {
198        Ok(Box::new(Cursor::new(self.contents)))
199    }
200}
201
202/// Receives source-byte progress while the archive writer consumes file data.
203///
204/// The writer may open source files multiple times when planning or replanning
205/// an archive. Progress callbacks report each archive member's logical source
206/// bytes at most once, capped by [`RegularFileSource::file_data_size`].
207pub trait ArchiveWriteProgressSink {
208    /// Reports newly consumed source bytes for one archive member.
209    fn source_bytes_read(&mut self, archive_path: &str, bytes: u64);
210}
211
212impl<F> ArchiveWriteProgressSink for F
213where
214    F: FnMut(&str, u64),
215{
216    fn source_bytes_read(&mut self, archive_path: &str, bytes: u64) {
217        self(archive_path, bytes);
218    }
219}
220
221struct SourceProgressState<'a> {
222    sink: &'a mut dyn ArchiveWriteProgressSink,
223    reported_by_path: BTreeMap<String, u64>,
224}
225
226impl<'a> SourceProgressState<'a> {
227    fn new(sink: &'a mut dyn ArchiveWriteProgressSink) -> Self {
228        Self {
229            sink,
230            reported_by_path: BTreeMap::new(),
231        }
232    }
233
234    fn record(&mut self, archive_path: &str, bytes: u64, file_data_size: u64) {
235        if bytes == 0 || file_data_size == 0 {
236            return;
237        }
238        let reported = self
239            .reported_by_path
240            .entry(archive_path.to_owned())
241            .or_default();
242        let capped_next = reported.saturating_add(bytes).min(file_data_size);
243        let delta = capped_next.saturating_sub(*reported);
244        if delta == 0 {
245            return;
246        }
247        *reported = capped_next;
248        self.sink.source_bytes_read(archive_path, delta);
249    }
250}
251
252struct ProgressRegularFileSource<'a, S> {
253    inner: &'a S,
254    state: Rc<RefCell<SourceProgressState<'a>>>,
255}
256
257impl<S: RegularFileSource> RegularFileSource for ProgressRegularFileSource<'_, S> {
258    fn archive_path(&self) -> &str {
259        self.inner.archive_path()
260    }
261
262    fn file_data_size(&self) -> u64 {
263        self.inner.file_data_size()
264    }
265
266    fn mode(&self) -> u32 {
267        self.inner.mode()
268    }
269
270    fn mtime(&self) -> u64 {
271        self.inner.mtime()
272    }
273
274    fn open(&self) -> Result<Box<dyn Read + '_>, ArchiveWriteError> {
275        Ok(Box::new(SourceProgressReader {
276            inner: self.inner.open()?,
277            archive_path: self.inner.archive_path().to_owned(),
278            file_data_size: self.inner.file_data_size(),
279            state: Rc::clone(&self.state),
280        }))
281    }
282}
283
284struct SourceProgressReader<'a> {
285    inner: Box<dyn Read + 'a>,
286    archive_path: String,
287    file_data_size: u64,
288    state: Rc<RefCell<SourceProgressState<'a>>>,
289}
290
291impl Read for SourceProgressReader<'_> {
292    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
293        let bytes = self.inner.read(buf)?;
294        self.state
295            .borrow_mut()
296            .record(&self.archive_path, bytes as u64, self.file_data_size);
297        Ok(bytes)
298    }
299}
300
301fn progress_sources<'a, S: RegularFileSource>(
302    files: &'a [S],
303    sink: &'a mut dyn ArchiveWriteProgressSink,
304) -> Vec<ProgressRegularFileSource<'a, S>> {
305    let state = Rc::new(RefCell::new(SourceProgressState::new(sink)));
306    files
307        .iter()
308        .map(|inner| ProgressRegularFileSource {
309            inner,
310            state: Rc::clone(&state),
311        })
312        .collect()
313}
314
315/// Streaming destination for archive volumes and optional bootstrap sidecar.
316///
317/// Calls arrive in archive order for each volume, but records are interleaved
318/// across volumes according to the archive stripe layout.
319pub trait ArchiveWriteSink {
320    fn begin_archive(&mut self, volume_count: usize) -> Result<(), ArchiveWriteError>;
321    fn write_volume(&mut self, volume_index: usize, bytes: &[u8]) -> Result<(), ArchiveWriteError>;
322    fn write_bootstrap_sidecar(&mut self, bytes: &[u8]) -> Result<(), ArchiveWriteError>;
323}
324
325#[derive(Debug, Clone, PartialEq, Eq)]
326pub struct WrittenArchiveSummary {
327    pub volume_count: usize,
328    pub archive_bytes: u64,
329    pub bootstrap_sidecar_bytes: u64,
330    pub archive_uuid: [u8; 16],
331    pub session_id: [u8; 16],
332    pub timings: WriterTimings,
333}
334
335#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
336pub struct WriterTimings {
337    pub total: Duration,
338    pub plan_payload: Duration,
339    pub plan_metadata: Duration,
340    pub emit_payload: Duration,
341    pub emit_metadata: Duration,
342}
343
344impl WriterTimings {
345    fn add_assign(&mut self, other: Self) {
346        self.total += other.total;
347        self.plan_payload += other.plan_payload;
348        self.plan_metadata += other.plan_metadata;
349        self.emit_payload += other.emit_payload;
350        self.emit_metadata += other.emit_metadata;
351    }
352}
353
354/// In-memory sink used by the compatibility writer APIs.
355#[derive(Debug, Clone, Default, PartialEq, Eq)]
356pub struct MemoryArchiveSink {
357    pub volumes: Vec<Vec<u8>>,
358    pub bootstrap_sidecar: Vec<u8>,
359}
360
361impl ArchiveWriteSink for MemoryArchiveSink {
362    fn begin_archive(&mut self, volume_count: usize) -> Result<(), ArchiveWriteError> {
363        self.volumes = vec![Vec::new(); volume_count];
364        self.bootstrap_sidecar.clear();
365        Ok(())
366    }
367
368    fn write_volume(&mut self, volume_index: usize, bytes: &[u8]) -> Result<(), ArchiveWriteError> {
369        let volume = self
370            .volumes
371            .get_mut(volume_index)
372            .ok_or(FormatError::WriterInvariant(
373                "volume sink index is out of bounds",
374            ))?;
375        volume.extend_from_slice(bytes);
376        Ok(())
377    }
378
379    fn write_bootstrap_sidecar(&mut self, bytes: &[u8]) -> Result<(), ArchiveWriteError> {
380        self.bootstrap_sidecar.extend_from_slice(bytes);
381        Ok(())
382    }
383}
384
385/// Completed archive artifacts produced by the compatibility writer APIs.
386///
387/// APIs returning this value build all volume bytes before returning. Use the
388/// sink writer when archive bytes should be delivered incrementally.
389#[derive(Debug, Clone, PartialEq, Eq)]
390pub struct WrittenArchive {
391    pub bytes: Vec<u8>,
392    pub volumes: Vec<Vec<u8>>,
393    pub bootstrap_sidecar: Vec<u8>,
394    pub archive_uuid: [u8; 16],
395    pub session_id: [u8; 16],
396    pub timings: WriterTimings,
397}
398
399#[derive(Debug, Clone)]
400struct TarMember {
401    path: Vec<u8>,
402    tar_member_group_start: u64,
403    tar_member_group_size: u64,
404    file_data_size: u64,
405    mode: u32,
406    mtime: u64,
407}
408
409#[derive(Debug, Clone)]
410struct PayloadFrame {
411    frame_index: u64,
412    envelope_index: u64,
413    member_index: usize,
414    offset_in_envelope: u32,
415    compressed_size: u32,
416    decompressed_size: u32,
417    flags: u32,
418    tar_stream_offset: u64,
419}
420
421#[derive(Debug, Clone)]
422struct FileRow {
423    path_hash: [u8; 8],
424    path: Vec<u8>,
425    member_index: usize,
426    member: TarMember,
427}
428
429#[derive(Debug, Clone)]
430struct PlannedIndexShard {
431    shard_index: u64,
432    plaintext: Vec<u8>,
433    file_count: u32,
434    first_path_hash: [u8; 8],
435    last_path_hash: [u8; 8],
436}
437
438#[derive(Debug, Clone)]
439struct PlannedDirectoryHintShard {
440    hint_shard_index: u64,
441    plaintext: Vec<u8>,
442    entry_count: u64,
443    first_dir_hash: [u8; 8],
444    last_dir_hash: [u8; 8],
445}
446
447#[derive(Debug, Clone)]
448#[cfg(test)]
449struct PayloadEnvelope {
450    envelope_index: u64,
451    plaintext: Vec<u8>,
452}
453
454#[derive(Debug, Clone)]
455struct PayloadObject {
456    envelope_index: u64,
457    plaintext_size: u32,
458    object: ObjectExtent,
459}
460
461#[derive(Debug, Clone)]
462struct EncryptedObject {
463    first_block_index: u64,
464    data_block_count: u32,
465    parity_block_count: u32,
466    encrypted_size: u32,
467    records: Vec<BlockRecord>,
468}
469
470#[derive(Debug, Clone, Copy, PartialEq, Eq)]
471struct ObjectExtent {
472    first_block_index: u64,
473    data_block_count: u32,
474    parity_block_count: u32,
475    encrypted_size: u32,
476}
477
478#[derive(Debug, Clone, Copy, PartialEq, Eq)]
479struct PlannedEncryptedObject {
480    data_block_count: u32,
481    parity_block_count: u32,
482    encrypted_size: u32,
483}
484
485#[derive(Debug, Clone, Copy, PartialEq, Eq)]
486enum MetadataObjectKind {
487    IndexRoot,
488    Dictionary,
489}
490
491impl MetadataObjectKind {
492    fn too_large_error(self) -> FormatError {
493        match self {
494            Self::IndexRoot => FormatError::WriterUnsupported("IndexRoot too large"),
495            Self::Dictionary => FormatError::WriterUnsupported("dictionary object too large"),
496        }
497    }
498}
499
500impl ObjectExtent {
501    fn new(first_block_index: u64, plan: PlannedEncryptedObject) -> Result<Self, FormatError> {
502        Ok(Self {
503            first_block_index,
504            data_block_count: plan.data_block_count,
505            parity_block_count: plan.parity_block_count,
506            encrypted_size: plan.encrypted_size,
507        })
508    }
509
510    fn next_block_index(self) -> Result<u64, FormatError> {
511        checked_u64_add(
512            self.first_block_index,
513            self.data_block_count as u64 + self.parity_block_count as u64,
514            "next_block_index",
515        )
516    }
517}
518
519impl From<&EncryptedObject> for ObjectExtent {
520    fn from(object: &EncryptedObject) -> Self {
521        Self {
522            first_block_index: object.first_block_index,
523            data_block_count: object.data_block_count,
524            parity_block_count: object.parity_block_count,
525            encrypted_size: object.encrypted_size,
526        }
527    }
528}
529
530#[derive(Debug, Clone)]
531struct PlannedDirectoryHintObject {
532    hint_shard_index: u64,
533    compressed: Vec<u8>,
534    extent: ObjectExtent,
535}
536
537struct WriterPlan {
538    options: WriterOptions,
539    archive_uuid: [u8; 16],
540    session_id: [u8; 16],
541    crypto_header: Vec<u8>,
542    tar_members: Vec<TarMember>,
543    frames: Vec<PayloadFrame>,
544    payload_objects: Vec<PayloadObject>,
545    index_root_plaintext: Vec<u8>,
546    compressed_index_root: Vec<u8>,
547    index_root_extent: ObjectExtent,
548    index_shard_objects: Vec<PlannedIndexShardObject>,
549    shard_entries: Vec<ShardEntry>,
550    compressed_dictionary: Option<Vec<u8>>,
551    dictionary_extent: Option<(ObjectExtent, u32)>,
552    directory_hint_objects: Vec<PlannedDirectoryHintObject>,
553    directory_hint_entries: Vec<DirectoryHintShardEntry>,
554    root_auth_footer_length: Option<u32>,
555    total_block_count: u64,
556}
557
558struct PlannedIndexShardObject {
559    shard_index: u64,
560    compressed: Vec<u8>,
561    extent: ObjectExtent,
562}
563
564struct PayloadPlanning {
565    tar_members: Vec<TarMember>,
566    frames: Vec<PayloadFrame>,
567    payload_objects: Vec<PayloadObject>,
568    payload_block_count: u64,
569    tar_total_size: u64,
570    content_sha256: [u8; 32],
571}
572
573struct PayloadEnvelopeBuilder {
574    envelope_index: u64,
575    plaintext: Vec<u8>,
576}
577
578#[derive(Debug, Clone, PartialEq, Eq)]
579pub(crate) struct StreamingRegularMember {
580    pub archive_path: Vec<u8>,
581    pub file_data_size: u64,
582    pub mode: u32,
583    pub mtime: u64,
584}
585
586struct WriterEmissionState {
587    volume_headers: Vec<[u8; VOLUME_HEADER_LEN]>,
588    bytes_written: Vec<u64>,
589    record_counts: Vec<u64>,
590    data_leaf_hashes: Vec<(u64, [u8; 32])>,
591    next_block_index: u64,
592}
593
594pub(crate) struct StreamingArchiveWriter<'a, O: ArchiveWriteSink> {
595    sink: &'a mut O,
596    options: WriterOptions,
597    archive_uuid: [u8; 16],
598    session_id: [u8; 16],
599    crypto_header: Vec<u8>,
600    subkeys: Subkeys,
601    tar_members: Vec<TarMember>,
602    frames: Vec<PayloadFrame>,
603    payload_objects: Vec<PayloadObject>,
604    payload_block_count: u64,
605    tar_total_size: u64,
606    hasher: Sha256,
607    next_frame_index: u64,
608    envelope: PayloadEnvelopeBuilder,
609    emission_state: WriterEmissionState,
610}
611
612pub fn write_archive(
613    files: &[RegularFile<'_>],
614    master_key: &MasterKey,
615    options: WriterOptions,
616) -> Result<WrittenArchive, FormatError> {
617    write_archive_inner(
618        files,
619        master_key,
620        options,
621        None,
622        &KdfParams::Raw,
623        None,
624        None,
625    )
626}
627
628pub fn write_archive_unencrypted(
629    files: &[RegularFile<'_>],
630    mut options: WriterOptions,
631) -> Result<WrittenArchive, FormatError> {
632    options.aead_algo = AeadAlgo::None;
633    let placeholder = MasterKey::from_raw_key(&[0; 32])?;
634    write_archive_inner(
635        files,
636        &placeholder,
637        options,
638        None,
639        &KdfParams::None,
640        None,
641        None,
642    )
643}
644
645pub fn write_archive_with_kdf(
646    files: &[RegularFile<'_>],
647    master_key: &MasterKey,
648    options: WriterOptions,
649    kdf_params: &KdfParams,
650) -> Result<WrittenArchive, FormatError> {
651    write_archive_inner(files, master_key, options, None, kdf_params, None, None)
652}
653
654pub fn write_archive_with_root_auth<F>(
655    files: &[RegularFile<'_>],
656    master_key: &MasterKey,
657    options: WriterOptions,
658    root_auth: RootAuthWriterConfig<'_>,
659    mut authenticator: F,
660) -> Result<WrittenArchive, FormatError>
661where
662    F: FnMut(&RootAuthSigningRequest) -> Result<Vec<u8>, FormatError>,
663{
664    write_archive_inner(
665        files,
666        master_key,
667        options,
668        None,
669        &KdfParams::Raw,
670        Some(root_auth),
671        Some(&mut authenticator),
672    )
673}
674
675pub fn write_archive_with_root_auth_and_kdf<F>(
676    files: &[RegularFile<'_>],
677    master_key: &MasterKey,
678    options: WriterOptions,
679    kdf_params: &KdfParams,
680    root_auth: RootAuthWriterConfig<'_>,
681    mut authenticator: F,
682) -> Result<WrittenArchive, FormatError>
683where
684    F: FnMut(&RootAuthSigningRequest) -> Result<Vec<u8>, FormatError>,
685{
686    write_archive_inner(
687        files,
688        master_key,
689        options,
690        None,
691        kdf_params,
692        Some(root_auth),
693        Some(&mut authenticator),
694    )
695}
696
697pub fn write_archive_with_dictionary_and_root_auth<F>(
698    files: &[RegularFile<'_>],
699    master_key: &MasterKey,
700    options: WriterOptions,
701    dictionary: &[u8],
702    root_auth: RootAuthWriterConfig<'_>,
703    mut authenticator: F,
704) -> Result<WrittenArchive, FormatError>
705where
706    F: FnMut(&RootAuthSigningRequest) -> Result<Vec<u8>, FormatError>,
707{
708    write_archive_inner(
709        files,
710        master_key,
711        options,
712        Some(dictionary),
713        &KdfParams::Raw,
714        Some(root_auth),
715        Some(&mut authenticator),
716    )
717}
718
719pub fn write_archive_with_dictionary_kdf_and_root_auth<F>(
720    files: &[RegularFile<'_>],
721    master_key: &MasterKey,
722    options: WriterOptions,
723    dictionary: &[u8],
724    kdf_params: &KdfParams,
725    root_auth: RootAuthWriterConfig<'_>,
726    mut authenticator: F,
727) -> Result<WrittenArchive, FormatError>
728where
729    F: FnMut(&RootAuthSigningRequest) -> Result<Vec<u8>, FormatError>,
730{
731    write_archive_inner(
732        files,
733        master_key,
734        options,
735        Some(dictionary),
736        kdf_params,
737        Some(root_auth),
738        Some(&mut authenticator),
739    )
740}
741
742pub fn write_archive_with_dictionary(
743    files: &[RegularFile<'_>],
744    master_key: &MasterKey,
745    options: WriterOptions,
746    dictionary: &[u8],
747) -> Result<WrittenArchive, FormatError> {
748    if dictionary.is_empty() {
749        return Err(FormatError::WriterUnsupported(
750            "dictionary archives require a non-empty dictionary",
751        ));
752    }
753    if files.is_empty() {
754        return Err(FormatError::WriterUnsupported(
755            "dictionary archives require at least one file",
756        ));
757    }
758    if dictionary.len() > u32::MAX as usize {
759        return Err(FormatError::WriterUnsupported(
760            "dictionary decompressed size exceeds u32",
761        ));
762    }
763    write_archive_inner(
764        files,
765        master_key,
766        options,
767        Some(dictionary),
768        &KdfParams::Raw,
769        None,
770        None,
771    )
772}
773
774pub fn write_archive_with_dictionary_and_kdf(
775    files: &[RegularFile<'_>],
776    master_key: &MasterKey,
777    options: WriterOptions,
778    dictionary: &[u8],
779    kdf_params: &KdfParams,
780) -> Result<WrittenArchive, FormatError> {
781    if dictionary.is_empty() {
782        return Err(FormatError::WriterUnsupported(
783            "dictionary archives require a non-empty dictionary",
784        ));
785    }
786    if files.is_empty() {
787        return Err(FormatError::WriterUnsupported(
788            "dictionary archives require at least one file",
789        ));
790    }
791    if dictionary.len() > u32::MAX as usize {
792        return Err(FormatError::WriterUnsupported(
793            "dictionary decompressed size exceeds u32",
794        ));
795    }
796    write_archive_inner(
797        files,
798        master_key,
799        options,
800        Some(dictionary),
801        kdf_params,
802        None,
803        None,
804    )
805}
806
807#[allow(clippy::too_many_arguments)]
808pub fn write_archive_sources_to_sink<S, O>(
809    files: &[S],
810    master_key: &MasterKey,
811    options: WriterOptions,
812    dictionary: Option<&[u8]>,
813    kdf_params: &KdfParams,
814    root_auth: Option<RootAuthWriterConfig<'_>>,
815    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
816    sink: &mut O,
817) -> Result<WrittenArchiveSummary, ArchiveWriteError>
818where
819    S: RegularFileSource,
820    O: ArchiveWriteSink,
821{
822    write_archive_stream_inner(
823        files,
824        master_key,
825        options,
826        dictionary,
827        kdf_params,
828        root_auth,
829        authenticator,
830        sink,
831    )
832}
833
834#[allow(clippy::too_many_arguments)]
835pub fn write_archive_sources_to_sink_with_progress<S, O>(
836    files: &[S],
837    master_key: &MasterKey,
838    options: WriterOptions,
839    dictionary: Option<&[u8]>,
840    kdf_params: &KdfParams,
841    root_auth: Option<RootAuthWriterConfig<'_>>,
842    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
843    sink: &mut O,
844    progress: &mut dyn ArchiveWriteProgressSink,
845) -> Result<WrittenArchiveSummary, ArchiveWriteError>
846where
847    S: RegularFileSource,
848    O: ArchiveWriteSink,
849{
850    let progress_files = progress_sources(files, progress);
851    write_archive_stream_inner(
852        &progress_files,
853        master_key,
854        options,
855        dictionary,
856        kdf_params,
857        root_auth,
858        authenticator,
859        sink,
860    )
861}
862
863#[allow(clippy::too_many_arguments)]
864pub fn write_archive_sources_to_sink_single_pass<S, O>(
865    files: &[S],
866    master_key: &MasterKey,
867    options: WriterOptions,
868    kdf_params: &KdfParams,
869    root_auth: Option<RootAuthWriterConfig<'_>>,
870    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
871    sink: &mut O,
872) -> Result<WrittenArchiveSummary, ArchiveWriteError>
873where
874    S: RegularFileSource,
875    O: ArchiveWriteSink,
876{
877    write_single_pass_archive_to_sink(
878        master_key,
879        options,
880        kdf_params,
881        root_auth,
882        authenticator,
883        sink,
884        |writer| {
885            for file in files {
886                let archive_path =
887                    normalize_lookup_file_path(file.archive_path(), options.max_path_length)?;
888                let mut reader = file.open()?;
889                writer.write_regular_member_from_reader(
890                    StreamingRegularMember {
891                        archive_path,
892                        file_data_size: file.file_data_size(),
893                        mode: file.mode(),
894                        mtime: file.mtime(),
895                    },
896                    reader.as_mut(),
897                )?;
898            }
899            Ok(())
900        },
901    )
902}
903
904#[allow(clippy::too_many_arguments)]
905pub fn write_archive_sources_to_sink_single_pass_with_progress<S, O>(
906    files: &[S],
907    master_key: &MasterKey,
908    options: WriterOptions,
909    kdf_params: &KdfParams,
910    root_auth: Option<RootAuthWriterConfig<'_>>,
911    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
912    sink: &mut O,
913    progress: &mut dyn ArchiveWriteProgressSink,
914) -> Result<WrittenArchiveSummary, ArchiveWriteError>
915where
916    S: RegularFileSource,
917    O: ArchiveWriteSink,
918{
919    let progress_files = progress_sources(files, progress);
920    write_archive_sources_to_sink_single_pass(
921        &progress_files,
922        master_key,
923        options,
924        kdf_params,
925        root_auth,
926        authenticator,
927        sink,
928    )
929}
930
931#[allow(clippy::too_many_arguments)]
932pub fn write_archive_sources_to_sink_ordered_parallel<S, O>(
933    files: &[S],
934    master_key: &MasterKey,
935    options: WriterOptions,
936    kdf_params: &KdfParams,
937    root_auth: Option<RootAuthWriterConfig<'_>>,
938    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
939    sink: &mut O,
940) -> Result<WrittenArchiveSummary, ArchiveWriteError>
941where
942    S: RegularFileSource,
943    O: ArchiveWriteSink,
944{
945    write_ordered_parallel_archive_to_sink(
946        files,
947        master_key,
948        options,
949        kdf_params,
950        root_auth,
951        authenticator,
952        sink,
953    )
954}
955
956#[allow(clippy::too_many_arguments)]
957pub fn write_archive_sources_to_sink_ordered_parallel_with_progress<S, O>(
958    files: &[S],
959    master_key: &MasterKey,
960    options: WriterOptions,
961    kdf_params: &KdfParams,
962    root_auth: Option<RootAuthWriterConfig<'_>>,
963    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
964    sink: &mut O,
965    progress: &mut dyn ArchiveWriteProgressSink,
966) -> Result<WrittenArchiveSummary, ArchiveWriteError>
967where
968    S: RegularFileSource,
969    O: ArchiveWriteSink,
970{
971    let progress_files = progress_sources(files, progress);
972    write_ordered_parallel_archive_to_sink(
973        &progress_files,
974        master_key,
975        options,
976        kdf_params,
977        root_auth,
978        authenticator,
979        sink,
980    )
981}
982
983fn write_archive_inner(
984    files: &[RegularFile<'_>],
985    master_key: &MasterKey,
986    options: WriterOptions,
987    dictionary: Option<&[u8]>,
988    kdf_params: &KdfParams,
989    root_auth: Option<RootAuthWriterConfig<'_>>,
990    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
991) -> Result<WrittenArchive, FormatError> {
992    let mut sink = MemoryArchiveSink::default();
993    let summary = write_archive_stream_inner(
994        files,
995        master_key,
996        options,
997        dictionary,
998        kdf_params,
999        root_auth,
1000        authenticator,
1001        &mut sink,
1002    )
1003    .map_err(format_error_from_archive_write_error)?;
1004    Ok(WrittenArchive {
1005        bytes: sink
1006            .volumes
1007            .first()
1008            .cloned()
1009            .ok_or(FormatError::WriterInvariant("no volumes emitted"))?,
1010        volumes: sink.volumes,
1011        bootstrap_sidecar: sink.bootstrap_sidecar,
1012        archive_uuid: summary.archive_uuid,
1013        session_id: summary.session_id,
1014        timings: summary.timings,
1015    })
1016}
1017
1018fn format_error_from_archive_write_error(error: ArchiveWriteError) -> FormatError {
1019    match error {
1020        ArchiveWriteError::Format(error) => error,
1021        ArchiveWriteError::Io(_) => {
1022            FormatError::WriterInvariant("in-memory archive writer returned I/O")
1023        }
1024    }
1025}
1026
1027fn writer_subkeys(
1028    master_key: &MasterKey,
1029    aead_algo: AeadAlgo,
1030    archive_uuid: &[u8; 16],
1031    session_id: &[u8; 16],
1032) -> Result<Subkeys, FormatError> {
1033    if aead_algo.is_encrypted() {
1034        Subkeys::derive(master_key, archive_uuid, session_id)
1035    } else {
1036        Ok(Subkeys::unencrypted_placeholder())
1037    }
1038}
1039
1040#[allow(clippy::too_many_arguments)]
1041fn write_archive_stream_inner<S, O>(
1042    files: &[S],
1043    master_key: &MasterKey,
1044    options: WriterOptions,
1045    dictionary: Option<&[u8]>,
1046    kdf_params: &KdfParams,
1047    root_auth: Option<RootAuthWriterConfig<'_>>,
1048    mut authenticator: Option<&mut RootAuthAuthenticator<'_>>,
1049    sink: &mut O,
1050) -> Result<WrittenArchiveSummary, ArchiveWriteError>
1051where
1052    S: RegularFileSource,
1053    O: ArchiveWriteSink,
1054{
1055    let total_started = Instant::now();
1056    validate_dictionary_inputs(files.is_empty(), dictionary)?;
1057    if let Some(root_auth) = root_auth {
1058        validate_root_auth_writer_config(root_auth)?;
1059    }
1060    let mut requested_options = options;
1061    if requested_options.target_volume_size.is_some() {
1062        requested_options.stripe_width = requested_options
1063            .stripe_width
1064            .max(requested_options.volume_loss_tolerance as u32 + 1);
1065    }
1066    let archive_uuid = requested_options
1067        .archive_uuid
1068        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1069    let session_id = requested_options
1070        .session_id
1071        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1072    let mut accumulated_timings = WriterTimings::default();
1073
1074    loop {
1075        let planned_options = plan_writer_options(requested_options)?;
1076        let timed_plan = build_writer_plan(
1077            files,
1078            master_key,
1079            planned_options,
1080            dictionary,
1081            kdf_params,
1082            archive_uuid,
1083            session_id,
1084            root_auth,
1085        )?;
1086        accumulated_timings.add_assign(timed_plan.timings);
1087        let plan = timed_plan.plan;
1088        if let Some(target_volume_size) = planned_options.target_volume_size {
1089            let required = required_stripe_width_for_plan(&plan, master_key, target_volume_size)?;
1090            if required > planned_options.stripe_width {
1091                requested_options.stripe_width = required;
1092                continue;
1093            }
1094        }
1095        let mut summary = emit_writer_plan(
1096            files,
1097            master_key,
1098            dictionary,
1099            root_auth,
1100            authenticator.take(),
1101            plan,
1102            sink,
1103        )?;
1104        summary.timings.add_assign(accumulated_timings);
1105        summary.timings.total = total_started.elapsed();
1106        return Ok(summary);
1107    }
1108}
1109
1110fn validate_dictionary_inputs(
1111    files_are_empty: bool,
1112    dictionary: Option<&[u8]>,
1113) -> Result<(), FormatError> {
1114    if let Some(dictionary) = dictionary {
1115        if dictionary.is_empty() {
1116            return Err(FormatError::WriterUnsupported(
1117                "dictionary archives require a non-empty dictionary",
1118            ));
1119        }
1120        if files_are_empty {
1121            return Err(FormatError::WriterUnsupported(
1122                "dictionary archives require at least one file",
1123            ));
1124        }
1125        if dictionary.len() > u32::MAX as usize {
1126            return Err(FormatError::WriterUnsupported(
1127                "dictionary decompressed size exceeds u32",
1128            ));
1129        }
1130    }
1131    Ok(())
1132}
1133
1134#[allow(clippy::too_many_arguments)]
1135struct TimedWriterPlan {
1136    plan: WriterPlan,
1137    timings: WriterTimings,
1138}
1139
1140#[allow(clippy::too_many_arguments)]
1141fn build_writer_plan<S: RegularFileSource>(
1142    files: &[S],
1143    master_key: &MasterKey,
1144    options: WriterOptions,
1145    dictionary: Option<&[u8]>,
1146    kdf_params: &KdfParams,
1147    archive_uuid: [u8; 16],
1148    session_id: [u8; 16],
1149    root_auth: Option<RootAuthWriterConfig<'_>>,
1150) -> Result<TimedWriterPlan, ArchiveWriteError> {
1151    let mut next_block_index = 0u64;
1152    let payload_started = Instant::now();
1153    let payload = plan_payload_stream(files, options, dictionary, &mut next_block_index)?;
1154    let plan_payload = payload_started.elapsed();
1155    let metadata_started = Instant::now();
1156    let plan = build_writer_plan_from_payload(
1157        payload,
1158        next_block_index,
1159        master_key,
1160        options,
1161        dictionary,
1162        kdf_params,
1163        archive_uuid,
1164        session_id,
1165        root_auth,
1166    )?;
1167    Ok(TimedWriterPlan {
1168        plan,
1169        timings: WriterTimings {
1170            plan_payload,
1171            plan_metadata: metadata_started.elapsed(),
1172            ..WriterTimings::default()
1173        },
1174    })
1175}
1176
1177#[allow(clippy::too_many_arguments)]
1178fn build_writer_plan_from_payload(
1179    payload: PayloadPlanning,
1180    mut next_block_index: u64,
1181    master_key: &MasterKey,
1182    mut options: WriterOptions,
1183    dictionary: Option<&[u8]>,
1184    kdf_params: &KdfParams,
1185    archive_uuid: [u8; 16],
1186    session_id: [u8; 16],
1187    root_auth: Option<RootAuthWriterConfig<'_>>,
1188) -> Result<WriterPlan, ArchiveWriteError> {
1189    let subkeys = writer_subkeys(master_key, options.aead_algo, &archive_uuid, &session_id)?;
1190    let (shard_file_rows, planned_index_shards) = if payload.tar_members.is_empty() {
1191        (Vec::new(), Vec::new())
1192    } else {
1193        let rows = sorted_file_rows(&payload.tar_members);
1194        let shard_file_rows = partition_file_rows(rows)?;
1195        let planned_index_shards = build_index_shard_plaintexts(
1196            &shard_file_rows,
1197            &payload.frames,
1198            &payload.payload_objects,
1199            options,
1200        )?;
1201        (shard_file_rows, planned_index_shards)
1202    };
1203
1204    let mut shard_entries = Vec::with_capacity(planned_index_shards.len());
1205    let mut index_shard_objects = Vec::with_capacity(planned_index_shards.len());
1206    for planned in planned_index_shards {
1207        let compressed =
1208            compress_zstd_frame_with_jobs(&planned.plaintext, options.zstd_level, options.jobs)?;
1209        let object_plan = plan_encrypted_object(
1210            compressed.len(),
1211            options.index_fec_data_shards,
1212            options.index_fec_parity_shards,
1213            options,
1214        )?;
1215        let extent = ObjectExtent::new(next_block_index, object_plan)?;
1216        next_block_index = extent.next_block_index()?;
1217        shard_entries.push(ShardEntry {
1218            shard_index: planned.shard_index,
1219            first_block_index: extent.first_block_index,
1220            data_block_count: extent.data_block_count,
1221            parity_block_count: extent.parity_block_count,
1222            encrypted_size: extent.encrypted_size,
1223            decompressed_size: u32_len(planned.plaintext.len(), "IndexShard")?,
1224            file_count: planned.file_count,
1225            first_path_hash: planned.first_path_hash,
1226            last_path_hash: planned.last_path_hash,
1227        });
1228        index_shard_objects.push(PlannedIndexShardObject {
1229            shard_index: planned.shard_index,
1230            compressed,
1231            extent,
1232        });
1233    }
1234
1235    let compressed_dictionary = dictionary
1236        .map(|dictionary| {
1237            compress_zstd_frame_with_jobs(dictionary, options.zstd_level, options.jobs)
1238        })
1239        .transpose()?;
1240    let dictionary_decompressed_size = dictionary
1241        .map(|dictionary| u32_len(dictionary.len(), "dictionary"))
1242        .transpose()?;
1243    let dictionary_plan = compressed_dictionary
1244        .as_ref()
1245        .map(|compressed| {
1246            plan_metadata_object_without_class(
1247                compressed.len(),
1248                options,
1249                MetadataObjectKind::Dictionary,
1250            )
1251        })
1252        .transpose()?;
1253    let dictionary_extent = dictionary_plan
1254        .map(|plan| ObjectExtent::new(next_block_index, plan))
1255        .transpose()?;
1256    let next_after_dictionary = if let Some(extent) = dictionary_extent {
1257        extent.next_block_index()?
1258    } else {
1259        next_block_index
1260    };
1261
1262    let planned_directory_hint_shards = if should_emit_directory_hints(payload.tar_members.len()) {
1263        build_directory_hint_plaintexts(&shard_file_rows, options)?
1264    } else {
1265        Vec::new()
1266    };
1267    let mut directory_hint_entries = Vec::with_capacity(planned_directory_hint_shards.len());
1268    let mut directory_hint_objects = Vec::with_capacity(planned_directory_hint_shards.len());
1269    let mut planned_next_block_index = next_after_dictionary;
1270    for planned in planned_directory_hint_shards {
1271        let compressed =
1272            compress_zstd_frame_with_jobs(&planned.plaintext, options.zstd_level, options.jobs)?;
1273        let object_plan = plan_encrypted_object(
1274            compressed.len(),
1275            options.index_fec_data_shards,
1276            options.index_fec_parity_shards,
1277            options,
1278        )?;
1279        let extent = ObjectExtent::new(planned_next_block_index, object_plan)?;
1280        planned_next_block_index = extent.next_block_index()?;
1281        directory_hint_entries.push(DirectoryHintShardEntry {
1282            hint_shard_index: planned.hint_shard_index,
1283            first_dir_hash: planned.first_dir_hash,
1284            last_dir_hash: planned.last_dir_hash,
1285            first_block_index: extent.first_block_index,
1286            data_block_count: extent.data_block_count,
1287            parity_block_count: extent.parity_block_count,
1288            encrypted_size: extent.encrypted_size,
1289            decompressed_size: u32_len(planned.plaintext.len(), "DirectoryHintTable")?,
1290            entry_count: planned.entry_count,
1291        });
1292        directory_hint_objects.push(PlannedDirectoryHintObject {
1293            hint_shard_index: planned.hint_shard_index,
1294            compressed,
1295            extent,
1296        });
1297    }
1298
1299    let dictionary_extent = dictionary_extent.zip(dictionary_decompressed_size);
1300    let index_root_plaintext = build_index_root_plaintext(IndexRootPlaintextInput {
1301        shard_entries: &shard_entries,
1302        frame_count: payload.frames.len() as u64,
1303        envelope_count: payload.payload_objects.len() as u64,
1304        file_count: payload.tar_members.len() as u64,
1305        payload_block_count: payload.payload_block_count,
1306        tar_total_size: payload.tar_total_size,
1307        content_sha256: payload.content_sha256,
1308        directory_hint_entries: &directory_hint_entries,
1309        dictionary_extent,
1310    });
1311    let compressed_index_root =
1312        compress_zstd_frame_with_jobs(&index_root_plaintext, options.zstd_level, options.jobs)?;
1313    let metadata_class = plan_index_root_metadata_class(
1314        options,
1315        compressed_index_root.len(),
1316        compressed_dictionary.as_ref().map(Vec::len),
1317    )?;
1318    options = metadata_class.options;
1319    let crypto_header = build_crypto_header(
1320        options,
1321        dictionary.is_some(),
1322        &subkeys,
1323        &archive_uuid,
1324        &session_id,
1325        kdf_params,
1326    )?;
1327    let index_root_extent = ObjectExtent::new(planned_next_block_index, metadata_class.index_root)?;
1328    let total_block_count = index_root_extent.next_block_index()?;
1329    let root_auth_footer_length = root_auth
1330        .map(|config| {
1331            root_auth_footer_wire_length(
1332                config.signer_identity.len(),
1333                config.authenticator_value_length as usize,
1334            )
1335        })
1336        .transpose()?;
1337
1338    Ok(WriterPlan {
1339        options,
1340        archive_uuid,
1341        session_id,
1342        crypto_header,
1343        tar_members: payload.tar_members,
1344        frames: payload.frames,
1345        payload_objects: payload.payload_objects,
1346        index_root_plaintext,
1347        compressed_index_root,
1348        index_root_extent,
1349        index_shard_objects,
1350        shard_entries,
1351        compressed_dictionary,
1352        dictionary_extent,
1353        directory_hint_objects,
1354        directory_hint_entries,
1355        root_auth_footer_length,
1356        total_block_count,
1357    })
1358}
1359
1360fn plan_payload_stream<S: RegularFileSource>(
1361    files: &[S],
1362    options: WriterOptions,
1363    dictionary: Option<&[u8]>,
1364    next_block_index: &mut u64,
1365) -> Result<PayloadPlanning, ArchiveWriteError> {
1366    let mut tar_members = Vec::with_capacity(files.len());
1367    let mut frames = Vec::new();
1368    let mut payload_objects = Vec::new();
1369    let mut tar_total_size = 0u64;
1370    let mut hasher = Sha256::new();
1371    let mut payload_block_count = 0u64;
1372    let mut next_frame_index = 0u64;
1373    let mut envelope = PayloadEnvelopeBuilder {
1374        envelope_index: 0,
1375        plaintext: Vec::new(),
1376    };
1377
1378    for (member_index, file) in files.iter().enumerate() {
1379        let path = normalize_lookup_file_path(file.archive_path(), options.max_path_length)?;
1380        let prefix = build_regular_file_member_prefix(
1381            &path,
1382            file.file_data_size(),
1383            file.mode(),
1384            file.mtime(),
1385        )?;
1386        let member_start = tar_total_size;
1387        let member_group_size = checked_u64_add(
1388            prefix.len() as u64,
1389            checked_u64_add(
1390                file.file_data_size(),
1391                padding_to_512_u64(file.file_data_size()),
1392                "tar member",
1393            )?,
1394            "tar member",
1395        )?;
1396        tar_members.push(TarMember {
1397            path,
1398            tar_member_group_start: member_start,
1399            tar_member_group_size: member_group_size,
1400            file_data_size: file.file_data_size(),
1401            mode: file.mode(),
1402            mtime: file.mtime(),
1403        });
1404        let mut reader = StreamingMemberReader::new(file.open()?, prefix, file.file_data_size());
1405        let mut member_offset = 0u64;
1406        while member_offset < member_group_size {
1407            let remaining = member_group_size - member_offset;
1408            let max_chunk = remaining.min(options.chunk_size as u64);
1409            let mut chunk = vec![0u8; to_usize_writer(max_chunk, "payload chunk")?];
1410            reader
1411                .read_exact(&mut chunk)
1412                .map_err(ArchiveWriteError::Io)?;
1413            let mut chunk_len = chunk.len();
1414            let frame = loop {
1415                let candidate = &chunk[..chunk_len];
1416                let frame = if let Some(dictionary) = dictionary {
1417                    compress_zstd_frame_with_dictionary_and_jobs(
1418                        candidate,
1419                        options.zstd_level,
1420                        dictionary,
1421                        options.jobs,
1422                    )?
1423                } else {
1424                    compress_zstd_frame_with_jobs(candidate, options.zstd_level, options.jobs)?
1425                };
1426                if payload_object_can_fit(frame.len(), options)? {
1427                    break frame;
1428                }
1429                if chunk_len == 1 {
1430                    return Err(FormatError::WriterUnsupported(
1431                        "single-byte payload frame exceeds envelope object limits",
1432                    )
1433                    .into());
1434                }
1435                chunk_len = (chunk_len / 2).max(1);
1436            };
1437            if chunk_len < chunk.len() {
1438                reader.push_back(chunk[chunk_len..].to_vec());
1439            }
1440            let chunk = &chunk[..chunk_len];
1441            hasher.update(chunk);
1442            append_payload_frame_to_plan(
1443                PayloadFramePlanState {
1444                    envelope: &mut envelope,
1445                    payload_objects: &mut payload_objects,
1446                    payload_block_count: &mut payload_block_count,
1447                    next_block_index,
1448                    frames: &mut frames,
1449                    next_frame_index: &mut next_frame_index,
1450                    options,
1451                },
1452                PayloadFramePlanInput {
1453                    frame: &frame,
1454                    decompressed_size: chunk_len,
1455                    member_index,
1456                    member_start,
1457                    member_offset,
1458                    member_group_size,
1459                },
1460            )?;
1461            member_offset = checked_u64_add(member_offset, chunk_len as u64, "payload chunk")?;
1462            tar_total_size = checked_u64_add(tar_total_size, chunk_len as u64, "tar stream")?;
1463        }
1464    }
1465
1466    if !envelope.plaintext.is_empty() {
1467        flush_payload_envelope_plan(
1468            &mut envelope,
1469            &mut payload_objects,
1470            &mut payload_block_count,
1471            next_block_index,
1472            options,
1473        )?;
1474    }
1475    let digest = hasher.finalize();
1476    let mut content_sha256 = [0u8; 32];
1477    content_sha256.copy_from_slice(&digest);
1478    Ok(PayloadPlanning {
1479        tar_members,
1480        frames,
1481        payload_objects,
1482        payload_block_count,
1483        tar_total_size,
1484        content_sha256,
1485    })
1486}
1487
1488#[allow(clippy::too_many_arguments)]
1489pub(crate) fn write_single_pass_archive_to_sink<O, F>(
1490    master_key: &MasterKey,
1491    options: WriterOptions,
1492    kdf_params: &KdfParams,
1493    root_auth: Option<RootAuthWriterConfig<'_>>,
1494    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
1495    sink: &mut O,
1496    drive_members: F,
1497) -> Result<WrittenArchiveSummary, ArchiveWriteError>
1498where
1499    O: ArchiveWriteSink,
1500    F: FnOnce(&mut StreamingArchiveWriter<'_, O>) -> Result<(), ArchiveWriteError>,
1501{
1502    let total_started = Instant::now();
1503    validate_single_pass_writer_options(options)?;
1504    if let Some(root_auth) = root_auth {
1505        validate_root_auth_writer_config(root_auth)?;
1506    }
1507    let options = plan_single_pass_writer_options(options)?;
1508    let archive_uuid = options
1509        .archive_uuid
1510        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1511    let session_id = options
1512        .session_id
1513        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1514    let subkeys = writer_subkeys(master_key, options.aead_algo, &archive_uuid, &session_id)?;
1515    let crypto_header = build_crypto_header(
1516        options,
1517        false,
1518        &subkeys,
1519        &archive_uuid,
1520        &session_id,
1521        kdf_params,
1522    )?;
1523    let emission_state =
1524        begin_writer_emission_state(sink, options, &crypto_header, archive_uuid, session_id)?;
1525
1526    let mut writer = StreamingArchiveWriter {
1527        sink,
1528        options,
1529        archive_uuid,
1530        session_id,
1531        crypto_header,
1532        subkeys,
1533        tar_members: Vec::new(),
1534        frames: Vec::new(),
1535        payload_objects: Vec::new(),
1536        payload_block_count: 0,
1537        tar_total_size: 0,
1538        hasher: Sha256::new(),
1539        next_frame_index: 0,
1540        envelope: PayloadEnvelopeBuilder {
1541            envelope_index: 0,
1542            plaintext: Vec::new(),
1543        },
1544        emission_state,
1545    };
1546    let emit_payload_started = Instant::now();
1547    drive_members(&mut writer)?;
1548    let emit_payload = emit_payload_started.elapsed();
1549    let mut summary = writer.finish(master_key, kdf_params, root_auth, authenticator)?;
1550    summary.timings.emit_payload += emit_payload;
1551    summary.timings.total = total_started.elapsed();
1552    Ok(summary)
1553}
1554
1555struct OrderedFrameJob {
1556    frame_index: u64,
1557    member_index: usize,
1558    member_start: u64,
1559    member_offset: u64,
1560    member_group_size: u64,
1561    plaintext: Vec<u8>,
1562}
1563
1564struct OrderedFrameResult {
1565    frame_index: u64,
1566    member_index: usize,
1567    member_start: u64,
1568    member_offset: u64,
1569    member_group_size: u64,
1570    decompressed_size: usize,
1571    frame: Vec<u8>,
1572}
1573
1574struct OrderedEnvelopeJob {
1575    envelope_index: u64,
1576    plaintext: Vec<u8>,
1577    extent: ObjectExtent,
1578}
1579
1580struct OrderedEnvelopeResult {
1581    envelope_index: u64,
1582    records: Vec<BlockRecord>,
1583}
1584
1585struct OrderedParallelState {
1586    tar_members: Vec<TarMember>,
1587    frames: Vec<PayloadFrame>,
1588    payload_objects: Vec<PayloadObject>,
1589    payload_block_count: u64,
1590    tar_total_size: u64,
1591    hasher: Sha256,
1592    next_frame_job_index: u64,
1593    next_frame_result_index: u64,
1594    next_frame_metadata_index: u64,
1595    frame_buffer: std::collections::BTreeMap<u64, OrderedFrameResult>,
1596    envelope: PayloadEnvelopeBuilder,
1597    next_payload_block_index: u64,
1598    next_envelope_result_index: u64,
1599    envelope_buffer: std::collections::BTreeMap<u64, OrderedEnvelopeResult>,
1600}
1601
1602impl OrderedParallelState {
1603    fn new(file_count: usize) -> Self {
1604        Self {
1605            tar_members: Vec::with_capacity(file_count),
1606            frames: Vec::new(),
1607            payload_objects: Vec::new(),
1608            payload_block_count: 0,
1609            tar_total_size: 0,
1610            hasher: Sha256::new(),
1611            next_frame_job_index: 0,
1612            next_frame_result_index: 0,
1613            next_frame_metadata_index: 0,
1614            frame_buffer: std::collections::BTreeMap::new(),
1615            envelope: PayloadEnvelopeBuilder {
1616                envelope_index: 0,
1617                plaintext: Vec::new(),
1618            },
1619            next_payload_block_index: 0,
1620            next_envelope_result_index: 0,
1621            envelope_buffer: std::collections::BTreeMap::new(),
1622        }
1623    }
1624}
1625
1626#[allow(clippy::too_many_arguments)]
1627fn write_ordered_parallel_archive_to_sink<S, O>(
1628    files: &[S],
1629    master_key: &MasterKey,
1630    options: WriterOptions,
1631    kdf_params: &KdfParams,
1632    root_auth: Option<RootAuthWriterConfig<'_>>,
1633    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
1634    sink: &mut O,
1635) -> Result<WrittenArchiveSummary, ArchiveWriteError>
1636where
1637    S: RegularFileSource,
1638    O: ArchiveWriteSink,
1639{
1640    let total_started = Instant::now();
1641    validate_single_pass_writer_options(options)?;
1642    if let Some(root_auth) = root_auth {
1643        validate_root_auth_writer_config(root_auth)?;
1644    }
1645    let options = plan_single_pass_writer_options(options)?;
1646    let archive_uuid = options
1647        .archive_uuid
1648        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1649    let session_id = options
1650        .session_id
1651        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1652    let subkeys = writer_subkeys(master_key, options.aead_algo, &archive_uuid, &session_id)?;
1653    let crypto_header = build_crypto_header(
1654        options,
1655        false,
1656        &subkeys,
1657        &archive_uuid,
1658        &session_id,
1659        kdf_params,
1660    )?;
1661    let mut emission_state =
1662        begin_writer_emission_state(sink, options, &crypto_header, archive_uuid, session_id)?;
1663
1664    let emit_payload_started = Instant::now();
1665    let mut ordered = OrderedParallelState::new(files.len());
1666    let worker_count = options.jobs.max(1);
1667    let frame_job_buffer = worker_count.saturating_mul(4).max(1);
1668    let envelope_job_buffer = worker_count.saturating_mul(2).max(1);
1669    let subkeys_for_workers = std::sync::Arc::new(subkeys.clone());
1670
1671    std::thread::scope(|scope| -> Result<(), ArchiveWriteError> {
1672        let (frame_job_tx, frame_job_rx) =
1673            std::sync::mpsc::sync_channel::<OrderedFrameJob>(frame_job_buffer);
1674        let (frame_result_tx, frame_result_rx) =
1675            std::sync::mpsc::channel::<Result<OrderedFrameResult, ArchiveWriteError>>();
1676        let frame_job_rx = std::sync::Arc::new(std::sync::Mutex::new(frame_job_rx));
1677
1678        let (envelope_job_tx, envelope_job_rx) =
1679            std::sync::mpsc::sync_channel::<OrderedEnvelopeJob>(envelope_job_buffer);
1680        let (envelope_result_tx, envelope_result_rx) =
1681            std::sync::mpsc::channel::<Result<OrderedEnvelopeResult, ArchiveWriteError>>();
1682        let envelope_job_rx = std::sync::Arc::new(std::sync::Mutex::new(envelope_job_rx));
1683
1684        let frame_handles = (0..worker_count)
1685            .map(|_| {
1686                let frame_job_rx = std::sync::Arc::clone(&frame_job_rx);
1687                let frame_result_tx = frame_result_tx.clone();
1688                scope.spawn(move || loop {
1689                    let job = {
1690                        let receiver = frame_job_rx
1691                            .lock()
1692                            .expect("ordered frame receiver poisoned");
1693                        receiver.recv()
1694                    };
1695                    let Ok(job) = job else {
1696                        break;
1697                    };
1698                    let is_error = match build_ordered_frame_result(job, options) {
1699                        Ok(result) => frame_result_tx.send(Ok(result)).is_err(),
1700                        Err(error) => {
1701                            let _ = frame_result_tx.send(Err(error));
1702                            true
1703                        }
1704                    };
1705                    if is_error {
1706                        break;
1707                    }
1708                })
1709            })
1710            .collect::<Vec<_>>();
1711        drop(frame_result_tx);
1712
1713        let envelope_handles = (0..worker_count)
1714            .map(|_| {
1715                let envelope_job_rx = std::sync::Arc::clone(&envelope_job_rx);
1716                let envelope_result_tx = envelope_result_tx.clone();
1717                let subkeys = std::sync::Arc::clone(&subkeys_for_workers);
1718                scope.spawn(move || loop {
1719                    let job = {
1720                        let receiver = envelope_job_rx
1721                            .lock()
1722                            .expect("ordered envelope receiver poisoned");
1723                        receiver.recv()
1724                    };
1725                    let Ok(job) = job else {
1726                        break;
1727                    };
1728                    let is_error = match build_ordered_envelope_result(
1729                        job,
1730                        &subkeys,
1731                        options,
1732                        archive_uuid,
1733                        session_id,
1734                    ) {
1735                        Ok(result) => envelope_result_tx.send(Ok(result)).is_err(),
1736                        Err(error) => {
1737                            let _ = envelope_result_tx.send(Err(error));
1738                            true
1739                        }
1740                    };
1741                    if is_error {
1742                        break;
1743                    }
1744                })
1745            })
1746            .collect::<Vec<_>>();
1747        drop(envelope_result_tx);
1748
1749        for (member_index, file) in files.iter().enumerate() {
1750            let path = normalize_lookup_file_path(file.archive_path(), options.max_path_length)?;
1751            let prefix = build_regular_file_member_prefix(
1752                &path,
1753                file.file_data_size(),
1754                file.mode(),
1755                file.mtime(),
1756            )?;
1757            let member_start = ordered.tar_total_size;
1758            let member_group_size = checked_u64_add(
1759                prefix.len() as u64,
1760                checked_u64_add(
1761                    file.file_data_size(),
1762                    padding_to_512_u64(file.file_data_size()),
1763                    "tar member",
1764                )?,
1765                "tar member",
1766            )?;
1767            ordered.tar_members.push(TarMember {
1768                path,
1769                tar_member_group_start: member_start,
1770                tar_member_group_size: member_group_size,
1771                file_data_size: file.file_data_size(),
1772                mode: file.mode(),
1773                mtime: file.mtime(),
1774            });
1775            let mut reader =
1776                StreamingMemberReader::new(file.open()?, prefix, file.file_data_size());
1777            let mut member_offset = 0u64;
1778            while member_offset < member_group_size {
1779                let remaining = member_group_size - member_offset;
1780                let read_len = remaining.min(options.chunk_size as u64);
1781                let mut plaintext = vec![0u8; to_usize_writer(read_len, "payload chunk")?];
1782                reader
1783                    .read_exact(&mut plaintext)
1784                    .map_err(ArchiveWriteError::Io)?;
1785                ordered.hasher.update(&plaintext);
1786                let frame_index = ordered.next_frame_job_index;
1787                ordered.next_frame_job_index =
1788                    checked_u64_add(ordered.next_frame_job_index, 1, "PayloadFrame.frame_index")?;
1789                send_ordered_frame_job(
1790                    OrderedFrameJob {
1791                        frame_index,
1792                        member_index,
1793                        member_start,
1794                        member_offset,
1795                        member_group_size,
1796                        plaintext,
1797                    },
1798                    &frame_job_tx,
1799                    &frame_result_rx,
1800                    &envelope_job_tx,
1801                    &envelope_result_rx,
1802                    &mut ordered,
1803                    sink,
1804                    options,
1805                    &mut emission_state,
1806                )?;
1807                member_offset = checked_u64_add(member_offset, read_len, "payload chunk")?;
1808                ordered.tar_total_size =
1809                    checked_u64_add(ordered.tar_total_size, read_len, "tar stream")?;
1810            }
1811        }
1812        drop(frame_job_tx);
1813
1814        while ordered.next_frame_result_index < ordered.next_frame_job_index {
1815            receive_ordered_frame_result(
1816                &frame_result_rx,
1817                &envelope_job_tx,
1818                &envelope_result_rx,
1819                &mut ordered,
1820                sink,
1821                options,
1822                &mut emission_state,
1823                true,
1824            )?;
1825        }
1826        flush_ordered_parallel_envelope(
1827            &envelope_job_tx,
1828            &envelope_result_rx,
1829            &mut ordered,
1830            sink,
1831            options,
1832            &mut emission_state,
1833        )?;
1834        drop(envelope_job_tx);
1835        while ordered.next_envelope_result_index < ordered.envelope.envelope_index {
1836            receive_ordered_envelope_result(
1837                &envelope_result_rx,
1838                &mut ordered,
1839                sink,
1840                options,
1841                &mut emission_state,
1842                true,
1843            )?;
1844        }
1845
1846        for handle in frame_handles {
1847            handle
1848                .join()
1849                .map_err(|_| FormatError::WriterInvariant("ordered frame worker panicked"))?;
1850        }
1851        for handle in envelope_handles {
1852            handle
1853                .join()
1854                .map_err(|_| FormatError::WriterInvariant("ordered envelope worker panicked"))?;
1855        }
1856        Ok(())
1857    })?;
1858    let emit_payload = emit_payload_started.elapsed();
1859
1860    emission_state.next_block_index = ordered.next_payload_block_index;
1861    let digest = ordered.hasher.finalize();
1862    let mut content_sha256 = [0u8; 32];
1863    content_sha256.copy_from_slice(&digest);
1864    let payload = PayloadPlanning {
1865        tar_members: ordered.tar_members,
1866        frames: ordered.frames,
1867        payload_objects: ordered.payload_objects,
1868        payload_block_count: ordered.payload_block_count,
1869        tar_total_size: ordered.tar_total_size,
1870        content_sha256,
1871    };
1872    let plan = build_writer_plan_from_payload(
1873        payload,
1874        emission_state.next_block_index,
1875        master_key,
1876        options,
1877        None,
1878        kdf_params,
1879        archive_uuid,
1880        session_id,
1881        root_auth,
1882    )?;
1883    if plan.options != options || plan.crypto_header != crypto_header {
1884        return Err(FormatError::WriterUnsupported(
1885            "ordered parallel metadata exceeded the predeclared header class",
1886        )
1887        .into());
1888    }
1889    let mut summary = emit_writer_plan_suffix(
1890        &subkeys,
1891        root_auth,
1892        authenticator,
1893        plan,
1894        sink,
1895        emission_state,
1896    )?;
1897    summary.timings.emit_payload += emit_payload;
1898    summary.timings.total = total_started.elapsed();
1899    Ok(summary)
1900}
1901
1902#[allow(clippy::too_many_arguments)]
1903pub(crate) fn write_ordered_parallel_stream_archive_to_sink<O, F>(
1904    master_key: &MasterKey,
1905    options: WriterOptions,
1906    kdf_params: &KdfParams,
1907    root_auth: Option<RootAuthWriterConfig<'_>>,
1908    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
1909    sink: &mut O,
1910    drive_members: F,
1911) -> Result<WrittenArchiveSummary, ArchiveWriteError>
1912where
1913    O: ArchiveWriteSink,
1914    F: FnOnce(&mut OrderedParallelArchiveWriter<'_, O>) -> Result<(), ArchiveWriteError>,
1915{
1916    let total_started = Instant::now();
1917    validate_single_pass_writer_options(options)?;
1918    if let Some(root_auth) = root_auth {
1919        validate_root_auth_writer_config(root_auth)?;
1920    }
1921    let options = plan_single_pass_writer_options(options)?;
1922    let archive_uuid = options
1923        .archive_uuid
1924        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1925    let session_id = options
1926        .session_id
1927        .unwrap_or_else(|| *Uuid::new_v4().as_bytes());
1928    let subkeys = writer_subkeys(master_key, options.aead_algo, &archive_uuid, &session_id)?;
1929    let crypto_header = build_crypto_header(
1930        options,
1931        false,
1932        &subkeys,
1933        &archive_uuid,
1934        &session_id,
1935        kdf_params,
1936    )?;
1937    let mut emission_state =
1938        begin_writer_emission_state(sink, options, &crypto_header, archive_uuid, session_id)?;
1939
1940    let emit_payload_started = Instant::now();
1941    let mut ordered = OrderedParallelState::new(0);
1942    let worker_count = options.jobs.max(1);
1943    let frame_job_buffer = worker_count.saturating_mul(4).max(1);
1944    let envelope_job_buffer = worker_count.saturating_mul(2).max(1);
1945    let subkeys_for_workers = std::sync::Arc::new(subkeys.clone());
1946
1947    std::thread::scope(|scope| -> Result<(), ArchiveWriteError> {
1948        let (frame_job_tx, frame_job_rx) =
1949            std::sync::mpsc::sync_channel::<OrderedFrameJob>(frame_job_buffer);
1950        let (frame_result_tx, frame_result_rx) =
1951            std::sync::mpsc::channel::<Result<OrderedFrameResult, ArchiveWriteError>>();
1952        let frame_job_rx = std::sync::Arc::new(std::sync::Mutex::new(frame_job_rx));
1953
1954        let (envelope_job_tx, envelope_job_rx) =
1955            std::sync::mpsc::sync_channel::<OrderedEnvelopeJob>(envelope_job_buffer);
1956        let (envelope_result_tx, envelope_result_rx) =
1957            std::sync::mpsc::channel::<Result<OrderedEnvelopeResult, ArchiveWriteError>>();
1958        let envelope_job_rx = std::sync::Arc::new(std::sync::Mutex::new(envelope_job_rx));
1959
1960        let frame_handles = (0..worker_count)
1961            .map(|_| {
1962                let frame_job_rx = std::sync::Arc::clone(&frame_job_rx);
1963                let frame_result_tx = frame_result_tx.clone();
1964                scope.spawn(move || loop {
1965                    let job = {
1966                        let receiver = frame_job_rx
1967                            .lock()
1968                            .expect("ordered frame receiver poisoned");
1969                        receiver.recv()
1970                    };
1971                    let Ok(job) = job else {
1972                        break;
1973                    };
1974                    let is_error = match build_ordered_frame_result(job, options) {
1975                        Ok(result) => frame_result_tx.send(Ok(result)).is_err(),
1976                        Err(error) => {
1977                            let _ = frame_result_tx.send(Err(error));
1978                            true
1979                        }
1980                    };
1981                    if is_error {
1982                        break;
1983                    }
1984                })
1985            })
1986            .collect::<Vec<_>>();
1987        drop(frame_result_tx);
1988
1989        let envelope_handles = (0..worker_count)
1990            .map(|_| {
1991                let envelope_job_rx = std::sync::Arc::clone(&envelope_job_rx);
1992                let envelope_result_tx = envelope_result_tx.clone();
1993                let subkeys = std::sync::Arc::clone(&subkeys_for_workers);
1994                scope.spawn(move || loop {
1995                    let job = {
1996                        let receiver = envelope_job_rx
1997                            .lock()
1998                            .expect("ordered envelope receiver poisoned");
1999                        receiver.recv()
2000                    };
2001                    let Ok(job) = job else {
2002                        break;
2003                    };
2004                    let is_error = match build_ordered_envelope_result(
2005                        job,
2006                        &subkeys,
2007                        options,
2008                        archive_uuid,
2009                        session_id,
2010                    ) {
2011                        Ok(result) => envelope_result_tx.send(Ok(result)).is_err(),
2012                        Err(error) => {
2013                            let _ = envelope_result_tx.send(Err(error));
2014                            true
2015                        }
2016                    };
2017                    if is_error {
2018                        break;
2019                    }
2020                })
2021            })
2022            .collect::<Vec<_>>();
2023        drop(envelope_result_tx);
2024
2025        {
2026            let mut writer = OrderedParallelArchiveWriter {
2027                frame_job_tx: &frame_job_tx,
2028                frame_result_rx: &frame_result_rx,
2029                envelope_job_tx: &envelope_job_tx,
2030                envelope_result_rx: &envelope_result_rx,
2031                ordered: &mut ordered,
2032                sink,
2033                options,
2034                emission_state: &mut emission_state,
2035            };
2036            drive_members(&mut writer)?;
2037        }
2038
2039        drop(frame_job_tx);
2040        while ordered.next_frame_result_index < ordered.next_frame_job_index {
2041            receive_ordered_frame_result(
2042                &frame_result_rx,
2043                &envelope_job_tx,
2044                &envelope_result_rx,
2045                &mut ordered,
2046                sink,
2047                options,
2048                &mut emission_state,
2049                true,
2050            )?;
2051        }
2052        flush_ordered_parallel_envelope(
2053            &envelope_job_tx,
2054            &envelope_result_rx,
2055            &mut ordered,
2056            sink,
2057            options,
2058            &mut emission_state,
2059        )?;
2060        drop(envelope_job_tx);
2061        while ordered.next_envelope_result_index < ordered.envelope.envelope_index {
2062            receive_ordered_envelope_result(
2063                &envelope_result_rx,
2064                &mut ordered,
2065                sink,
2066                options,
2067                &mut emission_state,
2068                true,
2069            )?;
2070        }
2071
2072        for handle in frame_handles {
2073            handle
2074                .join()
2075                .map_err(|_| FormatError::WriterInvariant("ordered frame worker panicked"))?;
2076        }
2077        for handle in envelope_handles {
2078            handle
2079                .join()
2080                .map_err(|_| FormatError::WriterInvariant("ordered envelope worker panicked"))?;
2081        }
2082        Ok(())
2083    })?;
2084    let emit_payload = emit_payload_started.elapsed();
2085
2086    emission_state.next_block_index = ordered.next_payload_block_index;
2087    let digest = ordered.hasher.finalize();
2088    let mut content_sha256 = [0u8; 32];
2089    content_sha256.copy_from_slice(&digest);
2090    let payload = PayloadPlanning {
2091        tar_members: ordered.tar_members,
2092        frames: ordered.frames,
2093        payload_objects: ordered.payload_objects,
2094        payload_block_count: ordered.payload_block_count,
2095        tar_total_size: ordered.tar_total_size,
2096        content_sha256,
2097    };
2098    let plan = build_writer_plan_from_payload(
2099        payload,
2100        emission_state.next_block_index,
2101        master_key,
2102        options,
2103        None,
2104        kdf_params,
2105        archive_uuid,
2106        session_id,
2107        root_auth,
2108    )?;
2109    if plan.options != options || plan.crypto_header != crypto_header {
2110        return Err(FormatError::WriterUnsupported(
2111            "ordered parallel metadata exceeded the predeclared header class",
2112        )
2113        .into());
2114    }
2115    let mut summary = emit_writer_plan_suffix(
2116        &subkeys,
2117        root_auth,
2118        authenticator,
2119        plan,
2120        sink,
2121        emission_state,
2122    )?;
2123    summary.timings.emit_payload += emit_payload;
2124    summary.timings.total = total_started.elapsed();
2125    Ok(summary)
2126}
2127
2128pub(crate) struct OrderedParallelArchiveWriter<'a, O: ArchiveWriteSink> {
2129    frame_job_tx: &'a std::sync::mpsc::SyncSender<OrderedFrameJob>,
2130    frame_result_rx: &'a std::sync::mpsc::Receiver<Result<OrderedFrameResult, ArchiveWriteError>>,
2131    envelope_job_tx: &'a std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2132    envelope_result_rx:
2133        &'a std::sync::mpsc::Receiver<Result<OrderedEnvelopeResult, ArchiveWriteError>>,
2134    ordered: &'a mut OrderedParallelState,
2135    sink: &'a mut O,
2136    options: WriterOptions,
2137    emission_state: &'a mut WriterEmissionState,
2138}
2139
2140impl<O: ArchiveWriteSink> OrderedParallelArchiveWriter<'_, O> {
2141    pub(crate) fn write_regular_member_from_reader(
2142        &mut self,
2143        member: StreamingRegularMember,
2144        payload: &mut dyn Read,
2145    ) -> Result<(), ArchiveWriteError> {
2146        let path = member.archive_path;
2147        validate_file_path_bytes(&path, self.options.max_path_length)?;
2148        let prefix = build_regular_file_member_prefix(
2149            &path,
2150            member.file_data_size,
2151            member.mode,
2152            member.mtime,
2153        )?;
2154        let member_start = self.ordered.tar_total_size;
2155        let member_group_size = checked_u64_add(
2156            prefix.len() as u64,
2157            checked_u64_add(
2158                member.file_data_size,
2159                padding_to_512_u64(member.file_data_size),
2160                "tar member",
2161            )?,
2162            "tar member",
2163        )?;
2164        let member_index = self.ordered.tar_members.len();
2165        self.ordered.tar_members.push(TarMember {
2166            path,
2167            tar_member_group_start: member_start,
2168            tar_member_group_size: member_group_size,
2169            file_data_size: member.file_data_size,
2170            mode: member.mode,
2171            mtime: member.mtime,
2172        });
2173
2174        let mut reader =
2175            StreamingMemberReader::new(Box::new(payload), prefix, member.file_data_size);
2176        let mut member_offset = 0u64;
2177        while member_offset < member_group_size {
2178            let remaining = member_group_size - member_offset;
2179            let read_len = remaining.min(self.options.chunk_size as u64);
2180            let mut plaintext = vec![0u8; to_usize_writer(read_len, "payload chunk")?];
2181            reader
2182                .read_exact(&mut plaintext)
2183                .map_err(ArchiveWriteError::Io)?;
2184            self.ordered.hasher.update(&plaintext);
2185            let frame_index = self.ordered.next_frame_job_index;
2186            self.ordered.next_frame_job_index = checked_u64_add(
2187                self.ordered.next_frame_job_index,
2188                1,
2189                "PayloadFrame.frame_index",
2190            )?;
2191            send_ordered_frame_job(
2192                OrderedFrameJob {
2193                    frame_index,
2194                    member_index,
2195                    member_start,
2196                    member_offset,
2197                    member_group_size,
2198                    plaintext,
2199                },
2200                self.frame_job_tx,
2201                self.frame_result_rx,
2202                self.envelope_job_tx,
2203                self.envelope_result_rx,
2204                self.ordered,
2205                self.sink,
2206                self.options,
2207                self.emission_state,
2208            )?;
2209            member_offset = checked_u64_add(member_offset, read_len, "payload chunk")?;
2210            self.ordered.tar_total_size =
2211                checked_u64_add(self.ordered.tar_total_size, read_len, "tar stream")?;
2212        }
2213        Ok(())
2214    }
2215}
2216
2217#[allow(clippy::too_many_arguments)]
2218fn send_ordered_frame_job<O: ArchiveWriteSink>(
2219    mut job: OrderedFrameJob,
2220    frame_job_tx: &std::sync::mpsc::SyncSender<OrderedFrameJob>,
2221    frame_result_rx: &std::sync::mpsc::Receiver<Result<OrderedFrameResult, ArchiveWriteError>>,
2222    envelope_job_tx: &std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2223    envelope_result_rx: &std::sync::mpsc::Receiver<
2224        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2225    >,
2226    ordered: &mut OrderedParallelState,
2227    sink: &mut O,
2228    options: WriterOptions,
2229    emission_state: &mut WriterEmissionState,
2230) -> Result<(), ArchiveWriteError> {
2231    loop {
2232        match frame_job_tx.try_send(job) {
2233            Ok(()) => {
2234                drain_ordered_frame_results(
2235                    frame_result_rx,
2236                    envelope_job_tx,
2237                    envelope_result_rx,
2238                    ordered,
2239                    sink,
2240                    options,
2241                    emission_state,
2242                )?;
2243                return Ok(());
2244            }
2245            Err(std::sync::mpsc::TrySendError::Full(returned)) => {
2246                job = returned;
2247                receive_ordered_frame_result(
2248                    frame_result_rx,
2249                    envelope_job_tx,
2250                    envelope_result_rx,
2251                    ordered,
2252                    sink,
2253                    options,
2254                    emission_state,
2255                    true,
2256                )?;
2257            }
2258            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
2259                return Err(FormatError::WriterInvariant("ordered frame worker stopped").into());
2260            }
2261        }
2262    }
2263}
2264
2265#[allow(clippy::too_many_arguments)]
2266fn drain_ordered_frame_results<O: ArchiveWriteSink>(
2267    frame_result_rx: &std::sync::mpsc::Receiver<Result<OrderedFrameResult, ArchiveWriteError>>,
2268    envelope_job_tx: &std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2269    envelope_result_rx: &std::sync::mpsc::Receiver<
2270        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2271    >,
2272    ordered: &mut OrderedParallelState,
2273    sink: &mut O,
2274    options: WriterOptions,
2275    emission_state: &mut WriterEmissionState,
2276) -> Result<(), ArchiveWriteError> {
2277    while receive_ordered_frame_result(
2278        frame_result_rx,
2279        envelope_job_tx,
2280        envelope_result_rx,
2281        ordered,
2282        sink,
2283        options,
2284        emission_state,
2285        false,
2286    )? {}
2287    Ok(())
2288}
2289
2290#[allow(clippy::too_many_arguments)]
2291fn receive_ordered_frame_result<O: ArchiveWriteSink>(
2292    frame_result_rx: &std::sync::mpsc::Receiver<Result<OrderedFrameResult, ArchiveWriteError>>,
2293    envelope_job_tx: &std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2294    envelope_result_rx: &std::sync::mpsc::Receiver<
2295        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2296    >,
2297    ordered: &mut OrderedParallelState,
2298    sink: &mut O,
2299    options: WriterOptions,
2300    emission_state: &mut WriterEmissionState,
2301    wait: bool,
2302) -> Result<bool, ArchiveWriteError> {
2303    let result = if wait {
2304        match frame_result_rx.recv() {
2305            Ok(result) => result?,
2306            Err(_) => {
2307                return Err(FormatError::WriterInvariant("ordered frame worker stopped").into())
2308            }
2309        }
2310    } else {
2311        match frame_result_rx.try_recv() {
2312            Ok(result) => result?,
2313            Err(std::sync::mpsc::TryRecvError::Empty) => return Ok(false),
2314            Err(std::sync::mpsc::TryRecvError::Disconnected) => return Ok(false),
2315        }
2316    };
2317    ordered.frame_buffer.insert(result.frame_index, result);
2318    while let Some(result) = ordered
2319        .frame_buffer
2320        .remove(&ordered.next_frame_result_index)
2321    {
2322        append_ordered_frame_result(
2323            result,
2324            envelope_job_tx,
2325            envelope_result_rx,
2326            ordered,
2327            sink,
2328            options,
2329            emission_state,
2330        )?;
2331        ordered.next_frame_result_index = checked_u64_add(
2332            ordered.next_frame_result_index,
2333            1,
2334            "PayloadFrame.frame_index",
2335        )?;
2336    }
2337    Ok(true)
2338}
2339
2340#[allow(clippy::too_many_arguments)]
2341fn append_ordered_frame_result<O: ArchiveWriteSink>(
2342    result: OrderedFrameResult,
2343    envelope_job_tx: &std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2344    envelope_result_rx: &std::sync::mpsc::Receiver<
2345        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2346    >,
2347    ordered: &mut OrderedParallelState,
2348    sink: &mut O,
2349    options: WriterOptions,
2350    emission_state: &mut WriterEmissionState,
2351) -> Result<(), ArchiveWriteError> {
2352    let next_len = checked_usize_add(
2353        ordered.envelope.plaintext.len(),
2354        result.frame.len(),
2355        "payload",
2356    )?;
2357    if !ordered.envelope.plaintext.is_empty()
2358        && (next_len > options.envelope_target_size as usize
2359            || !payload_object_can_fit(next_len, options)?)
2360    {
2361        flush_ordered_parallel_envelope(
2362            envelope_job_tx,
2363            envelope_result_rx,
2364            ordered,
2365            sink,
2366            options,
2367            emission_state,
2368        )?;
2369    }
2370    if ordered.envelope.plaintext.is_empty()
2371        && !payload_object_can_fit(result.frame.len(), options)?
2372    {
2373        return Err(
2374            FormatError::WriterUnsupported("payload frame exceeds envelope object limits").into(),
2375        );
2376    }
2377    let offset = u32_len(
2378        ordered.envelope.plaintext.len(),
2379        "FrameEntry.offset_in_envelope",
2380    )?;
2381    ordered.envelope.plaintext.extend_from_slice(&result.frame);
2382    let mut flags = 0u32;
2383    if result.member_offset == 0 {
2384        flags |= 0x0000_0001;
2385    }
2386    if checked_u64_add(
2387        result.member_offset,
2388        result.decompressed_size as u64,
2389        "payload chunk",
2390    )? == result.member_group_size
2391    {
2392        flags |= 0x0000_0002;
2393    }
2394    ordered.frames.push(PayloadFrame {
2395        frame_index: ordered.next_frame_metadata_index,
2396        envelope_index: ordered.envelope.envelope_index,
2397        member_index: result.member_index,
2398        offset_in_envelope: offset,
2399        compressed_size: u32_len(result.frame.len(), "FrameEntry.compressed_size")?,
2400        decompressed_size: u32_len(result.decompressed_size, "FrameEntry.decompressed_size")?,
2401        flags,
2402        tar_stream_offset: checked_u64_add(
2403            result.member_start,
2404            result.member_offset,
2405            "PayloadFrame.tar_stream_offset",
2406        )?,
2407    });
2408    ordered.next_frame_metadata_index = checked_u64_add(
2409        ordered.next_frame_metadata_index,
2410        1,
2411        "PayloadFrame.frame_index",
2412    )?;
2413    Ok(())
2414}
2415
2416#[allow(clippy::too_many_arguments)]
2417fn flush_ordered_parallel_envelope<O: ArchiveWriteSink>(
2418    envelope_job_tx: &std::sync::mpsc::SyncSender<OrderedEnvelopeJob>,
2419    envelope_result_rx: &std::sync::mpsc::Receiver<
2420        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2421    >,
2422    ordered: &mut OrderedParallelState,
2423    sink: &mut O,
2424    options: WriterOptions,
2425    emission_state: &mut WriterEmissionState,
2426) -> Result<(), ArchiveWriteError> {
2427    if ordered.envelope.plaintext.is_empty() {
2428        return Ok(());
2429    }
2430    let plaintext_size = u32_len(
2431        ordered.envelope.plaintext.len(),
2432        "EnvelopeEntry.plaintext_size",
2433    )?;
2434    let object_plan = plan_encrypted_object(
2435        ordered.envelope.plaintext.len(),
2436        options.fec_data_shards,
2437        options.fec_parity_shards,
2438        options,
2439    )?;
2440    let extent = ObjectExtent::new(ordered.next_payload_block_index, object_plan)?;
2441    ordered.next_payload_block_index = extent.next_block_index()?;
2442    ordered.payload_block_count = checked_u64_add(
2443        ordered.payload_block_count,
2444        extent.data_block_count as u64,
2445        "payload",
2446    )?;
2447    ordered.payload_objects.push(PayloadObject {
2448        envelope_index: ordered.envelope.envelope_index,
2449        plaintext_size,
2450        object: extent,
2451    });
2452    let mut job = OrderedEnvelopeJob {
2453        envelope_index: ordered.envelope.envelope_index,
2454        plaintext: std::mem::take(&mut ordered.envelope.plaintext),
2455        extent,
2456    };
2457    ordered.envelope.envelope_index =
2458        checked_u64_add(ordered.envelope.envelope_index, 1, "EnvelopeEntry")?;
2459    loop {
2460        match envelope_job_tx.try_send(job) {
2461            Ok(()) => {
2462                drain_ordered_envelope_results(
2463                    envelope_result_rx,
2464                    ordered,
2465                    sink,
2466                    options,
2467                    emission_state,
2468                )?;
2469                return Ok(());
2470            }
2471            Err(std::sync::mpsc::TrySendError::Full(returned)) => {
2472                job = returned;
2473                receive_ordered_envelope_result(
2474                    envelope_result_rx,
2475                    ordered,
2476                    sink,
2477                    options,
2478                    emission_state,
2479                    true,
2480                )?;
2481            }
2482            Err(std::sync::mpsc::TrySendError::Disconnected(_)) => {
2483                return Err(FormatError::WriterInvariant("ordered envelope worker stopped").into());
2484            }
2485        }
2486    }
2487}
2488
2489fn drain_ordered_envelope_results<O: ArchiveWriteSink>(
2490    envelope_result_rx: &std::sync::mpsc::Receiver<
2491        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2492    >,
2493    ordered: &mut OrderedParallelState,
2494    sink: &mut O,
2495    options: WriterOptions,
2496    emission_state: &mut WriterEmissionState,
2497) -> Result<(), ArchiveWriteError> {
2498    while receive_ordered_envelope_result(
2499        envelope_result_rx,
2500        ordered,
2501        sink,
2502        options,
2503        emission_state,
2504        false,
2505    )? {}
2506    Ok(())
2507}
2508
2509fn receive_ordered_envelope_result<O: ArchiveWriteSink>(
2510    envelope_result_rx: &std::sync::mpsc::Receiver<
2511        Result<OrderedEnvelopeResult, ArchiveWriteError>,
2512    >,
2513    ordered: &mut OrderedParallelState,
2514    sink: &mut O,
2515    options: WriterOptions,
2516    emission_state: &mut WriterEmissionState,
2517    wait: bool,
2518) -> Result<bool, ArchiveWriteError> {
2519    let result = if wait {
2520        match envelope_result_rx.recv() {
2521            Ok(result) => result?,
2522            Err(_) => {
2523                return Err(FormatError::WriterInvariant("ordered envelope worker stopped").into());
2524            }
2525        }
2526    } else {
2527        match envelope_result_rx.try_recv() {
2528            Ok(result) => result?,
2529            Err(std::sync::mpsc::TryRecvError::Empty) => return Ok(false),
2530            Err(std::sync::mpsc::TryRecvError::Disconnected) => return Ok(false),
2531        }
2532    };
2533    ordered
2534        .envelope_buffer
2535        .insert(result.envelope_index, result);
2536    while let Some(result) = ordered
2537        .envelope_buffer
2538        .remove(&ordered.next_envelope_result_index)
2539    {
2540        emit_ordered_envelope_result(result, sink, options, emission_state)?;
2541        ordered.next_envelope_result_index =
2542            checked_u64_add(ordered.next_envelope_result_index, 1, "EnvelopeEntry")?;
2543    }
2544    Ok(true)
2545}
2546
2547fn build_ordered_frame_result(
2548    job: OrderedFrameJob,
2549    options: WriterOptions,
2550) -> Result<OrderedFrameResult, ArchiveWriteError> {
2551    let frame = compress_zstd_frame_with_jobs(&job.plaintext, options.zstd_level, 1)?;
2552    Ok(OrderedFrameResult {
2553        frame_index: job.frame_index,
2554        member_index: job.member_index,
2555        member_start: job.member_start,
2556        member_offset: job.member_offset,
2557        member_group_size: job.member_group_size,
2558        decompressed_size: job.plaintext.len(),
2559        frame,
2560    })
2561}
2562
2563fn build_ordered_envelope_result(
2564    job: OrderedEnvelopeJob,
2565    subkeys: &Subkeys,
2566    options: WriterOptions,
2567    archive_uuid: [u8; 16],
2568    session_id: [u8; 16],
2569) -> Result<OrderedEnvelopeResult, ArchiveWriteError> {
2570    let mut local_next_block_index = job.extent.first_block_index;
2571    let object = encrypt_object(
2572        &job.plaintext,
2573        ObjectEncryptionContext {
2574            key: &subkeys.enc_key,
2575            nonce_seed: &subkeys.nonce_seed,
2576            domain: b"envelope",
2577            counter: job.envelope_index,
2578            data_kind: BlockKind::PayloadData,
2579            parity_kind: BlockKind::PayloadParity,
2580            data_shard_max: options.fec_data_shards,
2581            class_parity_shard_max: options.fec_parity_shards,
2582            archive_uuid: &archive_uuid,
2583            session_id: &session_id,
2584        },
2585        &mut local_next_block_index,
2586        options,
2587    )?;
2588    validate_planned_extent(&object, job.extent)?;
2589    Ok(OrderedEnvelopeResult {
2590        envelope_index: job.envelope_index,
2591        records: object.records,
2592    })
2593}
2594
2595fn emit_ordered_envelope_result<O: ArchiveWriteSink>(
2596    result: OrderedEnvelopeResult,
2597    sink: &mut O,
2598    options: WriterOptions,
2599    emission_state: &mut WriterEmissionState,
2600) -> Result<(), ArchiveWriteError> {
2601    for record in &result.records {
2602        emit_block_record(
2603            sink,
2604            options,
2605            &mut emission_state.bytes_written,
2606            &mut emission_state.record_counts,
2607            &mut emission_state.data_leaf_hashes,
2608            record,
2609        )?;
2610    }
2611    Ok(())
2612}
2613
2614fn validate_single_pass_writer_options(options: WriterOptions) -> Result<(), FormatError> {
2615    if options.volume_loss_tolerance != 0 {
2616        return Err(FormatError::WriterUnsupported(
2617            "streaming create cannot tolerate volume loss",
2618        ));
2619    }
2620    if options.target_volume_size.is_some() {
2621        return Err(FormatError::WriterUnsupported(
2622            "streaming create does not support target volume sizing",
2623        ));
2624    }
2625    Ok(())
2626}
2627
2628fn begin_writer_emission_state<O: ArchiveWriteSink>(
2629    sink: &mut O,
2630    options: WriterOptions,
2631    crypto_header: &[u8],
2632    archive_uuid: [u8; 16],
2633    session_id: [u8; 16],
2634) -> Result<WriterEmissionState, ArchiveWriteError> {
2635    let volume_count = usize::try_from(options.stripe_width)
2636        .map_err(|_| FormatError::WriterUnsupported("stripe_width"))?;
2637    sink.begin_archive(volume_count)?;
2638
2639    let mut state = WriterEmissionState {
2640        volume_headers: Vec::with_capacity(volume_count),
2641        bytes_written: vec![0u64; volume_count],
2642        record_counts: vec![0u64; volume_count],
2643        data_leaf_hashes: Vec::new(),
2644        next_block_index: 0,
2645    };
2646
2647    for volume_index in 0..volume_count {
2648        let volume_index_u32 = u32::try_from(volume_index)
2649            .map_err(|_| FormatError::WriterUnsupported("volume_index"))?;
2650        let volume_header = VolumeHeader {
2651            format_version: FORMAT_VERSION,
2652            volume_format_rev: VOLUME_FORMAT_REV,
2653            volume_index: volume_index_u32,
2654            stripe_width: options.stripe_width,
2655            archive_uuid,
2656            session_id,
2657            crypto_header_offset: VOLUME_HEADER_LEN as u32,
2658            crypto_header_length: u32_len(crypto_header.len(), "CryptoHeader")?,
2659            header_crc32c: 0,
2660        };
2661        let volume_header_bytes = volume_header.to_bytes();
2662        sink.write_volume(volume_index, &volume_header_bytes)?;
2663        sink.write_volume(volume_index, crypto_header)?;
2664        state.bytes_written[volume_index] = checked_u64_add(
2665            VOLUME_HEADER_LEN as u64,
2666            crypto_header.len() as u64,
2667            "volume header",
2668        )?;
2669        state.volume_headers.push(volume_header_bytes);
2670    }
2671
2672    Ok(state)
2673}
2674
2675fn plan_single_pass_writer_options(options: WriterOptions) -> Result<WriterOptions, FormatError> {
2676    let mut options = plan_writer_options(options)?;
2677    options.index_root_fec_data_shards = max_single_pass_index_root_data_shards(options)?;
2678    plan_writer_options(options)
2679}
2680
2681fn max_single_pass_index_root_data_shards(options: WriterOptions) -> Result<u16, FormatError> {
2682    let block_size_limit = (u32::MAX as u64 / options.block_size as u64).min(u16::MAX as u64);
2683    let mut low = MIN_INDEX_ROOT_FEC_DATA_SHARDS as u64;
2684    let mut high = block_size_limit;
2685    let mut best = low;
2686    while low <= high {
2687        let mid = low + (high - low) / 2;
2688        match compute_parity(mid, options) {
2689            Ok(parity)
2690                if mid + u64::from(parity) <= READER_MAX_INDEX_ROOT_FEC_CLASS_SHARDS as u64 =>
2691            {
2692                best = mid;
2693                low = mid + 1;
2694            }
2695            _ => {
2696                if mid == 0 {
2697                    break;
2698                }
2699                high = mid - 1;
2700            }
2701        }
2702    }
2703    u16::try_from(best).map_err(|_| FormatError::WriterUnsupported("index_root_fec_data_shards"))
2704}
2705
2706impl<O: ArchiveWriteSink> StreamingArchiveWriter<'_, O> {
2707    pub(crate) fn write_regular_member_from_reader(
2708        &mut self,
2709        member: StreamingRegularMember,
2710        payload: &mut dyn Read,
2711    ) -> Result<(), ArchiveWriteError> {
2712        let path = member.archive_path;
2713        validate_file_path_bytes(&path, self.options.max_path_length)?;
2714        let prefix = build_regular_file_member_prefix(
2715            &path,
2716            member.file_data_size,
2717            member.mode,
2718            member.mtime,
2719        )?;
2720        let member_start = self.tar_total_size;
2721        let member_group_size = checked_u64_add(
2722            prefix.len() as u64,
2723            checked_u64_add(
2724                member.file_data_size,
2725                padding_to_512_u64(member.file_data_size),
2726                "tar member",
2727            )?,
2728            "tar member",
2729        )?;
2730        let member_index = self.tar_members.len();
2731        self.tar_members.push(TarMember {
2732            path,
2733            tar_member_group_start: member_start,
2734            tar_member_group_size: member_group_size,
2735            file_data_size: member.file_data_size,
2736            mode: member.mode,
2737            mtime: member.mtime,
2738        });
2739
2740        let mut reader =
2741            StreamingMemberReader::new(Box::new(payload), prefix, member.file_data_size);
2742        let mut member_offset = 0u64;
2743        while member_offset < member_group_size {
2744            let remaining = member_group_size - member_offset;
2745            let max_chunk = remaining.min(self.options.chunk_size as u64);
2746            let mut chunk = vec![0u8; to_usize_writer(max_chunk, "payload chunk")?];
2747            reader
2748                .read_exact(&mut chunk)
2749                .map_err(ArchiveWriteError::Io)?;
2750            let mut chunk_len = chunk.len();
2751            let frame = loop {
2752                let candidate = &chunk[..chunk_len];
2753                let frame = compress_zstd_frame_with_jobs(
2754                    candidate,
2755                    self.options.zstd_level,
2756                    self.options.jobs,
2757                )?;
2758                if payload_object_can_fit(frame.len(), self.options)? {
2759                    break frame;
2760                }
2761                if chunk_len == 1 {
2762                    return Err(FormatError::WriterUnsupported(
2763                        "single-byte payload frame exceeds envelope object limits",
2764                    )
2765                    .into());
2766                }
2767                chunk_len = (chunk_len / 2).max(1);
2768            };
2769            if chunk_len < chunk.len() {
2770                reader.push_back(chunk[chunk_len..].to_vec());
2771            }
2772            let chunk = &chunk[..chunk_len];
2773            self.hasher.update(chunk);
2774            self.append_payload_frame(
2775                &frame,
2776                chunk_len,
2777                member_index,
2778                member_start,
2779                member_offset,
2780                member_group_size,
2781            )?;
2782            member_offset = checked_u64_add(member_offset, chunk_len as u64, "payload chunk")?;
2783            self.tar_total_size =
2784                checked_u64_add(self.tar_total_size, chunk_len as u64, "tar stream")?;
2785        }
2786        Ok(())
2787    }
2788
2789    fn append_payload_frame(
2790        &mut self,
2791        frame: &[u8],
2792        decompressed_size: usize,
2793        member_index: usize,
2794        member_start: u64,
2795        member_offset: u64,
2796        member_group_size: u64,
2797    ) -> Result<(), ArchiveWriteError> {
2798        let next_len = checked_usize_add(self.envelope.plaintext.len(), frame.len(), "payload")?;
2799        if !self.envelope.plaintext.is_empty()
2800            && (next_len > self.options.envelope_target_size as usize
2801                || !payload_object_can_fit(next_len, self.options)?)
2802        {
2803            self.flush_payload_envelope()?;
2804        }
2805        if self.envelope.plaintext.is_empty() && !payload_object_can_fit(frame.len(), self.options)?
2806        {
2807            return Err(FormatError::WriterUnsupported(
2808                "payload frame exceeds envelope object limits",
2809            )
2810            .into());
2811        }
2812        let offset = u32_len(
2813            self.envelope.plaintext.len(),
2814            "FrameEntry.offset_in_envelope",
2815        )?;
2816        self.envelope.plaintext.extend_from_slice(frame);
2817        let mut flags = 0u32;
2818        if member_offset == 0 {
2819            flags |= 0x0000_0001;
2820        }
2821        if checked_u64_add(member_offset, decompressed_size as u64, "payload chunk")?
2822            == member_group_size
2823        {
2824            flags |= 0x0000_0002;
2825        }
2826        self.frames.push(PayloadFrame {
2827            frame_index: self.next_frame_index,
2828            envelope_index: self.envelope.envelope_index,
2829            member_index,
2830            offset_in_envelope: offset,
2831            compressed_size: u32_len(frame.len(), "FrameEntry.compressed_size")?,
2832            decompressed_size: u32_len(decompressed_size, "FrameEntry.decompressed_size")?,
2833            flags,
2834            tar_stream_offset: checked_u64_add(
2835                member_start,
2836                member_offset,
2837                "PayloadFrame.tar_stream_offset",
2838            )?,
2839        });
2840        self.next_frame_index =
2841            checked_u64_add(self.next_frame_index, 1, "PayloadFrame.frame_index")?;
2842        Ok(())
2843    }
2844
2845    fn flush_payload_envelope(&mut self) -> Result<(), ArchiveWriteError> {
2846        if self.envelope.plaintext.is_empty() {
2847            return Ok(());
2848        }
2849        let plaintext_size = u32_len(
2850            self.envelope.plaintext.len(),
2851            "EnvelopeEntry.plaintext_size",
2852        )?;
2853        let object_plan = plan_encrypted_object(
2854            self.envelope.plaintext.len(),
2855            self.options.fec_data_shards,
2856            self.options.fec_parity_shards,
2857            self.options,
2858        )?;
2859        let extent = ObjectExtent::new(self.emission_state.next_block_index, object_plan)?;
2860        let object = encrypt_object(
2861            &self.envelope.plaintext,
2862            ObjectEncryptionContext {
2863                key: &self.subkeys.enc_key,
2864                nonce_seed: &self.subkeys.nonce_seed,
2865                domain: b"envelope",
2866                counter: self.envelope.envelope_index,
2867                data_kind: BlockKind::PayloadData,
2868                parity_kind: BlockKind::PayloadParity,
2869                data_shard_max: self.options.fec_data_shards,
2870                class_parity_shard_max: self.options.fec_parity_shards,
2871                archive_uuid: &self.archive_uuid,
2872                session_id: &self.session_id,
2873            },
2874            &mut self.emission_state.next_block_index,
2875            self.options,
2876        )?;
2877        validate_planned_extent(&object, extent)?;
2878        for record in &object.records {
2879            emit_block_record(
2880                self.sink,
2881                self.options,
2882                &mut self.emission_state.bytes_written,
2883                &mut self.emission_state.record_counts,
2884                &mut self.emission_state.data_leaf_hashes,
2885                record,
2886            )?;
2887        }
2888        self.payload_block_count = checked_u64_add(
2889            self.payload_block_count,
2890            extent.data_block_count as u64,
2891            "payload",
2892        )?;
2893        self.payload_objects.push(PayloadObject {
2894            envelope_index: self.envelope.envelope_index,
2895            plaintext_size,
2896            object: extent,
2897        });
2898        self.envelope.envelope_index =
2899            checked_u64_add(self.envelope.envelope_index, 1, "EnvelopeEntry")?;
2900        self.envelope.plaintext.clear();
2901        Ok(())
2902    }
2903
2904    fn finish(
2905        mut self,
2906        master_key: &MasterKey,
2907        kdf_params: &KdfParams,
2908        root_auth: Option<RootAuthWriterConfig<'_>>,
2909        authenticator: Option<&mut RootAuthAuthenticator<'_>>,
2910    ) -> Result<WrittenArchiveSummary, ArchiveWriteError> {
2911        self.flush_payload_envelope()?;
2912        let digest = self.hasher.finalize();
2913        let mut content_sha256 = [0u8; 32];
2914        content_sha256.copy_from_slice(&digest);
2915        let payload = PayloadPlanning {
2916            tar_members: self.tar_members,
2917            frames: self.frames,
2918            payload_objects: self.payload_objects,
2919            payload_block_count: self.payload_block_count,
2920            tar_total_size: self.tar_total_size,
2921            content_sha256,
2922        };
2923        let plan = build_writer_plan_from_payload(
2924            payload,
2925            self.emission_state.next_block_index,
2926            master_key,
2927            self.options,
2928            None,
2929            kdf_params,
2930            self.archive_uuid,
2931            self.session_id,
2932            root_auth,
2933        )?;
2934        if plan.options != self.options || plan.crypto_header != self.crypto_header {
2935            return Err(FormatError::WriterUnsupported(
2936                "streaming tar stdin metadata exceeded the predeclared header class",
2937            )
2938            .into());
2939        }
2940        emit_writer_plan_suffix(
2941            &self.subkeys,
2942            root_auth,
2943            authenticator,
2944            plan,
2945            self.sink,
2946            self.emission_state,
2947        )
2948    }
2949}
2950
2951struct PayloadFramePlanState<'a> {
2952    envelope: &'a mut PayloadEnvelopeBuilder,
2953    payload_objects: &'a mut Vec<PayloadObject>,
2954    payload_block_count: &'a mut u64,
2955    next_block_index: &'a mut u64,
2956    frames: &'a mut Vec<PayloadFrame>,
2957    next_frame_index: &'a mut u64,
2958    options: WriterOptions,
2959}
2960
2961struct PayloadFramePlanInput<'a> {
2962    frame: &'a [u8],
2963    decompressed_size: usize,
2964    member_index: usize,
2965    member_start: u64,
2966    member_offset: u64,
2967    member_group_size: u64,
2968}
2969
2970fn append_payload_frame_to_plan(
2971    state: PayloadFramePlanState<'_>,
2972    input: PayloadFramePlanInput<'_>,
2973) -> Result<(), FormatError> {
2974    let next_len = checked_usize_add(state.envelope.plaintext.len(), input.frame.len(), "payload")?;
2975    if !state.envelope.plaintext.is_empty()
2976        && (next_len > state.options.envelope_target_size as usize
2977            || !payload_object_can_fit(next_len, state.options)?)
2978    {
2979        flush_payload_envelope_plan(
2980            state.envelope,
2981            state.payload_objects,
2982            state.payload_block_count,
2983            state.next_block_index,
2984            state.options,
2985        )?;
2986    }
2987    if state.envelope.plaintext.is_empty()
2988        && !payload_object_can_fit(input.frame.len(), state.options)?
2989    {
2990        return Err(FormatError::WriterUnsupported(
2991            "payload frame exceeds envelope object limits",
2992        ));
2993    }
2994    let offset = u32_len(
2995        state.envelope.plaintext.len(),
2996        "FrameEntry.offset_in_envelope",
2997    )?;
2998    state.envelope.plaintext.extend_from_slice(input.frame);
2999    let mut flags = 0u32;
3000    if input.member_offset == 0 {
3001        flags |= 0x0000_0001;
3002    }
3003    if checked_u64_add(
3004        input.member_offset,
3005        input.decompressed_size as u64,
3006        "payload chunk",
3007    )? == input.member_group_size
3008    {
3009        flags |= 0x0000_0002;
3010    }
3011    state.frames.push(PayloadFrame {
3012        frame_index: *state.next_frame_index,
3013        envelope_index: state.envelope.envelope_index,
3014        member_index: input.member_index,
3015        offset_in_envelope: offset,
3016        compressed_size: u32_len(input.frame.len(), "FrameEntry.compressed_size")?,
3017        decompressed_size: u32_len(input.decompressed_size, "FrameEntry.decompressed_size")?,
3018        flags,
3019        tar_stream_offset: checked_u64_add(
3020            input.member_start,
3021            input.member_offset,
3022            "PayloadFrame.tar_stream_offset",
3023        )?,
3024    });
3025    *state.next_frame_index =
3026        checked_u64_add(*state.next_frame_index, 1, "PayloadFrame.frame_index")?;
3027    Ok(())
3028}
3029
3030fn flush_payload_envelope_plan(
3031    envelope: &mut PayloadEnvelopeBuilder,
3032    payload_objects: &mut Vec<PayloadObject>,
3033    payload_block_count: &mut u64,
3034    next_block_index: &mut u64,
3035    options: WriterOptions,
3036) -> Result<(), FormatError> {
3037    let plaintext_size = u32_len(envelope.plaintext.len(), "EnvelopeEntry.plaintext_size")?;
3038    let object_plan = plan_encrypted_object(
3039        envelope.plaintext.len(),
3040        options.fec_data_shards,
3041        options.fec_parity_shards,
3042        options,
3043    )?;
3044    let extent = ObjectExtent::new(*next_block_index, object_plan)?;
3045    *next_block_index = extent.next_block_index()?;
3046    *payload_block_count = checked_u64_add(
3047        *payload_block_count,
3048        extent.data_block_count as u64,
3049        "payload",
3050    )?;
3051    payload_objects.push(PayloadObject {
3052        envelope_index: envelope.envelope_index,
3053        plaintext_size,
3054        object: extent,
3055    });
3056    envelope.envelope_index = checked_u64_add(envelope.envelope_index, 1, "EnvelopeEntry")?;
3057    envelope.plaintext.clear();
3058    Ok(())
3059}
3060
3061fn required_stripe_width_for_plan(
3062    plan: &WriterPlan,
3063    master_key: &MasterKey,
3064    target_volume_size: u64,
3065) -> Result<u32, FormatError> {
3066    let subkeys = writer_subkeys(
3067        master_key,
3068        plan.options.aead_algo,
3069        &plan.archive_uuid,
3070        &plan.session_id,
3071    )?;
3072    let mut max_volume_size = 0u64;
3073    let mut max_overhead = 0u64;
3074    let block_record_len = plan.options.block_size as u64 + BLOCK_RECORD_FRAMING_LEN as u64;
3075    for volume_index in 0..plan.options.stripe_width {
3076        let block_count = striped_block_count(
3077            plan.total_block_count,
3078            plan.options.stripe_width,
3079            volume_index,
3080        );
3081        let volume_size = planned_v41_volume_size(plan, &subkeys, volume_index, block_count)?;
3082        max_volume_size = max_volume_size.max(volume_size);
3083        let record_bytes = checked_u64_mul(block_count, block_record_len, "volume records")?;
3084        let overhead =
3085            volume_size
3086                .checked_sub(record_bytes)
3087                .ok_or(FormatError::WriterInvariant(
3088                    "planned volume record overflow",
3089                ))?;
3090        max_overhead = max_overhead.max(overhead);
3091    }
3092    if max_volume_size <= target_volume_size {
3093        return Ok(plan.options.stripe_width);
3094    }
3095    if target_volume_size <= max_overhead {
3096        return Err(FormatError::WriterUnsupported(
3097            "volume-size is too small for per-volume metadata",
3098        ));
3099    }
3100
3101    let records_per_volume = (target_volume_size - max_overhead) / block_record_len;
3102    if records_per_volume == 0 {
3103        return Err(FormatError::WriterUnsupported(
3104            "volume-size is too small for the configured block-size",
3105        ));
3106    }
3107
3108    let required = ceil_div(plan.total_block_count, records_per_volume)?
3109        .max(plan.options.volume_loss_tolerance as u64 + 1)
3110        .max(1);
3111    u32::try_from(required).map_err(|_| FormatError::WriterUnsupported("volume count"))
3112}
3113
3114fn planned_v41_volume_size(
3115    plan: &WriterPlan,
3116    subkeys: &Subkeys,
3117    volume_index: u32,
3118    block_count: u64,
3119) -> Result<u64, FormatError> {
3120    let volume_header = VolumeHeader {
3121        format_version: FORMAT_VERSION,
3122        volume_format_rev: VOLUME_FORMAT_REV,
3123        volume_index,
3124        stripe_width: plan.options.stripe_width,
3125        archive_uuid: plan.archive_uuid,
3126        session_id: plan.session_id,
3127        crypto_header_offset: VOLUME_HEADER_LEN as u32,
3128        crypto_header_length: u32_len(plan.crypto_header.len(), "CryptoHeader")?,
3129        header_crc32c: 0,
3130    };
3131    let volume_header_bytes = volume_header.to_bytes();
3132    let block_record_len = plan.options.block_size as u64 + BLOCK_RECORD_FRAMING_LEN as u64;
3133    let block_record_bytes = checked_u64_mul(block_count, block_record_len, "volume records")?;
3134    let manifest_footer_offset = checked_u64_add(
3135        VOLUME_HEADER_LEN as u64 + plan.crypto_header.len() as u64,
3136        block_record_bytes,
3137        "volume records",
3138    )?;
3139    let manifest_footer = build_manifest_footer(
3140        subkeys,
3141        plan.options.aead_algo,
3142        plan.archive_uuid,
3143        plan.session_id,
3144        volume_index,
3145        plan.options.stripe_width,
3146        &plan.index_root_extent,
3147        plan.index_root_plaintext.len(),
3148    )?;
3149    let root_auth_footer = plan
3150        .root_auth_footer_length
3151        .map(|length| vec![0u8; length as usize]);
3152    let root_auth_footer_offset = root_auth_footer
3153        .as_ref()
3154        .map(|_| {
3155            checked_u64_add(
3156                manifest_footer_offset,
3157                MANIFEST_FOOTER_LEN as u64,
3158                "RootAuthFooterV1",
3159            )
3160        })
3161        .transpose()?;
3162    let trailer_offset = checked_u64_add(
3163        manifest_footer_offset,
3164        MANIFEST_FOOTER_LEN as u64 + u64::from(plan.root_auth_footer_length.unwrap_or(0)),
3165        "VolumeTrailer",
3166    )?;
3167    let trailer = build_volume_trailer(VolumeTrailerBuildInput {
3168        subkeys,
3169        aead_algo: plan.options.aead_algo,
3170        archive_uuid: plan.archive_uuid,
3171        session_id: plan.session_id,
3172        volume_index,
3173        block_count,
3174        bytes_written: trailer_offset,
3175        manifest_footer_offset,
3176        closed_at_ns: plan.options.closed_at_ns,
3177        root_auth_footer: root_auth_footer_offset.zip(plan.root_auth_footer_length),
3178    })?;
3179    let cmra_offset = checked_u64_add(trailer_offset, VOLUME_TRAILER_LEN as u64, "CMRA")?;
3180    let cmra = build_v41_cmra(CmraBuildInput {
3181        volume_header_bytes: &volume_header_bytes,
3182        crypto_header: &plan.crypto_header,
3183        block_count,
3184        manifest_footer_offset,
3185        manifest_footer: &manifest_footer,
3186        root_auth_footer_offset,
3187        root_auth_footer: root_auth_footer.as_deref(),
3188        trailer_offset,
3189        trailer: &trailer,
3190        cmra_offset,
3191        options: plan.options,
3192        archive_uuid: plan.archive_uuid,
3193        session_id: plan.session_id,
3194        volume_index,
3195    })?;
3196    checked_u64_add(
3197        checked_u64_add(cmra_offset, cmra.bytes.len() as u64, "CMRA")?,
3198        (CRITICAL_RECOVERY_LOCATOR_LEN * 2) as u64,
3199        "critical recovery locators",
3200    )
3201}
3202
3203fn striped_block_count(total_block_count: u64, stripe_width: u32, volume_index: u32) -> u64 {
3204    let volume_index = volume_index as u64;
3205    let stripe_width = stripe_width as u64;
3206    if total_block_count <= volume_index {
3207        0
3208    } else {
3209        (total_block_count - 1 - volume_index) / stripe_width + 1
3210    }
3211}
3212
3213fn emit_writer_plan<S, O>(
3214    files: &[S],
3215    master_key: &MasterKey,
3216    dictionary: Option<&[u8]>,
3217    root_auth: Option<RootAuthWriterConfig<'_>>,
3218    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
3219    plan: WriterPlan,
3220    sink: &mut O,
3221) -> Result<WrittenArchiveSummary, ArchiveWriteError>
3222where
3223    S: RegularFileSource,
3224    O: ArchiveWriteSink,
3225{
3226    let subkeys = writer_subkeys(
3227        master_key,
3228        plan.options.aead_algo,
3229        &plan.archive_uuid,
3230        &plan.session_id,
3231    )?;
3232    let mut state = begin_writer_emission_state(
3233        sink,
3234        plan.options,
3235        &plan.crypto_header,
3236        plan.archive_uuid,
3237        plan.session_id,
3238    )?;
3239
3240    let emit_payload_started = Instant::now();
3241    emit_payload_stream(
3242        files,
3243        dictionary,
3244        &subkeys,
3245        &plan,
3246        &mut state.next_block_index,
3247        sink,
3248        &mut state.bytes_written,
3249        &mut state.record_counts,
3250        &mut state.data_leaf_hashes,
3251    )?;
3252    let emit_payload = emit_payload_started.elapsed();
3253
3254    let mut summary =
3255        emit_writer_plan_suffix(&subkeys, root_auth, authenticator, plan, sink, state)?;
3256    summary.timings.emit_payload += emit_payload;
3257    Ok(summary)
3258}
3259
3260fn emit_writer_plan_suffix<O: ArchiveWriteSink>(
3261    subkeys: &Subkeys,
3262    root_auth: Option<RootAuthWriterConfig<'_>>,
3263    authenticator: Option<&mut RootAuthAuthenticator<'_>>,
3264    plan: WriterPlan,
3265    sink: &mut O,
3266    mut state: WriterEmissionState,
3267) -> Result<WrittenArchiveSummary, ArchiveWriteError> {
3268    let emit_metadata_started = Instant::now();
3269    let volume_count = plan.options.stripe_width as usize;
3270
3271    for planned in &plan.index_shard_objects {
3272        emit_encrypted_object(
3273            &planned.compressed,
3274            &subkeys.index_shard_key,
3275            &subkeys.index_nonce_seed,
3276            b"idxshard",
3277            planned.shard_index,
3278            BlockKind::IndexShardData,
3279            BlockKind::IndexShardParity,
3280            plan.options.index_fec_data_shards,
3281            plan.options.index_fec_parity_shards,
3282            &mut state.next_block_index,
3283            plan.options,
3284            &plan.archive_uuid,
3285            &plan.session_id,
3286            planned.extent,
3287            None,
3288            sink,
3289            &mut state.bytes_written,
3290            &mut state.record_counts,
3291            &mut state.data_leaf_hashes,
3292        )?;
3293    }
3294
3295    let dictionary_records = if let (Some(compressed), Some((extent, _))) =
3296        (plan.compressed_dictionary.as_ref(), plan.dictionary_extent)
3297    {
3298        let object = emit_encrypted_object(
3299            compressed,
3300            &subkeys.dictionary_key,
3301            &subkeys.index_nonce_seed,
3302            b"dict",
3303            0,
3304            BlockKind::DictionaryData,
3305            BlockKind::DictionaryParity,
3306            plan.options.index_root_fec_data_shards,
3307            plan.options.index_root_fec_parity_shards,
3308            &mut state.next_block_index,
3309            plan.options,
3310            &plan.archive_uuid,
3311            &plan.session_id,
3312            extent,
3313            Some(MetadataObjectKind::Dictionary),
3314            sink,
3315            &mut state.bytes_written,
3316            &mut state.record_counts,
3317            &mut state.data_leaf_hashes,
3318        )?;
3319        Some(object.records)
3320    } else {
3321        None
3322    };
3323
3324    for planned in &plan.directory_hint_objects {
3325        emit_encrypted_object(
3326            &planned.compressed,
3327            &subkeys.dir_hint_key,
3328            &subkeys.index_nonce_seed,
3329            b"dirhint",
3330            planned.hint_shard_index,
3331            BlockKind::DirectoryHintData,
3332            BlockKind::DirectoryHintParity,
3333            plan.options.index_fec_data_shards,
3334            plan.options.index_fec_parity_shards,
3335            &mut state.next_block_index,
3336            plan.options,
3337            &plan.archive_uuid,
3338            &plan.session_id,
3339            planned.extent,
3340            None,
3341            sink,
3342            &mut state.bytes_written,
3343            &mut state.record_counts,
3344            &mut state.data_leaf_hashes,
3345        )?;
3346    }
3347
3348    let index_root_object = emit_encrypted_object(
3349        &plan.compressed_index_root,
3350        &subkeys.index_root_key,
3351        &subkeys.index_nonce_seed,
3352        b"idxroot",
3353        0,
3354        BlockKind::IndexRootData,
3355        BlockKind::IndexRootParity,
3356        plan.options.index_root_fec_data_shards,
3357        plan.options.index_root_fec_parity_shards,
3358        &mut state.next_block_index,
3359        plan.options,
3360        &plan.archive_uuid,
3361        &plan.session_id,
3362        plan.index_root_extent,
3363        Some(MetadataObjectKind::IndexRoot),
3364        sink,
3365        &mut state.bytes_written,
3366        &mut state.record_counts,
3367        &mut state.data_leaf_hashes,
3368    )?;
3369    if state.next_block_index != plan.total_block_count {
3370        return Err(FormatError::WriterInvariant("streaming writer block plan mismatch").into());
3371    }
3372
3373    let volume_zero_manifest = build_manifest_footer(
3374        subkeys,
3375        plan.options.aead_algo,
3376        plan.archive_uuid,
3377        plan.session_id,
3378        0,
3379        plan.options.stripe_width,
3380        &plan.index_root_extent,
3381        plan.index_root_plaintext.len(),
3382    )?;
3383    let root_auth_footer = match root_auth {
3384        Some(config) => {
3385            let signer = authenticator.ok_or(FormatError::WriterInvariant(
3386                "missing root-auth authenticator",
3387            ))?;
3388            Some(build_root_auth_footer_from_leaf_hashes(
3389                config,
3390                signer,
3391                RootAuthFooterBuildInput {
3392                    archive_uuid: plan.archive_uuid,
3393                    session_id: plan.session_id,
3394                    options: plan.options,
3395                    crypto_header: &plan.crypto_header,
3396                    volume_zero_manifest: &volume_zero_manifest,
3397                    index_root_plaintext: &plan.index_root_plaintext,
3398                    index_root_extent: plan.index_root_extent,
3399                    dictionary_extent: plan.dictionary_extent,
3400                    shard_entries: &plan.shard_entries,
3401                    payload_objects: &plan.payload_objects,
3402                    directory_hint_entries: &plan.directory_hint_entries,
3403                    data_leaf_hashes: &state.data_leaf_hashes,
3404                },
3405            )?)
3406        }
3407        None => None,
3408    };
3409    let root_auth_footer_length = root_auth_footer
3410        .as_ref()
3411        .map(|footer| u32_len(footer.len(), "RootAuthFooterV1"))
3412        .transpose()?;
3413
3414    for volume_index in 0..volume_count {
3415        let volume_index_u32 = u32::try_from(volume_index)
3416            .map_err(|_| FormatError::WriterUnsupported("volume_index"))?;
3417        let manifest_footer_offset = state.bytes_written[volume_index];
3418        let manifest_footer = build_manifest_footer(
3419            subkeys,
3420            plan.options.aead_algo,
3421            plan.archive_uuid,
3422            plan.session_id,
3423            volume_index_u32,
3424            plan.options.stripe_width,
3425            &plan.index_root_extent,
3426            plan.index_root_plaintext.len(),
3427        )?;
3428        sink.write_volume(volume_index, &manifest_footer)?;
3429        state.bytes_written[volume_index] = checked_u64_add(
3430            state.bytes_written[volume_index],
3431            MANIFEST_FOOTER_LEN as u64,
3432            "ManifestFooter",
3433        )?;
3434
3435        let root_auth_footer_offset = if let Some(root_auth_footer) = root_auth_footer.as_ref() {
3436            let offset = state.bytes_written[volume_index];
3437            sink.write_volume(volume_index, root_auth_footer)?;
3438            state.bytes_written[volume_index] = checked_u64_add(
3439                state.bytes_written[volume_index],
3440                root_auth_footer.len() as u64,
3441                "RootAuthFooterV1",
3442            )?;
3443            Some(offset)
3444        } else {
3445            None
3446        };
3447
3448        let trailer_offset = state.bytes_written[volume_index];
3449        let trailer = build_volume_trailer(VolumeTrailerBuildInput {
3450            subkeys,
3451            aead_algo: plan.options.aead_algo,
3452            archive_uuid: plan.archive_uuid,
3453            session_id: plan.session_id,
3454            volume_index: volume_index_u32,
3455            block_count: state.record_counts[volume_index],
3456            bytes_written: trailer_offset,
3457            manifest_footer_offset,
3458            closed_at_ns: plan.options.closed_at_ns,
3459            root_auth_footer: root_auth_footer_offset.zip(root_auth_footer_length),
3460        })?;
3461        sink.write_volume(volume_index, &trailer)?;
3462        state.bytes_written[volume_index] = checked_u64_add(
3463            state.bytes_written[volume_index],
3464            VOLUME_TRAILER_LEN as u64,
3465            "VolumeTrailer",
3466        )?;
3467
3468        let cmra_offset = state.bytes_written[volume_index];
3469        let cmra = build_v41_cmra(CmraBuildInput {
3470            volume_header_bytes: &state.volume_headers[volume_index],
3471            crypto_header: &plan.crypto_header,
3472            block_count: state.record_counts[volume_index],
3473            manifest_footer_offset,
3474            manifest_footer: &manifest_footer,
3475            root_auth_footer_offset,
3476            root_auth_footer: root_auth_footer.as_deref(),
3477            trailer_offset,
3478            trailer: &trailer,
3479            cmra_offset,
3480            options: plan.options,
3481            archive_uuid: plan.archive_uuid,
3482            session_id: plan.session_id,
3483            volume_index: volume_index_u32,
3484        })?;
3485        sink.write_volume(volume_index, &cmra.bytes)?;
3486        state.bytes_written[volume_index] = checked_u64_add(
3487            state.bytes_written[volume_index],
3488            cmra.bytes.len() as u64,
3489            "CMRA",
3490        )?;
3491        let locator_base = CriticalRecoveryLocator {
3492            cmra_offset,
3493            cmra_length: u32_len(cmra.bytes.len(), "CMRA")?,
3494            volume_trailer_offset: trailer_offset,
3495            body_bytes_before_cmra: cmra_offset,
3496            archive_uuid_hint: plan.archive_uuid,
3497            session_id_hint: plan.session_id,
3498            volume_index_hint: volume_index_u32,
3499            locator_sequence: 1,
3500            cmra_shard_size: cmra.shard_size,
3501            cmra_data_shard_count: cmra.data_shard_count,
3502            cmra_parity_shard_count: cmra.parity_shard_count,
3503            cmra_image_length: cmra.image_length,
3504            cmra_image_sha256: cmra.image_sha256,
3505            locator_crc32c: 0,
3506        };
3507        let mirror = locator_base.to_bytes();
3508        sink.write_volume(volume_index, &mirror)?;
3509        let final_locator = CriticalRecoveryLocator {
3510            locator_sequence: 0,
3511            ..locator_base
3512        }
3513        .to_bytes();
3514        sink.write_volume(volume_index, &final_locator)?;
3515        state.bytes_written[volume_index] = checked_u64_add(
3516            state.bytes_written[volume_index],
3517            (CRITICAL_RECOVERY_LOCATOR_LEN * 2) as u64,
3518            "critical recovery locators",
3519        )?;
3520
3521        if volume_index == 0 {
3522            debug_assert_eq!(volume_zero_manifest, manifest_footer);
3523        }
3524    }
3525
3526    let bootstrap_sidecar_bytes = if plan.options.stripe_width == 1 {
3527        let sidecar = build_bootstrap_sidecar(
3528            subkeys,
3529            plan.options.aead_algo,
3530            plan.archive_uuid,
3531            plan.session_id,
3532            &volume_zero_manifest,
3533            &index_root_object.records,
3534            dictionary_records.as_deref(),
3535        )?;
3536        let sidecar_len = sidecar.len() as u64;
3537        sink.write_bootstrap_sidecar(&sidecar)?;
3538        sidecar_len
3539    } else {
3540        0
3541    };
3542
3543    Ok(WrittenArchiveSummary {
3544        volume_count,
3545        archive_bytes: state.bytes_written.iter().sum(),
3546        bootstrap_sidecar_bytes,
3547        archive_uuid: plan.archive_uuid,
3548        session_id: plan.session_id,
3549        timings: WriterTimings {
3550            emit_metadata: emit_metadata_started.elapsed(),
3551            ..WriterTimings::default()
3552        },
3553    })
3554}
3555
3556#[allow(clippy::too_many_arguments)]
3557fn emit_encrypted_object<O: ArchiveWriteSink>(
3558    payload: &[u8],
3559    key: &[u8; 32],
3560    nonce_seed: &[u8; 32],
3561    domain: &[u8],
3562    counter: u64,
3563    data_kind: BlockKind,
3564    parity_kind: BlockKind,
3565    data_shard_max: u16,
3566    class_parity_shard_max: u16,
3567    next_block_index: &mut u64,
3568    options: WriterOptions,
3569    archive_uuid: &[u8; 16],
3570    session_id: &[u8; 16],
3571    expected_extent: ObjectExtent,
3572    metadata_kind: Option<MetadataObjectKind>,
3573    sink: &mut O,
3574    bytes_written: &mut [u64],
3575    record_counts: &mut [u64],
3576    data_leaf_hashes: &mut Vec<(u64, [u8; 32])>,
3577) -> Result<EncryptedObject, ArchiveWriteError> {
3578    let object = encrypt_object(
3579        payload,
3580        ObjectEncryptionContext {
3581            key,
3582            nonce_seed,
3583            domain,
3584            counter,
3585            data_kind,
3586            parity_kind,
3587            data_shard_max,
3588            class_parity_shard_max,
3589            archive_uuid,
3590            session_id,
3591        },
3592        next_block_index,
3593        options,
3594    )
3595    .map_err(|error| match metadata_kind {
3596        Some(kind) => map_metadata_encrypt_error(error, kind),
3597        None => error,
3598    })?;
3599    validate_planned_extent(&object, expected_extent)?;
3600    for record in &object.records {
3601        emit_block_record(
3602            sink,
3603            options,
3604            bytes_written,
3605            record_counts,
3606            data_leaf_hashes,
3607            record,
3608        )?;
3609    }
3610    Ok(object)
3611}
3612
3613fn emit_block_record<O: ArchiveWriteSink>(
3614    sink: &mut O,
3615    options: WriterOptions,
3616    bytes_written: &mut [u64],
3617    record_counts: &mut [u64],
3618    data_leaf_hashes: &mut Vec<(u64, [u8; 32])>,
3619    record: &BlockRecord,
3620) -> Result<(), ArchiveWriteError> {
3621    let volume_index = (record.block_index % options.stripe_width as u64) as usize;
3622    let record_bytes = record.to_bytes();
3623    sink.write_volume(volume_index, &record_bytes)?;
3624    bytes_written[volume_index] = checked_u64_add(
3625        bytes_written[volume_index],
3626        record_bytes.len() as u64,
3627        "BlockRecord",
3628    )?;
3629    record_counts[volume_index] =
3630        checked_u64_add(record_counts[volume_index], 1, "BlockRecord count")?;
3631    if record.kind.is_data() {
3632        data_leaf_hashes.push((
3633            record.block_index,
3634            data_block_merkle_leaf_hash(
3635                record.block_index,
3636                record.kind,
3637                record.flags,
3638                &record.payload,
3639            ),
3640        ));
3641    }
3642    Ok(())
3643}
3644
3645#[allow(clippy::too_many_arguments)]
3646fn emit_payload_stream<S, O>(
3647    files: &[S],
3648    dictionary: Option<&[u8]>,
3649    subkeys: &Subkeys,
3650    plan: &WriterPlan,
3651    next_block_index: &mut u64,
3652    sink: &mut O,
3653    bytes_written: &mut [u64],
3654    record_counts: &mut [u64],
3655    data_leaf_hashes: &mut Vec<(u64, [u8; 32])>,
3656) -> Result<(), ArchiveWriteError>
3657where
3658    S: RegularFileSource,
3659    O: ArchiveWriteSink,
3660{
3661    let mut envelope = PayloadEnvelopeBuilder {
3662        envelope_index: 0,
3663        plaintext: Vec::new(),
3664    };
3665    let mut next_frame_index = 0u64;
3666
3667    for (member_index, file) in files.iter().enumerate() {
3668        let member = plan
3669            .tar_members
3670            .get(member_index)
3671            .ok_or(FormatError::WriterInvariant(
3672                "planned tar member is missing",
3673            ))?;
3674        let current_path =
3675            normalize_lookup_file_path(file.archive_path(), plan.options.max_path_length)?;
3676        if current_path != member.path
3677            || file.file_data_size() != member.file_data_size
3678            || file.mode() != member.mode
3679            || file.mtime() != member.mtime
3680        {
3681            return Err(FormatError::WriterInvariant(
3682                "file source changed between planning and emission",
3683            )
3684            .into());
3685        }
3686        let prefix = build_regular_file_member_prefix(
3687            &member.path,
3688            member.file_data_size,
3689            member.mode,
3690            member.mtime,
3691        )?;
3692        let mut reader = StreamingMemberReader::new(file.open()?, prefix, member.file_data_size);
3693        let mut member_offset = 0u64;
3694        while member_offset < member.tar_member_group_size {
3695            let remaining = member.tar_member_group_size - member_offset;
3696            let max_chunk = remaining.min(plan.options.chunk_size as u64);
3697            let mut chunk = vec![0u8; to_usize_writer(max_chunk, "payload chunk")?];
3698            reader
3699                .read_exact(&mut chunk)
3700                .map_err(ArchiveWriteError::Io)?;
3701            let mut chunk_len = chunk.len();
3702            let frame = loop {
3703                let candidate = &chunk[..chunk_len];
3704                let frame = if let Some(dictionary) = dictionary {
3705                    compress_zstd_frame_with_dictionary_and_jobs(
3706                        candidate,
3707                        plan.options.zstd_level,
3708                        dictionary,
3709                        plan.options.jobs,
3710                    )?
3711                } else {
3712                    compress_zstd_frame_with_jobs(
3713                        candidate,
3714                        plan.options.zstd_level,
3715                        plan.options.jobs,
3716                    )?
3717                };
3718                if payload_object_can_fit(frame.len(), plan.options)? {
3719                    break frame;
3720                }
3721                if chunk_len == 1 {
3722                    return Err(FormatError::WriterUnsupported(
3723                        "single-byte payload frame exceeds envelope object limits",
3724                    )
3725                    .into());
3726                }
3727                chunk_len = (chunk_len / 2).max(1);
3728            };
3729            if chunk_len < chunk.len() {
3730                reader.push_back(chunk[chunk_len..].to_vec());
3731            }
3732            append_payload_frame_to_emit(
3733                &mut envelope,
3734                &frame,
3735                chunk_len,
3736                member_index,
3737                member.tar_member_group_start,
3738                member_offset,
3739                member.tar_member_group_size,
3740                &mut next_frame_index,
3741                subkeys,
3742                plan,
3743                next_block_index,
3744                sink,
3745                bytes_written,
3746                record_counts,
3747                data_leaf_hashes,
3748            )?;
3749            member_offset = checked_u64_add(member_offset, chunk_len as u64, "payload chunk")?;
3750        }
3751    }
3752
3753    if !envelope.plaintext.is_empty() {
3754        flush_payload_envelope_emit(
3755            &mut envelope,
3756            subkeys,
3757            plan,
3758            next_block_index,
3759            sink,
3760            bytes_written,
3761            record_counts,
3762            data_leaf_hashes,
3763        )?;
3764    }
3765    if next_frame_index != plan.frames.len() as u64
3766        || envelope.envelope_index != plan.payload_objects.len() as u64
3767    {
3768        return Err(FormatError::WriterInvariant("streaming payload plan mismatch").into());
3769    }
3770    Ok(())
3771}
3772
3773#[allow(clippy::too_many_arguments)]
3774fn append_payload_frame_to_emit<O: ArchiveWriteSink>(
3775    envelope: &mut PayloadEnvelopeBuilder,
3776    frame: &[u8],
3777    decompressed_size: usize,
3778    member_index: usize,
3779    member_start: u64,
3780    member_offset: u64,
3781    member_group_size: u64,
3782    next_frame_index: &mut u64,
3783    subkeys: &Subkeys,
3784    plan: &WriterPlan,
3785    next_block_index: &mut u64,
3786    sink: &mut O,
3787    bytes_written: &mut [u64],
3788    record_counts: &mut [u64],
3789    data_leaf_hashes: &mut Vec<(u64, [u8; 32])>,
3790) -> Result<(), ArchiveWriteError> {
3791    let next_len = checked_usize_add(envelope.plaintext.len(), frame.len(), "payload")?;
3792    if !envelope.plaintext.is_empty()
3793        && (next_len > plan.options.envelope_target_size as usize
3794            || !payload_object_can_fit(next_len, plan.options)?)
3795    {
3796        flush_payload_envelope_emit(
3797            envelope,
3798            subkeys,
3799            plan,
3800            next_block_index,
3801            sink,
3802            bytes_written,
3803            record_counts,
3804            data_leaf_hashes,
3805        )?;
3806    }
3807    if envelope.plaintext.is_empty() && !payload_object_can_fit(frame.len(), plan.options)? {
3808        return Err(
3809            FormatError::WriterUnsupported("payload frame exceeds envelope object limits").into(),
3810        );
3811    }
3812    let offset = u32_len(envelope.plaintext.len(), "FrameEntry.offset_in_envelope")?;
3813    let mut flags = 0u32;
3814    if member_offset == 0 {
3815        flags |= 0x0000_0001;
3816    }
3817    if checked_u64_add(member_offset, decompressed_size as u64, "payload chunk")?
3818        == member_group_size
3819    {
3820        flags |= 0x0000_0002;
3821    }
3822    let expected =
3823        plan.frames
3824            .get(*next_frame_index as usize)
3825            .ok_or(FormatError::WriterInvariant(
3826                "planned payload frame is missing",
3827            ))?;
3828    let tar_stream_offset = checked_u64_add(
3829        member_start,
3830        member_offset,
3831        "PayloadFrame.tar_stream_offset",
3832    )?;
3833    if expected.envelope_index != envelope.envelope_index
3834        || expected.member_index != member_index
3835        || expected.offset_in_envelope != offset
3836        || expected.compressed_size != u32_len(frame.len(), "FrameEntry.compressed_size")?
3837        || expected.decompressed_size != u32_len(decompressed_size, "FrameEntry.decompressed_size")?
3838        || expected.flags != flags
3839        || expected.tar_stream_offset != tar_stream_offset
3840    {
3841        return Err(
3842            FormatError::WriterInvariant("emitted payload frame does not match plan").into(),
3843        );
3844    }
3845    envelope.plaintext.extend_from_slice(frame);
3846    *next_frame_index = checked_u64_add(*next_frame_index, 1, "PayloadFrame.frame_index")?;
3847    Ok(())
3848}
3849
3850#[allow(clippy::too_many_arguments)]
3851fn flush_payload_envelope_emit<O: ArchiveWriteSink>(
3852    envelope: &mut PayloadEnvelopeBuilder,
3853    subkeys: &Subkeys,
3854    plan: &WriterPlan,
3855    next_block_index: &mut u64,
3856    sink: &mut O,
3857    bytes_written: &mut [u64],
3858    record_counts: &mut [u64],
3859    data_leaf_hashes: &mut Vec<(u64, [u8; 32])>,
3860) -> Result<(), ArchiveWriteError> {
3861    let expected = plan
3862        .payload_objects
3863        .get(envelope.envelope_index as usize)
3864        .ok_or(FormatError::WriterInvariant(
3865            "planned payload envelope is missing",
3866        ))?;
3867    if expected.envelope_index != envelope.envelope_index
3868        || expected.plaintext_size
3869            != u32_len(envelope.plaintext.len(), "EnvelopeEntry.plaintext_size")?
3870    {
3871        return Err(
3872            FormatError::WriterInvariant("emitted payload envelope does not match plan").into(),
3873        );
3874    }
3875    emit_encrypted_object(
3876        &envelope.plaintext,
3877        &subkeys.enc_key,
3878        &subkeys.nonce_seed,
3879        b"envelope",
3880        envelope.envelope_index,
3881        BlockKind::PayloadData,
3882        BlockKind::PayloadParity,
3883        plan.options.fec_data_shards,
3884        plan.options.fec_parity_shards,
3885        next_block_index,
3886        plan.options,
3887        &plan.archive_uuid,
3888        &plan.session_id,
3889        expected.object,
3890        None,
3891        sink,
3892        bytes_written,
3893        record_counts,
3894        data_leaf_hashes,
3895    )?;
3896    envelope.envelope_index = checked_u64_add(envelope.envelope_index, 1, "EnvelopeEntry")?;
3897    envelope.plaintext.clear();
3898    Ok(())
3899}
3900
3901pub fn write_empty_archive(master_key: &MasterKey) -> Result<WrittenArchive, FormatError> {
3902    write_archive(&[], master_key, WriterOptions::default())
3903}
3904
3905fn plan_writer_options(mut options: WriterOptions) -> Result<WriterOptions, FormatError> {
3906    if options.jobs == 0 {
3907        return Err(FormatError::WriterUnsupported("jobs must be at least 1"));
3908    }
3909    if options.block_size < MIN_BLOCK_SIZE || options.block_size % 2 != 0 {
3910        return Err(FormatError::WriterUnsupported(
3911            "writer requires an even block size of at least 4096",
3912        ));
3913    }
3914    if options.stripe_width == 0 {
3915        return Err(FormatError::WriterUnsupported(
3916            "stripe_width must be non-zero",
3917        ));
3918    }
3919    if options.volume_loss_tolerance as u32 >= options.stripe_width {
3920        return Err(FormatError::WriterUnsupported(
3921            "volume_loss_tolerance must be less than stripe_width",
3922        ));
3923    }
3924    if options.stripe_width == 1 && options.volume_loss_tolerance != 0 {
3925        return Err(FormatError::WriterUnsupported(
3926            "single-volume archives cannot tolerate volume loss",
3927        ));
3928    }
3929    if matches!(options.target_volume_size, Some(0)) {
3930        return Err(FormatError::WriterUnsupported(
3931            "target_volume_size must be non-zero",
3932        ));
3933    }
3934    if options.bit_rot_buffer_pct > 100 {
3935        return Err(FormatError::WriterUnsupported(
3936            "bit_rot_buffer_pct must be at most 100",
3937        ));
3938    }
3939    if options.chunk_size == 0 || options.chunk_size > options.envelope_target_size {
3940        return Err(FormatError::WriterUnsupported(
3941            "chunk_size must be non-zero and no larger than envelope_target_size",
3942        ));
3943    }
3944    if options.fec_data_shards == 0
3945        || options.index_fec_data_shards == 0
3946        || options.index_root_fec_data_shards == 0
3947    {
3948        return Err(FormatError::WriterUnsupported(
3949            "FEC data shard class maxima must be non-zero",
3950        ));
3951    }
3952    options.index_root_fec_data_shards = options
3953        .index_root_fec_data_shards
3954        .max(MIN_INDEX_ROOT_FEC_DATA_SHARDS);
3955    options.fec_parity_shards =
3956        compute_parity_u16(options.fec_data_shards as u64, options, "fec_parity_shards")?;
3957    options.index_fec_parity_shards = compute_parity_u16(
3958        options.index_fec_data_shards as u64,
3959        options,
3960        "index_fec_parity_shards",
3961    )?;
3962    options.index_root_fec_parity_shards = compute_parity_u16(
3963        options.index_root_fec_data_shards as u64,
3964        options,
3965        "index_root_fec_parity_shards",
3966    )?;
3967    validate_writer_options_match_reader_caps(options)?;
3968    Ok(options)
3969}
3970
3971fn validate_writer_options_match_reader_caps(options: WriterOptions) -> Result<(), FormatError> {
3972    CryptoHeaderFixed {
3973        length: CRYPTO_HEADER_FIXED_LEN as u32,
3974        compression_algo: CompressionAlgo::ZstdFramed,
3975        aead_algo: options.aead_algo,
3976        fec_algo: FecAlgo::ReedSolomonGF16,
3977        kdf_algo: if options.aead_algo.is_encrypted() {
3978            KdfAlgo::Raw
3979        } else {
3980            KdfAlgo::None
3981        },
3982        chunk_size: options.chunk_size,
3983        envelope_target_size: options.envelope_target_size,
3984        block_size: options.block_size,
3985        fec_data_shards: options.fec_data_shards,
3986        fec_parity_shards: options.fec_parity_shards,
3987        index_fec_data_shards: options.index_fec_data_shards,
3988        index_fec_parity_shards: options.index_fec_parity_shards,
3989        index_root_fec_data_shards: options.index_root_fec_data_shards,
3990        index_root_fec_parity_shards: options.index_root_fec_parity_shards,
3991        stripe_width: options.stripe_width,
3992        volume_loss_tolerance: options.volume_loss_tolerance,
3993        bit_rot_buffer_pct: options.bit_rot_buffer_pct,
3994        has_dictionary: 0,
3995        max_path_length: options.max_path_length,
3996        expected_volume_size: options.target_volume_size.unwrap_or(0),
3997    }
3998    .validate_supported_profile()
3999}
4000
4001fn build_crypto_header(
4002    options: WriterOptions,
4003    has_dictionary: bool,
4004    subkeys: &Subkeys,
4005    archive_uuid: &[u8; 16],
4006    session_id: &[u8; 16],
4007    kdf_params: &KdfParams,
4008) -> Result<Vec<u8>, FormatError> {
4009    let kdf_payload = serialize_kdf_params(kdf_params)?;
4010    let length = CRYPTO_HEADER_FIXED_LEN
4011        .checked_add(kdf_payload.len())
4012        .and_then(|value| value.checked_add(CRYPTO_EXTENSION_HEADER_LEN))
4013        .and_then(|value| value.checked_add(CRYPTO_HEADER_HMAC_LEN))
4014        .ok_or(FormatError::WriterUnsupported(
4015            "CryptoHeader length overflow",
4016        ))?;
4017    let kdf_algo = match kdf_params {
4018        KdfParams::None => KdfAlgo::None,
4019        KdfParams::Raw => KdfAlgo::Raw,
4020        KdfParams::Argon2id { .. } => KdfAlgo::Argon2id,
4021    };
4022    match (options.aead_algo, kdf_algo) {
4023        (AeadAlgo::None, KdfAlgo::None) => {}
4024        (aead_algo, KdfAlgo::Raw | KdfAlgo::Argon2id) if aead_algo.is_encrypted() => {}
4025        _ => {
4026            return Err(FormatError::InvalidProtectionMode {
4027                aead_algo: options.aead_algo,
4028                kdf_algo,
4029            });
4030        }
4031    }
4032    let fixed = CryptoHeaderFixed {
4033        length: length as u32,
4034        compression_algo: CompressionAlgo::ZstdFramed,
4035        aead_algo: options.aead_algo,
4036        fec_algo: FecAlgo::ReedSolomonGF16,
4037        kdf_algo,
4038        chunk_size: options.chunk_size,
4039        envelope_target_size: options.envelope_target_size,
4040        block_size: options.block_size,
4041        fec_data_shards: options.fec_data_shards,
4042        fec_parity_shards: options.fec_parity_shards,
4043        index_fec_data_shards: options.index_fec_data_shards,
4044        index_fec_parity_shards: options.index_fec_parity_shards,
4045        index_root_fec_data_shards: options.index_root_fec_data_shards,
4046        index_root_fec_parity_shards: options.index_root_fec_parity_shards,
4047        stripe_width: options.stripe_width,
4048        volume_loss_tolerance: options.volume_loss_tolerance,
4049        bit_rot_buffer_pct: options.bit_rot_buffer_pct,
4050        has_dictionary: if has_dictionary { 1 } else { 0 },
4051        max_path_length: options.max_path_length,
4052        expected_volume_size: options.target_volume_size.unwrap_or(0),
4053    };
4054
4055    let mut bytes = fixed.to_bytes().to_vec();
4056    bytes.extend_from_slice(&kdf_payload);
4057    bytes.extend_from_slice(&0u16.to_le_bytes());
4058    bytes.extend_from_slice(&0u32.to_le_bytes());
4059    let hmac = compute_integrity_tag(
4060        HmacDomain::CryptoHeader,
4061        options.aead_algo,
4062        Some(&subkeys.mac_key),
4063        archive_uuid,
4064        session_id,
4065        &bytes,
4066    )?;
4067    bytes.extend_from_slice(&hmac);
4068    Ok(bytes)
4069}
4070
4071fn serialize_kdf_params(params: &KdfParams) -> Result<Vec<u8>, FormatError> {
4072    let mut bytes = Vec::new();
4073    match params {
4074        KdfParams::None => {
4075            bytes.extend_from_slice(&(KdfAlgo::None as u16).to_le_bytes());
4076        }
4077        KdfParams::Raw => {
4078            bytes.extend_from_slice(&(KdfAlgo::Raw as u16).to_le_bytes());
4079        }
4080        KdfParams::Argon2id {
4081            t_cost,
4082            m_cost_kib,
4083            parallelism,
4084            salt,
4085        } => {
4086            if *t_cost == 0 {
4087                return Err(FormatError::InvalidKdfParams("t_cost must be non-zero"));
4088            }
4089            if *parallelism == 0 {
4090                return Err(FormatError::InvalidKdfParams(
4091                    "parallelism must be non-zero",
4092                ));
4093            }
4094            let min_memory = parallelism
4095                .checked_mul(8)
4096                .ok_or(FormatError::InvalidKdfParams(
4097                    "m_cost_kib requirement overflow",
4098                ))?;
4099            if *m_cost_kib < min_memory {
4100                return Err(FormatError::InvalidKdfParams(
4101                    "m_cost_kib must be at least 8 * parallelism",
4102                ));
4103            }
4104            if !(8..=64).contains(&salt.len()) {
4105                return Err(FormatError::InvalidKdfParams(
4106                    "argon2id salt length must be 8..64",
4107                ));
4108            }
4109            let salt_len = u16::try_from(salt.len())
4110                .map_err(|_| FormatError::InvalidKdfParams("argon2id salt too long"))?;
4111            bytes.extend_from_slice(&(KdfAlgo::Argon2id as u16).to_le_bytes());
4112            bytes.extend_from_slice(&t_cost.to_le_bytes());
4113            bytes.extend_from_slice(&m_cost_kib.to_le_bytes());
4114            bytes.extend_from_slice(&parallelism.to_le_bytes());
4115            bytes.extend_from_slice(&salt_len.to_le_bytes());
4116            bytes.extend_from_slice(salt);
4117        }
4118    }
4119    Ok(bytes)
4120}
4121
4122#[cfg(test)]
4123fn build_tar_stream(
4124    files: &[RegularFile<'_>],
4125    max_path_length: u32,
4126) -> Result<(Vec<u8>, Vec<TarMember>), FormatError> {
4127    let mut stream = Vec::new();
4128    let mut members = Vec::with_capacity(files.len());
4129    for file in files {
4130        let path = normalize_lookup_file_path(file.path, max_path_length)?;
4131        let start = stream.len() as u64;
4132        let member_group =
4133            build_regular_file_member_group(&path, file.contents, file.mode, file.mtime)?;
4134        stream.extend_from_slice(&member_group);
4135        members.push(TarMember {
4136            path,
4137            tar_member_group_start: start,
4138            tar_member_group_size: member_group.len() as u64,
4139            file_data_size: file.contents.len() as u64,
4140            mode: file.mode,
4141            mtime: file.mtime,
4142        });
4143    }
4144    Ok((stream, members))
4145}
4146
4147#[cfg(test)]
4148fn build_payload_envelopes(
4149    tar_stream: &[u8],
4150    members: &[TarMember],
4151    options: WriterOptions,
4152    dictionary: Option<&[u8]>,
4153) -> Result<(Vec<PayloadEnvelope>, Vec<PayloadFrame>), FormatError> {
4154    let chunk_size = options.chunk_size as usize;
4155    if chunk_size == 0 {
4156        return Err(FormatError::WriterUnsupported(
4157            "chunk_size must be non-zero and no larger than envelope_target_size",
4158        ));
4159    }
4160    let envelope_target_size = options.envelope_target_size as usize;
4161    let mut envelopes = Vec::new();
4162    let mut current = PayloadEnvelope {
4163        envelope_index: 0,
4164        plaintext: Vec::new(),
4165    };
4166    let mut frames = Vec::new();
4167    let mut next_frame_index = 0u64;
4168
4169    for (member_index, member) in members.iter().enumerate() {
4170        let start = member.tar_member_group_start as usize;
4171        let end = checked_usize_add(start, member.tar_member_group_size as usize, "tar member")?;
4172        let member_bytes = tar_stream
4173            .get(start..end)
4174            .ok_or(FormatError::WriterInvariant(
4175                "tar member range is out of bounds",
4176            ))?;
4177        let mut member_offset = 0usize;
4178        while member_offset < member_bytes.len() {
4179            let mut chunk_len = (member_bytes.len() - member_offset).min(chunk_size);
4180            let frame = loop {
4181                let end = checked_usize_add(member_offset, chunk_len, "payload chunk")?;
4182                let chunk = &member_bytes[member_offset..end];
4183                let frame = if let Some(dictionary) = dictionary {
4184                    compress_zstd_frame_with_dictionary_and_jobs(
4185                        chunk,
4186                        options.zstd_level,
4187                        dictionary,
4188                        options.jobs,
4189                    )?
4190                } else {
4191                    compress_zstd_frame_with_jobs(chunk, options.zstd_level, options.jobs)?
4192                };
4193                if payload_object_can_fit(frame.len(), options)? {
4194                    break frame;
4195                }
4196                if chunk_len == 1 {
4197                    return Err(FormatError::WriterUnsupported(
4198                        "single-byte payload frame exceeds envelope object limits",
4199                    ));
4200                }
4201                chunk_len = (chunk_len / 2).max(1);
4202            };
4203            let next_len = checked_usize_add(current.plaintext.len(), frame.len(), "payload")?;
4204            if !current.plaintext.is_empty()
4205                && (next_len > envelope_target_size || !payload_object_can_fit(next_len, options)?)
4206            {
4207                envelopes.push(current);
4208                current = PayloadEnvelope {
4209                    envelope_index: envelopes.len() as u64,
4210                    plaintext: Vec::new(),
4211                };
4212            }
4213
4214            if current.plaintext.is_empty() && !payload_object_can_fit(frame.len(), options)? {
4215                return Err(FormatError::WriterUnsupported(
4216                    "payload frame exceeds envelope object limits",
4217                ));
4218            }
4219            let offset = u32_len(current.plaintext.len(), "FrameEntry.offset_in_envelope")?;
4220            current.plaintext.extend_from_slice(&frame);
4221            let is_first_member_frame = member_offset == 0;
4222            let is_last_member_frame =
4223                checked_usize_add(member_offset, chunk_len, "payload chunk")? == member_bytes.len();
4224            let mut flags = 0u32;
4225            if is_first_member_frame {
4226                flags |= 0x0000_0001;
4227            }
4228            if is_last_member_frame {
4229                flags |= 0x0000_0002;
4230            }
4231            frames.push(PayloadFrame {
4232                frame_index: next_frame_index,
4233                envelope_index: current.envelope_index,
4234                member_index,
4235                offset_in_envelope: offset,
4236                compressed_size: u32_len(frame.len(), "FrameEntry.compressed_size")?,
4237                decompressed_size: u32_len(chunk_len, "FrameEntry.decompressed_size")?,
4238                flags,
4239                tar_stream_offset: checked_u64_add(
4240                    member.tar_member_group_start,
4241                    u64::try_from(member_offset)
4242                        .map_err(|_| FormatError::WriterUnsupported("chunk offset"))?,
4243                    "PayloadFrame.tar_stream_offset",
4244                )?,
4245            });
4246            next_frame_index = checked_u64_add(next_frame_index, 1, "PayloadFrame.frame_index")?;
4247            member_offset = checked_usize_add(member_offset, chunk_len, "payload chunk")?;
4248        }
4249    }
4250    if !current.plaintext.is_empty() {
4251        envelopes.push(current);
4252    }
4253    Ok((envelopes, frames))
4254}
4255
4256fn sorted_file_rows(members: &[TarMember]) -> Vec<FileRow> {
4257    let mut rows = members
4258        .iter()
4259        .enumerate()
4260        .map(|(member_index, member)| FileRow {
4261            path_hash: hash_prefix(&member.path),
4262            path: member.path.clone(),
4263            member_index,
4264            member: member.clone(),
4265        })
4266        .collect::<Vec<_>>();
4267    rows.sort_by(|left, right| {
4268        (
4269            left.path_hash,
4270            left.path.as_slice(),
4271            left.member.tar_member_group_start,
4272        )
4273            .cmp(&(
4274                right.path_hash,
4275                right.path.as_slice(),
4276                right.member.tar_member_group_start,
4277            ))
4278    });
4279    rows
4280}
4281
4282fn partition_file_rows(rows: Vec<FileRow>) -> Result<Vec<Vec<FileRow>>, FormatError> {
4283    let mut shards = Vec::new();
4284    let mut start = 0usize;
4285    while start < rows.len() {
4286        let mut end = (start + DEFAULT_FILES_PER_INDEX_SHARD).min(rows.len());
4287        if end < rows.len() && rows[end - 1].path_hash == rows[end].path_hash {
4288            let boundary_hash = rows[end].path_hash;
4289            let mut run_start_in_shard = end - 1;
4290            while run_start_in_shard > start
4291                && rows[run_start_in_shard - 1].path_hash == boundary_hash
4292            {
4293                run_start_in_shard -= 1;
4294            }
4295            let mut full_run_start = run_start_in_shard;
4296            while full_run_start > 0 && rows[full_run_start - 1].path_hash == boundary_hash {
4297                full_run_start -= 1;
4298            }
4299            let mut full_run_end = end + 1;
4300            while full_run_end < rows.len() && rows[full_run_end].path_hash == boundary_hash {
4301                full_run_end += 1;
4302            }
4303            let full_run_len = full_run_end - full_run_start;
4304            end = if full_run_len <= MAX_HASH_PREFIX_RUN_FILES {
4305                full_run_end
4306            } else {
4307                (run_start_in_shard + MAX_HASH_PREFIX_RUN_FILES).min(full_run_end)
4308            };
4309        }
4310        if end - start > MAX_FILES_PER_INDEX_SHARD {
4311            return Err(FormatError::WriterUnsupported(
4312                "hash-prefix collision run exceeds max_files_per_index_shard",
4313            ));
4314        }
4315        shards.push(rows[start..end].to_vec());
4316        start = end;
4317    }
4318    Ok(shards)
4319}
4320
4321fn build_index_shard_plaintexts(
4322    shard_rows: &[Vec<FileRow>],
4323    frames: &[PayloadFrame],
4324    payloads: &[PayloadObject],
4325    options: WriterOptions,
4326) -> Result<Vec<PlannedIndexShard>, FormatError> {
4327    let mut planned = Vec::new();
4328    for rows in shard_rows {
4329        append_index_shards_for_rows(&mut planned, rows.clone(), frames, payloads, options)?;
4330    }
4331    Ok(planned)
4332}
4333
4334fn append_index_shards_for_rows(
4335    planned: &mut Vec<PlannedIndexShard>,
4336    rows: Vec<FileRow>,
4337    frames: &[PayloadFrame],
4338    payloads: &[PayloadObject],
4339    options: WriterOptions,
4340) -> Result<(), FormatError> {
4341    let shard_index =
4342        u64::try_from(planned.len()).map_err(|_| FormatError::WriterUnsupported("shard_index"))?;
4343    let candidate = build_index_shard_plaintext(shard_index, &rows, frames, payloads, options)?;
4344    let compressed =
4345        compress_zstd_frame_with_jobs(&candidate.plaintext, options.zstd_level, options.jobs)?;
4346    if index_object_can_fit(compressed.len(), options)? {
4347        planned.push(candidate);
4348        return Ok(());
4349    }
4350    if rows.len() == 1 {
4351        return Err(FormatError::WriterUnsupported(
4352            "single-file IndexShard exceeds index object limits",
4353        ));
4354    }
4355    let split_at = split_sorted_file_rows_for_object_limit(&rows);
4356    append_index_shards_for_rows(
4357        planned,
4358        rows[..split_at].to_vec(),
4359        frames,
4360        payloads,
4361        options,
4362    )?;
4363    append_index_shards_for_rows(
4364        planned,
4365        rows[split_at..].to_vec(),
4366        frames,
4367        payloads,
4368        options,
4369    )
4370}
4371
4372fn split_sorted_file_rows_for_object_limit(rows: &[FileRow]) -> usize {
4373    let midpoint = rows.len() / 2;
4374    if rows[midpoint - 1].path_hash != rows[midpoint].path_hash {
4375        return midpoint;
4376    }
4377
4378    let boundary_hash = rows[midpoint].path_hash;
4379    let mut left = midpoint;
4380    while left > 0 && rows[left - 1].path_hash == boundary_hash {
4381        left -= 1;
4382    }
4383    let mut right = midpoint;
4384    while right < rows.len() && rows[right].path_hash == boundary_hash {
4385        right += 1;
4386    }
4387
4388    match (left > 0, right < rows.len()) {
4389        (true, true) if midpoint - left <= right - midpoint => left,
4390        (true, true) => right,
4391        (true, false) => left,
4392        (false, true) => right,
4393        (false, false) => midpoint,
4394    }
4395}
4396
4397fn build_index_shard_plaintext(
4398    shard_index: u64,
4399    file_rows: &[FileRow],
4400    frames: &[PayloadFrame],
4401    payloads: &[PayloadObject],
4402    options: WriterOptions,
4403) -> Result<PlannedIndexShard, FormatError> {
4404    let mut string_pool = Vec::new();
4405    let mut file_entries = Vec::with_capacity(file_rows.len());
4406    let mut required_frame_indexes = BTreeSet::new();
4407    for row in file_rows {
4408        let path_offset = u32_len(string_pool.len(), "FileEntry.path_offset")?;
4409        string_pool.extend_from_slice(&row.path);
4410        let (first_frame_index, frame_count) = member_frame_range(row.member_index, frames)?;
4411        for offset in 0..frame_count as u64 {
4412            required_frame_indexes.insert(checked_u64_add(
4413                first_frame_index,
4414                offset,
4415                "FileEntry.frame_count",
4416            )?);
4417        }
4418        file_entries.push(FileEntry {
4419            path_hash: row.path_hash,
4420            path_offset,
4421            path_length: u32_len(row.path.len(), "FileEntry.path_length")?,
4422            first_frame_index,
4423            frame_count,
4424            offset_in_first_frame_plaintext: 0,
4425            tar_member_group_size: row.member.tar_member_group_size,
4426            file_data_size: row.member.file_data_size,
4427            flags: 0,
4428        });
4429    }
4430
4431    let frame_entries = frames
4432        .iter()
4433        .filter(|frame| required_frame_indexes.contains(&frame.frame_index))
4434        .map(|frame| FrameEntry {
4435            frame_index: frame.frame_index,
4436            envelope_index: frame.envelope_index,
4437            offset_in_envelope: frame.offset_in_envelope,
4438            compressed_size: frame.compressed_size,
4439            decompressed_size: frame.decompressed_size,
4440            flags: frame.flags,
4441            tar_stream_offset: frame.tar_stream_offset,
4442        })
4443        .collect::<Vec<_>>();
4444    let required_envelope_indexes = frame_entries
4445        .iter()
4446        .map(|frame| frame.envelope_index)
4447        .collect::<BTreeSet<_>>();
4448    let envelope_entries = payloads
4449        .iter()
4450        .filter(|payload| required_envelope_indexes.contains(&payload.envelope_index))
4451        .map(|payload| {
4452            let (first_frame_index, frame_count) =
4453                envelope_frame_range(payload.envelope_index, frames)?;
4454            Ok(EnvelopeEntry {
4455                envelope_index: payload.envelope_index,
4456                first_block_index: payload.object.first_block_index,
4457                data_block_count: payload.object.data_block_count,
4458                parity_block_count: payload.object.parity_block_count,
4459                encrypted_size: payload.object.encrypted_size,
4460                plaintext_size: payload.plaintext_size,
4461                first_frame_index,
4462                frame_count,
4463            })
4464        })
4465        .collect::<Result<Vec<_>, FormatError>>()?;
4466
4467    let plaintext = serialize_index_shard(
4468        shard_index,
4469        &file_entries,
4470        &frame_entries,
4471        &envelope_entries,
4472        &string_pool,
4473        options,
4474    )?;
4475    let first_path_hash = file_rows
4476        .first()
4477        .ok_or(FormatError::WriterInvariant("empty planned IndexShard"))?
4478        .path_hash;
4479    let last_path_hash = file_rows
4480        .last()
4481        .ok_or(FormatError::WriterInvariant("empty planned IndexShard"))?
4482        .path_hash;
4483    Ok(PlannedIndexShard {
4484        shard_index,
4485        plaintext,
4486        file_count: u32_len(file_rows.len(), "IndexShard.file_count")?,
4487        first_path_hash,
4488        last_path_hash,
4489    })
4490}
4491
4492fn serialize_index_shard(
4493    shard_index: u64,
4494    files: &[FileEntry],
4495    frames: &[FrameEntry],
4496    envelopes: &[EnvelopeEntry],
4497    string_pool: &[u8],
4498    _options: WriterOptions,
4499) -> Result<Vec<u8>, FormatError> {
4500    let mut cursor = INDEX_SHARD_HEADER_LEN;
4501    let file_table_offset = table_offset(files.len(), cursor)?;
4502    cursor = checked_usize_add(cursor, files.len() * FILE_ENTRY_LEN, "IndexShard")?;
4503    let frame_table_offset = table_offset(frames.len(), cursor)?;
4504    cursor = checked_usize_add(cursor, frames.len() * FRAME_ENTRY_LEN, "IndexShard")?;
4505    let envelope_table_offset = table_offset(envelopes.len(), cursor)?;
4506    cursor = checked_usize_add(cursor, envelopes.len() * ENVELOPE_ENTRY_LEN, "IndexShard")?;
4507    let string_pool_offset = table_offset(string_pool.len(), cursor)?;
4508
4509    let header = IndexShardHeader {
4510        version: 1,
4511        shard_index,
4512        file_count: u32_len(files.len(), "IndexShard.file_count")?,
4513        frame_count: u32_len(frames.len(), "IndexShard.frame_count")?,
4514        envelope_count: u32_len(envelopes.len(), "IndexShard.envelope_count")?,
4515        file_table_offset,
4516        frame_table_offset,
4517        envelope_table_offset,
4518        string_pool_offset,
4519        string_pool_size: u32_len(string_pool.len(), "IndexShard.string_pool_size")?,
4520    };
4521
4522    let mut bytes = Vec::with_capacity(cursor + string_pool.len());
4523    bytes.extend_from_slice(&header.to_bytes());
4524    for entry in files {
4525        bytes.extend_from_slice(&entry.to_bytes());
4526    }
4527    for entry in frames {
4528        bytes.extend_from_slice(&entry.to_bytes());
4529    }
4530    for entry in envelopes {
4531        bytes.extend_from_slice(&entry.to_bytes());
4532    }
4533    bytes.extend_from_slice(string_pool);
4534    Ok(bytes)
4535}
4536
4537fn build_directory_hint_plaintexts(
4538    shard_rows: &[Vec<FileRow>],
4539    options: WriterOptions,
4540) -> Result<Vec<PlannedDirectoryHintShard>, FormatError> {
4541    let mut map = BTreeMap::<Vec<u8>, BTreeSet<u32>>::new();
4542    for (shard_row_index, rows) in shard_rows.iter().enumerate() {
4543        let shard_row_index = u32::try_from(shard_row_index)
4544            .map_err(|_| FormatError::WriterUnsupported("directory hint shard row index"))?;
4545        for row in rows {
4546            add_directory_hint_rows(&mut map, shard_row_index, &row.path);
4547        }
4548    }
4549
4550    let rows = map
4551        .into_iter()
4552        .map(|(path, shard_rows)| (hash_prefix(&path), path, shard_rows))
4553        .collect::<Vec<_>>();
4554    let mut rows = rows;
4555    rows.sort_by(|left, right| (left.0, left.1.as_slice()).cmp(&(right.0, right.1.as_slice())));
4556
4557    let mut planned = Vec::new();
4558    for chunk in rows.chunks(DEFAULT_DIRECTORY_HINT_ENTRIES_PER_SHARD) {
4559        append_directory_hint_shards_for_rows(&mut planned, chunk.to_vec(), options)?;
4560    }
4561    Ok(planned)
4562}
4563
4564fn append_directory_hint_shards_for_rows(
4565    planned: &mut Vec<PlannedDirectoryHintShard>,
4566    rows: Vec<([u8; 8], Vec<u8>, BTreeSet<u32>)>,
4567    options: WriterOptions,
4568) -> Result<(), FormatError> {
4569    let hint_shard_index = u64::try_from(planned.len())
4570        .map_err(|_| FormatError::WriterUnsupported("hint_shard_index"))?;
4571    let candidate = build_directory_hint_plaintext(hint_shard_index, &rows)?;
4572    let compressed =
4573        compress_zstd_frame_with_jobs(&candidate.plaintext, options.zstd_level, options.jobs)?;
4574    if index_object_can_fit(compressed.len(), options)? {
4575        planned.push(candidate);
4576        return Ok(());
4577    }
4578    if rows.len() == 1 {
4579        return Err(FormatError::WriterUnsupported(
4580            "single DirectoryHintEntry exceeds index object limits",
4581        ));
4582    }
4583    let split_at = rows.len() / 2;
4584    append_directory_hint_shards_for_rows(planned, rows[..split_at].to_vec(), options)?;
4585    append_directory_hint_shards_for_rows(planned, rows[split_at..].to_vec(), options)
4586}
4587
4588fn add_directory_hint_rows(
4589    map: &mut BTreeMap<Vec<u8>, BTreeSet<u32>>,
4590    shard_row_index: u32,
4591    path: &[u8],
4592) {
4593    map.entry(Vec::new()).or_default().insert(shard_row_index);
4594    let mut cursor = 0usize;
4595    while let Some(position) = path[cursor..].iter().position(|byte| *byte == b'/') {
4596        let slash = cursor + position;
4597        if slash > 0 {
4598            map.entry(path[..slash].to_vec())
4599                .or_default()
4600                .insert(shard_row_index);
4601        }
4602        cursor = slash + 1;
4603    }
4604}
4605
4606fn build_directory_hint_plaintext(
4607    hint_shard_index: u64,
4608    rows: &[([u8; 8], Vec<u8>, BTreeSet<u32>)],
4609) -> Result<PlannedDirectoryHintShard, FormatError> {
4610    let mut entries = Vec::with_capacity(rows.len());
4611    let mut shard_row_indexes = Vec::new();
4612    let mut string_pool = Vec::new();
4613
4614    for (dir_hash, path, shard_rows) in rows {
4615        let path_offset = if path.is_empty() {
4616            0
4617        } else {
4618            u64::try_from(string_pool.len())
4619                .map_err(|_| FormatError::WriterUnsupported("DirectoryHintEntry.path_offset"))?
4620        };
4621        if !path.is_empty() {
4622            string_pool.extend_from_slice(path);
4623        }
4624        let shard_list_start_index = u32_len(
4625            shard_row_indexes.len(),
4626            "DirectoryHintEntry.shard_list_start_index",
4627        )?;
4628        shard_row_indexes.extend(shard_rows.iter().copied());
4629        entries.push(DirectoryHintEntry {
4630            dir_hash: *dir_hash,
4631            path_offset,
4632            path_length: u32_len(path.len(), "DirectoryHintEntry.path_length")?,
4633            shard_list_start_index,
4634            shard_count: u32_len(shard_rows.len(), "DirectoryHintEntry.shard_count")?,
4635        });
4636    }
4637
4638    let plaintext = serialize_directory_hint_table(
4639        hint_shard_index,
4640        &entries,
4641        &shard_row_indexes,
4642        &string_pool,
4643    )?;
4644    let first_dir_hash = rows
4645        .first()
4646        .ok_or(FormatError::WriterInvariant("empty directory hint shard"))?
4647        .0;
4648    let last_dir_hash = rows
4649        .last()
4650        .ok_or(FormatError::WriterInvariant("empty directory hint shard"))?
4651        .0;
4652    Ok(PlannedDirectoryHintShard {
4653        hint_shard_index,
4654        plaintext,
4655        entry_count: rows.len() as u64,
4656        first_dir_hash,
4657        last_dir_hash,
4658    })
4659}
4660
4661fn serialize_directory_hint_table(
4662    hint_shard_index: u64,
4663    entries: &[DirectoryHintEntry],
4664    shard_row_indexes: &[u32],
4665    string_pool: &[u8],
4666) -> Result<Vec<u8>, FormatError> {
4667    let entry_table_offset = table_offset(entries.len(), DIRECTORY_HINT_TABLE_LEN)?;
4668    let shard_list_cursor = checked_usize_add(
4669        DIRECTORY_HINT_TABLE_LEN,
4670        entries.len() * DIRECTORY_HINT_ENTRY_LEN,
4671        "DirectoryHintTable",
4672    )?;
4673    let shard_list_offset = table_offset(shard_row_indexes.len(), shard_list_cursor)?;
4674    let string_pool_cursor = checked_usize_add(
4675        shard_list_cursor,
4676        shard_row_indexes.len() * 4,
4677        "DirectoryHintTable",
4678    )?;
4679    let string_pool_offset = if string_pool.is_empty() {
4680        0
4681    } else {
4682        u64::try_from(string_pool_cursor)
4683            .map_err(|_| FormatError::WriterUnsupported("DirectoryHintTable.string_pool_offset"))?
4684    };
4685    let header = DirectoryHintTableHeader {
4686        version: 1,
4687        hint_shard_index,
4688        entry_count: entries.len() as u64,
4689        entry_table_offset: entry_table_offset as u64,
4690        shard_list_offset: shard_list_offset as u64,
4691        string_pool_offset,
4692        string_pool_size: string_pool.len() as u64,
4693    };
4694
4695    let mut bytes = Vec::with_capacity(string_pool_cursor + string_pool.len());
4696    bytes.extend_from_slice(&header.to_bytes());
4697    for entry in entries {
4698        bytes.extend_from_slice(&entry.to_bytes());
4699    }
4700    for row in shard_row_indexes {
4701        bytes.extend_from_slice(&row.to_le_bytes());
4702    }
4703    bytes.extend_from_slice(string_pool);
4704    Ok(bytes)
4705}
4706
4707#[derive(Debug, Clone, Copy)]
4708struct IndexRootPlaintextInput<'a> {
4709    shard_entries: &'a [ShardEntry],
4710    frame_count: u64,
4711    envelope_count: u64,
4712    file_count: u64,
4713    payload_block_count: u64,
4714    tar_total_size: u64,
4715    content_sha256: [u8; 32],
4716    directory_hint_entries: &'a [DirectoryHintShardEntry],
4717    dictionary_extent: Option<(ObjectExtent, u32)>,
4718}
4719
4720fn build_index_root_plaintext(input: IndexRootPlaintextInput<'_>) -> Vec<u8> {
4721    let mut header = IndexRootHeader::empty();
4722    header.frame_count = input.frame_count;
4723    header.envelope_count = input.envelope_count;
4724    header.file_count = input.file_count;
4725    header.payload_block_count = input.payload_block_count;
4726    header.tar_total_size = input.tar_total_size;
4727    header.content_sha256 = input.content_sha256;
4728    if let Some((dictionary, decompressed_size)) = input.dictionary_extent {
4729        header.dictionary_first_block = dictionary.first_block_index;
4730        header.dictionary_data_block_count = dictionary.data_block_count;
4731        header.dictionary_parity_block_count = dictionary.parity_block_count;
4732        header.dictionary_encrypted_size = dictionary.encrypted_size;
4733        header.dictionary_decompressed_size = decompressed_size;
4734    }
4735    let root = IndexRoot {
4736        header,
4737        shards: input.shard_entries.to_vec(),
4738        directory_hint_shards: input.directory_hint_entries.to_vec(),
4739    };
4740    root.to_bytes()
4741}
4742
4743#[derive(Debug, Clone, Copy, PartialEq, Eq)]
4744struct MetadataClassPlan {
4745    options: WriterOptions,
4746    index_root: PlannedEncryptedObject,
4747    dictionary: Option<PlannedEncryptedObject>,
4748}
4749
4750fn plan_index_root_metadata_class(
4751    mut options: WriterOptions,
4752    compressed_index_root_len: usize,
4753    compressed_dictionary_len: Option<usize>,
4754) -> Result<MetadataClassPlan, FormatError> {
4755    let index_root = plan_metadata_object_without_class(
4756        compressed_index_root_len,
4757        options,
4758        MetadataObjectKind::IndexRoot,
4759    )?;
4760    let dictionary = compressed_dictionary_len
4761        .map(|len| plan_metadata_object_without_class(len, options, MetadataObjectKind::Dictionary))
4762        .transpose()?;
4763    let required_data_shards = u32::from(options.index_root_fec_data_shards)
4764        .max(MIN_INDEX_ROOT_FEC_DATA_SHARDS as u32)
4765        .max(index_root.data_block_count)
4766        .max(dictionary.map(|plan| plan.data_block_count).unwrap_or(0));
4767    let required_data_shards = u16::try_from(required_data_shards)
4768        .map_err(|_| MetadataObjectKind::IndexRoot.too_large_error())?;
4769    options.index_root_fec_data_shards = required_data_shards;
4770    let required_parity_shards = compute_parity_u16(
4771        options.index_root_fec_data_shards as u64,
4772        options,
4773        "index_root_fec_parity_shards",
4774    )?;
4775    options.index_root_fec_parity_shards = options
4776        .index_root_fec_parity_shards
4777        .max(required_parity_shards);
4778    ensure_metadata_object_fits_class(index_root, options, MetadataObjectKind::IndexRoot)?;
4779    if let Some(dictionary) = dictionary {
4780        ensure_metadata_object_fits_class(dictionary, options, MetadataObjectKind::Dictionary)?;
4781    }
4782    Ok(MetadataClassPlan {
4783        options,
4784        index_root,
4785        dictionary,
4786    })
4787}
4788
4789fn plan_metadata_object_without_class(
4790    payload_len: usize,
4791    options: WriterOptions,
4792    kind: MetadataObjectKind,
4793) -> Result<PlannedEncryptedObject, FormatError> {
4794    let plan = plan_encrypted_object_without_class(payload_len, options)
4795        .map_err(|_| kind.too_large_error())?;
4796    if plan.data_block_count > u16::MAX as u32 || plan.parity_block_count > u16::MAX as u32 {
4797        return Err(kind.too_large_error());
4798    }
4799    validate_object_shard_total(plan.data_block_count, plan.parity_block_count)
4800        .map_err(|_| kind.too_large_error())?;
4801    Ok(plan)
4802}
4803
4804fn ensure_metadata_object_fits_class(
4805    plan: PlannedEncryptedObject,
4806    options: WriterOptions,
4807    kind: MetadataObjectKind,
4808) -> Result<(), FormatError> {
4809    if plan.data_block_count > options.index_root_fec_data_shards as u32 {
4810        return Err(kind.too_large_error());
4811    }
4812    if plan.parity_block_count > options.index_root_fec_parity_shards as u32 {
4813        return Err(kind.too_large_error());
4814    }
4815    validate_object_shard_total(plan.data_block_count, plan.parity_block_count)
4816        .map_err(|_| kind.too_large_error())
4817}
4818
4819fn payload_object_can_fit(payload_len: usize, options: WriterOptions) -> Result<bool, FormatError> {
4820    encrypted_object_can_fit(
4821        payload_len,
4822        options.fec_data_shards,
4823        options.fec_parity_shards,
4824        options,
4825    )
4826}
4827
4828fn index_object_can_fit(payload_len: usize, options: WriterOptions) -> Result<bool, FormatError> {
4829    encrypted_object_can_fit(
4830        payload_len,
4831        options.index_fec_data_shards,
4832        options.index_fec_parity_shards,
4833        options,
4834    )
4835}
4836
4837fn encrypted_object_can_fit(
4838    payload_len: usize,
4839    data_shard_max: u16,
4840    parity_shard_max: u16,
4841    options: WriterOptions,
4842) -> Result<bool, FormatError> {
4843    match plan_encrypted_object(payload_len, data_shard_max, parity_shard_max, options) {
4844        Ok(_) => Ok(true),
4845        Err(FormatError::WriterUnsupported("encrypted object exceeds u32 size limit"))
4846        | Err(FormatError::WriterUnsupported(
4847            "encrypted object exceeds its data shard class maximum",
4848        ))
4849        | Err(FormatError::WriterUnsupported(
4850            "encrypted object exceeds its parity shard class maximum",
4851        ))
4852        | Err(FormatError::WriterUnsupported(
4853            "encrypted object exceeds ReedSolomonGF16 shard limit",
4854        )) => Ok(false),
4855        Err(err) => Err(err),
4856    }
4857}
4858
4859fn plan_encrypted_object(
4860    payload_len: usize,
4861    data_shard_max: u16,
4862    parity_shard_max: u16,
4863    options: WriterOptions,
4864) -> Result<PlannedEncryptedObject, FormatError> {
4865    let plan = plan_encrypted_object_without_class(payload_len, options)?;
4866    if plan.data_block_count > data_shard_max as u32 {
4867        return Err(FormatError::WriterUnsupported(
4868            "encrypted object exceeds its data shard class maximum",
4869        ));
4870    }
4871    if plan.parity_block_count > parity_shard_max as u32 {
4872        return Err(FormatError::WriterUnsupported(
4873            "encrypted object exceeds its parity shard class maximum",
4874        ));
4875    }
4876    validate_object_shard_total(plan.data_block_count, plan.parity_block_count)?;
4877    Ok(plan)
4878}
4879
4880fn plan_encrypted_object_without_class(
4881    payload_len: usize,
4882    options: WriterOptions,
4883) -> Result<PlannedEncryptedObject, FormatError> {
4884    let (data_block_count, encrypted_size) = encrypted_object_data_extent(payload_len, options)?;
4885    let parity_block_count = compute_parity(data_block_count as u64, options)?;
4886    Ok(PlannedEncryptedObject {
4887        data_block_count,
4888        parity_block_count,
4889        encrypted_size,
4890    })
4891}
4892
4893fn encrypted_object_data_extent(
4894    payload_len: usize,
4895    options: WriterOptions,
4896) -> Result<(u32, u32), FormatError> {
4897    let block_size = options.block_size as usize;
4898    let tag_len = options.aead_algo.tag_len();
4899    let total_before_padding =
4900        payload_len
4901            .checked_add(tag_len)
4902            .ok_or(FormatError::WriterUnsupported(
4903                "encrypted object size overflow",
4904            ))?;
4905    let remainder = total_before_padding % block_size;
4906    let encrypted_size = if remainder == 0 {
4907        total_before_padding
4908            .checked_add(block_size)
4909            .ok_or(FormatError::WriterUnsupported(
4910                "encrypted object size overflow",
4911            ))?
4912    } else {
4913        total_before_padding
4914            .checked_add(block_size - remainder)
4915            .ok_or(FormatError::WriterUnsupported(
4916                "encrypted object size overflow",
4917            ))?
4918    };
4919    let encrypted_size = u32_len(encrypted_size, "encrypted_size")
4920        .map_err(|_| FormatError::WriterUnsupported("encrypted object exceeds u32 size limit"))?;
4921    Ok((encrypted_size / options.block_size, encrypted_size))
4922}
4923
4924#[derive(Debug, Clone, Copy)]
4925struct ObjectEncryptionContext<'a> {
4926    key: &'a [u8; 32],
4927    nonce_seed: &'a [u8; 32],
4928    domain: &'a [u8],
4929    counter: u64,
4930    data_kind: BlockKind,
4931    parity_kind: BlockKind,
4932    data_shard_max: u16,
4933    class_parity_shard_max: u16,
4934    archive_uuid: &'a [u8; 16],
4935    session_id: &'a [u8; 16],
4936}
4937
4938fn encrypt_object(
4939    payload: &[u8],
4940    context: ObjectEncryptionContext<'_>,
4941    next_block_index: &mut u64,
4942    options: WriterOptions,
4943) -> Result<EncryptedObject, FormatError> {
4944    let block_size = options.block_size as usize;
4945    let padded = suffix_pad_for_aead(payload, options.aead_algo.tag_len(), block_size)?;
4946    let nonce = derive_nonce(
4947        context.nonce_seed,
4948        context.domain,
4949        context.archive_uuid,
4950        context.session_id,
4951        context.counter,
4952        options.aead_algo.nonce_len(),
4953    )?;
4954    let aad = build_aad(
4955        context.domain,
4956        context.archive_uuid,
4957        context.session_id,
4958        context.counter,
4959    )?;
4960    let encrypted = aead_encrypt(options.aead_algo, context.key, &nonce, &aad, &padded)?;
4961    if encrypted.len() % block_size != 0 {
4962        return Err(FormatError::WriterInvariant(
4963            "encrypted object is not block aligned",
4964        ));
4965    }
4966    let encrypted_size = u32_len(encrypted.len(), "encrypted_size")?;
4967    let data_shards = encrypted
4968        .chunks(block_size)
4969        .map(|chunk| chunk.to_vec())
4970        .collect::<Vec<_>>();
4971    let data_block_count = u32_len(data_shards.len(), "data_block_count")?;
4972    if data_block_count == 0 {
4973        return Err(FormatError::WriterInvariant(
4974            "encrypted object has no data blocks",
4975        ));
4976    }
4977    if data_block_count > context.data_shard_max as u32 {
4978        return Err(FormatError::WriterUnsupported(
4979            "encrypted object exceeds its data shard class maximum",
4980        ));
4981    }
4982    let required_parity = compute_object_parity(
4983        data_block_count as u64,
4984        options,
4985        context.class_parity_shard_max as u32,
4986    )?;
4987    if required_parity > context.class_parity_shard_max as u32 {
4988        return Err(FormatError::WriterUnsupported(
4989            "encrypted object exceeds its parity shard class maximum",
4990        ));
4991    }
4992    validate_object_shard_total(data_block_count, required_parity)?;
4993    let parity_count = required_parity as u16;
4994    let parity_shards = if parity_count == 0 {
4995        Vec::new()
4996    } else {
4997        encode_parity_gf16(&data_shards, parity_count as usize)?
4998    };
4999
5000    let first_block_index = *next_block_index;
5001    let mut records = Vec::with_capacity(data_shards.len() + parity_shards.len());
5002    for (index, payload) in data_shards.into_iter().enumerate() {
5003        records.push(BlockRecord {
5004            block_index: checked_u64_add(first_block_index, index as u64, "BlockRecord")?,
5005            kind: context.data_kind,
5006            flags: if index + 1 == data_block_count as usize {
5007                0x01
5008            } else {
5009                0
5010            },
5011            payload,
5012            record_crc32c: 0,
5013        });
5014    }
5015    let parity_first_block = checked_u64_add(first_block_index, data_block_count as u64, "FEC")?;
5016    for (index, payload) in parity_shards.into_iter().enumerate() {
5017        records.push(BlockRecord {
5018            block_index: checked_u64_add(parity_first_block, index as u64, "BlockRecord")?,
5019            kind: context.parity_kind,
5020            flags: 0,
5021            payload,
5022            record_crc32c: 0,
5023        });
5024    }
5025
5026    *next_block_index = checked_u64_add(
5027        first_block_index,
5028        data_block_count as u64 + parity_count as u64,
5029        "next_block_index",
5030    )?;
5031
5032    Ok(EncryptedObject {
5033        first_block_index,
5034        data_block_count,
5035        parity_block_count: parity_count as u32,
5036        encrypted_size,
5037        records,
5038    })
5039}
5040
5041fn validate_planned_object(
5042    object: &EncryptedObject,
5043    expected: PlannedEncryptedObject,
5044) -> Result<(), FormatError> {
5045    if object.data_block_count != expected.data_block_count
5046        || object.parity_block_count != expected.parity_block_count
5047        || object.encrypted_size != expected.encrypted_size
5048    {
5049        return Err(FormatError::WriterInvariant(
5050            "encrypted object did not match planned sizing",
5051        ));
5052    }
5053    Ok(())
5054}
5055
5056fn validate_planned_extent(
5057    object: &EncryptedObject,
5058    expected: ObjectExtent,
5059) -> Result<(), FormatError> {
5060    validate_planned_object(
5061        object,
5062        PlannedEncryptedObject {
5063            data_block_count: expected.data_block_count,
5064            parity_block_count: expected.parity_block_count,
5065            encrypted_size: expected.encrypted_size,
5066        },
5067    )?;
5068    if object.first_block_index != expected.first_block_index {
5069        return Err(FormatError::WriterInvariant(
5070            "encrypted object did not match planned extent",
5071        ));
5072    }
5073    Ok(())
5074}
5075
5076fn map_metadata_encrypt_error(error: FormatError, kind: MetadataObjectKind) -> FormatError {
5077    match error {
5078        FormatError::WriterUnsupported("encrypted object exceeds u32 size limit")
5079        | FormatError::WriterUnsupported("encrypted object exceeds its data shard class maximum")
5080        | FormatError::WriterUnsupported(
5081            "encrypted object exceeds its parity shard class maximum",
5082        )
5083        | FormatError::WriterUnsupported("encrypted object exceeds ReedSolomonGF16 shard limit") => {
5084            kind.too_large_error()
5085        }
5086        other => other,
5087    }
5088}
5089
5090#[derive(Debug, Clone, Copy)]
5091struct RootAuthFooterBuildInput<'a> {
5092    archive_uuid: [u8; 16],
5093    session_id: [u8; 16],
5094    options: WriterOptions,
5095    crypto_header: &'a [u8],
5096    volume_zero_manifest: &'a [u8; MANIFEST_FOOTER_LEN],
5097    index_root_plaintext: &'a [u8],
5098    index_root_extent: ObjectExtent,
5099    dictionary_extent: Option<(ObjectExtent, u32)>,
5100    shard_entries: &'a [ShardEntry],
5101    payload_objects: &'a [PayloadObject],
5102    directory_hint_entries: &'a [DirectoryHintShardEntry],
5103    data_leaf_hashes: &'a [(u64, [u8; 32])],
5104}
5105
5106fn build_root_auth_footer_from_leaf_hashes(
5107    config: RootAuthWriterConfig<'_>,
5108    authenticator: &mut RootAuthAuthenticator<'_>,
5109    input: RootAuthFooterBuildInput<'_>,
5110) -> Result<Vec<u8>, FormatError> {
5111    let mut sorted_leaf_hashes = input.data_leaf_hashes.to_vec();
5112    sorted_leaf_hashes.sort_by_key(|(block_index, _)| *block_index);
5113    let leaf_hashes = sorted_leaf_hashes
5114        .iter()
5115        .map(|(_, leaf_hash)| *leaf_hash)
5116        .collect::<Vec<_>>();
5117    let total_data_block_count = u64::try_from(leaf_hashes.len())
5118        .map_err(|_| FormatError::WriterUnsupported("root-auth data block count"))?;
5119    let data_block_merkle_root = data_block_merkle_root_from_leaf_hashes(&leaf_hashes);
5120
5121    let parsed_crypto = CryptoHeader::parse(
5122        input.crypto_header,
5123        u32_len(input.crypto_header.len(), "CryptoHeader")?,
5124    )?;
5125    let footer_length = root_auth_footer_wire_length(
5126        config.signer_identity.len(),
5127        config.authenticator_value_length as usize,
5128    )?;
5129    let root_auth_descriptor_digest = root_auth_descriptor_digest(
5130        config.authenticator_id,
5131        config.signer_identity_type,
5132        config.signer_identity,
5133        config.authenticator_value_length,
5134        footer_length,
5135    )?;
5136    let signer_identity_digest =
5137        signer_identity_digest(config.signer_identity_type, config.signer_identity)?;
5138    let manifest_pre_hmac = manifest_footer_global_pre_hmac_bytes(input.volume_zero_manifest);
5139    let critical_metadata_digest = critical_metadata_digest(CriticalMetadataDigestInputs {
5140        archive_uuid: input.archive_uuid,
5141        session_id: input.session_id,
5142        stripe_width: input.options.stripe_width,
5143        total_volumes: input.options.stripe_width,
5144        compression_algo: parsed_crypto.fixed.compression_algo,
5145        aead_algo: parsed_crypto.fixed.aead_algo,
5146        fec_algo: parsed_crypto.fixed.fec_algo,
5147        kdf_algo: parsed_crypto.fixed.kdf_algo,
5148        crypto_header_pre_hmac_bytes: parsed_crypto.hmac_covered_bytes,
5149        chunk_size: parsed_crypto.fixed.chunk_size,
5150        envelope_target_size: parsed_crypto.fixed.envelope_target_size,
5151        block_size: parsed_crypto.fixed.block_size,
5152        fec_data_shards: parsed_crypto.fixed.fec_data_shards,
5153        fec_parity_shards: parsed_crypto.fixed.fec_parity_shards,
5154        index_fec_data_shards: parsed_crypto.fixed.index_fec_data_shards,
5155        index_fec_parity_shards: parsed_crypto.fixed.index_fec_parity_shards,
5156        index_root_fec_data_shards: parsed_crypto.fixed.index_root_fec_data_shards,
5157        index_root_fec_parity_shards: parsed_crypto.fixed.index_root_fec_parity_shards,
5158        volume_loss_tolerance: parsed_crypto.fixed.volume_loss_tolerance,
5159        bit_rot_buffer_pct: parsed_crypto.fixed.bit_rot_buffer_pct,
5160        has_dictionary: parsed_crypto.fixed.has_dictionary,
5161        manifest_footer_global_pre_hmac_bytes: &manifest_pre_hmac,
5162        index_root_first_block: input.index_root_extent.first_block_index,
5163        index_root_data_block_count: input.index_root_extent.data_block_count,
5164        index_root_parity_block_count: input.index_root_extent.parity_block_count,
5165        index_root_encrypted_size: input.index_root_extent.encrypted_size,
5166        index_root_decompressed_size: u32_len(input.index_root_plaintext.len(), "IndexRoot")?,
5167        root_auth_descriptor_digest,
5168    })?;
5169    let index_digest = index_digest(input.index_root_plaintext);
5170    let fec_layout_rows = writer_fec_layout_rows_from_extents(
5171        input.index_root_extent,
5172        u32_len(input.index_root_plaintext.len(), "IndexRoot")?,
5173        input.dictionary_extent,
5174        input.shard_entries,
5175        input.payload_objects,
5176        input.directory_hint_entries,
5177    );
5178    let expected_data_block_count = fec_layout_rows.iter().try_fold(0u64, |total, row| {
5179        if row.present {
5180            checked_u64_add(
5181                total,
5182                row.data_block_count as u64,
5183                "root-auth data block count",
5184            )
5185        } else {
5186            Ok(total)
5187        }
5188    })?;
5189    if expected_data_block_count != total_data_block_count {
5190        return Err(FormatError::WriterInvariant(
5191            "root-auth data block count does not match FEC layout",
5192        ));
5193    }
5194    let fec_layout_digest = fec_layout_digest(&fec_layout_rows)?;
5195    let archive_root = archive_root(ArchiveRootInputs {
5196        archive_uuid: input.archive_uuid,
5197        session_id: input.session_id,
5198        format_version: FORMAT_VERSION,
5199        volume_format_rev: VOLUME_FORMAT_REV,
5200        compression_algo: parsed_crypto.fixed.compression_algo,
5201        aead_algo: parsed_crypto.fixed.aead_algo,
5202        fec_algo: parsed_crypto.fixed.fec_algo,
5203        kdf_algo: parsed_crypto.fixed.kdf_algo,
5204        critical_metadata_digest,
5205        index_digest,
5206        fec_layout_digest,
5207        total_data_block_count,
5208        data_block_merkle_root,
5209        root_auth_descriptor_digest,
5210        signer_identity_digest,
5211    });
5212    let authenticator_value = authenticator(&RootAuthSigningRequest {
5213        archive_uuid: input.archive_uuid,
5214        session_id: input.session_id,
5215        archive_root,
5216    })?;
5217    if authenticator_value.len() != config.authenticator_value_length as usize {
5218        return Err(FormatError::WriterUnsupported(
5219            "root-auth authenticator length mismatch",
5220        ));
5221    }
5222
5223    RootAuthFooterV1 {
5224        archive_uuid: input.archive_uuid,
5225        session_id: input.session_id,
5226        authenticator_id: config.authenticator_id,
5227        signer_identity_type: config.signer_identity_type,
5228        signer_identity_bytes: config.signer_identity.to_vec(),
5229        authenticator_value,
5230        total_data_block_count,
5231        critical_metadata_digest,
5232        index_digest,
5233        fec_layout_digest,
5234        data_block_merkle_root,
5235        signer_identity_digest,
5236        archive_root,
5237        footer_crc32c: 0,
5238    }
5239    .to_bytes()
5240}
5241
5242fn writer_fec_layout_rows_from_extents(
5243    index_root_extent: ObjectExtent,
5244    index_root_plain_size: u32,
5245    dictionary_extent: Option<(ObjectExtent, u32)>,
5246    shard_entries: &[ShardEntry],
5247    payload_objects: &[PayloadObject],
5248    directory_hint_entries: &[DirectoryHintShardEntry],
5249) -> Vec<FecLayoutObjectRow> {
5250    let mut rows = Vec::new();
5251    rows.push(FecLayoutObjectRow {
5252        object_class: 1,
5253        present: true,
5254        object_id: 0,
5255        first_block_index: index_root_extent.first_block_index,
5256        data_block_count: index_root_extent.data_block_count,
5257        parity_block_count: index_root_extent.parity_block_count,
5258        encrypted_size: index_root_extent.encrypted_size,
5259        plain_size: index_root_plain_size,
5260    });
5261    if let Some((dictionary, decompressed_size)) = dictionary_extent {
5262        rows.push(FecLayoutObjectRow {
5263            object_class: 2,
5264            present: true,
5265            object_id: 0,
5266            first_block_index: dictionary.first_block_index,
5267            data_block_count: dictionary.data_block_count,
5268            parity_block_count: dictionary.parity_block_count,
5269            encrypted_size: dictionary.encrypted_size,
5270            plain_size: decompressed_size,
5271        });
5272    } else {
5273        rows.push(FecLayoutObjectRow {
5274            object_class: 2,
5275            present: false,
5276            object_id: 0,
5277            first_block_index: 0,
5278            data_block_count: 0,
5279            parity_block_count: 0,
5280            encrypted_size: 0,
5281            plain_size: 0,
5282        });
5283    }
5284    for entry in shard_entries {
5285        rows.push(FecLayoutObjectRow {
5286            object_class: 3,
5287            present: true,
5288            object_id: entry.shard_index,
5289            first_block_index: entry.first_block_index,
5290            data_block_count: entry.data_block_count,
5291            parity_block_count: entry.parity_block_count,
5292            encrypted_size: entry.encrypted_size,
5293            plain_size: entry.decompressed_size,
5294        });
5295    }
5296    for payload in payload_objects {
5297        rows.push(FecLayoutObjectRow {
5298            object_class: 4,
5299            present: true,
5300            object_id: payload.envelope_index,
5301            first_block_index: payload.object.first_block_index,
5302            data_block_count: payload.object.data_block_count,
5303            parity_block_count: payload.object.parity_block_count,
5304            encrypted_size: payload.object.encrypted_size,
5305            plain_size: payload.plaintext_size,
5306        });
5307    }
5308    for entry in directory_hint_entries {
5309        rows.push(FecLayoutObjectRow {
5310            object_class: 5,
5311            present: true,
5312            object_id: entry.hint_shard_index,
5313            first_block_index: entry.first_block_index,
5314            data_block_count: entry.data_block_count,
5315            parity_block_count: entry.parity_block_count,
5316            encrypted_size: entry.encrypted_size,
5317            plain_size: entry.decompressed_size,
5318        });
5319    }
5320    rows
5321}
5322
5323fn manifest_footer_global_pre_hmac_bytes(manifest_footer: &[u8; MANIFEST_FOOTER_LEN]) -> [u8; 104] {
5324    let mut bytes = [0u8; 104];
5325    bytes.copy_from_slice(&manifest_footer[..104]);
5326    bytes[36..40].fill(0);
5327    bytes
5328}
5329
5330fn root_auth_footer_wire_length(
5331    signer_identity_len: usize,
5332    authenticator_value_len: usize,
5333) -> Result<u32, FormatError> {
5334    validate_root_auth_variable_lengths_for_writer(signer_identity_len, authenticator_value_len)?;
5335    let len = crate::format::ROOT_AUTH_FOOTER_FIXED_LEN
5336        .checked_add(signer_identity_len)
5337        .and_then(|value| value.checked_add(authenticator_value_len))
5338        .and_then(|value| value.checked_add(4))
5339        .ok_or(FormatError::WriterUnsupported(
5340            "RootAuthFooterV1 length overflow",
5341        ))?;
5342    if len > READER_MAX_ROOT_AUTH_FOOTER_LEN as usize {
5343        return Err(FormatError::ReaderResourceLimitExceeded {
5344            field: "RootAuthFooterV1 length",
5345            cap: READER_MAX_ROOT_AUTH_FOOTER_LEN as u64,
5346            actual: len as u64,
5347        });
5348    }
5349    u32::try_from(len).map_err(|_| FormatError::WriterUnsupported("RootAuthFooterV1 length"))
5350}
5351
5352fn validate_root_auth_writer_config(config: RootAuthWriterConfig<'_>) -> Result<(), FormatError> {
5353    root_auth_footer_wire_length(
5354        config.signer_identity.len(),
5355        config.authenticator_value_length as usize,
5356    )?;
5357    Ok(())
5358}
5359
5360fn validate_root_auth_variable_lengths_for_writer(
5361    signer_identity_len: usize,
5362    authenticator_value_len: usize,
5363) -> Result<(), FormatError> {
5364    if signer_identity_len > READER_MAX_ROOT_AUTH_SIGNER_IDENTITY_LEN as usize {
5365        return Err(FormatError::ReaderResourceLimitExceeded {
5366            field: "RootAuthFooterV1 signer identity length",
5367            cap: READER_MAX_ROOT_AUTH_SIGNER_IDENTITY_LEN as u64,
5368            actual: signer_identity_len as u64,
5369        });
5370    }
5371    if authenticator_value_len > READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN as usize {
5372        return Err(FormatError::ReaderResourceLimitExceeded {
5373            field: "RootAuthFooterV1 authenticator value length",
5374            cap: READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN as u64,
5375            actual: authenticator_value_len as u64,
5376        });
5377    }
5378    Ok(())
5379}
5380
5381#[allow(clippy::too_many_arguments)]
5382fn build_manifest_footer(
5383    subkeys: &Subkeys,
5384    aead_algo: AeadAlgo,
5385    archive_uuid: [u8; 16],
5386    session_id: [u8; 16],
5387    volume_index: u32,
5388    total_volumes: u32,
5389    index_root_extent: &ObjectExtent,
5390    index_root_decompressed_size: usize,
5391) -> Result<[u8; MANIFEST_FOOTER_LEN], FormatError> {
5392    let mut footer = ManifestFooter {
5393        archive_uuid,
5394        session_id,
5395        volume_index,
5396        is_authoritative: 1,
5397        total_volumes,
5398        index_root_first_block: index_root_extent.first_block_index,
5399        index_root_data_block_count: index_root_extent.data_block_count,
5400        index_root_parity_block_count: index_root_extent.parity_block_count,
5401        index_root_encrypted_size: index_root_extent.encrypted_size,
5402        index_root_decompressed_size: u32_len(index_root_decompressed_size, "IndexRoot")?,
5403        manifest_hmac: [0u8; 32],
5404    };
5405    let mut bytes = footer.to_bytes();
5406    footer.manifest_hmac = compute_integrity_tag(
5407        HmacDomain::ManifestFooter,
5408        aead_algo,
5409        Some(&subkeys.mac_key),
5410        &archive_uuid,
5411        &session_id,
5412        &bytes[..104],
5413    )?;
5414    bytes = footer.to_bytes();
5415    Ok(bytes)
5416}
5417
5418#[derive(Debug, Clone, Copy)]
5419struct VolumeTrailerBuildInput<'a> {
5420    subkeys: &'a Subkeys,
5421    aead_algo: AeadAlgo,
5422    archive_uuid: [u8; 16],
5423    session_id: [u8; 16],
5424    volume_index: u32,
5425    block_count: u64,
5426    bytes_written: u64,
5427    manifest_footer_offset: u64,
5428    closed_at_ns: i64,
5429    root_auth_footer: Option<(u64, u32)>,
5430}
5431
5432fn build_volume_trailer(
5433    input: VolumeTrailerBuildInput<'_>,
5434) -> Result<[u8; VOLUME_TRAILER_LEN], FormatError> {
5435    let (root_auth_footer_offset, root_auth_footer_length, root_auth_flags) =
5436        match input.root_auth_footer {
5437            Some((offset, length)) => (offset, length, 0x0000_0001),
5438            None => (0, 0, 0),
5439        };
5440    let mut trailer = VolumeTrailer {
5441        archive_uuid: input.archive_uuid,
5442        session_id: input.session_id,
5443        volume_index: input.volume_index,
5444        block_count: input.block_count,
5445        bytes_written: input.bytes_written,
5446        manifest_footer_offset: input.manifest_footer_offset,
5447        manifest_footer_length: MANIFEST_FOOTER_LEN as u32,
5448        closed_at_ns: input.closed_at_ns,
5449        root_auth_footer_offset,
5450        root_auth_footer_length,
5451        root_auth_flags,
5452        trailer_hmac: [0u8; 32],
5453    };
5454    let mut bytes = trailer.to_bytes();
5455    trailer.trailer_hmac = compute_integrity_tag(
5456        HmacDomain::VolumeTrailer,
5457        input.aead_algo,
5458        Some(&input.subkeys.mac_key),
5459        &input.archive_uuid,
5460        &input.session_id,
5461        &bytes[..96],
5462    )?;
5463    bytes = trailer.to_bytes();
5464    Ok(bytes)
5465}
5466
5467struct BuiltCmra {
5468    bytes: Vec<u8>,
5469    shard_size: u32,
5470    data_shard_count: u16,
5471    parity_shard_count: u16,
5472    image_length: u32,
5473    image_sha256: [u8; 32],
5474}
5475
5476#[derive(Debug, Clone, Copy)]
5477struct CmraBuildInput<'a> {
5478    volume_header_bytes: &'a [u8; VOLUME_HEADER_LEN],
5479    crypto_header: &'a [u8],
5480    block_count: u64,
5481    manifest_footer_offset: u64,
5482    manifest_footer: &'a [u8; MANIFEST_FOOTER_LEN],
5483    root_auth_footer_offset: Option<u64>,
5484    root_auth_footer: Option<&'a [u8]>,
5485    trailer_offset: u64,
5486    trailer: &'a [u8; VOLUME_TRAILER_LEN],
5487    cmra_offset: u64,
5488    options: WriterOptions,
5489    archive_uuid: [u8; 16],
5490    session_id: [u8; 16],
5491    volume_index: u32,
5492}
5493
5494fn build_v41_cmra(input: CmraBuildInput<'_>) -> Result<BuiltCmra, FormatError> {
5495    let block_record_len = input.options.block_size as u64 + BLOCK_RECORD_FRAMING_LEN as u64;
5496    let block_records_offset = VOLUME_HEADER_LEN as u64 + input.crypto_header.len() as u64;
5497    let block_records_length = checked_u64_mul(
5498        input.block_count,
5499        block_record_len,
5500        "CMRA BlockRecord length overflow",
5501    )?;
5502    let manifest_end = input
5503        .manifest_footer_offset
5504        .checked_add(MANIFEST_FOOTER_LEN as u64)
5505        .ok_or(FormatError::WriterUnsupported("CMRA terminal overflow"))?;
5506    let root_auth_footer_length = input
5507        .root_auth_footer
5508        .map(|footer| u32_len(footer.len(), "RootAuthFooterV1"))
5509        .transpose()?;
5510    match (input.root_auth_footer_offset, root_auth_footer_length) {
5511        (Some(offset), Some(length)) => {
5512            if manifest_end != offset
5513                || offset
5514                    .checked_add(length as u64)
5515                    .ok_or(FormatError::WriterUnsupported("CMRA terminal overflow"))?
5516                    != input.trailer_offset
5517            {
5518                return Err(FormatError::WriterInvariant(
5519                    "RootAuthFooter does not sit between ManifestFooter and VolumeTrailer",
5520                ));
5521            }
5522        }
5523        (None, None) => {
5524            if manifest_end != input.trailer_offset {
5525                return Err(FormatError::WriterInvariant(
5526                    "ManifestFooter does not end at VolumeTrailer",
5527                ));
5528            }
5529        }
5530        _ => {
5531            return Err(FormatError::WriterInvariant(
5532                "RootAuthFooter offset/bytes mismatch",
5533            ));
5534        }
5535    }
5536    let body_bytes_before_cmra = input
5537        .trailer_offset
5538        .checked_add(VOLUME_TRAILER_LEN as u64)
5539        .ok_or(FormatError::WriterUnsupported("CMRA terminal overflow"))?;
5540    if body_bytes_before_cmra != input.cmra_offset {
5541        return Err(FormatError::WriterInvariant(
5542            "CMRA does not start after VolumeTrailer",
5543        ));
5544    }
5545
5546    let mut regions = vec![
5547        SerializedRegion {
5548            region_type: 1,
5549            offset: 0,
5550            bytes: input.volume_header_bytes.to_vec(),
5551        },
5552        SerializedRegion {
5553            region_type: 2,
5554            offset: VOLUME_HEADER_LEN as u64,
5555            bytes: input.crypto_header.to_vec(),
5556        },
5557        SerializedRegion {
5558            region_type: 3,
5559            offset: input.manifest_footer_offset,
5560            bytes: input.manifest_footer.to_vec(),
5561        },
5562    ];
5563    if let (Some(offset), Some(footer)) = (input.root_auth_footer_offset, input.root_auth_footer) {
5564        regions.push(SerializedRegion {
5565            region_type: 4,
5566            offset,
5567            bytes: footer.to_vec(),
5568        });
5569    }
5570    regions.push(SerializedRegion {
5571        region_type: 5,
5572        offset: input.trailer_offset,
5573        bytes: input.trailer.to_vec(),
5574    });
5575    let image = CriticalMetadataImage {
5576        archive_uuid: input.archive_uuid,
5577        session_id: input.session_id,
5578        volume_index: input.volume_index,
5579        stripe_width: input.options.stripe_width,
5580        layout_flags: if input.root_auth_footer.is_some() {
5581            0x0000_0001
5582        } else {
5583            0
5584        },
5585        volume_header_offset: 0,
5586        volume_header_length: VOLUME_HEADER_LEN as u32,
5587        crypto_header_offset: VOLUME_HEADER_LEN as u64,
5588        crypto_header_length: u32_len(input.crypto_header.len(), "CryptoHeader")?,
5589        block_records_offset,
5590        block_records_length,
5591        block_count: input.block_count,
5592        manifest_footer_offset: input.manifest_footer_offset,
5593        manifest_footer_length: MANIFEST_FOOTER_LEN as u32,
5594        root_auth_footer_offset: input.root_auth_footer_offset.unwrap_or(0),
5595        root_auth_footer_length: root_auth_footer_length.unwrap_or(0),
5596        volume_trailer_offset: input.trailer_offset,
5597        volume_trailer_length: VOLUME_TRAILER_LEN as u32,
5598        body_bytes_before_cmra,
5599        volume_header_sha256: sha256_bytes(input.volume_header_bytes),
5600        crypto_header_sha256: sha256_bytes(input.crypto_header),
5601        manifest_footer_sha256: sha256_bytes(input.manifest_footer),
5602        root_auth_footer_sha256: input
5603            .root_auth_footer
5604            .map(sha256_bytes)
5605            .unwrap_or([0u8; 32]),
5606        volume_trailer_sha256: sha256_bytes(input.trailer),
5607        regions,
5608    };
5609    let image_bytes = image.to_bytes()?;
5610    let image_sha256 = sha256_bytes(&image_bytes);
5611    let data_shard_count = ceil_div(image_bytes.len() as u64, CMRA_SHARD_SIZE as u64)?;
5612    let data_shard_count_u16 = u16::try_from(data_shard_count)
5613        .map_err(|_| FormatError::WriterUnsupported("CMRA data shard count"))?;
5614    let parity_lower = cmra_min_parity_shards(data_shard_count, input.options.bit_rot_buffer_pct)?;
5615    let parity_upper = cmra_min_parity_shards(data_shard_count, READER_MAX_CMRA_PARITY_PCT as u8)?;
5616    if parity_lower > parity_upper {
5617        return Err(FormatError::WriterUnsupported("CMRA parity bounds"));
5618    }
5619    let parity_shard_count_u16 = u16::try_from(parity_lower)
5620        .map_err(|_| FormatError::WriterUnsupported("CMRA parity shard count"))?;
5621
5622    let mut data_shards = Vec::with_capacity(data_shard_count as usize);
5623    for idx in 0..data_shard_count as usize {
5624        let start = idx * CMRA_SHARD_SIZE;
5625        let end = (start + CMRA_SHARD_SIZE).min(image_bytes.len());
5626        let mut shard = vec![0u8; CMRA_SHARD_SIZE];
5627        if start < image_bytes.len() {
5628            shard[..end - start].copy_from_slice(&image_bytes[start..end]);
5629        }
5630        data_shards.push(shard);
5631    }
5632    let parity_shards = encode_parity_gf16(&data_shards, parity_shard_count_u16 as usize)?;
5633
5634    let header = CriticalMetadataRecoveryHeader {
5635        shard_size: CMRA_SHARD_SIZE as u32,
5636        data_shard_count: data_shard_count_u16,
5637        parity_shard_count: parity_shard_count_u16,
5638        image_length: u32_len(image_bytes.len(), "CriticalMetadataImageV1")?,
5639        archive_uuid_hint: input.archive_uuid,
5640        session_id_hint: input.session_id,
5641        volume_index_hint: input.volume_index,
5642        image_sha256,
5643        header_crc32c: 0,
5644    };
5645    let mut cmra = Vec::new();
5646    cmra.extend_from_slice(&header.to_bytes());
5647    for (idx, payload) in data_shards.into_iter().enumerate() {
5648        let payload_len = if idx + 1 == data_shard_count as usize {
5649            let final_len = image_bytes.len() - idx * CMRA_SHARD_SIZE;
5650            if final_len == 0 {
5651                CMRA_SHARD_SIZE
5652            } else {
5653                final_len
5654            }
5655        } else {
5656            CMRA_SHARD_SIZE
5657        };
5658        cmra.extend_from_slice(
5659            &CriticalMetadataRecoveryShard {
5660                shard_index: u16::try_from(idx)
5661                    .map_err(|_| FormatError::WriterUnsupported("CMRA shard index"))?,
5662                shard_role: 0,
5663                shard_payload_length: u32_len(payload_len, "CMRA shard payload")?,
5664                payload,
5665                shard_crc32c: 0,
5666            }
5667            .to_bytes(CMRA_SHARD_SIZE)?,
5668        );
5669    }
5670    for (idx, payload) in parity_shards.into_iter().enumerate() {
5671        let shard_index = data_shard_count
5672            .checked_add(idx as u64)
5673            .ok_or(FormatError::WriterUnsupported("CMRA shard index overflow"))?;
5674        cmra.extend_from_slice(
5675            &CriticalMetadataRecoveryShard {
5676                shard_index: u16::try_from(shard_index)
5677                    .map_err(|_| FormatError::WriterUnsupported("CMRA shard index"))?,
5678                shard_role: 1,
5679                shard_payload_length: CMRA_SHARD_SIZE as u32,
5680                payload,
5681                shard_crc32c: 0,
5682            }
5683            .to_bytes(CMRA_SHARD_SIZE)?,
5684        );
5685    }
5686
5687    Ok(BuiltCmra {
5688        bytes: cmra,
5689        shard_size: CMRA_SHARD_SIZE as u32,
5690        data_shard_count: data_shard_count_u16,
5691        parity_shard_count: parity_shard_count_u16,
5692        image_length: u32_len(image_bytes.len(), "CriticalMetadataImageV1")?,
5693        image_sha256,
5694    })
5695}
5696
5697fn cmra_min_parity_shards(data_shard_count: u64, pct: u8) -> Result<u64, FormatError> {
5698    let by_pct = ceil_div(
5699        checked_u64_mul(data_shard_count, pct as u64, "CMRA parity overflow")?,
5700        100,
5701    )?;
5702    Ok(2u64.max(by_pct))
5703}
5704
5705fn compute_object_parity(
5706    data_block_count: u64,
5707    options: WriterOptions,
5708    class_parity_shard_max: u32,
5709) -> Result<u32, FormatError> {
5710    let computed = compute_parity(data_block_count, options)?;
5711    if computed > class_parity_shard_max {
5712        return Err(FormatError::WriterUnsupported(
5713            "encrypted object exceeds its parity shard class maximum",
5714        ));
5715    }
5716    Ok(computed)
5717}
5718
5719fn validate_object_shard_total(
5720    data_block_count: u32,
5721    parity_block_count: u32,
5722) -> Result<(), FormatError> {
5723    let total = checked_u64_add(
5724        data_block_count as u64,
5725        parity_block_count as u64,
5726        "encrypted object shard total overflow",
5727    )?;
5728    if total > MAX_REED_SOLOMON_GF16_SHARDS {
5729        return Err(FormatError::WriterUnsupported(
5730            "encrypted object exceeds ReedSolomonGF16 shard limit",
5731        ));
5732    }
5733    Ok(())
5734}
5735
5736fn compute_parity_u16(
5737    data_block_count: u64,
5738    options: WriterOptions,
5739    field: &'static str,
5740) -> Result<u16, FormatError> {
5741    let parity = compute_parity(data_block_count, options)?;
5742    u16::try_from(parity).map_err(|_| FormatError::WriterUnsupported(field))
5743}
5744
5745fn compute_parity(data_block_count: u64, options: WriterOptions) -> Result<u32, FormatError> {
5746    let min_parity = if options.volume_loss_tolerance > 0 || options.bit_rot_buffer_pct > 0 {
5747        1u64
5748    } else {
5749        0u64
5750    };
5751    let mut parity = 0u64;
5752    for _ in 0..100 {
5753        let total = data_block_count
5754            .checked_add(parity)
5755            .ok_or(FormatError::WriterUnsupported("parity total overflow"))?;
5756        let by_volume = checked_u64_mul(
5757            options.volume_loss_tolerance as u64,
5758            ceil_div(total, options.stripe_width as u64)?,
5759            "volume-loss parity overflow",
5760        )?;
5761        let by_bitrot = ceil_div(
5762            checked_u64_mul(
5763                total,
5764                options.bit_rot_buffer_pct as u64,
5765                "bit-rot parity overflow",
5766            )?,
5767            100,
5768        )?;
5769        let next = by_volume
5770            .checked_add(by_bitrot)
5771            .ok_or(FormatError::WriterUnsupported("parity overflow"))?
5772            .max(min_parity);
5773        if next == parity {
5774            return u32::try_from(next).map_err(|_| FormatError::WriterUnsupported("parity count"));
5775        }
5776        parity = next;
5777    }
5778    Err(FormatError::WriterUnsupported(
5779        "parity calculation did not converge",
5780    ))
5781}
5782
5783fn ceil_div(numerator: u64, denominator: u64) -> Result<u64, FormatError> {
5784    if denominator == 0 {
5785        return Err(FormatError::WriterUnsupported("division by zero"));
5786    }
5787    numerator
5788        .checked_add(denominator - 1)
5789        .ok_or(FormatError::WriterUnsupported("ceiling division overflow"))
5790        .map(|value| value / denominator)
5791}
5792
5793fn checked_u64_mul(lhs: u64, rhs: u64, field: &'static str) -> Result<u64, FormatError> {
5794    lhs.checked_mul(rhs)
5795        .ok_or(FormatError::WriterUnsupported(field))
5796}
5797
5798fn build_bootstrap_sidecar(
5799    subkeys: &Subkeys,
5800    aead_algo: AeadAlgo,
5801    archive_uuid: [u8; 16],
5802    session_id: [u8; 16],
5803    manifest_footer: &[u8; MANIFEST_FOOTER_LEN],
5804    index_root_records: &[BlockRecord],
5805    dictionary_records: Option<&[BlockRecord]>,
5806) -> Result<Vec<u8>, FormatError> {
5807    let index_records_len = index_root_records.iter().try_fold(0usize, |sum, record| {
5808        checked_usize_add(sum, record.to_bytes().len(), "bootstrap sidecar")
5809    })?;
5810    let dictionary_records_len = dictionary_records
5811        .unwrap_or(&[])
5812        .iter()
5813        .try_fold(0usize, |sum, record| {
5814            checked_usize_add(sum, record.to_bytes().len(), "bootstrap sidecar")
5815        })?;
5816    let manifest_offset = BOOTSTRAP_SIDECAR_HEADER_LEN as u64;
5817    let index_root_offset = manifest_offset + MANIFEST_FOOTER_LEN as u64;
5818    let dictionary_offset = if dictionary_records.is_some() {
5819        index_root_offset + index_records_len as u64
5820    } else {
5821        0
5822    };
5823    let mut header = BootstrapSidecarHeader {
5824        archive_uuid,
5825        session_id,
5826        flags: if dictionary_records.is_some() {
5827            0x07
5828        } else {
5829            0x03
5830        },
5831        manifest_footer_offset: manifest_offset,
5832        manifest_footer_length: MANIFEST_FOOTER_LEN as u32,
5833        index_root_records_offset: index_root_offset,
5834        index_root_records_length: index_records_len as u64,
5835        dictionary_records_offset: dictionary_offset,
5836        dictionary_records_length: dictionary_records_len as u64,
5837        sidecar_hmac: [0u8; 32],
5838        header_crc32c: 0,
5839    };
5840    let mut header_bytes = header.to_bytes();
5841    header.sidecar_hmac = compute_integrity_tag(
5842        HmacDomain::BootstrapSidecar,
5843        aead_algo,
5844        Some(&subkeys.mac_key),
5845        &archive_uuid,
5846        &session_id,
5847        &header_bytes[..92],
5848    )?;
5849    header_bytes = header.to_bytes();
5850
5851    let mut sidecar = Vec::with_capacity(
5852        BOOTSTRAP_SIDECAR_HEADER_LEN
5853            + MANIFEST_FOOTER_LEN
5854            + index_records_len
5855            + dictionary_records_len,
5856    );
5857    sidecar.extend_from_slice(&header_bytes);
5858    sidecar.extend_from_slice(manifest_footer);
5859    for record in index_root_records {
5860        sidecar.extend_from_slice(&record.to_bytes());
5861    }
5862    if let Some(dictionary_records) = dictionary_records {
5863        for record in dictionary_records {
5864            sidecar.extend_from_slice(&record.to_bytes());
5865        }
5866    }
5867    Ok(sidecar)
5868}
5869
5870struct StreamingMemberReader<'a> {
5871    prefix: Cursor<Vec<u8>>,
5872    file: Box<dyn Read + 'a>,
5873    remaining_file_bytes: u64,
5874    remaining_padding_bytes: usize,
5875    pushback: Vec<u8>,
5876}
5877
5878impl<'a> StreamingMemberReader<'a> {
5879    fn new(file: Box<dyn Read + 'a>, prefix: Vec<u8>, file_size: u64) -> Self {
5880        Self {
5881            prefix: Cursor::new(prefix),
5882            file,
5883            remaining_file_bytes: file_size,
5884            remaining_padding_bytes: padding_to_512_u64(file_size) as usize,
5885            pushback: Vec::new(),
5886        }
5887    }
5888
5889    fn push_back(&mut self, bytes: Vec<u8>) {
5890        if bytes.is_empty() {
5891            return;
5892        }
5893        if self.pushback.is_empty() {
5894            self.pushback = bytes;
5895        } else {
5896            let mut merged = bytes;
5897            merged.extend_from_slice(&self.pushback);
5898            self.pushback = merged;
5899        }
5900    }
5901}
5902
5903impl Read for StreamingMemberReader<'_> {
5904    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
5905        if out.is_empty() {
5906            return Ok(0);
5907        }
5908
5909        let mut written = 0usize;
5910        if !self.pushback.is_empty() {
5911            let count = out.len().min(self.pushback.len());
5912            out[..count].copy_from_slice(&self.pushback[..count]);
5913            self.pushback.drain(..count);
5914            written += count;
5915            if written == out.len() {
5916                return Ok(written);
5917            }
5918        }
5919
5920        let prefix_count = self.prefix.read(&mut out[written..])?;
5921        written += prefix_count;
5922        if written == out.len() {
5923            return Ok(written);
5924        }
5925
5926        if self.remaining_file_bytes > 0 {
5927            let max_file_read = (out.len() - written)
5928                .min(usize::try_from(self.remaining_file_bytes).unwrap_or(usize::MAX));
5929            let count = self.file.read(&mut out[written..written + max_file_read])?;
5930            if count == 0 {
5931                return Err(std::io::Error::new(
5932                    std::io::ErrorKind::UnexpectedEof,
5933                    "file source ended before declared size",
5934                ));
5935            }
5936            self.remaining_file_bytes -= count as u64;
5937            written += count;
5938            if written == out.len() {
5939                return Ok(written);
5940            }
5941            if self.remaining_file_bytes > 0 {
5942                return Ok(written);
5943            }
5944        }
5945
5946        if self.remaining_padding_bytes > 0 {
5947            let count = (out.len() - written).min(self.remaining_padding_bytes);
5948            out[written..written + count].fill(0);
5949            self.remaining_padding_bytes -= count;
5950            written += count;
5951        }
5952
5953        Ok(written)
5954    }
5955}
5956
5957fn build_regular_file_member_prefix(
5958    path: &[u8],
5959    file_size: u64,
5960    mode: u32,
5961    mtime: u64,
5962) -> Result<Vec<u8>, FormatError> {
5963    let mut out = Vec::new();
5964    let header_path = if path_requires_pax(path) {
5965        let pax_payload = build_pax_record("path", path)?;
5966        let pax_header = build_ustar_header(
5967            b"PaxHeaders/path",
5968            pax_payload.len() as u64,
5969            0o644,
5970            mtime,
5971            b'x',
5972        )?;
5973        out.extend_from_slice(&pax_header);
5974        out.extend_from_slice(&pax_payload);
5975        out.resize(out.len() + padding_to_512(pax_payload.len()), 0);
5976        pax_ustar_fallback_path(path)
5977    } else {
5978        path.to_vec()
5979    };
5980
5981    let header = build_ustar_header(&header_path, file_size, mode, mtime, b'0')?;
5982    out.extend_from_slice(&header);
5983    Ok(out)
5984}
5985
5986#[cfg(test)]
5987fn build_regular_file_member_group(
5988    path: &[u8],
5989    contents: &[u8],
5990    mode: u32,
5991    mtime: u64,
5992) -> Result<Vec<u8>, FormatError> {
5993    let mut out = build_regular_file_member_prefix(path, contents.len() as u64, mode, mtime)?;
5994    out.extend_from_slice(contents);
5995    out.resize(out.len() + padding_to_512(contents.len()), 0);
5996    Ok(out)
5997}
5998
5999fn path_requires_pax(path: &[u8]) -> bool {
6000    path.len() > 100 || !path.is_ascii()
6001}
6002
6003fn pax_ustar_fallback_path(path: &[u8]) -> Vec<u8> {
6004    path.rsplit(|byte| *byte == b'/')
6005        .next()
6006        .filter(|component| !component.is_empty() && component.len() <= 100 && component.is_ascii())
6007        .map(|component| component.to_vec())
6008        .unwrap_or_else(|| b"pax-file".to_vec())
6009}
6010
6011fn build_pax_record(key: &str, value: &[u8]) -> Result<Vec<u8>, FormatError> {
6012    let body_len = checked_usize_add(key.len(), 1, "PAX record")?;
6013    let body_len = checked_usize_add(body_len, value.len(), "PAX record")?;
6014    let body_len = checked_usize_add(body_len, 1, "PAX record")?;
6015    let mut digits = 1usize;
6016    loop {
6017        let len = checked_usize_add(digits, 1, "PAX record")?;
6018        let len = checked_usize_add(len, body_len, "PAX record")?;
6019        let next_digits = len.to_string().len();
6020        if next_digits == digits {
6021            let mut out = Vec::with_capacity(len);
6022            out.extend_from_slice(len.to_string().as_bytes());
6023            out.push(b' ');
6024            out.extend_from_slice(key.as_bytes());
6025            out.push(b'=');
6026            out.extend_from_slice(value);
6027            out.push(b'\n');
6028            return Ok(out);
6029        }
6030        digits = next_digits;
6031    }
6032}
6033
6034fn build_ustar_header(
6035    path: &[u8],
6036    size: u64,
6037    mode: u32,
6038    mtime: u64,
6039    typeflag: u8,
6040) -> Result<[u8; TAR_BLOCK_LEN], FormatError> {
6041    if path.len() > 100 {
6042        return Err(FormatError::WriterUnsupported(
6043            "ustar path exceeds name field",
6044        ));
6045    }
6046    let mut header = [0u8; TAR_BLOCK_LEN];
6047    header[0..path.len()].copy_from_slice(path);
6048    write_tar_octal(&mut header[100..108], mode as u64)?;
6049    write_tar_octal(&mut header[108..116], 0)?;
6050    write_tar_octal(&mut header[116..124], 0)?;
6051    write_tar_octal(&mut header[124..136], size)?;
6052    write_tar_octal(&mut header[136..148], mtime)?;
6053    header[148..156].fill(b' ');
6054    header[156] = typeflag;
6055    header[257..263].copy_from_slice(b"ustar\0");
6056    header[263..265].copy_from_slice(b"00");
6057    let checksum = header.iter().map(|byte| *byte as u32).sum::<u32>() as u64;
6058    write_tar_checksum(&mut header[148..156], checksum)?;
6059    Ok(header)
6060}
6061
6062fn write_tar_octal(field: &mut [u8], value: u64) -> Result<(), FormatError> {
6063    let digits = format!("{value:o}");
6064    if digits.len() + 1 > field.len() {
6065        return Err(FormatError::WriterUnsupported("tar octal field overflow"));
6066    }
6067    field.fill(0);
6068    let padding = field.len() - 1 - digits.len();
6069    for byte in &mut field[..padding] {
6070        *byte = b'0';
6071    }
6072    field[padding..padding + digits.len()].copy_from_slice(digits.as_bytes());
6073    Ok(())
6074}
6075
6076fn write_tar_checksum(field: &mut [u8], value: u64) -> Result<(), FormatError> {
6077    let digits = format!("{value:06o}");
6078    if digits.len() != 6 {
6079        return Err(FormatError::WriterUnsupported(
6080            "tar checksum field overflow",
6081        ));
6082    }
6083    field[0..6].copy_from_slice(digits.as_bytes());
6084    field[6] = 0;
6085    field[7] = b' ';
6086    Ok(())
6087}
6088
6089fn member_frame_range(
6090    member_index: usize,
6091    frames: &[PayloadFrame],
6092) -> Result<(u64, u32), FormatError> {
6093    let first = frames
6094        .iter()
6095        .find(|frame| frame.member_index == member_index)
6096        .map(|frame| frame.frame_index)
6097        .ok_or(FormatError::WriterInvariant("member frame is missing"))?;
6098    let count = frames
6099        .iter()
6100        .filter(|frame| frame.member_index == member_index)
6101        .count();
6102    Ok((first, u32_len(count, "FileEntry.frame_count")?))
6103}
6104
6105fn envelope_frame_range(
6106    envelope_index: u64,
6107    frames: &[PayloadFrame],
6108) -> Result<(u64, u32), FormatError> {
6109    let first = frames
6110        .iter()
6111        .find(|frame| frame.envelope_index == envelope_index)
6112        .map(|frame| frame.frame_index)
6113        .ok_or(FormatError::WriterInvariant("envelope frame is missing"))?;
6114    let count = frames
6115        .iter()
6116        .filter(|frame| frame.envelope_index == envelope_index)
6117        .count();
6118    Ok((first, u32_len(count, "EnvelopeEntry.frame_count")?))
6119}
6120
6121fn sha256_bytes(bytes: &[u8]) -> [u8; 32] {
6122    let digest = Sha256::digest(bytes);
6123    let mut out = [0u8; 32];
6124    out.copy_from_slice(&digest);
6125    out
6126}
6127
6128fn padding_to_512(len: usize) -> usize {
6129    let remainder = len % TAR_BLOCK_LEN;
6130    if remainder == 0 {
6131        0
6132    } else {
6133        TAR_BLOCK_LEN - remainder
6134    }
6135}
6136
6137fn padding_to_512_u64(len: u64) -> u64 {
6138    let remainder = len % TAR_BLOCK_LEN as u64;
6139    if remainder == 0 {
6140        0
6141    } else {
6142        TAR_BLOCK_LEN as u64 - remainder
6143    }
6144}
6145
6146fn table_offset(len: usize, cursor: usize) -> Result<u32, FormatError> {
6147    if len == 0 {
6148        Ok(0)
6149    } else {
6150        u32_len(cursor, "table offset")
6151    }
6152}
6153
6154fn u32_len(value: usize, field: &'static str) -> Result<u32, FormatError> {
6155    u32::try_from(value).map_err(|_| FormatError::WriterUnsupported(field))
6156}
6157
6158fn to_usize_writer(value: u64, field: &'static str) -> Result<usize, FormatError> {
6159    usize::try_from(value).map_err(|_| FormatError::WriterUnsupported(field))
6160}
6161
6162fn checked_usize_add(lhs: usize, rhs: usize, field: &'static str) -> Result<usize, FormatError> {
6163    lhs.checked_add(rhs)
6164        .ok_or(FormatError::WriterUnsupported(field))
6165}
6166
6167fn checked_u64_add(lhs: u64, rhs: u64, field: &'static str) -> Result<u64, FormatError> {
6168    lhs.checked_add(rhs)
6169        .ok_or(FormatError::WriterUnsupported(field))
6170}
6171
6172#[cfg(test)]
6173mod tests {
6174    use super::*;
6175    use crate::crypto::{verify_hmac, Subkeys};
6176    use crate::metadata::{DirectoryHintTable, MetadataLimits};
6177    use crate::reader::open_archive;
6178    use crate::tar_model::parse_tar_member_group;
6179    use crate::wire::{CriticalRecoveryLocator, CryptoHeader};
6180    use std::cell::RefCell;
6181    use std::io::{self, Read};
6182    use std::rc::Rc;
6183
6184    #[test]
6185    fn writer_defaults_use_v41_sizing_and_parallel_mode() {
6186        let options = WriterOptions::default();
6187
6188        assert_eq!(options.chunk_size, 256 * 1024);
6189        assert_eq!(options.envelope_target_size, 1024 * 1024);
6190        assert_eq!(options.block_size, 64 * 1024);
6191        assert_eq!(options.stripe_width, 8);
6192        assert_eq!(options.volume_loss_tolerance, 1);
6193        assert_eq!(options.fec_data_shards, 224);
6194        assert_eq!(options.index_fec_data_shards, 16);
6195        assert_eq!(
6196            options.index_root_fec_data_shards,
6197            MIN_INDEX_ROOT_FEC_DATA_SHARDS
6198        );
6199        assert_eq!(options.bit_rot_buffer_pct, 5);
6200        assert_eq!(options.jobs, default_jobs());
6201        assert!(options.jobs >= 1);
6202    }
6203
6204    #[test]
6205    fn writer_options_reject_zero_jobs() {
6206        let err = plan_writer_options(WriterOptions {
6207            jobs: 0,
6208            ..WriterOptions::default()
6209        })
6210        .unwrap_err();
6211
6212        assert_eq!(
6213            err,
6214            FormatError::WriterUnsupported("jobs must be at least 1")
6215        );
6216    }
6217
6218    #[test]
6219    fn production_writer_defaults_generate_distinct_v4_identities() {
6220        let master_key = MasterKey::from_raw_key(&[9u8; 32]).unwrap();
6221        let first = write_archive(&[], &master_key, WriterOptions::default()).unwrap();
6222        let second = write_archive(&[], &master_key, WriterOptions::default()).unwrap();
6223
6224        assert_ne!(first.archive_uuid, [0u8; 16]);
6225        assert_ne!(first.session_id, [0u8; 16]);
6226        assert_ne!(second.archive_uuid, [0u8; 16]);
6227        assert_ne!(second.session_id, [0u8; 16]);
6228        assert_ne!(first.archive_uuid, first.session_id);
6229        assert_ne!(first.archive_uuid, second.archive_uuid);
6230        assert_ne!(first.session_id, second.session_id);
6231
6232        for raw in [
6233            first.archive_uuid,
6234            first.session_id,
6235            second.archive_uuid,
6236            second.session_id,
6237        ] {
6238            let id = Uuid::from_bytes(raw);
6239            assert_eq!(id.get_version_num(), 4);
6240        }
6241
6242        let deterministic = WriterOptions {
6243            archive_uuid: Some([0x44; 16]),
6244            session_id: Some([0x55; 16]),
6245            ..WriterOptions::default()
6246        };
6247        let fixture = write_archive(&[], &master_key, deterministic).unwrap();
6248        assert_eq!(fixture.archive_uuid, [0x44; 16]);
6249        assert_eq!(fixture.session_id, [0x55; 16]);
6250    }
6251
6252    #[test]
6253    fn writer_partitions_multiple_default_sized_index_shards() {
6254        let members = (0..=DEFAULT_FILES_PER_INDEX_SHARD)
6255            .map(|idx| TarMember {
6256                path: format!("file-{idx:05}.txt").into_bytes(),
6257                tar_member_group_start: idx as u64 * 512,
6258                tar_member_group_size: 512,
6259                file_data_size: 0,
6260                mode: 0o644,
6261                mtime: 0,
6262            })
6263            .collect::<Vec<_>>();
6264
6265        let shards = partition_file_rows(sorted_file_rows(&members)).unwrap();
6266
6267        assert_eq!(shards.len(), 2);
6268        assert_eq!(shards[0].len(), DEFAULT_FILES_PER_INDEX_SHARD);
6269        assert_eq!(shards[1].len(), 1);
6270    }
6271
6272    #[test]
6273    fn writer_extends_shard_for_bounded_hash_prefix_run() {
6274        let mut rows = Vec::new();
6275        rows.extend((0..9_000).map(|idx| test_file_row(idx, [0u8; 8])));
6276        rows.extend((9_000..54_000).map(|idx| test_file_row(idx, [1u8; 8])));
6277        rows.push(test_file_row(54_000, [2u8; 8]));
6278
6279        let shards = partition_file_rows(rows).unwrap();
6280
6281        assert_eq!(shards.len(), 2);
6282        assert_eq!(shards[0].len(), 54_000);
6283        assert!(shards[0]
6284            .iter()
6285            .skip(9_000)
6286            .all(|row| row.path_hash == [1u8; 8]));
6287        assert_eq!(shards[1][0].path_hash, [2u8; 8]);
6288    }
6289
6290    #[test]
6291    fn writer_splits_oversized_hash_prefix_run_at_writer_ceiling() {
6292        let rows = (0..MAX_HASH_PREFIX_RUN_FILES + 1)
6293            .map(|idx| test_file_row(idx, [7u8; 8]))
6294            .collect::<Vec<_>>();
6295
6296        let shards = partition_file_rows(rows).unwrap();
6297
6298        assert_eq!(shards.len(), 2);
6299        assert_eq!(shards[0].len(), MAX_HASH_PREFIX_RUN_FILES);
6300        assert_eq!(shards[1].len(), 1);
6301    }
6302
6303    #[test]
6304    fn writer_builds_directory_hint_rows_for_ancestor_directories() {
6305        let shard_rows = vec![
6306            vec![FileRow {
6307                path_hash: hash_prefix(b"a/b/one.txt"),
6308                path: b"a/b/one.txt".to_vec(),
6309                member_index: 0,
6310                member: TarMember {
6311                    path: b"a/b/one.txt".to_vec(),
6312                    tar_member_group_start: 0,
6313                    tar_member_group_size: 512,
6314                    file_data_size: 0,
6315                    mode: 0o644,
6316                    mtime: 0,
6317                },
6318            }],
6319            vec![FileRow {
6320                path_hash: hash_prefix(b"a/c/two.txt"),
6321                path: b"a/c/two.txt".to_vec(),
6322                member_index: 1,
6323                member: TarMember {
6324                    path: b"a/c/two.txt".to_vec(),
6325                    tar_member_group_start: 512,
6326                    tar_member_group_size: 512,
6327                    file_data_size: 0,
6328                    mode: 0o644,
6329                    mtime: 0,
6330                },
6331            }],
6332        ];
6333
6334        let options = plan_writer_options(WriterOptions::default()).unwrap();
6335        let planned = build_directory_hint_plaintexts(&shard_rows, options).unwrap();
6336        assert_eq!(planned.len(), 1);
6337        let locating = DirectoryHintShardEntry {
6338            hint_shard_index: planned[0].hint_shard_index,
6339            first_dir_hash: planned[0].first_dir_hash,
6340            last_dir_hash: planned[0].last_dir_hash,
6341            first_block_index: 0,
6342            data_block_count: 1,
6343            parity_block_count: 0,
6344            encrypted_size: 4096,
6345            decompressed_size: planned[0].plaintext.len() as u32,
6346            entry_count: planned[0].entry_count,
6347        };
6348        let table = DirectoryHintTable::parse(
6349            &planned[0].plaintext,
6350            &locating,
6351            2,
6352            MetadataLimits::default(),
6353        )
6354        .unwrap();
6355
6356        let root = table.lookup_directory_index(b"").unwrap();
6357        assert_eq!(table.shard_rows_for_entry(root).unwrap(), &[0, 1]);
6358        let a = table.lookup_directory_index(b"a").unwrap();
6359        assert_eq!(table.shard_rows_for_entry(a).unwrap(), &[0, 1]);
6360        let ab = table.lookup_directory_index(b"a/b").unwrap();
6361        assert_eq!(table.shard_rows_for_entry(ab).unwrap(), &[0]);
6362    }
6363
6364    #[test]
6365    fn directory_hints_are_required_only_above_v41_threshold() {
6366        assert!(!should_emit_directory_hints(0));
6367        assert!(!should_emit_directory_hints(
6368            DIRECTORY_HINT_REQUIRED_FILE_COUNT
6369        ));
6370        assert!(should_emit_directory_hints(
6371            DIRECTORY_HINT_REQUIRED_FILE_COUNT + 1
6372        ));
6373    }
6374
6375    #[test]
6376    fn regular_file_writer_uses_local_pax_path_for_long_and_non_ascii_paths() {
6377        let long_path = format!("dir/{}.txt", "a".repeat(120));
6378        let unicode_path = "unicode/e\u{301}.txt";
6379        let files = [
6380            RegularFile::new(&long_path, b"long path"),
6381            RegularFile::new(unicode_path, b"unicode path"),
6382        ];
6383
6384        let (tar_stream, members) = build_tar_stream(&files, 4096).unwrap();
6385
6386        for (member, expected_path, expected_data) in [
6387            (&members[0], long_path.as_bytes(), b"long path".as_slice()),
6388            (
6389                &members[1],
6390                "unicode/\u{e9}.txt".as_bytes(),
6391                b"unicode path".as_slice(),
6392            ),
6393        ] {
6394            let start = member.tar_member_group_start as usize;
6395            let end = start + member.tar_member_group_size as usize;
6396            let group = &tar_stream[start..end];
6397            assert_eq!(group[156], b'x');
6398            let parsed = parse_tar_member_group(group, 4096).unwrap();
6399            assert_eq!(parsed.path, expected_path);
6400            assert_eq!(parsed.data, expected_data);
6401        }
6402    }
6403
6404    #[test]
6405    fn regular_file_writer_emits_no_global_metadata_or_tar_eof() {
6406        let long_path = format!("dir/{}.txt", "a".repeat(120));
6407        let files = [
6408            RegularFile::new("plain.txt", b"plain contents"),
6409            RegularFile::new(&long_path, b"long path contents"),
6410        ];
6411
6412        let (tar_stream, members) = build_tar_stream(&files, 4096).unwrap();
6413
6414        let member_bytes = members
6415            .iter()
6416            .map(|member| member.tar_member_group_size)
6417            .sum::<u64>();
6418        assert_eq!(tar_stream.len() as u64, member_bytes);
6419        assert!(!tar_stream[tar_stream.len() - TAR_BLOCK_LEN * 2..]
6420            .chunks(TAR_BLOCK_LEN)
6421            .all(|block| block.iter().all(|byte| *byte == 0)));
6422
6423        for member in members {
6424            let start = member.tar_member_group_start as usize;
6425            let end = start + member.tar_member_group_size as usize;
6426            assert_path_specific_member_group(&tar_stream[start..end]);
6427        }
6428    }
6429
6430    #[test]
6431    fn regular_file_writer_round_trips_mode_and_mtime() {
6432        let group =
6433            build_regular_file_member_group(b"script.sh", b"#!/bin/sh\n", 0o755, 1_700_000_000)
6434                .unwrap();
6435
6436        let parsed = parse_tar_member_group(&group, 4096).unwrap();
6437
6438        assert_eq!(parsed.mode, 0o755);
6439        assert_eq!(parsed.mtime, 1_700_000_000);
6440    }
6441
6442    #[test]
6443    fn writer_splits_large_payload_across_seekable_envelopes() {
6444        let master_key = MasterKey::from_raw_key(&[8u8; 32]).unwrap();
6445        let data = deterministic_bytes(2 * 1024 * 1024);
6446        let archive = write_archive(
6447            &[RegularFile::new("large.bin", &data)],
6448            &master_key,
6449            WriterOptions {
6450                stripe_width: 1,
6451                volume_loss_tolerance: 0,
6452                bit_rot_buffer_pct: 0,
6453                ..WriterOptions::default()
6454            },
6455        )
6456        .unwrap();
6457        let opened = open_archive(&archive.bytes, &master_key).unwrap();
6458
6459        assert_eq!(opened.list_files().unwrap()[0].path, "large.bin");
6460        assert_eq!(opened.extract_file("large.bin").unwrap(), Some(data));
6461        opened.verify().unwrap();
6462        assert!(opened.index_root.header.envelope_count > 1);
6463    }
6464
6465    #[test]
6466    fn split_member_frames_carry_exact_boundary_flags() {
6467        let data = deterministic_bytes(12 * 1024);
6468        let files = [RegularFile::new("large.bin", &data)];
6469        let options = WriterOptions {
6470            chunk_size: 1024,
6471            envelope_target_size: 64 * 1024,
6472            stripe_width: 1,
6473            volume_loss_tolerance: 0,
6474            bit_rot_buffer_pct: 0,
6475            ..WriterOptions::default()
6476        };
6477        let (tar_stream, members) = build_tar_stream(&files, options.max_path_length).unwrap();
6478        let (_, frames) = build_payload_envelopes(&tar_stream, &members, options, None).unwrap();
6479
6480        assert!(frames.len() > 2);
6481        assert_eq!(frames.first().unwrap().flags, 0x0000_0001);
6482        assert_eq!(frames.last().unwrap().flags, 0x0000_0002);
6483        assert!(frames[1..frames.len() - 1]
6484            .iter()
6485            .all(|frame| frame.flags == 0));
6486    }
6487
6488    #[test]
6489    fn writes_empty_archive_with_authentic_bootstrap_structures() {
6490        let master_key = MasterKey::from_raw_key(&[7u8; 32]).unwrap();
6491        let archive = write_empty_archive(&master_key).unwrap();
6492        let bytes = archive.bytes;
6493
6494        let volume_header = VolumeHeader::parse(&bytes[..VOLUME_HEADER_LEN]).unwrap();
6495        assert_eq!(volume_header.archive_uuid, archive.archive_uuid);
6496        assert_eq!(volume_header.session_id, archive.session_id);
6497
6498        let crypto_start = VOLUME_HEADER_LEN;
6499        let crypto_end = crypto_start + volume_header.crypto_header_length as usize;
6500        let crypto_header = CryptoHeader::parse(
6501            &bytes[crypto_start..crypto_end],
6502            volume_header.crypto_header_length,
6503        )
6504        .unwrap();
6505        let subkeys =
6506            Subkeys::derive(&master_key, &archive.archive_uuid, &archive.session_id).unwrap();
6507        verify_hmac(
6508            HmacDomain::CryptoHeader,
6509            &subkeys.mac_key,
6510            &archive.archive_uuid,
6511            &archive.session_id,
6512            crypto_header.hmac_covered_bytes,
6513            &crypto_header.header_hmac,
6514        )
6515        .unwrap();
6516
6517        let locator =
6518            CriticalRecoveryLocator::parse(&bytes[bytes.len() - CRITICAL_RECOVERY_LOCATOR_LEN..])
6519                .unwrap();
6520        let trailer_offset = locator.volume_trailer_offset as usize;
6521        let trailer =
6522            VolumeTrailer::parse(&bytes[trailer_offset..trailer_offset + VOLUME_TRAILER_LEN])
6523                .unwrap();
6524        assert_eq!(trailer.bytes_written, trailer_offset as u64);
6525        verify_hmac(
6526            HmacDomain::VolumeTrailer,
6527            &subkeys.mac_key,
6528            &archive.archive_uuid,
6529            &archive.session_id,
6530            &bytes[trailer_offset..trailer_offset + 96],
6531            &trailer.trailer_hmac,
6532        )
6533        .unwrap();
6534
6535        let manifest_offset = trailer.manifest_footer_offset as usize;
6536        let manifest_end = manifest_offset + MANIFEST_FOOTER_LEN;
6537        let manifest = ManifestFooter::parse(&bytes[manifest_offset..manifest_end]).unwrap();
6538        assert_eq!(manifest.is_authoritative, 1);
6539        assert_eq!(manifest.total_volumes, DEFAULT_STRIPE_WIDTH);
6540        verify_hmac(
6541            HmacDomain::ManifestFooter,
6542            &subkeys.mac_key,
6543            &archive.archive_uuid,
6544            &archive.session_id,
6545            &bytes[manifest_offset..manifest_offset + 104],
6546            &manifest.manifest_hmac,
6547        )
6548        .unwrap();
6549    }
6550
6551    #[test]
6552    fn parity_auto_scaling_matches_v41_examples() {
6553        let options = WriterOptions {
6554            fec_data_shards: 224,
6555            stripe_width: 8,
6556            volume_loss_tolerance: 1,
6557            bit_rot_buffer_pct: 5,
6558            ..WriterOptions::default()
6559        };
6560
6561        assert_eq!(compute_parity(224, options).unwrap(), 48);
6562        assert_eq!(compute_parity(17, options).unwrap(), 5);
6563    }
6564
6565    #[test]
6566    fn parity_auto_scaling_rejects_non_convergent_budget() {
6567        let err = compute_parity(
6568            1,
6569            WriterOptions {
6570                stripe_width: 2,
6571                volume_loss_tolerance: 1,
6572                bit_rot_buffer_pct: 50,
6573                ..WriterOptions::default()
6574            },
6575        )
6576        .unwrap_err();
6577
6578        assert_eq!(
6579            err,
6580            FormatError::WriterUnsupported("parity calculation did not converge")
6581        );
6582    }
6583
6584    #[test]
6585    fn zero_parity_is_allowed_when_no_recovery_margin_is_requested() {
6586        let planned = plan_writer_options(WriterOptions {
6587            bit_rot_buffer_pct: 0,
6588            stripe_width: 1,
6589            volume_loss_tolerance: 0,
6590            fec_parity_shards: 0,
6591            index_fec_parity_shards: 0,
6592            index_root_fec_parity_shards: 0,
6593            ..WriterOptions::default()
6594        })
6595        .unwrap();
6596
6597        assert_eq!(planned.fec_parity_shards, 0);
6598        assert_eq!(planned.index_fec_parity_shards, 0);
6599        assert_eq!(planned.index_root_fec_parity_shards, 0);
6600        assert_eq!(compute_parity(1, planned).unwrap(), 0);
6601    }
6602
6603    #[test]
6604    fn index_root_data_shard_maximum_obeys_v41_minimum() {
6605        let planned = plan_writer_options(WriterOptions {
6606            index_root_fec_data_shards: 1,
6607            ..WriterOptions::default()
6608        })
6609        .unwrap();
6610
6611        assert_eq!(
6612            planned.index_root_fec_data_shards,
6613            MIN_INDEX_ROOT_FEC_DATA_SHARDS
6614        );
6615    }
6616
6617    #[test]
6618    fn metadata_class_planning_raises_index_root_class_above_default() {
6619        let options = plan_writer_options(WriterOptions {
6620            block_size: MIN_BLOCK_SIZE,
6621            index_root_fec_parity_shards: 0,
6622            bit_rot_buffer_pct: 0,
6623            ..WriterOptions::default()
6624        })
6625        .unwrap();
6626        let index_root_payload_len = payload_len_for_encrypted_data_blocks(64, options);
6627
6628        let planned =
6629            plan_index_root_metadata_class(options, index_root_payload_len, None).unwrap();
6630
6631        assert_eq!(planned.index_root.data_block_count, 64);
6632        assert_eq!(planned.options.index_root_fec_data_shards, 64);
6633        assert_eq!(
6634            planned.options.index_root_fec_parity_shards,
6635            compute_parity_u16(
6636                planned.options.index_root_fec_data_shards as u64,
6637                planned.options,
6638                "index_root_fec_parity_shards",
6639            )
6640            .unwrap()
6641        );
6642    }
6643
6644    #[test]
6645    fn single_pass_writer_predeclares_metadata_class_before_payload_streaming() {
6646        let planned = plan_single_pass_writer_options(WriterOptions {
6647            block_size: MIN_BLOCK_SIZE,
6648            stripe_width: 1,
6649            volume_loss_tolerance: 0,
6650            bit_rot_buffer_pct: 0,
6651            index_root_fec_parity_shards: 0,
6652            ..WriterOptions::default()
6653        })
6654        .unwrap();
6655
6656        assert!(planned.index_root_fec_data_shards > DEFAULT_INDEX_ROOT_FEC_DATA_SHARDS);
6657        let index_root_payload_len = payload_len_for_encrypted_data_blocks(
6658            u32::from(planned.index_root_fec_data_shards - 1),
6659            planned,
6660        );
6661        let metadata_class =
6662            plan_index_root_metadata_class(planned, index_root_payload_len, None).unwrap();
6663
6664        assert_eq!(metadata_class.options, planned);
6665    }
6666
6667    #[test]
6668    fn metadata_class_planning_rejects_oversized_index_root() {
6669        let options = single_volume_metadata_test_options();
6670        let index_root_payload_len =
6671            payload_len_for_encrypted_data_blocks(u16::MAX as u32 + 1, options);
6672
6673        let err =
6674            plan_index_root_metadata_class(options, index_root_payload_len, None).unwrap_err();
6675
6676        assert_eq!(err, FormatError::WriterUnsupported("IndexRoot too large"));
6677    }
6678
6679    #[test]
6680    fn metadata_class_planning_rejects_index_root_u32_encrypted_size_overflow() {
6681        let options = single_volume_metadata_test_options();
6682        let index_root_payload_len = u32::MAX as usize - options.aead_algo.tag_len() + 1;
6683
6684        let err =
6685            plan_index_root_metadata_class(options, index_root_payload_len, None).unwrap_err();
6686
6687        assert_eq!(err, FormatError::WriterUnsupported("IndexRoot too large"));
6688    }
6689
6690    #[test]
6691    fn metadata_class_planning_rejects_oversized_dictionary() {
6692        let options = single_volume_metadata_test_options();
6693        let dictionary_payload_len =
6694            payload_len_for_encrypted_data_blocks(u16::MAX as u32 + 1, options);
6695
6696        let err =
6697            plan_index_root_metadata_class(options, 1, Some(dictionary_payload_len)).unwrap_err();
6698
6699        assert_eq!(
6700            err,
6701            FormatError::WriterUnsupported("dictionary object too large")
6702        );
6703    }
6704
6705    #[test]
6706    fn metadata_class_planning_rejects_gf16_total_overflow_for_dictionary() {
6707        let options = plan_writer_options(WriterOptions {
6708            block_size: MIN_BLOCK_SIZE,
6709            stripe_width: 8,
6710            volume_loss_tolerance: 1,
6711            bit_rot_buffer_pct: 5,
6712            ..WriterOptions::default()
6713        })
6714        .unwrap();
6715        let dictionary_payload_len = payload_len_for_encrypted_data_blocks(60_000, options);
6716
6717        let err =
6718            plan_index_root_metadata_class(options, 1, Some(dictionary_payload_len)).unwrap_err();
6719
6720        assert_eq!(
6721            err,
6722            FormatError::WriterUnsupported("dictionary object too large")
6723        );
6724    }
6725
6726    #[test]
6727    fn written_archive_authenticates_final_index_root_fec_class() {
6728        let master_key = MasterKey::from_raw_key(&[9u8; 32]).unwrap();
6729        let dictionary = deterministic_bytes(80 * 1024);
6730        let file = RegularFile::new("uses-dictionary.txt", b"payload");
6731        let archive = write_archive_with_dictionary(
6732            &[file],
6733            &master_key,
6734            WriterOptions {
6735                block_size: MIN_BLOCK_SIZE,
6736                stripe_width: 1,
6737                volume_loss_tolerance: 0,
6738                bit_rot_buffer_pct: 0,
6739                index_root_fec_parity_shards: 0,
6740                ..WriterOptions::default()
6741            },
6742            &dictionary,
6743        )
6744        .unwrap();
6745
6746        let volume_header = VolumeHeader::parse(&archive.bytes[..VOLUME_HEADER_LEN]).unwrap();
6747        let crypto_start = VOLUME_HEADER_LEN;
6748        let crypto_end = crypto_start + volume_header.crypto_header_length as usize;
6749        let crypto_header = CryptoHeader::parse(
6750            &archive.bytes[crypto_start..crypto_end],
6751            volume_header.crypto_header_length,
6752        )
6753        .unwrap();
6754        let subkeys =
6755            Subkeys::derive(&master_key, &archive.archive_uuid, &archive.session_id).unwrap();
6756        verify_hmac(
6757            HmacDomain::CryptoHeader,
6758            &subkeys.mac_key,
6759            &archive.archive_uuid,
6760            &archive.session_id,
6761            crypto_header.hmac_covered_bytes,
6762            &crypto_header.header_hmac,
6763        )
6764        .unwrap();
6765
6766        assert!(crypto_header.fixed.index_root_fec_data_shards > MIN_INDEX_ROOT_FEC_DATA_SHARDS);
6767        assert_eq!(crypto_header.fixed.index_root_fec_parity_shards, 0);
6768        let opened = open_archive(&archive.bytes, &master_key).unwrap();
6769        assert_eq!(
6770            opened.extract_file("uses-dictionary.txt").unwrap(),
6771            Some(b"payload".to_vec())
6772        );
6773        opened.verify().unwrap();
6774    }
6775
6776    #[test]
6777    fn object_parity_uses_per_object_recurrence_even_with_larger_class_max() {
6778        let options = WriterOptions {
6779            bit_rot_buffer_pct: 0,
6780            stripe_width: 1,
6781            volume_loss_tolerance: 0,
6782            fec_parity_shards: 1,
6783            ..WriterOptions::default()
6784        };
6785
6786        assert_eq!(compute_object_parity(1, options, 1).unwrap(), 0);
6787    }
6788
6789    #[test]
6790    fn object_total_shards_obeys_reed_solomon_limit() {
6791        assert!(validate_object_shard_total(65_535, 0).is_ok());
6792        assert_eq!(
6793            validate_object_shard_total(65_535, 1).unwrap_err(),
6794            FormatError::WriterUnsupported("encrypted object exceeds ReedSolomonGF16 shard limit")
6795        );
6796    }
6797
6798    #[test]
6799    fn argon2id_kdf_serialization_rejects_memory_requirement_overflow() {
6800        assert_eq!(
6801            serialize_kdf_params(&KdfParams::Argon2id {
6802                t_cost: 1,
6803                m_cost_kib: u32::MAX,
6804                parallelism: u32::MAX,
6805                salt: b"12345678".to_vec(),
6806            })
6807            .unwrap_err(),
6808            FormatError::InvalidKdfParams("m_cost_kib requirement overflow")
6809        );
6810    }
6811
6812    #[test]
6813    fn writer_options_reject_reader_resource_cap_excesses() {
6814        assert_eq!(
6815            plan_writer_options(WriterOptions {
6816                stripe_width: crate::format::READER_MAX_STRIPE_WIDTH + 1,
6817                volume_loss_tolerance: 0,
6818                ..WriterOptions::default()
6819            })
6820            .unwrap_err(),
6821            FormatError::ReaderResourceLimitExceeded {
6822                field: "stripe_width",
6823                cap: crate::format::READER_MAX_STRIPE_WIDTH as u64,
6824                actual: crate::format::READER_MAX_STRIPE_WIDTH as u64 + 1,
6825            }
6826        );
6827        assert_eq!(
6828            plan_writer_options(WriterOptions {
6829                block_size: crate::format::READER_MAX_BLOCK_SIZE + 2,
6830                ..WriterOptions::default()
6831            })
6832            .unwrap_err(),
6833            FormatError::ReaderResourceLimitExceeded {
6834                field: "block_size",
6835                cap: crate::format::READER_MAX_BLOCK_SIZE as u64,
6836                actual: crate::format::READER_MAX_BLOCK_SIZE as u64 + 2,
6837            }
6838        );
6839        assert_eq!(
6840            plan_writer_options(WriterOptions {
6841                chunk_size: crate::format::READER_MAX_CHUNK_SIZE + 1,
6842                envelope_target_size: crate::format::READER_MAX_CHUNK_SIZE + 1,
6843                ..WriterOptions::default()
6844            })
6845            .unwrap_err(),
6846            FormatError::ReaderResourceLimitExceeded {
6847                field: "chunk_size",
6848                cap: crate::format::READER_MAX_CHUNK_SIZE as u64,
6849                actual: crate::format::READER_MAX_CHUNK_SIZE as u64 + 1,
6850            }
6851        );
6852        assert_eq!(
6853            plan_writer_options(WriterOptions {
6854                max_path_length: crate::format::READER_MAX_PATH_LENGTH + 1,
6855                ..WriterOptions::default()
6856            })
6857            .unwrap_err(),
6858            FormatError::ReaderResourceLimitExceeded {
6859                field: "max_path_length",
6860                cap: crate::format::READER_MAX_PATH_LENGTH as u64,
6861                actual: crate::format::READER_MAX_PATH_LENGTH as u64 + 1,
6862            }
6863        );
6864        assert_eq!(
6865            plan_writer_options(WriterOptions {
6866                bit_rot_buffer_pct: 0,
6867                stripe_width: 1,
6868                volume_loss_tolerance: 0,
6869                fec_data_shards: crate::format::READER_MAX_FEC_CLASS_SHARDS as u16 + 1,
6870                ..WriterOptions::default()
6871            })
6872            .unwrap_err(),
6873            FormatError::ReaderResourceLimitExceeded {
6874                field: "fec_data_shards + fec_parity_shards",
6875                cap: crate::format::READER_MAX_FEC_CLASS_SHARDS as u64,
6876                actual: crate::format::READER_MAX_FEC_CLASS_SHARDS as u64 + 1,
6877            }
6878        );
6879    }
6880
6881    #[test]
6882    fn root_auth_writer_config_rejects_reader_cap_excess_before_authenticator() {
6883        let master_key = MasterKey::from_raw_key(&[7u8; 32]).unwrap();
6884        let mut authenticator_called = false;
6885        let err = write_archive_with_root_auth(
6886            &[RegularFile::new("signed.txt", b"payload")],
6887            &master_key,
6888            single_volume_metadata_test_options(),
6889            RootAuthWriterConfig {
6890                authenticator_id: 1,
6891                signer_identity_type: 1,
6892                signer_identity: b"signer",
6893                authenticator_value_length: READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN + 1,
6894            },
6895            |_| {
6896                authenticator_called = true;
6897                Ok(Vec::new())
6898            },
6899        )
6900        .unwrap_err();
6901
6902        assert!(!authenticator_called);
6903        assert_eq!(
6904            err,
6905            FormatError::ReaderResourceLimitExceeded {
6906                field: "RootAuthFooterV1 authenticator value length",
6907                cap: READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN as u64,
6908                actual: READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN as u64 + 1,
6909            }
6910        );
6911    }
6912
6913    #[test]
6914    fn root_auth_writer_accepts_128_kib_authenticator_value() {
6915        let master_key = MasterKey::from_raw_key(&[8u8; 32]).unwrap();
6916        let authenticator_value = vec![0x5a; READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN as usize];
6917        let expected_value = authenticator_value.clone();
6918        let archive = write_archive_with_root_auth(
6919            &[RegularFile::new("signed.txt", b"payload")],
6920            &master_key,
6921            single_volume_metadata_test_options(),
6922            RootAuthWriterConfig {
6923                authenticator_id: 0xcafe,
6924                signer_identity_type: 1,
6925                signer_identity: b"certificate-profile-signer",
6926                authenticator_value_length: READER_MAX_ROOT_AUTH_AUTHENTICATOR_VALUE_LEN,
6927            },
6928            |_| Ok(authenticator_value.clone()),
6929        )
6930        .unwrap();
6931
6932        let opened = open_archive(&archive.bytes, &master_key).unwrap();
6933        let footer = opened.root_auth_footer.as_ref().unwrap();
6934        assert_eq!(footer.authenticator_id, 0xcafe);
6935        assert_eq!(
6936            footer.authenticator_value.as_slice(),
6937            expected_value.as_slice()
6938        );
6939
6940        let verification = opened
6941            .verify_root_auth_with(|footer, _| {
6942                Ok(footer.authenticator_id == 0xcafe
6943                    && footer.authenticator_value.as_slice() == expected_value.as_slice())
6944            })
6945            .unwrap();
6946        assert_eq!(verification.authenticator_id, 0xcafe);
6947    }
6948
6949    #[test]
6950    fn streaming_writer_sink_round_trips_archive() {
6951        let files = [
6952            RegularFile::new("alpha.txt", b"alpha"),
6953            RegularFile::new("nested/beta.txt", b"beta payload"),
6954        ];
6955        let master_key = MasterKey::from_raw_key(&[7u8; 32]).unwrap();
6956        let mut sink = MemoryArchiveSink::default();
6957
6958        let summary = write_archive_sources_to_sink(
6959            &files,
6960            &master_key,
6961            single_volume_metadata_test_options(),
6962            None,
6963            &KdfParams::Raw,
6964            None,
6965            None,
6966            &mut sink,
6967        )
6968        .unwrap();
6969
6970        assert_eq!(summary.volume_count, 1);
6971        let opened = crate::reader::open_archive(&sink.volumes[0], &master_key).unwrap();
6972        assert_eq!(
6973            opened.extract_file("nested/beta.txt").unwrap(),
6974            Some(b"beta payload".to_vec())
6975        );
6976    }
6977
6978    #[test]
6979    fn streaming_writer_bounds_source_reads_and_sink_writes_for_large_file() {
6980        let file_size = 3 * 1024 * 1024;
6981        let stats = Rc::new(RefCell::new(GeneratedSourceStats::default()));
6982        let file = GeneratedFileSource {
6983            path: "large/generated.bin",
6984            len: file_size,
6985            stats: Rc::clone(&stats),
6986        };
6987        let master_key = MasterKey::from_raw_key(&[3u8; 32]).unwrap();
6988        let options = plan_writer_options(WriterOptions {
6989            block_size: MIN_BLOCK_SIZE,
6990            chunk_size: 16 * 1024,
6991            envelope_target_size: 64 * 1024,
6992            stripe_width: 1,
6993            volume_loss_tolerance: 0,
6994            bit_rot_buffer_pct: 0,
6995            fec_parity_shards: 0,
6996            index_fec_parity_shards: 0,
6997            index_root_fec_parity_shards: 0,
6998            ..WriterOptions::default()
6999        })
7000        .unwrap();
7001        let mut sink = TrackingArchiveSink::default();
7002
7003        let summary = write_archive_sources_to_sink_single_pass(
7004            &[file],
7005            &master_key,
7006            options,
7007            &KdfParams::Raw,
7008            None,
7009            None,
7010            &mut sink,
7011        )
7012        .unwrap();
7013
7014        let stats = stats.borrow();
7015        assert_eq!(stats.open_count, 1);
7016        assert_eq!(stats.total_read, file_size as u64);
7017        assert!(stats.max_read_request <= options.chunk_size as usize);
7018        assert_eq!(summary.volume_count, 1);
7019        assert_eq!(summary.archive_bytes, sink.volume_bytes.iter().sum());
7020        assert_eq!(
7021            summary.bootstrap_sidecar_bytes,
7022            sink.bootstrap_sidecar_bytes
7023        );
7024        assert!(sink.max_write_len <= 128 * 1024);
7025    }
7026
7027    #[test]
7028    fn sink_writer_progress_reports_source_bytes_during_multi_pass_write() {
7029        let file_size = 512 * 1024;
7030        let stats = Rc::new(RefCell::new(GeneratedSourceStats::default()));
7031        let file = GeneratedFileSource {
7032            path: "large/generated.bin",
7033            len: file_size,
7034            stats: Rc::clone(&stats),
7035        };
7036        let master_key = MasterKey::from_raw_key(&[4u8; 32]).unwrap();
7037        let options = WriterOptions {
7038            block_size: MIN_BLOCK_SIZE,
7039            chunk_size: 16 * 1024,
7040            envelope_target_size: 64 * 1024,
7041            stripe_width: 1,
7042            volume_loss_tolerance: 0,
7043            bit_rot_buffer_pct: 0,
7044            fec_parity_shards: 0,
7045            index_fec_parity_shards: 0,
7046            index_root_fec_parity_shards: 0,
7047            ..WriterOptions::default()
7048        };
7049        let mut sink = MemoryArchiveSink::default();
7050        let mut progress_events = Vec::new();
7051        let mut progress = |archive_path: &str, bytes: u64| {
7052            progress_events.push((archive_path.to_owned(), bytes));
7053        };
7054
7055        let summary = write_archive_sources_to_sink_with_progress(
7056            &[file],
7057            &master_key,
7058            options,
7059            None,
7060            &KdfParams::Raw,
7061            None,
7062            None,
7063            &mut sink,
7064            &mut progress,
7065        )
7066        .unwrap();
7067
7068        let reported_bytes = progress_events.iter().map(|(_, bytes)| *bytes).sum::<u64>();
7069        let stats = stats.borrow();
7070        assert!(stats.open_count > 1);
7071        assert_eq!(reported_bytes, file_size as u64);
7072        assert!(progress_events.len() > 1);
7073        assert!(progress_events
7074            .iter()
7075            .all(|(path, _)| path == "large/generated.bin"));
7076        assert_eq!(summary.volume_count, 1);
7077        assert!(!sink.volumes.is_empty());
7078    }
7079
7080    #[derive(Default)]
7081    struct GeneratedSourceStats {
7082        open_count: usize,
7083        total_read: u64,
7084        max_read_request: usize,
7085    }
7086
7087    struct GeneratedFileSource {
7088        path: &'static str,
7089        len: usize,
7090        stats: Rc<RefCell<GeneratedSourceStats>>,
7091    }
7092
7093    impl RegularFileSource for GeneratedFileSource {
7094        fn archive_path(&self) -> &str {
7095            self.path
7096        }
7097
7098        fn file_data_size(&self) -> u64 {
7099            self.len as u64
7100        }
7101
7102        fn mode(&self) -> u32 {
7103            0o644
7104        }
7105
7106        fn mtime(&self) -> u64 {
7107            0
7108        }
7109
7110        fn open(&self) -> Result<Box<dyn Read + '_>, ArchiveWriteError> {
7111            self.stats.borrow_mut().open_count += 1;
7112            Ok(Box::new(GeneratedReader {
7113                remaining: self.len,
7114                position: 0,
7115                stats: Rc::clone(&self.stats),
7116            }))
7117        }
7118    }
7119
7120    struct GeneratedReader {
7121        remaining: usize,
7122        position: usize,
7123        stats: Rc<RefCell<GeneratedSourceStats>>,
7124    }
7125
7126    impl Read for GeneratedReader {
7127        fn read(&mut self, out: &mut [u8]) -> io::Result<usize> {
7128            if self.remaining == 0 {
7129                return Ok(0);
7130            }
7131            let count = out.len().min(self.remaining);
7132            for (offset, byte) in out[..count].iter_mut().enumerate() {
7133                let position = self.position + offset;
7134                *byte = position.wrapping_mul(31).wrapping_add(17) as u8;
7135            }
7136            self.position += count;
7137            self.remaining -= count;
7138            let mut stats = self.stats.borrow_mut();
7139            stats.total_read += count as u64;
7140            stats.max_read_request = stats.max_read_request.max(out.len());
7141            Ok(count)
7142        }
7143    }
7144
7145    #[derive(Default)]
7146    struct TrackingArchiveSink {
7147        volume_bytes: Vec<u64>,
7148        bootstrap_sidecar_bytes: u64,
7149        max_write_len: usize,
7150    }
7151
7152    impl ArchiveWriteSink for TrackingArchiveSink {
7153        fn begin_archive(&mut self, volume_count: usize) -> Result<(), ArchiveWriteError> {
7154            self.volume_bytes = vec![0; volume_count];
7155            self.bootstrap_sidecar_bytes = 0;
7156            self.max_write_len = 0;
7157            Ok(())
7158        }
7159
7160        fn write_volume(
7161            &mut self,
7162            volume_index: usize,
7163            bytes: &[u8],
7164        ) -> Result<(), ArchiveWriteError> {
7165            let volume =
7166                self.volume_bytes
7167                    .get_mut(volume_index)
7168                    .ok_or(FormatError::WriterInvariant(
7169                        "tracking sink volume index is out of bounds",
7170                    ))?;
7171            *volume += bytes.len() as u64;
7172            self.max_write_len = self.max_write_len.max(bytes.len());
7173            Ok(())
7174        }
7175
7176        fn write_bootstrap_sidecar(&mut self, bytes: &[u8]) -> Result<(), ArchiveWriteError> {
7177            self.bootstrap_sidecar_bytes += bytes.len() as u64;
7178            self.max_write_len = self.max_write_len.max(bytes.len());
7179            Ok(())
7180        }
7181    }
7182
7183    fn deterministic_bytes(len: usize) -> Vec<u8> {
7184        let mut state = 0x4d41_4d45u32;
7185        let mut out = Vec::with_capacity(len);
7186        for _ in 0..len {
7187            state = state.wrapping_mul(1_664_525).wrapping_add(1_013_904_223);
7188            out.push((state >> 24) as u8);
7189        }
7190        out
7191    }
7192
7193    fn single_volume_metadata_test_options() -> WriterOptions {
7194        plan_writer_options(WriterOptions {
7195            block_size: MIN_BLOCK_SIZE,
7196            stripe_width: 1,
7197            volume_loss_tolerance: 0,
7198            bit_rot_buffer_pct: 0,
7199            index_root_fec_parity_shards: 0,
7200            ..WriterOptions::default()
7201        })
7202        .unwrap()
7203    }
7204
7205    fn payload_len_for_encrypted_data_blocks(
7206        data_block_count: u32,
7207        options: WriterOptions,
7208    ) -> usize {
7209        assert!(data_block_count > 0);
7210        if data_block_count == 1 {
7211            return 1;
7212        }
7213        let block_size = options.block_size as usize;
7214        (data_block_count as usize - 1) * block_size - options.aead_algo.tag_len() + 1
7215    }
7216
7217    fn assert_path_specific_member_group(group: &[u8]) {
7218        let mut cursor = 0usize;
7219        let mut saw_main = false;
7220        while cursor < group.len() {
7221            let header = &group[cursor..cursor + TAR_BLOCK_LEN];
7222            assert!(
7223                header.iter().any(|byte| *byte != 0),
7224                "writer emitted tar zero block inside member group"
7225            );
7226            let typeflag = header[156];
7227            assert_ne!(typeflag, b'g', "writer emitted global PAX metadata");
7228            assert!(
7229                !matches!(typeflag, b'V' | b'M' | b'N'),
7230                "writer emitted global GNU metadata"
7231            );
7232            assert!(
7233                matches!(typeflag, b'x' | b'0'),
7234                "writer emitted unexpected tar record type {typeflag:?}"
7235            );
7236            if typeflag == b'0' {
7237                saw_main = true;
7238            }
7239            let size = read_test_tar_octal(&header[124..136]);
7240            cursor += TAR_BLOCK_LEN + size + padding_to_512(size);
7241        }
7242        assert_eq!(cursor, group.len());
7243        assert!(saw_main);
7244    }
7245
7246    fn read_test_tar_octal(field: &[u8]) -> usize {
7247        let mut value = 0usize;
7248        for byte in field {
7249            match *byte {
7250                0 | b' ' => break,
7251                b'0'..=b'7' => {
7252                    value = value * 8 + usize::from(*byte - b'0');
7253                }
7254                other => panic!("malformed test tar octal byte {other:?}"),
7255            }
7256        }
7257        value
7258    }
7259
7260    fn test_file_row(idx: usize, path_hash: [u8; 8]) -> FileRow {
7261        let path = format!("file-{idx:05}.txt").into_bytes();
7262        FileRow {
7263            path_hash,
7264            path: path.clone(),
7265            member_index: idx,
7266            member: TarMember {
7267                path,
7268                tar_member_group_start: idx as u64 * 512,
7269                tar_member_group_size: 512,
7270                file_data_size: 0,
7271                mode: 0o644,
7272                mtime: 0,
7273            },
7274        }
7275    }
7276}