App 2.15.0: queued live data survives restarts

Unsent live samples were kept in memory only, so a reboot or a killed
process during a dead spot lost them. The queue is now saved to
files/telemetry_queue.json with AtomicFile, at most every 5 minutes and
only after a change (the flash is not written every 15 s), and at once
when the service stops; it is restored, oldest first and within the
2000-sample cap, when the service starts. A damaged file restores
nothing rather than failing. Tested on the emulator across a reboot.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01QLJfZQZeAKAwSqViTmLzki
This commit is contained in:
2026-10-18 17:56:25 +02:00
co-authored by Claude Opus 5.5
parent f63f48131e
commit f88249427f
6 changed files with 164 additions and 7 deletions
+2 -2
View File
@@ -34,8 +34,8 @@ android {
applicationId "com.stinnesbeck.nilsmg4"
minSdk 27
targetSdk 36
versionCode 82
versionName "2.14.4"
versionCode 83
versionName "2.15.0"
// Nothing about the server ships in the APK: its URL and ingest token are entered in
// the app's Server tab (ServerConfig), and the database is only reachable by the server.
@@ -141,7 +141,10 @@ public class AbrpUploadService extends Service {
running = false;
prefs.edit().putBoolean("service_running", false).apply();
if (scheduler != null) scheduler.shutdownNow();
if (telemetryUploader != null) telemetryUploader.stop();
if (telemetryUploader != null) {
telemetryUploader.persist(); // what is still queued survives the restart
telemetryUploader.stop();
}
if (steeringKeys != null) steeringKeys.unregister();
Log.i(TAG, "Service stopped");
}
@@ -9,8 +9,9 @@ import java.util.List;
*
* Bounded: while the car is offline the oldest samples are dropped once {@link #capacity}
* is reached, so a long dead spot costs the start of the gap, never the head unit's memory.
* Held in memory only — a process restart loses what is queued, which is the price of not
* writing to flash every 15 seconds.
* Kept in memory and saved to a file now and then (TelemetryUploader), so a reboot or a
* killed process on the way through a dead spot does not lose the drive. {@link #encode} and
* {@link #decode} are the file format: a JSON array of the samples as they are sent.
*
* Free of Android types so the policy is unit-tested on the JVM.
*/
@@ -29,7 +30,11 @@ final class TelemetryQueue {
TelemetryQueue(int capacity) { this.capacity = capacity; }
/** Counts every change, so the owner can tell whether there is anything new to save. */
private long version;
synchronized void add(String sampleJson) {
version++;
if (samples.size() >= capacity) {
samples.removeFirst();
dropped++;
@@ -49,6 +54,7 @@ final class TelemetryQueue {
/** Drops the oldest {@code count} samples once the server has confirmed them. */
synchronized void remove(int count) {
if (count > 0) version++;
for (int i = 0; i < count && !samples.isEmpty(); i++) samples.removeFirst();
}
@@ -57,6 +63,50 @@ final class TelemetryQueue {
synchronized int size() { return samples.size(); }
synchronized long version() { return version; }
/** Every queued sample, oldest first, as {@link #encode} writes them. */
synchronized String encode() {
org.json.JSONArray array = new org.json.JSONArray();
for (String sample : samples) array.put(sample);
return array.toString();
}
/**
* Puts saved samples back in front of anything queued since, oldest first and within the
* capacity. Unreadable input restores nothing: a damaged file costs the backlog, never
* the app.
*/
synchronized int restore(String saved) {
List<String> restored = decode(saved);
if (restored.isEmpty()) return 0;
ArrayDeque<String> merged = new ArrayDeque<>(restored);
merged.addAll(samples);
while (merged.size() > capacity) {
merged.removeFirst();
dropped++;
}
samples.clear();
samples.addAll(merged);
version++;
return restored.size();
}
static List<String> decode(String saved) {
List<String> out = new ArrayList<>();
if (saved == null || saved.isEmpty()) return out;
try {
org.json.JSONArray array = new org.json.JSONArray(saved);
for (int i = 0; i < array.length(); i++) {
String sample = array.optString(i, null);
if (sample != null && sample.startsWith("{")) out.add(sample);
}
} catch (org.json.JSONException e) {
out.clear();
}
return out;
}
/** Samples lost to the capacity limit since the queue was created. */
synchronized int dropped() { return dropped; }
}
@@ -59,6 +59,59 @@ final class TelemetryUploader {
TelemetryUploader(Context context) {
this.context = context.getApplicationContext();
load();
}
// ---------- Keeping the queue across restarts ----------
private static final String QUEUE_FILE = "telemetry_queue.json";
/**
* At most this often, and only after a change: the flash is not written every 15 s, and a
* killed process loses at most this much. A clean stop saves at once ({@link #persist}).
*/
private static final long SAVE_INTERVAL_MS = 5 * 60_000L;
private long savedVersion;
private long lastSaveMs;
private android.util.AtomicFile queueFile() {
return new android.util.AtomicFile(new java.io.File(context.getFilesDir(), QUEUE_FILE));
}
private synchronized void load() {
try {
byte[] bytes = queueFile().readFully();
int restored = queue.restore(new String(bytes, java.nio.charset.StandardCharsets.UTF_8));
if (restored > 0) Log.i(TAG, restored + " queued sample(s) restored");
} catch (java.io.FileNotFoundException e) {
// nothing saved yet
} catch (Exception e) {
Log.w(TAG, "Saved queue not read: " + e.getClass().getSimpleName());
}
savedVersion = queue.version();
lastSaveMs = SystemClock.elapsedRealtime();
}
private void maybeSave() {
if (SystemClock.elapsedRealtime() - lastSaveMs >= SAVE_INTERVAL_MS) persist();
}
/** Saves the queue if it changed since the last save. Atomic: a crash keeps the old file. */
synchronized void persist() {
long version = queue.version();
lastSaveMs = SystemClock.elapsedRealtime();
if (version == savedVersion) return;
android.util.AtomicFile file = queueFile();
java.io.FileOutputStream out = null;
try {
out = file.startWrite();
out.write(queue.encode().getBytes(java.nio.charset.StandardCharsets.UTF_8));
file.finishWrite(out);
savedVersion = version;
} catch (Exception e) {
if (out != null) file.failWrite(out);
Log.w(TAG, "Queue not saved: " + e.getClass().getSimpleName());
}
}
// ---------- Tick ----------
@@ -101,11 +154,15 @@ final class TelemetryUploader {
if (clockSet && (!idle || uploadDue)) {
queue.add(buildSample(nowMs, vehicle, session, settings, soc, charging, dcfc, parked, idle));
}
if (!uploadDue) return;
if (!uploadDue) {
maybeSave();
return;
}
// Cadence counts attempts, not successes: retrying every tick while offline would
// spend power exactly when it buys nothing.
lastUploadAttemptMs = nowMs;
flush();
maybeSave();
}
int queued() { return queue.size(); }
@@ -1,6 +1,7 @@
package com.evsuite.abrp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.List;
@@ -37,4 +38,46 @@ public class TelemetryQueueTest {
assertEquals(List.of("c", "d", "e"), queue.peekBatch());
assertEquals(2, queue.dropped());
}
@Test
public void aSavedQueueComesBackInOrder() {
TelemetryQueue q = new TelemetryQueue(10);
q.add("{\"utc\":1}");
q.add("{\"utc\":2}");
TelemetryQueue restarted = new TelemetryQueue(10);
restarted.add("{\"utc\":3}"); // taken after the restart
assertEquals(2, restarted.restore(q.encode()));
assertEquals(java.util.Arrays.asList("{\"utc\":1}", "{\"utc\":2}", "{\"utc\":3}"), restarted.peekBatch());
}
@Test
public void restoringKeepsTheCapacityByDroppingTheOldest() {
TelemetryQueue q = new TelemetryQueue(10);
for (int i = 1; i <= 4; i++) q.add("{\"utc\":" + i + "}");
TelemetryQueue small = new TelemetryQueue(3);
small.restore(q.encode());
assertEquals(java.util.Arrays.asList("{\"utc\":2}", "{\"utc\":3}", "{\"utc\":4}"), small.peekBatch());
assertEquals(1, small.dropped());
}
@Test
public void aDamagedFileRestoresNothing() {
TelemetryQueue q = new TelemetryQueue(10);
assertEquals(0, q.restore("[\"{\\\"utc\\\":1}\", trunc"));
assertEquals(0, q.restore(""));
assertEquals(0, q.restore(null));
assertEquals(0, q.size());
}
@Test
public void everyChangeMovesTheVersion() {
TelemetryQueue q = new TelemetryQueue(10);
long v0 = q.version();
q.add("{\"utc\":1}");
long v1 = q.version();
q.remove(0);
assertEquals(v1, q.version()); // nothing removed, nothing to save
q.remove(1);
assertTrue(q.version() > v1 && v1 > v0);
}
}