1use std::collections::{BTreeMap, BTreeSet};
8
9use serde::{Deserialize, Serialize};
10use serde_json::Value as JsonValue;
11use telltale_types::{Label, LocalTypeR, ValType};
12
13use crate::buffer::{BoundedBuffer, BufferConfig, SignedBuffer, SignedValue};
14use crate::coroutine::Value;
15use crate::instr::Endpoint;
16use crate::verification::{
17 signValue, signing_key_for_endpoint, verifySignedValue, verifying_key_for_endpoint, AuthTree,
18 DefaultVerificationModel, Hash, HashTag, Signature, VerificationModel,
19};
20
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
23pub struct ClosedSessionSummary {
24 pub sid: SessionId,
26 pub status: SessionStatus,
28 pub role_count: usize,
30 pub local_type_entries: usize,
32 pub edge_count: usize,
34 pub edge_handler_count: usize,
36 pub auth_leaf_count: usize,
38 pub auth_tree_count: usize,
40 pub auth_root_count: usize,
42 pub epoch: usize,
44}
45
46impl ClosedSessionSummary {
47 fn from_session(session: &SessionState) -> Self {
48 Self {
49 sid: session.sid,
50 status: session.status.clone(),
51 role_count: session.roles.len(),
52 local_type_entries: session.local_types.len(),
53 edge_count: session.buffers.len(),
54 edge_handler_count: session.edge_handlers.len(),
55 auth_leaf_count: session.auth_leaves.values().map(Vec::len).sum(),
56 auth_tree_count: session.auth_trees.len(),
57 auth_root_count: session.auth_roots.len(),
58 epoch: session.epoch,
59 }
60 }
61
62 fn retained_bytes_estimate(&self) -> usize {
63 std::mem::size_of::<Self>().saturating_add(serialized_bytes(self))
64 }
65}
66
67#[derive(Debug, Clone)]
69pub struct SessionOpenPlan {
70 pub(crate) roles: Vec<String>,
71 pub(crate) role_ids: BTreeMap<String, u16>,
72 pub(crate) initial_types: Vec<(String, LocalTypeR, LocalTypeR)>,
73 pub(crate) edge_blueprint: Vec<((u16, u16), String, String)>,
74 pub(crate) active_branch_roles: Vec<String>,
75}
76
77impl SessionOpenPlan {
78 fn collect_protocol_edges(
79 role: &str,
80 local_type: &LocalTypeR,
81 role_ids: &BTreeMap<String, u16>,
82 edges: &mut BTreeSet<(u16, u16)>,
83 ) {
84 match local_type {
85 LocalTypeR::End | LocalTypeR::Var(_) => {}
86 LocalTypeR::Mu { body, .. } => {
87 Self::collect_protocol_edges(role, body, role_ids, edges);
88 }
89 LocalTypeR::Send { partner, branches } => {
90 if let (Some(from_id), Some(to_id)) = (role_ids.get(role), role_ids.get(partner)) {
91 if from_id != to_id {
92 edges.insert((*from_id, *to_id));
93 }
94 }
95 for (_, _, continuation) in branches {
96 Self::collect_protocol_edges(role, continuation, role_ids, edges);
97 }
98 }
99 LocalTypeR::Recv { partner, branches } => {
100 if let (Some(from_id), Some(to_id)) = (role_ids.get(partner), role_ids.get(role)) {
101 if from_id != to_id {
102 edges.insert((*from_id, *to_id));
103 }
104 }
105 for (_, _, continuation) in branches {
106 Self::collect_protocol_edges(role, continuation, role_ids, edges);
107 }
108 }
109 }
110 }
111
112 #[must_use]
119 pub fn new(roles: &[String], initial_types: &BTreeMap<String, LocalTypeR>) -> Self {
120 let role_ids = SessionState::build_role_ids(roles);
121 let mut planned_types = Vec::with_capacity(roles.len());
122 let mut active_branch_roles = Vec::new();
123 for role in roles {
124 if let Some(original) = initial_types.get(role) {
125 let current = unfold_mu(original);
126 if SessionState::branch_shape(¤t).is_some() {
127 active_branch_roles.push(role.clone());
128 }
129 planned_types.push((role.clone(), current, original.clone()));
130 }
131 }
132
133 let mut protocol_edges = BTreeSet::new();
134 for role in roles {
135 if let Some(original) = initial_types.get(role) {
136 Self::collect_protocol_edges(role, original, &role_ids, &mut protocol_edges);
137 }
138 }
139 let mut edge_blueprint = Vec::with_capacity(protocol_edges.len());
140 for (from_id, to_id) in protocol_edges {
141 let from = roles
142 .get(usize::from(from_id))
143 .expect("sender role id must index the session-open role set")
144 .clone();
145 let to = roles
146 .get(usize::from(to_id))
147 .expect("receiver role id must index the session-open role set")
148 .clone();
149 edge_blueprint.push(((from_id, to_id), from, to));
150 }
151
152 Self {
153 roles: roles.to_vec(),
154 role_ids,
155 initial_types: planned_types,
156 edge_blueprint,
157 active_branch_roles,
158 }
159 }
160
161 #[must_use]
163 pub fn roles(&self) -> &[String] {
164 &self.roles
165 }
166
167 #[must_use]
169 pub fn edge_blueprint(&self) -> &[((u16, u16), String, String)] {
170 &self.edge_blueprint
171 }
172}
173
174#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
176pub struct SessionStoreMemoryUsage {
177 pub live_sessions: usize,
179 pub live_closed_sessions: usize,
181 pub archived_closed_sessions: usize,
183 pub live_local_type_entries: usize,
185 pub live_buffer_count: usize,
187 pub live_buffered_messages: usize,
189 pub live_edge_handler_count: usize,
191 pub live_auth_leaf_count: usize,
193 pub live_auth_tree_count: usize,
195 pub live_auth_root_count: usize,
197 pub retained_bytes: SessionStoreRetainedBytes,
199}
200
201#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
203pub struct SessionStoreRetainedBytes {
204 pub live_sessions: usize,
206 pub archived_closed: usize,
208 pub local_types: usize,
210 pub buffers: usize,
212 pub traces: usize,
214 pub auth: usize,
216 pub handlers: usize,
218 pub total: usize,
220}
221
222pub type SessionId = usize;
224
225pub type HandlerId = String;
227type HandlerNumericId = u16;
228type LabelNumericId = u16;
229type EdgeKey = (u16, u16);
230type LocalBranches<'a> = &'a [(Label, Option<ValType>, LocalTypeR)];
231type HandlerIndexBuild = (
232 BTreeMap<HandlerId, HandlerNumericId>,
233 Vec<HandlerId>,
234 BTreeMap<EdgeKey, HandlerNumericId>,
235 Option<HandlerNumericId>,
236);
237
238#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
239pub(crate) enum BranchDirection {
240 Send,
241 Recv,
242}
243
244#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
245pub(crate) struct CachedBranch {
246 pub(crate) direction: BranchDirection,
247 pub(crate) partner: String,
248 pub(crate) expected_type: Option<ValType>,
249 pub(crate) continuation: LocalTypeR,
250}
251
252pub const DEFAULT_HANDLER_ID: &str = "default_handler";
254
255fn default_handler_id() -> HandlerId {
256 DEFAULT_HANDLER_ID.to_string()
257}
258
259fn serialized_bytes<T: Serialize>(value: &T) -> usize {
260 bincode::serialized_size(value)
261 .ok()
262 .and_then(|bytes| usize::try_from(bytes).ok())
263 .unwrap_or(0)
264}
265
266#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
268pub struct Edge {
269 pub sid: SessionId,
271 pub sender: String,
273 pub receiver: String,
275}
276
277impl Edge {
278 #[must_use]
280 pub fn new(sid: SessionId, sender: impl Into<String>, receiver: impl Into<String>) -> Self {
281 Self {
282 sid,
283 sender: sender.into(),
284 receiver: receiver.into(),
285 }
286 }
287}
288
289#[derive(Debug, Deserialize)]
290struct EdgeJson {
291 sid: Option<SessionId>,
292 sender: String,
293 receiver: String,
294}
295
296pub fn decode_edge_json(
302 value: &JsonValue,
303 session_hint: Option<SessionId>,
304) -> Result<Edge, String> {
305 let raw: EdgeJson =
306 serde_json::from_value(value.clone()).map_err(|e| format!("invalid edge json: {e}"))?;
307
308 let sid = raw
309 .sid
310 .or(session_hint)
311 .ok_or_else(|| "missing sid in edge json".to_string())?;
312 Ok(Edge::new(sid, raw.sender, raw.receiver))
313}
314
315#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
317pub enum SessionStatus {
318 Active,
320 Draining,
322 Closed,
324 Cancelled,
326 Faulted {
328 reason: String,
330 },
331}
332
333#[derive(Debug, Clone, Serialize, Deserialize)]
335pub struct TypeEntry {
336 pub current: LocalTypeR,
338 pub original: LocalTypeR,
340}
341
342#[derive(Debug, Serialize)]
347pub struct SessionState {
348 pub sid: SessionId,
350 pub roles: Vec<String>,
352 #[serde(skip)]
354 role_ids: BTreeMap<String, u16>,
355 pub local_types: BTreeMap<Endpoint, TypeEntry>,
359 pub buffers: BTreeMap<Edge, SignedBuffer<Signature>>,
361 #[serde(skip)]
363 edge_lookup: BTreeMap<(u16, u16), Edge>,
364 #[serde(skip)]
366 handler_ids: BTreeMap<HandlerId, HandlerNumericId>,
367 #[serde(skip)]
369 handlers_by_id: Vec<HandlerId>,
370 #[serde(skip)]
372 edge_handler_lookup: BTreeMap<(u16, u16), HandlerNumericId>,
373 #[serde(skip)]
375 default_handler_id: Option<HandlerNumericId>,
376 #[serde(skip)]
378 label_ids: BTreeMap<String, LabelNumericId>,
379 #[serde(skip)]
381 labels_by_id: Vec<String>,
382 #[serde(skip)]
384 branch_lookup: BTreeMap<Endpoint, BTreeMap<LabelNumericId, CachedBranch>>,
385 pub auth_leaves: BTreeMap<Edge, Vec<Hash>>,
387 #[serde(default)]
389 pub auth_trees: BTreeMap<Edge, AuthTree>,
390 pub auth_roots: BTreeMap<Edge, Hash>,
392 pub edge_handlers: BTreeMap<Edge, HandlerId>,
394 #[serde(default = "default_handler_id")]
396 pub default_handler: HandlerId,
397 pub edge_traces: BTreeMap<Edge, Vec<ValType>>,
399 pub status: SessionStatus,
401 pub epoch: usize,
403}
404
405impl SessionState {
406 pub(crate) fn from_open_plan(
407 sid: SessionId,
408 plan: &SessionOpenPlan,
409 buffer_config: &BufferConfig,
410 ) -> Self {
411 let mut local_type_entries = Vec::with_capacity(plan.initial_types.len());
412 for (role, current, original) in &plan.initial_types {
413 local_type_entries.push((
414 Endpoint {
415 sid,
416 role: role.clone(),
417 },
418 TypeEntry {
419 current: current.clone(),
420 original: original.clone(),
421 },
422 ));
423 }
424 let local_types = local_type_entries.into_iter().collect();
425
426 let mut edge_entries = Vec::with_capacity(plan.edge_blueprint().len());
427 let mut buffer_entries = Vec::with_capacity(plan.edge_blueprint().len());
428 for (key, from, to) in plan.edge_blueprint() {
429 let edge = Edge::new(sid, from.clone(), to.clone());
430 edge_entries.push((*key, edge.clone()));
431 buffer_entries.push((edge, BoundedBuffer::new(buffer_config)));
432 }
433 let edge_lookup = edge_entries.into_iter().collect();
434 let buffers = buffer_entries.into_iter().collect();
435
436 let default_handler = default_handler_id();
437 let (handler_ids, handlers_by_id, edge_handler_lookup, default_handler_id) =
438 Self::build_handler_indexes(&plan.role_ids, &default_handler, &BTreeMap::new());
439
440 let mut state = Self {
441 sid,
442 roles: plan.roles.clone(),
443 role_ids: plan.role_ids.clone(),
444 local_types,
445 buffers,
446 edge_lookup,
447 handler_ids,
448 handlers_by_id,
449 edge_handler_lookup,
450 default_handler_id,
451 label_ids: BTreeMap::new(),
452 labels_by_id: Vec::new(),
453 branch_lookup: BTreeMap::new(),
454 auth_leaves: BTreeMap::new(),
455 auth_trees: BTreeMap::new(),
456 auth_roots: BTreeMap::new(),
457 edge_handlers: BTreeMap::new(),
458 default_handler,
459 edge_traces: BTreeMap::new(),
460 status: SessionStatus::Active,
461 epoch: 0,
462 };
463 for role in &plan.active_branch_roles {
464 state.refresh_endpoint_branch_lookup(&Endpoint {
465 sid,
466 role: role.clone(),
467 });
468 }
469 state
470 }
471
472 fn retained_session_core_bytes(&self) -> usize {
473 std::mem::size_of::<Self>()
474 .saturating_add(serialized_bytes(&self.sid))
475 .saturating_add(serialized_bytes(&self.roles))
476 .saturating_add(serialized_bytes(&self.role_ids))
477 .saturating_add(serialized_bytes(&self.edge_lookup))
478 .saturating_add(serialized_bytes(&self.handler_ids))
479 .saturating_add(serialized_bytes(&self.handlers_by_id))
480 .saturating_add(serialized_bytes(&self.edge_handler_lookup))
481 .saturating_add(serialized_bytes(&self.default_handler_id))
482 .saturating_add(serialized_bytes(&self.label_ids))
483 .saturating_add(serialized_bytes(&self.labels_by_id))
484 .saturating_add(serialized_bytes(&self.branch_lookup))
485 .saturating_add(serialized_bytes(&self.status))
486 .saturating_add(serialized_bytes(&self.epoch))
487 }
488
489 fn retained_local_type_bytes(&self) -> usize {
490 serialized_bytes(&self.local_types)
491 }
492
493 fn retained_buffer_bytes(&self) -> usize {
494 serialized_bytes(&self.buffers)
495 }
496
497 fn retained_trace_bytes(&self) -> usize {
498 serialized_bytes(&self.edge_traces)
499 }
500
501 fn retained_auth_bytes(&self) -> usize {
502 serialized_bytes(&self.auth_leaves)
503 .saturating_add(serialized_bytes(&self.auth_trees))
504 .saturating_add(serialized_bytes(&self.auth_roots))
505 }
506
507 fn retained_handler_bytes(&self) -> usize {
508 serialized_bytes(&self.edge_handlers)
509 .saturating_add(serialized_bytes(&self.default_handler))
510 }
511
512 fn rebuild_derived_indexes(&mut self) {
513 self.role_ids = Self::build_role_ids(&self.roles);
514 self.edge_lookup = Self::build_edge_lookup_from_buffers(&self.role_ids, &self.buffers);
515 self.refresh_handler_indexes();
516 self.label_ids = BTreeMap::new();
517 self.labels_by_id = Vec::new();
518 self.branch_lookup = BTreeMap::new();
519 let endpoints: Vec<Endpoint> = self.local_types.keys().cloned().collect();
520 for endpoint in endpoints {
521 self.refresh_endpoint_branch_lookup(&endpoint);
522 }
523 }
524
525 pub(crate) fn refresh_handler_indexes(&mut self) {
526 let (handler_ids, handlers_by_id, edge_handler_lookup, default_handler_id) =
527 Self::build_handler_indexes(&self.role_ids, &self.default_handler, &self.edge_handlers);
528 self.handler_ids = handler_ids;
529 self.handlers_by_id = handlers_by_id;
530 self.edge_handler_lookup = edge_handler_lookup;
531 self.default_handler_id = default_handler_id;
532 }
533
534 pub(crate) fn build_role_ids(roles: &[String]) -> BTreeMap<String, u16> {
535 roles
536 .iter()
537 .enumerate()
538 .map(|(idx, role)| {
539 (
540 role.clone(),
541 u16::try_from(idx).expect("role count should fit in u16"),
542 )
543 })
544 .collect()
545 }
546
547 pub(crate) fn build_edge_lookup_from_buffers(
548 role_ids: &BTreeMap<String, u16>,
549 buffers: &BTreeMap<Edge, SignedBuffer<Signature>>,
550 ) -> BTreeMap<EdgeKey, Edge> {
551 let mut lookup = BTreeMap::new();
552 for edge in buffers.keys() {
553 let Some(from_id) = role_ids.get(&edge.sender) else {
554 continue;
555 };
556 let Some(to_id) = role_ids.get(&edge.receiver) else {
557 continue;
558 };
559 lookup.insert((*from_id, *to_id), edge.clone());
560 }
561 lookup
562 }
563
564 pub(crate) fn build_handler_indexes(
565 role_ids: &BTreeMap<String, u16>,
566 default_handler: &str,
567 edge_handlers: &BTreeMap<Edge, HandlerId>,
568 ) -> HandlerIndexBuild {
569 let mut handler_ids = BTreeMap::new();
570 let mut handlers_by_id = Vec::new();
571 let intern_handler = |handler: &str,
572 handler_ids: &mut BTreeMap<HandlerId, HandlerNumericId>,
573 handlers_by_id: &mut Vec<HandlerId>|
574 -> HandlerNumericId {
575 if let Some(id) = handler_ids.get(handler) {
576 return *id;
577 }
578 let id = u16::try_from(handlers_by_id.len()).expect("handler count should fit in u16");
579 let owned = handler.to_string();
580 handler_ids.insert(owned.clone(), id);
581 handlers_by_id.push(owned);
582 id
583 };
584
585 let default_handler_id = (!default_handler.is_empty())
586 .then(|| intern_handler(default_handler, &mut handler_ids, &mut handlers_by_id));
587
588 let mut edge_handler_lookup = BTreeMap::new();
589 for (edge, handler) in edge_handlers {
590 let Some(from_id) = role_ids.get(&edge.sender) else {
591 continue;
592 };
593 let Some(to_id) = role_ids.get(&edge.receiver) else {
594 continue;
595 };
596 let handler_id = intern_handler(handler, &mut handler_ids, &mut handlers_by_id);
597 edge_handler_lookup.insert((*from_id, *to_id), handler_id);
598 }
599
600 (
601 handler_ids,
602 handlers_by_id,
603 edge_handler_lookup,
604 default_handler_id,
605 )
606 }
607
608 fn edge_for_roles(&self, from: &str, to: &str) -> Option<&Edge> {
609 let from_id = self.role_ids.get(from)?;
610 let to_id = self.role_ids.get(to)?;
611 self.edge_lookup.get(&(*from_id, *to_id))
612 }
613
614 fn edge_key_for_roles(&self, from: &str, to: &str) -> Option<(u16, u16)> {
615 let from_id = self.role_ids.get(from)?;
616 let to_id = self.role_ids.get(to)?;
617 Some((*from_id, *to_id))
618 }
619
620 fn intern_label(&mut self, label: &str) -> LabelNumericId {
621 if let Some(id) = self.label_ids.get(label) {
622 return *id;
623 }
624 let id = u16::try_from(self.labels_by_id.len()).expect("label count should fit in u16");
625 let owned = label.to_string();
626 self.label_ids.insert(owned.clone(), id);
627 self.labels_by_id.push(owned);
628 id
629 }
630
631 fn intern_handler_binding(&mut self, handler: &str) -> HandlerNumericId {
632 if let Some(id) = self.handler_ids.get(handler) {
633 return *id;
634 }
635 let id = u16::try_from(self.handlers_by_id.len()).expect("handler count should fit in u16");
636 let owned = handler.to_string();
637 self.handler_ids.insert(owned.clone(), id);
638 self.handlers_by_id.push(owned);
639 id
640 }
641
642 fn handler_by_id(&self, handler_id: HandlerNumericId) -> Option<&HandlerId> {
643 self.handlers_by_id.get(usize::from(handler_id))
644 }
645
646 fn branch_shape(local_type: &LocalTypeR) -> Option<(BranchDirection, &str, LocalBranches<'_>)> {
647 match local_type {
648 LocalTypeR::Send { partner, branches } => {
649 Some((BranchDirection::Send, partner.as_str(), branches.as_slice()))
650 }
651 LocalTypeR::Recv { partner, branches } => {
652 Some((BranchDirection::Recv, partner.as_str(), branches.as_slice()))
653 }
654 _ => None,
655 }
656 }
657
658 pub(crate) fn refresh_endpoint_branch_lookup(&mut self, ep: &Endpoint) {
659 self.branch_lookup.remove(ep);
660 let Some(entry) = self.local_types.get(ep) else {
661 return;
662 };
663 let Some((direction, partner, branches)) = Self::branch_shape(&entry.current) else {
664 return;
665 };
666 let partner = partner.to_string();
667 let branches: Vec<(String, Option<ValType>, LocalTypeR)> = branches
668 .iter()
669 .map(|(label, expected_type, continuation)| {
670 (
671 label.name.clone(),
672 expected_type.clone(),
673 continuation.clone(),
674 )
675 })
676 .collect();
677
678 let mut endpoint_lookup = BTreeMap::new();
679 for (label, expected_type, continuation) in branches {
680 let label_id = self.intern_label(&label);
681 endpoint_lookup.insert(
682 label_id,
683 CachedBranch {
684 direction,
685 partner: partner.clone(),
686 expected_type,
687 continuation,
688 },
689 );
690 }
691 if !endpoint_lookup.is_empty() {
692 self.branch_lookup.insert(ep.clone(), endpoint_lookup);
693 }
694 }
695
696 #[must_use]
698 pub(crate) fn lookup_branch_resolution(
699 &self,
700 ep: &Endpoint,
701 label: &str,
702 ) -> Option<&CachedBranch> {
703 let label_id = self.label_ids.get(label)?;
704 self.branch_lookup.get(ep)?.get(label_id)
705 }
706
707 fn update_auth_tree(&mut self, edge: &Edge, signed: &SignedValue<Signature>) {
708 let bytes = bincode::serialize(signed).unwrap_or_default();
709 let leaf = DefaultVerificationModel::hash(HashTag::MerkleLeaf, &bytes);
710 self.auth_leaves.entry(edge.clone()).or_default().push(leaf);
711 let tree = self
712 .auth_trees
713 .entry(edge.clone())
714 .or_insert_with(|| AuthTree::new(Vec::new()));
715 tree.append_leaf(leaf);
716 self.auth_roots.insert(edge.clone(), tree.root());
717 }
718
719 pub fn send_signed(
725 &mut self,
726 from: &str,
727 to: &str,
728 signed: &SignedValue<Signature>,
729 ) -> Result<crate::buffer::EnqueueResult, String> {
730 let edge = self
731 .edge_for_roles(from, to)
732 .cloned()
733 .ok_or_else(|| format!("no buffer for edge {from} → {to}"))?;
734 let buf = self
735 .buffers
736 .get_mut(&edge)
737 .ok_or_else(|| format!("no buffer for edge {from} → {to}"))?;
738 let result = buf.enqueue(signed.clone());
739 if matches!(result, crate::buffer::EnqueueResult::Ok) {
740 self.update_auth_tree(&edge, signed);
741 }
742 Ok(result)
743 }
744
745 pub fn send(
753 &mut self,
754 from: &str,
755 to: &str,
756 val: Value,
757 ) -> Result<crate::buffer::EnqueueResult, String> {
758 let signer = signing_key_for_endpoint(&Endpoint {
759 sid: self.sid,
760 role: from.to_string(),
761 });
762 let signature = signValue(&val, &signer);
763 self.send_signed(
764 from,
765 to,
766 &SignedValue {
767 payload: val,
768 signature,
769 sequence_no: 0,
770 },
771 )
772 }
773
774 pub fn send_with_sequence(
780 &mut self,
781 from: &str,
782 to: &str,
783 val: Value,
784 sequence_no: u64,
785 ) -> Result<crate::buffer::EnqueueResult, String> {
786 let signer = signing_key_for_endpoint(&Endpoint {
787 sid: self.sid,
788 role: from.to_string(),
789 });
790 let signature = signValue(&val, &signer);
791 self.send_signed(
792 from,
793 to,
794 &SignedValue {
795 payload: val,
796 signature,
797 sequence_no,
798 },
799 )
800 }
801
802 pub fn recv_signed(&mut self, from: &str, to: &str) -> Option<SignedValue<Signature>> {
804 let edge = self.edge_for_roles(from, to)?.clone();
805 self.buffers.get_mut(&edge).and_then(|buf| buf.dequeue())
806 }
807
808 pub fn recv_verified_signed(
814 &mut self,
815 from: &str,
816 to: &str,
817 ) -> Result<Option<SignedValue<Signature>>, String> {
818 let sender = Endpoint {
819 sid: self.sid,
820 role: from.to_string(),
821 };
822 let verifying = verifying_key_for_endpoint(&sender);
823 let signed = self.recv_signed(from, to);
824 let Some(signed) = signed else {
825 return Ok(None);
826 };
827 if !verifySignedValue(&signed.payload, &signed.signature, &verifying) {
828 return Err(format!(
829 "signature verification failed on edge {from} -> {to}"
830 ));
831 }
832 Ok(Some(signed))
833 }
834
835 pub fn recv_verified(&mut self, from: &str, to: &str) -> Result<Option<Value>, String> {
841 Ok(self
842 .recv_verified_signed(from, to)?
843 .map(|signed| signed.payload))
844 }
845
846 pub fn recv(&mut self, from: &str, to: &str) -> Option<Value> {
848 self.recv_verified(from, to).ok().flatten()
849 }
850
851 #[must_use]
853 pub fn has_message(&self, from: &str, to: &str) -> bool {
854 let Some(edge) = self.edge_for_roles(from, to) else {
855 return false;
856 };
857 self.buffers.get(edge).is_some_and(|buf| !buf.is_empty())
858 }
859
860 #[must_use]
862 pub fn lookup_handler_for_roles(&self, from: &str, to: &str) -> Option<&HandlerId> {
863 if self.edge_handlers.is_empty() {
864 return None;
865 }
866 let edge_key = self.edge_key_for_roles(from, to)?;
867 let handler_id = self.edge_handler_lookup.get(&edge_key)?;
868 self.handler_by_id(*handler_id)
869 }
870
871 #[must_use]
873 pub fn default_handler_binding(&self) -> Option<&HandlerId> {
874 if self.default_handler.is_empty() {
875 return None;
876 }
877 let handler_id = self.default_handler_id?;
878 self.handler_by_id(handler_id)
879 }
880
881 #[must_use]
883 pub fn has_bound_handler(&self) -> bool {
884 !self.default_handler.is_empty() || !self.edge_handlers.is_empty()
885 }
886}
887
888#[derive(Debug, Deserialize)]
889struct SessionStateSerde {
890 sid: SessionId,
891 roles: Vec<String>,
892 local_types: BTreeMap<Endpoint, TypeEntry>,
893 buffers: BTreeMap<Edge, SignedBuffer<Signature>>,
894 auth_leaves: BTreeMap<Edge, Vec<Hash>>,
895 #[serde(default)]
896 auth_trees: BTreeMap<Edge, AuthTree>,
897 auth_roots: BTreeMap<Edge, Hash>,
898 edge_handlers: BTreeMap<Edge, HandlerId>,
899 #[serde(default = "default_handler_id")]
900 default_handler: HandlerId,
901 edge_traces: BTreeMap<Edge, Vec<ValType>>,
902 status: SessionStatus,
903 epoch: usize,
904}
905
906impl<'de> Deserialize<'de> for SessionState {
907 fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
908 where
909 D: serde::Deserializer<'de>,
910 {
911 let raw = SessionStateSerde::deserialize(deserializer)?;
912 let mut session = Self {
913 sid: raw.sid,
914 roles: raw.roles,
915 role_ids: BTreeMap::new(),
916 local_types: raw.local_types,
917 buffers: raw.buffers,
918 edge_lookup: BTreeMap::new(),
919 handler_ids: BTreeMap::new(),
920 handlers_by_id: Vec::new(),
921 edge_handler_lookup: BTreeMap::new(),
922 default_handler_id: None,
923 label_ids: BTreeMap::new(),
924 labels_by_id: Vec::new(),
925 branch_lookup: BTreeMap::new(),
926 auth_leaves: raw.auth_leaves,
927 auth_trees: raw.auth_trees,
928 auth_roots: raw.auth_roots,
929 edge_handlers: raw.edge_handlers,
930 default_handler: raw.default_handler,
931 edge_traces: raw.edge_traces,
932 status: raw.status,
933 epoch: raw.epoch,
934 };
935 session.rebuild_derived_indexes();
936 Ok(session)
937 }
938}
939
940#[derive(Debug, Default, Serialize, Deserialize)]
945pub struct SessionStore {
946 sessions: BTreeMap<SessionId, SessionState>,
947 #[serde(default)]
948 archived_closed: Vec<ClosedSessionSummary>,
949 next_id: SessionId,
950}
951
952impl SessionStore {
953 #[must_use]
955 pub fn new() -> Self {
956 Self::default()
957 }
958
959 #[allow(clippy::needless_pass_by_value)]
963 pub fn open_with_sid(
964 &mut self,
965 sid: SessionId,
966 roles: Vec<String>,
967 buffer_config: &BufferConfig,
968 initial_types: &BTreeMap<String, LocalTypeR>,
969 ) -> SessionId {
970 let plan = SessionOpenPlan::new(&roles, initial_types);
971 self.open_with_sid_from_plan(sid, &plan, buffer_config)
972 }
973
974 pub fn open_with_sid_from_plan(
976 &mut self,
977 sid: SessionId,
978 plan: &SessionOpenPlan,
979 buffer_config: &BufferConfig,
980 ) -> SessionId {
981 let state = SessionState::from_open_plan(sid, plan, buffer_config);
982 self.sessions.insert(sid, state);
983 self.next_id = self.next_id.max(sid.saturating_add(1));
984 sid
985 }
986
987 #[allow(clippy::needless_pass_by_value)]
991 pub fn open(
992 &mut self,
993 roles: Vec<String>,
994 buffer_config: &BufferConfig,
995 initial_types: &BTreeMap<String, LocalTypeR>,
996 ) -> SessionId {
997 let sid = self.next_id;
998 self.open_with_sid(sid, roles, buffer_config, initial_types)
999 }
1000
1001 #[must_use]
1003 pub fn next_session_id(&self) -> SessionId {
1004 self.next_id
1005 }
1006
1007 #[must_use]
1013 pub fn lookup_type(&self, ep: &Endpoint) -> Option<&LocalTypeR> {
1014 self.sessions
1015 .get(&ep.sid)?
1016 .local_types
1017 .get(ep)
1018 .map(|e| &e.current)
1019 }
1020
1021 pub fn update_type(&mut self, ep: &Endpoint, new_type: LocalTypeR) {
1025 if let Some(session) = self.sessions.get_mut(&ep.sid) {
1026 if let Some(entry) = session.local_types.get_mut(ep) {
1027 entry.current = new_type;
1028 }
1029 session.refresh_endpoint_branch_lookup(ep);
1030 }
1031 }
1032
1033 pub fn update_original(&mut self, ep: &Endpoint, new_original: LocalTypeR) {
1035 if let Some(session) = self.sessions.get_mut(&ep.sid) {
1036 if let Some(entry) = session.local_types.get_mut(ep) {
1037 entry.original = new_original;
1038 }
1039 }
1040 }
1041
1042 #[must_use]
1044 pub fn original_type(&self, ep: &Endpoint) -> Option<&LocalTypeR> {
1045 self.sessions
1046 .get(&ep.sid)?
1047 .local_types
1048 .get(ep)
1049 .map(|e| &e.original)
1050 }
1051
1052 pub fn remove_type(&mut self, ep: &Endpoint) {
1054 if let Some(session) = self.sessions.get_mut(&ep.sid) {
1055 session.local_types.remove(ep);
1056 session.branch_lookup.remove(ep);
1057 }
1058 }
1059
1060 #[must_use]
1064 pub fn get(&self, sid: SessionId) -> Option<&SessionState> {
1065 self.sessions.get(&sid)
1066 }
1067
1068 pub fn get_mut(&mut self, sid: SessionId) -> Option<&mut SessionState> {
1070 self.sessions.get_mut(&sid)
1071 }
1072
1073 pub fn iter(&self) -> impl Iterator<Item = &SessionState> {
1075 self.sessions.values()
1076 }
1077
1078 pub fn close(&mut self, sid: SessionId) -> Result<(), String> {
1084 let session = self
1085 .sessions
1086 .get_mut(&sid)
1087 .ok_or_else(|| format!("session {sid} not found"))?;
1088
1089 session.status = SessionStatus::Closed;
1090 session.buffers.clear();
1091 session.edge_traces.clear();
1092 session.epoch = session.epoch.saturating_add(1);
1093 Ok(())
1094 }
1095
1096 #[must_use]
1098 pub fn closed_session_ids(&self) -> Vec<SessionId> {
1099 self.sessions
1100 .iter()
1101 .filter_map(|(sid, session)| {
1102 matches!(
1103 session.status,
1104 SessionStatus::Closed
1105 | SessionStatus::Cancelled
1106 | SessionStatus::Faulted { .. }
1107 )
1108 .then_some(*sid)
1109 })
1110 .collect()
1111 }
1112
1113 pub fn reap_sessions(&mut self, session_ids: &[SessionId]) -> Vec<ClosedSessionSummary> {
1120 let mut reaped = Vec::new();
1121 for sid in session_ids {
1122 let Some(session) = self.sessions.get(sid) else {
1123 continue;
1124 };
1125 if !matches!(
1126 session.status,
1127 SessionStatus::Closed | SessionStatus::Cancelled | SessionStatus::Faulted { .. }
1128 ) {
1129 continue;
1130 }
1131
1132 let session = self
1133 .sessions
1134 .remove(sid)
1135 .expect("session existence checked before removal");
1136 let summary = ClosedSessionSummary::from_session(&session);
1137 self.archived_closed.push(summary.clone());
1138 reaped.push(summary);
1139 }
1140 reaped
1141 }
1142
1143 pub fn reap_closed(&mut self) -> Vec<ClosedSessionSummary> {
1145 let sids = self.closed_session_ids();
1146 self.reap_sessions(&sids)
1147 }
1148
1149 #[must_use]
1151 pub fn active_count(&self) -> usize {
1152 self.sessions
1153 .values()
1154 .filter(|s| s.status == SessionStatus::Active)
1155 .count()
1156 }
1157
1158 #[must_use]
1160 pub fn live_count(&self) -> usize {
1161 self.sessions.len()
1162 }
1163
1164 #[must_use]
1166 pub fn session_ids(&self) -> Vec<SessionId> {
1167 self.sessions.keys().copied().collect()
1168 }
1169
1170 #[must_use]
1172 pub fn archived_closed(&self) -> &[ClosedSessionSummary] {
1173 &self.archived_closed
1174 }
1175
1176 #[must_use]
1178 pub fn memory_usage(&self) -> SessionStoreMemoryUsage {
1179 let mut usage = SessionStoreMemoryUsage {
1180 live_sessions: self.sessions.len(),
1181 archived_closed_sessions: self.archived_closed.len(),
1182 ..SessionStoreMemoryUsage::default()
1183 };
1184 usage.retained_bytes.archived_closed = self
1185 .archived_closed
1186 .iter()
1187 .map(ClosedSessionSummary::retained_bytes_estimate)
1188 .sum();
1189
1190 for session in self.sessions.values() {
1191 if matches!(
1192 session.status,
1193 SessionStatus::Closed | SessionStatus::Cancelled | SessionStatus::Faulted { .. }
1194 ) {
1195 usage.live_closed_sessions += 1;
1196 }
1197 usage.live_local_type_entries += session.local_types.len();
1198 usage.live_buffer_count += session.buffers.len();
1199 usage.live_buffered_messages += session
1200 .buffers
1201 .values()
1202 .map(BoundedBuffer::len)
1203 .sum::<usize>();
1204 usage.live_edge_handler_count += session.edge_handlers.len();
1205 usage.live_auth_leaf_count += session.auth_leaves.values().map(Vec::len).sum::<usize>();
1206 usage.live_auth_tree_count += session.auth_trees.len();
1207 usage.live_auth_root_count += session.auth_roots.len();
1208 usage.retained_bytes.live_sessions += session.retained_session_core_bytes();
1209 usage.retained_bytes.local_types += session.retained_local_type_bytes();
1210 usage.retained_bytes.buffers += session.retained_buffer_bytes();
1211 usage.retained_bytes.traces += session.retained_trace_bytes();
1212 usage.retained_bytes.auth += session.retained_auth_bytes();
1213 usage.retained_bytes.handlers += session.retained_handler_bytes();
1214 }
1215 usage.retained_bytes.total = usage
1216 .retained_bytes
1217 .live_sessions
1218 .saturating_add(usage.retained_bytes.archived_closed)
1219 .saturating_add(usage.retained_bytes.local_types)
1220 .saturating_add(usage.retained_bytes.buffers)
1221 .saturating_add(usage.retained_bytes.traces)
1222 .saturating_add(usage.retained_bytes.auth)
1223 .saturating_add(usage.retained_bytes.handlers);
1224
1225 usage
1226 }
1227
1228 #[must_use]
1230 pub fn lookup_handler(&self, edge: &Edge) -> Option<&HandlerId> {
1231 self.sessions
1232 .get(&edge.sid)?
1233 .lookup_handler_for_roles(&edge.sender, &edge.receiver)
1234 }
1235
1236 #[must_use]
1238 pub fn default_handler_for_session(&self, sid: SessionId) -> Option<&HandlerId> {
1239 self.sessions.get(&sid)?.default_handler_binding()
1240 }
1241
1242 pub fn set_default_handler_for_session(&mut self, sid: SessionId, handler: HandlerId) {
1244 if let Some(session) = self.sessions.get_mut(&sid) {
1245 let handler_id = session.intern_handler_binding(&handler);
1246 session.default_handler = handler;
1247 session.default_handler_id = Some(handler_id);
1248 }
1249 }
1250
1251 pub fn update_handler(&mut self, edge: &Edge, handler: HandlerId) {
1253 if let Some(session) = self.sessions.get_mut(&edge.sid) {
1254 let handler_id = session.intern_handler_binding(&handler);
1255 if let Some(edge_key) = session.edge_key_for_roles(&edge.sender, &edge.receiver) {
1256 session.edge_handler_lookup.insert(edge_key, handler_id);
1257 }
1258 session.edge_handlers.insert(edge.clone(), handler);
1259 }
1260 }
1261
1262 #[must_use]
1264 pub fn lookup_trace(&self, edge: &Edge) -> Option<&[ValType]> {
1265 self.sessions
1266 .get(&edge.sid)?
1267 .edge_traces
1268 .get(edge)
1269 .map(Vec::as_slice)
1270 }
1271
1272 pub fn update_trace(&mut self, edge: &Edge, trace: Vec<ValType>) {
1274 if let Some(session) = self.sessions.get_mut(&edge.sid) {
1275 session.edge_traces.insert(edge.clone(), trace);
1276 }
1277 }
1278}
1279
1280#[must_use]
1286pub fn unfold_mu(lt: &LocalTypeR) -> LocalTypeR {
1288 match lt {
1289 LocalTypeR::Mu { body, .. } => unfold_mu(body),
1290 other => other.clone(),
1291 }
1292}
1293
1294#[must_use]
1299pub fn unfold_if_var(cont: &LocalTypeR, original: &LocalTypeR) -> LocalTypeR {
1300 match cont {
1301 LocalTypeR::Var(_) => unfold_mu(original),
1302 LocalTypeR::Mu { .. } => unfold_mu(cont),
1303 other => other.clone(),
1304 }
1305}
1306
1307#[must_use]
1313pub fn unfold_if_var_with_scope(
1314 cont: &LocalTypeR,
1315 original: &LocalTypeR,
1316) -> (LocalTypeR, Option<LocalTypeR>) {
1317 match cont {
1318 LocalTypeR::Var(_) => (unfold_mu(original), None),
1319 LocalTypeR::Mu { .. } => (unfold_mu(cont), Some(cont.clone())),
1320 other => (other.clone(), None),
1321 }
1322}
1323
1324#[cfg(test)]
1325mod tests {
1326 use super::*;
1327 use serde_json::json;
1328 use telltale_types::Label;
1329
1330 fn default_types() -> BTreeMap<String, LocalTypeR> {
1331 let mut m = BTreeMap::new();
1332 m.insert(
1333 "A".to_string(),
1334 LocalTypeR::mu(
1335 "step",
1336 LocalTypeR::Send {
1337 partner: "B".into(),
1338 branches: vec![(Label::new("msg"), None, LocalTypeR::var("step"))],
1339 },
1340 ),
1341 );
1342 m.insert(
1343 "B".to_string(),
1344 LocalTypeR::mu(
1345 "step",
1346 LocalTypeR::Recv {
1347 partner: "A".into(),
1348 branches: vec![(Label::new("msg"), None, LocalTypeR::var("step"))],
1349 },
1350 ),
1351 );
1352 m
1353 }
1354
1355 fn single_send_recv_types() -> BTreeMap<String, LocalTypeR> {
1356 let mut types = BTreeMap::new();
1357 types.insert(
1358 "A".to_string(),
1359 LocalTypeR::send("B", Label::new("msg"), LocalTypeR::End),
1360 );
1361 types.insert(
1362 "B".to_string(),
1363 LocalTypeR::recv("A", Label::new("msg"), LocalTypeR::End),
1364 );
1365 types
1366 }
1367
1368 #[test]
1369 fn test_session_open_with_types() {
1370 let mut store = SessionStore::new();
1371 let types = default_types();
1372 let sid = store.open(
1373 vec!["A".into(), "B".into()],
1374 &BufferConfig::default(),
1375 &types,
1376 );
1377
1378 let ep_a = Endpoint {
1379 sid,
1380 role: "A".into(),
1381 };
1382 let ep_b = Endpoint {
1383 sid,
1384 role: "B".into(),
1385 };
1386
1387 assert!(matches!(
1389 store.lookup_type(&ep_a),
1390 Some(LocalTypeR::Send { .. })
1391 ));
1392 assert!(matches!(
1393 store.lookup_type(&ep_b),
1394 Some(LocalTypeR::Recv { .. })
1395 ));
1396 }
1397
1398 #[test]
1399 fn test_type_advance_and_unfold() {
1400 let mut store = SessionStore::new();
1401 let types = default_types();
1402 let sid = store.open(
1403 vec!["A".into(), "B".into()],
1404 &BufferConfig::default(),
1405 &types,
1406 );
1407
1408 let ep_a = Endpoint {
1409 sid,
1410 role: "A".into(),
1411 };
1412
1413 let lt = store.lookup_type(&ep_a).unwrap().clone();
1415 let (_, _vt, continuation) = match < {
1416 LocalTypeR::Send { branches, .. } => branches.first().unwrap().clone(),
1417 _ => panic!("expected Send"),
1418 };
1419
1420 let original = store.original_type(&ep_a).unwrap();
1422 let resolved = unfold_if_var(&continuation, original);
1423 assert!(matches!(resolved, LocalTypeR::Send { .. }));
1424
1425 store.update_type(&ep_a, resolved);
1427 assert!(matches!(
1428 store.lookup_type(&ep_a),
1429 Some(LocalTypeR::Send { .. })
1430 ));
1431 }
1432
1433 #[test]
1434 fn test_branch_lookup_tracks_type_updates_and_removals() {
1435 let mut store = SessionStore::new();
1436 let sid = store.open(
1437 vec!["A".into(), "B".into()],
1438 &BufferConfig::default(),
1439 &default_types(),
1440 );
1441 let ep_a = Endpoint {
1442 sid,
1443 role: "A".into(),
1444 };
1445
1446 let initial = store
1447 .get(sid)
1448 .expect("session exists")
1449 .lookup_branch_resolution(&ep_a, "msg")
1450 .expect("initial branch must be cached");
1451 assert_eq!(initial.direction, BranchDirection::Send);
1452 assert_eq!(initial.partner, "B");
1453 assert_eq!(initial.expected_type, None);
1454
1455 store.update_type(
1456 &ep_a,
1457 LocalTypeR::Send {
1458 partner: "B".into(),
1459 branches: vec![(Label::new("alt"), Some(ValType::Nat), LocalTypeR::End)],
1460 },
1461 );
1462
1463 let updated_session = store.get(sid).expect("session exists after type update");
1464 assert!(updated_session
1465 .lookup_branch_resolution(&ep_a, "msg")
1466 .is_none());
1467 let updated = updated_session
1468 .lookup_branch_resolution(&ep_a, "alt")
1469 .expect("updated branch must be cached");
1470 assert_eq!(updated.direction, BranchDirection::Send);
1471 assert_eq!(updated.partner, "B");
1472 assert_eq!(updated.expected_type, Some(ValType::Nat));
1473
1474 store.remove_type(&ep_a);
1475 assert!(store
1476 .get(sid)
1477 .expect("session exists after remove")
1478 .lookup_branch_resolution(&ep_a, "alt")
1479 .is_none());
1480 }
1481
1482 #[test]
1483 fn test_session_send_recv() {
1484 let mut store = SessionStore::new();
1485 let sid = store.open(
1486 vec!["A".into(), "B".into()],
1487 &BufferConfig::default(),
1488 &single_send_recv_types(),
1489 );
1490
1491 let session = store.get_mut(sid).unwrap();
1492 session.send("A", "B", Value::Nat(42)).unwrap();
1493 assert!(session.has_message("A", "B"));
1494 assert!(!session.has_message("B", "A"));
1495
1496 let val = session.recv("A", "B");
1497 assert_eq!(val, Some(Value::Nat(42)));
1498 }
1499
1500 #[test]
1501 fn test_open_allocates_only_protocol_reachable_edges() {
1502 let mut types = BTreeMap::new();
1503 types.insert(
1504 "A".to_string(),
1505 LocalTypeR::send("B", Label::new("msg"), LocalTypeR::End),
1506 );
1507 types.insert(
1508 "B".to_string(),
1509 LocalTypeR::recv("A", Label::new("msg"), LocalTypeR::End),
1510 );
1511 types.insert("C".to_string(), LocalTypeR::End);
1512
1513 let mut store = SessionStore::new();
1514 let sid = store.open(
1515 vec!["A".into(), "B".into(), "C".into()],
1516 &BufferConfig::default(),
1517 &types,
1518 );
1519
1520 let session = store.get_mut(sid).expect("session exists");
1521 assert_eq!(session.buffers.len(), 1);
1522 assert!(session.buffers.contains_key(&Edge::new(sid, "A", "B")));
1523 assert!(session.send("A", "B", Value::Nat(42)).is_ok());
1524 assert!(session.has_message("A", "B"));
1525 assert_eq!(session.recv("A", "B"), Some(Value::Nat(42)));
1526 assert_eq!(
1527 session.send("B", "A", Value::Nat(7)).unwrap_err(),
1528 "no buffer for edge B → A"
1529 );
1530 assert_eq!(
1531 session.send("A", "C", Value::Nat(9)).unwrap_err(),
1532 "no buffer for edge A → C"
1533 );
1534 }
1535
1536 #[test]
1537 fn test_close_clears_buffers_and_traces_even_when_messages_pending() {
1538 let mut store = SessionStore::new();
1539 let sid = store.open(
1540 vec!["A".into(), "B".into()],
1541 &BufferConfig::default(),
1542 &single_send_recv_types(),
1543 );
1544 let edge = Edge::new(sid, "A", "B");
1545 store
1546 .get_mut(sid)
1547 .expect("session exists")
1548 .send("A", "B", Value::Nat(7))
1549 .expect("enqueue pending message");
1550 store.update_trace(&edge, vec![ValType::Nat]);
1551
1552 store.close(sid).expect("close session");
1553 let session = store.get(sid).expect("session exists after close");
1554 assert_eq!(session.status, SessionStatus::Closed);
1555 assert!(session.buffers.is_empty());
1556 assert!(session.edge_traces.is_empty());
1557 }
1558
1559 #[test]
1560 fn test_reap_closed_archives_and_removes_session() {
1561 let mut store = SessionStore::new();
1562 let sid = store.open(
1563 vec!["A".into(), "B".into()],
1564 &BufferConfig::default(),
1565 &default_types(),
1566 );
1567
1568 store.close(sid).expect("close session");
1569 let summaries = store.reap_closed();
1570
1571 assert_eq!(summaries.len(), 1);
1572 assert_eq!(summaries[0].sid, sid);
1573 assert!(store.get(sid).is_none());
1574 assert_eq!(store.archived_closed().len(), 1);
1575 }
1576
1577 #[test]
1578 fn test_memory_usage_tracks_live_and_archived_closed_sessions() {
1579 let mut store = SessionStore::new();
1580 let sid = store.open(
1581 vec!["A".into(), "B".into()],
1582 &BufferConfig::default(),
1583 &default_types(),
1584 );
1585
1586 let before_close = store.memory_usage();
1587 assert_eq!(before_close.live_sessions, 1);
1588 assert_eq!(before_close.live_closed_sessions, 0);
1589 assert!(before_close.retained_bytes.total > 0);
1590 assert!(before_close.retained_bytes.live_sessions > 0);
1591 assert!(before_close.retained_bytes.buffers > 0);
1592
1593 store.close(sid).expect("close session");
1594 let after_close = store.memory_usage();
1595 assert_eq!(after_close.live_sessions, 1);
1596 assert_eq!(after_close.live_closed_sessions, 1);
1597 assert!(after_close.retained_bytes.total > 0);
1598
1599 store.reap_closed();
1600 let after_reap = store.memory_usage();
1601 assert_eq!(after_reap.live_sessions, 0);
1602 assert_eq!(after_reap.live_closed_sessions, 0);
1603 assert_eq!(after_reap.archived_closed_sessions, 1);
1604 assert_eq!(after_reap.retained_bytes.live_sessions, 0);
1605 assert_eq!(after_reap.retained_bytes.local_types, 0);
1606 assert_eq!(after_reap.retained_bytes.buffers, 0);
1607 assert!(after_reap.retained_bytes.archived_closed > 0);
1608 }
1609
1610 #[test]
1611 fn test_session_state_roundtrip_preserves_internal_role_edge_indexes() {
1612 let mut store = SessionStore::new();
1613 let sid = store.open(
1614 vec!["A".into(), "B".into(), "C".into()],
1615 &BufferConfig::default(),
1616 &{
1617 let mut types = single_send_recv_types();
1618 types.insert("C".to_string(), LocalTypeR::End);
1619 types
1620 },
1621 );
1622
1623 let session = store.get_mut(sid).expect("session exists");
1624 session.default_handler = "handler/default".to_string();
1625 session
1626 .edge_handlers
1627 .insert(Edge::new(sid, "A", "B"), "handler/ab".to_string());
1628 session.rebuild_derived_indexes();
1629 session
1630 .send("A", "B", Value::Nat(7))
1631 .expect("send should succeed");
1632
1633 let encoded = bincode::serialize(session).expect("serialize session");
1634 let mut decoded: SessionState = bincode::deserialize(&encoded).expect("deserialize");
1635
1636 assert_eq!(
1637 decoded.default_handler_binding().map(String::as_str),
1638 Some("handler/default")
1639 );
1640 assert_eq!(
1641 decoded
1642 .lookup_handler_for_roles("A", "B")
1643 .map(String::as_str),
1644 Some("handler/ab")
1645 );
1646 assert!(decoded.has_bound_handler());
1647 assert!(decoded
1648 .lookup_branch_resolution(
1649 &Endpoint {
1650 sid,
1651 role: "A".into()
1652 },
1653 "msg"
1654 )
1655 .is_some());
1656 assert!(decoded.has_message("A", "B"));
1657 assert_eq!(decoded.recv("A", "B"), Some(Value::Nat(7)));
1658 assert!(!decoded.has_message("A", "B"));
1659 }
1660
1661 #[test]
1662 fn test_namespace_isolation() {
1663 let mut store = SessionStore::new();
1664 let sid1 = store.open(
1665 vec!["A".into(), "B".into()],
1666 &BufferConfig::default(),
1667 &single_send_recv_types(),
1668 );
1669 let sid2 = store.open(
1670 vec!["A".into(), "B".into()],
1671 &BufferConfig::default(),
1672 &single_send_recv_types(),
1673 );
1674
1675 assert_ne!(sid1, sid2);
1676
1677 store
1678 .get_mut(sid1)
1679 .unwrap()
1680 .send("A", "B", Value::Nat(1))
1681 .unwrap();
1682 assert!(!store.get(sid2).unwrap().has_message("A", "B"));
1683 }
1684
1685 #[test]
1686 fn test_remove_type() {
1687 let mut store = SessionStore::new();
1688 let types = default_types();
1689 let sid = store.open(
1690 vec!["A".into(), "B".into()],
1691 &BufferConfig::default(),
1692 &types,
1693 );
1694
1695 let ep_a = Endpoint {
1696 sid,
1697 role: "A".into(),
1698 };
1699 assert!(store.lookup_type(&ep_a).is_some());
1700
1701 store.remove_type(&ep_a);
1702 assert!(store.lookup_type(&ep_a).is_none());
1703 }
1704
1705 #[test]
1706 fn test_cross_session_role_name_edge_collision_regression() {
1707 let mut store = SessionStore::new();
1708 let sid1 = store.open(
1709 vec!["A".into(), "B".into()],
1710 &BufferConfig::default(),
1711 &single_send_recv_types(),
1712 );
1713 let sid2 = store.open(
1714 vec!["A".into(), "B".into()],
1715 &BufferConfig::default(),
1716 &single_send_recv_types(),
1717 );
1718
1719 let e1 = Edge::new(sid1, "A", "B");
1720 let e2 = Edge::new(sid2, "A", "B");
1721 assert_ne!(e1, e2, "edges from distinct sessions must not collide");
1722 assert!(store
1723 .get(sid1)
1724 .expect("sid1 exists")
1725 .buffers
1726 .contains_key(&e1));
1727 assert!(store
1728 .get(sid2)
1729 .expect("sid2 exists")
1730 .buffers
1731 .contains_key(&e2));
1732 }
1733
1734 #[test]
1735 fn test_edge_handler_and_trace_bindings() {
1736 let mut store = SessionStore::new();
1737 let sid = store.open(
1738 vec!["A".into(), "B".into()],
1739 &BufferConfig::default(),
1740 &BTreeMap::new(),
1741 );
1742 let edge = Edge::new(sid, "A", "B");
1743
1744 assert!(store.lookup_handler(&edge).is_none());
1745 assert_eq!(
1746 store.default_handler_for_session(sid).map(String::as_str),
1747 Some(DEFAULT_HANDLER_ID)
1748 );
1749 store.update_handler(&edge, "handler/send".to_string());
1750 assert_eq!(
1751 store.lookup_handler(&edge).map(String::as_str),
1752 Some("handler/send")
1753 );
1754 store.set_default_handler_for_session(sid, "handler/default".to_string());
1755 assert_eq!(
1756 store.default_handler_for_session(sid).map(String::as_str),
1757 Some("handler/default")
1758 );
1759 assert!(
1760 store.get(sid).expect("session exists").has_bound_handler(),
1761 "internal handler ids must remain populated after updates"
1762 );
1763
1764 assert!(store.lookup_trace(&edge).is_none());
1765 store.update_trace(&edge, vec![ValType::Nat]);
1766 assert_eq!(store.lookup_trace(&edge), Some([ValType::Nat].as_slice()));
1767 }
1768
1769 #[test]
1770 fn test_decode_edge_json_requires_sid_sender_receiver() {
1771 let sid_qualified = json!({
1772 "sid": 7,
1773 "sender": "A",
1774 "receiver": "B"
1775 });
1776 let e = decode_edge_json(&sid_qualified, None).expect("decode sid-qualified edge");
1777 assert_eq!(e, Edge::new(7, "A", "B"));
1778
1779 let no_sid = json!({
1780 "sender": "A",
1781 "receiver": "B"
1782 });
1783 let e2 = decode_edge_json(&no_sid, Some(11)).expect("decode edge with sid hint");
1784 assert_eq!(e2, Edge::new(11, "A", "B"));
1785
1786 let legacy = json!({
1787 "from": "A",
1788 "to": "B",
1789 "sid": 11
1790 });
1791 let err = decode_edge_json(&legacy, None).expect_err("legacy edge shape must be rejected");
1792 assert!(err.contains("invalid edge json"), "unexpected error: {err}");
1793 }
1794}