use std::collections::BTreeMap;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use telltale_types::{LocalTypeR, ValType};
use crate::buffer::{BoundedBuffer, BufferConfig, SignedBuffer, SignedValue};
use crate::coroutine::Value;
use crate::instr::Endpoint;
use crate::verification::{
signValue, signing_key_for_endpoint, verifySignedValue, verifying_key_for_endpoint, AuthTree,
DefaultVerificationModel, Hash, HashTag, Signature, VerificationModel,
};
pub type SessionId = usize;
pub type HandlerId = String;
pub const DEFAULT_HANDLER_ID: &str = "default_handler";
fn default_handler_id() -> HandlerId {
DEFAULT_HANDLER_ID.to_string()
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
pub struct Edge {
pub sid: SessionId,
pub sender: String,
pub receiver: String,
}
impl Edge {
#[must_use]
pub fn new(sid: SessionId, sender: impl Into<String>, receiver: impl Into<String>) -> Self {
Self {
sid,
sender: sender.into(),
receiver: receiver.into(),
}
}
}
#[derive(Debug, Deserialize)]
struct EdgeJson {
sid: Option<SessionId>,
sender: String,
receiver: String,
}
pub fn decode_edge_json(
value: &JsonValue,
session_hint: Option<SessionId>,
) -> Result<Edge, String> {
let raw: EdgeJson =
serde_json::from_value(value.clone()).map_err(|e| format!("invalid edge json: {e}"))?;
let sid = raw
.sid
.or(session_hint)
.ok_or_else(|| "missing sid in edge json".to_string())?;
Ok(Edge::new(sid, raw.sender, raw.receiver))
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum SessionStatus {
Active,
Draining,
Closed,
Cancelled,
Faulted {
reason: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TypeEntry {
pub current: LocalTypeR,
pub original: LocalTypeR,
}
#[derive(Debug, Serialize, Deserialize)]
pub struct SessionState {
pub sid: SessionId,
pub roles: Vec<String>,
pub local_types: BTreeMap<Endpoint, TypeEntry>,
pub buffers: BTreeMap<Edge, SignedBuffer<Signature>>,
pub auth_leaves: BTreeMap<Edge, Vec<Hash>>,
#[serde(default)]
pub auth_trees: BTreeMap<Edge, AuthTree>,
pub auth_roots: BTreeMap<Edge, Hash>,
pub edge_handlers: BTreeMap<Edge, HandlerId>,
#[serde(default = "default_handler_id")]
pub default_handler: HandlerId,
pub edge_traces: BTreeMap<Edge, Vec<ValType>>,
pub status: SessionStatus,
pub epoch: usize,
}
impl SessionState {
fn update_auth_tree(&mut self, edge: &Edge, signed: &SignedValue<Signature>) {
let bytes = serde_json::to_vec(signed).unwrap_or_default();
let leaf = DefaultVerificationModel::hash(HashTag::MerkleLeaf, &bytes);
self.auth_leaves.entry(edge.clone()).or_default().push(leaf);
let tree = self
.auth_trees
.entry(edge.clone())
.or_insert_with(|| AuthTree::new(Vec::new()));
tree.append_leaf(leaf);
self.auth_roots.insert(edge.clone(), tree.root());
}
pub fn send_signed(
&mut self,
from: &str,
to: &str,
signed: &SignedValue<Signature>,
) -> Result<crate::buffer::EnqueueResult, String> {
let edge = Edge::new(self.sid, from, to);
let buf = self
.buffers
.get_mut(&edge)
.ok_or_else(|| format!("no buffer for edge {from} → {to}"))?;
let result = buf.enqueue(signed.clone());
if matches!(result, crate::buffer::EnqueueResult::Ok) {
self.update_auth_tree(&edge, signed);
}
Ok(result)
}
pub fn send(
&mut self,
from: &str,
to: &str,
val: Value,
) -> Result<crate::buffer::EnqueueResult, String> {
let signer = signing_key_for_endpoint(&Endpoint {
sid: self.sid,
role: from.to_string(),
});
let signature = signValue(&val, &signer);
self.send_signed(
from,
to,
&SignedValue {
payload: val,
signature,
},
)
}
pub fn recv_signed(&mut self, from: &str, to: &str) -> Option<SignedValue<Signature>> {
let edge = Edge::new(self.sid, from, to);
self.buffers.get_mut(&edge).and_then(|buf| buf.dequeue())
}
pub fn recv_verified(&mut self, from: &str, to: &str) -> Result<Option<Value>, String> {
let sender = Endpoint {
sid: self.sid,
role: from.to_string(),
};
let verifying = verifying_key_for_endpoint(&sender);
let signed = self.recv_signed(from, to);
let Some(signed) = signed else {
return Ok(None);
};
if !verifySignedValue(&signed.payload, &signed.signature, &verifying) {
return Err(format!(
"signature verification failed on edge {from} -> {to}"
));
}
Ok(Some(signed.payload))
}
pub fn recv(&mut self, from: &str, to: &str) -> Option<Value> {
self.recv_verified(from, to).ok().flatten()
}
#[must_use]
pub fn has_message(&self, from: &str, to: &str) -> bool {
let edge = Edge::new(self.sid, from, to);
self.buffers.get(&edge).is_some_and(|buf| !buf.is_empty())
}
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct SessionStore {
sessions: BTreeMap<SessionId, SessionState>,
next_id: SessionId,
}
impl SessionStore {
#[must_use]
pub fn new() -> Self {
Self::default()
}
pub fn open_with_sid(
&mut self,
sid: SessionId,
roles: Vec<String>,
buffer_config: &BufferConfig,
initial_types: &BTreeMap<String, LocalTypeR>,
) -> SessionId {
let mut local_types = BTreeMap::new();
for role in &roles {
if let Some(lt) = initial_types.get(role) {
let ep = Endpoint {
sid,
role: role.clone(),
};
local_types.insert(
ep,
TypeEntry {
current: unfold_mu(lt),
original: lt.clone(),
},
);
}
}
let mut buffers = BTreeMap::new();
for from in &roles {
for to in &roles {
if from != to {
let edge = Edge::new(sid, from.clone(), to.clone());
buffers.insert(edge, BoundedBuffer::new(buffer_config));
}
}
}
let state = SessionState {
sid,
roles,
local_types,
buffers,
auth_leaves: BTreeMap::new(),
auth_trees: BTreeMap::new(),
auth_roots: BTreeMap::new(),
edge_handlers: BTreeMap::new(),
default_handler: default_handler_id(),
edge_traces: BTreeMap::new(),
status: SessionStatus::Active,
epoch: 0,
};
self.sessions.insert(sid, state);
self.next_id = self.next_id.max(sid.saturating_add(1));
sid
}
pub fn open(
&mut self,
roles: Vec<String>,
buffer_config: &BufferConfig,
initial_types: &BTreeMap<String, LocalTypeR>,
) -> SessionId {
let sid = self.next_id;
self.open_with_sid(sid, roles, buffer_config, initial_types)
}
#[must_use]
pub fn next_session_id(&self) -> SessionId {
self.next_id
}
#[must_use]
pub fn lookup_type(&self, ep: &Endpoint) -> Option<&LocalTypeR> {
self.sessions
.get(&ep.sid)?
.local_types
.get(ep)
.map(|e| &e.current)
}
pub fn update_type(&mut self, ep: &Endpoint, new_type: LocalTypeR) {
if let Some(session) = self.sessions.get_mut(&ep.sid) {
if let Some(entry) = session.local_types.get_mut(ep) {
entry.current = new_type;
}
}
}
pub fn update_original(&mut self, ep: &Endpoint, new_original: LocalTypeR) {
if let Some(session) = self.sessions.get_mut(&ep.sid) {
if let Some(entry) = session.local_types.get_mut(ep) {
entry.original = new_original;
}
}
}
#[must_use]
pub fn original_type(&self, ep: &Endpoint) -> Option<&LocalTypeR> {
self.sessions
.get(&ep.sid)?
.local_types
.get(ep)
.map(|e| &e.original)
}
pub fn remove_type(&mut self, ep: &Endpoint) {
if let Some(session) = self.sessions.get_mut(&ep.sid) {
session.local_types.remove(ep);
}
}
#[must_use]
pub fn get(&self, sid: SessionId) -> Option<&SessionState> {
self.sessions.get(&sid)
}
pub fn get_mut(&mut self, sid: SessionId) -> Option<&mut SessionState> {
self.sessions.get_mut(&sid)
}
pub fn iter(&self) -> impl Iterator<Item = &SessionState> {
self.sessions.values()
}
pub fn close(&mut self, sid: SessionId) -> Result<(), String> {
let session = self
.sessions
.get_mut(&sid)
.ok_or_else(|| format!("session {sid} not found"))?;
session.status = SessionStatus::Closed;
session.buffers.clear();
session.edge_traces.clear();
session.epoch = session.epoch.saturating_add(1);
Ok(())
}
#[must_use]
pub fn active_count(&self) -> usize {
self.sessions
.values()
.filter(|s| s.status == SessionStatus::Active)
.count()
}
#[must_use]
pub fn session_ids(&self) -> Vec<SessionId> {
self.sessions.keys().copied().collect()
}
#[must_use]
pub fn lookup_handler(&self, edge: &Edge) -> Option<&HandlerId> {
self.sessions.get(&edge.sid)?.edge_handlers.get(edge)
}
#[must_use]
pub fn default_handler_for_session(&self, sid: SessionId) -> Option<&HandlerId> {
Some(&self.sessions.get(&sid)?.default_handler)
}
pub fn set_default_handler_for_session(&mut self, sid: SessionId, handler: HandlerId) {
if let Some(session) = self.sessions.get_mut(&sid) {
session.default_handler = handler;
}
}
pub fn update_handler(&mut self, edge: &Edge, handler: HandlerId) {
if let Some(session) = self.sessions.get_mut(&edge.sid) {
session.edge_handlers.insert(edge.clone(), handler);
}
}
#[must_use]
pub fn lookup_trace(&self, edge: &Edge) -> Option<&[ValType]> {
self.sessions
.get(&edge.sid)?
.edge_traces
.get(edge)
.map(Vec::as_slice)
}
pub fn update_trace(&mut self, edge: &Edge, trace: Vec<ValType>) {
if let Some(session) = self.sessions.get_mut(&edge.sid) {
session.edge_traces.insert(edge.clone(), trace);
}
}
}
#[must_use]
pub fn unfold_mu(lt: &LocalTypeR) -> LocalTypeR {
match lt {
LocalTypeR::Mu { body, .. } => unfold_mu(body),
other => other.clone(),
}
}
#[must_use]
pub fn unfold_if_var(cont: &LocalTypeR, original: &LocalTypeR) -> LocalTypeR {
match cont {
LocalTypeR::Var(_) => unfold_mu(original),
LocalTypeR::Mu { .. } => unfold_mu(cont),
other => other.clone(),
}
}
#[must_use]
pub fn unfold_if_var_with_scope(
cont: &LocalTypeR,
original: &LocalTypeR,
) -> (LocalTypeR, Option<LocalTypeR>) {
match cont {
LocalTypeR::Var(_) => (unfold_mu(original), None),
LocalTypeR::Mu { .. } => (unfold_mu(cont), Some(cont.clone())),
other => (other.clone(), None),
}
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
use telltale_types::Label;
fn default_types() -> BTreeMap<String, LocalTypeR> {
let mut m = BTreeMap::new();
m.insert(
"A".to_string(),
LocalTypeR::mu(
"step",
LocalTypeR::Send {
partner: "B".into(),
branches: vec![(Label::new("msg"), None, LocalTypeR::var("step"))],
},
),
);
m.insert(
"B".to_string(),
LocalTypeR::mu(
"step",
LocalTypeR::Recv {
partner: "A".into(),
branches: vec![(Label::new("msg"), None, LocalTypeR::var("step"))],
},
),
);
m
}
#[test]
fn test_session_open_with_types() {
let mut store = SessionStore::new();
let types = default_types();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&types,
);
let ep_a = Endpoint {
sid,
role: "A".into(),
};
let ep_b = Endpoint {
sid,
role: "B".into(),
};
assert!(matches!(
store.lookup_type(&ep_a),
Some(LocalTypeR::Send { .. })
));
assert!(matches!(
store.lookup_type(&ep_b),
Some(LocalTypeR::Recv { .. })
));
}
#[test]
fn test_type_advance_and_unfold() {
let mut store = SessionStore::new();
let types = default_types();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&types,
);
let ep_a = Endpoint {
sid,
role: "A".into(),
};
let lt = store.lookup_type(&ep_a).unwrap().clone();
let (_, _vt, continuation) = match < {
LocalTypeR::Send { branches, .. } => branches.first().unwrap().clone(),
_ => panic!("expected Send"),
};
let original = store.original_type(&ep_a).unwrap();
let resolved = unfold_if_var(&continuation, original);
assert!(matches!(resolved, LocalTypeR::Send { .. }));
store.update_type(&ep_a, resolved);
assert!(matches!(
store.lookup_type(&ep_a),
Some(LocalTypeR::Send { .. })
));
}
#[test]
fn test_session_send_recv() {
let mut store = SessionStore::new();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let session = store.get_mut(sid).unwrap();
session.send("A", "B", Value::Nat(42)).unwrap();
assert!(session.has_message("A", "B"));
assert!(!session.has_message("B", "A"));
let val = session.recv("A", "B");
assert_eq!(val, Some(Value::Nat(42)));
}
#[test]
fn test_close_clears_buffers_and_traces_even_when_messages_pending() {
let mut store = SessionStore::new();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let edge = Edge::new(sid, "A", "B");
store
.get_mut(sid)
.expect("session exists")
.send("A", "B", Value::Nat(7))
.expect("enqueue pending message");
store.update_trace(&edge, vec![ValType::Nat]);
store.close(sid).expect("close session");
let session = store.get(sid).expect("session exists after close");
assert_eq!(session.status, SessionStatus::Closed);
assert!(session.buffers.is_empty());
assert!(session.edge_traces.is_empty());
}
#[test]
fn test_namespace_isolation() {
let mut store = SessionStore::new();
let sid1 = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let sid2 = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
assert_ne!(sid1, sid2);
store
.get_mut(sid1)
.unwrap()
.send("A", "B", Value::Nat(1))
.unwrap();
assert!(!store.get(sid2).unwrap().has_message("A", "B"));
}
#[test]
fn test_remove_type() {
let mut store = SessionStore::new();
let types = default_types();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&types,
);
let ep_a = Endpoint {
sid,
role: "A".into(),
};
assert!(store.lookup_type(&ep_a).is_some());
store.remove_type(&ep_a);
assert!(store.lookup_type(&ep_a).is_none());
}
#[test]
fn test_cross_session_role_name_edge_collision_regression() {
let mut store = SessionStore::new();
let sid1 = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let sid2 = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let e1 = Edge::new(sid1, "A", "B");
let e2 = Edge::new(sid2, "A", "B");
assert_ne!(e1, e2, "edges from distinct sessions must not collide");
assert!(store
.get(sid1)
.expect("sid1 exists")
.buffers
.contains_key(&e1));
assert!(store
.get(sid2)
.expect("sid2 exists")
.buffers
.contains_key(&e2));
}
#[test]
fn test_edge_handler_and_trace_bindings() {
let mut store = SessionStore::new();
let sid = store.open(
vec!["A".into(), "B".into()],
&BufferConfig::default(),
&BTreeMap::new(),
);
let edge = Edge::new(sid, "A", "B");
assert!(store.lookup_handler(&edge).is_none());
store.update_handler(&edge, "handler/send".to_string());
assert_eq!(
store.lookup_handler(&edge).map(String::as_str),
Some("handler/send")
);
assert!(store.lookup_trace(&edge).is_none());
store.update_trace(&edge, vec![ValType::Nat]);
assert_eq!(store.lookup_trace(&edge), Some([ValType::Nat].as_slice()));
}
#[test]
fn test_decode_edge_json_requires_sid_sender_receiver() {
let sid_qualified = json!({
"sid": 7,
"sender": "A",
"receiver": "B"
});
let e = decode_edge_json(&sid_qualified, None).expect("decode sid-qualified edge");
assert_eq!(e, Edge::new(7, "A", "B"));
let no_sid = json!({
"sender": "A",
"receiver": "B"
});
let e2 = decode_edge_json(&no_sid, Some(11)).expect("decode edge with sid hint");
assert_eq!(e2, Edge::new(11, "A", "B"));
let legacy = json!({
"from": "A",
"to": "B",
"sid": 11
});
let err = decode_edge_json(&legacy, None).expect_err("legacy edge shape must be rejected");
assert!(err.contains("invalid edge json"), "unexpected error: {err}");
}
}