shell_tunnel/fs/transfer.rs
1//! In-flight upload sessions.
2//!
3//! Bytes land in a staging file and are renamed into place only after the whole
4//! transfer verifies. A partial file therefore never appears at the destination
5//! — a consumer polling that path sees nothing or sees the finished article.
6
7use std::collections::HashMap;
8use std::io::{Seek, SeekFrom, Write};
9use std::path::{Path, PathBuf};
10use std::sync::{Mutex, RwLock};
11use std::time::{Duration, Instant};
12
13use crate::fs::sha256::Hasher;
14use crate::fs::UPLOAD_DIR;
15
16/// Chunk size advertised to clients, and the ceiling a chunk may not exceed.
17///
18/// Four rather than eight MiB: the relay's body ceiling is 8 MiB
19/// (`relay::MAX_BODY`) and a WebSocket frame plus a JSON header ride on top, so
20/// sitting on the ceiling turns a 413 into a one-byte accident. The relay's
21/// 120s request timeout is also tight for 8 MiB over a slow link.
22pub const DEFAULT_CHUNK_SIZE: usize = 4 * 1024 * 1024;
23
24/// Largest value `--fs-chunk-size` may name. At or above the relay's ceiling
25/// every relayed transfer would 413, and the symptom looks like a server bug.
26pub const MAX_CHUNK_SIZE: usize = 8 * 1024 * 1024;
27
28/// How long a session may sit idle before it is swept.
29pub const SESSION_TTL: Duration = Duration::from_secs(3600);
30
31/// Largest number of upload sessions this process holds open at once.
32///
33/// A session holds an open file handle for up to `SESSION_TTL` (an hour). The
34/// only credential `POST /uploads` requires is `fs.write` — so without a cap,
35/// a token scoped to nothing but `fs.write` could open enough sessions to
36/// exhaust the process's file descriptors, and fd exhaustion is process-wide:
37/// it would degrade `exec` and `session` routes too, which that token has no
38/// capability over at all. That makes this a capability-boundary issue, not
39/// merely a disk-quota one, so it belongs in this endpoint rather than being
40/// left to an operator-configured limit nobody has asked for yet.
41///
42/// 128 is a fixed constant rather than a CLI knob: generous enough that no
43/// legitimate concurrent-upload workload should hit it, small enough that the
44/// worst case (128 open file handles) is nowhere near typical per-process fd
45/// limits (1024+ on Linux, comparable on Windows). Not configurable — YAGNI
46/// until an operator actually needs a different number.
47const MAX_CONCURRENT_UPLOADS: usize = 128;
48
49/// Why an upload operation was refused.
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub enum UploadError {
52 /// No such session (unknown id, or already completed/cancelled/expired).
53 NotFound,
54 /// The chunk did not start where the session expects.
55 OffsetMismatch { expected: u64 },
56 /// Another live session already targets this destination.
57 Conflict,
58 /// The chunk exceeds the advertised chunk size.
59 TooLarge,
60 /// This chunk would push the session past the size declared at creation.
61 SizeExceeded,
62 /// Too many sessions are already open (see `MAX_CONCURRENT_UPLOADS`).
63 TooManySessions,
64 /// The assembled bytes do not hash to what was declared.
65 Checksum {
66 expected: String,
67 actual: String,
68 /// The destination the rejected session was headed for. Widened for
69 /// this field for the same reason `UploadStore::cancel`'s return
70 /// type was: an audit event for a terminal upload outcome should be
71 /// able to name its subject. `dest_rel` is not consumed before this
72 /// variant is built — `take_for_complete` only borrows it (for
73 /// `self.release(&dest_rel)`) beforehand — so there was never a
74 /// reason it could not be carried here; the field was simply never
75 /// added.
76 dest_rel: String,
77 },
78 /// The filesystem refused.
79 Io {
80 /// Already-rendered detail (`ToString` of the underlying
81 /// `io::Error`). A `String`, not the `io::Error` itself: `UploadError`
82 /// derives `Clone`/`PartialEq`/`Eq`, neither of which `io::Error`
83 /// implements.
84 detail: String,
85 /// The underlying `io::Error`'s `raw_os_error()`, carried alongside
86 /// `detail` so a caller can tell ENOSPC apart from an unrelated
87 /// failure without a locale-dependent match on the rendered message
88 /// — see `platform::is_out_of_space`, which `upload_error_response`
89 /// (`src/api/fs.rs`) uses this to answer.
90 raw_os_error: Option<i32>,
91 },
92}
93
94impl From<std::io::Error> for UploadError {
95 fn from(e: std::io::Error) -> Self {
96 UploadError::Io {
97 raw_os_error: e.raw_os_error(),
98 detail: e.to_string(),
99 }
100 }
101}
102
103/// A session whose bytes are all in and whose digest has been computed.
104#[derive(Debug, Clone)]
105pub struct FinishedUpload {
106 pub dest_rel: String,
107 pub part_path: PathBuf,
108 pub bytes: u64,
109 pub digest: String,
110 pub expected: String,
111}
112
113/// One in-flight upload.
114struct Session {
115 dest_rel: String,
116 /// Absolute canonicalized path to the staging file. Always built from a
117 /// canonical prefix (the staging directory derived from an absolute
118 /// destination). `has_live_part_under` compares this against caller-supplied
119 /// input using `starts_with`, which requires both paths to be canonical.
120 /// Building `part_path` from a non-canonicalized destination path would
121 /// break that comparison silently, causing the query to return `false` even
122 /// when a session is actually under the queried directory.
123 part_path: PathBuf,
124 declared_size: u64,
125 declared_sha256: String,
126 offset: u64,
127 hasher: Hasher,
128 file: std::fs::File,
129 touched: Instant,
130}
131
132/// All in-flight uploads for this process.
133///
134/// State lives in memory only. A restart loses sessions and the client starts
135/// over; persisting them would mean reconstructing a partial hash across
136/// processes, which is a durability feature nobody has asked for yet. The
137/// staging files a restart leaves behind are swept on startup
138/// (`sweep_orphan_parts`).
139///
140/// Invariant: no method ever holds the `sessions` lock and the `claimed`
141/// lock at the same time. `create` takes `claimed` (in its own block, which
142/// closes before anything else runs) and only later, separately, takes
143/// `sessions`; `cancel`, `sweep`, and `take_for_complete` take `sessions`
144/// first and always drop that guard — explicitly, where it is not the last
145/// use in the enclosing statement — before reaching `claimed` through
146/// `release`/`release_destination`. That the two methods' orderings are
147/// opposite (`claimed` before `sessions` in one, `sessions` before `claimed`
148/// in the other) would be a textbook two-lock deadlock *if* either ever held
149/// both at once; because neither does, the orderings never actually nest and
150/// there is nothing to cycle on. Preserving this is what makes `sessions`
151/// vs. `claimed` safe to reason about independently of `append`'s
152/// documented (non-deadlocking) contention with `sweep` — see `append`'s
153/// doc comment for that argument. Breaking this invariant — folding a
154/// `claimed` access inside a still-held `sessions` guard, or vice versa —
155/// would reintroduce a real deadlock that no existing test would catch.
156pub struct UploadStore {
157 sessions: RwLock<HashMap<String, Mutex<Session>>>,
158 /// Destinations currently claimed, so two sessions cannot race to one path,
159 /// each mapped to the staging directory that destination stages through.
160 ///
161 /// A map, not a set. It was a set for as long as nothing read a value out
162 /// of it (an earlier version stored the claiming session's id, which
163 /// nothing looked up — `sessions` is the source of truth for which id owns
164 /// which destination). `release` now does read one: it reclaims an empty
165 /// staging directory once no remaining claim stages through it, and "which
166 /// directory, and is anyone else using it" is a question only this map can
167 /// answer under a single lock. Machine-wide, staging follows each
168 /// destination, so two claims routinely name two different directories and
169 /// a set could not tell them apart.
170 ///
171 /// Its length still doubles as the live-session count for
172 /// `MAX_CONCURRENT_UPLOADS`: every live session claims exactly one
173 /// destination and every destination is claimed by at most one session, so
174 /// checking `claimed.len()` under `claimed`'s own lock is an atomic
175 /// admission check — two concurrent callers cannot both read a count under
176 /// the cap and then both insert, because the check and the insert share
177 /// one critical section.
178 claimed: Mutex<HashMap<String, PathBuf>>,
179 chunk_size: usize,
180 counter: std::sync::atomic::AtomicU64,
181}
182
183impl UploadStore {
184 pub fn new(chunk_size: usize) -> Self {
185 Self {
186 sessions: RwLock::new(HashMap::new()),
187 claimed: Mutex::new(HashMap::new()),
188 // Upper bound one *less* than `MAX_CHUNK_SIZE`, matching what
189 // `--fs-chunk-size`'s own startup check enforces (`main.rs`
190 // exits for `size >= MAX_CHUNK_SIZE`). Clamping to
191 // `MAX_CHUNK_SIZE` itself (an earlier version did) is not
192 // reachable through the CLI today, but it is worse than
193 // unreachable: it would silently *accept* exactly the value the
194 // CLI's own check exists to refuse, for any future caller that
195 // constructs a store directly rather than through the CLI.
196 chunk_size: chunk_size.clamp(1, MAX_CHUNK_SIZE - 1),
197 counter: std::sync::atomic::AtomicU64::new(0),
198 }
199 }
200
201 /// The chunk size clients are told to use.
202 pub fn chunk_size(&self) -> usize {
203 self.chunk_size
204 }
205
206 /// Where staging files live for an upload landing at `dest_abs`.
207 ///
208 /// Inside a jail that is one directory at the root, as it has always been.
209 /// Machine-wide there is no single place it could be: `complete` publishes
210 /// by `rename`, which is only atomic within a filesystem, so staging has to
211 /// sit on the same one as the destination. Windows makes this unavoidable
212 /// rather than merely preferable — a staging directory on `C:` cannot be
213 /// renamed onto `D:` at all.
214 ///
215 /// Taking the destination's own parent, rather than the volume root, keeps
216 /// that guarantee on Unix too, where a mount point below `/` is a different
217 /// filesystem and `/` is usually not writable by the account running this.
218 ///
219 /// The cost is that machine-wide staging is no longer one enumerable
220 /// directory, which is what `sweep_orphan_parts` needs — see its doc.
221 pub fn staging_dir(root: &crate::fs::FsRoot, dest_abs: &Path) -> PathBuf {
222 match root.jail_path() {
223 Some(jail) => jail.join(UPLOAD_DIR),
224 None => match dest_abs.parent() {
225 Some(parent) => parent.join(UPLOAD_DIR),
226 // A destination with no parent is a filesystem anchor, which
227 // `resolve_for_create` already refuses as a create target.
228 None => PathBuf::from(UPLOAD_DIR),
229 },
230 }
231 }
232
233 /// 살아있는 세션 중 스테이징 파일이 `dir` 아래에 있는 것이 하나라도 있는가.
234 ///
235 /// 트리 삭제가 진행 중인 업로드를 지우지 않기 위한 조회다. 근거는 **세션
236 /// 목록이지 디스크가 아니다**: 이전 실행이 남긴 고아 `.part`는 아무도
237 /// 소유하지 않으므로 "살아있음"이 아니고, 그것까지 살아있다고 답하면
238 /// 스윕이 아직 닿지 않은 트리가 무기한 삭제 불가가 된다.
239 ///
240 /// `sessions` 락만 잡는다. `claimed`을 함께 잡으면 이 타입의 락 불변식이
241 /// 깨진다 — 그 이유는 `UploadStore`의 doc comment에 있다.
242 pub fn has_live_part_under(&self, dir: &Path) -> bool {
243 let Ok(sessions) = self.sessions.read() else {
244 // 락이 오염됐다면 "없다"고 답할 근거가 없다. 삭제를 막는 쪽이
245 // 안전하다 — 이 조회의 유일한 소비자가 그렇게 쓴다.
246 return true;
247 };
248 // 캐노니칼화 실패(경로가 존재하지 않음 등)는 안전하게 "있다"고 답한다.
249 // `part_path`는 항상 정규화된 절대경로이고, 정규화되지 않은 경로와는
250 // 비교할 수 없다. 확인할 수 없는 경우 삭제를 막는 쪽이 안전하다
251 // — 이것은 위의 락 오염 처리와 동일한 입장이다.
252 let Ok(canonical_dir) = std::fs::canonicalize(dir) else {
253 return true;
254 };
255 sessions.values().any(|session| {
256 session
257 .lock()
258 .map(|s| s.part_path.starts_with(&canonical_dir))
259 .unwrap_or(true)
260 })
261 }
262
263 /// Open a session for `dest_rel`, which need not exist yet.
264 ///
265 /// Does *not* sweep expired sessions itself, even opportunistically — an
266 /// earlier version did, right here, before anything else ran. That
267 /// silently discarded whatever session the sweep reclaimed: `UploadStore`
268 /// has no `AuditSink` to record with, so a sweep run from inside this
269 /// method structurally cannot leave a trail. The caller
270 /// (`create_upload_blocking`, `src/api/fs.rs`) now sweeps via the
271 /// audit-aware `sweep_expired_uploads` immediately before calling this,
272 /// preserving the ordering the old internal call existed for: reclaim
273 /// stale capacity before the cap check below runs, so a session old
274 /// enough to matter is freed the moment somebody next asks for a new one
275 /// — the same guarantee, just recorded now instead of silent.
276 pub fn create(
277 &self,
278 root: &crate::fs::FsRoot,
279 dest_abs: &Path,
280 dest_rel: String,
281 size: u64,
282 sha256: String,
283 ) -> Result<String, UploadError> {
284 // Computed before the claim rather than after it, and recorded *with*
285 // the claim: `release` reclaims this directory once it is empty and no
286 // remaining claim stages through it, and it decides that under the
287 // same lock this insert takes. Claiming first is therefore what makes
288 // the directory safe to create afterwards — a concurrent `release`
289 // cannot be between "no other claim" and `remove_dir` while this claim
290 // is already in the map, so it can never delete the directory out from
291 // under the `create_dir_all` and `create_new` below.
292 let staging = Self::staging_dir(root, dest_abs);
293
294 // Claim the destination first: a second session for the same path is a
295 // silent-overwrite race, and last-writer-wins loses data quietly.
296 {
297 let mut claimed = self.claimed.lock().map_err(|_| poisoned())?;
298 if claimed.contains_key(&dest_rel) {
299 return Err(UploadError::Conflict);
300 }
301 if claimed.len() >= MAX_CONCURRENT_UPLOADS {
302 return Err(UploadError::TooManySessions);
303 }
304 claimed.insert(dest_rel.clone(), staging.clone());
305 }
306
307 if let Err(e) = std::fs::create_dir_all(&staging) {
308 self.release(&dest_rel);
309 return Err(e.into());
310 }
311
312 let serial = self
313 .counter
314 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
315 // `{serial:016x}` is fixed-width lowercase hex of a server-generated
316 // counter — never caller input — so `staging.join` below only ever
317 // appends exactly one ordinary, `..`-free, drive-prefix-free
318 // component. The postcondition `delete_file_blocking` needs for a
319 // caller-supplied name (`src/api/fs.rs`) has nothing to check here.
320 let id = format!("up-{serial:016x}");
321 let part_path = staging.join(format!("{id}.part"));
322
323 // `create_new` (`O_EXCL` on Unix, `CREATE_NEW` on Windows), not
324 // `File::create` (`O_CREAT|O_TRUNC`, no `O_EXCL`). Containment is a
325 // property of the moment of resolution, and a session here can live
326 // for up to `SESSION_TTL` — long enough for `part_path`'s final
327 // component to become a symlink pointing outside the root before this
328 // call runs. `File::create` would follow it and write outside the
329 // jail; `create_new` fails `EEXIST` on an existing name instead of
330 // following it, symlink or not. `part_path` above is safe by
331 // construction (server-generated id), so nothing should exist at this
332 // exact name yet — `create_new` is what makes "should" load-bearing
333 // instead of assumed.
334 let file = match std::fs::OpenOptions::new()
335 .write(true)
336 .create_new(true)
337 .open(&part_path)
338 {
339 Ok(file) => file,
340 Err(e) => {
341 self.release(&dest_rel);
342 return Err(e.into());
343 }
344 };
345
346 let session = Session {
347 dest_rel: dest_rel.clone(),
348 part_path,
349 declared_size: size,
350 declared_sha256: sha256,
351 offset: 0,
352 hasher: Hasher::new(),
353 file,
354 touched: Instant::now(),
355 };
356
357 // Matched rather than `?`-chained (an earlier version chained
358 // `.write().map_err(...)?.insert(...)`): a poisoned `sessions` lock
359 // must not leak the claim or the staging file already created above
360 // — `session` is not moved into the map on this path, so its
361 // `part_path` is still reachable to clean up. `sweep`/`cancel` only
362 // ever reclaim a claim through a *live* entry in `sessions`; if this
363 // session never made it into that map, nothing else will ever
364 // release it.
365 let mut sessions = match self.sessions.write() {
366 Ok(sessions) => sessions,
367 Err(_) => {
368 std::fs::remove_file(&session.part_path).ok();
369 self.release(&dest_rel);
370 return Err(poisoned());
371 }
372 };
373 sessions.insert(id.clone(), Mutex::new(session));
374 drop(sessions);
375
376 // The claim was inserted at the top of this function and nothing on
377 // the path to here removes it — every `release` above is followed by
378 // `return Err`. A re-insert stood here and was a no-op for a set; for
379 // a map it would be worse than redundant, since it could resurrect a
380 // claim that a `sweep` running between the `sessions.insert` above and
381 // this line had just released, leaving a destination claimed by a
382 // session that no longer exists.
383 Ok(id)
384 }
385
386 /// How many bytes the session has accepted so far.
387 pub fn offset(&self, id: &str) -> Option<u64> {
388 let sessions = self.sessions.read().ok()?;
389 let session = sessions.get(id)?.lock().ok()?;
390 Some(session.offset)
391 }
392
393 /// Append one chunk, returning the offset to send next.
394 ///
395 /// The offset is checked rather than trusted: a retried request that
396 /// already landed would otherwise be written twice and corrupt the hash.
397 ///
398 /// This holds the session's own `Mutex` (and the store-wide `sessions`
399 /// `RwLock` in its shared read mode) for the full duration of the
400 /// `seek`+`write_all` below, which is blocking disk I/O — not merely an
401 /// in-memory update. That means a concurrent `sweep` (which needs
402 /// `sessions`' *write* lock) blocks until this call finishes, and so does
403 /// any other caller of `append`/`take_for_complete`/`cancel` for this same
404 /// session id (nothing else can, since only one request should ever be
405 /// live per session anyway). Concurrent `append` calls for *different*
406 /// sessions are unaffected — `RwLock` read access is shared. Accepted:
407 /// bounded by one write's duration. This is not the only possible
408 /// shape, though: `HashMap<String, Arc<Mutex<Session>>>` would let this
409 /// function clone the `Arc`, drop the `sessions` read guard immediately,
410 /// and hold only the session's own `Mutex` across the write — removing
411 /// the contention with `sweep`/`create` entirely while keeping the same
412 /// per-session serialization (two chunks for the *same* session still
413 /// cannot interleave, since the session `Mutex` alone already prevents
414 /// that). That is a real improvement, tracked separately rather than
415 /// made here: it restructures the state machine's storage at the tail
416 /// of an already-large task, and the contention it would remove is
417 /// bounded and documented, not a correctness gap.
418 ///
419 /// One failure mode this contention analysis does not cover: if a writer
420 /// ever panicked while holding `sessions`' *write* guard (`create`'s
421 /// insert, `take_for_complete`'s or `sweep`'s remove), `std::sync::RwLock`
422 /// poisons permanently — every later `.read()` and `.write()` on it,
423 /// including this method's and `sweep`'s own, then fails identically and
424 /// forever, not "eventually swept once the panic clears." Unreachable
425 /// today, specifically because every write-lock section in this file is a
426 /// plain `HashMap` insert or remove with no disk I/O and nothing else
427 /// that can panic — not because poisoning itself is impossible. That is
428 /// the condition this note depends on, not a permanent property of the
429 /// type: if a future change adds a fallible operation (a write, a
430 /// panicking conversion, anything that can unwind) inside one of those
431 /// three write-lock sections, this analysis no longer holds and the
432 /// unreachability claim needs re-checking against whatever was added.
433 ///
434 /// This says nothing about the per-session `Mutex` acquired below, which
435 /// *is* held across the `seek`+`write_all` disk I/O this doc comment
436 /// itself describes. It is unreachable for the same underlying reason,
437 /// not the same argument: `seek` and `write_all` report failure through
438 /// `Result`, propagated with `?` rather than unwound, so nothing in that
439 /// critical section can panic either.
440 pub fn append(&self, id: &str, offset: u64, bytes: &[u8]) -> Result<u64, UploadError> {
441 if bytes.len() > self.chunk_size {
442 return Err(UploadError::TooLarge);
443 }
444
445 let sessions = self.sessions.read().map_err(|_| poisoned())?;
446 let cell = sessions.get(id).ok_or(UploadError::NotFound)?;
447 let mut session = cell.lock().map_err(|_| poisoned())?;
448
449 if offset != session.offset {
450 return Err(UploadError::OffsetMismatch {
451 expected: session.offset,
452 });
453 }
454
455 // Refused before a single byte is written: without this, a session
456 // can stream arbitrarily far past what it declared, and the mismatch
457 // is only ever caught at `complete` — after every byte has already
458 // hit disk. `checked_add` rather than a plain `+`: `offset` is
459 // caller-supplied (via `Content-Range`) and could in principle be
460 // adversarially close to `u64::MAX`; overflow is treated the same as
461 // exceeding the declared size, not as a wrapped-around pass.
462 let next_offset = offset.checked_add(bytes.len() as u64);
463 if next_offset.map_or(true, |next| next > session.declared_size) {
464 return Err(UploadError::SizeExceeded);
465 }
466
467 session
468 .file
469 .seek(SeekFrom::Start(offset))
470 .map_err(UploadError::from)?;
471 session.file.write_all(bytes).map_err(UploadError::from)?;
472
473 session.hasher.update(bytes);
474 session.offset += bytes.len() as u64;
475 session.touched = Instant::now();
476 Ok(session.offset)
477 }
478
479 /// Finish a session: verify the digest and hand back the staging file.
480 ///
481 /// The session is always removed from `sessions`. The destination's
482 /// *claim*, however, survives a successful call — see
483 /// `release_destination`'s doc comment for why. A failed checksum is
484 /// different: it is terminal (the bytes on disk are known-wrong, and
485 /// leaving them resumable would invite a client to retry into the same
486 /// wrong result), and terminal means no rename will ever follow, so the
487 /// claim is released immediately in that case — there is nothing left for
488 /// a caller to finish acting on.
489 pub fn take_for_complete(&self, id: &str) -> Result<FinishedUpload, UploadError> {
490 let cell = self
491 .sessions
492 .write()
493 .map_err(|_| poisoned())?
494 .remove(id)
495 .ok_or(UploadError::NotFound)?;
496 let session = cell.into_inner().map_err(|_| poisoned())?;
497
498 let Session {
499 dest_rel,
500 part_path,
501 declared_size,
502 declared_sha256,
503 offset,
504 hasher,
505 file,
506 ..
507 } = session;
508 drop(file);
509
510 let digest = hasher.finish();
511 if declared_size != offset || digest != declared_sha256 {
512 std::fs::remove_file(&part_path).ok();
513 self.release(&dest_rel);
514 return Err(UploadError::Checksum {
515 expected: declared_sha256,
516 actual: digest,
517 dest_rel,
518 });
519 }
520
521 // Deliberately not released here — see `release_destination`.
522 Ok(FinishedUpload {
523 dest_rel,
524 part_path,
525 bytes: offset,
526 digest: digest.clone(),
527 expected: declared_sha256,
528 })
529 }
530
531 /// Release a destination's claim once the caller has finished acting on
532 /// the `FinishedUpload` a prior `take_for_complete` handed back — after
533 /// the rename lands, or after giving up on it (whichever the caller's
534 /// last step was).
535 ///
536 /// Not folded into `take_for_complete` itself: an earlier version of this
537 /// function released the claim there, immediately on success — which
538 /// opened a window between "session removed, claim released" and
539 /// "staging file renamed into place" where a second `create` for the same
540 /// `dest_rel` could succeed and start its own rename racing the first's,
541 /// defeating the reason `claimed` exists at all. Keeping the claim alive
542 /// until the caller explicitly releases it closes that window; the caller
543 /// (`complete_upload` in `src/api/fs.rs`) calls this on every exit path
544 /// after `take_for_complete` succeeds, success or failure of the rename
545 /// alike, so the claim is always released exactly once.
546 pub fn release_destination(&self, dest_rel: &str) {
547 self.release(dest_rel);
548 }
549
550 /// Discard a session and its staging file.
551 ///
552 /// Returns the destination and bytes received so far when a session
553 /// existed to cancel — `None` means no such session (unknown, already
554 /// completed, already cancelled, or already expired). Widened from a
555 /// plain `bool` for the same reason `sweep` returns
556 /// `(id, destination, bytes_received)` instead of just dropping what it
557 /// finds: the caller (`cancel_upload` in `src/api/fs.rs`) records a
558 /// terminal audit event, and an event that cannot name which file was
559 /// cancelled answers only "a session ended", not "what happened to this
560 /// file" — the question an audit trail exists to answer.
561 pub fn cancel(&self, id: &str) -> Option<(String, u64)> {
562 let Ok(mut sessions) = self.sessions.write() else {
563 return None;
564 };
565 let cell = sessions.remove(id)?;
566 // Load-bearing, not tidiness: `release` below takes the `claimed`
567 // lock, and `UploadStore`'s struct-level invariant is that `sessions`
568 // and `claimed` are never held at once. Removing this `drop` would
569 // still compile — `sessions` is unused after this point — but would
570 // hold the `sessions` write guard across the `claimed` acquisition,
571 // breaking that invariant silently.
572 drop(sessions);
573 let Ok(session) = cell.into_inner() else {
574 return None;
575 };
576 // Staging file first, claim second. `release` reclaims the staging
577 // directory when the last claim through it goes, and `remove_dir`
578 // refuses a directory that still holds this session's `.part` — so
579 // releasing first leaves exactly the empty directory this is meant to
580 // clear. The reverse order is safe for the claim too: nothing else can
581 // take this destination while the claim is still held.
582 std::fs::remove_file(&session.part_path).ok();
583 self.release(&session.dest_rel);
584 Some((session.dest_rel, session.offset))
585 }
586
587 /// Drop sessions idle for longer than `ttl`.
588 ///
589 /// Returns `(id, destination, bytes_received)` for each, so the caller can
590 /// record a terminal audit event. A session that begins and never ends
591 /// leaves a trail showing only a beginning, which is not a trail.
592 pub fn sweep(&self, ttl: Duration) -> Vec<(String, String, u64)> {
593 let mut expired = Vec::new();
594 let Ok(sessions) = self.sessions.read() else {
595 return expired;
596 };
597 let stale: Vec<String> = sessions
598 .iter()
599 .filter(|(_, cell)| {
600 cell.lock()
601 .map(|s| s.touched.elapsed() >= ttl)
602 .unwrap_or(false)
603 })
604 .map(|(id, _)| id.clone())
605 .collect();
606 // Load-bearing: `std::sync::RwLock` has no upgrade from a read guard
607 // to a write guard, so holding this one into the loop below (which
608 // needs `sessions.write()`) would deadlock this thread against
609 // itself — a different hazard from the `drop` inside the loop below.
610 drop(sessions);
611
612 for id in stale {
613 let Ok(mut sessions) = self.sessions.write() else {
614 break;
615 };
616 let Some(cell) = sessions.remove(&id) else {
617 continue;
618 };
619 // Load-bearing, not tidiness — same reason as `cancel`'s:
620 // `release` below takes `claimed`, and `UploadStore`'s
621 // struct-level invariant is that `sessions` and `claimed` are
622 // never held at once.
623 drop(sessions);
624 if let Ok(session) = cell.into_inner() {
625 // Staging file before claim, for the reason `cancel` states.
626 std::fs::remove_file(&session.part_path).ok();
627 self.release(&session.dest_rel);
628 expired.push((id, session.dest_rel, session.offset));
629 }
630 }
631 expired
632 }
633
634 /// Drop a destination's claim, and reclaim its staging directory if that
635 /// was the last claim staging through it.
636 ///
637 /// The directory is this API's own artifact, and until now nothing removed
638 /// it: `.part` files were swept but the `.shell-tunnel-uploads` directory
639 /// holding them stayed forever, invisible to `list` and refused by `stat`
640 /// and `delete` alike — an artifact the file API created and the file API
641 /// could not remove. Machine-wide that is one per directory anyone has
642 /// ever uploaded to. Whoever made it cleans it up, which is also why this
643 /// does not instead relax the reservation guard on the delete route.
644 ///
645 /// **The `remove_dir` runs while the lock is held, deliberately.** Deciding
646 /// "no other claim stages here" and then releasing the lock before the
647 /// syscall reopens exactly the window this ordering closes: a `create` can
648 /// insert its claim and run `create_dir_all` in that gap, and the removal
649 /// would then delete the directory that create is about to place a `.part`
650 /// into. The cost is that `create`'s admission check can wait on one
651 /// `remove_dir`; shortening this critical section is not the optimisation
652 /// it looks like.
653 ///
654 /// `remove_dir`, never `remove_dir_all`: a directory holding another
655 /// session's staging file, or anything else, fails the call and is left
656 /// alone. The error is discarded because every reason it can fail — not
657 /// empty, already gone, no permission — is a reason to do nothing.
658 ///
659 /// Not extended to the orphan sweeps. `sweep_orphan_parts_in` is called
660 /// *from* `create`, one line before the directory is needed, and
661 /// `sweep_orphan_parts` runs on an interval without holding this lock;
662 /// removing a directory from either would race the very creation this
663 /// method's ordering protects. A directory left empty by a previous run's
664 /// crash is instead reclaimed the next time an upload through it finishes.
665 fn release(&self, dest_rel: &str) {
666 if let Ok(mut claimed) = self.claimed.lock() {
667 let Some(staging) = claimed.remove(dest_rel) else {
668 return;
669 };
670 if !claimed.values().any(|other| *other == staging) {
671 std::fs::remove_dir(&staging).ok();
672 }
673 }
674 }
675}
676
677impl std::fmt::Debug for UploadStore {
678 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
679 f.debug_struct("UploadStore")
680 .field("chunk_size", &self.chunk_size)
681 .finish_non_exhaustive()
682 }
683}
684
685fn poisoned() -> UploadError {
686 UploadError::Io {
687 detail: "internal lock poisoned".to_string(),
688 raw_os_error: None,
689 }
690}
691
692/// Remove staging files left behind by a previous run.
693///
694/// Sessions do not survive a restart, so any `.part` still present is
695/// unreachable — nothing can resume it and nothing will complete it.
696///
697/// Returns `(upload_id, bytes)` for each file removed, so a caller can record
698/// a terminal audit event per orphan — same reason `UploadStore::sweep` and
699/// `cancel` return what they do, rather than dropping what they find. Two
700/// things are recoverable here and one is not: `upload_id` is the filename
701/// stem (`up-{serial:016x}`, never caller input, so parsing it back out is
702/// safe), and `bytes` is the file's size, which always equals what
703/// `append` had written — but the *destination* lived only in the in-memory
704/// `Session` a restart already discarded before this function ever runs, so
705/// there is nothing here to recover it from. See `AuditEvent::with_upload_id`
706/// for how a caller correlates this back to the `upload.start` that does
707/// have it.
708///
709/// An empty `Vec` covers both "nothing to sweep" and "the staging directory
710/// could not be read at all" (most commonly: no upload has ever run against
711/// this root, so it was never created). Not distinguished, same reasoning as
712/// before this was widened: nothing consumes that distinction.
713///
714/// **Machine-wide scope sweeps nothing here, and that is a real gap rather
715/// than an oversight.** With no `--fs-root`, staging follows each destination
716/// to its own directory (see [`UploadStore::staging_dir`]), so the set of
717/// places a `.part` could be left is every directory anyone has ever uploaded
718/// to — not enumerable without walking every drive, which is not a thing a
719/// startup path should do. What covers it instead is
720/// [`sweep_orphan_parts_in`], called against a single destination's staging
721/// directory when an upload next targets it. The practical difference: inside
722/// a jail an orphan is reclaimed at the next restart; machine-wide it is
723/// reclaimed the next time something uploads to the same directory. Both are
724/// invisible to `list`, which refuses the staging directory by name either
725/// way.
726pub fn sweep_orphan_parts(root: &crate::fs::FsRoot) -> Vec<(String, u64)> {
727 let Some(jail) = root.jail_path() else {
728 return Vec::new();
729 };
730 // No age floor: this runs at startup and on an interval against a jail's
731 // single staging directory, where "a `.part` exists" already implies no
732 // session owns it — sessions do not survive a restart, and the interval
733 // caller sweeps expired sessions first.
734 sweep_orphan_parts_in(&jail.join(UPLOAD_DIR), Duration::ZERO)
735}
736
737/// [`sweep_orphan_parts`] against one staging directory, removing only files
738/// that have been untouched for at least `min_age`.
739///
740/// Split out so machine-wide uploads have a reclaim path at all: the caller
741/// that knows a destination knows its staging directory, even though no
742/// startup path can enumerate every such directory.
743///
744/// **`min_age` is what keeps this from destroying a live upload, and it is not
745/// optional for a runtime caller.** Machine-wide staging is shared by every
746/// upload heading for the same directory, so a sweep run when a second session
747/// is created will see the *first* session's `.part` — a file that is very much
748/// owned. Removing it does not fail the writes that follow: the session holds
749/// an open handle, so `append` keeps succeeding against a name that no longer
750/// exists, every chunk answers 200, and only `complete` fails, with
751/// `ENOENT` — after the client has uploaded the whole file. That shape (accept
752/// everything, then lose it at publication) is the worst available, and it is
753/// what an unconditional sweep here produced.
754///
755/// A live session's file is protected because writing to it updates its mtime,
756/// and a session that has gone quiet for longer than the caller's floor has
757/// already been reclaimed by `sweep_expired_uploads`, which the API layer runs
758/// first. An orphan from a previous run has no such protection, which is the
759/// point.
760pub fn sweep_orphan_parts_in(staging: &Path, min_age: Duration) -> Vec<(String, u64)> {
761 let Ok(entries) = std::fs::read_dir(staging) else {
762 return Vec::new();
763 };
764 let mut removed = Vec::new();
765 for entry in entries.flatten() {
766 let path = entry.path();
767 if path.extension().and_then(|e| e.to_str()) != Some("part") {
768 continue;
769 }
770 // Read before removing: there is no size to report once the file is
771 // gone. `std::fs::metadata(&path)` — a fresh stat — rather than the
772 // cheaper `entry.metadata()`: on Windows, `DirEntry::metadata()`
773 // returns the `WIN32_FIND_DATA` captured by the `read_dir`
774 // enumeration itself, which can under-report the size of a file
775 // still open elsewhere for writing (verified: a session whose
776 // staging file was just appended to and never closed reported `0`
777 // bytes here, on this platform, until this was changed to a fresh
778 // stat). A second, different way the same `DirEntry` API is not what
779 // it appears to be: `list`'s own walk (`src/api/fs.rs`) already notes
780 // that `DirEntry::metadata` is lstat-like there, so a symlink looks
781 // in-root when `metadata` would follow it out — that one is about
782 // *which* file the metadata describes, this one is about *how current*
783 // it is, but both come from trusting the enumeration's cached view
784 // instead of asking the filesystem again. Not reachable in
785 // production here — the whole reason a `.part` file is orphaned is
786 // that the process that held it open is gone — but a test exercising
787 // this without a real restart can still hit it, and the fresh call
788 // costs one extra syscall per file, on a path that runs once at
789 // startup.
790 let meta = std::fs::metadata(&path);
791 // Age is read from the same fresh stat, and a file whose age cannot be
792 // established is left alone rather than removed: this is the guard that
793 // stands between a runtime sweep and a live upload, and a guard that
794 // fails open is not one. `Duration::ZERO` makes it a no-op for the
795 // startup caller, where nothing can be live.
796 if !min_age.is_zero() {
797 // `map_or(true, ..)` rather than `is_none_or`: the latter is stable
798 // since 1.82 and this crate's MSRV is 1.78. Same meaning — an
799 // unreadable or unknowable age counts as young, so the file stays.
800 let young_or_unknown = meta
801 .as_ref()
802 .ok()
803 .and_then(|m| m.modified().ok())
804 .and_then(|modified| modified.elapsed().ok())
805 .map_or(true, |age| age < min_age);
806 if young_or_unknown {
807 continue;
808 }
809 }
810 let bytes = meta.map(|m| m.len()).unwrap_or(0);
811 let Some(id) = path.file_stem().and_then(|s| s.to_str()) else {
812 // Not a name this process ever generated (`up-{serial:016x}.part`
813 // is always valid UTF-8) — nothing to correlate an event to, so
814 // the file is removed but not reported.
815 std::fs::remove_file(&path).ok();
816 continue;
817 };
818 let id = id.to_string();
819 if std::fs::remove_file(&path).is_ok() {
820 removed.push((id, bytes));
821 }
822 }
823 removed
824}
825
826#[cfg(test)]
827mod tests {
828 use super::*;
829 use crate::fs::FsRoot;
830
831 fn store() -> (tempfile::TempDir, FsRoot, UploadStore) {
832 let dir = tempfile::tempdir().expect("tempdir");
833 let root = FsRoot::new(dir.path()).expect("root");
834 let store = UploadStore::new(DEFAULT_CHUNK_SIZE);
835 (dir, root, store)
836 }
837
838 impl UploadStore {
839 /// `create` from a root-relative destination, resolving it the way the
840 /// API layer does.
841 ///
842 /// `create` takes the resolved absolute destination because staging
843 /// has to land on the destination's own filesystem when no `--fs-root`
844 /// narrows the scope (see `staging_dir`). These tests all run against a
845 /// jail, where that resolution is uninteresting — doing it here rather
846 /// than passing some hand-built path keeps them exercising the same
847 /// path the real caller takes.
848 fn create_rel(
849 &self,
850 root: &FsRoot,
851 dest: &str,
852 size: u64,
853 sha256: String,
854 ) -> Result<String, UploadError> {
855 let absolute = root.resolve_for_create(dest).expect("destination resolves");
856 self.create(root, &absolute, dest.to_string(), size, sha256)
857 }
858 }
859
860 /// The staging directory of a jailed root.
861 ///
862 /// `staging_dir` takes a destination because machine-wide scope has to put
863 /// staging on the destination's own filesystem. A jail ignores it, so these
864 /// tests name that explicitly rather than threading a value none of them
865 /// care about through every call.
866 fn staging_of(root: &FsRoot) -> PathBuf {
867 UploadStore::staging_dir(root, Path::new("ignored-when-jailed"))
868 }
869
870 /// SHA-256 of b"hello world".
871 const HELLO_DIGEST: &str = "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9";
872
873 /// 트리 삭제가 진행 중인 업로드를 지우지 않으려면, 어떤 디렉터리 아래에
874 /// 살아있는 세션의 스테이징 파일이 있는지 물을 수 있어야 한다. 고아
875 /// `.part`는 "살아있음"이 아니다 — 그것까지 살아있다고 답하면 스윕이 늦은
876 /// 트리가 무기한 삭제 불가가 된다.
877 #[test]
878 fn a_live_session_is_visible_under_its_staging_directory() {
879 let (dir, root, store) = store();
880 let staging = staging_of(&root);
881
882 assert!(
883 !store.has_live_part_under(dir.path()),
884 "세션이 없으면 아무것도 살아있지 않다"
885 );
886
887 let id = store
888 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
889 .expect("create");
890
891 assert!(
892 store.has_live_part_under(dir.path()),
893 "루트 아래에서 보인다"
894 );
895 assert!(
896 store.has_live_part_under(&staging),
897 "스테이징 자신 아래에서도 보인다"
898 );
899 // 캐노니칼화할 수 있도록 존재하는 디렉터리 생성
900 let elsewhere = dir.path().join("elsewhere");
901 std::fs::create_dir_all(&elsewhere).expect("mkdir elsewhere");
902 assert!(
903 !store.has_live_part_under(&elsewhere),
904 "관계없는 디렉터리 아래에서는 보이지 않는다"
905 );
906
907 store.cancel(&id);
908 assert!(
909 !store.has_live_part_under(dir.path()),
910 "취소된 세션은 살아있지 않다"
911 );
912 }
913
914 /// 세션이 없는 채 남은 `.part`(이전 실행의 고아)는 살아있지 않다.
915 /// 이 테스트는 고아 `.part` 파일이 디스크에 존재해도, 세션 목록에
916 /// 없으면 "살아있음"이 아님을 증명한다. 구현이 디스크를 조회한다면
917 /// 이 테스트는 실패할 것이다.
918 #[test]
919 fn an_orphan_part_file_is_not_a_live_session() {
920 let (dir, root, store) = store();
921 let staging = staging_of(&root);
922 std::fs::create_dir_all(&staging).expect("mkdir staging");
923
924 // 살아있는 세션을 생성
925 let live_id = store
926 .create_rel(&root, "upload1.bin", 5, "0".repeat(64))
927 .expect("create live session");
928
929 // 세션이 있으므로 true를 반환
930 assert!(
931 store.has_live_part_under(dir.path()),
932 "살아있는 세션이 있으므로 true를 반환한다"
933 );
934
935 // 세션을 취소하고, 고아 `.part` 파일을 그 자리에 남김.
936 // Cancelling now reclaims the staging directory as well, so it has to
937 // be recreated before an orphan can be planted in it — which is also
938 // the shape of the real case: a previous run's directory, remade by
939 // whichever upload comes next.
940 store.cancel(&live_id);
941 std::fs::create_dir_all(&staging).expect("remake staging");
942 let orphan_path = staging.join("up-0000000000000000.part");
943 std::fs::write(&orphan_path, b"orphan content").expect("write orphan");
944 assert!(
945 orphan_path.exists(),
946 "고아 파일이 디스크에 실제로 존재해야 함"
947 );
948
949 // 고아 파일이 있어도 세션 목록에 없으므로 false를 반환해야 한다.
950 // 구현이 디스크를 조회한다면 이 assertion이 실패할 것이다.
951 assert!(
952 !store.has_live_part_under(dir.path()),
953 "고아 파일이 있어도 세션 목록이 기준이므로 false를 반환한다"
954 );
955 }
956
957 /// 존재하지 않는 경로(캐노니칼화 불가)에 대한 조회는 안전하게 "있다"고 답한다.
958 /// `part_path`는 항상 정규화된 절대경로이므로, 정규화되지 않은 경로와는
959 /// 비교할 수 없다. 알 수 없는 경우 삭제를 거부하는 쪽이 안전하다.
960 #[test]
961 fn a_nonexistent_path_cannot_be_canonicalized_so_answers_true() {
962 let (dir, root, store) = store();
963 let id = store
964 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
965 .expect("create");
966
967 // 존재하지 않는 경로: canonicalize가 실패한다
968 let nonexistent = dir.path().join("does-not-exist");
969 assert!(!nonexistent.exists(), "path must not exist for this test");
970
971 // 캐노니칼화할 수 없으므로 안전하게 "있다"고 답해야 한다
972 assert!(
973 store.has_live_part_under(&nonexistent),
974 "캐노니칼화 불가능한 경로는 안전하게 true를 반환한다"
975 );
976
977 store.cancel(&id);
978 }
979
980 #[test]
981 fn a_session_starts_at_offset_zero() {
982 let (_dir, root, store) = store();
983 let id = store
984 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
985 .expect("create");
986 assert_eq!(store.offset(&id), Some(0));
987 }
988
989 #[test]
990 fn chunks_advance_the_offset() {
991 let (_dir, root, store) = store();
992 let id = store
993 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
994 .expect("create");
995
996 assert_eq!(store.append(&id, 0, b"hello ").expect("first"), 6);
997 assert_eq!(store.append(&id, 6, b"world").expect("second"), 11);
998 }
999
1000 #[test]
1001 fn a_chunk_at_the_wrong_offset_is_refused_with_the_expected_one() {
1002 let (_dir, root, store) = store();
1003 let id = store
1004 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1005 .expect("create");
1006 store.append(&id, 0, b"hello ").expect("first");
1007
1008 assert_eq!(
1009 store.append(&id, 0, b"again"),
1010 Err(UploadError::OffsetMismatch { expected: 6 })
1011 );
1012 }
1013
1014 #[test]
1015 fn two_sessions_may_not_target_the_same_path() {
1016 let (_dir, root, store) = store();
1017 store
1018 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1019 .expect("first");
1020 assert_eq!(
1021 store.create_rel(&root, "out.bin", 11, HELLO_DIGEST.into()),
1022 Err(UploadError::Conflict)
1023 );
1024 }
1025
1026 #[test]
1027 fn a_matching_checksum_completes() {
1028 let (_dir, root, store) = store();
1029 let id = store
1030 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1031 .expect("create");
1032 store.append(&id, 0, b"hello world").expect("append");
1033
1034 let finished = store.take_for_complete(&id).expect("complete");
1035 assert_eq!(finished.bytes, 11);
1036 assert_eq!(finished.digest, HELLO_DIGEST);
1037 assert_eq!(finished.dest_rel, "out.bin");
1038 }
1039
1040 #[test]
1041 fn a_mismatched_checksum_is_refused() {
1042 let (_dir, root, store) = store();
1043 let wrong = "0".repeat(64);
1044 let id = store
1045 .create_rel(&root, "out.bin", 11, wrong.clone())
1046 .expect("create");
1047 store.append(&id, 0, b"hello world").expect("append");
1048
1049 match store.take_for_complete(&id) {
1050 Err(UploadError::Checksum {
1051 expected,
1052 actual,
1053 dest_rel,
1054 }) => {
1055 assert_eq!(expected, wrong);
1056 assert_eq!(actual, HELLO_DIGEST);
1057 assert_eq!(dest_rel, "out.bin");
1058 }
1059 other => panic!("expected a checksum refusal, got {other:?}"),
1060 }
1061 // The session is gone and the staging file with it.
1062 assert_eq!(store.offset(&id), None);
1063 }
1064
1065 #[test]
1066 fn a_chunk_above_the_ceiling_is_refused() {
1067 let (_dir, root, store) = store();
1068 let id = store
1069 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1070 .expect("create");
1071 let oversized = vec![0_u8; DEFAULT_CHUNK_SIZE + 1];
1072 assert_eq!(store.append(&id, 0, &oversized), Err(UploadError::TooLarge));
1073 }
1074
1075 #[test]
1076 fn a_chunk_that_would_exceed_the_declared_size_is_refused() {
1077 let (_dir, root, store) = store();
1078 // Declares 5 bytes; the digest is irrelevant here since the size
1079 // check runs at `append` time, well before any checksum comparison.
1080 let id = store
1081 .create_rel(&root, "out.bin", 5, HELLO_DIGEST.into())
1082 .expect("create");
1083 assert_eq!(
1084 store.append(&id, 0, b"hello world"),
1085 Err(UploadError::SizeExceeded)
1086 );
1087 // Refused before anything was written: the offset must not have moved.
1088 assert_eq!(store.offset(&id), Some(0));
1089 }
1090
1091 #[test]
1092 fn a_chunk_landing_exactly_on_the_declared_size_is_accepted() {
1093 let (_dir, root, store) = store();
1094 let id = store
1095 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1096 .expect("create");
1097 // Exactly 11 bytes against a declared size of 11 — the boundary
1098 // `a_chunk_that_would_exceed_the_declared_size_is_refused` does not
1099 // cover, and the one `>` (not `>=`) in the check depends on.
1100 assert_eq!(store.append(&id, 0, b"hello world").expect("append"), 11);
1101 }
1102
1103 #[test]
1104 fn cancelling_removes_the_session_and_frees_the_destination() {
1105 let (_dir, root, store) = store();
1106 let id = store
1107 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1108 .expect("create");
1109 store.append(&id, 0, b"hello ").expect("append");
1110
1111 let (destination, bytes) = store.cancel(&id).expect("session existed");
1112 assert_eq!(destination, "out.bin");
1113 assert_eq!(bytes, 6);
1114 assert_eq!(store.offset(&id), None);
1115 // The destination is claimable again.
1116 assert!(store
1117 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1118 .is_ok());
1119 }
1120
1121 #[test]
1122 fn sweeping_drops_sessions_past_their_ttl() {
1123 let (_dir, root, store) = store();
1124 let staging = root.jail_path().expect("jailed").join(UPLOAD_DIR);
1125 let id = store
1126 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1127 .expect("create");
1128 assert!(staging.is_dir(), "staging exists while the session is live");
1129
1130 assert_eq!(store.sweep(Duration::ZERO).len(), 1);
1131 assert_eq!(store.offset(&id), None);
1132 // The sweep path reclaims the directory too, and it only can because
1133 // it removes the staging file *before* releasing the claim: `release`
1134 // reclaims under the claim lock, and `remove_dir` refuses a directory
1135 // that still holds a `.part`. This asserts that ordering from the
1136 // outside, where reversing it leaves an empty directory behind.
1137 assert!(
1138 !staging.exists(),
1139 "a swept session must not leave its staging directory behind"
1140 );
1141 }
1142
1143 /// `create` used to sweep opportunistically (with the real, fixed
1144 /// `SESSION_TTL`) before doing anything else. That call is gone — moved
1145 /// to the caller, which sweeps through the audit-aware
1146 /// `sweep_expired_uploads` instead (`src/api/fs.rs`) — because
1147 /// `UploadStore` has no `AuditSink` to record with, so a sweep run from
1148 /// inside this method could never leave a trail. This is the regression
1149 /// guard for that move: even a session that a zero-TTL sweep would call
1150 /// stale must survive an unrelated `create` call untouched, proving
1151 /// `create` itself no longer reclaims anything — only an explicit
1152 /// `sweep`/`sweep_expired_uploads` call does.
1153 #[test]
1154 fn create_does_not_sweep_expired_sessions_itself() {
1155 let (_dir, root, store) = store();
1156 let id = store
1157 .create_rel(&root, "old.bin", 11, HELLO_DIGEST.into())
1158 .expect("create");
1159
1160 store
1161 .create_rel(&root, "new.bin", 11, HELLO_DIGEST.into())
1162 .expect("second create");
1163
1164 assert_eq!(
1165 store.offset(&id),
1166 Some(0),
1167 "create must not silently reclaim a stale session; only an explicit sweep call may"
1168 );
1169 }
1170
1171 /// The strongest test in this module: `create` opens the staging file
1172 /// with `create_new`, which must fail (`EEXIST`) rather than follow an
1173 /// existing symlink at that exact name. Planted *before* any session
1174 /// exists, exploiting that a fresh store's counter starts at 0 — so the
1175 /// first session's id, and therefore its staging path, is predictable
1176 /// (`up-0000000000000000.part`).
1177 ///
1178 /// Two assertions, not one: the create must fail, *and* the outside
1179 /// target must be untouched. Checking only the error would still pass a
1180 /// version that wrote through the link and then failed for an unrelated
1181 /// reason afterward.
1182 #[test]
1183 fn a_pre_existing_symlink_at_the_predicted_staging_path_cannot_be_written_through() {
1184 let outer = tempfile::tempdir().expect("outer tempdir");
1185 let root_dir = outer.path().join("root");
1186 std::fs::create_dir_all(&root_dir).expect("mkdir root");
1187 let root = FsRoot::new(&root_dir).expect("root");
1188 let store = UploadStore::new(DEFAULT_CHUNK_SIZE);
1189
1190 let secret = outer.path().join("secret.txt");
1191 std::fs::write(&secret, b"outside-secret").expect("write secret");
1192
1193 let staging = staging_of(&root);
1194 std::fs::create_dir_all(&staging).expect("mkdir staging");
1195 let predicted = staging.join("up-0000000000000000.part");
1196
1197 #[cfg(unix)]
1198 let linked = std::os::unix::fs::symlink(&secret, &predicted).is_ok();
1199 #[cfg(windows)]
1200 let linked = std::os::windows::fs::symlink_file(&secret, &predicted).is_ok();
1201 #[cfg(not(any(unix, windows)))]
1202 let linked = false;
1203 if !linked {
1204 return; // symlink privilege unavailable on this runner; skip
1205 }
1206
1207 let result = store.create_rel(&root, "app-new.bin", 11, HELLO_DIGEST.into());
1208 assert!(
1209 matches!(result, Err(UploadError::Io { .. })),
1210 "create_new must refuse a pre-existing symlink at the staging path \
1211 rather than follow it, got {result:?}"
1212 );
1213 assert_eq!(
1214 std::fs::read(&secret).expect("read secret"),
1215 b"outside-secret",
1216 "the outside target must be untouched: the open must fail before \
1217 any write reaches it"
1218 );
1219 }
1220
1221 #[test]
1222 fn a_cap_limits_concurrent_sessions_and_releasing_one_frees_a_slot() {
1223 let (_dir, root, store) = store();
1224 let mut ids = Vec::with_capacity(MAX_CONCURRENT_UPLOADS);
1225 for i in 0..MAX_CONCURRENT_UPLOADS {
1226 let id = store
1227 .create_rel(&root, &format!("f{i}.bin"), 1, HELLO_DIGEST.into())
1228 .unwrap_or_else(|e| panic!("session {i} should fit under the cap: {e:?}"));
1229 ids.push(id);
1230 }
1231
1232 assert_eq!(
1233 store.create_rel(&root, "one-too-many.bin", 1, HELLO_DIGEST.into()),
1234 Err(UploadError::TooManySessions)
1235 );
1236
1237 // Freeing one slot makes room for exactly one more.
1238 assert!(store.cancel(&ids[0]).is_some());
1239 assert!(store
1240 .create_rel(&root, "one-too-many.bin", 1, HELLO_DIGEST.into())
1241 .is_ok());
1242 }
1243
1244 #[test]
1245 fn completing_an_upload_keeps_the_destination_claimed_until_explicitly_released() {
1246 let (_dir, root, store) = store();
1247 let id = store
1248 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1249 .expect("create");
1250 store.append(&id, 0, b"hello world").expect("append");
1251 let finished = store.take_for_complete(&id).expect("complete");
1252
1253 // The caller has not renamed the staging file into place yet (has not
1254 // called `release_destination`), so the destination must still be
1255 // refused to a second session — otherwise two sessions could both be
1256 // mid-publication to the same path.
1257 assert_eq!(
1258 store.create_rel(&root, "out.bin", 11, HELLO_DIGEST.into()),
1259 Err(UploadError::Conflict)
1260 );
1261
1262 store.release_destination(&finished.dest_rel);
1263
1264 // Now that the caller is done with it, the destination is claimable again.
1265 assert!(store
1266 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1267 .is_ok());
1268 }
1269
1270 #[test]
1271 fn sweep_orphan_parts_removes_leftover_part_files_and_nothing_else() {
1272 let dir = tempfile::tempdir().expect("tempdir");
1273 let root = FsRoot::new(dir.path()).expect("root");
1274 let staging = staging_of(&root);
1275 std::fs::create_dir_all(&staging).expect("mkdir staging");
1276 std::fs::write(staging.join("up-0000000000000000.part"), b"leftover")
1277 .expect("write orphan");
1278 std::fs::write(staging.join("up-0000000000000001.part"), b"leftover2")
1279 .expect("write second orphan");
1280 // Not a `.part` file — proves the extension filter, not "delete
1281 // everything in the directory".
1282 std::fs::write(staging.join("keep.txt"), b"not a part file").expect("write keep");
1283
1284 let mut removed = sweep_orphan_parts(&root);
1285 removed.sort();
1286 assert_eq!(
1287 removed,
1288 vec![
1289 ("up-0000000000000000".to_string(), 8),
1290 ("up-0000000000000001".to_string(), 9),
1291 ],
1292 "each orphan must be reported by its id (the filename stem) and the bytes it held, so a caller can audit it"
1293 );
1294 assert!(!staging.join("up-0000000000000000.part").exists());
1295 assert!(!staging.join("up-0000000000000001.part").exists());
1296 assert!(
1297 staging.join("keep.txt").exists(),
1298 "only .part files are orphans; anything else in staging must survive"
1299 );
1300 }
1301
1302 #[test]
1303 fn a_poisoned_sessions_lock_does_not_leak_the_claim_or_the_staging_file() {
1304 let (_dir, root, store) = store();
1305
1306 // Poison `sessions` by panicking while holding its write guard.
1307 // `catch_unwind` keeps the panic from taking the test process down;
1308 // the guard's `Drop` still runs during the unwind and marks the
1309 // lock poisoned regardless of the panic being caught afterward.
1310 let poisoned = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
1311 let _guard = store.sessions.write().expect("lock not yet poisoned");
1312 panic!("poison it");
1313 }));
1314 assert!(
1315 poisoned.is_err(),
1316 "the closure must have panicked while holding the write guard"
1317 );
1318
1319 let outcome = store.create_rel(&root, "out.bin", 11, HELLO_DIGEST.into());
1320 assert!(
1321 matches!(outcome, Err(UploadError::Io { .. })),
1322 "a poisoned sessions lock must surface as an Io error, got {outcome:?}"
1323 );
1324
1325 // Recover the lock — a real caller cannot do this, but the test does,
1326 // purely to inspect whether the failed attempt above left anything
1327 // behind. If it did, this second `create` for the same destination
1328 // would come back `Err(Conflict)` instead of succeeding.
1329 store.sessions.clear_poison();
1330 assert!(
1331 store
1332 .create_rel(&root, "out.bin", 11, HELLO_DIGEST.into())
1333 .is_ok(),
1334 "the destination must not still be claimed by the failed attempt"
1335 );
1336
1337 let staging = staging_of(&root);
1338 let leftover_parts = std::fs::read_dir(&staging)
1339 .expect("staging dir")
1340 .flatten()
1341 .filter(|e| e.path().extension().and_then(|x| x.to_str()) == Some("part"))
1342 .count();
1343 assert_eq!(
1344 leftover_parts, 1,
1345 "only the second, successful session's staging file should remain"
1346 );
1347 }
1348}