1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
use crate::chunking::Hasher;
use crate::reconstruction::io_ext::FileExt;
use blake3::Hash;
use std::fs::{File, OpenOptions};
use std::io::{self, Seek, SeekFrom};
use std::path::{Path, PathBuf};
#[derive(Debug, Clone)]
pub struct ChunkSource {
pub offset: u64, // Where in the SOURCE to read from
pub length: usize,
}
pub struct ReconstructionTask {
pub target_offset: u64, // Where to write in the NEW temp file
pub source: ChunkSource,
}
pub struct Reconstructor {
temp_file: File,
local_source_path: PathBuf,
}
impl Reconstructor {
/// Creates a new reconstructor.
/// Opens the temporary file and pre-allocates it (Sparse file).
pub fn new(target_path: &Path, local_source_path: &Path, final_size: u64) -> io::Result<Self> {
let temp_path = target_path.with_extension("ffsync-tmp");
let file = OpenOptions::new()
.read(true) // Needed for final verification
.write(true)
.create(true)
.truncate(true) // Overwrite if partial exists (for now)
.open(&temp_path)?;
// Pre-allocate file size. This is crucial for performance and ensuring
// we don't run out of disk space halfway through.
file.set_len(final_size)?;
Ok(Self {
temp_file: file,
local_source_path: local_source_path.to_path_buf(),
})
}
/// Opens an existing temp file for resume (D10): no truncation, no
/// creation. The caller must have verified the temp file exists; its
/// content is the resume state. Size is ensured (extends with zeros if
/// a crash happened mid-pre-allocation; a no-op otherwise).
pub fn open_existing(
target_path: &Path,
local_source_path: &Path,
final_size: u64,
) -> io::Result<Self> {
let temp_path = target_path.with_extension("ffsync-tmp");
let file = OpenOptions::new().read(true).write(true).open(&temp_path)?;
if file.metadata()?.len() < final_size {
file.set_len(final_size)?;
}
Ok(Self {
temp_file: file,
local_source_path: local_source_path.to_path_buf(),
})
}
/// The temp file path for a given target path (single source of truth
/// for the naming convention).
pub fn temp_path_for(target_path: &Path) -> PathBuf {
target_path.with_extension("ffsync-tmp")
}
/// Process a single chunk whose bytes are already in memory.
///
/// This is the path the sync state machines use: the receiver has
/// either decompressed network data or verified local bytes in a
/// buffer, and only needs them placed at `task.target_offset`.
pub fn write_chunk(&self, task: &ReconstructionTask, data: &[u8]) -> io::Result<()> {
self.temp_file.write_at(data, task.target_offset)?;
Ok(())
}
/// Process a single chunk by copying it from the local source file.
///
/// The bytes are read through `scratch_buffer` (resized as needed) so
/// repeated calls do not reallocate. Bench and integration tests use
/// this to simulate the locals-first write order without a peer; the
/// protocol itself resolves local chunks itself because it must verify
/// each chunk's hash before writing (D12).
pub fn copy_local_chunk(
&self,
task: &ReconstructionTask,
scratch_buffer: &mut Vec<u8>,
) -> io::Result<()> {
// Open a fresh handle per chunk. For millions of chunks we might
// want to keep a thread-local handle open; for now, open it here to
// keep the struct `Send`-agnostic (see D7 for the batching/parallel
// work this eventually needs).
let source_file = File::open(&self.local_source_path)?;
if scratch_buffer.len() < task.source.length {
scratch_buffer.resize(task.source.length, 0);
}
let slice = &mut scratch_buffer[0..task.source.length];
source_file.read_at(slice, task.source.offset)?;
self.temp_file.write_at(slice, task.target_offset)?;
Ok(())
}
/// Finalize: verify the reconstructed file's chunk-tree root and rename.
///
/// `expected_root` is the sender's `chunk_root` over its ordered chunk
/// list. Verifying requires re-scanning the temp file with the same CDC
/// parameters and computing the root over that chunk list — the chunk
/// tree is *the* verification identity (see `chunking::chunk_root`).
pub fn finalize_chunk_root(
self,
expected_root: Hash,
final_path: &Path,
pipeline: &crate::chunking::PipelineConfig,
cdc: (usize, usize, usize),
) -> io::Result<()> {
use crate::chunking::{Blake3Hasher, CHUNK_MIN, ChunkEvent, FastCDCChunker, process_file};
let _ = CHUNK_MIN;
let temp_path = final_path.with_extension("ffsync-tmp");
let rx = process_file(
&temp_path,
move || FastCDCChunker::new(cdc.0, cdc.1, cdc.2),
Blake3Hasher,
pipeline,
)?;
let mut chunks = Vec::new();
while let Ok(event) = rx.recv() {
if let ChunkEvent::Chunk(c) = event {
chunks.push(c);
}
}
// Canonical order (offsets are unique; hash workers race).
chunks.sort_by_key(|c| c.offset);
let actual_root = crate::chunking::chunk_root(&chunks);
if actual_root != expected_root {
let _ = std::fs::remove_file(&temp_path);
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Reconstructed chunk-root mismatch",
));
}
drop(self.temp_file);
std::fs::rename(temp_path, final_path)?;
Ok(())
}
/// Legacy finalize using the standard BLAKE3 file hash.
pub fn finalize<H: Hasher>(
mut self,
hasher_impl: &H,
expected_hash: Hash,
final_path: &Path,
) -> io::Result<()> {
self.temp_file.seek(SeekFrom::Start(0))?;
// This keeps the abstraction clean. The Reconstructor doesn't need to know
// *how* to hash, just that it needs to hash this file handle.
let final_hash = hasher_impl.hash_reader(&mut self.temp_file)?;
if final_hash != expected_hash {
// Explicitly delete the corrupted temp file
// (Use metadata to get the path, or reconstruct it)
let _ = std::fs::remove_file(final_path.with_extension("ffsync-tmp"));
return Err(io::Error::new(
io::ErrorKind::InvalidData,
"Reconstructed hash mismatch",
));
}
drop(self.temp_file);
std::fs::rename(final_path.with_extension("ffsync-tmp"), final_path)?;
Ok(())
}
}