rovenue/events/
queue.rs

1// =============================================================
2// queue.rs — durable, process-kill-safe `funnel_*` event queue (spec D4;
3// the `paywall_*` arm went with the rail — only funnel envelopes queue now)
4// =============================================================
5//
6// Mirrors `sessions/{buffer,dispatcher}.rs`'s discipline (peek -> POST ->
7// delete-on-2xx, retain on 5xx/network) but combined into a single type
8// since /v1/events takes one envelope per POST (no batching like
9// /v1/sdk/sessions), and persists into its own bounded (100, drop-oldest)
10// cache-store lane instead of the unbounded-until-flush session buffer.
11//
12// Funnel sessions enqueue pre-stamped envelopes instead of posting inline —
13// an app killed between "funnel shown" and "beacon sent" no longer silently
14// loses the impression. Delivery is at-least-once: the server dedupes on
15// the deterministic event id, so a replayed drain after a crash-before-delete
16// is safe.
17
18use std::sync::atomic::{AtomicBool, Ordering};
19use std::sync::Arc;
20
21use crate::cache::store::QueuedEventRow;
22use crate::cache::CacheStore;
23use crate::error::{RovenueError, RovenueResult};
24use crate::events::{EventEnvelope, EventsClient};
25use crate::logging::{LogLevel, Logger};
26
27/// The event families this queue carries.
28///
29/// Funnels share it rather than getting their own (delivery spec D6): a
30/// second durable queue is a second thing that can lose events, and this
31/// one already survives a process kill. Only `funnel_` envelopes are
32/// accepted now — the `paywall_` arm went with the rail (nothing produces
33/// paywall events any more, and the server rejects them).
34const QUEUED_EVENT_PREFIXES: [&str; 1] = ["funnel_"];
35
36pub struct EventQueue {
37    store: Arc<CacheStore>,
38    events: Arc<EventsClient>,
39    logger: Option<Arc<Logger>>,
40    in_flight: AtomicBool,
41}
42
43impl EventQueue {
44    pub fn new(store: Arc<CacheStore>, events: Arc<EventsClient>) -> Self {
45        Self {
46            store,
47            events,
48            logger: None,
49            in_flight: AtomicBool::new(false),
50        }
51    }
52
53    pub fn with_logger(mut self, logger: Arc<Logger>) -> Self {
54        self.logger = Some(logger);
55        self
56    }
57
58    /// Validate `envelope_json` (must parse as an `EventEnvelope` whose
59    /// `eventType` starts with one of `QUEUED_EVENT_PREFIXES` — anything
60    /// else is rejected as `InvalidArgument`, this is not a
61    /// general-purpose enqueue), persist it
62    /// (bounded at 100 entries, drop-oldest — see
63    /// `CacheStore::append_queued_event`), then trigger a best-effort
64    /// background drain. The caller (funnel sessions emit pre-stamped
65    /// envelopes) is responsible for stamping
66    /// `version`/`eventId`/`subscriberId` before calling this — those
67    /// fields must already be baked into `envelope_json` so a drain that
68    /// runs after a `log_out()` still posts under the identity the event
69    /// was actually attributed to.
70    pub fn enqueue(self: &Arc<Self>, envelope_json: &str) -> RovenueResult<()> {
71        let envelope: EventEnvelope =
72            serde_json::from_str(envelope_json).map_err(|_| RovenueError::InvalidArgument())?;
73        if !QUEUED_EVENT_PREFIXES
74            .iter()
75            .any(|prefix| envelope.event_type.starts_with(prefix))
76        {
77            return Err(RovenueError::InvalidArgument());
78        }
79        self.store.append_queued_event(envelope_json)?;
80        self.trigger_drain();
81        Ok(())
82    }
83
84    /// Spawns a background thread to run [`drain_once`](Self::drain_once).
85    /// Fire-and-forget and non-blocking — callers on the configure()/
86    /// set_foreground() paths must never wait on network I/O here.
87    pub fn trigger_drain(self: &Arc<Self>) {
88        let me = Arc::clone(self);
89        std::thread::spawn(move || me.drain_once());
90    }
91
92    /// Peek (oldest first) -> POST `/v1/events` -> delete on 2xx. On
93    /// connectivity failure / 5xx/429, or a non-poison 4xx (401/403 auth
94    /// blip, 408, etc. — see [`RovenueError::is_poison`]), the entry (and
95    /// everything behind it) is retained and the drain stops —
96    /// at-least-once delivery, safe because the server dedupes on the
97    /// deterministic event id and bounded because the queue itself is
98    /// capped at 100 with drop-oldest. Only a genuine envelope rejection
99    /// (400/422) is poison: retaining it would wedge every later entry
100    /// behind it forever, so it is dropped (loudly logged) and the drain
101    /// continues with the next entry.
102    ///
103    /// Single-in-flight: a drain already running makes this call a no-op —
104    /// the in-progress drain will pick up anything enqueued meanwhile since
105    /// it re-peeks the store on every iteration. There is a narrow window
106    /// between the drain loop seeing an empty queue and the flag actually
107    /// clearing where a concurrent `enqueue()` can land its own no-op
108    /// drain; see the re-arm check below.
109    pub fn drain_once(&self) {
110        if self
111            .in_flight
112            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
113            .is_err()
114        {
115            return;
116        }
117        let stopped_early = self.drain_inner();
118        if stopped_early {
119            // Retained an entry for retry-later (connectivity/5xx/429/
120            // non-poison-4xx) — do not immediately re-attempt, that would
121            // spin instead of waiting for the next trigger.
122            return;
123        }
124
125        // Lost-wakeup guard (Phase-D review finding): the queue looked
126        // empty and the flag is about to clear, but a concurrent
127        // `enqueue()` can append a row and call `trigger_drain()` in the
128        // gap between this thread's loop exiting and the
129        // `in_flight.store(false, ..)` inside `drain_inner()` above — that
130        // spawned drain sees the flag still `true` and no-ops. Nothing is
131        // lost (the row stays in the store), only delayed until the next
132        // trigger. Re-check once and, if work appeared, re-arm a single
133        // extra pass ourselves. Bounded to exactly one re-arm — if another
134        // thread already grabbed the flag first, the CAS below fails and we
135        // simply stop; that thread now owns the new entry.
136        if self.next_row().is_some()
137            && self
138                .in_flight
139                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
140                .is_ok()
141        {
142            self.drain_inner();
143        }
144    }
145
146    /// Runs the peek -> POST -> delete-on-2xx loop until the queue is empty
147    /// or an entry must be retained, then clears `in_flight`. Returns
148    /// `true` if the loop stopped early with an entry retained (retry
149    /// later), `false` if it ran the queue empty naturally.
150    fn drain_inner(&self) -> bool {
151        let mut stopped_early = false;
152        while let Some(row) = self.next_row() {
153            if !self.drain_one(&row) {
154                stopped_early = true;
155                break;
156            }
157        }
158        self.in_flight.store(false, Ordering::SeqCst);
159        stopped_early
160    }
161
162    fn next_row(&self) -> Option<QueuedEventRow> {
163        self.store.list_queued_events(1).ok()?.into_iter().next()
164    }
165
166    /// Returns `true` when the caller should continue to the next entry,
167    /// `false` when the drain should stop (connectivity/5xx — keep this
168    /// entry and everything behind it for the next trigger).
169    fn drain_one(&self, row: &QueuedEventRow) -> bool {
170        let envelope: EventEnvelope = match serde_json::from_str(&row.envelope_json) {
171            Ok(e) => e,
172            Err(_) => {
173                // Can't happen via enqueue()'s own validation, but a
174                // persisted-and-now-unparseable row would wedge the queue
175                // forever if retained — poison-delete defensively.
176                self.log_poison(row.id, None);
177                let _ = self.store.delete_queued_events(&[row.id]);
178                return true;
179            }
180        };
181        match self
182            .events
183            .post(&envelope, envelope.subscriber_id.as_deref())
184        {
185            Ok(()) => {
186                let _ = self.store.delete_queued_events(&[row.id]);
187                true
188            }
189            Err(err) => {
190                if err.is_poison() {
191                    self.log_poison(row.id, err.http_status);
192                    let _ = self.store.delete_queued_events(&[row.id]);
193                    true
194                } else {
195                    false
196                }
197            }
198        }
199    }
200
201    /// A producer could not enqueue `event_type` (storage failure) and the
202    /// event is lost. Producers stay best-effort — this makes the loss
203    /// visible instead of silent.
204    pub(crate) fn log_dropped(&self, event_type: &str, err: &RovenueError) {
205        let Some(logger) = &self.logger else { return };
206        let message = format!("rovenue: dropped {event_type} event, could not enqueue it: {err:?}");
207        let event_type = event_type.to_string();
208        logger.log(
209            LogLevel::Warn,
210            move || message,
211            move || {
212                let mut f = std::collections::HashMap::new();
213                f.insert("op".to_string(), "event_queue_enqueue".to_string());
214                f.insert("event_type".to_string(), event_type);
215                f
216            },
217        );
218    }
219
220    fn log_poison(&self, id: i64, http_status: Option<u16>) {
221        let Some(logger) = &self.logger else { return };
222        let message = format!(
223            "rovenue: dropping poisoned queued event id={id} (permanent rejection, http_status={http_status:?}) — retaining it would wedge the queue"
224        );
225        logger.log(
226            LogLevel::Error,
227            move || message,
228            move || {
229                let mut f = std::collections::HashMap::new();
230                f.insert("op".to_string(), "event_queue_drain".to_string());
231                if let Some(s) = http_status {
232                    f.insert("http_status".to_string(), s.to_string());
233                }
234                f
235            },
236        );
237    }
238
239    /// Test-only: number of buffered (undrained) queued events.
240    #[doc(hidden)]
241    pub fn test_queue_len(&self) -> usize {
242        self.store
243            .list_queued_events(usize::MAX)
244            .map(|r| r.len())
245            .unwrap_or(0)
246    }
247}
248
249#[cfg(test)]
250mod tests {
251    use super::*;
252    use crate::transport::http_client::HttpClient;
253
254    fn envelope_json(event_id: &str, event_type: &str) -> String {
255        format!(
256            r#"{{"eventType":"{event_type}","occurredAt":"2026-05-28T10:00:00Z","eventId":"{event_id}","subscriberId":"rov_test"}}"#
257        )
258    }
259
260    fn setup(base_url: String) -> (Arc<CacheStore>, Arc<EventQueue>) {
261        let store = Arc::new(CacheStore::open_in_memory().unwrap());
262        setup_with_store(base_url, store)
263    }
264
265    fn setup_with_store(
266        base_url: String,
267        store: Arc<CacheStore>,
268    ) -> (Arc<CacheStore>, Arc<EventQueue>) {
269        let http = Arc::new(HttpClient::new(base_url, "pk_test".to_string()).with_max_attempts(1));
270        let events = Arc::new(EventsClient::new(http));
271        let queue = Arc::new(EventQueue::new(Arc::clone(&store), events));
272        (store, queue)
273    }
274
275    // -----------------------------------------------------------
276    // Validation
277    // -----------------------------------------------------------
278
279    #[test]
280    fn enqueue_rejects_malformed_json() {
281        let (_store, queue) = setup("http://127.0.0.1:1".to_string());
282        let err = queue.enqueue("not json").unwrap_err();
283        assert_eq!(err.kind, crate::error::ErrorKind::InvalidArgument);
284    }
285
286    #[test]
287    fn enqueue_rejects_non_funnel_event_types() {
288        let (_store, queue) = setup("http://127.0.0.1:1".to_string());
289        let err = queue
290            .enqueue(&envelope_json("evt_1", "purchase"))
291            .unwrap_err();
292        assert_eq!(err.kind, crate::error::ErrorKind::InvalidArgument);
293        assert_eq!(queue.test_queue_len(), 0);
294    }
295
296    #[test]
297    fn enqueue_accepts_funnel_prefixed_event_types() {
298        let (store, queue) = setup("http://127.0.0.1:1".to_string());
299        queue
300            .enqueue(&envelope_json("evt_1", "funnel_session_started"))
301            .unwrap();
302        queue
303            .enqueue(&envelope_json("evt_2", "funnel_session_completed"))
304            .unwrap();
305        assert_eq!(store.list_queued_events(10).unwrap().len(), 2);
306    }
307
308    // -----------------------------------------------------------
309    // Drain semantics (call drain_once() directly — deterministic, no
310    // background-thread races; enqueue()'s own trigger_drain() is exercised
311    // separately below).
312    // -----------------------------------------------------------
313
314    #[test]
315    fn drain_once_deletes_after_successful_post() {
316        let mut server = mockito::Server::new();
317        let m = server
318            .mock("POST", "/v1/events")
319            .with_status(202)
320            .with_header("content-type", "application/json")
321            .create();
322
323        let (store, queue) = setup(server.url());
324        store
325            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
326            .unwrap();
327
328        queue.drain_once();
329
330        m.assert();
331        assert_eq!(store.list_queued_events(10).unwrap().len(), 0);
332    }
333
334    #[test]
335    fn drain_once_retains_and_stops_on_503() {
336        let mut server = mockito::Server::new();
337        let m1 = server
338            .mock("POST", "/v1/events")
339            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_1""#.into()))
340            .with_status(503)
341            .create();
342        // A second, later entry that WOULD succeed if attempted — proves the
343        // drain stopped after the first failure rather than skipping ahead.
344        let m2 = server
345            .mock("POST", "/v1/events")
346            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_2""#.into()))
347            .with_status(202)
348            .expect(0)
349            .create();
350
351        let (store, queue) = setup(server.url());
352        store
353            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
354            .unwrap();
355        store
356            .append_queued_event(&envelope_json("evt_2", "funnel_session_completed"))
357            .unwrap();
358
359        queue.drain_once();
360
361        m1.assert();
362        m2.assert();
363        let remaining = store.list_queued_events(10).unwrap();
364        assert_eq!(
365            remaining.len(),
366            2,
367            "both entries retained — the drain must stop, not skip ahead"
368        );
369        assert!(remaining[0].envelope_json.contains("evt_1"));
370        assert!(remaining[1].envelope_json.contains("evt_2"));
371    }
372
373    #[test]
374    fn drain_once_retains_and_stops_on_network_error() {
375        // Nothing listening on this port — a connection error, not an HTTP
376        // status, exercising the "network" half of "5xx/network".
377        let (store, queue) = setup("http://127.0.0.1:1".to_string());
378        store
379            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
380            .unwrap();
381
382        queue.drain_once();
383
384        assert_eq!(store.list_queued_events(10).unwrap().len(), 1);
385    }
386
387    #[test]
388    fn drain_once_poison_deletes_4xx_and_continues() {
389        let mut server = mockito::Server::new();
390        let m1 = server
391            .mock("POST", "/v1/events")
392            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_poison""#.into()))
393            .with_status(400)
394            .create();
395        let m2 = server
396            .mock("POST", "/v1/events")
397            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_ok""#.into()))
398            .with_status(202)
399            .create();
400
401        let (store, queue) = setup(server.url());
402        store
403            .append_queued_event(&envelope_json("evt_poison", "funnel_session_started"))
404            .unwrap();
405        store
406            .append_queued_event(&envelope_json("evt_ok", "funnel_session_completed"))
407            .unwrap();
408
409        queue.drain_once();
410
411        m1.assert();
412        m2.assert();
413        assert_eq!(
414            store.list_queued_events(10).unwrap().len(),
415            0,
416            "the poisoned entry must not wedge the good entry behind it"
417        );
418    }
419
420    /// Regression (Phase-D review finding 1): a 401 (invalid/rotated API
421    /// key) must be RETAINED, not poison-deleted — a key rotation or
422    /// transient auth failure must not permanently drop queued beacons.
423    #[test]
424    fn drain_once_retains_and_stops_on_401() {
425        let mut server = mockito::Server::new();
426        let m1 = server
427            .mock("POST", "/v1/events")
428            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_1""#.into()))
429            .with_status(401)
430            .create();
431        let m2 = server
432            .mock("POST", "/v1/events")
433            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_2""#.into()))
434            .with_status(202)
435            .expect(0)
436            .create();
437
438        let (store, queue) = setup(server.url());
439        store
440            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
441            .unwrap();
442        store
443            .append_queued_event(&envelope_json("evt_2", "funnel_session_completed"))
444            .unwrap();
445
446        queue.drain_once();
447
448        m1.assert();
449        m2.assert();
450        let remaining = store.list_queued_events(10).unwrap();
451        assert_eq!(
452            remaining.len(),
453            2,
454            "a 401 must retain the entry (and everything behind it), not poison-delete it"
455        );
456    }
457
458    /// Regression (Phase-D review finding 1): same as the 401 case — 403
459    /// (forbidden) is also an auth-adjacent failure, not a malformed
460    /// envelope, so it must be retained too.
461    #[test]
462    fn drain_once_retains_and_stops_on_403() {
463        let mut server = mockito::Server::new();
464        let m = server
465            .mock("POST", "/v1/events")
466            .with_status(403)
467            .expect(1)
468            .create();
469
470        let (store, queue) = setup(server.url());
471        store
472            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
473            .unwrap();
474
475        queue.drain_once();
476
477        m.assert();
478        assert_eq!(
479            store.list_queued_events(10).unwrap().len(),
480            1,
481            "a 403 must retain the entry, not poison-delete it"
482        );
483    }
484
485    /// Regression (Phase-D review finding 1): 408 (request timeout) is
486    /// non-retryable per `ErrorKind::is_retryable()` (it maps to the
487    /// generic `InvalidRequest` kind) but is not a malformed-payload
488    /// rejection either — it must be retained, matching the review's
489    /// explicit call-out.
490    #[test]
491    fn drain_once_retains_and_stops_on_408() {
492        let mut server = mockito::Server::new();
493        let m = server
494            .mock("POST", "/v1/events")
495            .with_status(408)
496            .expect(1)
497            .create();
498
499        let (store, queue) = setup(server.url());
500        store
501            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
502            .unwrap();
503
504        queue.drain_once();
505
506        m.assert();
507        assert_eq!(
508            store.list_queued_events(10).unwrap().len(),
509            1,
510            "a 408 must retain the entry, not poison-delete it"
511        );
512    }
513
514    /// 422 (unprocessable entity) is a genuine envelope rejection — same
515    /// bucket as 400 — and must still poison-delete after the narrowed
516    /// predicate.
517    #[test]
518    fn drain_once_poison_deletes_422() {
519        let mut server = mockito::Server::new();
520        let m1 = server
521            .mock("POST", "/v1/events")
522            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_poison""#.into()))
523            .with_status(422)
524            .create();
525        let m2 = server
526            .mock("POST", "/v1/events")
527            .match_body(mockito::Matcher::Regex(r#""eventId":"evt_ok""#.into()))
528            .with_status(202)
529            .create();
530
531        let (store, queue) = setup(server.url());
532        store
533            .append_queued_event(&envelope_json("evt_poison", "funnel_session_started"))
534            .unwrap();
535        store
536            .append_queued_event(&envelope_json("evt_ok", "funnel_session_completed"))
537            .unwrap();
538
539        queue.drain_once();
540
541        m1.assert();
542        m2.assert();
543        assert_eq!(
544            store.list_queued_events(10).unwrap().len(),
545            0,
546            "422 is a genuine envelope rejection and must still be poison-deleted"
547        );
548    }
549
550    #[test]
551    fn drain_once_processes_oldest_first() {
552        let mut server = mockito::Server::new();
553        let seen = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
554        let seen_clone = Arc::clone(&seen);
555        let m = server
556            .mock("POST", "/v1/events")
557            .with_status(202)
558            .with_body_from_request(move |req| {
559                let empty = Vec::new();
560                let body = String::from_utf8_lossy(req.body().unwrap_or(&empty)).to_string();
561                seen_clone.lock().unwrap().push(body);
562                Vec::new()
563            })
564            .expect(3)
565            .create();
566
567        let (store, queue) = setup(server.url());
568        store
569            .append_queued_event(&envelope_json("evt_a", "funnel_session_started"))
570            .unwrap();
571        store
572            .append_queued_event(&envelope_json("evt_b", "funnel_session_completed"))
573            .unwrap();
574        store
575            .append_queued_event(&envelope_json("evt_c", "funnel_session_started"))
576            .unwrap();
577
578        queue.drain_once();
579
580        m.assert();
581        let order = seen.lock().unwrap();
582        assert_eq!(order.len(), 3);
583        assert!(order[0].contains("evt_a"));
584        assert!(order[1].contains("evt_b"));
585        assert!(order[2].contains("evt_c"));
586    }
587
588    // -----------------------------------------------------------
589    // Single in-flight
590    // -----------------------------------------------------------
591
592    #[test]
593    fn drain_once_is_a_noop_when_already_in_flight() {
594        let mut server = mockito::Server::new();
595        let m = server
596            .mock("POST", "/v1/events")
597            .with_status(202)
598            .expect(0)
599            .create();
600
601        let (store, queue) = setup(server.url());
602        store
603            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
604            .unwrap();
605
606        // Simulate a drain already in progress.
607        queue.in_flight.store(true, Ordering::SeqCst);
608        queue.drain_once();
609        m.assert();
610        assert_eq!(
611            store.list_queued_events(10).unwrap().len(),
612            1,
613            "a concurrent drain must not touch the queue"
614        );
615
616        // Once the "in-progress" drain finishes, the next trigger works normally.
617        queue.in_flight.store(false, Ordering::SeqCst);
618        queue.drain_once();
619        assert_eq!(store.list_queued_events(10).unwrap().len(), 0);
620    }
621
622    // -----------------------------------------------------------
623    // Lost-wakeup re-arm (Phase-D review finding 2)
624    //
625    // The actual race — a concurrent enqueue() landing in the handful of
626    // CPU instructions between drain_inner()'s loop deciding the queue is
627    // empty and `in_flight` actually clearing — has no I/O or scheduling
628    // point in between for a real OS thread to land in, so it cannot be
629    // reproduced deterministically via real concurrency (any attempt would
630    // be flaky at best, since Rust gives no hook to pause a thread exactly
631    // there). These are therefore DIRECT unit tests of the re-arm logic:
632    // they drive `drain_inner()`/`next_row()`/the `in_flight` CAS by hand,
633    // the same private primitives `drain_once()` itself composes, with the
634    // "concurrent enqueue" interleaved at exactly the point the gap opens.
635    // -----------------------------------------------------------
636
637    #[test]
638    fn rearm_picks_up_entry_that_appears_right_after_the_first_pass_finds_the_queue_empty() {
639        let mut server = mockito::Server::new();
640        let m = server.mock("POST", "/v1/events").with_status(202).create();
641        let (store, queue) = setup(server.url());
642
643        // Step 1 of drain_once(): win the CAS, exactly as a real drain would.
644        assert!(queue
645            .in_flight
646            .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
647            .is_ok());
648
649        // Step 2: run the first drain_inner() pass while the queue is
650        // still empty — mirrors a drain that wins the CAS a beat before a
651        // concurrent enqueue() lands. Must report a natural (non-retained)
652        // exit and clear the flag.
653        assert!(
654            !queue.drain_inner(),
655            "an empty queue must not report stopped_early"
656        );
657        assert!(!queue.in_flight.load(Ordering::SeqCst));
658
659        // Step 3: the "concurrent enqueue()" lands in the gap. Pre-fix,
660        // its own trigger_drain() would have been the ONLY thing that
661        // could pick this up — and pre-fix, that's a no-op window this
662        // finding is about (append happening exactly while in_flight was
663        // still true from this thread's point of view). We've modeled
664        // that moment as already past, so the append here is exactly what
665        // the re-arm check must catch instead.
666        store
667            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
668            .unwrap();
669
670        // Step 4: replay drain_once()'s post-drain_inner re-arm check by
671        // hand — this is the exact logic drain_once() runs after its own
672        // drain_inner() call returns `false`.
673        assert!(queue.next_row().is_some(), "the row must be visible");
674        assert!(
675            queue
676                .in_flight
677                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
678                .is_ok(),
679            "re-arm CAS must succeed — nothing else holds the flag here"
680        );
681        queue.drain_inner();
682
683        m.assert();
684        assert_eq!(
685            store.list_queued_events(10).unwrap().len(),
686            0,
687            "the re-arm pass must deliver work that appeared right after the \
688             first pass found the queue empty, without waiting for another trigger"
689        );
690        assert!(!queue.in_flight.load(Ordering::SeqCst));
691    }
692
693    #[test]
694    fn drain_once_end_to_end_rearms_when_an_entry_is_already_queued() {
695        // Complement to the direct test above: exercises the real, public
696        // `drain_once()` entry point (not hand-driven) to prove the
697        // production re-arm branch — not just the primitives — delivers
698        // an entry that's present by the time the natural-empty check
699        // runs. With the queue already non-empty at call time this always
700        // takes the first-pass path today, but pins the end-to-end
701        // contract so a future refactor of `drain_once()`'s two-pass shape
702        // can't silently break delivery.
703        let mut server = mockito::Server::new();
704        let m = server.mock("POST", "/v1/events").with_status(202).create();
705        let (store, queue) = setup(server.url());
706        store
707            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
708            .unwrap();
709
710        queue.drain_once();
711
712        m.assert();
713        assert_eq!(store.list_queued_events(10).unwrap().len(), 0);
714        assert!(!queue.in_flight.load(Ordering::SeqCst));
715    }
716
717    /// Bounded re-arm (Phase-D review finding 2): when the first pass
718    /// stops early because an entry must be RETAINED (retry-later —
719    /// connectivity/5xx/429/non-poison-4xx), `drain_once()` must NOT
720    /// re-arm and immediately retry — that would spin instead of waiting
721    /// for the next explicit trigger. Proven here by asserting exactly one
722    /// POST attempt even though the retained entry is still present (and
723    /// therefore `next_row()` would return `Some` if the re-arm check ran).
724    #[test]
725    fn drain_once_does_not_rearm_after_a_retry_later_stop() {
726        let mut server = mockito::Server::new();
727        let m = server
728            .mock("POST", "/v1/events")
729            .with_status(503)
730            .expect(1)
731            .create();
732
733        let (store, queue) = setup(server.url());
734        store
735            .append_queued_event(&envelope_json("evt_1", "funnel_session_started"))
736            .unwrap();
737
738        queue.drain_once();
739
740        m.assert();
741        assert_eq!(
742            store.list_queued_events(10).unwrap().len(),
743            1,
744            "retained entry must still be present — proves no spin-retry happened"
745        );
746        assert!(
747            !queue.in_flight.load(Ordering::SeqCst),
748            "the flag must still be cleared so the next explicit trigger can run"
749        );
750    }
751
752    // -----------------------------------------------------------
753    // Kill-safety: enqueue via one store instance, drain via a fresh one
754    // opened against the SAME on-disk backing file — proves the queue
755    // survives a process restart, not just an in-memory struct's lifetime.
756    // -----------------------------------------------------------
757
758    #[test]
759    fn kill_safety_enqueue_survives_reopening_the_store() {
760        let dir = tempfile::tempdir().unwrap();
761        let path = dir.path().join("event_queue_test.sqlite3");
762
763        {
764            let store = Arc::new(CacheStore::open(&path).unwrap());
765            let (_s, queue) = setup_with_store("http://127.0.0.1:1".to_string(), store);
766            queue
767                .enqueue(&envelope_json("evt_1", "funnel_session_started"))
768                .unwrap();
769            // `queue`/`store` drop here — simulates the process dying before
770            // the background drain (spawned by enqueue()'s trigger_drain())
771            // gets to run.
772        }
773
774        let mut server = mockito::Server::new();
775        let m = server.mock("POST", "/v1/events").with_status(202).create();
776        let store2 = Arc::new(CacheStore::open(&path).unwrap());
777        assert_eq!(
778            store2.list_queued_events(10).unwrap().len(),
779            1,
780            "the entry must be visible to a brand-new store instance over the same file"
781        );
782        let (_s2, queue2) = setup_with_store(server.url(), store2);
783        queue2.drain_once();
784
785        m.assert();
786        assert_eq!(_s2.list_queued_events(10).unwrap().len(), 0);
787    }
788
789    // -----------------------------------------------------------
790    // enqueue() triggers a background drain (integration-ish; polls with a
791    // bounded timeout, consistent with this crate's existing style for
792    // asserting on background-thread work — see polling/scheduler.rs).
793    // -----------------------------------------------------------
794
795    #[test]
796    fn enqueue_triggers_a_background_drain() {
797        let mut server = mockito::Server::new();
798        let m = server.mock("POST", "/v1/events").with_status(202).create();
799        let (store, queue) = setup(server.url());
800
801        queue
802            .enqueue(&envelope_json("evt_1", "funnel_session_started"))
803            .unwrap();
804
805        let mut drained = false;
806        for _ in 0..50 {
807            if store.list_queued_events(10).unwrap().is_empty() {
808                drained = true;
809                break;
810            }
811            std::thread::sleep(std::time::Duration::from_millis(20));
812        }
813        assert!(drained, "enqueue() must trigger a background drain");
814        m.assert();
815    }
816}
rovenue/events/queue.rs

rovenue/events/
queue.rs
