From 33e5ef1a1dfdf0bbb59a0f1f97018a4acaa9682f Mon Sep 17 00:00:00 2001 From: Eric Moore Date: Sun, 25 Oct 2026 17:09:46 -0500 Subject: [PATCH] =?UTF-8?q?contacts:=20a=20stalled=20key=20fetch=20says=20?= =?UTF-8?q?so=20=E2=80=94=20and=20the=20Pull=20is=20re-queued=20until=20th?= =?UTF-8?q?e=20key=20lands?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The five-platform cross-chat stage (2026-10-25): the linux node pasted the Android owner's fedcode v3, got contacts.awaiting_key_body, and the key never arrived. The canonical held the stranger's key:record:v1 row at federation tier and both nodes held each other's trust-root acceptance, but every hop that could explain the stall logs at DEBUG (edge's "recovery round: asked this peer", the responder's "identifier lookup refused: no mutual trust root"), so neither node's log said anything. - The pending-contacts pass now re-queues the Key Pull every pass. Edge's recovery queue removes a name when its Pull is sent and nothing re-adds it if the reply is lost, so one dropped answer stranded the contact until the person pasted again. - After 60 s it WARNs "KEY FETCH STALLED" (again every 5 min) with what this node knows: wait time, asks, whether the runtime exists, edge's recovery queue depth, peers in no-route backoff, and the five places to look next. - COMPLETED now logs waited_secs and asked. The pending clock keeps the first paste, so waited_secs is the person's real wait. - tests/contact_key_pull_e2e.rs drives the production path in process (request_key_body, send_recovery_pull, the responder's subject_holdings, apply) and proves the canonical answers: Summary(1), Diff, Deliver, admitted 1. The fault is upstream of the round, on the asking node. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QriZxPx5aJvHXCGaTZmvKF --- src/contacts_chat.rs | 116 +++++++++-- tests/contact_key_pull_e2e.rs | 363 ++++++++++++++++++++++++++++++++++ 2 files changed, 466 insertions(+), 13 deletions(-) create mode 100644 tests/contact_key_pull_e2e.rs diff --git a/src/contacts_chat.rs b/src/contacts_chat.rs index 15e72ec3..f6b4045b 100644 --- a/src/contacts_chat.rs +++ b/src/contacts_chat.rs @@ -5280,15 +5280,25 @@ pub(crate) mod pending_contacts { pub const PENDING_TTL: Duration = Duration::from_secs(24 * 3600); /// How often the pass looks for arrived key bodies. pub const PASS_EVERY: Duration = Duration::from_secs(10); + /// A key still missing after this long is STALLED, and the pass says so + /// out loud. Two anti-entropy cadences (~30 s each): one for the Pull, one + /// for slack. The cross-chat stage of 2026-10-25 waited minutes behind + /// nothing but DEBUG lines on both nodes. + pub const STALL_AFTER: Duration = Duration::from_secs(60); + /// How often a stalled key is reported again. + pub const STALL_REPORT_EVERY: Duration = Duration::from_secs(300); type Key = (String, String); - type Due = (Key, ChatState, Owner, String, Instant); + type Due = (Key, ChatState, Owner, String, Instant, u32); struct Pending { st: ChatState, owner: Owner, key_id: String, since: Instant, + /// How many times the pass re-queued the Pull. + asked: u32, + last_report: Option, } fn table() -> &'static Mutex> { @@ -5297,21 +5307,26 @@ pub(crate) mod pending_contacts { } /// Hold `pasted` for the owner until `key_id`'s record arrives. Idempotent - /// per (owner, code): a second paste refreshes nothing but the clock. + /// per (owner, code): a second paste changes nothing — the clock keeps the + /// FIRST paste, so a stall report's `waited_secs` is the person's real wait. pub(crate) fn record(st: &ChatState, owner: &Owner, pasted: &str, key_id: &str) { let k = (owner.key_id.clone(), pasted.to_owned()); let fresh = { let mut t = table().lock().expect("pending contacts lock"); let fresh = !t.contains_key(&k); - t.insert( - k, - Pending { - st: st.clone(), - owner: owner.clone(), - key_id: key_id.to_owned(), - since: Instant::now(), - }, - ); + if fresh { + t.insert( + k, + Pending { + st: st.clone(), + owner: owner.clone(), + key_id: key_id.to_owned(), + since: Instant::now(), + asked: 1, + last_report: None, + }, + ); + } fresh }; if fresh { @@ -5335,6 +5350,57 @@ pub(crate) mod pending_contacts { } } + /// Re-queue the Key Pull for `key_id` in edge's recovery queue. `false` + /// when there is no replication runtime or edge refused the name. + async fn request_key_body(st: &ChatState, key_id: &str) -> bool { + let Some(runtime) = crate::compose::held_replication_runtime() else { + return false; + }; + let directory = st.engine.federation_directory(); + let lens = ciris_edge::contact::PersistLens::new(directory.as_ref()) + .with_replication(runtime.bridge()); + ciris_edge::contact::DirectoryLens::request_key_body(&lens, key_id).await + } + + /// The STALLED report: every fact this node holds about why a pasted + /// person's key has not arrived, and where the rest of the evidence is. + fn report_stall(owner: &Owner, key_id: &str, waited: Duration, asked: u32, requeued: bool) { + let runtime = crate::compose::held_replication_runtime(); + let queued = runtime + .as_ref() + .map(|r| r.bridge().tracked_missing_signers()); + let no_route: Vec = runtime + .as_ref() + .map(|r| { + r.no_route_backoff() + .into_iter() + .map(|e| format!("{}({} failed probes)", e.peer_key_id, e.failed_probes)) + .collect() + }) + .unwrap_or_default(); + tracing::warn!( + owner = %owner.key_id, + key_id, + waited_secs = waited.as_secs(), + asked, + requeued, + replication_runtime = runtime.is_some(), + edge_recovery_queue = ?queued, + peers_with_no_route = ?no_route, + "contacts: KEY FETCH STALLED — a pasted person's key has not arrived. This node \ + queues a Key Pull for it every pass; the next Key round toward ANY peer sends it, \ + and the peer answers only if it holds the key AND shares a trust root with this \ + node. Where to look: (1) requeued=false or replication_runtime=false → this \ + node cannot ask at all; (2) peers_with_no_route lists peers the Pull cannot \ + reach; (3) here, `CRPL reply DROPPED … kind=Key` means the answer arrived and \ + was discarded (CIRISEdge#634); (4) on the answering peer, `identifier lookup \ + refused: no mutual trust root` (DEBUG) or `identifier lookup RATE LIMITED` means \ + it declined; (5) on the peer, does it hold the key at all — `federation_keys` \ + row for this key_id. Edge logs the Pull itself at DEBUG only \ + (`recovery round: asked this peer`)" + ); + } + /// One pass: complete, discard or expire every pending contact. pub(crate) async fn pass() { let due: Vec = { @@ -5347,11 +5413,12 @@ pub(crate) mod pending_contacts { p.owner.clone(), p.key_id.clone(), p.since, + p.asked, ) }) .collect() }; - for (k, st, owner, key_id, since) in due { + for (k, st, owner, key_id, since, asked) in due { if since.elapsed() > PENDING_TTL { table().lock().expect("lock").remove(&k); tracing::info!(owner = %owner.key_id, key_id, "contacts: pending contact expired — the key never arrived; the person can paste the code again"); @@ -5365,6 +5432,29 @@ pub(crate) mod pending_contacts { Ok(Some(_)) ); if !held { + // Edge's recovery queue REMOVES a name when its Pull is sent, and + // nothing re-adds it if the reply never lands — so a single lost + // reply strands the contact until the person pastes again. Re-queue + // every pass (idempotent: the queue is keyed by name). + let requeued = request_key_body(&st, &key_id).await; + let report = { + let mut t = table().lock().expect("lock"); + t.get_mut(&k).and_then(|p| { + if requeued { + p.asked += 1; + } + let due = since.elapsed() >= STALL_AFTER + && p.last_report + .is_none_or(|at| at.elapsed() >= STALL_REPORT_EVERY); + if due { + p.last_report = Some(Instant::now()); + } + due.then_some(p.asked) + }) + }; + if let Some(asked) = report { + report_stall(&owner, &key_id, since.elapsed(), asked, requeued); + } continue; } let resp = add_contact_for_owner(st, owner.clone(), k.1.clone()).await; @@ -5374,7 +5464,7 @@ pub(crate) mod pending_contacts { } table().lock().expect("lock").remove(&k); if status.is_success() { - tracing::info!(owner = %owner.key_id, key_id, "contacts: pending contact COMPLETED — the key arrived and matched the code"); + tracing::info!(owner = %owner.key_id, key_id, waited_secs = since.elapsed().as_secs(), asked, "contacts: pending contact COMPLETED — the key arrived and matched the code"); } else { tracing::warn!(owner = %owner.key_id, key_id, status = status.as_u16(), "contacts: pending contact discarded — the add path refused it once the key arrived (see the refusal logged just above)"); } diff --git a/tests/contact_key_pull_e2e.rs b/tests/contact_key_pull_e2e.rs new file mode 100644 index 00000000..2973b74b --- /dev/null +++ b/tests/contact_key_pull_e2e.rs @@ -0,0 +1,363 @@ +//! **A pasted stranger's Key reaches the asker over a real recovery round.** +//! +//! The five-platform cross-chat stage, 2026-10-25: the linux node pasted the +//! Android owner's fedcode v3, got `contacts.awaiting_key_body`, and the Key never +//! arrived. The canonical HELD the stranger's `key:record:v1` row (federation +//! tier), both nodes held each other's `delegates_to(node → humanity-accord)`, and +//! the only peer the linux node had was the canonical — yet no logged hop named +//! the stranger. This drives the exact production path in process: the server's +//! `request_key_body` (edge's `PersistLens::with_replication`), the scheduler's +//! recovery Pull (`send_recovery_pull`), the responder's `subject_holdings`, and +//! the asker's apply. +#![cfg(feature = "test-anchor")] + +use std::sync::Arc; + +use ed25519_dalek::SigningKey; + +use ciris_edge::replication::{ + DirectoryStateAdapter, DriveStep, EnvelopeKind, FederationDirectoryReplicationBridge, + MutableDirectoryStateAdapter, ReplicationCoordinator, ReplicationMessage, SessionRole, +}; +use ciris_keyring::MlDsa65SoftwareSigner; +use ciris_persist::federation::types::{cohort_scope, identity_type}; +use ciris_persist::prelude::{Engine, LocalSigner}; +use ciris_server::attest::{register_key, KeySigner}; + +fn init_tracing() { + use std::sync::Once; + static ONCE: Once = Once::new(); + ONCE.call_once(|| { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "ciris_edge=debug,ciris_persist=warn".into()), + ) + .with_test_writer() + .try_init(); + }); +} + +fn genesis_holders() -> Vec { + use ciris_persist::federation::operational::test_support::Identity; + (0..3) + .map(|i| Identity::new(&format!("test-accord-holder-{i}"))) + .collect() +} + +fn arm_test_trust_root() { + let holders = genesis_holders(); + let eds: Vec = holders + .iter() + .map(|h| h.member().ed25519_public_key_base64) + .collect(); + let pqcs: Vec = holders + .iter() + .filter_map(|h| h.member().mldsa65_public_key_base64) + .collect(); + std::env::set_var("CIRIS_TESTING_MODE", "true"); + std::env::set_var("CIRIS_TEST_TRUST_ROOT", eds.join(",")); + std::env::set_var("CIRIS_TEST_TRUST_ROOT_PQC", pqcs.join(",")); +} + +fn pen(alias: &str, seed: u8) -> (LocalSigner, String) { + let ed = SigningKey::from_bytes(&[seed; 32]); + let key_id = ciris_verify_core::fedcode::derive_key_id(alias, ed.verifying_key().as_bytes()); + let pqc = Arc::new( + MlDsa65SoftwareSigner::from_seed_bytes(&[seed ^ 0x5A; 32], format!("{key_id}-pqc")) + .expect("ML-DSA-65 seed"), + ); + let signer = LocalSigner::from_parts( + SigningKey::from_bytes(&[seed; 32]), + key_id.clone(), + Some(pqc), + Some(format!("{key_id}-pqc")), + ); + (signer, key_id) +} + +struct Node { + engine: Arc, + key_id: String, +} + +/// A node whose engine key self-registers as `node` (what every boot writes). +async fn node(alias: &str, seed: u8) -> Node { + let (engine_pen, _) = pen(alias, seed); + let engine = Arc::new( + Engine::with_signer(Arc::new(engine_pen), "sqlite::memory:") + .await + .expect("engine"), + ); + let key_id = engine.local_derived_key_id().await.expect("derived key id"); + register_key( + &engine, + KeySigner::Engine(&engine), + &key_id, + identity_type::NODE, + serde_json::Value::Null, + ) + .await + .expect("node self-registers"); + Node { engine, key_id } +} + +/// Copy `who`'s own key-plane rows into `on` (what first contact carries). +async fn carry_key(on: &Node, from: &Engine, key_id: &str) { + let dir = from.federation_directory(); + let rec = dir + .lookup_public_key(key_id) + .await + .expect("read key") + .unwrap_or_else(|| panic!("{key_id} held at the source")); + on.engine + .federation_directory() + .put_public_key(ciris_persist::federation::SignedKeyRecord { record: rec }) + .await + .unwrap_or_else(|e| panic!("{} holds {key_id}: {e}", on.key_id)); +} + +/// Each node's OWN `delegates_to(node → root)` at federation, held by both. +async fn accept_shared_root(nodes: &[&Node], root: &str, serve_node_key_id: &str) { + for n in nodes { + ciris_persist::federation::operational::test_support::establish_trust_root_side( + n.engine.federation_directory().as_ref(), + root, + serve_node_key_id, + "infra:serve", + ) + .await + .expect("root-side trust legs"); + let trust_edge = serde_json::json!({ + "dimension": ciris_persist::federation::trust_root::TRUST_ACCEPTS_DIMENSION, + "scope": ["infra:attest", "infra:serve"], + }); + let core = ciris_persist::federation::envelope::EnvelopeCore::from_value(trust_edge) + .expect("trust edge envelope"); + let mut te = ciris_persist::federation::EmitAttestationInput::with_envelope( + ciris_persist::federation::types::attestation_type::DELEGATES_TO, + core, + cohort_scope::FEDERATION, + ); + te.attested_key_id = Some(root.to_string()); + te.subject_key_ids = vec![root.to_string()]; + n.engine + .emit_attestation_self(te) + .await + .expect("own delegates_to"); + } + for n in nodes { + let mine = n + .engine + .federation_directory() + .list_attestations_by(&n.key_id) + .await + .expect("own rows"); + for other in nodes.iter().filter(|o| o.key_id != n.key_id) { + for row in mine.iter().filter(|a| a.attested_key_id == root) { + other + .engine + .federation_directory() + .put_attestation(ciris_persist::federation::SignedAttestation { + attestation: row.clone(), + }) + .await + .expect("peer holds the acceptance"); + } + } + } +} + +fn bridge(n: &Node) -> Arc { + Arc::new( + FederationDirectoryReplicationBridge::new( + n.engine.federation_directory(), + Arc::new(Vec::new), + ) + .with_local_key_id(Some(n.key_id.clone())), + ) +} + +struct Loopback { + to_peer: tokio::sync::mpsc::Sender>, +} + +#[async_trait::async_trait] +impl ciris_edge::transport::Transport for Loopback { + fn id(&self) -> ciris_edge::transport::TransportId { + ciris_edge::transport::TransportId::HTTP + } + async fn send( + &self, + _destination_key_id: &str, + envelope_bytes: &[u8], + ) -> Result + { + let _ = self.to_peer.send(envelope_bytes.to_vec()).await; + Ok(ciris_edge::transport::TransportSendOutcome::Delivered) + } + async fn listen( + &self, + _sink: tokio::sync::mpsc::Sender, + ) -> Result<(), ciris_edge::transport::TransportError> { + std::future::pending::<()>().await; + Ok(()) + } +} + +fn msg_name(m: &ReplicationMessage) -> String { + match m { + ReplicationMessage::Summary(s) => format!("Summary({} refs)", s.refs.len()), + ReplicationMessage::Diff(_) => "Diff".into(), + ReplicationMessage::Deliver(d) => format!("Deliver({:?})", d), + ReplicationMessage::Fetch(f) => format!("Fetch({})", f.want.len()), + ReplicationMessage::Pull(p) => format!("Pull({})", p.subject_key_id), + ReplicationMessage::CursorPull(_) => "CursorPull".into(), + } +} + +async fn send_step(coord: &ReplicationCoordinator, step: &DriveStep) { + let msgs: &[ReplicationMessage] = match step { + DriveStep::SendThenWait(m) | DriveStep::SendThenComplete(m, _) => m, + DriveStep::Complete(_) | DriveStep::Refused => &[], + }; + for m in msgs { + coord.send_message(m).await.expect("loopback send"); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_pasted_strangers_key_arrives_from_the_canonical_over_a_recovery_pull() { + init_tracing(); + arm_test_trust_root(); + let asker = node("ciris-node-bootstrap-asker", 0x31).await; + let canonical = node("ciris-canonical-1-probe", 0x41).await; + + // The stranger: a person's fedID, self-registered at the canonical (as the + // Android owner's key reached it 21:45:43). + let (stranger_pen, stranger) = pen("qa-node-stranger", 0x51); + register_key( + &canonical.engine, + KeySigner::Local(&stranger_pen), + &stranger, + identity_type::USER, + serde_json::Value::Null, + ) + .await + .expect("stranger registered at the canonical"); + + // First contact: each holds the other's Key, and both accept one root. + carry_key(&canonical, &asker.engine, &asker.key_id).await; + carry_key(&asker, &canonical.engine, &canonical.key_id).await; + accept_shared_root(&[&asker, &canonical], "probe-root", &canonical.key_id).await; + + assert!( + asker + .engine + .federation_directory() + .lookup_public_key(&stranger) + .await + .unwrap() + .is_none(), + "precondition: the asker does not hold the stranger" + ); + + let asker_bridge = bridge(&asker); + let canon_bridge = bridge(&canonical); + + // What the canonical would serve to this requester, asked directly. + let direct = ciris_edge::replication::StateProvider::subject_refs( + &DirectoryStateAdapter::new(canon_bridge.clone()).with_peer(asker.key_id.clone()), + EnvelopeKind::Key, + &stranger, + ) + .await; + eprintln!( + "canonical subject_refs(Key, stranger) for the asker: {} ref(s)", + direct.len() + ); + + // The server's contacts path, verbatim. + let asker_dir = asker.engine.federation_directory(); + let lens = ciris_edge::contact::PersistLens::new(asker_dir.as_ref()) + .with_replication(asker_bridge.clone()); + assert!( + ciris_edge::contact::DirectoryLens::request_key_body(&lens, &stranger).await, + "the asker queues the Key Pull" + ); + + let (a_tx, mut a_rx) = tokio::sync::mpsc::channel::>(64); + let (c_tx, mut c_rx) = tokio::sync::mpsc::channel::>(64); + let asker_coord = ReplicationCoordinator::new( + Arc::new(Loopback { to_peer: c_tx }), + canonical.key_id.clone(), + EnvelopeKind::Key, + SessionRole::Initiator, + Arc::new( + DirectoryStateAdapter::new(asker_bridge.clone()).with_peer(canonical.key_id.clone()), + ), + Arc::new(MutableDirectoryStateAdapter::new(asker_bridge.clone())), + ); + let canon_coord = ReplicationCoordinator::new( + Arc::new(Loopback { to_peer: a_tx }), + asker.key_id.clone(), + EnvelopeKind::Key, + SessionRole::Responder, + Arc::new(DirectoryStateAdapter::new(canon_bridge.clone()).with_peer(asker.key_id.clone())), + Arc::new(MutableDirectoryStateAdapter::new(canon_bridge)), + ); + + // The scheduler's recovery round (run_one_recovery_round), by hand. + asker_coord.begin_round().await; + let pulled = asker_coord + .send_recovery_pull() + .await + .expect("recovery pull sends"); + assert_eq!( + pulled.as_deref(), + Some(stranger.as_str()), + "the queued name is Pulled" + ); + + for _ in 0..24 { + let mut progressed = false; + while let Ok(bytes) = c_rx.try_recv() { + let msg = + ReplicationCoordinator::parse_inbound_bytes(&bytes).expect("parse at canonical"); + eprintln!(" canonical <- {}", msg_name(&msg)); + let s = canon_coord + .drive_round_step(Some(msg)) + .await + .expect("responder step"); + eprintln!(" canonical => {s:?}"); + send_step(&canon_coord, &s).await; + progressed = true; + } + while let Ok(bytes) = a_rx.try_recv() { + let msg = ReplicationCoordinator::parse_inbound_bytes(&bytes).expect("parse at asker"); + eprintln!(" asker <- {}", msg_name(&msg)); + let s = asker_coord + .drive_round_step(Some(msg)) + .await + .expect("initiator step"); + eprintln!(" asker => {s:?}"); + send_step(&asker_coord, &s).await; + progressed = true; + } + if !progressed { + break; + } + } + + let held = asker + .engine + .federation_directory() + .lookup_public_key(&stranger) + .await + .unwrap(); + assert!( + held.is_some(), + "the stranger's Key reached the asker (canonical offered {} ref(s) directly)", + direct.len() + ); +}