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
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
//! Tap channels follow their live subscribers and admission reservations.
use super::{
AttachedTap, ConsoleError, ConsoleFailure, ConsoleHub, MAX_LIVE_TAPS, TAP_SUBJECT_PREFIX,
TOPOLOGY_SUBJECT, TapError, TapFilter, add, failed, len_u64, lock, pending_total, set,
};
use std::collections::VecDeque;
impl ConsoleHub {
/// Whether a new subscription to `channel` is within the live-tap limit:
/// always for a channel that is not a tap or whose tap is live.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the tap table is poisoned.
pub fn tap_room(&self, channel: &str) -> Result<bool, TapError> {
if !channel.starts_with(TAP_SUBJECT_PREFIX) {
return Ok(true);
}
let taps = lock(&self.taps, "console taps")?;
Ok(taps.live.contains_key(channel)
|| taps.reserved.contains_key(channel)
|| taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS)
}
/// Admits a subscribe to `channel` against [`MAX_LIVE_TAPS`] and, for a
/// tap not yet live, reserves its slot in the same lock, so two racing
/// subscribes cannot both take the last slot. Answers false when no slot
/// is left. The attach that follows takes the reservation up;
/// [`Self::release_tap`] gives it back when the subscribe fails.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the tap table is poisoned.
pub fn reserve_tap(&self, channel: &str) -> Result<bool, TapError> {
if !channel.starts_with(TAP_SUBJECT_PREFIX) {
return Ok(true);
}
let mut taps = lock(&self.taps, "console taps")?;
let admitted = if taps.live.contains_key(channel) {
true
} else if let Some(holders) = taps.reserved.get_mut(channel) {
*holders = holders.saturating_add(1);
true
} else if taps.live.len() + taps.reserved.len() < MAX_LIVE_TAPS {
taps.reserved.insert(channel.to_owned(), 1);
true
} else {
false
};
drop(taps);
Ok(admitted)
}
/// Gives back one reservation of `channel` for a subscribe that failed
/// after admission. When it was the last and no tap is live there, the
/// tap channel admission registered is taken off the roster by `retire`,
/// under the taps lock, so no subscribe reserving the tap again can see
/// its entry go: a refused subscribe leaves the roster as it found it. A
/// failed `retire` queues the subject for the pump; a poisoned table is
/// counted as a detach failure.
pub fn release_tap(
&self,
channel: &str,
retire: impl FnOnce(&str) -> Result<(), ConsoleError>,
) {
if !channel.starts_with(TAP_SUBJECT_PREFIX) {
return;
}
let Ok(mut taps) = lock(&self.taps, "console taps") else {
failed(
&self.counts.tap_detach_failures,
1,
ConsoleFailure::TapDetach,
);
return;
};
let last = match taps.reserved.get_mut(channel) {
Some(holders) if *holders > 1 => {
*holders -= 1;
false
}
Some(_) => {
taps.reserved.remove(channel);
true
}
None => false,
};
if !last || taps.live.contains_key(channel) || retire(channel).is_ok() {
return;
}
taps.retiring.push(channel.to_owned());
drop(taps);
failed(
&self.counts.tap_detach_failures,
1,
ConsoleFailure::TapDetach,
);
self.wake();
}
/// Counts a subscription to `channel`. A new `console.topology`
/// subscriber makes the latest document due again, so it is sent the
/// whole topology (R7); the first to a `console.tap.<target>` starts that
/// target's tap. Any other channel is ignored.
///
/// # Errors
/// Returns [`ConsoleError::TapLimit`] when a new tap would pass
/// [`MAX_LIVE_TAPS`], or [`ConsoleError::Tap`] when a lock is poisoned;
/// either is counted in `tap_attach_failures`.
pub fn attach(&self, channel: &str) -> Result<(), ConsoleError> {
if channel == TOPOLOGY_SUBJECT {
let mut state = self.lock_topology()?;
state.subscribers = state.subscribers.saturating_add(1);
state.stale = true;
drop(state);
self.wake();
return Ok(());
}
let Some(target) = channel.strip_prefix(TAP_SUBJECT_PREFIX) else {
return Ok(());
};
self.attach_tap(channel, target).inspect_err(|_| {
failed(
&self.counts.tap_attach_failures,
1,
ConsoleFailure::TapAttach,
);
})
}
fn attach_tap(&self, channel: &str, target: &str) -> Result<(), ConsoleError> {
let mut taps = lock(&self.taps, "console taps")?;
match taps.reserved.get_mut(channel) {
Some(holders) if *holders > 1 => *holders -= 1,
Some(_) => {
taps.reserved.remove(channel);
}
None => {}
}
if let Some(entry) = taps.live.get_mut(channel) {
entry.subscribers = entry.subscribers.saturating_add(1);
return Ok(());
}
if taps.live.len() >= MAX_LIVE_TAPS {
return Err(ConsoleError::TapLimit {
subject: channel.to_owned(),
limit: MAX_LIVE_TAPS,
});
}
let subscription = self.tap.subscribe(TapFilter {
subject: target.to_owned(),
})?;
taps.retiring.retain(|subject| subject != channel);
let generation = taps.generations;
taps.generations = taps.generations.wrapping_add(1);
taps.live.insert(
channel.to_owned(),
AttachedTap {
subscription,
subscribers: 1,
pending: VecDeque::new(),
generation,
},
);
set(&self.counts.taps_attached, len_u64(taps.live.len()));
drop(taps);
Ok(())
}
/// Uncounts a subscription to `channel`. The last one to leave a
/// `console.tap.<target>` stops that target's tap and queues the channel
/// for roster removal by the next pump. Safe on the channel actor's
/// thread: it takes only hub locks and wakes without waiting.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when a lock is poisoned; counted in
/// `tap_detach_failures`.
pub fn detach(&self, channel: &str) -> Result<(), TapError> {
self.detach_inner(channel).inspect_err(|_| {
failed(
&self.counts.tap_detach_failures,
1,
ConsoleFailure::TapDetach,
);
})
}
fn detach_inner(&self, channel: &str) -> Result<(), TapError> {
if channel == TOPOLOGY_SUBJECT {
let mut state = self.lock_topology()?;
state.subscribers = state.subscribers.saturating_sub(1);
drop(state);
return Ok(());
}
if !channel.starts_with(TAP_SUBJECT_PREFIX) {
return Ok(());
}
let mut taps = lock(&self.taps, "console taps")?;
let last = taps.live.get_mut(channel).is_some_and(|entry| {
entry.subscribers = entry.subscribers.saturating_sub(1);
entry.subscribers == 0
});
if !last {
return Ok(());
}
// R9: its pending records and unread ring leave with the tap, since
// nobody is left to read them; each is counted by name.
if let Some(entry) = taps.live.remove(channel) {
add(
&self.counts.tap_records_abandoned,
len_u64(entry.pending.len()),
);
add(
&self.counts.tap_messages_abandoned,
len_u64(entry.subscription.held().unwrap_or_default()),
);
}
taps.retiring.push(channel.to_owned());
set(&self.counts.taps_attached, len_u64(taps.live.len()));
let pending = pending_total(&taps.live);
drop(taps);
set(&self.counts.console_pending_records, pending);
self.wake();
Ok(())
}
/// Subjects with a live tap, sorted.
///
/// # Errors
/// Returns [`TapError::Poisoned`] when the tap table is poisoned.
pub fn attached_subjects(&self) -> Result<Vec<String>, TapError> {
Ok(lock(&self.taps, "console taps")?
.live
.keys()
.cloned()
.collect())
}
}