Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Input bounds wall: every request-shaped resource has
3 : //! a configured cap, rejected with the spec-shaped error. One middleware
4 : //! enforces the transport-level caps (URI length 414, body size 413, JSON
5 : //! nesting 400) — size and depth are checked BEFORE any parse. The
6 : //! per-feature caps (batch count, joinLevel, @context fetch
7 : //! count, q= complexity, result ceiling) live at their parse points.
8 : //! Rejections are counted and exported via /q/health.
9 :
10 : use axum::body::{Body, Bytes};
11 : use axum::http::{Request, StatusCode};
12 : use axum::middleware::Next;
13 : use axum::response::{IntoResponse, Response};
14 : use std::sync::atomic::{AtomicU64, Ordering};
15 :
16 : /// Hard caps (v1: compile-time constants — a config file is a later knob;
17 : /// every value is spec-shaped on rejection).
18 : /// → bare 413 (6.3.4). Deployment knob (ANTARES_MAX_BODY_BYTES): the spec
19 : /// names no ceiling; 4 MiB is the DoS bound, raised where a trusted
20 : /// producer legitimately sends bigger batches. Read once at first use.
21 70 : pub static MAX_BODY_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
22 70 : std::env::var("ANTARES_MAX_BODY_BYTES")
23 70 : .ok()
24 70 : .and_then(|v| v.parse().ok())
25 70 : .filter(|n| *n > 0)
26 70 : .unwrap_or(4 * 1024 * 1024)
27 70 : });
28 : pub const MAX_URI_BYTES: usize = 8 * 1024; // → bare 414
29 : pub const MAX_JSON_DEPTH: usize = 64; // → 400 BadRequestData
30 : /// → 403 TooManyResults (5.5.6) on the NGSI-LD API, 413 on a surface
31 : /// (`antares_ops::Refusal::Limit`). The most bytes of one in-process answer
32 : /// a handler buffers before reading it (a temporal candidate set, a
33 : /// snapshot fill page, a surface's query): the page size bounds the Entity
34 : /// count, not the bytes, and every concurrent request holds its buffer.
35 : /// `AppState::max_answer_bytes` carries it, so a test can lower it.
36 : pub const MAX_ANSWER_BYTES: usize = 64 * 1024 * 1024;
37 : /// The fixed detail of an answer past [`MAX_ANSWER_BYTES`].
38 : pub const ANSWER_TOO_LARGE: &str = "the answer is larger than the broker buffers for one request";
39 : /// → 400 BadRequestData. Maximum coordinate positions in a QUERY geometry
40 : /// (4.10 geoQ, 4.23 ordering reference). The spec sets no ceiling, and the
41 : /// geometry is not bounded by the URI length on the POST query path — the
42 : /// body carries it. Every position is an edge the DE-9IM relate walks once
43 : /// per candidate entity, so the work a single request can buy is capped
44 : /// here: 1024 positions describe an administrative boundary at street
45 : /// resolution, and are already more than the 8 KiB URI ceiling can carry.
46 : pub use antares_ql::geo::MAX_GEO_VERTICES;
47 : /// → 400 BadRequestData. Deployment knob (ANTARES_MAX_BATCH_ITEMS): the
48 : /// spec sets no batch ceiling — 1000 is this broker's DoS-bounds default,
49 : /// raised where a trusted producer legitimately batches larger (e.g. a
50 : /// full-fleet upsert). Read once at first use.
51 38 : pub static MAX_BATCH_ITEMS: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
52 38 : std::env::var("ANTARES_MAX_BATCH_ITEMS")
53 38 : .ok()
54 38 : .and_then(|v| v.parse().ok())
55 38 : .filter(|n| *n > 0)
56 38 : .unwrap_or(1_000)
57 38 : });
58 : /// Deployment knob (ANTARES_MAX_LIMIT): the most Entities one query page
59 : /// answers (`limit`, 6.3.9; past it 403 TooManyResults). 1,000 unless set;
60 : /// read once at first use.
61 72 : pub static MAX_LIMIT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
62 72 : std::env::var("ANTARES_MAX_LIMIT")
63 72 : .ok()
64 72 : .and_then(|v| v.parse().ok())
65 72 : .filter(|n| *n > 0)
66 72 : .unwrap_or(1_000)
67 72 : });
68 : /// Deployment knob (ANTARES_DEFAULT_LIMIT): the Entities a query page answers
69 : /// when the request names no `limit` (6.3.9 leaves it to the broker); the
70 : /// next page is Linked. 100 unless set, never above [`MAX_LIMIT`].
71 72 : pub static DEFAULT_LIMIT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
72 72 : std::env::var("ANTARES_DEFAULT_LIMIT")
73 72 : .ok()
74 72 : .and_then(|v| v.parse().ok())
75 72 : .filter(|n| *n > 0)
76 72 : .unwrap_or(100)
77 72 : .min(*MAX_LIMIT)
78 72 : });
79 : /// Deployment knob (ANTARES_TEMPORAL_INSTANCE_LIMIT): NGSI-LD 6.3.10, the
80 : /// most instances of one Attribute a temporal response serves; past it the
81 : /// representation is cut and answered 206 with a Content-Range. 1,000 unless
82 : /// set: a week of a 10-minute sensor in one answer. The ETSI suite
83 : /// creates 20 instances and expects the cut, so its brokers run with 9. The
84 : /// answer stays bounded by [`MAX_ANSWER_BYTES`]. Read once at first use into
85 : /// `AppState::temporal_instance_limit`, which a test or the wasm broker
86 : /// (no process env; it reads `globalThis`) sets per state.
87 72 : pub static TEMPORAL_INSTANCE_LIMIT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
88 72 : std::env::var("ANTARES_TEMPORAL_INSTANCE_LIMIT")
89 72 : .ok()
90 72 : .and_then(|v| v.parse().ok())
91 72 : .filter(|n| *n > 0)
92 72 : .unwrap_or(1_000)
93 72 : });
94 :
95 : /// [`TEMPORAL_INSTANCE_LIMIT`].
96 5018 : pub fn temporal_instance_limit() -> usize {
97 5018 : *TEMPORAL_INSTANCE_LIMIT
98 5018 : }
99 :
100 : /// Deployment knob (ANTARES_TEMPORAL_DEFAULT_INSTANCES): the instances of one
101 : /// Attribute a temporal response serves when the request names no
102 : /// lastN/firstN; past it the answer is cut and 206 like the ceiling's.
103 : /// 100 unless set, never above [`temporal_instance_limit`], so the ETSI
104 : /// brokers (ceiling 9) keep 9. Read once at first use into
105 : /// `AppState::temporal_default_instances`.
106 : pub static TEMPORAL_DEFAULT_INSTANCES: std::sync::LazyLock<usize> =
107 72 : std::sync::LazyLock::new(|| {
108 72 : std::env::var("ANTARES_TEMPORAL_DEFAULT_INSTANCES")
109 72 : .ok()
110 72 : .and_then(|v| v.parse().ok())
111 72 : .filter(|n| *n > 0)
112 72 : .unwrap_or(100)
113 72 : .min(temporal_instance_limit())
114 72 : });
115 :
116 : /// [`TEMPORAL_DEFAULT_INSTANCES`].
117 4938 : pub fn temporal_default_instances() -> usize {
118 4938 : *TEMPORAL_DEFAULT_INSTANCES
119 4938 : }
120 : // Deployment knob (ANTARES_MAX_FED_RESPONSE_BYTES): ceiling on one forwarded
121 : // (4.3.6) response body. The spec sets no ceiling; an over-cap peer part
122 : // fails like an unparseable payload (Table 6.3.17-1, warning 111) instead of
123 : // ballooning broker memory — one misbehaving peer must not break the
124 : // 500 MB RSS budget. Read once at first use.
125 38 : pub static MAX_FED_RESPONSE_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
126 38 : std::env::var("ANTARES_MAX_FED_RESPONSE_BYTES")
127 38 : .ok()
128 38 : .and_then(|v| v.parse().ok())
129 38 : .filter(|n| *n > 0)
130 38 : .unwrap_or(16 * 1024 * 1024)
131 38 : });
132 : // Deployment knob (ANTARES_FED_INFLIGHT): forwarded requests in flight for
133 : // the whole process. Per-request fan-out is bounded below; across requests
134 : // nothing was, and 6 000 open federated queries × 34 sources each held
135 : // 7.7 GB of buffers and connections. Callers over the cap wait their turn.
136 38 : pub static MAX_FED_INFLIGHT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
137 38 : std::env::var("ANTARES_FED_INFLIGHT")
138 38 : .ok()
139 38 : .and_then(|v| v.parse().ok())
140 38 : .filter(|n| *n > 0)
141 38 : .unwrap_or(256)
142 38 : });
143 : pub static FED_INFLIGHT: std::sync::LazyLock<tokio::sync::Semaphore> =
144 28 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*MAX_FED_INFLIGHT));
145 : // Deployment knob (ANTARES_FED_FANOUT): concurrent forwards per distributed
146 : // read. 4.3.6.1 orders the MERGE (4.5.5), never the requests, so forwards
147 : // run concurrently; this bounds how many at once per request.
148 46 : pub static MAX_FED_FANOUT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
149 46 : std::env::var("ANTARES_FED_FANOUT")
150 46 : .ok()
151 46 : .and_then(|v| v.parse().ok())
152 46 : .filter(|n| *n > 0)
153 46 : .unwrap_or(8)
154 46 : });
155 : /// 6.3.17: `NGSILD-Warning` values relayed from ONE Context Source. The
156 : /// clause makes a peer's warnings part of this broker's answer (4.3.6.4 puts
157 : /// a deeper hop's abnormality on the response that survived it), so the list
158 : /// is written by the peer but sent by this broker to a client that never
159 : /// addressed it. Table 6.3.17-1 defines four codes and a cascade adds a few
160 : /// per hop, so eight carries a real cascade; past it a source would make the
161 : /// response grow faster than the fan-out does, and would crowd out the
162 : /// warnings the clause obliges this broker to raise about the other sources.
163 : pub const MAX_PEER_WARNINGS: usize = 8;
164 :
165 : /// → 500 InternalError. How many [`crate::AppState::call`] frames one
166 : /// request may be inside. The in-process handle is the façade seam, and a
167 : /// façade legitimately calls the broker while another façade legitimately
168 : /// calls it, so the ceiling is a depth rather than a refusal of the second
169 : /// call. Without one a `/x/` route that translates into a request its own
170 : /// surface serves recurses until the stack ends, and each frame builds a
171 : /// router of its own. Eight carries a façade over a façade over the broker
172 : /// several times and still ends the loop in milliseconds, an order of
173 : /// magnitude below what ends the process: with the guard removed, a route
174 : /// asking for thirty-two hops overflows a 2 MiB thread stack in a debug
175 : /// build. The count is per task — work a handler spawns starts a new chain,
176 : /// which is right, since a notification is not inside the request that
177 : /// caused it.
178 : pub const MAX_IN_PROCESS_CALL_DEPTH: usize = 8;
179 :
180 : pub const MAX_JOIN_LEVEL: usize = 10; // → 400 BadRequestData
181 : /// → 400 BadRequestData. Documents one @context resolution may fetch, owned
182 : /// by the loader that enforces it (`antares_jsonld`), and the ceiling on how
183 : /// many DISTINCT @contexts one batch may name — without the second, the item
184 : /// count multiplies the first.
185 : pub use antares_jsonld::MAX_CONTEXT_URLS as MAX_CONTEXT_FETCHES;
186 : /// Linked-entity lookup budget per `q=`, owned by the shared evaluator.
187 : pub use antares_ql::eval::MAX_Q_LINK_LOOKUPS;
188 : /// Regex compile ceiling and retention caps, owned by the shared cache
189 : /// (`antares_ql::regex`).
190 : pub use antares_ql::regex::{MAX_REGEX_CACHE, MAX_REGEX_CACHE_BYTES, MAX_REGEX_PROGRAM_BYTES};
191 : /// → 403 TooComplexQuery. The AST size cap, owned by the parser that
192 : /// enforces it (`antares_ql::parse_q`).
193 : pub use antares_ql::MAX_Q_NODES;
194 :
195 : /// → 403 TooManyResults (5.5.6). Deployment knob
196 : /// (ANTARES_DISCOVERY_SCAN_MAX): documents ONE unpaginated whole-tenant fold
197 : /// may read or hold. The folds that carry it are the ones the spec gives no
198 : /// window to push into the store: `/types` and `/attributes` (5.7.5-5.7.10
199 : /// define no pagination) and the registration query (5.10.2.4 filters first
200 : /// and pages second, so the page cannot be pushed down). Without it one
201 : /// request over a tenant at the 100 000-registration target holds every match
202 : /// at once. Read once at first use.
203 34 : pub static MAX_FOLD_DOCS: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
204 34 : std::env::var("ANTARES_DISCOVERY_SCAN_MAX")
205 34 : .ok()
206 34 : .and_then(|v| v.parse().ok())
207 34 : .filter(|n| *n > 0)
208 34 : .unwrap_or(100_000)
209 34 : });
210 :
211 : /// The ceilings on the notification pipeline. Not input bounds: no request
212 : /// names them and none is rejected against them. They are published beside
213 : /// the input caps for the same reason the regex-cache caps above are — an
214 : /// operator reading `/q/health` has to be able to tell a ceiling from a
215 : /// coincidence, and reaching one of these is what a dropped change or a
216 : /// stalled fan-out looks like from outside.
217 : ///
218 : /// Depth of the change→matcher ring, the same size the local bus uses. A
219 : /// full ring drops the batch and counts it
220 : /// (`antares_notification_changes_dropped_total`), so this number is the
221 : /// back-pressure a deployment has before delivery loss starts.
222 : pub const CHANGE_QUEUE: usize = 1024;
223 : /// Groups one subscription may have waiting for delivery. Each subscription
224 : /// is served on a lane of its own, so a slow or dead endpoint backs up its
225 : /// lane and never the matching of every other change; a full lane
226 : /// drops the group for that subscription alone and counts it
227 : /// (`antares_notification_lane_dropped_total`).
228 : pub const DELIVERY_LANE: usize = 64;
229 : /// What every delivery lane together may hold, in estimated heap bytes of
230 : /// the entities waiting in them. [`DELIVERY_LANE`] counts groups, and a
231 : /// group is up to a drain's worth of entities copied for that subscription
232 : /// alone, so twenty subscriptions behind slow receivers under a batch upsert
233 : /// flood held 64 groups of 100 3 KB entities each and took the process past
234 : /// its 1536 MiB limit in seconds. Past this the group is dropped
235 : /// for that subscription and counted, as a full lane drops it. 128 MiB
236 : /// leaves room for a running batch, which costs as much again.
237 : /// Deployment knob (ANTARES_DELIVERY_LANE_BYTES), read once at first use.
238 34 : pub static DELIVERY_LANE_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
239 34 : count_from(
240 34 : std::env::var("ANTARES_DELIVERY_LANE_BYTES").ok().as_deref(),
241 34 : 128 * 1024 * 1024,
242 : usize::MAX,
243 : )
244 34 : });
245 : /// Batch write bodies in flight at once, in bytes. One batch costs about a
246 : /// hundred times its body while it runs: the parsed items, their expansion
247 : /// for the policy check, the stored pre-images, the forward copy, the rows
248 : /// and the change events, so a 1000-entity upsert of 3 MB held about 300 MB
249 : /// and eight of them at once took the process past its 1536 MiB limit;
250 : /// two at once were 480 MB live, so the default runs one batch of
251 : /// that size beside small ones. A batch waits for its share and is answered 503 with
252 : /// `Retry-After` when none frees inside [`BATCH_ADMIT_WAIT`]; one batch is
253 : /// always admitted alone, however large. Deployment knob
254 : /// (ANTARES_BATCH_INFLIGHT_BYTES), read once at first use.
255 30 : pub static BATCH_INFLIGHT_BYTES: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
256 30 : count_from(
257 30 : std::env::var("ANTARES_BATCH_INFLIGHT_BYTES")
258 30 : .ok()
259 30 : .as_deref(),
260 30 : 4 * 1024 * 1024,
261 30 : u32::MAX as usize,
262 : )
263 30 : });
264 : /// How long a batch waits for its share of [`BATCH_INFLIGHT_BYTES`]. A waiting
265 : /// batch holds only its body, and the batches ahead of it take well under a
266 : /// second each, so it waits long enough for a queue of dozens to drain; a
267 : /// client timeout of 30 s still sees the 503 first.
268 : pub const BATCH_ADMIT_WAIT: std::time::Duration = std::time::Duration::from_secs(20);
269 : static BATCH_SLOTS: std::sync::LazyLock<tokio::sync::Semaphore> =
270 16 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*BATCH_INFLIGHT_BYTES));
271 :
272 : /// Batch writes running at once, whatever their size. Each holds a database
273 : /// connection for as long as it runs, so a stream of small batches has to
274 : /// leave the rest of the pool to the reads: 811 vehicle upserts in two
275 : /// minutes held all 20 of a broker's and a visitor's read waited 4 s for
276 : /// one. Deployment knob (ANTARES_BATCH_INFLIGHT), read once at first
277 : /// use; keep it under `ANTARES_PG_POOL`.
278 30 : pub static BATCH_INFLIGHT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
279 30 : count_from(
280 30 : std::env::var("ANTARES_BATCH_INFLIGHT").ok().as_deref(),
281 : 8,
282 30 : u32::MAX as usize,
283 : )
284 30 : });
285 : static BATCH_RUNNING: std::sync::LazyLock<tokio::sync::Semaphore> =
286 16 : std::sync::LazyLock::new(|| tokio::sync::Semaphore::new(*BATCH_INFLIGHT));
287 :
288 : /// The admission of a batch body of `bytes`, held for as long as the batch
289 : /// runs: its share of [`BATCH_INFLIGHT_BYTES`] and one of [`BATCH_INFLIGHT`]
290 : /// slots; `None` when they did not free inside [`BATCH_ADMIT_WAIT`].
291 924 : pub async fn admit_batch(bytes: usize) -> Option<Admitted> {
292 924 : admit_both(
293 924 : (&BATCH_SLOTS, *BATCH_INFLIGHT_BYTES),
294 924 : (&BATCH_RUNNING, *BATCH_INFLIGHT),
295 924 : bytes,
296 924 : BATCH_ADMIT_WAIT,
297 924 : )
298 924 : .await
299 924 : }
300 :
301 : /// A running batch's share of the byte budget and its slot.
302 : pub type Admitted = (
303 : tokio::sync::SemaphorePermit<'static>,
304 : tokio::sync::SemaphorePermit<'static>,
305 : );
306 :
307 940 : async fn admit_both<'a>(
308 940 : bytes_budget: (&'a tokio::sync::Semaphore, usize),
309 940 : running: (&'a tokio::sync::Semaphore, usize),
310 940 : bytes: usize,
311 940 : wait: std::time::Duration,
312 940 : ) -> Option<(
313 940 : tokio::sync::SemaphorePermit<'a>,
314 940 : tokio::sync::SemaphorePermit<'a>,
315 940 : )> {
316 : // One wait for both: the slot gets what the byte share left of it.
317 : // Clock rule: std Instant panics on wasm32 (every batch write aborted the
318 : // wasm broker).
319 : #[cfg(not(target_arch = "wasm32"))]
320 940 : let started = std::time::Instant::now();
321 : #[cfg(target_arch = "wasm32")]
322 : let started = web_time::Instant::now();
323 940 : let share = admit(bytes_budget.0, bytes_budget.1, bytes, wait).await?;
324 940 : let slot = admit(
325 940 : running.0,
326 940 : running.1,
327 940 : 1,
328 940 : wait.saturating_sub(started.elapsed()),
329 940 : )
330 940 : .await?;
331 936 : Some((share, slot))
332 940 : }
333 :
334 1904 : async fn admit(
335 1904 : slots: &tokio::sync::Semaphore,
336 1904 : budget: usize,
337 1904 : bytes: usize,
338 1904 : wait: std::time::Duration,
339 1904 : ) -> Option<tokio::sync::SemaphorePermit<'_>> {
340 : // capped at the whole budget, so the largest body is admitted once it runs alone
341 1904 : let weight = bytes.clamp(1, budget) as u32;
342 : #[cfg(not(target_arch = "wasm32"))]
343 1904 : let admitted = tokio::time::timeout(wait, slots.acquire_many(weight))
344 1904 : .await
345 1904 : .ok()
346 1904 : .and_then(Result::ok);
347 : // the wasm build has no tokio timer: a batch is admitted only when its share is free now
348 : #[cfg(target_arch = "wasm32")]
349 : let admitted = {
350 : let _ = wait;
351 : slots.try_acquire_many(weight).ok()
352 : };
353 1904 : match admitted {
354 1892 : Some(permit) => Some(permit),
355 : _ => {
356 12 : metrics::counter!("antares_batch_overloaded_total").increment(1);
357 12 : None
358 : }
359 : }
360 1904 : }
361 : /// Notifications in flight at once per drain: one serial POST at a time
362 : /// capped a 9-subscription fan-out at ~600 POST/s and overflowed the ring.
363 : /// Deployment knob (ANTARES_DELIVERY_WIDTH): what a width is worth is a
364 : /// property of the subscribers, not of this broker — a slot is held for as
365 : /// long as the endpoint takes to answer, so a fleet of local sinks and a
366 : /// fleet of remote endpoints sitting at their 30 s timeout (Table 5.2.15-1)
367 : /// are served by different numbers, and the one that fits is measured
368 : /// against a deployment rather than compiled in. Read once at first use.
369 30 : pub static DELIVERY_WIDTH: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
370 30 : count_from(
371 30 : std::env::var("ANTARES_DELIVERY_WIDTH").ok().as_deref(),
372 : 64,
373 : usize::MAX,
374 : )
375 30 : });
376 : /// Of that width, what one tenant may hold. A Subscription belongs to one
377 : /// tenant (5.2.12), and a delivery to an endpoint that accepts and never
378 : /// answers holds its slot for the endpoint's whole timeout — up to 30 s
379 : /// (Table 5.2.15-1). Sharing the width with no per-tenant bound, a single
380 : /// tenant with enough dead endpoints holds every slot and nothing leaves the
381 : /// broker for anyone else. The share is a fraction of the width and not a
382 : /// fair split of it: a tenant delivering alone still gets several slots, and
383 : /// eight of them is what a full width of 64 divides into before the
384 : /// per-tenant queue becomes the bottleneck for an ordinary fan-out.
385 : /// Deployment knob (ANTARES_DELIVERY_WIDTH_PER_TENANT), ceilinged at the
386 : /// width above: a share larger than the width is not a share, and
387 : /// publishing one would name a ceiling that can never fire. Read once at
388 : /// first use.
389 30 : pub static DELIVERY_WIDTH_PER_TENANT: std::sync::LazyLock<usize> = std::sync::LazyLock::new(|| {
390 30 : count_from(
391 30 : std::env::var("ANTARES_DELIVERY_WIDTH_PER_TENANT")
392 30 : .ok()
393 30 : .as_deref(),
394 : 8,
395 30 : *DELIVERY_WIDTH,
396 : )
397 30 : });
398 :
399 : /// One configured count: the default stands in for anything unset,
400 : /// unparseable or zero — a zero would mint a semaphore that admits nobody
401 : /// and stop delivery altogether — and the ceiling bounds what a deployment
402 : /// can ask for where a larger number would be meaningless.
403 186 : fn count_from(raw: Option<&str>, default: usize, ceiling: usize) -> usize {
404 186 : raw.and_then(|v| v.parse().ok())
405 186 : .filter(|n| *n > 0)
406 186 : .unwrap_or(default)
407 186 : .min(ceiling)
408 186 : }
409 : /// Destinations and registrations the egress breaker tracks. Both maps are
410 : /// keyed by client-supplied strings, so they need a bound: at the ceiling the
411 : /// least recently recorded entry is dropped, which costs at most a forgotten
412 : /// failure count for a destination nobody has touched in a while.
413 : pub const MAX_TRACKED_DESTINATIONS: usize = 4096;
414 :
415 : /// Rejection counters, exported by /q/health.
416 : #[derive(Default)]
417 : pub struct LimitStats {
418 : pub uri_too_long: AtomicU64,
419 : pub body_too_large: AtomicU64,
420 : pub body_too_deep: AtomicU64,
421 : }
422 :
423 : impl LimitStats {
424 112 : pub fn snapshot(&self) -> serde_json::Value {
425 112 : serde_json::json!({
426 112 : "maxBodyBytes": *MAX_BODY_BYTES,
427 112 : "maxUriBytes": MAX_URI_BYTES,
428 112 : "maxJsonDepth": MAX_JSON_DEPTH,
429 112 : "maxGeoVertices": MAX_GEO_VERTICES,
430 112 : "maxBatchItems": *MAX_BATCH_ITEMS,
431 112 : "maxLimit": *MAX_LIMIT,
432 112 : "defaultLimit": *DEFAULT_LIMIT,
433 112 : "temporalInstanceLimit": temporal_instance_limit(),
434 112 : "temporalDefaultInstances": temporal_default_instances(),
435 112 : "maxAnswerBytes": MAX_ANSWER_BYTES,
436 112 : "maxFoldDocs": *MAX_FOLD_DOCS,
437 112 : "maxFedResponseBytes": *MAX_FED_RESPONSE_BYTES,
438 112 : "maxFedFanout": *MAX_FED_FANOUT,
439 112 : "maxFedInflight": *MAX_FED_INFLIGHT,
440 112 : "maxJoinLevel": MAX_JOIN_LEVEL,
441 112 : "maxInProcessCallDepth": MAX_IN_PROCESS_CALL_DEPTH,
442 112 : "maxPeerWarnings": MAX_PEER_WARNINGS,
443 112 : "maxContextFetches": MAX_CONTEXT_FETCHES,
444 112 : "maxQNodes": MAX_Q_NODES,
445 112 : "maxQLinkLookups": MAX_Q_LINK_LOOKUPS,
446 112 : "maxRegexCache": MAX_REGEX_CACHE,
447 112 : "maxRegexCacheBytes": MAX_REGEX_CACHE_BYTES,
448 112 : "maxRegexProgramBytes": MAX_REGEX_PROGRAM_BYTES,
449 112 : "changeQueue": CHANGE_QUEUE,
450 112 : "deliveryLane": DELIVERY_LANE,
451 112 : "batchInflight": *BATCH_INFLIGHT,
452 112 : "batchInflightBytes": *BATCH_INFLIGHT_BYTES,
453 112 : "deliveryLaneBytes": *DELIVERY_LANE_BYTES,
454 112 : "deliveryWidth": *DELIVERY_WIDTH,
455 112 : "deliveryWidthPerTenant": *DELIVERY_WIDTH_PER_TENANT,
456 112 : "maxTrackedDestinations": MAX_TRACKED_DESTINATIONS,
457 112 : "rejectedUriTooLong": self.uri_too_long.load(Ordering::Relaxed),
458 112 : "rejectedBodyTooLarge": self.body_too_large.load(Ordering::Relaxed),
459 112 : "rejectedBodyTooDeep": self.body_too_deep.load(Ordering::Relaxed),
460 : })
461 112 : }
462 : }
463 :
464 : /// Maximum brace/bracket nesting of a JSON byte stream, string-aware.
465 : /// A scan, not a parse — depth is checked before serde ever runs.
466 46638 : pub fn json_depth(bytes: &[u8]) -> usize {
467 46638 : let (mut depth, mut max, mut in_str, mut esc) = (0usize, 0usize, false, false);
468 13938744 : for &b in bytes {
469 13938744 : if in_str {
470 9104240 : if esc {
471 4128 : esc = false;
472 9100112 : } else if b == b'\\' {
473 4128 : esc = true;
474 9095984 : } else if b == b'"' {
475 768230 : in_str = false;
476 8327754 : }
477 9104240 : continue;
478 4834504 : }
479 4834504 : match b {
480 768230 : b'"' => in_str = true,
481 991866 : b'{' | b'[' => {
482 991866 : depth = depth.saturating_add(1);
483 991866 : max = max.max(depth);
484 991866 : }
485 991880 : b'}' | b']' => depth = depth.saturating_sub(1),
486 2082528 : _ => {}
487 : }
488 : }
489 46638 : max
490 46638 : }
491 :
492 55968 : pub(crate) async fn bounds_layer(
493 55968 : axum::extract::State(st): axum::extract::State<crate::AppState>,
494 55968 : req: Request<Body>,
495 55968 : next: Next,
496 55968 : ) -> Response {
497 55968 : if req.uri().to_string().len() > MAX_URI_BYTES {
498 16 : st.limits.uri_too_long.fetch_add(1, Ordering::Relaxed);
499 16 : return StatusCode::URI_TOO_LONG.into_response(); // bare, like 6.3.4
500 55952 : }
501 : // 6.3.4: "For HTTP POST, PATCH and PUT HTTP requests implementations shall
502 : // check … Content-Length header shall include the length of the request
503 : // payload body", and its absence "shall result in just a 411 HTTP status
504 : // code (without any payload body)" — restated in 6.3.2. Scoped to HTTP/1.x:
505 : // HTTP/2 and later carry length in the framing layer and legitimately omit
506 : // the header, so demanding it there would reject conformant clients.
507 : // The clause grants NO exemption for `Transfer-Encoding: chunked` — a
508 : // chunked POST without Content-Length is exactly the case 411 covers, so it
509 : // is deliberately not carved out here.
510 : //
511 : // The ONE deviation, made explicit rather than implied: the check is scoped
512 : // to HTTP/1.x. 6.3.4 is written against RFC 7230/7231 and predates any h2
513 : // consideration; HTTP/2 carries length in its framing and conformant h2
514 : // clients routinely omit the header, so applying it there would reject
515 : // requests the spec never meant to describe. Recorded in docs/ics.yaml.
516 55952 : if matches!(req.method().as_str(), "POST" | "PATCH" | "PUT")
517 35748 : && req.version() <= axum::http::Version::HTTP_11
518 35748 : && !req
519 35748 : .headers()
520 35748 : .contains_key(axum::http::header::CONTENT_LENGTH)
521 : {
522 32 : return StatusCode::LENGTH_REQUIRED.into_response(); // bare 411
523 55920 : }
524 55920 : let has_body = matches!(req.method().as_str(), "POST" | "PATCH" | "PUT" | "DELETE");
525 55920 : if !has_body {
526 10014 : return next.run(req).await;
527 45906 : }
528 45906 : let (parts, body) = req.into_parts();
529 : // What the client says it is sending. The read below still decides — a
530 : // declared length is a claim, and answering on the claim alone would cut
531 : // the client off mid-body, where the RST that follows can take the
532 : // response with it.
533 45906 : let fits_its_claim = parts
534 45906 : .headers
535 45906 : .get(axum::http::header::CONTENT_LENGTH)
536 45906 : .and_then(|v| v.to_str().ok())
537 45906 : .and_then(|v| v.parse::<usize>().ok())
538 45906 : .is_some_and(|n| n <= *MAX_BODY_BYTES);
539 45906 : let bytes: Bytes = match axum::body::to_bytes(body, *MAX_BODY_BYTES).await {
540 45888 : Ok(b) => b,
541 : // `to_bytes` reports the length limit and any transport failure the
542 : // same way. A body whose declared length fits the cap cannot have
543 : // exceeded it, so what failed was the delivery: that is a bad request
544 : // and NOT a size rejection — counting it as one would make
545 : // `rejectedBodyTooLarge` read client aborts as clients hitting the
546 : // cap. A body that declared more than the cap, or declared nothing
547 : // (chunked), can only have hit the cap.
548 4 : Err(_) if fits_its_claim => {
549 4 : return crate::negotiate::ApiError::from(antares_model::NgsiError::InvalidRequest(
550 4 : "request body was not delivered completely".into(),
551 4 : ))
552 4 : .into_response();
553 : }
554 : Err(_) => {
555 14 : st.limits.body_too_large.fetch_add(1, Ordering::Relaxed);
556 14 : return StatusCode::PAYLOAD_TOO_LARGE.into_response(); // bare 413
557 : }
558 : };
559 : // An absent (or unreadable) Content-Type is parsed as JSON downstream —
560 : // 6.3.4 mandates Content-Length, not Content-Type — so it is scanned
561 : // here too, or the nesting cap has a hole exactly where the parser has
562 : // none. A header that names another media type keeps its 415.
563 45888 : let is_json = parts
564 45888 : .headers
565 45888 : .get(axum::http::header::CONTENT_TYPE)
566 45888 : .is_none_or(|v| match v.to_str() {
567 35784 : Ok(ct) => ct.contains("json"),
568 : // A header the parser cannot read is not a header naming another
569 : // media type: `negotiate::content_type` reports it as the empty
570 : // string, which every route that tolerates an absent
571 : // Content-Type reads as absent and parses. Scanning it is what
572 : // keeps the cap ahead of the parser on those routes.
573 6 : Err(_) => true,
574 35790 : });
575 45888 : if is_json && json_depth(&bytes) > MAX_JSON_DEPTH {
576 16 : st.limits.body_too_deep.fetch_add(1, Ordering::Relaxed);
577 16 : return crate::negotiate::ApiError::from(antares_model::NgsiError::BadRequestData(
578 16 : format!("JSON nesting exceeds the {MAX_JSON_DEPTH}-level limit"),
579 16 : ))
580 16 : .into_response();
581 45872 : }
582 45872 : next.run(Request::from_parts(parts, Body::from(bytes)))
583 45872 : .await
584 55964 : }
585 :
586 : #[cfg(test)]
587 : mod tests {
588 : use super::*;
589 :
590 : /// The scan is a bound, not a parser: unbalanced closers must not
591 : /// underflow, and the value it reports is the one the middleware compares
592 : /// against MAX_JSON_DEPTH, so the accept/reject boundary is exact.
593 : #[test]
594 4 : fn depth_scan_survives_unbalanced_and_boundary_input() {
595 4 : assert_eq!(json_depth(b"]]]]"), 0, "stray closers must not underflow");
596 4 : assert_eq!(json_depth(b"}}}{"), 1);
597 4 : assert_eq!(json_depth(b""), 0);
598 4 : assert_eq!(
599 4 : json_depth(br#""{{{{""#),
600 : 0,
601 : "a bare string carries no depth"
602 : );
603 4 : assert_eq!(
604 4 : json_depth(br#"{"a": "\\"}"#),
605 : 1,
606 : "an escaped backslash ends the escape"
607 : );
608 4 : let at_cap = "[".repeat(MAX_JSON_DEPTH) + &"]".repeat(MAX_JSON_DEPTH);
609 4 : assert_eq!(json_depth(at_cap.as_bytes()), MAX_JSON_DEPTH);
610 4 : assert!(
611 4 : json_depth(at_cap.as_bytes()) <= MAX_JSON_DEPTH,
612 : "exactly at the cap is accepted"
613 : );
614 4 : let over = "[".repeat(MAX_JSON_DEPTH + 1) + &"]".repeat(MAX_JSON_DEPTH + 1);
615 4 : assert!(
616 4 : json_depth(over.as_bytes()) > MAX_JSON_DEPTH,
617 : "one over is rejected"
618 : );
619 4 : }
620 :
621 : /// A cap `/q/health` publishes is the cap that fires: an OR chain of
622 : /// `MAX_Q_NODES - 1` terms (the chain node plus its terms) parses, one
623 : /// term more is 5.5.6 TooComplexQuery.
624 : #[test]
625 4 : fn the_published_q_node_cap_is_the_enforced_one() {
626 8 : let chain = |k: usize| vec!["a==1"; k].join("|");
627 4 : assert!(antares_ql::parse_q(&chain(MAX_Q_NODES - 1)).is_ok());
628 4 : assert!(matches!(
629 4 : antares_ql::parse_q(&chain(MAX_Q_NODES)),
630 : Err(antares_model::NgsiError::TooComplexQuery(_))
631 : ));
632 4 : }
633 :
634 : /// Every cap belongs in `/q/health` AND in the operator's chapter: the
635 : /// admin-API chapter prints the whole `limits` object as the answer a
636 : /// memory-store broker gives, and an operator reads a bound from there.
637 : /// The test above pins the key set in code; nothing pinned the chapter,
638 : /// and `maxFedInflight` was published for a release without ever being
639 : /// written down. A cap added here now fails until the chapter has it.
640 : #[test]
641 4 : fn the_documented_health_limits_are_the_published_ones() {
642 4 : let book = std::fs::read_to_string(concat!(
643 : env!("CARGO_MANIFEST_DIR"),
644 : "/../../docs/src/admin-api.md"
645 : ))
646 4 : .expect("the admin API chapter");
647 4 : let block = book
648 4 : .split_once("\"limits\": {")
649 4 : .expect("the limits object in the sample response")
650 4 : .1;
651 4 : let block = block.split_once('}').expect("the end of the object").0;
652 4 : let mut documented: Vec<&str> = block
653 4 : .lines()
654 144 : .filter_map(|l| l.trim().strip_prefix('"'))
655 136 : .filter_map(|l| l.split_once('"'))
656 4 : .map(|(k, _)| k)
657 4 : .collect();
658 4 : documented.sort_unstable();
659 4 : let snap = LimitStats::default().snapshot();
660 4 : let mut published: Vec<&str> = snap
661 4 : .as_object()
662 4 : .expect("object")
663 4 : .keys()
664 4 : .map(String::as_str)
665 4 : .collect();
666 4 : published.sort_unstable();
667 4 : assert_eq!(
668 : documented, published,
669 : "the admin API chapter's limits object is not the one /q/health answers"
670 : );
671 4 : }
672 :
673 : /// /q/health publishes the caps and the rejection counters — and nothing
674 : /// else: no configuration paths, no environment variable values, no
675 : /// internal error text.
676 : #[test]
677 4 : fn health_snapshot_reports_the_caps_and_nothing_internal() {
678 4 : let stats = LimitStats::default();
679 4 : stats.uri_too_long.fetch_add(3, Ordering::Relaxed);
680 4 : let snap = stats.snapshot();
681 4 : let obj = snap.as_object().expect("object");
682 4 : let mut keys: Vec<&str> = obj.keys().map(String::as_str).collect();
683 4 : keys.sort_unstable();
684 4 : assert_eq!(
685 : keys,
686 : [
687 : "batchInflight",
688 : "batchInflightBytes",
689 : "changeQueue",
690 : "defaultLimit",
691 : "deliveryLane",
692 : "deliveryLaneBytes",
693 : "deliveryWidth",
694 : "deliveryWidthPerTenant",
695 : "maxAnswerBytes",
696 : "maxBatchItems",
697 : "maxBodyBytes",
698 : "maxContextFetches",
699 : "maxFedFanout",
700 : "maxFedInflight",
701 : "maxFedResponseBytes",
702 : "maxFoldDocs",
703 : "maxGeoVertices",
704 : "maxInProcessCallDepth",
705 : "maxJoinLevel",
706 : "maxJsonDepth",
707 : "maxLimit",
708 : "maxPeerWarnings",
709 : "maxQLinkLookups",
710 : "maxQNodes",
711 : "maxRegexCache",
712 : "maxRegexCacheBytes",
713 : "maxRegexProgramBytes",
714 : "maxTrackedDestinations",
715 : "maxUriBytes",
716 : "rejectedBodyTooDeep",
717 : "rejectedBodyTooLarge",
718 : "rejectedUriTooLong",
719 : "temporalDefaultInstances",
720 : "temporalInstanceLimit",
721 : ],
722 : "no member beyond the caps, the pipeline ceilings and the counters"
723 : );
724 4 : assert_eq!(snap["rejectedUriTooLong"], 3);
725 4 : assert!(
726 136 : obj.values().all(|v| v.is_number()),
727 : "every member is a number — no strings to leak paths through"
728 : );
729 4 : }
730 :
731 : /// The nesting cap is checked BEFORE any parse — including on the path
732 : /// that carries no Content-Type header at all, which the body parser
733 : /// accepts and parses as JSON (6.3.4 only mandates Content-Length).
734 : /// An unparseable Content-Type follows the same rule.
735 : #[tokio::test]
736 4 : async fn over_depth_body_without_content_type_is_still_rejected() {
737 : use tower::ServiceExt;
738 4 : let st = crate::AppState::new("http://localhost:0".into());
739 4 : let app = axum::Router::new()
740 4 : .route(
741 4 : "/x",
742 4 : axum::routing::post(|| async { StatusCode::NO_CONTENT }),
743 : )
744 4 : .layer(axum::middleware::from_fn_with_state(st, bounds_layer));
745 4 : let deep = "[".repeat(MAX_JSON_DEPTH + 5) + &"]".repeat(MAX_JSON_DEPTH + 5);
746 : // The third case is the one a header map can hold and `to_str`
747 : // cannot read: a Content-Type carrying a byte outside UTF-8. Every
748 : // route reads it as an absent Content-Type, so the scan must too.
749 4 : let unreadable = axum::http::HeaderValue::from_bytes(b"application/\xffjson")
750 4 : .expect("header value from raw bytes");
751 4 : assert!(
752 4 : unreadable.to_str().is_err(),
753 : "the case under test is a header value that cannot be read as text"
754 : );
755 12 : for ct in [
756 4 : None,
757 4 : Some(axum::http::HeaderValue::from_static("application/json")),
758 4 : Some(unreadable),
759 4 : ] {
760 12 : let mut req = Request::post("/x")
761 12 : .header(axum::http::header::CONTENT_LENGTH, deep.len().to_string());
762 12 : if let Some(ct) = ct.clone() {
763 8 : req = req.header(axum::http::header::CONTENT_TYPE, ct);
764 8 : }
765 12 : let resp = app
766 12 : .clone()
767 12 : .oneshot(req.body(Body::from(deep.clone())).expect("req"))
768 12 : .await
769 12 : .expect("resp");
770 12 : assert_eq!(
771 12 : resp.status(),
772 4 : StatusCode::BAD_REQUEST,
773 4 : "an over-deep body must not reach the handler (content-type {ct:?})"
774 4 : );
775 4 : }
776 4 : }
777 :
778 : /// 6.3.4's 413 is about size, and `/q/health` publishes how often it
779 : /// fired. A body the client abandoned mid-flight is a different event: it
780 : /// is not over the cap, it must not be counted as one, and a declared
781 : /// length over the cap must be refused before the broker buffers a byte
782 : /// of it.
783 : #[tokio::test]
784 4 : async fn a_broken_body_is_not_an_over_cap_body() {
785 : use tower::ServiceExt;
786 4 : let st = crate::AppState::new("http://localhost:0".into());
787 4 : let app = axum::Router::new()
788 4 : .route(
789 4 : "/x",
790 4 : axum::routing::post(|| async { StatusCode::NO_CONTENT }),
791 : )
792 4 : .layer(axum::middleware::from_fn_with_state(
793 4 : st.clone(),
794 : bounds_layer,
795 : ));
796 :
797 : // the transport gave up: a declared length the body never delivers
798 4 : let broken = Body::from_stream(futures_util::stream::once(async {
799 4 : Err::<axum::body::Bytes, std::io::Error>(std::io::Error::other("reset"))
800 4 : }));
801 4 : let resp = app
802 4 : .clone()
803 4 : .oneshot(
804 4 : Request::post("/x")
805 4 : .header(axum::http::header::CONTENT_LENGTH, "100")
806 4 : .body(broken)
807 4 : .expect("req"),
808 4 : )
809 4 : .await
810 4 : .expect("resp");
811 4 : assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
812 4 : assert_eq!(
813 4 : st.limits.body_too_large.load(Ordering::Relaxed),
814 : 0,
815 : "a client abort is not a size rejection"
816 : );
817 :
818 : // over the cap for real: 6.3.4's 413, counted
819 4 : let over = vec![b'x'; *MAX_BODY_BYTES + 1];
820 4 : let resp = app
821 4 : .oneshot(
822 4 : Request::post("/x")
823 4 : .header(axum::http::header::CONTENT_LENGTH, over.len().to_string())
824 4 : .body(Body::from(over))
825 4 : .expect("req"),
826 4 : )
827 4 : .await
828 4 : .expect("resp");
829 4 : assert_eq!(resp.status(), StatusCode::PAYLOAD_TOO_LARGE);
830 4 : assert_eq!(st.limits.body_too_large.load(Ordering::Relaxed), 1);
831 4 : }
832 :
833 : /// Every cap this module declares is in the payload `/q/health` serves.
834 : /// A cap an operator cannot read is one they cannot tell from a
835 : /// coincidence when a request is refused or a change is dropped, so the
836 : /// list is read out of this file's own source rather than kept by hand:
837 : /// a cap added below without a member above fails here.
838 : #[test]
839 4 : fn every_declared_cap_is_published() {
840 4 : let src = include_str!("bounds.rs");
841 4 : let published = LimitStats::default().snapshot();
842 4 : let published = published.as_object().expect("an object");
843 4 : let mut missing = Vec::new();
844 4008 : for line in src.lines() {
845 4008 : let line = line.trim_start();
846 : // `pub use` re-exports name their cap in another crate; the
847 : // declarations here are the ones this file owns.
848 8016 : for kw in ["pub const ", "pub static "] {
849 8016 : let Some(rest) = line.strip_prefix(kw) else {
850 7908 : continue;
851 : };
852 108 : let Some((name, ty)) = rest.split_once(':') else {
853 0 : continue;
854 : };
855 108 : let name = name.trim();
856 : // A cap is a count; the semaphore built from one is not a
857 : // second cap and has nothing of its own to publish.
858 108 : if !ty.contains("usize") {
859 12 : continue;
860 96 : }
861 : // MAX_URI_BYTES → maxUriBytes
862 96 : let mut camel = String::new();
863 280 : for (i, word) in name.split('_').enumerate() {
864 280 : let lower = word.to_lowercase();
865 280 : if i == 0 {
866 96 : camel.push_str(&lower);
867 96 : } else {
868 184 : let mut c = lower.chars();
869 184 : if let Some(f) = c.next() {
870 184 : camel.extend(f.to_uppercase());
871 184 : camel.push_str(c.as_str());
872 184 : }
873 : }
874 : }
875 96 : if !published.contains_key(&camel) {
876 0 : missing.push(format!("{name} (expected {camel:?})"));
877 96 : }
878 : }
879 : }
880 4 : assert!(
881 4 : missing.is_empty(),
882 : "caps declared here and absent from /q/health: {missing:?}"
883 : );
884 4 : }
885 :
886 : /// A configured count is read once and never re-read, so a value that
887 : /// would stop delivery has no second chance to be corrected: zero mints
888 : /// a semaphore that admits nobody, and a per-tenant share above the
889 : /// width names a ceiling that can never fire. Both fall back rather
890 : /// than take the number as given.
891 : #[test]
892 4 : fn a_configured_count_falls_back_rather_than_disabling_delivery() {
893 4 : assert_eq!(count_from(None, 64, usize::MAX), 64, "unset is the default");
894 4 : assert_eq!(count_from(Some("weeks"), 64, usize::MAX), 64);
895 4 : assert_eq!(count_from(Some(""), 64, usize::MAX), 64);
896 4 : assert_eq!(
897 4 : count_from(Some("0"), 64, usize::MAX),
898 : 64,
899 : "zero would admit nobody"
900 : );
901 4 : assert_eq!(count_from(Some("-8"), 64, usize::MAX), 64);
902 4 : assert_eq!(count_from(Some("256"), 64, usize::MAX), 256);
903 4 : assert_eq!(
904 4 : count_from(Some("512"), 8, 64),
905 : 64,
906 : "a share over the width is the width"
907 : );
908 4 : assert_eq!(
909 4 : count_from(None, 8, 4),
910 : 4,
911 : "the ceiling binds the default too"
912 : );
913 4 : }
914 :
915 : /// A guard on the compiled defaults rather than on the clamp: both
916 : /// counts are read once per process, so a test cannot set one without
917 : /// racing every other test in the binary for the read. What it holds is
918 : /// that the two numbers shipped stay a share and a total, and that
919 : /// neither default is ever lowered to zero — which would admit nobody.
920 : #[test]
921 4 : fn the_published_tenant_share_fits_inside_the_published_width() {
922 4 : let snap = LimitStats::default().snapshot();
923 4 : let width = snap["deliveryWidth"].as_u64().expect("a number");
924 4 : let share = snap["deliveryWidthPerTenant"].as_u64().expect("a number");
925 4 : assert!(share <= width, "share {share} exceeds width {width}");
926 4 : assert!(width > 0 && share > 0, "neither may be zero");
927 4 : }
928 :
929 : #[test]
930 4 : fn depth_scan_is_string_aware() {
931 4 : assert_eq!(json_depth(br#"{"a": [1, {"b": 2}]}"#), 3);
932 4 : assert_eq!(
933 4 : json_depth(br#"{"a": "}]}]}]{[{["}"#),
934 : 1,
935 : "braces in strings don't count"
936 : );
937 4 : assert_eq!(
938 4 : json_depth(br#"{"a": "\"}"}"#),
939 : 1,
940 : "escaped quotes stay in-string"
941 : );
942 4 : let deep = "[".repeat(100) + &"]".repeat(100);
943 4 : assert_eq!(json_depth(deep.as_bytes()), 100);
944 4 : }
945 :
946 : /// batches run inside their byte budget. A second one that does
947 : /// not fit waits and is refused when nothing frees; a body larger than
948 : /// the whole budget still runs alone; a finished batch frees its share.
949 : #[tokio::test]
950 4 : async fn batches_wait_for_their_share_of_the_in_flight_budget() {
951 4 : let slots = tokio::sync::Semaphore::new(10);
952 4 : let wait = std::time::Duration::from_millis(50);
953 4 : let first = admit(&slots, 10, 6, wait).await.expect("fits");
954 4 : assert!(
955 4 : admit(&slots, 10, 6, wait).await.is_none(),
956 : "6 + 6 is past 10"
957 : );
958 4 : let small = admit(&slots, 10, 4, wait).await.expect("6 + 4 fits");
959 4 : drop((first, small));
960 4 : let huge = admit(&slots, 10, 1 << 30, wait)
961 4 : .await
962 4 : .expect("alone, a huge body runs");
963 4 : assert!(
964 4 : admit(&slots, 10, 0, wait).await.is_none(),
965 : "an empty body still counts one"
966 : );
967 4 : drop(huge);
968 4 : assert!(admit(&slots, 10, 0, wait).await.is_some());
969 4 : assert_eq!(slots.available_permits(), 10, "every share came back");
970 4 : }
971 :
972 : /// small batches each fit the byte budget, yet only as many run
973 : /// as there are slots, so the rest of the pool stays with the reads; a
974 : /// batch refused a slot gives its byte share back.
975 : #[tokio::test]
976 4 : async fn only_so_many_batches_run_at_once_however_small() {
977 4 : let (bytes, running) = (
978 4 : tokio::sync::Semaphore::new(1000),
979 4 : tokio::sync::Semaphore::new(2),
980 4 : );
981 4 : let wait = std::time::Duration::from_millis(50);
982 16 : let admit = || admit_both((&bytes, 1000), (&running, 2), 10, wait);
983 4 : let first = admit().await.expect("a slot");
984 4 : let second = admit().await.expect("a slot");
985 4 : assert!(
986 4 : admit().await.is_none(),
987 : "two run, the third waits and is refused"
988 : );
989 4 : assert_eq!(
990 4 : bytes.available_permits(),
991 : 980,
992 : "the refused one's share came back"
993 : );
994 4 : drop(first);
995 4 : let third = admit().await.expect("a freed slot");
996 4 : drop((second, third));
997 4 : assert_eq!(
998 4 : (bytes.available_permits(), running.available_permits()),
999 4 : (1000, 2)
1000 4 : );
1001 4 : }
1002 : }
|
