Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Periodic (timeInterval) subscriptions: due times, the cross-replica claim and the sweep tick.
3 :
4 : use super::*;
5 :
6 : /// When an interval subscription is next due, in epoch millis: one
7 : /// `timeInterval` after the last Notification it sent (Table 5.2.14.2-1
8 : /// `lastNotification`), or after its creation while it has sent none. Without
9 : /// either anchor it is due immediately.
10 222 : pub(super) fn due_at_ms(sub: &Value, interval: f64) -> i64 {
11 222 : let anchor = sub
12 222 : .get("notification")
13 222 : .and_then(|n| n.get("lastNotification"))
14 222 : .and_then(Value::as_str)
15 222 : .or_else(|| sub.get("createdAt").and_then(Value::as_str));
16 222 : match anchor.and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok()) {
17 222 : Some(last) => last
18 222 : .timestamp_millis()
19 222 : .saturating_add(interval_offset_ms(interval)),
20 0 : None => i64::MIN,
21 : }
22 222 : }
23 :
24 : /// One period of a periodic Subscription (5.2.12 `timeInterval`) in
25 : /// milliseconds.
26 : ///
27 : /// Table 5.2.12-1 bounds the member only as greater than 0, so the seconds a
28 : /// client names can exceed what epoch milliseconds hold. The cast saturates,
29 : /// and every caller adds it with `saturating_add`, which puts an interval the
30 : /// broker cannot schedule at the end of representable time. Adding it plainly
31 : /// wraps the sum negative, and a negative firing instant reads as permanently
32 : /// due: the subscription then fires its whole query on every tick, and the
33 : /// same value poisons the process-wide sweep clock those minima feed.
34 418 : pub(super) fn interval_offset_ms(interval: f64) -> i64 {
35 418 : (interval * 1000.0) as i64
36 418 : }
37 :
38 : /// timeInterval subscriptions: fire when due, with all matching entities.
39 : /// Multi-instance: claim one interval firing under the subscription row
40 : /// lock — N matcher pods race, exactly one wins (single-winner by
41 : /// lock, no leader election). The due-check reruns INSIDE the lock; the
42 : /// winner stamps `lastNotification` as its claim, losers see not-due and
43 : /// roll back. Only engaged in bus=nats mode — single-process behaviour (and
44 : /// its 046_12 bookkeeping ordering) is untouched.
45 : ///
46 : /// `None` = the firing is not this pod's. `Some(prev)` = claimed, carrying
47 : /// the `lastNotification` the claim overwrote: a firing that turns out to
48 : /// have nothing to send gives it back through [`release_interval`], because
49 : /// Table 5.2.14.2-1 stamps the instant a notification was SENT and 5.8.6
50 : /// sends none when nothing matches.
51 4 : pub(super) async fn claim_interval(
52 4 : st: &AppState,
53 4 : tenant: &TenantId,
54 4 : kind: Kind,
55 4 : sub: &Value,
56 4 : interval: f64,
57 4 : ) -> Option<Option<Value>> {
58 4 : let id = sub.get("id").and_then(Value::as_str)?;
59 4 : let mut prev: Option<Value> = None;
60 4 : let res = st
61 4 : .store
62 4 : .mutate(tenant, kind, id, |doc| {
63 4 : if chrono::Utc::now().timestamp_millis() < due_at_ms(doc, interval) {
64 0 : return Err(());
65 4 : }
66 4 : let Some(sub_doc) = doc.as_object_mut() else {
67 : // a stored Subscription is an object; one that is not carries no
68 : // notification member to stamp, and Err(()) is the same "nothing
69 : // written" the not-due branch above returns
70 0 : return Err(());
71 : };
72 4 : if let Some(n) = sub_doc
73 4 : .entry("notification")
74 4 : .or_insert_with(|| json!({}))
75 4 : .as_object_mut()
76 4 : {
77 4 : prev = n.insert("lastNotification".into(), Value::String(now_iso()));
78 4 : }
79 4 : Ok(())
80 4 : })
81 4 : .await;
82 4 : matches!(res, Ok(Some(Ok(())))).then_some(prev)
83 4 : }
84 :
85 : /// Give a claimed firing back (5.8.6: nothing matched, so nothing was sent).
86 : /// The stamp returns to what [`claim_interval`] found, which keeps
87 : /// `lastNotification` truthful exactly as the single-process path does; the
88 : /// sweep parks the subscription until its next interval in this pod's mirror.
89 4 : pub(super) async fn release_interval(
90 4 : st: &AppState,
91 4 : tenant: &TenantId,
92 4 : kind: Kind,
93 4 : id: &str,
94 4 : prev: Option<Value>,
95 4 : ) {
96 4 : let res = st
97 4 : .store
98 4 : .mutate::<(), ()>(tenant, kind, id, |doc| {
99 4 : if let Some(n) = doc
100 4 : .as_object_mut()
101 4 : .and_then(|o| o.get_mut("notification"))
102 4 : .and_then(Value::as_object_mut)
103 : {
104 4 : match &prev {
105 0 : Some(v) => n.insert("lastNotification".into(), v.clone()),
106 4 : None => n.remove("lastNotification"),
107 : };
108 0 : }
109 4 : Ok(())
110 4 : })
111 4 : .await;
112 4 : if let Err(e) = res {
113 0 : tracing::warn!("releasing the interval claim for {id} failed: {e}");
114 4 : }
115 4 : }
116 :
117 : /// One sweep of the interval subscriptions (5.8.6, 5.11.7): "If a
118 : /// Subscription defines a timeInterval member, a Notification shall be sent
119 : /// periodically, when the time interval (in seconds) specified in such value
120 : /// field is reached, regardless of Attribute changes."
121 : ///
122 : /// The sweep runs on a fixed tick, so its idle cost is what has to stay
123 : /// small. Two things keep it off the store. A tick that cannot fire anything
124 : /// returns before enumerating tenants: each sweep records the earliest instant
125 : /// a subscription it saw can next be due, and a write zeroes that clock —
126 : /// through the mirror for Subscriptions, through `csub_written` for the
127 : /// Context Source Registration Subscription half, which is mirrored by clock
128 : /// rather than by document. A lost signal is repaired within
129 : /// `CSUB_SWEEP_BACKSTOP_MS`. And a due subscription reads only the Entities
130 : /// its own selector can match instead of its tenant's entity set.
131 9449 : pub async fn interval_tick(st: &AppState) {
132 : use std::sync::atomic::Ordering::Relaxed;
133 9449 : let now_ms = chrono::Utc::now().timestamp_millis();
134 9449 : let clocks = st.sub_mirror.as_ref().map(|m| {
135 9423 : (
136 9423 : m.next_sub_sweep_ms.load(Relaxed),
137 9423 : m.next_csub_sweep_ms.load(Relaxed),
138 9423 : )
139 9423 : });
140 9449 : let (sweep_subs, sweep_csubs) = match clocks {
141 9423 : Some((sub_clock, csub_clock)) => (now_ms >= sub_clock, now_ms >= csub_clock),
142 : // Never-wired fallback: no clock to keep, so every tick sweeps.
143 26 : None => (true, true),
144 : };
145 9449 : if !sweep_subs && !sweep_csubs {
146 8894 : return;
147 555 : }
148 : // Earliest next-due instant seen by this sweep, per half.
149 555 : let mut next_sub = i64::MAX;
150 555 : let mut next_csub = i64::MAX;
151 555 : let tenants = read_or_warn(
152 555 : st.store.subscription_tenants().await,
153 555 : "the tenants with subscriptions",
154 : )
155 555 : .into_iter()
156 16409 : .filter_map(|t| TenantId::new_internal(&t).ok());
157 : // Each Tenant is swept on its own task. A Tenant waits for its own share
158 : // of the delivery width before it reads its next page, so a Tenant whose
159 : // endpoint hangs holds that wait for up to the endpoint's whole timeout:
160 : // swept in turn, every Tenant after it would wait as long, and one
161 : // Tenant would set every other Tenant's periodic deadline.
162 : #[cfg(not(target_arch = "wasm32"))]
163 : {
164 555 : let mut sweeps = tokio::task::JoinSet::new();
165 16403 : let mut fold = |joined: Option<Result<(i64, i64), tokio::task::JoinError>>| match joined {
166 16356 : Some(Ok((sub, csub))) => {
167 16356 : next_sub = next_sub.min(sub);
168 16356 : next_csub = next_csub.min(csub);
169 16356 : }
170 : // a sweep that died decided nothing about its Tenant: the next
171 : // tick sweeps again
172 0 : Some(Err(_)) => {
173 0 : note_panic();
174 0 : next_sub = next_sub.min(now_ms);
175 0 : next_csub = next_csub.min(now_ms);
176 0 : }
177 0 : None => {}
178 16356 : };
179 16409 : for tenant in tenants {
180 : // ponytail: at most tenant_sweep_width() Tenants are swept at once
181 16368 : if sweeps.len() >= tenant_sweep_width() {
182 15134 : fold(sweeps.join_next().await);
183 1234 : }
184 16368 : let st = st.clone();
185 16368 : sweeps.spawn(async move {
186 16368 : sweep_tenant(&st, &tenant, now_ms, sweep_subs, sweep_csubs).await
187 16356 : });
188 : }
189 1777 : while let Some(joined) = sweeps.join_next().await {
190 1222 : fold(Some(joined));
191 1222 : }
192 : }
193 : #[cfg(target_arch = "wasm32")]
194 : for tenant in tenants {
195 : let (sub, csub) = sweep_tenant(st, &tenant, now_ms, sweep_subs, sweep_csubs).await;
196 : next_sub = next_sub.min(sub);
197 : next_csub = next_csub.min(csub);
198 : }
199 543 : if let (Some(m), Some((sub_clock, csub_clock))) = (&st.sub_mirror, clocks) {
200 : // A periodic subscription written DURING the sweep has already zeroed
201 : // the clock: the exchange then fails, the zero stands and the next
202 : // tick sweeps rather than waiting out an interval computed without it.
203 529 : if sweep_subs {
204 361 : let _ = m
205 361 : .next_sub_sweep_ms
206 361 : .compare_exchange(sub_clock, next_sub, Relaxed, Relaxed);
207 392 : }
208 529 : if sweep_csubs {
209 459 : let _ = m.next_csub_sweep_ms.compare_exchange(
210 459 : csub_clock,
211 459 : next_csub.min(now_ms.saturating_add(CSUB_SWEEP_BACKSTOP_MS)),
212 459 : Relaxed,
213 459 : Relaxed,
214 459 : );
215 459 : }
216 14 : }
217 9437 : }
218 :
219 : /// How many Tenants one interval sweep visits at once. Each one reads the
220 : /// store while it runs, so the width takes at most half the connection
221 : /// pool (`ANTARES_PG_POOL`, default 20, which the broker has validated) and
222 : /// never more than 8: the request path keeps the other half. A Tenant
223 : /// waiting for its delivery share holds no connection.
224 : #[cfg(not(target_arch = "wasm32"))]
225 16368 : fn tenant_sweep_width() -> usize {
226 : static WIDTH: std::sync::OnceLock<usize> = std::sync::OnceLock::new();
227 16368 : *WIDTH.get_or_init(|| sweep_width_for(std::env::var("ANTARES_PG_POOL").ok().as_deref()))
228 16368 : }
229 :
230 : #[cfg(not(target_arch = "wasm32"))]
231 44 : fn sweep_width_for(pool: Option<&str>) -> usize {
232 44 : let pool = pool.and_then(|v| v.parse::<usize>().ok()).unwrap_or(20);
233 44 : (pool / 2).clamp(1, 8)
234 44 : }
235 :
236 : /// One Tenant's half of [`interval_tick`]: fires its due periodic
237 : /// Subscriptions and Context Source Registration Subscriptions and answers
238 : /// the earliest instant each half can next be due. It returns when its
239 : /// deliveries have settled, so ticks never overlap and a subscription cannot
240 : /// be fired twice for one period.
241 16368 : async fn sweep_tenant(
242 16368 : st: &AppState,
243 16368 : tenant: &TenantId,
244 16368 : now_ms: i64,
245 16368 : sweep_subs: bool,
246 16368 : sweep_csubs: bool,
247 16368 : ) -> (i64, i64) {
248 16368 : let tenant = tenant.clone();
249 16368 : let mut next_sub = i64::MAX;
250 16368 : let mut next_csub = i64::MAX;
251 : // A delivery costs up to the endpoint's whole timeout (Table 5.2.15-1,
252 : // 30 s at the ceiling). Awaited in turn, one unresponsive endpoint becomes
253 : // the deadline of every other subscriber's periodic notification. The
254 : // deliveries therefore run together, under the Tenant's share of the
255 : // width the change path uses.
256 : #[cfg(not(target_arch = "wasm32"))]
257 16368 : let mut sending = tokio::task::JoinSet::new();
258 : // Same source the matcher reads: the indexed mirror, with the store
259 : // list only as the never-wired fallback.
260 16368 : let subs = match (&st.sub_mirror, sweep_subs) {
261 11355 : (_, false) => Vec::new(),
262 4983 : (Some(m), _) => m.periodic_docs(tenant.as_str()),
263 30 : (None, _) => read_or_warn(
264 30 : st.store.list(&tenant, Kind::Subscription).await,
265 30 : "the periodic Subscriptions",
266 : )
267 30 : .into_iter()
268 30 : .map(Arc::new)
269 30 : .collect(),
270 : };
271 16368 : for sub in subs {
272 166 : let Some(interval) = sub.get("timeInterval").and_then(Value::as_f64) else {
273 0 : continue;
274 : };
275 166 : if !is_active(&sub) {
276 4 : continue;
277 162 : }
278 162 : let sub_id = sub.get("id").and_then(Value::as_str).unwrap_or_default();
279 : // a firing that had nothing to send ends that interval (below)
280 162 : let quiet = st
281 162 : .sub_mirror
282 162 : .as_ref()
283 162 : .and_then(|m| m.quiet_until(tenant.as_str(), sub_id));
284 162 : let due_at = due_at_ms(&sub, interval).max(quiet.unwrap_or(i64::MIN));
285 162 : if now_ms < due_at {
286 22 : next_sub = next_sub.min(due_at);
287 22 : continue;
288 140 : }
289 : // Due: the following firing is one interval away. Recorded before
290 : // the claim, so a pod that LOSES the race (another one is firing
291 : // this subscription right now) keeps sweeping on the interval
292 : // instead of parking on an anchor only the winner advanced.
293 140 : next_sub = next_sub.min(now_ms.saturating_add(interval_offset_ms(interval)));
294 : // a stored q that no longer parses (one percent-encoded before
295 : // 5.2.12's read-as-written) matches nothing, as conditions_match
296 : // decides: no Entity to read and no Notification to send, so
297 : // nothing to claim either (the claim stamped
298 : // lastNotification for a firing that never ran)
299 140 : if sub_str(&sub, "q").is_some_and(|q| antares_ql::regex::q_node(q).is_none()) {
300 4 : continue;
301 136 : }
302 136 : let claim = if st.nats {
303 4 : match claim_interval(st, &tenant, Kind::Subscription, &sub, interval).await {
304 4 : Some(prev) => Some(prev),
305 0 : None => continue,
306 : }
307 : } else {
308 132 : None
309 : };
310 136 : let ctx = sub_context(st, &tenant, &sub).await;
311 136 : let now = now_iso();
312 : // 5.8.6: the periodic Notification "shall include all the
313 : // subscribed Entities that match the query, geoquery and Scope
314 : // query conditions" — so the read is exactly this subscription's
315 : // own selector (5.2.33) and conditions, never the tenant's entity
316 : // set. Only predicates a store reproduces without hiding a
317 : // candidate are offered (ids when every selector entry names one,
318 : // types when the index proves them plain, q/scopeQ/geoQ under the
319 : // store's own rule that SQL removes rows and never decides them);
320 : // the selector_match/conditions_match pair below stays the
321 : // arbiter, exactly as on the query path.
322 136 : let type_groups: Vec<Vec<String>> = match index_keys(&sub) {
323 136 : Keys::Types(ts) => ts.into_iter().map(|t| vec![t]).collect(),
324 0 : _ => Vec::new(),
325 : };
326 136 : let ids = selector_ids(&sub);
327 : // The filter borrows a term expander that is not Sync, so it lives
328 : // and dies inside this block: held across the delivery await it
329 : // would make the whole interval task non-Send.
330 : // the match set is read a page at a time and each
331 : // page's matches leave as their own Notification, so a type-only
332 : // selector over a large tenant never holds its whole set; a store
333 : // that cannot decide the filter answers it whole, once.
334 136 : let page_size = st.max_limit.max(1);
335 136 : let mut offset = 0usize;
336 136 : let mut sent = false;
337 : loop {
338 184 : let (rows, last) = {
339 184 : let expand = |t: &str| ctx.expand_key(t);
340 184 : let id_refs: Vec<&str> = ids.iter().flatten().map(String::as_str).collect();
341 : // q is read as written (5.2.12); parses shared per distinct
342 : // expression text, as in conditions_match — the sweep
343 : // re-runs per due subscription
344 184 : let q_ast = sub_str(&sub, "q").and_then(antares_ql::regex::q_node);
345 184 : let geo = sub.get("geoQ").and_then(Value::as_object).and_then(|g| {
346 0 : let key = serde_json::to_string(g).unwrap_or_default();
347 0 : antares_ql::regex::geo_query(&key, || {
348 0 : antares_ql::geo::GeoQuery::from_params(&geo_params(g))
349 0 : .ok()
350 0 : .flatten()
351 0 : })
352 0 : });
353 184 : let geo_spec = geo.as_ref().and_then(|g| g.to_sql_spec(&ctx));
354 184 : let filter = antares_store::filter::EntityFilter {
355 184 : ids: ids.as_ref().map(|_| id_refs.as_slice()),
356 184 : types: (!type_groups.is_empty()).then_some(type_groups.as_slice()),
357 184 : q: q_ast.as_deref(),
358 184 : scope_q: sub_str(&sub, "scopeQ"),
359 184 : geo: geo_spec.as_slice(),
360 184 : expand: &expand,
361 184 : page: Some(antares_store::filter::Page {
362 184 : offset: i64::try_from(offset).unwrap_or(i64::MAX),
363 184 : limit: i64::try_from(page_size).unwrap_or(i64::MAX),
364 184 : count: false,
365 184 : }),
366 184 : ..Default::default()
367 : };
368 184 : match st.store.query_entities(&tenant, &filter).await {
369 184 : Ok(o) => {
370 184 : let last = !o.paged || o.rows.len() < page_size;
371 184 : (o.rows, last)
372 : }
373 0 : Err(e) => {
374 0 : tracing::warn!("notification path: reading the Entities a periodic Subscription notifies about failed: {e}");
375 0 : (Vec::new(), true)
376 : }
377 : }
378 : };
379 184 : offset = offset.saturating_add(rows.len());
380 184 : let mut matching: Vec<Value> = Vec::new();
381 196 : for d in rows {
382 196 : if !selector_match(&sub, &d, &ctx)
383 196 : || !linked_eval(st, &tenant, None, |l| conditions_match(&sub, &d, &ctx, l))
384 196 : .await
385 : {
386 0 : continue;
387 196 : }
388 196 : matching.extend(
389 196 : build_data(st, &tenant, &sub, &ctx, None, Some(&d), &[], false, &now).await,
390 : );
391 : }
392 184 : if !matching.is_empty() {
393 160 : sent = true;
394 : #[cfg(not(target_arch = "wasm32"))]
395 : {
396 : // The slot is taken BEFORE the next page is read: a
397 : // page waiting for one is a page held in memory, and
398 : // against a slow endpoint the read would run ahead to
399 : // the end of the match set.
400 160 : let permits = delivery_permits(&tenant).await;
401 152 : let (st, tenant, ctx, sub) =
402 152 : (st.clone(), tenant.clone(), Arc::clone(&ctx), sub.clone());
403 152 : sending.spawn(async move {
404 152 : let _permits = permits;
405 152 : deliver(&st, &tenant, &sub, matching, &ctx).await;
406 88 : });
407 : }
408 : #[cfg(target_arch = "wasm32")]
409 : deliver(st, &tenant, &sub, matching, &ctx).await;
410 24 : }
411 176 : if last {
412 128 : break;
413 48 : }
414 : }
415 128 : if !sent {
416 : // 5.8.6: "If there are no matching Entities, no Notification
417 : // is sent" — lastNotification stays untouched, and a claim
418 : // taken to win the firing is given back here, or the
419 : // multi-pod path would stamp an instant nothing was sent at.
420 : // The firing still happened: the next one is when the next
421 : // interval is reached, so the subscription parks in the
422 : // mirror until then instead of re-running its query on every
423 : // tick (the clock above already points there). A write to it
424 : // clears the park.
425 16 : if let Some(prev) = claim {
426 4 : release_interval(st, &tenant, Kind::Subscription, sub_id, prev).await;
427 12 : }
428 16 : match &st.sub_mirror {
429 8 : Some(m) => m.park_quiet(
430 8 : tenant.as_str(),
431 8 : sub_id,
432 8 : &sub,
433 8 : now_ms.saturating_add(interval_offset_ms(interval)),
434 8 : ),
435 8 : None => next_sub = next_sub.min(due_at),
436 : }
437 112 : }
438 : }
439 : // csource timeInterval subs: periodic CSourceNotification with all
440 : // matching registrations, independent of changes (5.11.7). Skipped by
441 : // an empty list, not a return: the deliveries spawned above must settle
442 : // below (a dropped JoinSet aborts them).
443 16360 : let csubs = if sweep_csubs {
444 12213 : read_or_warn(
445 12318 : docs_where(st, &tenant, Kind::CSourceSubscription, |s| {
446 379 : s.get("timeInterval").is_some()
447 379 : })
448 12213 : .await,
449 12213 : "the periodic Context Source Registration Subscriptions",
450 : )
451 : } else {
452 4147 : Vec::new()
453 : };
454 16360 : for sub in csubs {
455 40 : let Some(interval) = sub.get("timeInterval").and_then(Value::as_f64) else {
456 0 : continue;
457 : };
458 40 : if !is_active(&sub) {
459 0 : continue;
460 40 : }
461 40 : let due_at = due_at_ms(&sub, interval);
462 40 : if now_ms < due_at {
463 4 : next_csub = next_csub.min(due_at);
464 4 : continue;
465 36 : }
466 36 : next_csub = next_csub.min(now_ms.saturating_add(interval_offset_ms(interval)));
467 : // 5.11.7 sends the periodic CSourceNotification whatever the
468 : // matching set is, so this claim is never given back.
469 36 : if st.nats
470 0 : && claim_interval(st, &tenant, Kind::CSourceSubscription, &sub, interval)
471 0 : .await
472 0 : .is_none()
473 : {
474 0 : continue;
475 36 : }
476 36 : let ctx = sub_context(st, &tenant, &sub).await;
477 36 : let spec = crate::registry::spec_for_subscription(&sub);
478 36 : let data: Vec<Value> = read_or_warn(
479 36 : docs_where(st, &tenant, Kind::Registration, |r| {
480 36 : crate::registry::csr_matches_subscription(&sub, r, &ctx)
481 36 : })
482 36 : .await,
483 36 : "the registrations a periodic Context Source Notification carries",
484 : )
485 36 : .into_iter()
486 36 : .map(|r| {
487 36 : let mut p =
488 36 : crate::registry::present_registration(&filter_csr(&spec, &r, &ctx), &ctx, false);
489 36 : arrayify_entity_types(&mut p);
490 36 : p
491 36 : })
492 36 : .collect();
493 : #[cfg(not(target_arch = "wasm32"))]
494 : {
495 : // the Tenant's share and a slot, as for Subscriptions (a
496 : // bare slot let one Tenant hold them all), taken before the
497 : // next subscription's registrations are read
498 36 : let permits = delivery_permits(&tenant).await;
499 32 : let (st, tenant, ctx) = (st.clone(), tenant.clone(), Arc::clone(&ctx));
500 32 : sending.spawn(async move {
501 32 : let _permits = permits;
502 32 : deliver_csource(&st, &tenant, &sub, data, &ctx, "newlyMatching").await;
503 0 : });
504 : }
505 : #[cfg(target_arch = "wasm32")]
506 : deliver_csource(st, &tenant, &sub, data, &ctx, "newlyMatching").await;
507 : }
508 : #[cfg(not(target_arch = "wasm32"))]
509 16444 : while let Some(joined) = sending.join_next().await {
510 88 : if joined.is_err() {
511 0 : note_panic();
512 88 : }
513 : }
514 16356 : (next_sub, next_csub)
515 16356 : }
516 :
517 : #[cfg(all(test, not(target_arch = "wasm32")))]
518 : mod tests {
519 : /// SECV-03: the sweep never takes more than half a small pool, and
520 : /// keeps 8 on the default one.
521 : #[test]
522 4 : fn the_sweep_width_leaves_the_request_path_half_the_pool() {
523 4 : assert_eq!(super::sweep_width_for(None), 8);
524 4 : assert_eq!(super::sweep_width_for(Some("20")), 8);
525 4 : assert_eq!(super::sweep_width_for(Some("8")), 4);
526 4 : assert_eq!(super::sweep_width_for(Some("2")), 1);
527 4 : assert_eq!(super::sweep_width_for(Some("1")), 1);
528 4 : }
529 : }
|
