Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The change drain and its delivery lanes: batching, per-(tenant, subscription) folding, the byte ledger and the per-tenant delivery slots.
3 :
4 : use super::*;
5 :
6 : /// Groups dropped from a full delivery lane, since process start.
7 : #[cfg(all(test, not(target_arch = "wasm32")))]
8 8 : pub(crate) fn lane_drops() -> u64 {
9 8 : LANE_DROPS.load(std::sync::atomic::Ordering::Relaxed)
10 8 : }
11 :
12 : /// Changes one drain of the queue folds into one delivery pass — a batch
13 : /// request's N writes arrive as N events back to back and leave as ONE
14 : /// notification per matching subscription. Bounded so a flood cannot hold
15 : /// the first notification back indefinitely.
16 : pub(super) const CHANGE_BATCH: usize = 256;
17 :
18 : /// The estimated heap one drain's groups may hold before they leave for their
19 : /// lanes: an eighth of what the lanes hold together.
20 : #[cfg(not(target_arch = "wasm32"))]
21 : pub(super) static DRAIN_BYTES: std::sync::LazyLock<usize> =
22 22 : std::sync::LazyLock::new(|| *crate::bounds::DELIVERY_LANE_BYTES / 8);
23 :
24 : /// One matched (subscription, entity) pair before delivery.
25 : pub(super) struct Matched {
26 : pub(super) tenant: TenantId,
27 : /// Shared with the mirror that holds it: a change is evaluated against
28 : /// every candidate subscription, so owning a copy here made the cost of
29 : /// a change scale with the size of those documents.
30 : pub(super) sub: Arc<Value>,
31 : pub(super) ctx: Arc<Context>,
32 : pub(super) data: Vec<Value>,
33 : }
34 :
35 4 : pub async fn process_change(
36 4 : st: &AppState,
37 4 : tenant_str: &str,
38 4 : before: Option<Value>,
39 4 : after: Option<Value>,
40 4 : ) {
41 4 : process_changes(st, vec![(tenant_str.to_owned(), before, after)]).await;
42 4 : }
43 :
44 : /// 5.8.6: "the Notification ... data ... shall contain the Entities that
45 : /// match" — every change of one drain that matches the same subscription
46 : /// travels in one notification, so a batch of N entities is one POST with N
47 : /// data entries (and timesSent moves by one), never N POSTs.
48 8 : pub(crate) async fn process_changes(st: &AppState, changes: Vec<Change>) {
49 8 : process_queued(
50 8 : st,
51 24 : changes.into_iter().map(|c| (u64::MAX, c)).collect(),
52 8 : None,
53 : )
54 8 : .await;
55 8 : }
56 :
57 : /// [`process_changes`] of changes each stamped with the [`crate::mirror::tick`]
58 : /// they were queued at.
59 : ///
60 : /// With `done` (the matcher queue's drain), each group is handed to its
61 : /// subscription's lane and the drain goes on to the next batch: a slow or
62 : /// dead endpoint holds its own lane, never the matching of every other
63 : /// change. `done` falls when the last group of the batch has left.
64 : /// Without it (a bus event, a direct call), the groups are delivered before
65 : /// this returns.
66 89304 : pub(super) async fn process_queued(
67 89304 : st: &AppState,
68 89304 : changes: Vec<(u64, Change)>,
69 89304 : done: Option<Arc<BatchDone>>,
70 89304 : ) {
71 89304 : let mut groups: Vec<Matched> = Vec::new();
72 : // (tenant, subscription id) → its group. A scan for the group would be
73 : // linear in the subscriptions already matched, and a drain of
74 : // CHANGE_BATCH changes over S matching subscriptions walks it S times per
75 : // change: quadratic in the one dimension this broker is built to grow
76 : // (100 000 subscriptions), on the notification hot path.
77 89304 : let mut index: std::collections::HashMap<(String, String), usize> =
78 89304 : std::collections::HashMap::new();
79 89304 : let mut ctx_memo = CtxMemo::new();
80 : // estimated heap of the groups this drain holds
81 : #[cfg(not(target_arch = "wasm32"))]
82 89304 : let mut held = 0usize;
83 140905 : for (queued, (tenant_str, before, after)) in changes {
84 140905 : let matched = matches_for(st, &tenant_str, before, after, queued, &mut ctx_memo);
85 : #[cfg(not(target_arch = "wasm32"))]
86 140905 : let matched = match tokio::time::timeout(STEP_DEADLINE, matched).await {
87 140905 : Ok(m) => m,
88 : Err(_) => {
89 0 : note_stall("match", &tenant_str, "");
90 0 : continue;
91 : }
92 : };
93 : #[cfg(target_arch = "wasm32")]
94 : let matched = matched.await;
95 140905 : for m in matched {
96 : #[cfg(not(target_arch = "wasm32"))]
97 : {
98 9038 : held = m
99 9038 : .data
100 9038 : .iter()
101 9062 : .fold(held, |n, e| n.saturating_add(heap_estimate(e)));
102 : }
103 9038 : let key = (
104 9038 : m.tenant.as_str().to_owned(),
105 9038 : m.sub
106 9038 : .get("id")
107 9038 : .and_then(Value::as_str)
108 9038 : .unwrap_or_default()
109 9038 : .to_owned(),
110 9038 : );
111 9038 : match index.get(&key) {
112 153 : Some(&i) => {
113 153 : if let Some(group) = groups.get_mut(i) {
114 153 : group.data.extend(m.data);
115 153 : }
116 : }
117 8885 : None => {
118 8885 : index.insert(key, groups.len());
119 8885 : groups.push(m);
120 8885 : }
121 : }
122 : }
123 : // A drain of CHANGE_BATCH large changes over many subscriptions held
124 : // a shaped copy per (subscription, entity) until its end: 256 changes
125 : // of 3 KB entities times 20 subscriptions was 750 MB. Past
126 : // DRAIN_BYTES the groups so far leave for their lanes; a lane is
127 : // FIFO, so a subscription's changes still leave in drain order.
128 : #[cfg(not(target_arch = "wasm32"))]
129 140905 : if let Some(done) = &done {
130 140881 : if held > *DRAIN_BYTES {
131 0 : for g in groups.drain(..) {
132 0 : to_lane(st, g, done);
133 0 : }
134 0 : index.clear();
135 0 : held = 0;
136 140881 : }
137 24 : }
138 : }
139 : // Groups are distinct subscriptions, so they leave concurrently; a
140 : // subscription's changes in this drain are already one group. On a lane
141 : // its groups leave in drain order; awaited, the next pass starts only
142 : // after this one. Either way per-subscription order holds.
143 : #[cfg(not(target_arch = "wasm32"))]
144 89304 : if let Some(done) = done {
145 89296 : for g in groups {
146 8751 : to_lane(st, g, &done);
147 8751 : }
148 89296 : return;
149 8 : }
150 : // Each group is its own task: a group's bookkeeping reads and writes
151 : // await the store, and one task's `for_each_concurrent` would still put
152 : // every group's round-trips on a single task's poll budget.
153 : #[cfg(not(target_arch = "wasm32"))]
154 : {
155 8 : let mut set = tokio::task::JoinSet::new();
156 134 : for g in groups {
157 134 : let st = st.clone();
158 134 : set.spawn(async move { deliver_group(&st, g).await });
159 : }
160 142 : while let Some(joined) = set.join_next().await {
161 134 : if joined.is_err() {
162 0 : note_panic();
163 134 : }
164 : }
165 : }
166 : #[cfg(target_arch = "wasm32")]
167 : {
168 : drop(done);
169 : use futures_util::StreamExt;
170 : futures_util::stream::iter(groups)
171 : .for_each_concurrent(*DELIVERY_WIDTH, |g| async move {
172 : deliver(st, &g.tenant, &g.sub, g.data, &g.ctx).await;
173 : })
174 : .await;
175 : }
176 89304 : }
177 :
178 : /// One group out: the delivery permits, then the delivery, each bounded by
179 : /// [`STEP_DEADLINE`].
180 : #[cfg(not(target_arch = "wasm32"))]
181 4661 : pub(super) async fn deliver_group(st: &AppState, g: Matched) {
182 4661 : let id = g.sub.get("id").and_then(Value::as_str).unwrap_or_default();
183 : // The slots bound how many deliveries can wait on a peer. An in-process
184 : // sink has no peer to wait on, and its store calls join its tenant's
185 : // shared statements, so a fan-out to 1 000 sockets no longer
186 : // queues behind 8 slots for work that never leaves the process.
187 4661 : let _permits = if in_process_sink(st, &g.sub) {
188 116 : None
189 : } else {
190 4545 : match tokio::time::timeout(STEP_DEADLINE, delivery_permits(&g.tenant)).await {
191 4545 : Ok(permits) => Some(permits),
192 : Err(_) => {
193 0 : note_stall("permits", g.tenant.as_str(), id);
194 0 : tell_lost(st, &g);
195 0 : return;
196 : }
197 : }
198 : };
199 4661 : let delivered = deliver(st, &g.tenant, &g.sub, g.data, &g.ctx);
200 4661 : if tokio::time::timeout(STEP_DEADLINE, delivered)
201 4661 : .await
202 4610 : .is_err()
203 0 : {
204 0 : note_stall("deliver", g.tenant.as_str(), id);
205 4610 : }
206 4610 : }
207 :
208 : /// A drained batch still owed to its subscribers: while any of its groups
209 : /// waits on a lane, the batch counts as pending, so a shutdown drain that
210 : /// waits for zero closes the pool only after the last delivery.
211 : pub(crate) struct BatchDone {
212 : pub(super) pending: Arc<std::sync::atomic::AtomicUsize>,
213 : pub(super) taken: usize,
214 : }
215 :
216 : impl Drop for BatchDone {
217 89340 : fn drop(&mut self) {
218 89340 : self.pending
219 89340 : .fetch_sub(self.taken, std::sync::atomic::Ordering::SeqCst);
220 89340 : }
221 : }
222 :
223 : /// One group waiting on its subscription's lane, holding its batch pending
224 : /// and its bytes on the lanes' ledger.
225 : #[cfg(not(target_arch = "wasm32"))]
226 : pub(super) struct LaneJob {
227 : pub(super) st: AppState,
228 : pub(super) group: Matched,
229 : pub(super) _batch: Arc<BatchDone>,
230 : pub(super) _held: Held,
231 : }
232 :
233 : /// The estimated heap bytes every delivery lane holds, against
234 : /// [`crate::bounds::DELIVERY_LANE_BYTES`].
235 : #[cfg(not(target_arch = "wasm32"))]
236 : pub(super) struct Ledger(pub(super) std::sync::atomic::AtomicUsize);
237 :
238 : #[cfg(not(target_arch = "wasm32"))]
239 : pub(super) static LANE_LEDGER: Ledger = Ledger(std::sync::atomic::AtomicUsize::new(0));
240 :
241 : #[cfg(not(target_arch = "wasm32"))]
242 : impl Ledger {
243 : /// `bytes` more on the ledger, or `None` when they would take it past
244 : /// `budget`. A group larger than the whole budget is still taken on an
245 : /// empty ledger, so one big change is delivered rather than never.
246 9075 : pub(super) fn take(&'static self, bytes: usize, budget: usize) -> Option<Held> {
247 : use std::sync::atomic::Ordering;
248 9075 : self.0
249 9075 : .try_update(Ordering::AcqRel, Ordering::Acquire, |used| {
250 9075 : let after = used.saturating_add(bytes);
251 9075 : (used == 0 || after <= budget).then_some(after)
252 9075 : })
253 9075 : .ok()?;
254 9067 : metrics::gauge!("antares_notification_lane_bytes").increment(bytes as f64);
255 9067 : Some(Held {
256 9067 : ledger: self,
257 9067 : bytes,
258 9067 : })
259 9075 : }
260 :
261 : #[cfg(test)]
262 12 : pub(super) fn used(&self) -> usize {
263 12 : self.0.load(std::sync::atomic::Ordering::Acquire)
264 12 : }
265 : }
266 :
267 : /// Bytes a waiting group holds on the [`Ledger`], given back when the group
268 : /// is delivered or dropped.
269 : #[cfg(not(target_arch = "wasm32"))]
270 : pub(super) struct Held {
271 : pub(super) ledger: &'static Ledger,
272 : pub(super) bytes: usize,
273 : }
274 :
275 : #[cfg(not(target_arch = "wasm32"))]
276 : impl Drop for Held {
277 9067 : fn drop(&mut self) {
278 9067 : self.ledger
279 9067 : .0
280 9067 : .fetch_sub(self.bytes, std::sync::atomic::Ordering::AcqRel);
281 9067 : metrics::gauge!("antares_notification_lane_bytes").decrement(self.bytes as f64);
282 9067 : }
283 : }
284 :
285 : /// What `v` costs on the heap, estimated from how serde_json lays it out:
286 : /// an object is a `BTreeMap`, whose every leaf node is about 640 bytes
287 : /// however few of its eleven slots it fills, so the two-member `{"type",
288 : /// "value"}` of an attribute instance costs a whole node (counting
289 : /// a slot per member read 268 MB of lanes that held 620 MB); an array is its
290 : /// capacity of slots; a string its bytes, rounded to the allocator's
291 : /// 16-byte classes. Cheap next to the serialization a delivery does.
292 : #[cfg(not(target_arch = "wasm32"))]
293 18134 : pub(super) fn heap_estimate(v: &Value) -> usize {
294 18134 : std::mem::size_of::<Value>().saturating_add(owned_heap(v))
295 18134 : }
296 :
297 : /// The heap `v` owns beyond its own slot.
298 : #[cfg(not(target_arch = "wasm32"))]
299 113704 : pub(super) fn owned_heap(v: &Value) -> usize {
300 : const LEAF: usize = 640;
301 : const SLOT: usize = std::mem::size_of::<Value>();
302 152832 : let text = |len: usize| len.next_multiple_of(16);
303 113704 : match v {
304 57742 : Value::String(s) => text(s.len()),
305 224 : Value::Array(items) => items
306 224 : .iter()
307 480 : .fold(items.capacity().saturating_mul(SLOT), |n, item| {
308 480 : n.saturating_add(owned_heap(item))
309 480 : }),
310 : // nodes fill to about two thirds once a map has more than one
311 37648 : Value::Object(map) => map.iter().fold(
312 37648 : LEAF.saturating_mul(map.len().div_ceil(8).max(1)),
313 95090 : |n, (k, item)| {
314 95090 : n.saturating_add(text(k.len()))
315 95090 : .saturating_add(owned_heap(item))
316 95090 : },
317 : ),
318 18090 : _ => 0,
319 : }
320 113704 : }
321 :
322 : /// (broker, tenant, subscription id) → the lane its groups wait on. An
323 : /// entry lives while its lane has work: the lane's task removes it when it
324 : /// finds the lane empty, under this lock, so a group is never sent to a lane
325 : /// nobody serves. The broker part keeps two brokers of one process (tests
326 : /// run one each, with the same tenants and subscription ids) on lanes of
327 : /// their own, each served on its own runtime.
328 : #[cfg(not(target_arch = "wasm32"))]
329 : pub(super) type Lanes = std::sync::Mutex<HashMap<LaneKey, tokio::sync::mpsc::Sender<LaneJob>>>;
330 :
331 : /// A lane's key: [`AppState::broker`], tenant, subscription id.
332 : #[cfg(not(target_arch = "wasm32"))]
333 : pub(super) type LaneKey = (usize, String, String);
334 : #[cfg(not(target_arch = "wasm32"))]
335 : pub(super) static LANES: std::sync::LazyLock<Lanes> = std::sync::LazyLock::new(Lanes::default);
336 :
337 : /// Hand one group to its subscription's lane, opening the lane when there is
338 : /// none. A full lane drops the group for that subscription alone, counted
339 : /// and named: the drain never waits for a subscriber.
340 : #[cfg(not(target_arch = "wasm32"))]
341 9015 : pub(super) fn to_lane(st: &AppState, group: Matched, batch: &Arc<BatchDone>) {
342 : use tokio::sync::mpsc::error::TrySendError;
343 9015 : let key = (
344 9015 : st.broker(),
345 9015 : group.tenant.as_str().to_owned(),
346 9015 : group
347 9015 : .sub
348 9015 : .get("id")
349 9015 : .and_then(Value::as_str)
350 9015 : .unwrap_or_default()
351 9015 : .to_owned(),
352 9015 : );
353 9015 : let bytes = group
354 9015 : .data
355 9015 : .iter()
356 9064 : .fold(0usize, |n, entity| n.saturating_add(heap_estimate(entity)));
357 9015 : let Some(held) = LANE_LEDGER.take(bytes, *crate::bounds::DELIVERY_LANE_BYTES) else {
358 : // every lane together holds its budget: this subscription loses the
359 : // group, counted and named like a full lane's, and the process keeps
360 : // its memory
361 0 : note_lane_drop(&key, group.data.len());
362 0 : tell_lost(st, &group);
363 0 : return;
364 : };
365 9015 : let job = LaneJob {
366 9015 : st: st.clone(),
367 9015 : group,
368 9015 : _batch: batch.clone(),
369 9015 : _held: held,
370 9015 : };
371 : // A poisoned map is a panic in some past hand-off; what it holds is
372 : // senders, rebuilt on the next group either way.
373 9015 : let mut lanes = LANES.lock().unwrap_or_else(|p| p.into_inner());
374 9015 : let job = match lanes.get(&key) {
375 4734 : Some(lane) => match lane.try_send(job) {
376 897 : Ok(()) => return,
377 3837 : Err(TrySendError::Full(job)) => {
378 3837 : note_lane_drop(&key, job.group.data.len());
379 3837 : tell_lost(st, &job.group);
380 3837 : return;
381 : }
382 : // its task is gone (a panic past the guard): open a new lane
383 0 : Err(TrySendError::Closed(job)) => job,
384 : },
385 4281 : None => job,
386 : };
387 4281 : let (lane, rx) = tokio::sync::mpsc::channel(DELIVERY_LANE);
388 4281 : if lane.try_send(job).is_err() {
389 : // a fresh channel always has room; nothing to serve otherwise
390 0 : return;
391 4281 : }
392 4281 : lanes.insert(key.clone(), lane);
393 4281 : drop(lanes);
394 4281 : tokio::spawn(run_lane(key, rx));
395 9015 : }
396 :
397 : /// Serve one subscription's lane in order until it is empty, then retire it.
398 : #[cfg(not(target_arch = "wasm32"))]
399 4281 : pub(super) async fn run_lane(key: LaneKey, mut rx: tokio::sync::mpsc::Receiver<LaneJob>) {
400 : // a group the last fold had no room for leads the next notification
401 4277 : let mut carry = None;
402 : loop {
403 8753 : let job = match carry.take().map_or_else(|| rx.try_recv().ok(), Some) {
404 4523 : Some(job) => job,
405 : None => {
406 : // Looked at again under the map's lock, which every sender
407 : // holds while it sends: a group cannot arrive between this
408 : // look and the removal.
409 4230 : let mut lanes = LANES.lock().unwrap_or_else(|p| p.into_inner());
410 4230 : match rx.try_recv() {
411 4 : Ok(job) => job,
412 : Err(_) => {
413 4226 : lanes.remove(&key);
414 4226 : return;
415 : }
416 : }
417 : }
418 : };
419 4527 : let (st, group, batches, rest) = fold(job, &mut rx);
420 4527 : carry = rest;
421 4527 : guarded(async move { deliver_group(&st, group).await }).await;
422 4476 : drop(batches);
423 : }
424 4226 : }
425 :
426 : /// A fold: the state, the merged group, the batches and ledger bytes it
427 : /// holds until delivered, and the job it had no room for.
428 : #[cfg(not(target_arch = "wasm32"))]
429 : pub(super) type Folded = (
430 : AppState,
431 : Matched,
432 : Vec<(Arc<BatchDone>, Held)>,
433 : Option<LaneJob>,
434 : );
435 :
436 : /// What waits behind `job` on its lane leaves in the same notification, as
437 : /// a drain folds one subscription's changes into one group: a lane keeps up
438 : /// by sending fewer, larger notifications, not by queueing one per change.
439 : /// The fold stops before a group that would take the notification past
440 : /// [`CHANGE_BATCH`] entities and hands that group back, so a backed-up lane
441 : /// sends several bounded POSTs, never its whole depth in one. The newest
442 : /// group carries the subscription as it reads now; the batches stay pending
443 : /// until the delivery ends. A group of another state (a second broker in
444 : /// the process, with the same tenant and subscription id) leads the next
445 : /// notification: each leaves through its own state's sinks.
446 : #[cfg(not(target_arch = "wasm32"))]
447 4555 : pub(super) fn fold(job: LaneJob, rx: &mut tokio::sync::mpsc::Receiver<LaneJob>) -> Folded {
448 : let LaneJob {
449 4555 : st,
450 4555 : mut group,
451 4555 : _batch,
452 4555 : _held,
453 4555 : } = job;
454 4555 : let mut batches = vec![(_batch, _held)];
455 4704 : while let Ok(next) = rx.try_recv() {
456 169 : if group.data.len().saturating_add(next.group.data.len()) > CHANGE_BATCH
457 157 : || !next.st.same(&st)
458 : {
459 20 : return (st, group, batches, Some(next));
460 149 : }
461 149 : let mut data = std::mem::take(&mut group.data);
462 149 : data.extend(next.group.data);
463 149 : group = Matched { data, ..next.group };
464 149 : batches.push((next._batch, next._held));
465 : }
466 4535 : (st, group, batches, None)
467 4555 : }
468 :
469 : /// A group this process dropped, told to the subscription's binding, which
470 : /// says so in band where it can (a socket's `x-gap`).
471 : #[cfg(not(target_arch = "wasm32"))]
472 3837 : pub(super) fn tell_lost(st: &AppState, g: &Matched) {
473 3837 : if let (Some(uri), Some(id)) = (
474 3837 : g.sub.as_object().and_then(super::endpoint_uri),
475 3837 : g.sub.get("id").and_then(Value::as_str),
476 : ) {
477 3837 : if let Some(sink) = st.sinks.sink_for_uri(uri) {
478 3837 : sink.lost(uri, g.tenant.as_str(), id);
479 3837 : }
480 0 : }
481 3837 : }
482 :
483 : /// Groups a full lane refused: counted by entity, and named by tenant and
484 : /// subscription in the log, so an operator can tell which subscriber fell
485 : /// behind and lost changes.
486 : #[cfg(not(target_arch = "wasm32"))]
487 3837 : pub(super) fn note_lane_drop(key: &LaneKey, entities: usize) {
488 3837 : let n = LANE_DROPS
489 3837 : .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
490 3837 : .saturating_add(1);
491 3837 : metrics::counter!("antares_notification_lane_dropped_total").increment(entities as u64);
492 3837 : if n == 1 || n.is_multiple_of(100) {
493 40 : tracing::warn!(
494 : "subscription {} of tenant {} is slower than its changes: {entities} entities dropped from its delivery lane ({n} groups so far, all subscriptions)",
495 : key.2,
496 : key.1
497 : );
498 3797 : }
499 3837 : }
500 :
501 : #[cfg(not(target_arch = "wasm32"))]
502 : pub(super) static LANE_DROPS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
503 :
504 : /// How long one step of a drain may take: matching one change, waiting for
505 : /// one group's delivery permits, or that delivery. A drain waits for all of
506 : /// its groups and the next drain for it, so a step that never ends stops
507 : /// notifications for every Subscription of the process. A send is
508 : /// at most 30 s (`endpoint_timeout_ms`), so a live step stays well inside.
509 : #[cfg(not(target_arch = "wasm32"))]
510 : pub(super) const STEP_DEADLINE: std::time::Duration = std::time::Duration::from_secs(60);
511 :
512 : /// A step abandoned at [`STEP_DEADLINE`]: counted and named, so the wait
513 : /// that never ended is in the log.
514 : #[cfg(not(target_arch = "wasm32"))]
515 0 : pub(super) fn note_stall(step: &'static str, tenant: &str, sub: &str) {
516 0 : metrics::counter!("antares_notification_stalls_total", "step" => step).increment(1);
517 0 : tracing::warn!(
518 : "notification step {step} abandoned after {}s (tenant {tenant}, subscription {sub}); the drain goes on",
519 0 : STEP_DEADLINE.as_secs()
520 : );
521 0 : }
522 :
523 : /// A change the full queue refused: counted, and said once per thousand so
524 : /// a 40 % delivery gap is visible in the log and not only on the counter.
525 0 : pub(super) fn note_drop() {
526 0 : let n = CHANGES_DROPPED
527 0 : .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
528 0 : .saturating_add(1);
529 0 : metrics::counter!("antares_notification_changes_dropped_total").increment(1);
530 0 : if n == 1 || n.is_multiple_of(1000) {
531 0 : tracing::warn!("notification change queue full: {n} changes dropped so far (delivery slower than the write rate)");
532 0 : }
533 0 : }
534 :
535 : #[cfg(not(target_arch = "wasm32"))]
536 : pub(super) static DELIVERY_SLOTS: std::sync::LazyLock<tokio::sync::Semaphore> =
537 16 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*DELIVERY_WIDTH));
538 :
539 : #[cfg(not(target_arch = "wasm32"))]
540 : pub(super) type TenantSlots =
541 : std::sync::Mutex<std::collections::HashMap<String, Arc<tokio::sync::Semaphore>>>;
542 : #[cfg(not(target_arch = "wasm32"))]
543 : pub(super) static TENANT_SLOTS: std::sync::LazyLock<TenantSlots> =
544 : std::sync::LazyLock::new(TenantSlots::default);
545 :
546 : /// The tenant's share, created on first use. An entry lives only while a
547 : /// delivery is holding it: a lookup that has to create one first drops every
548 : /// entry nothing else still references, so this map is bounded by the
549 : /// delivery width and not by how many of the broker's 10 000 tenants have
550 : /// ever had a subscription fire.
551 : #[cfg(not(target_arch = "wasm32"))]
552 4741 : pub(super) fn tenant_slots(tenant: &TenantId) -> Arc<tokio::sync::Semaphore> {
553 : // A poisoned map is a panic in some past delivery, not a reason to stop
554 : // delivering: what it holds is rebuildable and bounded either way.
555 4741 : let mut map = TENANT_SLOTS.lock().unwrap_or_else(|p| p.into_inner());
556 4741 : if let Some(s) = map.get(tenant.as_str()) {
557 4673 : return Arc::clone(s);
558 68 : }
559 68 : map.retain(|_, s| Arc::strong_count(s) > 1);
560 68 : let slots = Arc::new(tokio::sync::Semaphore::new(*DELIVERY_WIDTH_PER_TENANT));
561 68 : map.insert(tenant.as_str().to_owned(), Arc::clone(&slots));
562 68 : slots
563 4741 : }
564 :
565 : /// Both permits one delivery holds until it settles. The tenant's share is
566 : /// taken first, so a tenant already at its share waits there instead of
567 : /// waiting inside one of the broker's slots.
568 : #[cfg(not(target_arch = "wasm32"))]
569 4741 : pub(super) async fn delivery_permits(tenant: &TenantId) -> impl Sized {
570 4741 : let share = tenant_slots(tenant).acquire_owned().await;
571 4729 : (share, DELIVERY_SLOTS.acquire().await)
572 4729 : }
573 :
574 : /// Whether the Subscription's notifications stay in this process: its
575 : /// binding opens no socket (an in-process one) or the forward transport takes them.
576 : /// The same test `deliver_as` polices egress by.
577 : #[cfg(not(target_arch = "wasm32"))]
578 4661 : pub(super) fn in_process_sink(st: &AppState, sub: &Value) -> bool {
579 4661 : sub.as_object()
580 4661 : .and_then(super::endpoint_uri)
581 4661 : .is_some_and(|uri| {
582 4661 : in_process(st, uri).is_some()
583 4655 : || st.sinks.sink_for_uri(uri).is_some_and(|s| !s.network())
584 4661 : })
585 4661 : }
|
