Record a decided tab retirement even when the write outlasts its budget

CHANGELOG.md

@@ -13,4 +13,5 @@ First public release. Industrious is a Rust port of Calvino, a maintained fork o
 - `serve` prints its access token only when it generates one, adds `--token-file`, compares tokens without early exit, refuses an empty token and answers HTTP 413 `payload_too_large` for run requests over 1,432,704,342 bytes (Oracle's limit).
 - File selection fails closed: `.gitignore` rules apply from the root of the repository holding each selected path, a `.gitignore` that cannot be read or parsed is an error, files read into an API prompt or a browser text bundle must be UTF-8 text rather than being garbled (browser uploads and ZIP bundles still send the original bytes), and credential and private-key files are refused by name and skipped in expansion.
 - File sections in rendered prompts and bundles keep every line that ended, dropping only the final line ending, so a file ending in a whitespace-only line (as a diff can) arrives whole.
+- A tab retirement that has reached its outcome always records it: a slow disk write at the end of the 5 s budget no longer reports a false cleanup failure and leaves the record undecided, with the tab already closed.
 - Licensed under FSL-1.1-ALv2; each version also becomes available under Apache-2.0 two years after its release.

src/browser_disposition.rs

@@ -386,7 +386,7 @@ impl TargetJournal {
                     Self::saved_retirement_id(&record)?;
                 }
                 remaining(operation_deadline)?;
-                // Leave a tail for the disposition write if a CDP operation stalls.
+                // Stop a stalled CDP operation early so its disposition normally lands in budget.
                 let result =
                     tokio::time::timeout(remaining(operation_deadline)?, retire(record.clone()))
                         .await;
@@ -426,8 +426,13 @@ impl TargetJournal {
             &reason,
             &absence,
         );
-        let saved = self.save_until(&record, deadline)?;
-        lock.release()?;
+        // The outcome is decided, so its record outlasts the deadline: a stalled write that gave
+        // up here would report a false failure and leave the retirement undecided. The deadline
+        // bounds the operation and the lock above. This write stays under the record's lock.
+        let saved = self.save(&record)?;
+        // The lock would otherwise inherit the spent deadline, and a release that fails once is
+        // not retried, so a contended release would strand it.
+        lock.release_until(Instant::now() + Duration::from_secs(5))?;
         Ok(saved["disposition"].clone())
     }
     pub(crate) async fn retire_settled_before_dispatch<F, Fut>(
@@ -1526,6 +1531,72 @@ mod tests {
         assert!(journal.retirement_id(&creator, &changed).is_err());
         assert!(journal.retirement_id(&creator, &commit).is_err());
     }
+    /// Blocks the thread past the budget, as a stalled disk would; a timer would yield instead.
+    fn stall_past(budget: Duration) {
+        std::thread::sleep(budget + Duration::from_millis(50));
+    }
+    #[tokio::test]
+    async fn a_decision_reached_as_the_budget_ends_is_published_despite_the_stall() {
+        for (budget, absence, decision) in [
+            (Duration::from_secs(2), "confirmed-absent", "retired"),
+            (Duration::from_millis(500), "unknown", "retained"),
+        ] {
+            let home = tempfile::tempdir().unwrap();
+            let (journal, creator, commit) = native(home.path());
+            journal.commit(&creator, &commit).unwrap();
+            journal.finalise(&commit).unwrap();
+            let deadline = Instant::now() + budget;
+            // Blocking before the operation starts stands in for a stall between the evidence and
+            // its durable write; the operation still completes on its first poll, as a real close
+            // can.
+            let result = journal
+                .retire_record_until(&creator, deadline, |_| {
+                    stall_past(budget);
+                    std::future::ready(Ok((
+                        absence.to_owned(),
+                        "PUBLIC stalled evidence".to_owned(),
+                    )))
+                })
+                .await
+                .unwrap();
+            assert_eq!(result["decision"], decision);
+            assert_eq!(result["absence"], absence);
+            assert!(Instant::now() >= deadline);
+            let saved = journal.read(&creator["allocationId"]).unwrap();
+            assert_eq!(saved["disposition"], result);
+        }
+    }
+    #[tokio::test]
+    async fn a_late_publication_releases_its_lock_despite_contention() {
+        let home = tempfile::tempdir().unwrap();
+        let (journal, creator, commit) = native(home.path());
+        journal.commit(&creator, &commit).unwrap();
+        journal.finalise(&commit).unwrap();
+        let budget = Duration::from_millis(500);
+        let lock = journal
+            .file(&creator["allocationId"])
+            .unwrap()
+            .with_extension("lock");
+        let contender = lock.clone();
+        let holder = std::cell::Cell::new(None);
+        let result = journal
+            .retire_record_until(&creator, Instant::now() + budget, |_| {
+                stall_past(budget);
+                holder.set(Some(crate::storage::hold_recovery_mutex(
+                    &contender,
+                    Duration::from_millis(500),
+                )));
+                std::future::ready(Ok((
+                    "unknown".to_owned(),
+                    "PUBLIC stalled evidence".to_owned(),
+                )))
+            })
+            .await
+            .unwrap();
+        assert_eq!(result["decision"], "retained");
+        assert!(!lock.exists());
+        holder.take().unwrap().join().unwrap();
+    }
     #[test]
     fn creator_replacement_and_directory_replacement_refuse_writes() {
         let home = tempfile::tempdir().unwrap();

src/storage.rs

@@ -552,6 +552,20 @@ fn normalise_windows_path_identity(raw: &str) -> String {
     raw.to_lowercase()
 }
 
+/// Holds a lock's recovery mutex on another thread, returning once it is held.
+#[cfg(test)]
+pub(crate) fn hold_recovery_mutex(path: &Path, hold: Duration) -> std::thread::JoinHandle<()> {
+    let path = path.to_owned();
+    let (held, wait) = std::sync::mpsc::channel();
+    let holder = std::thread::spawn(move || {
+        let mutex = RecoveryMutex::acquire(&path, Duration::from_secs(5)).unwrap();
+        held.send(()).unwrap();
+        std::thread::sleep(hold);
+        drop(mutex);
+    });
+    wait.recv().unwrap();
+    holder
+}
 struct RecoveryMutex {
     #[cfg(unix)]
     _file: File,