Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Graceful shutdown drain.
3 : //!
4 : //! The ORDER is the whole feature, and it exists because of one asymmetry: a
5 : //! load balancer learns this instance is going away only by polling
6 : //! `/q/health`, but the orchestrator kills it on its own schedule. So the
7 : //! health endpoint must go unhealthy while the socket still works, and only
8 : //! then may the socket close.
9 : //!
10 : //! 1. flip `draining` → `/q/health` answers 503 (see `antares_api::health`)
11 : //! 2. keep accepting for `ANTARES_DRAIN_DELAY_MS` — the LB's notice window;
12 : //! this is the step people skip, and skipping it is what turns a rolling
13 : //! update into a burst of connection-refused
14 : //! 3. stop accepting
15 : //! 4. wait for in-flight connections, bounded by `ANTARES_DRAIN_DEADLINE_SECS`
16 : //! 5. wait for the outbox to empty — same deadline, whatever step 4 left of it
17 : //! (see the note in `drain`)
18 : //! 6. close the pools
19 : //!
20 : //! The two numbers are for different jobs and are easy to confuse: the delay
21 : //! is the LB's notice window in MILLIseconds (default 2000), the deadline is
22 : //! the ceiling on in-flight work in SECONDS (default 20). Steps 4 and 5 share
23 : //! that one deadline.
24 : //!
25 : //! Operational contract: the container `stopGracePeriod` (compose
26 : //! `stop_grace_period`, K8s `terminationGracePeriodSeconds`) MUST exceed
27 : //! delay + deadline, or the orchestrator turns a drain into a kill. The
28 : //! defaults below (2 s + 20 s) overrun Docker's 10 s default as soon as the
29 : //! in-flight work is not short; the reference manifests set both explicitly.
30 :
31 : use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
32 : use std::sync::Arc;
33 : use std::time::{Duration, Instant};
34 :
35 : /// The LB's notice window (default 2 s): how long to keep serving AFTER going
36 : /// unhealthy, sized so a load balancer's health poll actually observes the 503
37 : /// before the socket goes. It is NOT the in-flight ceiling — that is
38 : /// `drain_deadline`, and it is 10× longer.
39 73 : pub fn drain_delay() -> Result<Duration, String> {
40 73 : env_num("ANTARES_DRAIN_DELAY_MS", 2000).map(Duration::from_millis)
41 73 : }
42 :
43 : /// The real shutdown deadline (default 20 s): the ceiling on waiting for
44 : /// in-flight work — connections, then the outbox — once the listener is
45 : /// closed. Both share it; it does not extend the notice window above.
46 63 : pub fn drain_deadline() -> Result<Duration, String> {
47 63 : env_num("ANTARES_DRAIN_DEADLINE_SECS", 20).map(Duration::from_secs)
48 63 : }
49 :
50 : /// Absent = the documented default; present-but-unparsable is fatal. A
51 : /// misread drain window silently running at the default is the same class of
52 : /// misconfiguration as an unknown key, and is refused the same way. Zero is a
53 : /// real choice on both knobs (no notice window / close at once).
54 136 : fn env_num(key: &str, default: u64) -> Result<u64, String> {
55 136 : match std::env::var(key) {
56 98 : Err(std::env::VarError::NotPresent) => Ok(default),
57 0 : Err(e) => Err(format!("{key} is unreadable: {e}")),
58 38 : Ok(v) => v
59 38 : .parse::<u64>()
60 38 : .map_err(|e| format!("{key} must be a non-negative integer, got {v:?} ({e})")),
61 : }
62 136 : }
63 :
64 : /// Resolves on SIGTERM or SIGINT. SIGTERM is the one that matters —
65 : /// it is what every orchestrator sends — and listening only for ctrl_c (the
66 : /// v0 behaviour) meant a `docker stop` or a pod eviction dropped every
67 : /// in-flight request on the floor.
68 8 : pub async fn signal() {
69 : #[cfg(unix)]
70 : {
71 : use tokio::signal::unix::{signal, SignalKind};
72 8 : let mut term = match signal(SignalKind::terminate()) {
73 8 : Ok(s) => s,
74 0 : Err(e) => {
75 0 : tracing::warn!("cannot listen for SIGTERM ({e}); ctrl-c only");
76 0 : let _ = tokio::signal::ctrl_c().await;
77 0 : return;
78 : }
79 : };
80 8 : tokio::select! {
81 8 : _ = term.recv() => tracing::info!("SIGTERM received"),
82 8 : _ = tokio::signal::ctrl_c() => tracing::info!("SIGINT received"),
83 : }
84 : }
85 : #[cfg(not(unix))]
86 : {
87 : let _ = tokio::signal::ctrl_c().await;
88 : }
89 8 : }
90 :
91 : /// Steps 4–6. Steps 1–3 belong to the accept loop, which owns the listener.
92 : /// The deadline is passed in, not read here: the composition root parses every
93 : /// config value once, at startup, so a garbage window fails before serving.
94 16 : pub async fn drain(
95 16 : inflight: &Arc<AtomicUsize>,
96 16 : pending_changes: &AtomicUsize,
97 16 : store: &dyn antares_store::CurrentStateDriver,
98 16 : temporal: &dyn antares_store::TemporalDriver,
99 16 : deadline: Duration,
100 16 : flush_outbox: bool,
101 16 : ) {
102 16 : let started = Instant::now();
103 : // A request is not over when its response is: the remote leg of a
104 : // distributed subscription, an initial Context Source notification and a
105 : // forwarded notification all run as tasks after the 2xx, and a stop that
106 : // dropped them left the subscription chain half-built on every roll.
107 : // The matcher queue is part of a request too: a change accepted before
108 : // the listener closed still owes its notifications (5.8.6).
109 66 : while inflight.load(Ordering::Relaxed) > 0
110 38 : || antares_api::background_tasks() > 0
111 14 : || pending_changes.load(Ordering::SeqCst) > 0
112 : {
113 52 : if started.elapsed() >= deadline {
114 2 : tracing::warn!(
115 : "drain deadline {deadline:?} hit with {} connection(s), {} task(s) and {} change batch(es) still open — closing anyway",
116 0 : inflight.load(Ordering::Relaxed),
117 0 : antares_api::background_tasks(),
118 0 : pending_changes.load(Ordering::SeqCst)
119 : );
120 2 : break;
121 50 : }
122 50 : tokio::time::sleep(Duration::from_millis(25)).await;
123 : }
124 : // The outbox is drained by a background task on the api pods. Stopping
125 : // here — after the last request has committed its row, before the pool
126 : // closes — gives that task the chance to publish what is still pending,
127 : // so a rolling update does not leave events sitting in the table until
128 : // another pod's fallback poll finds them. Stores without an outbox
129 : // (memory, file) answer an empty page and fall straight through.
130 : //
131 : // Two residuals, stated rather than papered over. The table is shared, so
132 : // this waits for rows OTHER pods are still producing too, and under
133 : // sustained write load it therefore runs to the deadline; and the deadline
134 : // is the same one step 4 just spent, so a slow in-flight wait can leave
135 : // the flush no time at all. Either way the rows stay committed — the
136 : // fallback poll on a surviving pod publishes them — so the ceiling costs
137 : // latency, never an event.
138 16 : if flush_outbox {
139 : loop {
140 744 : match store.outbox_peek(1).await {
141 744 : Ok(rows) if rows.is_empty() => break,
142 735 : Ok(_) => {}
143 0 : Err(e) => {
144 0 : tracing::warn!("outbox flush gave up: {e}");
145 0 : break;
146 : }
147 : }
148 735 : if started.elapsed() >= deadline {
149 1 : tracing::warn!("drain deadline {deadline:?} hit with outbox rows still pending");
150 1 : break;
151 734 : }
152 734 : tokio::time::sleep(Duration::from_millis(25)).await;
153 : }
154 6 : }
155 16 : store.close().await;
156 : // Both seams, because the temporal half may be a store of its own
157 : // (`ANTARES_TEMPORAL` naming a second backend) with its own pool. When
158 : // one instance serves both, the second call lands on an already-closed
159 : // pool and does nothing.
160 16 : temporal.close().await;
161 16 : tracing::info!("drain complete in {:?}", started.elapsed());
162 16 : }
163 :
164 : /// Step 1, so the flip and the log line stay in one place.
165 10 : pub fn begin(draining: &Arc<AtomicBool>, delay: Duration) {
166 10 : draining.store(true, Ordering::Relaxed);
167 : // Immediate, not sampler-paced — a roll must be visible on a
168 : // dashboard for its whole (short) duration.
169 10 : metrics::gauge!("antares_draining").set(1.0);
170 10 : tracing::info!("draining: /q/health now 503 for {delay:?} before the listener closes");
171 10 : }
172 :
173 : #[cfg(test)]
174 : mod tests {
175 : use super::*;
176 :
177 : /// A temporal driver that only records whether it was closed; every
178 : /// operation answers "no temporal store", the same shape as `NoTemporal`.
179 : struct CountsCloses(Arc<AtomicUsize>);
180 :
181 : impl CountsCloses {
182 0 : fn off<T>() -> Result<T, antares_model::NgsiError> {
183 0 : Err(antares_model::NgsiError::OperationNotSupported(
184 0 : "test driver".into(),
185 0 : ))
186 0 : }
187 : }
188 :
189 : #[async_trait::async_trait]
190 : impl antares_store::TemporalDriver for CountsCloses {
191 2 : async fn close(&self) {
192 : self.0.fetch_add(1, Ordering::SeqCst);
193 2 : }
194 : async fn temporal_append(
195 : &self,
196 : _t: &antares_model::TenantId,
197 : _id: &str,
198 : _shell: &serde_json::Value,
199 : _add: &serde_json::Value,
200 0 : ) -> Result<(), antares_model::NgsiError> {
201 : Self::off()
202 0 : }
203 : async fn query_temporal(
204 : &self,
205 : _t: &antares_model::TenantId,
206 : _f: &antares_store::filter::TemporalFilter<'_>,
207 0 : ) -> Result<antares_store::filter::TemporalOutcome, antares_model::NgsiError> {
208 : Self::off()
209 0 : }
210 : async fn get_temporal(
211 : &self,
212 : _t: &antares_model::TenantId,
213 : _id: &str,
214 : _f: &antares_store::filter::TemporalFilter<'_>,
215 0 : ) -> Result<Option<serde_json::Value>, antares_model::NgsiError> {
216 : Self::off()
217 0 : }
218 : async fn get(
219 : &self,
220 : _t: &antares_model::TenantId,
221 : _id: &str,
222 0 : ) -> Result<Option<serde_json::Value>, antares_model::NgsiError> {
223 : Self::off()
224 0 : }
225 : async fn create(
226 : &self,
227 : _t: &antares_model::TenantId,
228 : _id: &str,
229 : _d: serde_json::Value,
230 0 : ) -> Result<bool, antares_model::NgsiError> {
231 : Self::off()
232 0 : }
233 : async fn upsert(
234 : &self,
235 : _t: &antares_model::TenantId,
236 : _id: &str,
237 : _d: serde_json::Value,
238 0 : ) -> Result<bool, antares_model::NgsiError> {
239 : Self::off()
240 0 : }
241 : async fn delete(
242 : &self,
243 : _t: &antares_model::TenantId,
244 : _id: &str,
245 0 : ) -> Result<bool, antares_model::NgsiError> {
246 : Self::off()
247 0 : }
248 : async fn list(
249 : &self,
250 : _t: &antares_model::TenantId,
251 0 : ) -> Result<Vec<serde_json::Value>, antares_model::NgsiError> {
252 : Self::off()
253 0 : }
254 : async fn mutate_boxed<'a>(
255 : &self,
256 : _t: &antares_model::TenantId,
257 : _id: &str,
258 : _f: antares_store::MutateFn<'a>,
259 0 : ) -> Result<Option<Result<(), ()>>, antares_model::NgsiError> {
260 : Self::off()
261 0 : }
262 : }
263 :
264 : /// The drain closes BOTH driver seams. With `ANTARES_TEMPORAL` naming a
265 : /// backend of its own the temporal half is a second store holding its own
266 : /// connection pool; closing only the current-state store left that pool
267 : /// open for process teardown to sever, with whatever it still owed
268 : /// in flight.
269 : #[tokio::test(flavor = "multi_thread")]
270 2 : async fn drain_closes_the_temporal_driver_too() {
271 2 : let closes = Arc::new(AtomicUsize::new(0));
272 2 : let temporal = CountsCloses(Arc::clone(&closes));
273 2 : let store = antares_sql::store::any::AnyStore::Mem(antares_sql::store::Store::default());
274 2 : drain(
275 2 : &Arc::new(AtomicUsize::new(0)),
276 2 : &AtomicUsize::new(0),
277 2 : &store,
278 2 : &temporal,
279 2 : Duration::from_millis(50),
280 2 : false,
281 2 : )
282 2 : .await;
283 2 : assert_eq!(
284 2 : closes.load(Ordering::SeqCst),
285 2 : 1,
286 2 : "the temporal driver must be closed by the drain, not by process exit"
287 2 : );
288 2 : }
289 :
290 : /// The book states these two defaults, and their SUM is an operator
291 : /// contract: the container stop grace period has to exceed it or the
292 : /// orchestrator turns a drain into a kill. Nothing tied the stated
293 : /// numbers to the ones the binary uses, and the delay drifted to a value
294 : /// 1.5 s shorter than the truth in three chapters at once — a grace
295 : /// period sized from the book would then have been under the real drain.
296 : /// `dev/check-env-docs.sh` proves each variable is documented; this
297 : /// proves the documented numbers are the ones that run.
298 : #[test]
299 2 : fn the_documented_drain_defaults_are_the_ones_the_binary_uses() {
300 2 : let book = std::fs::read_to_string(concat!(
301 : env!("CARGO_MANIFEST_DIR"),
302 : "/../../docs/src/configuration.md"
303 : ))
304 2 : .expect("the configuration chapter");
305 4 : let stated = |var: &str| -> String {
306 4 : let row = book
307 4 : .lines()
308 314 : .find(|l| l.starts_with(&format!("| `{var}` |")))
309 4 : .unwrap_or_else(|| panic!("{var} has no row in the configuration table"));
310 4 : row.split('|')
311 4 : .nth(2)
312 4 : .expect("the default column")
313 4 : .trim()
314 4 : .trim_matches('`')
315 4 : .to_owned()
316 4 : };
317 2 : std::env::remove_var("ANTARES_DRAIN_DELAY_MS");
318 2 : std::env::remove_var("ANTARES_DRAIN_DEADLINE_SECS");
319 2 : assert_eq!(
320 2 : stated("ANTARES_DRAIN_DELAY_MS"),
321 2 : drain_delay().expect("absent").as_millis().to_string(),
322 : "the book's notice window is not the compiled one"
323 : );
324 2 : assert_eq!(
325 2 : stated("ANTARES_DRAIN_DEADLINE_SECS"),
326 2 : drain_deadline().expect("absent").as_secs().to_string(),
327 : "the book's in-flight ceiling is not the compiled one"
328 : );
329 :
330 : // The shipped compose files justify their stop_grace_period against
331 : // the same two defaults, in a comment an operator copies the number
332 : // out of. Both said 0.5 s long after the delay became 2 s.
333 2 : let secs = drain_delay().expect("absent").as_secs_f64();
334 2 : let want = format!(
335 : "drain delay ({} s)",
336 2 : if secs.fract() == 0.0 {
337 2 : format!("{secs:.0}")
338 : } else {
339 0 : format!("{secs}")
340 : }
341 : );
342 4 : for name in ["docker-compose-ha.yml", "docker-compose-roles.yml"] {
343 4 : let path = format!(
344 : concat!(env!("CARGO_MANIFEST_DIR"), "/../../compose-files/{}"),
345 : name
346 : );
347 4 : let text = std::fs::read_to_string(&path).expect("the compose file");
348 4 : assert!(
349 4 : text.contains(&want),
350 : "{name} does not justify stop_grace_period against \"{want}\""
351 : );
352 : }
353 2 : }
354 :
355 : /// Both drain knobs in ONE test: the environment is process-global, so
356 : /// parsing them from parallel test threads would race.
357 : ///
358 : /// Contract: absent = the documented default; present-but-unparsable is
359 : /// FATAL, never a silent default — a misread timeout is exactly the class
360 : /// of misconfiguration the unknown-key policy exists to catch. The two
361 : /// defaults are different numbers for different jobs: 2000 MILLIseconds of
362 : /// LB notice, 20 SECONDS of in-flight ceiling.
363 : #[test]
364 2 : fn drain_knobs_default_when_absent_and_refuse_garbage() {
365 2 : std::env::remove_var("ANTARES_DRAIN_DELAY_MS");
366 2 : std::env::remove_var("ANTARES_DRAIN_DEADLINE_SECS");
367 2 : assert_eq!(drain_delay().expect("absent"), Duration::from_millis(2000));
368 2 : assert_eq!(drain_deadline().expect("absent"), Duration::from_secs(20));
369 2 : assert_ne!(
370 2 : drain_delay().expect("absent"),
371 2 : drain_deadline().expect("absent"),
372 : "the notice window and the in-flight ceiling are not the same number"
373 : );
374 :
375 2 : std::env::set_var("ANTARES_DRAIN_DELAY_MS", "750");
376 2 : std::env::set_var("ANTARES_DRAIN_DEADLINE_SECS", "10");
377 2 : assert_eq!(drain_delay().expect("set"), Duration::from_millis(750));
378 2 : assert_eq!(drain_deadline().expect("set"), Duration::from_secs(10));
379 :
380 : // Zero is a real choice on both knobs (no notice window / close at
381 : // once), so it must NOT be rejected with the garbage.
382 2 : std::env::set_var("ANTARES_DRAIN_DELAY_MS", "0");
383 2 : std::env::set_var("ANTARES_DRAIN_DEADLINE_SECS", "0");
384 2 : assert_eq!(drain_delay().expect("zero"), Duration::ZERO);
385 2 : assert_eq!(drain_deadline().expect("zero"), Duration::ZERO);
386 :
387 14 : for bad in [
388 2 : "soon",
389 2 : "",
390 2 : "-1",
391 2 : "2.5",
392 2 : "500ms",
393 2 : "99999999999999999999999",
394 2 : " 5",
395 2 : ] {
396 14 : std::env::set_var("ANTARES_DRAIN_DELAY_MS", bad);
397 14 : let err =
398 14 : drain_delay().expect_err(&format!("ANTARES_DRAIN_DELAY_MS={bad:?} must be fatal"));
399 14 : assert!(
400 14 : err.contains("ANTARES_DRAIN_DELAY_MS"),
401 : "the error must name the key: {err}"
402 : );
403 : }
404 2 : std::env::set_var("ANTARES_DRAIN_DELAY_MS", "2000");
405 8 : for bad in ["soon", "", "-1", "20.0"] {
406 8 : std::env::set_var("ANTARES_DRAIN_DEADLINE_SECS", bad);
407 8 : let err = drain_deadline().expect_err(&format!(
408 8 : "ANTARES_DRAIN_DEADLINE_SECS={bad:?} must be fatal"
409 8 : ));
410 8 : assert!(err.contains("ANTARES_DRAIN_DEADLINE_SECS"), "{err}");
411 : }
412 2 : std::env::remove_var("ANTARES_DRAIN_DELAY_MS");
413 2 : std::env::remove_var("ANTARES_DRAIN_DEADLINE_SECS");
414 2 : }
415 :
416 6 : fn rt() -> tokio::runtime::Runtime {
417 6 : tokio::runtime::Builder::new_current_thread()
418 6 : .enable_all()
419 6 : .build()
420 6 : .expect("runtime")
421 6 : }
422 :
423 6 : fn mem_store() -> antares_sql::store::any::AnyStore {
424 6 : antares_sql::store::any::AnyStore::Mem(antares_sql::store::Store::default())
425 6 : }
426 :
427 : /// A request's follow-up work (the remote leg of a distributed
428 : /// subscription, a forwarded notification) runs after its response; the
429 : /// drain waits for it like it waits for the request itself.
430 : #[test]
431 2 : fn drain_waits_for_request_born_tasks() {
432 2 : rt().block_on(async {
433 2 : let inflight = Arc::new(AtomicUsize::new(0));
434 2 : let done = Arc::new(AtomicBool::new(false));
435 2 : let d = done.clone();
436 2 : antares_api::spawn(async move {
437 2 : tokio::time::sleep(Duration::from_millis(300)).await;
438 2 : d.store(true, Ordering::SeqCst);
439 2 : });
440 2 : let store = mem_store();
441 2 : drain(
442 2 : &inflight,
443 2 : &AtomicUsize::new(0),
444 2 : &store,
445 2 : &antares_store::NoTemporal,
446 2 : Duration::from_secs(5),
447 2 : false,
448 2 : )
449 2 : .await;
450 2 : assert!(
451 2 : done.load(Ordering::SeqCst),
452 : "drain returned before the task finished"
453 : );
454 2 : assert_eq!(antares_api::background_tasks(), 0);
455 2 : });
456 2 : }
457 :
458 : /// Nothing in flight = nothing to wait for: the drain must not sit out
459 : /// its deadline, and the outbox flush must not hang a store that has no
460 : /// outbox (memory/file).
461 : #[test]
462 2 : fn drain_returns_at_once_when_nothing_is_in_flight() {
463 2 : let inflight = Arc::new(AtomicUsize::new(0));
464 2 : let store = mem_store();
465 2 : let started = Instant::now();
466 2 : rt().block_on(drain(
467 2 : &inflight,
468 2 : &AtomicUsize::new(0),
469 2 : &store,
470 2 : &antares_store::NoTemporal,
471 2 : Duration::from_secs(20),
472 : true,
473 : ));
474 2 : assert!(
475 2 : started.elapsed() < Duration::from_secs(1),
476 : "an idle drain waited {:?} — it must not burn the deadline",
477 0 : started.elapsed()
478 : );
479 2 : }
480 :
481 : /// The deadline is a CEILING, not a promise: a connection that never
482 : /// finishes must not hold the process open forever.
483 : #[test]
484 2 : fn drain_gives_up_at_the_deadline_with_a_stuck_connection() {
485 2 : let inflight = Arc::new(AtomicUsize::new(1)); // never released
486 2 : let store = mem_store();
487 2 : let started = Instant::now();
488 2 : rt().block_on(drain(
489 2 : &inflight,
490 2 : &AtomicUsize::new(0),
491 2 : &store,
492 2 : &antares_store::NoTemporal,
493 2 : Duration::from_millis(300),
494 : false,
495 : ));
496 2 : let waited = started.elapsed();
497 2 : assert!(
498 2 : waited >= Duration::from_millis(300),
499 : "the drain must actually wait for in-flight work: {waited:?}"
500 : );
501 2 : assert!(
502 2 : waited < Duration::from_secs(3),
503 : "the drain must give up AT the deadline, not later: {waited:?}"
504 : );
505 2 : }
506 :
507 : /// The outbox flush must WAIT while rows are still pending, and only when
508 : /// it is asked to: a pod whose own drain is off publishes nothing, so
509 : /// waiting there would only burn the deadline. Needs a live database —
510 : /// the outbox table exists on the Pg arm alone, so the memory store can
511 : /// never exercise the wait (it answers an empty page and falls through).
512 : #[tokio::test(flavor = "multi_thread")]
513 : #[ignore = "needs a live database (ANTARES_TEST_DATABASE_URL)"]
514 0 : async fn outbox_flush_waits_for_pending_rows_and_only_when_asked() {
515 : use antares_sql::store::any::{AnyStore, PgBackend};
516 : use antares_sql::store::Kind;
517 0 : let url = std::env::var("ANTARES_TEST_DATABASE_URL")
518 0 : .expect("ANTARES_TEST_DATABASE_URL: this test is asked for by name where a DB exists");
519 : // a nested fn, not a closure: connecting is awaited, and an async
520 : // closure is not a stable language feature
521 0 : async fn connect(url: &str) -> AnyStore {
522 0 : AnyStore::Pg(PgBackend::new(
523 0 : antares_sql::store::pg::connect(url, 5)
524 0 : .await
525 0 : .expect("connect+migrate"),
526 : ))
527 0 : }
528 0 : let run = std::time::SystemTime::now()
529 0 : .duration_since(std::time::UNIX_EPOCH)
530 0 : .expect("clock")
531 0 : .as_millis();
532 0 : let tenant = antares_model::TenantId::new(&format!("drain{run}")).expect("tenant");
533 0 : let id = format!("urn:ngsi-ld:DrainProbe:{run}");
534 0 : let inflight = Arc::new(AtomicUsize::new(0));
535 :
536 : // One committed-but-unpublished row: outbox on, then a write. Nothing
537 : // publishes it here — a unit test wires no bus drain task.
538 0 : let store = connect(&url).await;
539 0 : store.set_outbox(true);
540 0 : store
541 0 : .create(
542 0 : &tenant,
543 0 : Kind::Entity,
544 0 : &id,
545 0 : serde_json::json!({"id": id.as_str(), "type": "DrainProbe"}),
546 0 : )
547 0 : .await
548 0 : .expect("write");
549 0 : let mine: Vec<i64> = store
550 0 : .outbox_peek(500)
551 0 : .await
552 0 : .expect("peek")
553 0 : .into_iter()
554 0 : .filter(|(_, t, _)| t == tenant.as_str())
555 0 : .map(|(seq, ..)| seq)
556 0 : .collect();
557 0 : assert!(
558 0 : !mine.is_empty(),
559 : "the write enqueued no outbox row — the rest of this test would prove nothing"
560 : );
561 :
562 : // Not asked to flush: the pending row may not delay the close at all.
563 0 : let t0 = Instant::now();
564 0 : drain(
565 0 : &inflight,
566 0 : &AtomicUsize::new(0),
567 0 : &store,
568 0 : &antares_store::NoTemporal,
569 0 : Duration::from_millis(400),
570 0 : false,
571 0 : )
572 0 : .await;
573 0 : let closed = t0.elapsed();
574 0 : assert!(
575 0 : closed < Duration::from_millis(250),
576 : "flush_outbox=false waited {closed:?} on a row it never intended to publish"
577 : );
578 :
579 : // Asked to flush: the rows stay pending, so the flush must hold the
580 : // process to its deadline rather than exit on top of them.
581 0 : let store = connect(&url).await;
582 0 : let t0 = Instant::now();
583 0 : drain(
584 0 : &inflight,
585 0 : &AtomicUsize::new(0),
586 0 : &store,
587 0 : &antares_store::NoTemporal,
588 0 : Duration::from_millis(400),
589 0 : true,
590 0 : )
591 0 : .await;
592 0 : let waited = t0.elapsed();
593 0 : assert!(
594 0 : waited >= Duration::from_millis(400),
595 : "the flush returned after {waited:?} with rows still pending"
596 : );
597 0 : assert!(
598 0 : waited < Duration::from_secs(5),
599 : "the flush must give up AT the deadline, not later: {waited:?}"
600 : );
601 :
602 : // Leave the shared table as it was found: ack only our own seqs (a
603 : // blanket ack would delete another test's pending rows) and drop the
604 : // probe entity with the outbox off, so the delete enqueues nothing.
605 0 : let store = connect(&url).await;
606 0 : store
607 0 : .delete(&tenant, Kind::Entity, &id)
608 0 : .await
609 0 : .expect("probe cleanup");
610 0 : store.outbox_ack(&mine).await.expect("outbox cleanup");
611 0 : assert!(
612 0 : store
613 0 : .outbox_peek(500)
614 0 : .await
615 0 : .expect("peek")
616 0 : .into_iter()
617 0 : .all(|(_, t, _)| t != tenant.as_str()),
618 0 : "the test left its own rows in the shared outbox"
619 0 : );
620 0 : }
621 :
622 : /// Step 1 is the flag the health endpoint reads; nothing else may flip it.
623 : #[test]
624 2 : fn begin_flips_the_health_flag() {
625 2 : let draining = Arc::new(AtomicBool::new(false));
626 2 : assert!(!draining.load(Ordering::Relaxed));
627 2 : begin(&draining, Duration::from_millis(500));
628 2 : assert!(draining.load(Ordering::Relaxed), "/q/health must go 503");
629 2 : }
630 : }
|
