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
168pub 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
202pub trait ArchiveWriteProgressSink {
208 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
315pub 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#[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#[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(¶llelism.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}