ava1: release the console page's finished jobs once it has seen them

ava1_jobs_release_seen(me): every finished operation or copy owned by
the console page keeps only the grace (AVA1_OP_GRACE_MS) after the page
first read it finished, instead of a copy holding one of the 32 shared
job slots for the full 10-minute park age. Operations reuse
ava1_op_status_delivered; copies are restamped and the reaper keeps a
delivered copy only for the grace. A paired PC's jobs are never touched.

ava1-ctest: the 32-job cap with list_all (full table refuses a 33rd for
the page and a PC; releasing the page's own makes room, PC jobs stay)
and the grace (page jobs gone after ~10 s, PC op and copy still listed).

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

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

@@ -2507,6 +2507,42 @@ extern "C" {
     fn ava1_jobs_list_all(me: *const u8, out: *mut u8, cap: usize, out_len: *mut usize) -> c_int;
     fn ava1_jobs_owner_is(id: *const u8, me: *const u8) -> c_int;
     fn ava1_job_cancel_any(id: *const u8) -> c_int;
+    fn ava1_jobs_release_seen(me: *const u8);
+    fn ava1_server_set_rpc_peer(peer: *const u8);
+    fn ava1_data_rpc(
+        method: u16,
+        body: *const u8,
+        len: u32,
+        out: *mut u8,
+        cap: usize,
+        out_len: *mut usize,
+    ) -> c_int;
+}
+
+/// `ava1_jobs_release_seen` (the console page read /api/jobs): `me`'s finished operations and
+/// copies keep only the grace.
+pub fn c_jobs_release_seen(me: &[u8; 32]) {
+    unsafe { ava1_jobs_release_seen(me.as_ptr()) }
+}
+
+/// A data-layer RPC (`job.copy`, `job.run`, ...) as `me`, the way the console page's routes call
+/// it (`ava1_server_set_rpc_peer` on the calling thread, then the data plane): (status, reply).
+pub fn c_data_rpc_as(me: &[u8; 32], method: u16, body: &[u8]) -> (i32, Vec<u8>) {
+    let mut out = vec![0u8; 64 * 1024];
+    let mut n = 0usize;
+    let rc = unsafe {
+        ava1_server_set_rpc_peer(me.as_ptr());
+        ava1_data_rpc(
+            method,
+            body.as_ptr(),
+            body.len() as u32,
+            out.as_mut_ptr(),
+            out.len(),
+            &mut n,
+        )
+    };
+    out.truncate(n);
+    (rc, out)
 }
 
 /// `ava1_jobs_list_all` (the console page's job list): every job the data layer holds, of

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

@@ -184,3 +184,167 @@ async fn job_list_rpc_unchanged_for_pcs() {
     assert_eq!(all, vec![ja, jb]);
     drop(t.srv);
 }
+
+const FSCK_NOW: &[u8] = br#"{"device":"/dev/md1","loops":0}"#;
+
+fn fsck_body(job: [u8; 16]) -> Vec<u8> {
+    JobRun {
+        job_id: job,
+        op: gen::JOB_OP_FSCK,
+        args: FSCK_NOW.to_vec(),
+    }
+    .to_bytes()
+    .unwrap()
+}
+
+/// `job.run` of a quick fsck as the console page (the page's routes call the data layer with its
+/// identity set on the thread): the status.
+fn run_as_console(job: [u8; 16]) -> i32 {
+    c_data_rpc_as(&CONSOLE, gen::METHOD_JOB_RUN, &fsck_body(job)).0
+}
+
+/// `job.copy` as the console page.
+fn copy_as_console(job: [u8; 16], src: &std::path::Path, dest: &std::path::Path) -> i32 {
+    let body = JobCopy {
+        job_id: job,
+        src: src.to_str().unwrap().into(),
+        dest: dest.to_str().unwrap().into(),
+        flags: 0,
+    };
+    c_data_rpc_as(&CONSOLE, gen::METHOD_JOB_COPY, &body.to_bytes().unwrap()).0
+}
+
+/// Waits until every job in `ids` is listed and finished.
+async fn all_finished(ids: &[[u8; 16]]) {
+    for _ in 0..400 {
+        let all = list_all(&CONSOLE);
+        if ids
+            .iter()
+            .all(|id| all.iter().any(|e| e.job_id == *id && e.state != 0))
+        {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(25)).await;
+    }
+    panic!("jobs did not finish");
+}
+
+/// Final review: the 32-slot job table is shared by every owner. Finished jobs hold their slot
+/// until they are released; the console page lists all 32 and, once it has read them finished,
+/// releases only its own, which frees room for new jobs.
+#[tokio::test(flavor = "multi_thread")]
+async fn table_holds_32_jobs_and_seen_console_jobs_make_room() {
+    let t = two("any-cap32", 0).await;
+    let mut pc = Vec::new();
+    let mut mine = Vec::new();
+    for n in 0..32u8 {
+        let id = [0x40 + n; 16];
+        // At most eight operations run at once: a quick one is retried while the others end.
+        let mut st = gen::ERR_BUSY as i32;
+        for _ in 0..200 {
+            st = if n % 2 == 0 {
+                run_fsck_now(&t.a, id).await as i32
+            } else {
+                run_as_console(id)
+            };
+            if st != gen::ERR_BUSY as i32 {
+                break;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+        assert_eq!(st, gen::STATUS_OK as i32, "job {n}");
+        if n % 2 == 0 {
+            pc.push(id)
+        } else {
+            mine.push(id)
+        }
+    }
+    let every: Vec<_> = pc.iter().chain(mine.iter()).copied().collect();
+    all_finished(&every).await;
+    // list_all fits all 32 in one reply, whoever owns them.
+    let all = list_all(&CONSOLE);
+    assert_eq!(all.len(), 32);
+    for id in &mine {
+        assert_eq!(c_jobs_owner_is(id, &CONSOLE), 1);
+    }
+    // Full, and nothing was delivered: a 33rd job is refused, for the page as for a PC.
+    assert_eq!(run_as_console([0x70; 16]), gen::ERR_BUSY as i32);
+    assert_eq!(run_fsck_now(&t.b, [0x71; 16]).await, gen::ERR_BUSY);
+    // The page read them finished: its own are released (oldest first when room is needed) ...
+    c_jobs_release_seen(&CONSOLE);
+    assert_eq!(list_all(&CONSOLE).len(), 32, "the grace keeps them listed");
+    assert_eq!(run_as_console([0x70; 16]), gen::STATUS_OK as i32);
+    assert_eq!(run_fsck_now(&t.b, [0x71; 16]).await, gen::STATUS_OK);
+    // ... and a PC's job never is.
+    let all = list_all(&CONSOLE);
+    assert_eq!(all.len(), 32);
+    for id in &pc {
+        assert!(
+            all.iter().any(|e| e.job_id == *id),
+            "a PC's job stays listed"
+        );
+    }
+    drop(t.srv);
+}
+
+/// A finished copy or operation the console page started is released a grace after the page first
+/// read it finished (not the 10-minute park age); a paired PC's finished jobs stay listed.
+#[tokio::test(flavor = "multi_thread")]
+async fn seen_console_jobs_leave_after_the_grace_pc_jobs_stay() {
+    let t = two("any-seen", 0).await;
+    write_tree(&t.d.join("usb/g"), 5, |_| 1024);
+    let (my_op, my_copy, pc_op, pc_copy) = ([0x81; 16], [0x82; 16], [0x83; 16], [0x84; 16]);
+    assert_eq!(run_as_console(my_op), gen::STATUS_OK as i32);
+    assert_eq!(
+        copy_as_console(my_copy, &t.d.join("usb/g"), &t.d.join("data/mine")),
+        gen::STATUS_OK as i32
+    );
+    assert_eq!(run_fsck_now(&t.a, pc_op).await, gen::STATUS_OK);
+    assert_eq!(
+        copy(&t.a, pc_copy, &t.d.join("usb/g"), &t.d.join("data/pc")).await,
+        gen::STATUS_OK
+    );
+    all_finished(&[my_op, my_copy, pc_op, pc_copy]).await;
+    assert_eq!(c_jobs_owner_is(&my_copy, &CONSOLE), 1);
+    c_jobs_release_seen(&CONSOLE);
+    let started = std::time::Instant::now();
+    let listed = |id: [u8; 16]| list_all(&CONSOLE).iter().any(|e| e.job_id == id);
+    assert!(
+        listed(my_op) && listed(my_copy),
+        "still listed during the grace"
+    );
+    // The page reads /api/jobs about once a second.
+    let mut gone = false;
+    while started.elapsed() < Duration::from_secs(20) {
+        c_jobs_release_seen(&CONSOLE);
+        if !listed(my_op) && !listed(my_copy) {
+            gone = true;
+            break;
+        }
+        tokio::time::sleep(Duration::from_millis(250)).await;
+    }
+    assert!(
+        gone,
+        "the page's finished jobs are released after the grace"
+    );
+    assert!(
+        started.elapsed() >= Duration::from_secs(9),
+        "not before the grace: {:?}",
+        started.elapsed()
+    );
+    assert!(listed(pc_op), "a PC's finished operation stays");
+    assert!(listed(pc_copy), "a PC's finished copy stays");
+    // The PC still reads its own results.
+    assert_eq!(
+        code(&t.a, gen::METHOD_JOB_STATUS, pc_copy).await,
+        gen::STATUS_OK
+    );
+    drop(t.srv);
+}
+
+async fn run_fsck_now(s: &Session, job: [u8; 16]) -> u16 {
+    s.rpc(gen::METHOD_JOB_RUN, &fsck_body(job))
+        .await
+        .unwrap()
+        .status
+}

payload/ava1/ava1_job.c

@@ -377,7 +377,9 @@ void ava1_job_reap(uint64_t now_ms) {
         if (j->kind == AVA1_JOB_OPKIND
                 ? (fin && now_ms - j->parked_at_ms > ava1_op_keep_ms(__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE)))
                 : j->kind == AVA1_JOB_COPY || j->c2c
-                ? (fin && now_ms - j->parked_at_ms > age)
+                /* a copy the console page has seen finished (ava1_jobs_release_seen) keeps only the grace */
+                ? (fin && now_ms - j->parked_at_ms >
+                              (__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE) ? ava1_op_keep_ms(1) : age))
                 : (now_ms - j->parked_at_ms > age || (fin && now_ms - j->parked_at_ms > done_age))) {
             gone[n++] = j;
             unlist(j);

payload/ava1/ava1_job.h

@@ -92,7 +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;
-    int op_delivered;               /* an operation job: its terminal status was first delivered at parked_at_ms (the grace starts) */
+    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 */
     uint8_t c2c_by[32];             /* ... and the device that asked for it, which may read its status or cancel it */
     uint8_t kind, policy;

payload/ava1/ava1_op.c

@@ -460,6 +460,45 @@ int ava1_jobs_list_all(const uint8_t me[32], uint8_t *out, size_t cap, size_t *o
     return list_jobs(NULL, out, cap, out_len);
 }
 
+typedef struct {
+    const uint8_t *me;
+    uint8_t id[AVA1_MAX_JOBS][16];
+    unsigned n;
+} seen_t;
+
+static void seen_one(ava1_job_t *j, void *ctx) {
+    seen_t *s = ctx;
+    if (s->n >= AVA1_MAX_JOBS || memcmp(j->owner, s->me, 32) != 0) return;
+    if (j->kind != AVA1_JOB_OPKIND && j->kind != AVA1_JOB_COPY) return;
+    if (!__atomic_load_n(&j->finished, __ATOMIC_ACQUIRE)) return;
+    /* the listing's rule (list_one): a move is running until its source is gone */
+    if (j->kind == AVA1_JOB_COPY && __atomic_load_n(&j->copy_move, __ATOMIC_ACQUIRE) &&
+        !__atomic_load_n(&j->copy_delete_done, __ATOMIC_ACQUIRE))
+        return;
+    memcpy(s->id[s->n++], j->id, 16);
+}
+
+void ava1_jobs_release_seen(const uint8_t me[32]) {
+    seen_t *s = calloc(1, sizeof *s);
+    unsigned i;
+    if (!s) return;
+    s->me = me;
+    ava1_job_foreach(seen_one, s); /* under the table lock: only collects ids */
+    for (i = 0; i < s->n; i++) {
+        ava1_job_t *j = ava1_job_find(s->id[i]);
+        if (!j) continue;
+        if (j->kind == AVA1_JOB_OPKIND) {
+            ava1_op_status_delivered(j, 1);
+        } else if (!__atomic_load_n(&j->op_delivered, __ATOMIC_ACQUIRE)) {
+            /* a copy: the reaper releases it once the grace from now has passed */
+            __atomic_store_n(&j->parked_at_ms, ava1_mono_ms(), __ATOMIC_RELEASE);
+            __atomic_store_n(&j->op_delivered, 1, __ATOMIC_RELEASE);
+        }
+        ava1_job_put(j);
+    }
+    free(s);
+}
+
 int ava1_jobs_owner_is(const uint8_t id[16], const uint8_t me[32]) {
     ava1_job_t *j = ava1_job_find(id);
     int mine;

payload/ava1/ava1_op.h

@@ -57,6 +57,15 @@ int ava1_op_list_rpc(const uint8_t owner[32], uint8_t *out, size_t cap, size_t *
 int ava1_jobs_list_all(const uint8_t me[32], uint8_t *out, size_t cap, size_t *out_len);
 /* 1 when job `id` is listed and `me` owns it, else 0 (an absent job included). */
 int ava1_jobs_owner_is(const uint8_t id[16], const uint8_t me[32]);
+/* The console page has read the job list: every FINISHED operation or copy that `me` owns is
+ * treated as delivered, exactly as a JOB_STATUS reply delivers an operation's end
+ * (ava1_op_status_delivered): it stays listed for the grace (AVA1_OP_GRACE_MS, capped by the park
+ * age) after the first such read, so other readers still see how it ended, and is then released
+ * (an operation also by the next read after the grace, and at once when the table is full). Without
+ * this a page-started copy stayed listed the full park age (10 min) and 32 of them filled the table
+ * every owner shares. A move still deleting its source is not finished. Jobs of any other owner (a
+ * paired PC's) are never touched. Never call it with the table lock held. */
+void ava1_jobs_release_seen(const uint8_t me[32]);
 /* Encodes the Status of an operation job (job.status / job.run reply). */
 int ava1_op_encode_status(ava1_job_t *j, uint8_t *out, size_t cap, size_t *out_len);
 /* job.cancel on an operation job: raises the flag and returns (never waits); the worker stops at