helper: ended uploads leave the job table on a live session (#475)

The Files screen uploads a pick of files one AVA1 job per file, all on the
engine's one shared session. The console's job table holds 32 jobs, and an
upload that ended stayed listed (with its job thread and workers) until its
session ended: only parked jobs were ever reaped, and a session is parked only
when it ends. So the 33rd file of a pick got ERR_BUSY "too many jobs, or this
job is still closing" for 12 retries and failed, and every later file did the
same until the helper was replaced (a new session). Reproduced on the Phat
(FW 13.60): file 24 of the Switch-like set failed busy, and the table never
drained while the session lived.

- A finished upload or download (not a copy, op or console-to-console send)
  stamps concluded_ms once its JobDone is out (or the peer's came in).
- The reaper lets such a job go 10 s after that, attached or not, unless it
  still has files settling (unswept), as SPEC §11.7 intended.
- A JobOpen that finds the table full takes the slot of the concluded
  transfer that ended longest ago and nobody else holds, instead of BUSY.

Tests: C unit test of the eviction and reap rules (wire_upload), and a host
test of 40 single-file jobs in a row on one session (failed at file 32 before).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RxCGb8Asp1A1AQYiv1kbgS

engine/crates/ava1-ctest/csrc/test_shim.c

@@ -2469,6 +2469,92 @@ int ava1_test_reap_rules(const char *jobs_dir) {
     return rc;
 }
 
+/* #475: an upload that ended on a session still running leaves the table on its own. Marks a listed
+ * job as a concluded upload (what ava1_apply_fail does once its JobDone is out), ended `ago_ms` ago. */
+static void conclude(uint8_t tag, uint64_t ago_ms) {
+    uint8_t id[16];
+    ava1_job_t *j;
+    memset(id, tag, 16);
+    if (!(j = ava1_job_find(id))) return;
+    j->kind = AVA1_JOB_UPLOAD;
+    __atomic_store_n(&j->finished, 1, __ATOMIC_RELEASE);
+    __atomic_store_n(&j->concluded_ms, (ava1_mono_ms() - ago_ms) | 1u, __ATOMIC_RELEASE);
+    ava1_job_put(j);
+}
+
+/* 0, or the step that failed. A full table of uploads attached to one live session: the ended ones
+ * make room for the next JobOpen (the one that ended longest ago and nobody holds goes first), the
+ * ones still running or still settling never do, and the reaper lets an ended one go 10 s after its
+ * end although its session never let go of it. */
+int ava1_test_ended_transfers(const char *jobs_dir) {
+    static const uint8_t SID[16] = { 0xe1 };
+    ava1_data_cfg_t dc;
+    ava1_job_t *j, *held = NULL;
+    uint8_t id[16];
+    int rc = 0, t;
+    memset(&dc, 0, sizeof dc);
+    snprintf(dc.jobs_dir, sizeof dc.jobs_dir, "%s", jobs_dir);
+    if (ava1_data_start(&dc) != 0) return -100;
+    /* 32 running uploads on one session: the table is full and nothing may be evicted */
+    for (t = 0; t < AVA1_MAX_JOBS && !rc; t++) {
+        if (!(j = mk((uint8_t)(0x40 + t), SID))) rc = -1;
+        else {
+            j->kind = AVA1_JOB_UPLOAD;
+            ava1_job_put(j);
+        }
+    }
+    if (!rc && (j = mk(0x70, SID)) != NULL) {
+        ava1_job_put(j);
+        rc = -2; /* a full table of running jobs took another */
+    }
+    if (!rc) {
+        /* 0x40 ended longest ago but a reader holds it; 0x41 ended next; 0x42 ended but still settles */
+        conclude(0x40, 5000);
+        conclude(0x41, 4000);
+        conclude(0x42, 6000);
+        memset(id, 0x42, 16);
+        if ((j = ava1_job_find(id)) != NULL) {
+            __atomic_store_n(&j->unswept_n, 1, __ATOMIC_RELAXED);
+            ava1_job_put(j);
+        }
+        memset(id, 0x40, 16);
+        held = ava1_job_find(id);
+        if (!(j = mk(0x71, SID))) rc = -3; /* 0x41 makes room */
+        else ava1_job_put(j);
+        if (!rc && listed(0x41)) rc = -4;
+        if (!rc && (!listed(0x40) || !listed(0x42))) rc = -5;
+        if (held) ava1_job_put(held);
+        if (!rc && !(j = mk(0x72, SID))) rc = -6; /* now 0x40 makes room */
+        else if (!rc) ava1_job_put(j);
+        if (!rc && listed(0x40)) rc = -7;
+        if (!rc && (j = mk(0x73, SID)) != NULL) { /* only the settling 0x42 has ended: still full */
+            ava1_job_put(j);
+            rc = -8;
+        }
+    }
+    if (!rc) {
+        /* the reaper: an ended upload still attached goes 10 s after its end, not before; a running one stays */
+        conclude(0x43, 0);
+        ava1_job_reap(ava1_mono_ms() + 5000u);
+        if (!listed(0x43)) rc = -9;
+        ava1_job_reap(ava1_mono_ms() + 11000u);
+        if (!rc && listed(0x43)) rc = -10;
+        if (!rc && (!listed(0x44) || !listed(0x42))) rc = -11;
+        /* a copy is read by job id after it ends: never taken this way */
+        memset(id, 0x45, 16);
+        if (!rc && (j = ava1_job_find(id)) != NULL) {
+            j->kind = AVA1_JOB_COPY;
+            __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE);
+            ava1_job_put(j);
+            ava1_job_reap(ava1_mono_ms() + 11000u);
+            if (!listed(0x45)) rc = -12;
+        }
+    }
+    ava1_job_free_all();
+    ava1_data_stop();
+    return rc;
+}
+
 /* N3: a job a cancel unlisted, still referenced by the caller and one other holder. */
 int ava1_test_retire_unlisted(void) {
     ava1_data_cfg_t dc;

engine/crates/ava1-ctest/src/lib.rs

@@ -460,6 +460,7 @@ pub mod ffi {
         pub fn ava1_test_data_delays(open_ms: u32, map_ms: u32);
         pub fn ava1_test_job_attached(id: *const u8) -> c_int;
         pub fn ava1_test_reap_rules(jobs_dir: *const c_char) -> c_int;
+        pub fn ava1_test_ended_transfers(jobs_dir: *const c_char) -> c_int;
         pub fn ava1_test_retire_unlisted() -> c_int;
         pub fn ava1_test_job_counts(id: *const u8, out: *mut u64) -> c_int;
         pub fn ava1_test_data_knob(name: *const c_char, v: u32) -> c_int;
@@ -1071,6 +1072,14 @@ pub fn c_reap_rules(jobs: &Path) -> i32 {
     unsafe { ffi::ava1_test_reap_rules(j.as_ptr()) }
 }
 
+/// Ended uploads on a live session leave the job table (#475), checked in C on a private data
+/// layer (0 = all hold, else the step that failed). Holds the C server lock.
+pub fn c_ended_transfers(jobs: &Path) -> i32 {
+    let _lock = C_SERVER.lock().unwrap_or_else(|e| e.into_inner());
+    let j = CString::new(jobs.to_str().unwrap()).unwrap();
+    unsafe { ffi::ava1_test_ended_transfers(j.as_ptr()) }
+}
+
 extern "C" {
     fn ava1_test_firmware(kernel_version: *const c_char, out: *mut c_char, cap: usize);
 }

engine/crates/ava1-ctest/tests/many_jobs_one_session.rs

@@ -0,0 +1,99 @@
+#![cfg(unix)]
+//! Issue #475: the Files screen uploads a pick of many files one job per file, all on the
+//! engine's one shared session. The console's job table holds 32 jobs, and an upload that ended
+//! stayed listed until its session ended, so the 33rd file was answered BUSY for 12 retries and
+//! failed ("the console stayed busy"). Ended uploads must make room on a session that runs on.
+mod common;
+
+use std::sync::atomic::{AtomicBool, AtomicU64};
+use std::sync::Arc;
+
+use ava1::keys::Identity;
+use ava1::peers::PeerStore;
+use ava1_ctest::{c_job_count, CServer, LogOpts};
+use common::{dir, SECRET};
+use ps5upload_ava1::upload::upload_file_in;
+use ps5upload_ava1::Pool;
+use ps5upload_core::transfer::TransferConfig;
+
+fn cfg(addr: &str) -> TransferConfig {
+    let mut c = TransferConfig::new(addr);
+    c.progress_bytes = Some(Arc::new(AtomicU64::new(0)));
+    c.progress_files = Some(Arc::new(AtomicU64::new(0)));
+    c.progress_files_finalized = Some(Arc::new(AtomicU64::new(0)));
+    c.progress_bytes_finalized = Some(Arc::new(AtomicU64::new(0)));
+    c.cancel = Some(Arc::new(AtomicBool::new(false)));
+    c
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn forty_single_file_jobs_in_a_row_on_one_session_all_land() {
+    let t = dir("many-jobs");
+    let ava = t.join("ava");
+    std::fs::create_dir_all(&ava).unwrap();
+    let me = Identity::load_or_create(&ava.join("identity")).unwrap();
+    PeerStore::load(&t.join("srv-peers"))
+        .unwrap()
+        .add(me.public(), "engine")
+        .unwrap();
+    PeerStore::load(&ava.join("peers"))
+        .unwrap()
+        .add(Identity::from_secret(SECRET).public(), "C receiver")
+        .unwrap();
+    let srv = CServer::start_data_opts(
+        SECRET,
+        &t.join("srv-peers"),
+        &t.join("jobs"),
+        200,
+        2000,
+        2000,
+        0,
+        0,
+        LogOpts::ON,
+    );
+    let src = t.join("src");
+    let dest = t.join("dest");
+    std::fs::create_dir_all(&src).unwrap();
+    std::fs::create_dir_all(&dest).unwrap();
+    let pool = Arc::new(Pool::new(ava.clone()).with_addr(srv.addr()));
+    let mut slowest = std::time::Duration::ZERO;
+    for n in 0..40u32 {
+        let f = src.join(format!("f{n}.nca"));
+        std::fs::write(&f, vec![n as u8; 1000 + n as usize * 37]).unwrap();
+        let mut id = [0x47u8; 16];
+        id[..4].copy_from_slice(&n.to_le_bytes());
+        let (p, c, d, s) = (
+            pool.clone(),
+            cfg(&srv.addr()),
+            dest.join(format!("f{n}.nca")).to_str().unwrap().to_string(),
+            f.clone(),
+        );
+        let started = std::time::Instant::now();
+        tokio::task::spawn_blocking(move || upload_file_in(&p, &c, id, &d, &s))
+            .await
+            .unwrap()
+            .unwrap_or_else(|e| panic!("file {n}: {e:#} (jobs listed: {})", c_job_count()));
+        slowest = slowest.max(started.elapsed());
+        assert!(c_job_count() <= 32);
+    }
+    for n in 0..40u32 {
+        assert_eq!(
+            std::fs::read(dest.join(format!("f{n}.nca"))).unwrap(),
+            std::fs::read(src.join(format!("f{n}.nca"))).unwrap()
+        );
+    }
+    // A BUSY answer costs the sender at least one backoff (250 ms and doubling); a job that
+    // waited for a slot shows up as a slow file.
+    assert!(
+        slowest < std::time::Duration::from_secs(5),
+        "a file waited for a table slot: slowest {slowest:?}"
+    );
+    // The session is still up (the pool holds it): the ended jobs leave anyway.
+    ava1_ctest::c_reap_far();
+    assert_eq!(
+        c_job_count(),
+        0,
+        "ended jobs stayed listed on a live session"
+    );
+    drop(pool);
+}

engine/crates/ava1-ctest/tests/wire_upload.rs

@@ -640,6 +640,15 @@ fn the_reaper_takes_parked_jobs_only_and_never_one_just_attached() {
     assert_eq!(ava1_ctest::c_reap_rules(&d.join("jobs")), 0);
 }
 
+#[test]
+fn ended_uploads_leave_the_table_while_their_session_runs_on() {
+    // #475, in C: a session runs job after job, so the uploads that ended on it make room for
+    // the next JobOpen (the one that ended longest ago and nobody holds goes first) and leave
+    // 10 s after their end; running or still-settling jobs never do, nor a copy.
+    let d = dir("wire-ended");
+    assert_eq!(ava1_ctest::c_ended_transfers(&d.join("jobs")), 0);
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn the_server_restarts_on_the_same_port() {
     let d = dir("wire-restart");

payload/ava1/ava1_apply.c

@@ -524,6 +524,7 @@ void ava1_apply_fail(ava1_job_t *j, uint16_t status, const char *what, int err,
     }
     pthread_mutex_unlock(&j->mu);
     emit_done(j);
+    __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE); /* the table may now let it go (ava1_job.c) */
 }
 
 void ava1_apply_done_again(ava1_job_t *j) {

payload/ava1/ava1_job.c

@@ -71,8 +71,36 @@ ava1_job_t *ava1_job_find(const uint8_t id[16]) {
     return j;
 }
 
+/* An upload or download that ended and has nothing left to settle: what it still holds (its threads, its
+ * slot) serves only a late Status or a reopen, and a reopen replays its journal (an upload) or starts over (a
+ * download). A copy, an operation and a send to another console keep their own rules: their status is read
+ * by job id after they end. `concluded_ms` (set after `finished`, once the job's JobDone is out: its last commit
+ * and journal append are behind it) and `unswept_n` are read without j->mu, as in the reaper. */
+static int transfer_ended(const ava1_job_t *j) {
+    return (j->kind == AVA1_JOB_UPLOAD || j->kind == AVA1_JOB_DOWNLOAD) && !j->c2c &&
+           __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) && !__atomic_load_n(&j->unswept_n, __ATOMIC_RELAXED);
+}
+
+/* Caller holds T.mu, and the table is full: the ended transfer nobody else holds (only the table's
+ * reference) that ended longest ago, or -1. A session runs job after job (the Files screen sends a pick of
+ * files one job each, #475), so its ended jobs must make room for the next one rather than answer BUSY. */
+static int evictable_locked(void) {
+    int i, best = -1;
+    for (i = 0; i < AVA1_MAX_JOBS; i++) {
+        ava1_job_t *j = T.jobs[i];
+        if (!j || j->refs != 1 || !transfer_ended(j)) continue;
+        if (best < 0 || __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) <
+                            __atomic_load_n(&T.jobs[best]->concluded_ms, __ATOMIC_ACQUIRE))
+            best = i;
+    }
+    return best;
+}
+
+static void unlist(ava1_job_t *j);
+
 static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const uint8_t *sid) {
     ava1_job_t *j = calloc(1, sizeof *j);
+    ava1_job_t *evicted = NULL;
     int i, slot = -1;
     if (!j) return NULL;
     memcpy(j->id, id, 16);
@@ -100,6 +128,11 @@ static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const u
         }
         if (!T.jobs[i] && slot == -1) slot = i;
     }
+    if (slot == -1 && (i = evictable_locked()) >= 0) {
+        evicted = T.jobs[i];
+        unlist(evicted); /* its id stays retiring until the put below has destroyed it */
+        slot = i;
+    }
     if (slot >= 0)
         for (i = 0; i < (int)T.retiring_n; i++)
             if (memcmp(T.retiring[i], id, 16) == 0) {
@@ -112,6 +145,7 @@ static ava1_job_t *create(const uint8_t id[16], const uint8_t owner[32], const u
     if (slot >= 0 && T.retiring_n >= AVA1_MAX_JOBS) slot = -2;
     if (slot >= 0) T.jobs[slot] = j;
     pthread_mutex_unlock(&T.mu);
+    if (evicted) ava1_job_put(evicted); /* the table's reference, the last one: joins its idle threads */
     if (slot < 0) {
         pthread_mutex_destroy(&j->mu);
         pthread_mutex_destroy(&j->jnl_mu);
@@ -357,7 +391,18 @@ void ava1_job_reap(uint64_t now_ms) {
          * j->mu: that lock can be held across disk I/O, and waiting for it under T.mu would
          * stall every reader thread's job lookup. It only ever goes 0 -> 1, so a stale read
          * reaps later. */
-        if (!j || j->attached || j->parked_at_ms == 0) continue;
+        if (!j) continue;
+        if (j->attached) {
+            /* An upload or download ends on the session that ran it, and that session goes on to the
+             * next job: the ended one leaves the table 10 s after it ended, as if its session had let go
+             * of it (SPEC.md §11.7), not when the session ends (#475). */
+            if (transfer_ended(j) && now_ms > __atomic_load_n(&j->concluded_ms, __ATOMIC_ACQUIRE) + done_age) {
+                gone[n++] = j;
+                unlist(j);
+            }
+            continue;
+        }
+        if (j->parked_at_ms == 0) continue;
         fin = __atomic_load_n(&j->finished, __ATOMIC_ACQUIRE);
         /* Files still settling (durable-by-log) keep their job alive for the full park age: it owns the
          * sweep, and the engine polls its Status for `unswept`. */

payload/ava1/ava1_job.h

@@ -92,6 +92,8 @@ struct ava1_job {
     int attached, refs, stopping, finished, prepared;
     int discard_parts;              /* a cancelled local copy: destroy removes the .ava-part files this job itself was writing (final review #9) */
     uint64_t parked_at_ms;
+    uint64_t concluded_ms;          /* an upload or download: when its end went out (our JobDone) or came in (the peer's),
+                                     * nothing of it in flight after; 0 before. The table lets it go from then (#475) */
     int op_delivered;               /* an operation job, or a console-page copy (ava1_jobs_release_seen): its terminal status was
                                      * first delivered at parked_at_ms (the grace starts) */
     int c2c;                        /* a send to another console (SPEC.md §18): kept the full park age once ended */

payload/ava1/ava1_send.c

@@ -734,6 +734,7 @@ static int on_frame(ava1_job_t *j, uint8_t type, const uint8_t *body, size_t len
         }
         s->stop = 1;
         j->finished = 1;
+        __atomic_store_n(&j->concluded_ms, ava1_mono_ms() | 1u, __ATOMIC_RELEASE); /* the peer ended it: the table may let it go */
         pthread_cond_broadcast(&j->cv);
         pthread_mutex_unlock(&j->mu);
         return 0;

protocol/ava1/SPEC.md

@@ -720,7 +720,10 @@ or `job.copy` whose root is, or lies inside or around, the root of another job t
 (and, for a move, its source) is answered `ERR_BUSY`; parked jobs count, because they can resume.
 When a session ends its jobs are parked, not ended: they stay listed, detached, for 10 minutes
 and then leave the table; their journal stays on disk (§14.3), so a later `JobOpen` resumes them.
-A finished upload or download leaves the table 10 s after its session lets go of it; a finished local job
+A finished upload or download leaves the table 10 s after it ended (its `JobDone` went out or came in), whether
+or not its session goes on: a session runs job after job, so the jobs it ended must not hold the table until it
+ends. A `JobOpen` that finds the table full takes the slot of the ended upload or download (with nothing left
+to settle) that ended longest ago instead of answering `ERR_BUSY` (a reopen of an upload replays its journal; a download starts over). A finished local job
 (`job.copy`) stays listed for the full park age so `job.status` can still answer. Either peer ends a
 job with `JobCancel{job_id, reason}`, where `reason` is the `ERR_*` code the ender wants reported
 (`ERR_CANCELLED` for a user's cancel, `ERR_IO` or `ERR_VERIFY` when a sender's source fails); the