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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
use super::{
Arc, ConnectionNotifier, ConnectionRuntime, MountKind, PassPrincipal, Registry, ServerError,
WorkerRegistration, lock,
};
impl ConnectionRuntime {
pub(in crate::server::connection) fn pressure(&self, connection: u64, response: &str) {
if let Some(signals) = self.signals.get() {
signals.event(crate::server::self_streams::events::ServerEvent::Pressure {
connection,
response,
});
}
}
/// Returns the configured connection-keyed notifier, if any.
pub(in crate::server::connection) fn notifier(&self) -> Option<&Arc<dyn ConnectionNotifier>> {
self.notifier.as_ref()
}
/// Offers a channel publish to the notifier's observability-drain tap, returning
/// `true` when the application consumed it (so the connection process skips the
/// normal fan-out). `false` when no notifier is installed (liminal standalone) or
/// the notifier did not recognise the channel, so the caller can invoke it
/// unconditionally and fall through to the normal publish path.
pub(in crate::server::connection) fn notifier_channel_publish(
&self,
pid: u64,
channel: &str,
payload: &[u8],
) -> bool {
self.notifier
.as_ref()
.is_some_and(|notifier| notifier.on_channel_publish(pid, channel, payload))
}
/// SRV-012 R7: connection `pid`'s pass names `participant`, so the
/// topology shows it on that connection from now on, in lowercase hex.
pub(in crate::server::connection) fn pass_attached(
&self,
pid: u64,
mount: MountKind,
participant: &[u8],
) {
const HEX: &[u8; 16] = b"0123456789abcdef";
if let Some(signals) = self.signals.get() {
let mut text = String::with_capacity(participant.len() * 2);
for byte in participant {
text.push(char::from(HEX[usize::from(byte >> 4)]));
text.push(char::from(HEX[usize::from(byte & 0x0f)]));
}
signals.topology(crate::console::TopologyChange::ConnectionOpened {
id: pid,
mount: mount.into(),
participant: Some(text),
});
}
}
/// SRV-012 `to_stations`: the subscription whose library subscriber is
/// `subscriber` belongs to connection `pid`, so a publish it receives
/// names `pid`'s station.
pub(in crate::server::connection) fn receiver(&self, pid: u64, subscriber: u64) {
if let Some(signals) = self.signals.get() {
signals.receiver(subscriber, pid);
}
}
/// SRV-012 R7: connection `pid` starts (`present`) or stops using `line`
/// one way; a worker's flows are its line's stops.
pub(in crate::server::connection) fn flow(
&self,
pid: u64,
line: &str,
role: crate::console::FlowRole,
present: bool,
) {
if let Some(signals) = self.signals.get() {
signals.topology(crate::console::TopologyChange::Flow {
connection: pid,
line: line.to_owned(),
role,
present,
});
}
}
/// Stores `registration` on the connection record for `pid`, so the close
/// path can later fire `on_worker_unregistered` for exactly the connections
/// that registered. A missing record (the connection already closed) is a
/// no-op.
///
/// # Errors
/// Returns [`ServerError`] when the connection registry mutex is poisoned.
pub(in crate::server::connection) fn set_registration(
&self,
pid: u64,
registration: WorkerRegistration,
) -> Result<(), ServerError> {
let identity = registration.identity.clone();
let subjects = vec![registration.task_queue.clone()];
let activities = registration
.activities
.iter()
.map(|activity| crate::console::StationActivity {
name: activity.name.clone(),
input_schema_json: activity.input_schema_json.clone(),
output_schema_json: activity.output_schema_json.clone(),
})
.collect();
let mut records = lock(&self.records, "connection registry")?;
let Registry {
connections,
directory,
} = &mut *records;
let stored = match connections.get_mut(&pid) {
Some(record) => {
// SRV-027 R1: the directory follows the record under the one
// lock; a replacement registration moves the subject.
if let Some(earlier) = record.registration.as_ref() {
directory.forget(pid, &earlier.task_queue);
self.services.site_registration_closed(pid)?;
}
directory.register(pid, ®istration.task_queue);
record.registration = Some(registration);
true
}
None => false,
};
drop(records);
// SRV-012 R7: a registered worker is a station on the topology, by
// its identity, until its connection closes; SRV-027 R3: with its
// subjects and activities.
if stored && let Some(signals) = self.signals.get() {
signals.topology(crate::console::TopologyChange::Worker {
connection: pid,
identity,
subjects,
activities,
});
}
Ok(())
}
/// Stores the registry principal a successful pass-stamped `Connect`
/// verified on the connection record for `pid`, then fires
/// `on_pass_attached` — so the close path can later fire the matching
/// `on_pass_detached` for exactly the connections that attached, with the
/// same principal. The store precedes the fire and the registry lock is
/// released before it, so the notifier may call back into the supervisor.
/// A missing record (the connection already closed) stores nothing and
/// fires nothing: an attach with no record could never be paired with a
/// detach.
///
/// # Errors
/// Returns [`ServerError`] when the connection registry mutex is poisoned.
pub(in crate::server::connection) fn attach_pass_principal(
&self,
pid: u64,
principal: &PassPrincipal,
) -> Result<(), ServerError> {
let stored = lock(&self.records, "connection registry")?
.connections
.get_mut(&pid)
.is_some_and(|record| {
record.pass_principal = Some(principal.clone());
true
});
if stored {
if let Some(notifier) = self.notifier.as_ref() {
notifier.on_pass_attached(pid, principal);
}
}
Ok(())
}
}
impl ConnectionRuntime {
pub(in crate::server::connection) fn clear_registration(
&self,
pid: u64,
) -> Result<(), ServerError> {
let mut records = lock(&self.records, "connection registry")?;
if let Some(registration) = records
.connections
.get_mut(&pid)
.and_then(|record| record.registration.take())
{
records.directory.forget(pid, ®istration.task_queue);
self.services.site_registration_closed(pid)?;
}
drop(records);
if let Some(notifier) = self.notifier.as_ref() {
notifier.on_worker_unregistered(pid);
}
Ok(())
}
}