takanawa_core/
part_file.rs

1use std::ffi::OsString;
2use std::fs::{self, File, OpenOptions};
3use std::io::{Read, Seek, SeekFrom, Write};
4use std::path::{Path, PathBuf};
5
6use fs2::FileExt;
7
8use crate::chunk::{ChunkPlan, normalize_chunk_size};
9use crate::metadata::{PartMetadata, RemoteInfo, slot_size_for};
10use crate::{HashConfig, HashVerifier, Result, TakanawaError, hash_url};
11
12/// Resumable on-disk part file with metadata slots and an exclusive lock.
13#[derive(Debug)]
14pub struct PartFile {
15    file: File,
16    lock_file: File,
17    lock_path: PathBuf,
18    part_path: PathBuf,
19    slot_size: u64,
20    active_slot: u8,
21    metadata: PartMetadata,
22}
23
24impl PartFile {
25    /// Opens an existing compatible part file or creates a new one.
26    ///
27    /// The target file must not already exist. The companion `.part.lock` file
28    /// is locked for the lifetime of the returned value.
29    ///
30    /// # Errors
31    ///
32    /// Returns an error if the target exists, the part file is locked by
33    /// another process, existing metadata is corrupt or incompatible, the part
34    /// file size is unexpected, or filesystem operations fail.
35    pub fn open_or_create(
36        target_path: &Path,
37        url: &str,
38        remote: &RemoteInfo,
39        chunk_size: u64,
40        hash: HashConfig,
41    ) -> Result<Self> {
42        if target_path.exists() {
43            return Err(TakanawaError::TargetExists(target_path.to_owned()));
44        }
45
46        let chunk_size = normalize_chunk_size(chunk_size)?;
47        let part_path = part_path_for(target_path);
48        let lock_path = part_lock_path_for(target_path);
49        let lock_file = acquire_lock(&lock_path)?;
50        let slot_size = slot_size_for(remote.content_len, chunk_size)?;
51        let expected_len = remote
52            .content_len
53            .checked_add(slot_size.checked_mul(2).ok_or_else(|| {
54                TakanawaError::InvalidConfig("part file length overflow".to_owned())
55            })?)
56            .ok_or_else(|| TakanawaError::InvalidConfig("part file length overflow".to_owned()))?;
57        let url_hash = hash_url(url);
58
59        if part_path.exists() {
60            let mut file = OpenOptions::new().read(true).write(true).open(&part_path)?;
61            let actual_len = file.metadata()?.len();
62            if actual_len != expected_len {
63                return Err(TakanawaError::PartSizeMismatch {
64                    expected: expected_len,
65                    actual: actual_len,
66                });
67            }
68
69            let (metadata, active_slot) =
70                read_best_metadata(&mut file, remote.content_len, slot_size)?;
71            metadata.ensure_compatible(url_hash, remote, chunk_size, hash)?;
72            return Ok(Self {
73                file,
74                lock_file,
75                lock_path,
76                part_path,
77                slot_size,
78                active_slot,
79                metadata,
80            });
81        }
82
83        let mut file = OpenOptions::new()
84            .read(true)
85            .write(true)
86            .create_new(true)
87            .open(&part_path)?;
88        file.set_len(expected_len)?;
89
90        let metadata = PartMetadata::new(url_hash, remote, chunk_size, hash)?;
91        let slot = metadata.encode_slot(slot_size)?;
92        file.seek(SeekFrom::Start(remote.content_len))?;
93        file.write_all(&slot)?;
94        file.sync_all()?;
95
96        Ok(Self {
97            file,
98            lock_file,
99            lock_path,
100            part_path,
101            slot_size,
102            active_slot: 0,
103            metadata,
104        })
105    }
106
107    #[must_use]
108    /// Returns the current part metadata.
109    pub const fn metadata(&self) -> &PartMetadata {
110        &self.metadata
111    }
112
113    #[must_use]
114    /// Returns indexes of chunks that still need to be downloaded.
115    pub fn incomplete_chunks(&self) -> Vec<u64> {
116        self.metadata.bitmap.incomplete_indices()
117    }
118
119    /// Writes and commits a complete chunk.
120    ///
121    /// Already completed chunks are ignored.
122    ///
123    /// # Errors
124    ///
125    /// Returns an error if `index` is outside the chunk plan, `bytes` does not
126    /// exactly match the chunk length, metadata cannot be updated, or I/O fails.
127    pub fn write_chunk(&mut self, index: u64, bytes: &[u8]) -> Result<()> {
128        let plan = ChunkPlan::new(self.metadata.content_len, self.metadata.chunk_size)?;
129        let chunk = plan.chunk(index)?;
130        if bytes.len() != usize::try_from(chunk.len).unwrap_or(usize::MAX) {
131            return Err(TakanawaError::HttpProtocol(format!(
132                "chunk {index} length mismatch: expected {}, got {}",
133                chunk.len,
134                bytes.len()
135            )));
136        }
137        if self.metadata.bitmap.is_complete(index)? {
138            return Ok(());
139        }
140
141        self.write_chunk_bytes(index, 0, bytes)?;
142        self.commit_chunk(index)
143    }
144
145    /// Writes bytes into a chunk without marking the chunk complete.
146    ///
147    /// This supports streaming partial responses. Call [`Self::commit_chunk`]
148    /// only after the full chunk has been written.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error if the chunk index or write range is invalid, the byte
153    /// length cannot fit in file offsets, or I/O fails.
154    pub fn write_chunk_bytes(&mut self, index: u64, chunk_offset: u64, bytes: &[u8]) -> Result<()> {
155        let plan = ChunkPlan::new(self.metadata.content_len, self.metadata.chunk_size)?;
156        let chunk = plan.chunk(index)?;
157        let len = u64::try_from(bytes.len()).map_err(|_| {
158            TakanawaError::InvalidConfig(format!(
159                "chunk {index} write length does not fit in file offsets"
160            ))
161        })?;
162        let end = chunk_offset.checked_add(len).ok_or_else(|| {
163            TakanawaError::InvalidConfig(format!("chunk {index} write offset overflow"))
164        })?;
165        if end > chunk.len {
166            return Err(TakanawaError::InvalidConfig(format!(
167                "chunk {index} write range {chunk_offset}..{end} exceeds chunk length {}",
168                chunk.len
169            )));
170        }
171        if bytes.is_empty() || self.metadata.bitmap.is_complete(index)? {
172            return Ok(());
173        }
174
175        self.file
176            .seek(SeekFrom::Start(chunk.start + chunk_offset))?;
177        self.file.write_all(bytes)?;
178        Ok(())
179    }
180
181    /// Marks a previously written chunk complete and persists metadata.
182    ///
183    /// Already completed chunks are ignored.
184    ///
185    /// # Errors
186    ///
187    /// Returns an error if `index` is outside the chunk plan, metadata
188    /// generation overflows, metadata cannot be encoded, or I/O fails.
189    pub fn commit_chunk(&mut self, index: u64) -> Result<()> {
190        let plan = ChunkPlan::new(self.metadata.content_len, self.metadata.chunk_size)?;
191        let _chunk = plan.chunk(index)?;
192        if self.metadata.bitmap.is_complete(index)? {
193            return Ok(());
194        }
195
196        self.file.sync_data()?;
197
198        self.metadata.bitmap.mark_complete(index)?;
199        self.commit_metadata()
200    }
201
202    /// Verifies and promotes the part file to the final target path.
203    ///
204    /// This consumes the part file, truncates away metadata slots, renames the
205    /// `.part` file to `target_path`, and releases the lock.
206    ///
207    /// # Errors
208    ///
209    /// Returns an error if the target exists, not all chunks are complete, hash
210    /// verification fails, or filesystem operations fail.
211    pub fn finalize(mut self, target_path: &Path) -> Result<()> {
212        if target_path.exists() {
213            return Err(TakanawaError::TargetExists(target_path.to_owned()));
214        }
215        if !self.metadata.all_complete() {
216            return Err(TakanawaError::InvalidConfig(
217                "cannot finalize an incomplete part file".to_owned(),
218            ));
219        }
220
221        if !self.verify_hash()? {
222            return Err(TakanawaError::HashMismatch);
223        }
224
225        let PartFile {
226            file,
227            lock_file,
228            lock_path,
229            part_path,
230            metadata,
231            ..
232        } = self;
233        file.set_len(metadata.content_len)?;
234        file.sync_all()?;
235        drop(file);
236        fs::rename(&part_path, target_path)?;
237        sync_parent_dir(target_path);
238        drop(lock_file);
239        let _ = fs::remove_file(lock_path);
240        Ok(())
241    }
242
243    fn commit_metadata(&mut self) -> Result<()> {
244        self.metadata.generation = self.metadata.generation.checked_add(1).ok_or_else(|| {
245            TakanawaError::InvalidConfig("metadata generation overflow".to_owned())
246        })?;
247        self.active_slot = (self.metadata.generation % 2) as u8;
248        let slot = self.metadata.encode_slot(self.slot_size)?;
249        let offset = self.metadata.content_len + u64::from(self.active_slot) * self.slot_size;
250        self.file.seek(SeekFrom::Start(offset))?;
251        self.file.write_all(&slot)?;
252        self.file.sync_all()?;
253        Ok(())
254    }
255
256    fn verify_hash(&mut self) -> Result<bool> {
257        let Some(mut verifier) = HashVerifier::new(self.metadata.hash) else {
258            return Ok(true);
259        };
260
261        let mut remaining = self.metadata.content_len;
262        let mut buffer = vec![0; 1024 * 1024];
263        self.file.seek(SeekFrom::Start(0))?;
264        while remaining > 0 {
265            let read_len = usize::try_from(remaining.min(buffer.len() as u64))
266                .expect("bounded by buffer length");
267            self.file.read_exact(&mut buffer[..read_len])?;
268            verifier.update(&buffer[..read_len]);
269            remaining -= read_len as u64;
270        }
271        Ok(verifier.finish())
272    }
273}
274
275#[must_use]
276/// Returns the companion `.part` path for a target file.
277pub fn part_path_for(target_path: &Path) -> PathBuf {
278    let mut value: OsString = target_path.as_os_str().to_owned();
279    value.push(".part");
280    PathBuf::from(value)
281}
282
283#[must_use]
284/// Returns the companion `.part.lock` path for a target file.
285pub fn part_lock_path_for(target_path: &Path) -> PathBuf {
286    let mut value: OsString = target_path.as_os_str().to_owned();
287    value.push(".part.lock");
288    PathBuf::from(value)
289}
290
291fn acquire_lock(lock_path: &Path) -> Result<File> {
292    let lock_file = OpenOptions::new()
293        .read(true)
294        .write(true)
295        .create(true)
296        .truncate(false)
297        .open(lock_path)?;
298    match lock_file.try_lock_exclusive() {
299        Ok(()) => Ok(lock_file),
300        Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
301            Err(TakanawaError::PartBusy(lock_path.to_owned()))
302        }
303        Err(err) => Err(TakanawaError::Io(err)),
304    }
305}
306
307fn read_best_metadata(
308    file: &mut File,
309    content_len: u64,
310    slot_size: u64,
311) -> Result<(PartMetadata, u8)> {
312    let slot_len = usize::try_from(slot_size)
313        .map_err(|_| TakanawaError::PartCorrupt("slot size overflow".to_owned()))?;
314    let mut slots = Vec::new();
315
316    for slot_index in 0..2_u8 {
317        let offset = content_len + u64::from(slot_index) * slot_size;
318        let mut buffer = vec![0; slot_len];
319        file.seek(SeekFrom::Start(offset))?;
320        file.read_exact(&mut buffer)?;
321        if let Ok(metadata) = PartMetadata::decode_slot(&buffer) {
322            slots.push((metadata, slot_index));
323        }
324    }
325
326    slots
327        .into_iter()
328        .max_by_key(|(metadata, _)| metadata.generation)
329        .ok_or_else(|| TakanawaError::PartCorrupt("no valid metadata slot found".to_owned()))
330}
331
332#[cfg(unix)]
333fn sync_parent_dir(target_path: &Path) {
334    if let Some(parent) = target_path.parent() {
335        if let Ok(dir) = File::open(parent) {
336            let _ = dir.sync_all();
337        }
338    }
339}
340
341#[cfg(not(unix))]
342fn sync_parent_dir(_target_path: &Path) {}
343
344#[cfg(test)]
345mod tests {
346    use std::fs;
347
348    use tempfile::TempDir;
349
350    use super::*;
351
352    fn hex_array<const N: usize>(value: impl AsRef<str>) -> [u8; N] {
353        hex::decode(value.as_ref()).unwrap().try_into().unwrap()
354    }
355
356    fn remote(content_len: u64) -> RemoteInfo {
357        RemoteInfo {
358            content_len,
359            etag: Some("etag".to_owned()),
360            last_modified: Some("now".to_owned()),
361        }
362    }
363
364    #[test]
365    fn resumes_valid_part() {
366        let dir = TempDir::new().unwrap();
367        let target = dir.path().join("file.bin");
368        {
369            let mut part = PartFile::open_or_create(
370                &target,
371                "https://example.test/file",
372                &remote(6),
373                3,
374                HashConfig::None,
375            )
376            .unwrap();
377            part.write_chunk(0, b"abc").unwrap();
378        }
379
380        let part = PartFile::open_or_create(
381            &target,
382            "https://example.test/file",
383            &remote(6),
384            3,
385            HashConfig::None,
386        )
387        .unwrap();
388
389        assert_eq!(part.metadata().completed_chunks(), 1);
390        assert_eq!(part.incomplete_chunks(), vec![1]);
391    }
392
393    #[test]
394    fn partial_chunk_write_is_not_committed_on_reopen() {
395        let dir = TempDir::new().unwrap();
396        let target = dir.path().join("file.bin");
397        {
398            let mut part = PartFile::open_or_create(
399                &target,
400                "https://example.test/file",
401                &remote(6),
402                3,
403                HashConfig::None,
404            )
405            .unwrap();
406            part.write_chunk_bytes(0, 0, b"ab").unwrap();
407        }
408
409        let part = PartFile::open_or_create(
410            &target,
411            "https://example.test/file",
412            &remote(6),
413            3,
414            HashConfig::None,
415        )
416        .unwrap();
417
418        assert_eq!(part.metadata().completed_chunks(), 0);
419        assert_eq!(part.incomplete_chunks(), vec![0, 1]);
420    }
421
422    #[test]
423    fn partial_chunk_can_be_overwritten_and_committed() {
424        let dir = TempDir::new().unwrap();
425        let target = dir.path().join("file.bin");
426        let mut part = PartFile::open_or_create(
427            &target,
428            "https://example.test/file",
429            &remote(6),
430            3,
431            HashConfig::None,
432        )
433        .unwrap();
434        part.write_chunk_bytes(0, 0, b"xx").unwrap();
435        part.write_chunk_bytes(0, 0, b"abc").unwrap();
436        part.commit_chunk(0).unwrap();
437        part.write_chunk(1, b"def").unwrap();
438        part.finalize(&target).unwrap();
439
440        assert_eq!(fs::read(&target).unwrap(), b"abcdef");
441    }
442
443    #[test]
444    fn partial_chunk_write_rejects_out_of_bounds_ranges() {
445        let dir = TempDir::new().unwrap();
446        let target = dir.path().join("file.bin");
447        let mut part = PartFile::open_or_create(
448            &target,
449            "https://example.test/file",
450            &remote(6),
451            3,
452            HashConfig::None,
453        )
454        .unwrap();
455
456        let err = part.write_chunk_bytes(0, 2, b"bc").unwrap_err();
457
458        assert!(matches!(err, TakanawaError::InvalidConfig(_)));
459    }
460
461    #[test]
462    fn rejects_part_size_mismatch() {
463        let dir = TempDir::new().unwrap();
464        let target = dir.path().join("file.bin");
465        let part_path = part_path_for(&target);
466        fs::write(&part_path, b"too short").unwrap();
467
468        let err = PartFile::open_or_create(
469            &target,
470            "https://example.test/file",
471            &remote(6),
472            3,
473            HashConfig::None,
474        )
475        .unwrap_err();
476
477        assert!(matches!(err, TakanawaError::PartSizeMismatch { .. }));
478    }
479
480    #[test]
481    fn finalizes_with_supported_hashes() {
482        let cases = [
483            HashConfig::Sha1(hex_array::<20>("a9993e364706816aba3e25717850c26c9cd0d89d")),
484            HashConfig::Sha256(hex_array::<32>(
485                "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
486            )),
487            HashConfig::Sha512(hex_array::<64>(concat!(
488                "ddaf35a193617abacc417349ae20413112e6fa4e89a97ea20a9eeee64b55d39a",
489                "2192992a274fc1a836ba3c23a3feebbd454d4423643ce80e2a9ac94fa54ca49f",
490            ))),
491            HashConfig::Md5(hex_array::<16>("900150983cd24fb0d6963f7d28e17f72")),
492            HashConfig::Crc32(hex_array::<4>("352441c2")),
493        ];
494
495        for hash in cases {
496            let dir = TempDir::new().unwrap();
497            let target = dir.path().join("file.bin");
498            let mut part =
499                PartFile::open_or_create(&target, "https://example.test/file", &remote(3), 3, hash)
500                    .unwrap();
501            part.write_chunk(0, b"abc").unwrap();
502            part.finalize(&target).unwrap();
503
504            assert_eq!(fs::read(&target).unwrap(), b"abc");
505        }
506    }
507
508    #[test]
509    fn finalizes_and_strips_metadata() {
510        let dir = TempDir::new().unwrap();
511        let target = dir.path().join("file.bin");
512        let mut part = PartFile::open_or_create(
513            &target,
514            "https://example.test/file",
515            &remote(6),
516            3,
517            HashConfig::None,
518        )
519        .unwrap();
520        part.write_chunk(1, b"def").unwrap();
521        part.write_chunk(0, b"abc").unwrap();
522        part.finalize(&target).unwrap();
523
524        assert_eq!(fs::read(&target).unwrap(), b"abcdef");
525        assert!(!part_path_for(&target).exists());
526    }
527}
takanawa_core/part_file.rs

takanawa_core/
part_file.rs
