Skip to main content

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}