Day 10: The durable upload pipeline

Yesterday ended with fixes leaving the service isolate every two seconds. Today follows the other copy of those fixes, the one that goes to your Phoenix app. It is written to a local database first, then drained to POST /api/races/:id/positions in batches, so that a tunnel, a dead zone or a killed process costs latency rather than distance. You wrote the receiving end; today you line the two up.

Time and files

About 40 minutes: 10 on the queue, 15 on the uploader, 10 on how the service drives it and reports back, 5 on the backend side.

Have open: app/lib/tracking/upload/fix_queue.dart, app/lib/tracking/upload/position_uploader.dart, app/lib/tracking/upload/race_upload_target.dart, app/lib/tracking/location_service.dart, app/lib/features/race/race_notifier.dart, app/test/tracking/upload/position_uploader_test.dart, and on the backend backend/lib/footrace_web/controllers/api/race_controller.ex and backend/lib/racing/positions/ingest.ex.

Before you start

Day 9 established three things you need now. The service runs in its own isolate and cannot read Riverpod, the keychain or AppConfig. What it needs arrives through FlutterForegroundTask.saveData before the service starts. And onRepeatEvent fires every 2 s, which is when uploads happen. Day 6 covered the app’s Dio client and ApiException; you will see why the pipeline does not use them. The reference page is Tracking; the end-to-end picture is in The race flow.

sequenceDiagram
    participant H as LocationTaskHandler
    participant Q as FixQueue (sembast file)
    participant U as PositionUploader
    participant P as Phoenix RaceController
    participant UI as RaceNotifier (UI isolate)
    H->>Q: enqueue(raceId, fixes)
    H->>U: drain()
    loop until empty, stop, or failure
        U->>Q: pending(limit 50)
        U->>P: POST /api/races/:id/positions
        P-->>U: 200 {race, competitors, competitor}
        U->>Q: markUploaded(keys)
    end
    U-->>H: UploadReport
    H-->>UI: sendDataToMain({type: upload, ...report})

1. The target and the queue

What the service is told. app/lib/tracking/upload/race_upload_target.dart:4-24 is a plain value class with four fields (race id, API base URL, bearer token, queue file path) and the four string keys they are saved under. It exists so that both isolates agree on the key names. The UI isolate writes them in startBackgroundTracking (app/lib/tracking/location_service.dart:333-348); the service reads them back in _openUploadPipeline (app/lib/tracking/location_service.dart:77-108). If any of the four is missing, lines 86-91 return early and the service only reports fixes, never uploads: that is the diagnostics-screen mode from Day 9.

The queue. Open app/lib/tracking/upload/fix_queue.dart. The doc comment on lines 33-37 says why it is sembast: it is pure Dart. Most storage in Flutter (SQLite via sqflite, SharedPreferences, the keychain) is a plugin, meaning Dart code that calls native code over a platform channel. The service handler cannot rely on those, so the queue uses a database written entirely in Dart that reads and writes an ordinary file through dart:io.

sembast is a document store: stores of records, each record a JSON-like map with a key. Line 43 creates a store with auto-increment int keys:

final _store = intMapStoreFactory.store(_storeName);

If you have used Room or Core Data, think of a single table with no schema, or of a DETS table with integer keys if you prefer the Elixir picture. Two constructors (lines 45-49): open(path) on a file for the app, inMemory() for tests. These are static methods returning Future<FixQueue>, because opening is async and Dart constructors cannot be.

Each operation is a few lines:

• enqueue (lines 54-71) adds one record per fix inside _db.transaction. The map stores lat/lng, accuracy, altitude, speed, an ISO-8601 UTC timestamp, createdAt, and uploadedAt: null. Pending means uploadedAt is null.
• pending (lines 74-87) is the query the uploader runs: same race, not uploaded, sorted by timestamp then key, limit 50 by default. Finder, Filter.and, SortOrder read like a tiny Ecto query.
• markUploaded (lines 97-104) sets uploadedAt on the given keys. Rows are not deleted on success, so you can still count what went out.
• clearRace (lines 107-110) deletes everything for a race. pruneUploaded (lines 113-124) deletes uploaded rows older than a day; the service calls it each time it opens the queue (app/lib/tracking/location_service.dart:94).

QueuedFix.toPosition (app/lib/tracking/upload/fix_queue.dart:23-30) builds the JSON body for one fix: latitude, longitude, timestamp (UTC ISO-8601), accuracy, altitude, speed. Its doc comment names the backend module it targets, which is your cue to check the other side. In backend/lib/racing/positions/ingest.ex:61-76, parse_one requires latitude, longitude and a readable timestamp, and takes the rest as optional numbers. timestamp/1 accepts ISO-8601 (lines 101-106) or Unix milliseconds (lines 94-99), and a missing one means “now” (line 92). The app always sends one, so a fix replayed after a dead zone lands at the time it was taken, not the time it arrived.

2. PositionUploader: batches, backoff, stop codes

Open app/lib/tracking/upload/position_uploader.dart. Start with the doc comment on lines 103-109; it is the contract:

At-least-once: a fix is marked uploaded only after a 200; the server dedupes by timestamp. Network failures back off 1, 2, 4 … 30 s and stay pending. 409 (race over) and 403 stop the uploader for good; 422 (malformed) drops that batch since retrying cannot fix it; 401 stops until the service is restarted with a fresh token.

The HTTP client. PositionClient (lines 21-24) is an abstract class with one method, so tests can fake it. HttpPositionClient (lines 26-66) uses package:http, not the app’s Dio client. The comment on lines 19-20 gives the reason: Dio and its interceptors live in the UI isolate’s provider graph. So this client sets its own Authorization: Bearer header (line 46), wraps the batch as {'positions': [...]} (line 50), times out after 15 s (line 52), and tolerates a non-JSON body (lines 53-61). It does not throw on a 4xx or 5xx; it returns an UploadResponse and lets the uploader decide. It only throws when there is no response at all.

The drain loop. drain() (lines 140-214). Read it top to bottom:

1. Lines 141-147: if a stop condition was hit earlier, or a drain is already running, return a report with the current pending count and do nothing else. _stopped is sticky for the life of the uploader.
2. Line 150: if the last pass failed, sleep the backoff now, at the start of the next drain. The sleep function is injected (line 117, defaulting to Future.delayed on line 120) so tests can record sleeps instead of waiting.
3. Lines 153-209: loop. Take up to 50 pending fixes (line 154), stop if none (line 155), post them (lines 157-159).
4. A thrown exception (lines 160-168) is a network failure: grow the backoff, return a report with an error string. The fixes stay pending.
5. A 2xx (lines 169-175) marks the batch uploaded, keeps the body as lastReply, resets the backoff and continues to the next 50.
6. Anything else goes to the switch (lines 176-201).

switch (response.statusCode) {
  case 409:
    _stopped = UploadStop.raceOver;
  case 403:
    _stopped = UploadStop.forbidden;
  case 401:
    _stopped = UploadStop.unauthorized;
  case 422:
    // Malformed batch: retrying cannot fix it. Drop it and go on.
    await _queue.markUploaded(batch.map((q) => q.key));
    return UploadReport(/* ... */);
  default:
    _backoff = _nextBackoff();
    return UploadReport(/* ... */);
}
return UploadReport(/* ... */ stop: _stopped);

Dart 3 cases do not fall through, so 409, 403 and 401 each just set _stopped and then reach the return after the switch. “Drop” on 422 means mark uploaded, not delete. A 5xx takes the default branch and backs off exactly like a lost connection.

_nextBackoff (lines 216-220) is 1 s, then doubling, capped at maxBackoff (30 s, line 116).

The report. UploadReport (lines 72-99) carries pending, uploaded, the last 200 body as reply, an error string and the stop reason. toMap() (lines 92-98) flattens it into primitives for the isolate port, with stop as the enum’s .name string.

Now line these up with your controller. backend/lib/footrace_web/router.ex:136 routes the POST to RaceController.put_position, inside the :authorized_api pipeline.

Server says	Where in Phoenix	App does
200 + {race, competitors, competitor}	backend/lib/footrace_web/controllers/api/race_controller.ex:57-63	mark uploaded, reset backoff, keep reply
401 {error: "token_expired"} etc.	Guardian, backend/lib/footrace/users/error_handler.ex:13-20	stop (unauthorized)
403 not a participant	race_controller.ex:65-68	stop (forbidden)
409 race already ended	race_controller.ex:70-73, from open_race at lines 98-103 or Ingest.record at backend/lib/racing/positions/ingest.ex:155-156	stop (raceOver), queue cleared
422 bad position	race_controller.ex:75-87	mark batch uploaded, carry on next pass
5xx / no response	(crash, proxy, offline)	back off 1 → 30 s, keep pending

Two server details matter for the client. Ingest.parse sorts each batch by timestamp (backend/lib/racing/positions/ingest.ex:47), so order within a batch is forgiving. And Ingest.record drops positions that arrive while the race process is in :countdown but still returns {:ok, state} (lines 143-145), so the app sees a 200 and marks them uploaded. That is intended: the tracker starts three seconds before the gun (Day 11) and those fixes must not count.

Could not verify: “the server dedupes by timestamp”

The comment on app/lib/tracking/upload/position_uploader.dart:105-106 relies on server-side dedupe. In the backend checkout read for this page (d7b9c91, not fetched, so possibly behind its remote), I could not find any. Ingest.insert_marker! inserts every position (backend/lib/racing/positions/ingest.ex:160-183), I found no unique index on markers, and RaceProcess.update_distance adds the haversine distance from the last position to each new one with no timestamp check (backend/lib/racing/race_process.ex:402-413). If a 200 is lost in transit and the batch is re-sent, the server would apply it twice. Check the backend’s current main before relying on this.

3. How the service drives it, and how the UI hears

Back to app/lib/tracking/location_service.dart. Every flush calls _persistAndUpload (line 151), lines 165-202:

final toStore = List<GpsFix>.of(_pendingEnqueue);
_pendingEnqueue.clear();
_uploadInFlight ??= () async {
  try {
    if (toStore.isNotEmpty) await queue.enqueue(raceId, toStore);
    final report = await uploader.drain();
    FlutterForegroundTask.sendDataToMain(<String, Object?>{
      'type': 'upload',
      'raceId': raceId,
      ...report.toMap(),
    });
    if (report.stop == UploadStop.raceOver) {
      await queue.clearRace(raceId);
    }
  } catch (e) {
    // ...
  } finally {
    _uploadInFlight = null;
  }
}();
if (toStore.isNotEmpty && _uploadInFlight != null) {
  // A drain was already running: store these now so nothing is lost if
  // the service dies before the next flush.
  queue.enqueue(raceId, toStore).catchError(/* ... */);
}

Some Dart here worth unpacking:

• () async { ... }() defines an async closure and calls it at once, an IIFE. It returns a Future<void> straight away; the body runs synchronously up to its first await, then continues later on the event loop. A JS (async () => { ... })() behaves the same.
• ??= assigns only when the left side is null. So if a drain is already in flight, the closure is never created, and the 2 s flush does not start a second one.
• ...report.toMap() is the spread operator inside a map literal, like {...a} in JS.
• _uploadInFlight is also awaited in onDestroy (line 215), so stopping the service waits for the current upload before closing the HTTP client and the database.

Read lines 194-201 slowly, together with line 175. The intent is “if a drain was already running, store these fixes now”. But after line 175 runs, _uploadInFlight is non-null in both cases: either it was already set, or line 175 just set it to the new closure’s future (an async function never completes synchronously, so its finally cannot have cleared it yet). As written, when no drain was running, the fixes are enqueued twice: once at line 177 and once at line 197. Duplicate rows with the same timestamp sort next to each other in pending, and a zero-length hop adds no distance on the server, so the effect may be small; but together with the dedupe caution above it is worth a test. A bool captured before line 175 (final wasRunning = _uploadInFlight != null;) would express the intent. I could not find a test that drives LocationTaskHandler directly.

On the UI side, the 'upload' message arrives through the same callback as the batches. RaceNotifier._onTrackerMessage (app/lib/features/race/race_notifier.dart:201-234) switches on data['type']. For 'upload' (lines 214-232) it turns reply, the last 200 body, into a RaceSnapshot and applies it exactly as it would a channel raceUpdate (lines 216-224). That is the fallback the comment on lines 145-146 promises: with no socket, uploads still bring the server state back every two seconds. A stop of unauthorized or forbidden becomes a trackingError banner (lines 229-232). raceOver gets no special case: the reply before it, or the channel’s raceFinished, ends the race.

Two things follow from the design:

• The token is copied. The UI reads it from TokenStore when tracking starts (app/lib/features/race/race_notifier.dart:304) and saves it in the plugin’s plain prefs. A 401 mid-race stops uploads until the service is restarted with a new one; the stopBackgroundTracking comment (app/lib/tracking/location_service.dart:368) is why the copy is removed when the race ends.
• Pending rows can outlive a race. Only raceOver triggers clearRace, and pruneUploaded only removes uploaded rows. Rows left pending after a 401 or 403, or after the service was killed past the finish, appear to stay in fix_queue.db until something else clears them. I found nothing that does.

Try it

1. Run the uploader tests and read one.

   flutter test test/tracking/upload/position_uploader_test.dart
   flutter test test/tracking/upload/fix_queue_test.dart

   Open app/test/tracking/upload/position_uploader_test.dart:40-51: the setUp builds a real in-memory FixQueue, a FakePositionClient that replays a list of responses or exceptions, and a sleep that records durations. Then read the backoff test at line 66.

2. Predict, then check, a backoff sequence. Add a test after the one at line 136 (“5xx backs off like a network failure”). Enqueue fixAt(1), then queue up [Exception('offline'), Exception('offline'), const UploadResponse(500, {}), FakePositionClient.ok] and call drain() four times. Before running it, write down what sleeps and uploader.currentBackoff will be after each call, and what the fourth report’s uploaded is. Then assert it. (Answer: sleeps [], [1s], [1s, 2s], [1s, 2s, 4s]; backoff 1 s, 2 s, 4 s, then zero; uploaded 1.)

3. Optional, on the backend side. Read backend/test/footrace_web/controllers/api/race_controller_test.exs from line 92, the describe for this endpoint, and find the test whose batch is sent out of order (line 133). That is the server half of “order within a batch is forgiving”.

Check yourself

1. Why does HttpPositionClient use package:http instead of the app’s Dio ApiClient?
2. A batch gets a 422. What happens to those fixes, and to the next batch?
3. The phone is offline for 90 seconds during a race. Roughly what does the uploader do, and how much distance does the server lose once the phone is back?
4. Where does the UI’s race state come from when the WebSocket is down?
5. What is the difference between clearRace and pruneUploaded, and which code calls each?

Answers

1. Dio, its interceptors and the token store live in the UI isolate’s Riverpod graph. The service isolate cannot reach them, so it gets a minimal, plugin-free client and a token copied in through the start payload (app/lib/tracking/upload/position_uploader.dart:19-20).
2. They are marked uploaded (dropped) because retrying cannot fix a malformed batch; the drain returns with an error and the next flush’s drain carries on with the next 50 (lines 183-191).
3. Every flush enqueues fixes; each drain fails, backs off 1, 2, 4, 8, 16, then 30 s (the sleep happens at the start of the next drain, while later flushes just enqueue). When the connection is back, the next drain posts the backlog in batches of 50 with the original timestamps. The server rebuilds distance from those positions, so in principle nothing is lost, only delayed.
4. From the reply field of each 'upload' report: the last 200 body, parsed into a RaceSnapshot in RaceNotifier._onTrackerMessage (app/lib/features/race/race_notifier.dart:214-224).
5. clearRace deletes every row for one race, uploaded or not; the service calls it after a 409 (app/lib/tracking/location_service.dart:184-186). pruneUploaded deletes uploaded rows older than a day across all races; the service calls it when it opens the queue (line 94).

Next

Day 11: RaceNotifier, the race screen, the result.