Merge #475: finished uploads leave the job table

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