1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
//! The pending-reply servicing every door runs each slice: the link calls
//! and the routed calls that settled (SRV-027 R2), the deadlines that
//! passed, and each conversation's buffered replies, correlated through the
//! pending-reply table. One body for the raw wire and the WebSocket door
//! (AGT-006 R2): a door that kept its own copy lacked the routed leg, and a
//! worker's answer to a call through it timed out on the table.
use liminal::protocol::Frame;
use super::state::ConnectionProcessState;
use super::supervisor::runtime::ConnectionRuntime;
/// The frames the connection owes this slice, in the order the raw wire
/// has always written them. Non-blocking: a reply not yet buffered is left
/// for the next wake.
///
/// # Errors
/// A far call whose authenticated answer could not be released.
pub(super) fn settled_frames(
state: &mut ConnectionProcessState,
runtime: &ConnectionRuntime,
) -> Result<Vec<Frame>, String> {
let mut frames = super::link_route::settled_calls(state, runtime)?;
frames.extend(state.pending_replies.expire_due(std::time::Instant::now()));
frames.extend(super::worker_route::settled(state, runtime));
for conversation_id in state.pending_replies.conversations_with_entries() {
let Some(conversation) = state.conversations.get(&conversation_id) else {
// The conversation closed while entries were pending: sweep them
// rather than poll a conversation that no longer exists.
state.pending_replies.remove_conversation(conversation_id);
continue;
};
// Every buffered reply this slice, each matched to its exact
// operation; a reply that consumed a tombstone is discarded, never
// delivered late.
while let Some((op_id, reply)) = conversation.try_receive_reply() {
if let Some(frame) = state
.pending_replies
.match_reply(conversation_id, op_id, reply)
{
frames.push(frame);
}
}
}
Ok(frames)
}