//! The peer request store: admission is idempotent, and only the status CAS takes a request off the
//! dispatch worklist. The peer gate suite that used to cover this was retired, so the contract is
//! stated here against the store directly.
use anyhow::Result;
use klbr_runtime::{
protocol::SessionId,
store::{PeerAdmitOutcome, PeerRequestRecord, PeerSessionRecord, Store},
};
use serde_json::json;
async fn open_peer(store: &Store, sender: &SessionId, peer: &str) -> Result<()> {
let peer_id: SessionId = peer.parse().map_err(anyhow::Error::msg)?;
store.ensure_session(&peer_id).await?;
store.ensure_session(sender).await?;
store
.record_peer_session(PeerSessionRecord {
peer_id: peer.to_string(),
owner_session_id: sender.to_string(),
init_mode: "independent".to_string(),
profile: None,
workspace: None,
kernel_target: None,
budget: None,
branch_parent_revision: None,
created_ms: 900,
})
.await?;
Ok(())
}
fn request(id: &str, sender: &str, peer: &str, at: i64) -> PeerRequestRecord {
PeerRequestRecord {
request_id: id.to_string(),
sender_session_id: sender.to_string(),
peer_id: peer.to_string(),
text: "review the diff".to_string(),
evidence_refs: vec!["event:41".to_string()],
status: "admitted".to_string(),
admitted_at_ms: at,
completed_at_ms: None,
result: None,
error: None,
}
}
/// A re-sent request is not a second request: the durable record is what the sender gets back.
#[tokio::test]
async fn a_re_admitted_request_is_the_same_request() -> Result<()> {
let temp = tempfile::tempdir()?;
let store = Store::open(temp.path().join("peers.sqlite"))?;
open_peer(
&store,
&"s-main".parse().map_err(anyhow::Error::msg)?,
"s-peer",
)
.await?;
let first = store
.admit_peer_request(request("req-1", "s-main", "s-peer", 1_000))
.await?;
assert!(
matches!(first, PeerAdmitOutcome::Admitted(_)),
"first admission must insert: {first:?}"
);
let again = store
.admit_peer_request(request("req-1", "s-main", "s-peer", 1_000))
.await?;
match again {
PeerAdmitOutcome::AlreadyAdmitted(existing) => {
assert_eq!(existing.request_id, "req-1");
assert_eq!(existing.evidence_refs, ["event:41"]);
assert_eq!(existing.admitted_at_ms, 1_000);
}
other => panic!("a repeat must not insert again, got {other:?}"),
}
// The worklist is the durable one, and it holds the request until it is answered.
let worklist = store.admitted_peer_requests().await?;
assert_eq!(worklist.len(), 1, "exactly the one request is dispatchable");
// Durability: a new handle on the same file sees the same admitted request.
drop(store);
let reopened = Store::open(temp.path().join("peers.sqlite"))?;
assert_eq!(reopened.admitted_peer_requests().await?.len(), 1);
Ok(())
}
/// Answering is a compare-and-set: the first writer wins and the worklist drains.
#[tokio::test]
async fn answering_a_request_removes_it_from_the_dispatch_worklist() -> Result<()> {
let temp = tempfile::tempdir()?;
let store = Store::open(temp.path().join("peers.sqlite"))?;
open_peer(
&store,
&"s-main".parse().map_err(anyhow::Error::msg)?,
"s-peer",
)
.await?;
store
.admit_peer_request(request("req-2", "s-main", "s-peer", 2_000))
.await?;
assert!(
store
.complete_peer_request("req-2", json!({"text": "looks fine"}), 2_100)
.await?
);
assert!(
store.admitted_peer_requests().await?.is_empty(),
"an answered request is not dispatched again"
);
// A second answer (a retrying dispatcher) must not overwrite the first one's result.
assert!(
!store
.complete_peer_request("req-2", json!({"text": "second answer"}), 2_200)
.await?
);
let stored = store.get_peer_request("req-2").await?.expect("admitted");
assert_eq!(stored.result, Some(json!({"text": "looks fine"})));
assert_eq!(stored.status, "completed");
Ok(())
}
/// A request is addressed to a session that exists: the foreign key refuses an invented peer, so
/// dispatch can never be handed a request nobody can answer.
#[tokio::test]
async fn a_request_to_an_unopened_peer_is_refused() -> Result<()> {
let temp = tempfile::tempdir()?;
let store = Store::open(temp.path().join("peers.sqlite"))?;
let sender: SessionId = "s-main".parse().map_err(anyhow::Error::msg)?;
store.ensure_session(&sender).await?;
let refused = store
.admit_peer_request(request("req-3", "s-main", "s-not-opened", 3_000))
.await;
assert!(refused.is_err(), "an unopened peer must not be addressable");
assert!(store.admitted_peer_requests().await?.is_empty());
Ok(())
}