opentracker

opentracker/crates/otserver/src/writer.rs
writer.rs
⎇
Raw
1//! The single writer task.
2//!
3//! Everything that writes to SQLite sends a [`WriteOp`] down an `mpsc` channel.
4//! One task drains it, batching everything that arrives within 250 ms (or 512
5//! operations, whichever comes first) into one transaction.
6//!
7//! Three properties follow, and all three matter:
8//!
9//! * **No `SQLITE_BUSY`, ever.** There is exactly one writer, so there is nothing
10//! to contend with.
11//! * **fsyncs are amortised.** 100 devices reporting once a minute is ~4
12//! transactions per second, not 100.
13//! * **The UDP loop never blocks on the database.** A full channel is answered
14//! with `THROTTLE` and the datagram is dropped; the client will retry. Applying
15//! backpressure to the receive loop instead would turn a storage stall into
16//! packet loss for *every* device, including the ones the server could still
17//! serve.
18
19use std::time::Duration;
20
21use anyhow::{Context, Result};
22use otproto::Point;
23use sqlx::{Sqlite, SqlitePool, Transaction};
24use tokio::sync::mpsc;
25use tracing::{debug, error, warn};
26
27/// Channel depth. Deep enough to absorb a burst of queue flushes from every
28/// device at once, shallow enough that a stalled disk is noticed in seconds
29/// rather than after the process has eaten a gigabyte of positions.
30const CHANNEL_DEPTH: usize = 8192;
31
32/// Longest a write waits to be committed.
33const BATCH_WINDOW: Duration = Duration::from_millis(250);
34
35/// Most operations in one transaction.
36const BATCH_MAX: usize = 512;
37
38#[derive(Debug)]
39pub enum WriteOp {
40 /// Points from one authenticated `LOC`, all belonging to one account.
41 Points {
42 user_id: i64,
43 /// Provenance only.
44 src_token_id: i64,
45 points: Vec<Point>,
46 recv_at: i64,
47 },
48 /// Per-datagram telemetry for a token. Written on every authenticated
49 /// packet, which is why it must be batched rather than done inline.
50 TokenSeen {
51 token_id: i64,
52 at: i64,
53 src_ip: String,
54 src_port: u16,
55 transport: &'static str,
56 },
57 /// Recorded from `HELLO`.
58 TokenHello {
59 token_id: i64,
60 app_version: i64,
61 os_api_level: i64,
62 },
63 Audit {
64 user_id: Option<i64>,
65 at: i64,
66 action: String,
67 detail: String,
68 src_ip: Option<String>,
69 },
70}
71
72/// Send handle. Cheap to clone; hand one to every task that needs to write.
73#[derive(Clone)]
74pub struct WriteHandle {
75 tx: mpsc::Sender<WriteOp>,
76}
77
78/// Whether a write was accepted.
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum Accepted {
81 Yes,
82 /// The channel is full. The caller should signal `THROTTLE` and drop.
83 Saturated,
84 /// The writer task is gone — we are shutting down.
85 Closed,
86}
87
88impl WriteHandle {
89 /// Never awaits, never blocks. This is what the UDP loop calls.
90 pub fn try_send(&self, op: WriteOp) -> Accepted {
91 match self.tx.try_send(op) {
92 Ok(()) => Accepted::Yes,
93 Err(mpsc::error::TrySendError::Full(_)) => Accepted::Saturated,
94 Err(mpsc::error::TrySendError::Closed(_)) => Accepted::Closed,
95 }
96 }
97
98 /// For HTTP handlers, which can afford to wait for a slot.
99 pub async fn send(&self, op: WriteOp) -> Result<()> {
100 self.tx.send(op).await.context("writer task has stopped")
101 }
102
103 /// Approximate free capacity, for `/metrics` and for deciding when to warn.
104 pub fn capacity(&self) -> usize {
105 self.tx.capacity()
106 }
107}
108
109/// Start the writer task. Returns the handle and a join handle for shutdown.
110pub fn spawn(pool: SqlitePool) -> (WriteHandle, tokio::task::JoinHandle<()>) {
111 let (tx, rx) = mpsc::channel(CHANNEL_DEPTH);
112 let task = tokio::spawn(run(pool, rx));
113 (WriteHandle { tx }, task)
114}
115
116async fn run(pool: SqlitePool, mut rx: mpsc::Receiver<WriteOp>) {
117 let mut batch: Vec<WriteOp> = Vec::with_capacity(BATCH_MAX);
118
119 loop {
120 // Block until there is something to do — no idle polling.
121 let Some(first) = rx.recv().await else {
122 break; // all senders dropped: shutdown
123 };
124 batch.push(first);
125
126 // Then take whatever else shows up inside the window.
127 let deadline = tokio::time::Instant::now() + BATCH_WINDOW;
128 while batch.len() < BATCH_MAX {
129 match tokio::time::timeout_at(deadline, rx.recv()).await {
130 Ok(Some(op)) => batch.push(op),
131 Ok(None) => break, // channel closed; commit what we have
132 Err(_) => break, // window elapsed
133 }
134 }
135
136 if let Err(e) = commit(&pool, &mut batch).await {
137 // Losing a batch of positions is bad but survivable; the clients
138 // still hold them unacked and will retry. Dying here is not
139 // survivable, so log loudly and carry on.
140 error!(error = %e, "write batch failed; clients will retry");
141 batch.clear();
142 }
143 }
144
145 // Drain whatever is left so a graceful SIGTERM does not lose points.
146 while let Ok(op) = rx.try_recv() {
147 batch.push(op);
148 if batch.len() >= BATCH_MAX
149 && let Err(e) = commit(&pool, &mut batch).await
150 {
151 error!(error = %e, "final write batch failed");
152 batch.clear();
153 }
154 }
155 if !batch.is_empty()
156 && let Err(e) = commit(&pool, &mut batch).await
157 {
158 error!(error = %e, "final write batch failed");
159 }
160 debug!("writer task stopped");
161}
162
163async fn commit(pool: &SqlitePool, batch: &mut Vec<WriteOp>) -> Result<()> {
164 let count = batch.len();
165 let mut tx = pool.begin().await.context("begin transaction")?;
166 for op in batch.drain(..) {
167 apply(&mut tx, op).await?;
168 }
169 tx.commit().await.context("commit transaction")?;
170 if count > BATCH_MAX / 2 {
171 warn!(
172 count,
173 "large write batch — the writer may be falling behind"
174 );
175 } else {
176 debug!(count, "committed write batch");
177 }
178 Ok(())
179}
180
181async fn apply(tx: &mut Transaction<'_, Sqlite>, op: WriteOp) -> Result<()> {
182 match op {
183 WriteOp::Points {
184 user_id,
185 src_token_id,
186 points,
187 recv_at,
188 } => {
189 for p in points {
190 insert_point(tx, user_id, src_token_id, &p, recv_at).await?;
191 }
192 }
193
194 WriteOp::TokenSeen {
195 token_id,
196 at,
197 src_ip,
198 src_port,
199 transport,
200 } => {
201 // The source address is recorded but never used for authentication,
202 // which is exactly why a Wi-Fi to LTE handoff needs no protocol work.
203 sqlx::query(
204 "UPDATE tokens SET last_seen_at = ?, last_src_ip = ?, last_src_port = ?, \
205 last_transport = ? WHERE token_id = ?",
206 )
207 .bind(at)
208 .bind(src_ip)
209 .bind(i64::from(src_port))
210 .bind(transport)
211 .bind(token_id)
212 .execute(&mut **tx)
213 .await
214 .context("updating token telemetry")?;
215 }
216
217 WriteOp::TokenHello {
218 token_id,
219 app_version,
220 os_api_level,
221 } => {
222 sqlx::query("UPDATE tokens SET app_version = ?, os_api_level = ? WHERE token_id = ?")
223 .bind(app_version)
224 .bind(os_api_level)
225 .bind(token_id)
226 .execute(&mut **tx)
227 .await
228 .context("recording HELLO")?;
229 }
230
231 WriteOp::Audit {
232 user_id,
233 at,
234 action,
235 detail,
236 src_ip,
237 } => {
238 sqlx::query(
239 "INSERT INTO audit_log (at, user_id, action, detail, src_ip) VALUES (?, ?, ?, ?, ?)",
240 )
241 .bind(at)
242 .bind(user_id)
243 .bind(action)
244 .bind(detail)
245 .bind(src_ip)
246 .execute(&mut **tx)
247 .await
248 .context("writing audit log")?;
249 }
250 }
251 Ok(())
252}
253
254/// Insert one point and, if it is newer than what we have, update the live marker.
255///
256/// The `ON CONFLICT` clause is the load-bearing part of the whole storage design:
257///
258/// * A **retry or replay** carries a `ts` that already exists, so it collapses
259/// into the existing row. This is why the protocol needs no replay window.
260/// * **Two phones on one account reporting in the same second** would otherwise
261/// resolve to whichever packet landed last, which is arbitrary. The
262/// `WHERE excluded.acc_dm < points.acc_dm` guard keeps the better fix.
263async fn insert_point(
264 tx: &mut Transaction<'_, Sqlite>,
265 user_id: i64,
266 src_token_id: i64,
267 p: &Point,
268 recv_at: i64,
269) -> Result<()> {
270 let ts = i64::from(p.ts);
271 let acc = p.acc_dm.map(i64::from);
272
273 sqlx::query(
274 "INSERT INTO points \
275 (user_id, ts, lat, lon, acc_dm, alt_m, spd_cms, brg_cdeg, bat_pct, flags, recv_at, src_token_id) \
276 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
277 ON CONFLICT (user_id, ts) DO UPDATE SET \
278 lat = excluded.lat, lon = excluded.lon, acc_dm = excluded.acc_dm, \
279 alt_m = excluded.alt_m, spd_cms = excluded.spd_cms, brg_cdeg = excluded.brg_cdeg, \
280 bat_pct = excluded.bat_pct, flags = excluded.flags, recv_at = excluded.recv_at, \
281 src_token_id = excluded.src_token_id \
282 WHERE excluded.acc_dm IS NOT NULL \
283 AND (points.acc_dm IS NULL OR excluded.acc_dm < points.acc_dm)",
284 )
285 .bind(user_id)
286 .bind(ts)
287 .bind(p.lat_e7)
288 .bind(p.lon_e7)
289 .bind(acc)
290 .bind(p.alt_m.map(i64::from))
291 .bind(p.spd_cms.map(i64::from))
292 .bind(p.brg_cdeg.map(i64::from))
293 .bind(p.bat_pct.map(i64::from))
294 .bind(i64::from(p.flags.0))
295 .bind(recv_at)
296 .bind(src_token_id)
297 .execute(&mut **tx)
298 .await
299 .context("inserting point")?;
300
301 // The live marker only moves forward in client time. A phone that has been
302 // in a drawer and wakes up with an old queued fix must not drag the dot back.
303 sqlx::query(
304 "INSERT INTO user_latest \
305 (user_id, ts, lat, lon, acc_dm, alt_m, spd_cms, brg_cdeg, bat_pct, flags, recv_at, src_token_id) \
306 VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \
307 ON CONFLICT (user_id) DO UPDATE SET \
308 ts = excluded.ts, lat = excluded.lat, lon = excluded.lon, acc_dm = excluded.acc_dm, \
309 alt_m = excluded.alt_m, spd_cms = excluded.spd_cms, brg_cdeg = excluded.brg_cdeg, \
310 bat_pct = excluded.bat_pct, flags = excluded.flags, recv_at = excluded.recv_at, \
311 src_token_id = excluded.src_token_id \
312 WHERE excluded.ts > user_latest.ts",
313 )
314 .bind(user_id)
315 .bind(ts)
316 .bind(p.lat_e7)
317 .bind(p.lon_e7)
318 .bind(acc)
319 .bind(p.alt_m.map(i64::from))
320 .bind(p.spd_cms.map(i64::from))
321 .bind(p.brg_cdeg.map(i64::from))
322 .bind(p.bat_pct.map(i64::from))
323 .bind(i64::from(p.flags.0))
324 .bind(recv_at)
325 .bind(src_token_id)
326 .execute(&mut **tx)
327 .await
328 .context("updating live position")?;
329
330 Ok(())
331}
332
333#[cfg(test)]
334mod tests {
335 use super::*;
336 use otproto::point::Flags;
337
338 async fn seeded() -> (crate::db::Db, tempfile::TempDir) {
339 let dir = tempfile::tempdir().expect("temp dir");
340 let db = crate::db::Db::open(&dir.path().join("t.db"))
341 .await
342 .expect("open");
343 sqlx::query(
344 "INSERT INTO users (id, username, pw_hash, display_name, created_at, pw_changed_at) \
345 VALUES (1, 'a', 'x', 'A', 0, 0)",
346 )
347 .execute(&db.write)
348 .await
349 .expect("user");
350 (db, dir)
351 }
352
353 fn point(ts: u32, lat: i32, acc: Option<u16>) -> Point {
354 Point {
355 acc_dm: acc,
356 flags: Flags::NONE,
357 ..Point::new(ts, lat, 0)
358 }
359 }
360
361 async fn drain(handle: &WriteHandle, task: tokio::task::JoinHandle<()>) {
362 drop(handle.clone());
363 // Dropping the last handle ends the task; the caller keeps one, so use an
364 // explicit timeout instead of awaiting forever.
365 let _ = tokio::time::timeout(Duration::from_millis(50), task).await;
366 }
367
368 #[tokio::test]
369 async fn a_batch_of_points_lands_in_one_transaction() {
370 let (db, _dir) = seeded().await;
371 let (handle, task) = spawn(db.write.clone());
372
373 handle
374 .send(WriteOp::Points {
375 user_id: 1,
376 src_token_id: 7,
377 points: (0..40)
378 .map(|i| point(1000 + i, i as i32, Some(50)))
379 .collect(),
380 recv_at: 2000,
381 })
382 .await
383 .expect("send");
384
385 tokio::time::sleep(Duration::from_millis(400)).await;
386 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
387 .fetch_one(&db.read)
388 .await
389 .expect("count");
390 assert_eq!(count, 40);
391 drain(&handle, task).await;
392 }
393
394 #[tokio::test]
395 async fn the_live_marker_only_moves_forward_in_client_time() {
396 let (db, _dir) = seeded().await;
397 let (handle, task) = spawn(db.write.clone());
398
399 // Newest first, then an older queued fix — the drawer-phone scenario.
400 for (ts, lat) in [(2000u32, 100), (1000, 999)] {
401 handle
402 .send(WriteOp::Points {
403 user_id: 1,
404 src_token_id: 7,
405 points: vec![point(ts, lat, Some(50))],
406 recv_at: 3000,
407 })
408 .await
409 .expect("send");
410 }
411 tokio::time::sleep(Duration::from_millis(400)).await;
412
413 let (ts, lat): (i64, i64) =
414 sqlx::query_as("SELECT ts, lat FROM user_latest WHERE user_id = 1")
415 .fetch_one(&db.read)
416 .await
417 .expect("latest");
418 assert_eq!(
419 (ts, lat),
420 (2000, 100),
421 "an older fix must not drag the marker back"
422 );
423
424 // Both points are still in the history, though.
425 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
426 .fetch_one(&db.read)
427 .await
428 .expect("count");
429 assert_eq!(count, 2);
430 drain(&handle, task).await;
431 }
432
433 #[tokio::test]
434 async fn a_replayed_point_is_idempotent() {
435 let (db, _dir) = seeded().await;
436 let (handle, task) = spawn(db.write.clone());
437
438 for _ in 0..3 {
439 handle
440 .send(WriteOp::Points {
441 user_id: 1,
442 src_token_id: 7,
443 points: vec![point(1000, 42, Some(80))],
444 recv_at: 2000,
445 })
446 .await
447 .expect("send");
448 }
449 tokio::time::sleep(Duration::from_millis(400)).await;
450
451 let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM points")
452 .fetch_one(&db.read)
453 .await
454 .expect("count");
455 assert_eq!(count, 1, "replay must not create duplicate rows");
456 drain(&handle, task).await;
457 }
458
459 #[tokio::test]
460 async fn a_same_second_collision_keeps_the_better_accuracy() {
461 let (db, _dir) = seeded().await;
462 let (handle, task) = spawn(db.write.clone());
463
464 for (acc, lat) in [(200u16, 1), (30, 2), (500, 3)] {
465 handle
466 .send(WriteOp::Points {
467 user_id: 1,
468 src_token_id: 7,
469 points: vec![point(1000, lat, Some(acc))],
470 recv_at: 2000,
471 })
472 .await
473 .expect("send");
474 }
475 tokio::time::sleep(Duration::from_millis(400)).await;
476
477 let (acc, lat): (i64, i64) =
478 sqlx::query_as("SELECT acc_dm, lat FROM points WHERE user_id = 1 AND ts = 1000")
479 .fetch_one(&db.read)
480 .await
481 .expect("point");
482 assert_eq!(
483 (acc, lat),
484 (30, 2),
485 "the better fix must win, whatever the arrival order"
486 );
487 drain(&handle, task).await;
488 }
489}
490