KU-K1 review: red — record_last_failure_transport erases the reachability slot

Opus's review: freemkv-library's server calls take_online_decode_reachability()
after resolve (server/ripper/mod.rs:505, :1361) to classify a no-key without a
second probe. record_last_failure_transport currently TAKES the thread-local
slot, so the server sees None and fires probe_online_reachability — a second
key-service call, and a definite 422 misread as an outage. Red: the slot is
gone after get_unit_keys/get_fmts_indexes. Also adds the forensic-path,
mid-body-stall, transport-then-5xx-reset and MultiSource forwarding tests
requested in review; the latter fails since MultiSource has no override yet.

src/lib.rs

@@ -250,6 +250,35 @@ mod tests {
         assert!(keydb_then_online.answer_depends_on_samples());
     }
 
+    // A source whose `last_failure_was_transport` is fixed to the given value.
+    struct FixedTransport(bool);
+    impl KeySource for FixedTransport {
+        fn get_unit_keys(&self, _ctx: &dyn ResolveCtx) -> Result<Vec<UnitKey>, libfreemkv::Error> {
+            Ok(Vec::new())
+        }
+        fn last_failure_was_transport(&self) -> bool {
+            self.0
+        }
+    }
+
+    // For consistency with `answer_depends_on_samples` (KU3-6): a composed
+    // source's last-failure verdict is `any(inner)` too, so `resolve` retries
+    // the composition whenever ANY inner source's last failure was transport.
+    #[test]
+    fn multi_source_last_failure_was_transport_is_any_inner() {
+        let none_transport = MultiSource::new(vec![
+            Box::new(FixedTransport(false)) as Box<dyn KeySource>,
+            Box::new(FixedTransport(false)) as Box<dyn KeySource>,
+        ]);
+        assert!(!none_transport.last_failure_was_transport());
+
+        let one_transport = MultiSource::new(vec![
+            Box::new(FixedTransport(false)) as Box<dyn KeySource>,
+            Box::new(FixedTransport(true)) as Box<dyn KeySource>,
+        ]);
+        assert!(one_transport.last_failure_was_transport());
+    }
+
     /// The `mkb` generation must reach each inner source so its OWN revocation
     /// filter can act; a source that drops everything at a given generation is
     /// honoured by the union.

src/online.rs

@@ -2943,4 +2943,76 @@ mod tests {
             "a success must reset the flag"
         );
     }
+
+    // Regression: freemkv-library's server takes the reachability slot itself
+    // (`take_online_decode_reachability`) after `resolve`, to classify a no-key
+    // without a second probe. `last_failure_was_transport` must PEEK, not TAKE,
+    // or the server sees `None` and re-asks the service — never call it twice.
+    #[test]
+    fn recording_last_failure_does_not_erase_the_reachability_slot() {
+        let addr = stub_server(Stub::Answer(422));
+        let src = source_via("kuk1f.test", T_IDLE, move || Ok(vec![addr]));
+        assert!(src.get_unit_keys(&ctx_with_mkb(0)).is_err());
+        assert_eq!(
+            take_last_decode_reachability(),
+            Some(DecodeReachability::Status(422)),
+            "get_unit_keys must leave the slot for a later caller (e.g. the server) to read"
+        );
+
+        let addr = stub_server(Stub::Answer(422));
+        let src = source_via("kuk1g.test", T_IDLE, move || Ok(vec![addr]));
+        assert!(src.get_fmts_indexes(&ctx_with_mkb(0)).is_err());
+        assert_eq!(
+            take_last_decode_reachability(),
+            Some(DecodeReachability::Status(422)),
+            "get_fmts_indexes must leave the slot too"
+        );
+    }
+
+    // The base and forensic paths share `query_with`, so the flag must update
+    // through `get_fmts_indexes` exactly as it does through `get_unit_keys`.
+    #[test]
+    fn last_failure_was_transport_true_through_get_fmts_indexes() {
+        let addr = stub_server(Stub::NeverAnswer);
+        let src = source_via("kuk1h.test", T_IDLE, move || Ok(vec![addr]));
+        assert!(src.get_fmts_indexes(&ctx_with_mkb(0)).is_err());
+        assert!(
+            src.last_failure_was_transport(),
+            "get_fmts_indexes must classify the failure too"
+        );
+    }
+
+    // Stop-design-v5 T16: a reply that stalls mid-BODY (not just before the
+    // first byte) is still "no bytes moved" past idle — transport-class.
+    #[test]
+    fn last_failure_was_transport_true_after_mid_body_stall() {
+        let addr = stub_server(Stub::StallBody);
+        let src = source_via("kuk1i.test", T_IDLE, move || Ok(vec![addr]));
+        assert!(src.get_unit_keys(&ctx_with_mkb(0)).is_err());
+        assert!(
+            src.last_failure_was_transport(),
+            "a mid-body stall is transport-class"
+        );
+    }
+
+    // J15: a definite answer after a transport failure must clear the flag — the
+    // NEXT verdict always wins, so a stale `true` never survives past one query.
+    #[test]
+    fn last_failure_was_transport_resets_from_true_on_5xx() {
+        let refused: SocketAddr = ([127, 0, 0, 1], 1).into();
+        let answering = stub_server(Stub::Answer(503));
+        let target = Arc::new(Mutex::new(refused));
+        let for_resolve = target.clone();
+        let src = source_via("kuk1j.test", T_IDLE, move || {
+            Ok(vec![*for_resolve.lock().unwrap()])
+        });
+        assert!(src.get_unit_keys(&ctx_with_mkb(0)).is_err());
+        assert!(src.last_failure_was_transport());
+        *target.lock().unwrap() = answering;
+        assert!(src.get_unit_keys(&ctx_with_mkb(0)).is_err());
+        assert!(
+            !src.last_failure_was_transport(),
+            "a 5xx answer must clear a prior transport verdict"
+        );
+    }
 }