syncweb_core/folder/
archive_export.rs

1use std::{
2    collections::BTreeSet,
3    path::{Path, PathBuf},
4    sync::Arc,
5};
6
7use async_compression::tokio::write::ZstdEncoder;
8use iroh_blobs::Hash;
9use semver::Version;
10use serde::{Deserialize, Serialize};
11use tokio::{
12    fs,
13    io::{AsyncReadExt, AsyncWrite, AsyncWriteExt},
14    sync::Mutex,
15};
16use uuid::Uuid;
17
18use crate::{
19    daemon::ManagedPool,
20    error::{Result, SyncwebError},
21    filter::{FilterAction, FilterEngine, FilterEntry},
22    node::blob_store::BlobStore,
23};
24
25use super::{CollectionEntry, CollectionManifest};
26
27const CAR_VERSION: u64 = 1;
28const CID_VERSION: u64 = 1;
29const RAW_CODEC: u64 = 0x55;
30const BLAKE3_MULTIHASH: u64 = 0x1e;
31const HASH_SIZE: u64 = 32;
32const COPY_BUFFER_SIZE: usize = 64 * 1024;
33
34/// Options controlling which collection version and entries are included.
35#[derive(Clone, Debug, Default)]
36#[non_exhaustive]
37pub struct DropExportOptions {
38    version: Option<String>,
39    filter: Option<FilterEngine>,
40}
41
42impl DropExportOptions {
43    #[must_use]
44    pub fn with_version(mut self, version: impl Into<String>) -> Self {
45        self.version = Some(version.into());
46        self
47    }
48
49    #[must_use]
50    pub fn with_filter(mut self, filter: FilterEngine) -> Self {
51        self.filter = Some(filter);
52        self
53    }
54
55    #[must_use]
56    pub fn version(&self) -> Option<&str> {
57        self.version.as_deref()
58    }
59
60    #[must_use]
61    pub const fn filter(&self) -> Option<&FilterEngine> {
62        self.filter.as_ref()
63    }
64}
65
66/// Information about a completed archive export.
67#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
68#[non_exhaustive]
69pub struct DropExportResult {
70    pub output: PathBuf,
71    pub collection_id: uuid::Uuid,
72    pub version: String,
73    pub manifest: Hash,
74    pub entry_count: usize,
75    pub block_count: usize,
76    pub archive_size: u64,
77}
78
79/// Streams collection manifests and content blobs into atomic `.car.zst` files.
80#[derive(Clone)]
81pub struct DropExporter {
82    blob_store: BlobStore,
83    export_lock: Arc<Mutex<()>>,
84}
85
86impl DropExporter {
87    #[must_use]
88    pub fn new(blob_store: BlobStore) -> Self {
89        Self {
90            blob_store,
91            export_lock: Arc::new(Mutex::new(())),
92        }
93    }
94
95    /// Export one manifest without filtering.
96    ///
97    /// # Errors
98    ///
99    /// Returns an error if the manifest is invalid, content is unavailable, or
100    /// the archive cannot be written.
101    pub async fn export_archive(
102        &self,
103        manifest: &CollectionManifest,
104        output: impl AsRef<Path>,
105    ) -> Result<DropExportResult> {
106        self.export_drop_with_options(
107            std::slice::from_ref(manifest),
108            output,
109            DropExportOptions::default(),
110            None,
111        )
112        .await
113    }
114
115    /// Export one or more versions, selecting the requested version or latest
116    /// semver version when no version is specified.
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if manifests disagree about their collection, the
121    /// requested version is unavailable, or the archive cannot be written.
122    pub async fn export_manifests(
123        &self,
124        manifests: &[CollectionManifest],
125        output: impl AsRef<Path>,
126        options: DropExportOptions,
127    ) -> Result<DropExportResult> {
128        self.export_drop_with_options(manifests, output, options, None).await
129    }
130
131    /// Export a manifest using filtering and version-selection options.
132    ///
133    /// A filtered manifest is deliberately unsigned because changing its
134    /// entries changes the signed content. The archive remains content
135    /// addressed and each included blob is verified while it is streamed.
136    ///
137    /// # Errors
138    ///
139    /// Returns an error if the selected manifest or any referenced blob is
140    /// invalid or unavailable, or if the archive cannot be written.
141    pub async fn export_drop_with_options(
142        &self,
143        manifests: &[CollectionManifest],
144        output: impl AsRef<Path>,
145        options: DropExportOptions,
146        pool: Option<&ManagedPool>,
147    ) -> Result<DropExportResult> {
148        let _export_guard = self.export_lock.lock().await;
149        let selected = in_pool(pool, || select_manifest(manifests, options.version()))?;
150        let filtered = in_pool(pool, || filter_manifest(selected, options.filter()))?;
151        let (manifest_bytes, manifest_hash, entries) = in_pool(pool, || -> Result<_> {
152            let manifest_bytes = filtered.to_bytes()?;
153            let manifest_hash = Hash::new(&manifest_bytes);
154            let entries = unique_entries(&filtered.entries);
155            Ok((manifest_bytes, manifest_hash, entries))
156        })?;
157        for entry in &entries {
158            if !self.blob_store.has(entry.content_id).await? {
159                return Err(SyncwebError::InvalidConfig(format!(
160                    "drop content is missing from the blob store: {}",
161                    entry.logical_path.display()
162                )));
163            }
164        }
165
166        let output_path = output.as_ref().to_path_buf();
167        let parent = output_path.parent().unwrap_or_else(|| Path::new("."));
168        fs::create_dir_all(parent).await?;
169        let staging = parent.join(format!(".syncweb-drop-{}", Uuid::new_v4()));
170        let write_result = self
171            .write_archive(&staging, &manifest_bytes, manifest_hash, &entries, pool)
172            .await;
173        if let Err(error) = write_result {
174            remove_if_present(&staging).await?;
175            return Err(error);
176        }
177        if let Err(error) = fs::rename(&staging, &output_path).await {
178            remove_if_present(&staging).await?;
179            return Err(SyncwebError::operation("failed to finalize drop archive", error));
180        }
181        let archive_size = fs::metadata(&output_path)
182            .await
183            .map_err(|error| SyncwebError::operation("failed to inspect drop archive", error))?
184            .len();
185        Ok(DropExportResult {
186            output: output_path,
187            collection_id: filtered.collection_id,
188            version: filtered.version,
189            manifest: manifest_hash,
190            entry_count: filtered.entries.len(),
191            block_count: entries.len().saturating_add(1),
192            archive_size,
193        })
194    }
195
196    async fn write_archive(
197        &self,
198        staging: &Path,
199        manifest_bytes: &[u8],
200        manifest_hash: Hash,
201        entries: &[CollectionEntry],
202        pool: Option<&ManagedPool>,
203    ) -> Result<()> {
204        let file = fs::File::create(staging)
205            .await
206            .map_err(|error| SyncwebError::operation("failed to create drop archive", error))?;
207        let mut encoder = ZstdEncoder::new(file);
208        let header = car_header(manifest_hash)?;
209        encoder.write_all(&header).await?;
210        write_section(&mut encoder, manifest_hash, manifest_bytes).await?;
211        let temporary_dir = staging.parent().unwrap_or_else(|| Path::new("."));
212        for entry in entries {
213            write_blob_section(&self.blob_store, &mut encoder, entry, temporary_dir, pool).await?;
214        }
215        encoder
216            .shutdown()
217            .await
218            .map_err(|error| SyncwebError::operation("failed to finish drop compression", error))?;
219        Ok(())
220    }
221}
222
223/// Export one collection manifest into a compressed CAR archive.
224///
225/// # Errors
226///
227/// Returns an error if the manifest or referenced content is invalid or the
228/// archive cannot be written.
229pub async fn export_archive(
230    blob_store: BlobStore,
231    manifest: &CollectionManifest,
232    output: impl AsRef<Path>,
233    filter: Option<&FilterEngine>,
234) -> Result<DropExportResult> {
235    let options = filter.map_or_else(DropExportOptions::default, |value| {
236        DropExportOptions::default().with_filter(value.clone())
237    });
238    DropExporter::new(blob_store)
239        .export_drop_with_options(std::slice::from_ref(manifest), output, options, None)
240        .await
241}
242
243fn in_pool<F, R>(pool: Option<&ManagedPool>, operation: F) -> R
244where
245    F: FnOnce() -> R + Send,
246    R: Send,
247{
248    match pool {
249        Some(managed_pool) => managed_pool.install(operation),
250        None => operation(),
251    }
252}
253
254fn select_manifest(manifests: &[CollectionManifest], requested: Option<&str>) -> Result<CollectionManifest> {
255    let mut candidates = manifests.iter();
256    let Some(first) = candidates.next() else {
257        return Err(SyncwebError::InvalidConfig(
258            "at least one collection manifest is required".to_owned(),
259        ));
260    };
261    first.validate()?;
262    let collection_id = first.collection_id;
263    let mut selected = first.clone();
264    for manifest in candidates {
265        manifest.validate()?;
266        if manifest.collection_id != collection_id {
267            return Err(SyncwebError::InvalidConfig(
268                "drop manifests must belong to the same collection".to_owned(),
269            ));
270        }
271        if requested.is_none() && manifest.is_upgrade_from(&selected)? {
272            selected = manifest.clone();
273        }
274    }
275    if let Some(version) = requested {
276        Version::parse(version)
277            .map_err(|error| SyncwebError::InvalidConfig(format!("invalid requested collection version: {error}")))?;
278        selected = manifests
279            .iter()
280            .find(|manifest| manifest.version == version)
281            .cloned()
282            .ok_or_else(|| SyncwebError::InvalidConfig(format!("collection version {version} is unavailable")))?;
283        if selected.collection_id != collection_id {
284            return Err(SyncwebError::InvalidConfig(
285                "drop manifests must belong to the same collection".to_owned(),
286            ));
287        }
288    }
289    Ok(selected)
290}
291
292fn filter_manifest(manifest: CollectionManifest, filter: Option<&FilterEngine>) -> Result<CollectionManifest> {
293    let Some(engine) = filter else {
294        return Ok(manifest);
295    };
296    let version = Version::parse(&manifest.version)
297        .map_err(|error| SyncwebError::InvalidConfig(format!("invalid collection version: {error}")))?;
298    let mut filtered = manifest;
299    filtered.entries.retain(|entry| {
300        let filter_entry = FilterEntry::new(&entry.logical_path, entry.size).with_version(version.clone());
301        engine.evaluate(&filter_entry) == FilterAction::Accept
302    });
303    filtered.signature = None;
304    filtered.public_key = None;
305    filtered.validate()?;
306    Ok(filtered)
307}
308
309fn unique_entries(entries: &[CollectionEntry]) -> Vec<CollectionEntry> {
310    let mut hashes = BTreeSet::new();
311    entries
312        .iter()
313        .filter(|entry| hashes.insert(entry.content_id))
314        .cloned()
315        .collect()
316}
317
318async fn write_blob_section(
319    blob_store: &BlobStore,
320    encoder: &mut ZstdEncoder<fs::File>,
321    entry: &CollectionEntry,
322    temporary_dir: &Path,
323    pool: Option<&ManagedPool>,
324) -> Result<()> {
325    let temporary_path = temporary_dir.join(format!(".syncweb-drop-blob-{}", Uuid::new_v4()));
326    let result = stream_blob(blob_store, encoder, entry, &temporary_path, pool).await;
327    let cleanup = remove_if_present(&temporary_path).await;
328    match (result, cleanup) {
329        (Ok(()), Ok(())) => Ok(()),
330        (Err(error), Ok(())) | (Ok(()), Err(error)) => Err(error),
331        (Err(error), Err(cleanup_error)) => Err(SyncwebError::operation(
332            "failed to clean up temporary drop blob",
333            format!("{error}; cleanup failed: {cleanup_error}"),
334        )),
335    }
336}
337
338async fn stream_blob(
339    blob_store: &BlobStore,
340    encoder: &mut ZstdEncoder<fs::File>,
341    entry: &CollectionEntry,
342    temporary_path: &Path,
343    pool: Option<&ManagedPool>,
344) -> Result<()> {
345    blob_store.export_to_path(entry.content_id, temporary_path).await?;
346    let metadata = fs::metadata(temporary_path)
347        .await
348        .map_err(|error| SyncwebError::operation("failed to inspect exported drop blob", error))?;
349    if metadata.len() != entry.size {
350        return Err(SyncwebError::InvalidConfig(format!(
351            "drop blob size does not match manifest for {}",
352            entry.logical_path.display()
353        )));
354    }
355    write_section_prefix(encoder, entry.content_id, entry.size).await?;
356    let mut source = fs::File::open(temporary_path)
357        .await
358        .map_err(|error| SyncwebError::operation("failed to open exported drop blob", error))?;
359    let mut buffer = vec![0_u8; COPY_BUFFER_SIZE];
360    let mut hasher = blake3::Hasher::new();
361    let mut copied = 0_u64;
362    loop {
363        let read = source.read(&mut buffer).await?;
364        if read == 0 {
365            break;
366        }
367        let chunk = buffer
368            .get(..read)
369            .ok_or_else(|| SyncwebError::operation("blob reader returned an invalid chunk", read))?;
370        encoder.write_all(chunk).await?;
371        in_pool(pool, || hasher.update(chunk));
372        copied = copied
373            .checked_add(u64::try_from(read).map_err(|error| SyncwebError::operation("drop blob is too large", error))?)
374            .ok_or_else(|| SyncwebError::operation("drop blob size overflow", "u64 limit exceeded"))?;
375    }
376    let actual = in_pool(pool, || Hash::from_bytes(*hasher.finalize().as_bytes()));
377    if copied != entry.size || actual != entry.content_id {
378        return Err(SyncwebError::InvalidConfig(format!(
379            "drop blob hash does not match manifest for {}",
380            entry.logical_path.display()
381        )));
382    }
383    Ok(())
384}
385
386async fn write_section<W: AsyncWrite + Unpin>(writer: &mut W, hash: Hash, bytes: &[u8]) -> Result<()> {
387    write_section_prefix(
388        writer,
389        hash,
390        u64::try_from(bytes.len())
391            .map_err(|error| SyncwebError::operation("manifest is too large for a drop archive", error))?,
392    )
393    .await?;
394    writer.write_all(bytes).await?;
395    Ok(())
396}
397
398async fn write_section_prefix<W: AsyncWrite + Unpin>(writer: &mut W, hash: Hash, size: u64) -> Result<()> {
399    let cid = cid_for_hash(hash);
400    let section_size = u64::try_from(cid.len())
401        .map_err(|error| SyncwebError::operation("CAR section is too large", error))?
402        .checked_add(size)
403        .ok_or_else(|| SyncwebError::operation("CAR section size overflow", "u64 limit exceeded"))?;
404    let mut encoded_size = Vec::with_capacity(10);
405    encode_varint(section_size, &mut encoded_size);
406    writer.write_all(&encoded_size).await?;
407    writer.write_all(&cid).await?;
408    Ok(())
409}
410
411fn car_header(root: Hash) -> Result<Vec<u8>> {
412    let cid = cid_for_hash(root);
413    let cid_length =
414        u64::try_from(cid.len()).map_err(|error| SyncwebError::operation("CAR root CID is too large", error))?;
415    let mut header = Vec::with_capacity(cid.len().saturating_add(8));
416    header.push(0x0a);
417    encode_varint(cid_length, &mut header);
418    header.extend_from_slice(&cid);
419    header.push(0x10);
420    encode_varint(CAR_VERSION, &mut header);
421    let header_length =
422        u64::try_from(header.len()).map_err(|error| SyncwebError::operation("CAR header is too large", error))?;
423    let mut encoded = Vec::with_capacity(header.len().saturating_add(10));
424    encode_varint(header_length, &mut encoded);
425    encoded.extend_from_slice(&header);
426    Ok(encoded)
427}
428
429fn cid_for_hash(hash: Hash) -> Vec<u8> {
430    let mut cid = Vec::with_capacity(2 + 2 + 2 + 32);
431    encode_varint(CID_VERSION, &mut cid);
432    encode_varint(RAW_CODEC, &mut cid);
433    encode_varint(BLAKE3_MULTIHASH, &mut cid);
434    encode_varint(HASH_SIZE, &mut cid);
435    cid.extend_from_slice(hash.as_bytes());
436    cid
437}
438
439fn encode_varint(mut value: u64, output: &mut Vec<u8>) {
440    while value >= 0x80 {
441        output.push(value.to_le_bytes()[0] | 0x80);
442        value >>= 7;
443    }
444    output.push(value.to_le_bytes()[0]);
445}
446
447async fn remove_if_present(path: &Path) -> Result<()> {
448    match fs::remove_file(path).await {
449        Ok(()) => Ok(()),
450        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
451        Err(error) => Err(SyncwebError::operation(
452            "failed to remove temporary archive file",
453            error,
454        )),
455    }
456}
syncweb_core/folder/archive_export.rs

syncweb_core/folder/
archive_export.rs
