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
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
//! The service route keeps admission before the actual dispatch.
use super::{
ConnectionProcessState, ConnectionRuntime, FrameAction, HotAction, MembershipAction, Request,
SERVER_ERROR_CODE, conversation_error, lys_decision, membership_refusal,
route_conversation_message, unsupported_by_front_door,
};
use liminal::protocol::{CONVERSATION_REPLY_REQUESTED_FLAG, Frame, MessageEnvelope};
use std::time::Instant;
pub(in crate::server::connection) fn conversation_message(
pid: u64,
runtime: &ConnectionRuntime,
state: &mut ConnectionProcessState,
request: Request,
envelope: &MessageEnvelope,
) -> FrameAction {
let Request {
flags,
stream_id,
conversation_id,
} = request;
let services = runtime.services();
// A capability-scoped adapter serves no conversations, so a conversation could
// never have been opened: reject with a clear "not supported" message rather
// than the misleading "not open" one. Full mode keeps the existing path.
if !services.supports_channel_operations() {
return conversation_error(
stream_id,
conversation_id,
&unsupported_by_front_door("conversation messaging"),
);
}
let Some(conversation) = state.conversations.get(&conversation_id) else {
return conversation_error(
stream_id,
conversation_id,
"conversation is not open on this connection",
);
};
if let Some(response) =
route_conversation_message(runtime, state, request, conversation.subject(), envelope)
{
return response;
}
// ACCESS-001 R2, `ConversationMessage` = Hot(Call, ConversationBinding):
// each message is its own decision on the subject the conversation was
// opened on, so a grant that lapsed since the open refuses here.
match services.admit_site_call(
state.lys_session.as_deref(),
conversation.subject(),
super::unix_seconds_now().saturating_mul(1000),
) {
Err(error) => return conversation_error(stream_id, conversation_id, &error.to_string()),
Ok(false) => {
if let Some(refused) = lys_decision(
runtime,
state,
conversation.subject(),
HotAction::Call,
|message| Frame::ConversationError {
flags: 0,
stream_id,
conversation_id,
reason_code: SERVER_ERROR_CODE,
message: Some(message),
},
) {
return refused;
}
}
Ok(true) => {}
}
// SRV-027 R2: a subject a registered worker serves carries requests to
// that worker, pushed on the subject and answered on this conversation;
// the message never reaches the channel.
if let Some(worker) = runtime.serving(conversation.subject()) {
let subject = conversation.subject().to_owned();
return crate::server::connection::worker_route::routed(
pid, runtime, state, request, &subject, envelope, worker,
);
}
// Pre-existing fire-and-forget semantics: without the reply-requested flag the
// message is forwarded and the server stays silent on success, exactly as
// before. The reply leg is purely additive and only runs when the client
// explicitly asked for a correlated reply on this frame.
if flags & CONVERSATION_REPLY_REQUESTED_FLAG == 0 {
if let Err(error) = services.conversation_message(conversation, envelope, None) {
return conversation_error(stream_id, conversation_id, &error.to_string());
}
return FrameAction::NoResponse;
}
// R1(vi) (§1.2(3b)) — THE LIVE BEHAVIOR CHANGE. The old path BLOCKED the
// connection slice for up to 5 s draining the participant's reply
// (`receive_reply`); now the reply-requested operation rides the bounded
// per-connection pending-reply table and the slice returns immediately.
//
// ADMISSION COMES FIRST (cap-before-mutation, review round 1 item 1): the
// reservation is made BEFORE the message is forwarded, so a cap-refused
// request — including the tombstone self-wedge — NEVER reaches the
// participant. A refused-but-forwarded request would produce an orphan
// reply with no table entry, which could later FIFO-match a younger
// admitted operation on the same conversation: wrong-reply delivery, the
// corruption class the tombstone ruling kills. If the forward itself fails
// after admission, the exact reservation is rolled back (no reply can ever
// arrive for a message the participant never received, so the entry must
// not linger to time out into a spurious tombstone).
//
// ACCESS-006 R3: the reply is data released to this caller, so on a
// protected channel its read is decided before the request is forwarded
// and carried to the reply's final release.
if let Some(refused) = membership_refusal(state, conversation.subject(), MembershipAction::Read)
{
return conversation_error(stream_id, conversation_id, &refused);
}
let release = state
.membership
.release_check(conversation.subject(), MembershipAction::Read);
let op_id = match state
.pending_replies
.admit(conversation_id, stream_id, Instant::now())
{
Ok(op_id) => op_id,
Err(error) => {
return FrameAction::Respond(
crate::server::connection::pending_reply::cap_refusal_frame(
stream_id,
conversation_id,
&error,
),
);
}
};
if let Some(release) = release {
state.pending_replies.bind_release(op_id, release);
}
if let Err(error) = services.conversation_message(conversation, envelope, Some(op_id)) {
state.pending_replies.cancel(op_id);
return conversation_error(stream_id, conversation_id, &error.to_string());
}
state.pending_replies.dispatched(op_id, Instant::now());
FrameAction::NoResponse
}