Flake batch 5: task-bundle lock, storm config publish, macOS thread inventory

benchmarks/aft-search/slice-descriptors/land-flake-batch-5.json

@@ -0,0 +1,12 @@
+{
+  "slice_class": "non_ranking",
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "fixtures": [
+    "harness-goldens"
+  ],
+  "slices": [
+    "flake-batch-5"
+  ],
+  "note": "Test-only: timing and race fixes for merge-queue flakes (generous positive waits, waits on real signals). No product search behavior changed and no changed path matches RANKING_FENCE_PREFIXES."
+}

crates/aft/src/subc/idle_root_retention_tests.rs

@@ -450,7 +450,7 @@ mod thread_inventory {
         struct ThreadState {
             _x: [u64; 29],
             fp: u64,
-            _lr: u64,
+            lr: u64,
             _sp: u64,
             pc: u64,
             _cpsr: u32,
@@ -598,11 +598,22 @@ mod thread_inventory {
             (result == 0 && read == 16).then_some(frame)
         }
 
+        /// The registers of a suspended thread that locate its stack.
+        struct Registers {
+            pc: u64,
+            fp: u64,
+            /// The link register on arm64, which holds the running
+            /// function's return address until that function saves it in its
+            /// frame record. x86_64 keeps the return address on the stack
+            /// instead, so it has none.
+            lr: Option<u64>,
+        }
+
         /// A thread's saved registers and return addresses.
         struct Stack {
-            /// Program counter and frame pointer; `None` when the thread could
-            /// not be suspended or its registers could not be read.
-            registers: Option<(u64, u64)>,
+            /// `None` when the thread could not be suspended or its registers
+            /// could not be read.
+            registers: Option<Registers>,
             /// Return addresses, innermost first.
             returns: Vec<u64>,
         }
@@ -656,28 +667,51 @@ mod thread_inventory {
             unsafe {
                 thread_resume(port);
             }
+            #[cfg(target_arch = "aarch64")]
+            let lr = Some(state.lr & ADDRESS_MASK);
+            #[cfg(target_arch = "x86_64")]
+            let lr = None;
             Stack {
-                registers: got_state.then_some((state.pc & ADDRESS_MASK, state.fp & ADDRESS_MASK)),
+                registers: got_state.then_some(Registers {
+                    pc: state.pc & ADDRESS_MASK,
+                    fp: state.fp & ADDRESS_MASK,
+                    lr,
+                }),
                 returns: frames[..depth].to_vec(),
             }
         }
 
+        /// What `dladdr` knows about `address`; `None` outside every image.
+        fn lookup(address: u64) -> Option<libc::Dl_info> {
+            // SAFETY: Dl_info is plain C data that dladdr fills.
+            let mut info: libc::Dl_info = unsafe { std::mem::zeroed() };
+            // SAFETY: dladdr only reads loader tables for the address.
+            (unsafe { libc::dladdr(address as *const libc::c_void, &mut info) } != 0)
+                .then_some(info)
+        }
+
+        /// The demangled name of the symbol recorded in `info`, if `dladdr`
+        /// found one.
+        fn symbol_name(info: &libc::Dl_info) -> Option<String> {
+            if info.dli_sname.is_null() {
+                return None;
+            }
+            // SAFETY: dladdr returned a NUL-terminated symbol name.
+            let name = unsafe { CStr::from_ptr(info.dli_sname) }.to_string_lossy();
+            // `{:#}` leaves out the hash suffix; other names pass through.
+            Some(format!("{:#}", rustc_demangle::demangle(&name)))
+        }
+
         /// The symbol containing `address`, demangled, or the image and
         /// offset when the image exports no symbol there. A return address
         /// points after its call, so it is looked up one byte back.
         fn symbol(address: u64, return_address: bool) -> String {
             let address = address.saturating_sub(u64::from(return_address));
-            // SAFETY: Dl_info is plain C data that dladdr fills.
-            let mut info: libc::Dl_info = unsafe { std::mem::zeroed() };
-            // SAFETY: dladdr only reads loader tables for the address.
-            if unsafe { libc::dladdr(address as *const libc::c_void, &mut info) } == 0 {
+            let Some(info) = lookup(address) else {
                 return format!("{address:#x}");
-            }
-            if !info.dli_sname.is_null() {
-                // SAFETY: dladdr returned a NUL-terminated symbol name.
-                let name = unsafe { CStr::from_ptr(info.dli_sname) }.to_string_lossy();
-                // `{:#}` leaves out the hash suffix; other names pass through.
-                return format!("{:#}", rustc_demangle::demangle(&name));
+            };
+            if let Some(name) = symbol_name(&info) {
+                return name;
             }
             let image = if info.dli_fname.is_null() {
                 "?".to_string()
@@ -693,28 +727,65 @@ mod thread_inventory {
             name == "start_wqthread" || name == "_pthread_wqthread"
         }
 
+        fn is_entry_frame(name: &str) -> bool {
+            ENTRY_PREFIXES.iter().any(|prefix| name.starts_with(prefix))
+        }
+
+        /// The functions the registers name that the saved frames cannot:
+        /// the running function (from the program counter), then on arm64
+        /// the function the link register returns into.
+        ///
+        /// The frame pointer chain yields return addresses: once the running
+        /// function has saved its frame record, the chain names its caller
+        /// and everything outside it, but never the running function itself. A thread caught executing its spawn
+        /// closure's own code has no saved frame naming the closure; neither
+        /// has one caught entering or leaving a callee (before the callee
+        /// saves its frame record, or after it restores it), because the
+        /// frame pointer still points at the closure's record, which names
+        /// only the closure's caller. In the second case the closure is where
+        /// the link register returns to.
+        ///
+        /// The program counter always lies in code. The link register can be
+        /// reused as scratch once its function has saved it, so it is used
+        /// only when it falls in a named symbol, and only after the program
+        /// counter.
+        fn running_functions(registers: &Registers) -> Vec<String> {
+            let mut names = vec![symbol(registers.pc, false)];
+            if let Some(lr) = registers.lr.filter(|&lr| lr != 0) {
+                if let Some(name) = lookup(lr - 1).as_ref().and_then(symbol_name) {
+                    names.push(name);
+                }
+            }
+            names
+        }
+
         /// How a thread started, from its stack: the two outermost frames and
-        /// the first frame past the entry and spawn machinery, which for a
+        /// the first function past the entry and spawn machinery, which for a
         /// Rust thread is the closure it was spawned with. Also whether it is
         /// a kernel workqueue thread.
         ///
+        /// That function is searched for in the saved frames from the
+        /// outermost in, then in the functions the registers name (see
+        /// [`running_functions`]). When none qualifies, only the two
+        /// outermost frames are shown.
+        ///
         /// The kernel creates workqueue threads ahead of demand and parks
         /// them before they ever run: such a thread has all-zero registers
         /// and no thread-local storage. `pthread_create` (which `std::thread`,
         /// rayon and tokio use) sets a new thread's registers and storage
         /// before it is scheduled, so no AFT thread looks like that.
         fn describe_start(tls_base: u64, stack: Stack) -> (Option<String>, bool) {
-            let Some((pc, fp)) = stack.registers else {
+            let Some(registers) = stack.registers else {
                 return (Some("<stack not readable>".to_string()), false);
             };
             if stack.returns.is_empty() {
-                if pc == 0 && fp == 0 && tls_base == 0 {
+                if registers.pc == 0 && registers.fp == 0 && tls_base == 0 {
                     return (
                         Some("<kernel workqueue thread that has not run yet>".to_string()),
                         true,
                     );
                 }
-                let name = symbol(pc, false);
+                let name = symbol(registers.pc, false);
                 let os_workqueue = is_workqueue_entry(&name);
                 return (Some(format!("{name} (no saved frames)")), os_workqueue);
             }
@@ -729,16 +800,92 @@ mod thread_inventory {
                 .take(2)
                 .any(|name| is_workqueue_entry(name));
             let mut chain: Vec<&str> = outer_first.iter().take(2).map(String::as_str).collect();
+            let running = running_functions(&registers);
             if let Some(routine) = outer_first
                 .iter()
-                .find(|name| !ENTRY_PREFIXES.iter().any(|prefix| name.starts_with(prefix)))
+                .chain(&running)
+                .find(|name| !is_entry_frame(name))
             {
                 if !chain.contains(&routine.as_str()) {
                     chain.push(routine);
                 }
             }
             (Some(chain.join(" > ")), os_workqueue)
         }
+
+        /// A thread caught where no saved frame names its spawn closure (see
+        /// [`running_functions`]) must still be reported with the closure,
+        /// not with the entry and runtime frames alone. The stack here is
+        /// built by hand so that case is certain: every saved frame is a
+        /// standard-library function, as the frames outside a Rust spawn
+        /// closure are.
+        #[test]
+        fn start_routine_comes_from_the_registers_when_no_saved_frame_names_it() {
+            #[inline(never)]
+            fn spawned_work_marker() {
+                std::hint::black_box(());
+            }
+            let marker = spawned_work_marker as fn() as usize as u64;
+            let runtime_function = std::thread::yield_now as fn() as usize as u64;
+            let outer_runtime_function = std::thread::park as fn() as usize as u64;
+            // Return addresses, innermost first: each points one byte into
+            // its function, as `symbol` expects after looking one byte back.
+            let runtime_returns = vec![runtime_function + 1, outer_runtime_function + 1];
+
+            // Running the closure's own code: only the program counter
+            // names it.
+            let (start, os_workqueue) = describe_start(
+                1,
+                Stack {
+                    registers: Some(Registers {
+                        pc: marker,
+                        fp: 1,
+                        lr: None,
+                    }),
+                    returns: runtime_returns.clone(),
+                },
+            );
+            let start = start.unwrap_or_default();
+            assert!(start.contains("spawned_work_marker"), "{start}");
+            assert!(start.starts_with("std::"), "{start}");
+            assert!(!os_workqueue, "{start}");
+
+            // This synthetic register state models an arm64 thread entering
+            // a runtime callee before the callee saved its frame record: the
+            // program counter is in the callee, and only the link register,
+            // not yet saved, returns into the closure.
+            let (start, _) = describe_start(
+                1,
+                Stack {
+                    registers: Some(Registers {
+                        pc: runtime_function,
+                        fp: 1,
+                        lr: Some(marker + 4),
+                    }),
+                    returns: runtime_returns.clone(),
+                },
+            );
+            let start = start.unwrap_or_default();
+            assert!(start.contains("spawned_work_marker"), "{start}");
+
+            // A saved frame that names the closure wins over the running
+            // function, which is whatever the closure called into last.
+            let callee = libc::getpid as unsafe extern "C" fn() -> libc::pid_t as usize as u64;
+            let (start, _) = describe_start(
+                1,
+                Stack {
+                    registers: Some(Registers {
+                        pc: callee,
+                        fp: 1,
+                        lr: None,
+                    }),
+                    returns: vec![marker + 4, outer_runtime_function + 1],
+                },
+            );
+            let start = start.unwrap_or_default();
+            assert!(start.contains("spawned_work_marker"), "{start}");
+            assert!(!start.contains("getpid"), "{start}");
+        }
     }
 }
 

crates/aft/src/subc/tool_provider.rs

@@ -5696,7 +5696,9 @@ mod route_tests {
 
     #[tokio::test]
     async fn occipital_task_observation_is_limited_to_bound_session() {
-        use crate::bash_background::persistence::{task_paths, write_task, PersistedTask};
+        use crate::bash_background::persistence::{
+            create_task_layout, write_task_at, PersistedTask,
+        };
         use crate::bash_background::BgTaskStatus;
 
         let (dir, root) = test_support::test_root("occipital-tasks");
@@ -5710,7 +5712,13 @@ mod route_tests {
                 "foreign-output-marker",
             ),
         ] {
-            let paths = task_paths(&storage, session, id).unwrap();
+            // Allocate the bundle through the locked production path so the
+            // session directory is non-empty before the detached persisted-task
+            // GC from the first replay can see it. A bare task_paths +
+            // write_task leaves an empty session directory that the sweep can
+            // remove between creation and the first file write.
+            let task = create_task_layout(&storage, session, id).unwrap();
+            let paths = &task.paths;
             let mut metadata = PersistedTask::starting(
                 id.into(),
                 session.into(),
@@ -5722,7 +5730,7 @@ mod route_tests {
                 false,
             );
             metadata.mark_terminal(BgTaskStatus::Completed, Some(0), None);
-            write_task(&paths.json, &metadata).unwrap();
+            write_task_at(&task, &metadata).unwrap();
             std::fs::write(&paths.stdout, output).unwrap();
             std::fs::write(&paths.stderr, "").unwrap();
             std::fs::write(&paths.exit, "0\n").unwrap();

crates/aft/tests/integration/subc_storm_test.rs

@@ -2432,6 +2432,29 @@ fn send_bind(
     send_bind_epoch(tx, channel, 1, corr, root, session, doc);
 }
 
+/// Publish the storm root's project config atomically so a concurrent rebind
+/// never observes a half-written document. Several storm routes share one
+/// root with independent rebind schedules, so one route's write can land
+/// while another route's bind is reading the same file; a direct
+/// truncate-and-write lets that reader see a torn file and fail the bind
+/// with `invalid_config_document`. A sibling temp file plus rename keeps
+/// every observed document complete.
+fn publish_storm_project_config(root: &Path, doc: &Value) {
+    let project_cfg = root.join(".cortexkit").join("aft.jsonc");
+    std::fs::create_dir_all(project_cfg.parent().expect("project cfg parent")).expect("cfg dir");
+    let mut temporary =
+        tempfile::NamedTempFile::new_in(project_cfg.parent().expect("project cfg parent"))
+            .expect("create project config temp");
+    std::io::Write::write_all(
+        &mut temporary,
+        &serde_json::to_vec(doc).expect("serialize storm config"),
+    )
+    .expect("write project config temp");
+    temporary
+        .persist(&project_cfg)
+        .expect("publish project config");
+}
+
 fn send_bind_epoch(
     tx: &mpsc::UnboundedSender<Frame>,
     channel: u16,
@@ -2441,10 +2464,7 @@ fn send_bind_epoch(
     session: &str,
     doc: Value,
 ) {
-    let project_cfg = root.join(".cortexkit").join("aft.jsonc");
-    std::fs::create_dir_all(project_cfg.parent().expect("project cfg parent")).expect("cfg dir");
-    std::fs::write(&project_cfg, serde_json::to_string(&doc).expect("cfg json"))
-        .expect("write cfg");
+    publish_storm_project_config(root, &doc);
     let request = ModuleControlRequest::RouteBind {
         route_channel: channel,
         epoch,
@@ -2472,15 +2492,7 @@ async fn write_measure_bind(
     root: &Path,
     session: &str,
 ) {
-    let project_cfg = root.join(".cortexkit").join("aft.jsonc");
-    std::fs::create_dir_all(project_cfg.parent().expect("measurement config parent"))
-        .expect("create measurement config directory");
-    std::fs::write(
-        &project_cfg,
-        serde_json::to_string(&storm_project_config(false, false, false, 0))
-            .expect("measurement config json"),
-    )
-    .expect("write measurement config");
+    publish_storm_project_config(root, &storm_project_config(false, false, false, 0));
     let corr = 700_000 + u64::from(channel);
     let request = ModuleControlRequest::RouteBind {
         route_channel: channel,