rovenue/sessions/
dispatcher.rs

1use std::sync::Arc;
2use std::time::Duration;
3
4use crate::error::RovenueResult;
5use crate::polling::PollingScheduler;
6use crate::transport::http_client::HttpClient;
7
8use super::buffer::SessionBuffer;
9
10/// Most events one POST carries.
11const BATCH_LIMIT: usize = 200;
12
13pub struct SessionDispatcher {
14    buffer: Arc<SessionBuffer>,
15    http: Arc<HttpClient>,
16    /// Captured at configure() time from the host app's bundle/PackageInfo.
17    /// Serialized as `""` when None to preserve the pre-0.7 wire format
18    /// the backend expects.
19    app_version: Option<String>,
20}
21
22impl SessionDispatcher {
23    pub fn new(
24        buffer: Arc<SessionBuffer>,
25        http: Arc<HttpClient>,
26        app_version: Option<String>,
27    ) -> Self {
28        Self {
29            buffer,
30            http,
31            app_version,
32        }
33    }
34
35    /// Send every buffered event, oldest first, each under the identity it
36    /// was recorded under (a log out in between changes nothing). One POST
37    /// carries the leading run of up to 200 rows that share a `rovenue_id`;
38    /// runs are sent in order until the buffer is empty or a POST fails.
39    ///
40    /// Rows are deleted ONLY after the server confirms receipt (2xx). On a
41    /// 503 (Kafka down) or network error the run is retained and retried on
42    /// the next tick — the API returns 503 precisely to signal "keep it".
43    /// This is at-least-once delivery; the server derives a deterministic
44    /// eventId from the event's stable fields so a replayed batch dedupes in
45    /// ClickHouse. Growth is bounded by the buffer's FIFO cap (newest 1000 on
46    /// append). Returns the number of events sent.
47    pub fn flush_once(&self) -> RovenueResult<usize> {
48        let mut sent = 0;
49        loop {
50            let rows = self.buffer.peek(BATCH_LIMIT)?;
51            let Some(first) = rows.first() else {
52                return Ok(sent);
53            };
54            let sub_id = first.rovenue_id.clone();
55            let run: Vec<_> = rows
56                .into_iter()
57                .take_while(|r| r.rovenue_id == sub_id)
58                .collect();
59            let ids: Vec<i64> = run.iter().map(|r| r.id).collect();
60            // Unattributable (never expected: the identity is minted at
61            // configure, before anything can be recorded). Kept, it would
62            // block every later row behind it.
63            if sub_id.is_empty() {
64                self.buffer.delete(&ids)?;
65                continue;
66            }
67            let app_version = self.app_version.as_deref().unwrap_or("");
68            let events: Vec<_> = run
69                .iter()
70                .map(|r| {
71                    serde_json::json!({
72                        "type": r.kind,
73                        "occurredAt": r.occurred_at,
74                        "durationMs": r.duration_ms,
75                        "appVersion": app_version,
76                        "sdkVersion": crate::version::SDK_VERSION,
77                    })
78                })
79                .collect();
80            // Propagates on failure (5xx/network/429, or a non-poison 4xx —
81            // e.g. 401/403 auth blip, 408 — see `RovenueError::is_poison`)
82            // WITHOUT deleting → retained. A genuine envelope rejection
83            // (400/422) is the one exception: peek returns id ASC, so
84            // retaining a run the server will never accept re-sends it every
85            // tick and head-of-line-blocks all newer telemetry until the FIFO
86            // cap evicts it. Drop it instead. Auth failures/key rotation must
87            // NOT drop telemetry — they're retried on the next tick until the
88            // key is fixed (shared predicate with `events::queue`, the durable
89            // event queue's twin drain site — see Phase-D review
90            // finding 1).
91            if let Err(err) = self.http.post_sessions(&sub_id, &events) {
92                if err.is_poison() {
93                    self.buffer.delete(&ids)?;
94                }
95                return Err(err);
96            }
97            self.buffer.delete(&ids)?;
98            sent += run.len();
99        }
100    }
101
102    pub fn start(self: Arc<Self>, scheduler: &PollingScheduler) {
103        let me = Arc::clone(&self);
104        scheduler.register("sessions", Duration::from_secs(30), move || {
105            let _ = me.flush_once();
106        });
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113    use crate::cache::CacheStore;
114    use crate::sessions::SessionEventKind;
115
116    fn setup(base_url: String) -> (Arc<SessionBuffer>, Arc<SessionDispatcher>) {
117        let store = Arc::new(CacheStore::open_in_memory().unwrap());
118        let buffer = Arc::new(SessionBuffer::new(Arc::clone(&store)));
119        // max_attempts=1 so a 503 fails fast without exponential backoff.
120        let http = Arc::new(HttpClient::new(base_url, "pk_test".to_string()).with_max_attempts(1));
121        let dispatcher = Arc::new(SessionDispatcher::new(
122            Arc::clone(&buffer),
123            http,
124            Some("1.0.0".to_string()),
125        ));
126        (buffer, dispatcher)
127    }
128
129    /// Regression (P1): on a 503 (Kafka down) / network error the batch must be
130    /// RETAINED for the next flush — the API returns 503 precisely so the SDK
131    /// keeps it. Previously flush drained-then-discarded, losing the events.
132    #[test]
133    fn retains_events_when_server_returns_503() {
134        let mut server = mockito::Server::new();
135        let _m = server
136            .mock("POST", "/v1/sdk/sessions")
137            .with_status(503)
138            .with_body(r#"{"error":{"code":"TELEMETRY_UNAVAILABLE"}}"#)
139            .create();
140
141        let (buffer, dispatcher) = setup(server.url());
142        buffer
143            .record(
144                "rov_test",
145                SessionEventKind::Open,
146                "2026-05-28T10:00:00Z",
147                None,
148            )
149            .unwrap();
150
151        // The flush fails; the events must NOT be deleted.
152        let result = dispatcher.flush_once();
153        assert!(result.is_err(), "a 503 must surface as an error");
154        assert_eq!(
155            buffer.peek(100).unwrap().len(),
156            1,
157            "events must be retained for retry after a 503"
158        );
159    }
160
161    /// A permanently-rejected batch (4xx other than 429) must be DISCARDED,
162    /// not retained: peek returns rows in id ASC order, so a retained
163    /// poison batch would be re-sent on every tick and head-of-line-block
164    /// all newer telemetry behind it until the FIFO cap evicts it.
165    #[test]
166    fn discards_batch_on_permanent_4xx_rejection() {
167        let mut server = mockito::Server::new();
168        let _m = server
169            .mock("POST", "/v1/sdk/sessions")
170            .with_status(400)
171            .with_body(r#"{"error":{"code":"VALIDATION_ERROR"}}"#)
172            .create();
173
174        let (buffer, dispatcher) = setup(server.url());
175        buffer
176            .record(
177                "rov_test",
178                SessionEventKind::Open,
179                "2026-05-28T10:00:00Z",
180                None,
181            )
182            .unwrap();
183
184        let result = dispatcher.flush_once();
185        assert!(result.is_err(), "a 400 must still surface as an error");
186        assert_eq!(
187            buffer.peek(100).unwrap().len(),
188            0,
189            "a 4xx-rejected batch must be dropped, not poison the queue"
190        );
191    }
192
193    /// Regression (Phase-D review finding 1): a 401 (invalid/rotated API
194    /// key) must be RETAINED, not dropped — a key rotation or transient
195    /// auth failure must not permanently drop queued session telemetry.
196    #[test]
197    fn retains_batch_on_401() {
198        let mut server = mockito::Server::new();
199        let _m = server
200            .mock("POST", "/v1/sdk/sessions")
201            .with_status(401)
202            .with_body(r#"{"error":{"code":"INVALID_API_KEY"}}"#)
203            .create();
204
205        let (buffer, dispatcher) = setup(server.url());
206        buffer
207            .record(
208                "rov_test",
209                SessionEventKind::Open,
210                "2026-05-28T10:00:00Z",
211                None,
212            )
213            .unwrap();
214
215        let result = dispatcher.flush_once();
216        assert!(result.is_err(), "a 401 must still surface as an error");
217        assert_eq!(
218            buffer.peek(100).unwrap().len(),
219            1,
220            "a 401 must retain the batch, not drop it"
221        );
222    }
223
224    /// Same as the 401 case — 403 (forbidden) is auth-adjacent, not a
225    /// malformed batch, so it must be retained too.
226    #[test]
227    fn retains_batch_on_403() {
228        let mut server = mockito::Server::new();
229        let _m = server
230            .mock("POST", "/v1/sdk/sessions")
231            .with_status(403)
232            .with_body(r#"{"error":{"code":"FORBIDDEN"}}"#)
233            .create();
234
235        let (buffer, dispatcher) = setup(server.url());
236        buffer
237            .record(
238                "rov_test",
239                SessionEventKind::Open,
240                "2026-05-28T10:00:00Z",
241                None,
242            )
243            .unwrap();
244
245        let result = dispatcher.flush_once();
246        assert!(result.is_err(), "a 403 must still surface as an error");
247        assert_eq!(
248            buffer.peek(100).unwrap().len(),
249            1,
250            "a 403 must retain the batch, not drop it"
251        );
252    }
253
254    /// 408 (request timeout) is non-retryable per
255    /// `ErrorKind::is_retryable()` (maps to the generic `InvalidRequest`
256    /// kind) but is not a malformed-batch rejection either — must be
257    /// retained, matching the review's explicit call-out.
258    #[test]
259    fn retains_batch_on_408() {
260        let mut server = mockito::Server::new();
261        let _m = server
262            .mock("POST", "/v1/sdk/sessions")
263            .with_status(408)
264            .create();
265
266        let (buffer, dispatcher) = setup(server.url());
267        buffer
268            .record(
269                "rov_test",
270                SessionEventKind::Open,
271                "2026-05-28T10:00:00Z",
272                None,
273            )
274            .unwrap();
275
276        let result = dispatcher.flush_once();
277        assert!(result.is_err(), "a 408 must still surface as an error");
278        assert_eq!(
279            buffer.peek(100).unwrap().len(),
280            1,
281            "a 408 must retain the batch, not drop it"
282        );
283    }
284
285    /// 422 (unprocessable entity) is a genuine envelope rejection — same
286    /// bucket as 400 — and must still be dropped after the narrowed
287    /// predicate.
288    #[test]
289    fn discards_batch_on_422() {
290        let mut server = mockito::Server::new();
291        let _m = server
292            .mock("POST", "/v1/sdk/sessions")
293            .with_status(422)
294            .with_body(r#"{"error":{"code":"VALIDATION_ERROR"}}"#)
295            .create();
296
297        let (buffer, dispatcher) = setup(server.url());
298        buffer
299            .record(
300                "rov_test",
301                SessionEventKind::Open,
302                "2026-05-28T10:00:00Z",
303                None,
304            )
305            .unwrap();
306
307        let result = dispatcher.flush_once();
308        assert!(result.is_err(), "a 422 must still surface as an error");
309        assert_eq!(
310            buffer.peek(100).unwrap().len(),
311            0,
312            "422 is a genuine envelope rejection and must still be dropped"
313        );
314    }
315
316    #[test]
317    fn deletes_events_after_successful_post() {
318        let mut server = mockito::Server::new();
319        let _m = server
320            .mock("POST", "/v1/sdk/sessions")
321            .with_status(200)
322            .with_header("content-type", "application/json")
323            .with_body("{}")
324            .create();
325
326        let (buffer, dispatcher) = setup(server.url());
327        buffer
328            .record(
329                "rov_test",
330                SessionEventKind::Open,
331                "2026-05-28T10:00:00Z",
332                None,
333            )
334            .unwrap();
335
336        let flushed = dispatcher.flush_once().unwrap();
337        assert_eq!(flushed, 1);
338        assert_eq!(
339            buffer.peek(100).unwrap().len(),
340            0,
341            "events must be removed only after the server confirms receipt"
342        );
343    }
344}
rovenue/sessions/dispatcher.rs

rovenue/sessions/
dispatcher.rs
