From 343c0c404e138fee980e3ef7b08e2a42e52235b4 Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Fri, 10 Apr 2026 15:36:12 -0600 Subject: [PATCH 1/6] feat: offline disk overflow + sendBatchDirect Add offline disk overflow capability to PersistableEventQueue: - Events that overflow from memory while offline are buffered and batched to a separate disk file (offline-overflow.v1.json) - On reconnect, overflow drains directly to network via Dispatcher.sendBatchDirect() without loading into memory queue - Respects maxOfflineDiskEvents cap (default 10000) - EventDiskStore now accepts configurable filename for separate stores - MetaRouterAnalyticsClient wires overflow enable/disable on network transitions and triggers drain on reconnect Tests cover: overflow buffering, batch writes, disk cap enforcement, drain-to-network success/failure, app-kill-relaunch scenario, toggle behavior, sendBatchDirect, and configurable filename. --- .../analytics/MetaRouterAnalyticsClient.kt | 20 +- .../analytics/dispatcher/Dispatcher.kt | 46 ++- .../analytics/queue/PersistableEventQueue.kt | 154 +++++++- .../analytics/storage/EventDiskStore.kt | 13 +- .../analytics/dispatcher/DispatcherTest.kt | 63 +++ .../queue/PersistableEventQueueTest.kt | 371 ++++++++++++++++++ .../analytics/storage/EventDiskStoreTest.kt | 29 ++ 7 files changed, 684 insertions(+), 12 deletions(-) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt index 18fb3c1..bb5e196 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt @@ -150,9 +150,12 @@ class MetaRouterAnalyticsClient private constructor( eventQueue = injectedEventQueue } else { val diskStore = injectedDiskStore ?: EventDiskStore.create(context) + val overflowDiskStore = EventDiskStore.create(context, filename = "offline-overflow.v1.json") val pQueue = PersistableEventQueue( maxCapacity = options.maxQueueEvents, - diskStore = diskStore + diskStore = diskStore, + maxOfflineDiskEvents = options.maxOfflineDiskEvents, + overflowDiskStore = overflowDiskStore ) pQueue.rehydrate() eventQueue = pQueue @@ -179,17 +182,27 @@ class MetaRouterAnalyticsClient private constructor( networkMonitor = DebouncedNetworkMonitor(rawMonitor, scope) networkMonitor.start { connected -> if (connected) { + persistableEventQueue?.setOfflineOverflowEnabled(false) circuitBreaker.reset() dispatcher.resumeFromOffline() - scope.launch { dispatcher.flush() } + // Two independent flush paths: + scope.launch { dispatcher.flush() } // (1) memory queue -> network + scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } // (2) disk -> network directly } else { dispatcher.pauseForOffline() + persistableEventQueue?.setOfflineOverflowEnabled(true) } } - // If starting offline, pause dispatcher immediately + // If starting offline, pause dispatcher and enable overflow immediately if (!networkMonitor.isConnected) { dispatcher.pauseForOffline() + persistableEventQueue?.setOfflineOverflowEnabled(true) + } + + // If online at launch and overflow exists from previous session, drain directly to network + if (networkMonitor.isConnected) { + scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } } startEventProcessor() @@ -418,6 +431,7 @@ class MetaRouterAnalyticsClient private constructor( } flush() persistableEventQueue?.flushToDisk() + persistableEventQueue?.flushOverflowBufferToDisk() dispatcher.pause() } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt index 772d958..af656a9 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt @@ -191,6 +191,50 @@ class Dispatcher( ) } + // ===== Direct Send (for disk overflow drain) ===== + + /** + * Send a batch of events directly to the network, bypassing the memory queue. + * Used by PersistableEventQueue.drainDiskOverflowToNetwork() to flush overflow + * events from disk without loading them into the memory queue. + * + * @return true if the batch was sent successfully, false on any failure + */ + suspend fun sendBatchDirect(events: List): Boolean { + if (events.isEmpty()) return true + + val sentAt = getIso8601Timestamp() + val batch = events.map { it.copy(sentAt = sentAt) } + + val url = "${options.getNormalizedIngestionHost()}${config.endpointPath}" + val body: ByteArray + try { + val payload = BatchPayload(batch = batch) + body = json.encodeToString(payload).toByteArray(Charsets.UTF_8) + } catch (e: Exception) { + Logger.error("Failed to encode direct batch of ${batch.size} events: ${e.message}") + return false + } + + val headers = mutableMapOf() + if (tracingEnabled) { + headers["Trace"] = "true" + } + + return try { + val response = networkClient.postJson( + url = url, + body = body, + timeoutMs = config.timeoutMs, + additionalHeaders = if (headers.isNotEmpty()) headers else null + ) + response.statusCode in 200..299 + } catch (e: Exception) { + Logger.warn("Direct batch send failed: ${e.message}") + false + } + } + // ===== Internal ===== /** @@ -368,7 +412,7 @@ class Dispatcher( } @kotlinx.serialization.Serializable -private data class BatchPayload( +internal data class BatchPayload( val batch: List ) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt index d43382d..8b9a0e6 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt @@ -1,5 +1,6 @@ package com.metarouter.analytics.queue +import com.metarouter.analytics.dispatcher.Dispatcher import com.metarouter.analytics.storage.EventDiskStore import com.metarouter.analytics.storage.QueueSnapshot import com.metarouter.analytics.types.EnrichedEventPayload @@ -30,7 +31,10 @@ class PersistableEventQueue( private val flushThresholdEvents: Int = 500, private val flushThresholdBytes: Long = 2L * 1024 * 1024, private val maxCapacityBytes: Long = 5L * 1024 * 1024, - private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000 + private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000, + private val maxOfflineDiskEvents: Int = 10000, + private val overflowDiskStore: EventDiskStore? = null, + private val overflowBufferBatchThreshold: Int = 100 ) : EventQueueInterface { companion object { @@ -60,6 +64,11 @@ class PersistableEventQueue( // Cache of per-event byte sizes (parallel to memoryQueue order) private val eventSizes = ArrayDeque() + // Offline overflow state + @Volatile + private var offlineOverflowEnabled = false + private val overflowBuffer = ArrayDeque() + @Synchronized override fun size(): Int = memoryQueue.size @@ -78,7 +87,15 @@ class PersistableEventQueue( // Drop oldest if at event count capacity if (memoryQueue.size >= maxCapacity) { - dropOldest("event count capacity") + if (offlineOverflowEnabled && overflowDiskStore != null) { + // Overflow to buffer instead of dropping + val oldest = memoryQueue.removeFirst() + val size = eventSizes.removeFirst() + estimatedBytes -= size + bufferOverflow(oldest) + } else { + dropOldest("event count capacity") + } } memoryQueue.addLast(event) @@ -141,9 +158,11 @@ class PersistableEventQueue( memoryQueue.clear() eventSizes.clear() estimatedBytes = 0L + overflowBuffer.clear() diskStore.delete() + overflowDiskStore?.delete() if (count > 0) { - Logger.log("Cleared $count events from queue and deleted disk snapshot") + Logger.log("Cleared $count events from queue and deleted disk snapshots") } } @@ -240,6 +259,135 @@ class PersistableEventQueue( @Synchronized fun estimatedSizeBytes(): Long = estimatedBytes + // ===== Offline Overflow ===== + + /** + * Enable or disable offline overflow mode. When enabled and memory queue is full, + * oldest events overflow to buffer (then disk) instead of being dropped. + */ + @Synchronized + fun setOfflineOverflowEnabled(enabled: Boolean) { + offlineOverflowEnabled = enabled + } + + /** + * Buffer an evicted event for batched disk write. + * Flushes buffer to disk when batch threshold reached. + * Must be called within a synchronized block. + */ + private fun bufferOverflow(event: EnrichedEventPayload) { + overflowBuffer.addLast(event) + if (overflowBuffer.size >= overflowBufferBatchThreshold) { + flushOverflowBufferToDiskInternal() + } + } + + /** + * Flush the in-memory overflow buffer to the overflow disk store. + * Called when buffer reaches batch threshold, on app backgrounding, + * or on offline->online transition before draining. + */ + @Synchronized + fun flushOverflowBufferToDisk() { + flushOverflowBufferToDiskInternal() + } + + /** + * Internal implementation — assumes lock is already held or called within synchronized block. + */ + private fun flushOverflowBufferToDiskInternal() { + val store = overflowDiskStore ?: return + if (overflowBuffer.isEmpty()) return + + val batch = overflowBuffer.toList() + overflowBuffer.clear() + + val existing = try { + store.read()?.events ?: emptyList() + } catch (e: Exception) { + Logger.warn("Failed to read overflow disk store: ${e.message}") + emptyList() + } + + var combined = existing + batch + if (combined.size > maxOfflineDiskEvents) { + val dropCount = combined.size - maxOfflineDiskEvents + combined = combined.drop(dropCount) + Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") + } + + store.write(QueueSnapshot(version = 1, events = combined)) + Logger.log("Flushed ${batch.size} overflow events to disk (total on disk: ${combined.size})") + } + + /** + * Drain overflow events directly from disk to network in batches. + * Does NOT load events into the memory queue — sends via dispatcher's direct path. + * Called on offline->online transition as an independent flush pipeline. + * + * @return number of events successfully drained + */ + suspend fun drainDiskOverflowToNetwork(dispatcher: Dispatcher): Int { + val store = overflowDiskStore ?: return 0 + + // First, flush any remaining buffer to disk + synchronized(this) { + flushOverflowBufferToDiskInternal() + } + + var totalDrained = 0 + + while (true) { + val snapshot = store.read() ?: return totalDrained + val now = System.currentTimeMillis() + val events = if (eventTTLMs > 0) { + snapshot.events.filter { !isExpired(it, now) } + } else { + snapshot.events + } + + if (events.isEmpty()) { + store.delete() + return totalDrained + } + + // Take a batch from the front (oldest first) + val batchSize = minOf(100, events.size) + val batch = events.take(batchSize) + val remaining = events.drop(batchSize) + + // Send batch directly to network via dispatcher's HTTP path + val success = dispatcher.sendBatchDirect(batch) + if (success) { + totalDrained += batch.size + if (remaining.isEmpty()) { + store.delete() + Logger.log("Offline overflow disk drain complete ($totalDrained events)") + return totalDrained + } else { + store.write(QueueSnapshot(version = 1, events = remaining)) + } + } else { + // Network failed — stop draining, will retry on next online transition + Logger.warn("Offline overflow drain paused — network send failed ($totalDrained drained so far)") + return totalDrained + } + } + } + + /** + * Get count of events in the offline overflow buffer + disk store. + */ + @Synchronized + fun offlineOverflowCount(): Int { + val diskCount = try { + overflowDiskStore?.read()?.events?.size ?: 0 + } catch (e: Exception) { + 0 + } + return overflowBuffer.size + diskCount + } + private fun dropOldest(reason: String) { memoryQueue.removeFirstOrNull()?.let { dropped -> val size = eventSizes.removeFirstOrNull() ?: 0L diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt index 9c6b099..805616d 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt @@ -24,7 +24,7 @@ import java.io.File * * Thread safety: callers must synchronize externally (PersistableEventQueue does this). */ -class EventDiskStore(private val baseDir: File) { +class EventDiskStore(private val baseDir: File, private val filename: String = SNAPSHOT_FILENAME) { companion object { private const val DIRECTORY_NAME = "metarouter" @@ -35,8 +35,11 @@ class EventDiskStore(private val baseDir: File) { /** * Production factory: uses Context.noBackupFilesDir as base directory. * Falls back to Context.filesDir, then system temp directory. + * + * @param filename Override the default snapshot filename. Used for separate + * overflow storage (e.g., "offline-overflow.v1.json"). */ - fun create(context: Context): EventDiskStore { + fun create(context: Context, filename: String = SNAPSHOT_FILENAME): EventDiskStore { val baseDir = listOf( { context.noBackupFilesDir }, { context.filesDir } @@ -51,7 +54,7 @@ class EventDiskStore(private val baseDir: File) { null } } ?: File(System.getProperty("java.io.tmpdir", "/tmp"), "metarouter-fallback") - return EventDiskStore(baseDir) + return EventDiskStore(baseDir, filename) } } @@ -64,10 +67,10 @@ class EventDiskStore(private val baseDir: File) { get() = File(baseDir, "$DIRECTORY_NAME/$SUBDIRECTORY_NAME") private val snapshotFile: File - get() = File(snapshotDir, SNAPSHOT_FILENAME) + get() = File(snapshotDir, filename) private val tmpFile: File - get() = File(snapshotDir, "$SNAPSHOT_FILENAME.tmp") + get() = File(snapshotDir, "$filename.tmp") /** * Write a snapshot to disk. Uses atomic write-to-tmp-then-rename. diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt index 86fa564..7409d2c 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt @@ -1050,6 +1050,69 @@ class DispatcherTest { dispatcher.stop() } + // ===== sendBatchDirect Tests ===== + + @Test + fun `sendBatchDirect returns true on success`() = runTest { + val dispatcher = createDispatcher() + networkClient.nextResponse = NetworkResponse(200, emptyMap(), null) + + val events = listOf(createEvent("direct-1"), createEvent("direct-2")) + val result = dispatcher.sendBatchDirect(events) + + assertTrue(result) + assertEquals(1, networkClient.requests.size) + } + + @Test + fun `sendBatchDirect returns false on server error`() = runTest { + val dispatcher = createDispatcher() + networkClient.nextResponse = NetworkResponse(500, emptyMap(), null) + + val events = listOf(createEvent("direct-1")) + val result = dispatcher.sendBatchDirect(events) + + assertFalse(result) + } + + @Test + fun `sendBatchDirect returns false on network error`() = runTest { + val dispatcher = createDispatcher() + networkClient.nextException = java.io.IOException("Connection refused") + + val events = listOf(createEvent("direct-1")) + val result = dispatcher.sendBatchDirect(events) + + assertFalse(result) + } + + @Test + fun `sendBatchDirect returns true for empty list`() = runTest { + val dispatcher = createDispatcher() + + val result = dispatcher.sendBatchDirect(emptyList()) + + assertTrue(result) + assertEquals(0, networkClient.requests.size) // No request made + } + + @Test + fun `sendBatchDirect does not affect memory queue`() = runTest { + val dispatcher = createDispatcher() + networkClient.nextResponse = NetworkResponse(200, emptyMap(), null) + + // Add events to memory queue + queue.enqueue(createEvent("queue-1")) + assertEquals(1, queue.size()) + + // Send different events via direct path + val directEvents = listOf(createEvent("direct-1")) + dispatcher.sendBatchDirect(directEvents) + + // Memory queue should be untouched + assertEquals(1, queue.size()) + } + // ===== Helper Methods ===== private fun TestScope.createDispatcher(): Dispatcher { diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt index 50d20a2..5482a75 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt @@ -1,8 +1,19 @@ package com.metarouter.analytics.queue +import android.util.Log +import com.metarouter.analytics.InitOptions +import com.metarouter.analytics.dispatcher.Dispatcher +import com.metarouter.analytics.dispatcher.DispatcherConfig +import com.metarouter.analytics.network.CircuitBreaker +import com.metarouter.analytics.network.FakeNetworkClient +import com.metarouter.analytics.network.NetworkResponse import com.metarouter.analytics.storage.EventDiskStore import com.metarouter.analytics.storage.QueueSnapshot import com.metarouter.analytics.types.* +import io.mockk.every +import io.mockk.mockkStatic +import io.mockk.unmockkStatic +import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Assert.* import org.junit.Before @@ -577,6 +588,366 @@ class PersistableEventQueueTest { assertEquals("batch2-a", drained[0].messageId) } + // ===== Offline Overflow ===== + + @Test + fun `memory queue overflow while offline buffers then flushes to disk`() { + val overflowDir = createTempDirectory("metarouter-overflow-test").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 3, // Low threshold for testing + eventTTLMs = 0 + ) + + smallQueue.setOfflineOverflowEnabled(true) + + // Fill queue to capacity + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + assertEquals(5, smallQueue.size()) + + // Enqueue more — should overflow to buffer, then to disk at threshold=3 + repeat(4) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + + // Memory queue should still be at max capacity + assertEquals(5, smallQueue.size()) + + // Flush remaining buffer to disk + smallQueue.flushOverflowBufferToDisk() + + // Verify overflow events are on disk + // Queue was: fill-0..fill-4. When overflow-0 comes in, fill-0 is evicted to buffer. + // overflow-1 → fill-1 evicted. overflow-2 → fill-2 evicted (buffer hits threshold=3, auto-flush). + // overflow-3 → fill-3 evicted. Explicit flush writes remaining buffer. + val snapshot = overflowDiskStore.read() + assertNotNull(snapshot) + assertEquals(4, snapshot!!.events.size) + assertEquals("fill-0", snapshot.events[0].messageId) + assertEquals("fill-1", snapshot.events[1].messageId) + assertEquals("fill-2", snapshot.events[2].messageId) + assertEquals("fill-3", snapshot.events[3].messageId) + + overflowDir.deleteRecursively() + } + + @Test + fun `overflow writes are batched not per-event`() { + val overflowDir = createTempDirectory("metarouter-overflow-batch").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 10, // Won't trigger during this test + eventTTLMs = 0 + ) + + smallQueue.setOfflineOverflowEnabled(true) + + // Fill queue + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + + // Overflow 8 events — batch threshold is 10, so no auto-flush to disk + repeat(8) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + + // No disk write should have happened yet (buffer < threshold) + assertNull(overflowDiskStore.read()) + + // Explicit flush writes all at once + smallQueue.flushOverflowBufferToDisk() + val snapshot = overflowDiskStore.read() + assertNotNull(snapshot) + assertEquals(8, snapshot!!.events.size) + + overflowDir.deleteRecursively() + } + + @Test + fun `offline disk overflow respects maxOfflineDiskEvents cap`() { + val overflowDir = createTempDirectory("metarouter-overflow-cap").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 10, // Cap at 10 + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 5, + eventTTLMs = 0 + ) + + smallQueue.setOfflineOverflowEnabled(true) + + // Fill queue + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + + // Overflow 15 events (exceeds maxOfflineDiskEvents=10) + repeat(15) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + smallQueue.flushOverflowBufferToDisk() + + val snapshot = overflowDiskStore.read() + assertNotNull(snapshot) + assertTrue("Should cap at maxOfflineDiskEvents", snapshot!!.events.size <= 10) + + overflowDir.deleteRecursively() + } + + @Test + fun `drainDiskOverflowToNetwork sends directly to network`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-drain").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 100, + eventTTLMs = 0 + ) + + // Pre-populate overflow disk + val events = (1..5).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + // Set up dispatcher with fake network + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) + val initOptions = InitOptions( + writeKey = "test-key", + ingestionHost = "https://api.example.com", + flushIntervalSeconds = 10 + ) + val dispatcher = Dispatcher( + options = initOptions, + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(5, drained) + assertEquals(1, fakeClient.requests.size) // Batch of 5 events + assertNull(overflowDiskStore.read()) // Disk file deleted + + // Memory queue should be untouched (0 events — drain goes directly to network) + assertEquals(0, overflowQueue.size()) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drainDiskOverflowToNetwork stops on network failure`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-fail").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 100, + eventTTLMs = 0 + ) + + // Pre-populate overflow disk + val events = (1..5).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + // Set up dispatcher that fails + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(500, emptyMap(), null) + val initOptions = InitOptions( + writeKey = "test-key", + ingestionHost = "https://api.example.com", + flushIntervalSeconds = 10 + ) + val dispatcher = Dispatcher( + options = initOptions, + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(0, drained) + // Events should still be on disk + val remaining = overflowDiskStore.read() + assertNotNull(remaining) + assertEquals(5, remaining!!.events.size) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `clear deletes overflow disk file`() { + val overflowDir = createTempDirectory("metarouter-overflow-clear").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // Write overflow data + overflowDiskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("overflow-1")))) + overflowQueue.enqueue(createTestEvent("memory-1")) + + overflowQueue.clear() + + assertEquals(0, overflowQueue.size()) + assertNull(diskStore.read()) + assertNull(overflowDiskStore.read()) + + overflowDir.deleteRecursively() + } + + @Test + fun `app killed while offline, relaunch online, disk overflow drains to network`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-relaunch").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + // Simulate previous session: overflow events were written to disk + val previousEvents = (1..3).map { createTestEvent("prev-session-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = previousEvents)) + + // New session — create fresh queue (simulating app relaunch) + PersistableEventQueue.resetRehydrationFlag() + val freshQueue = PersistableEventQueue( + maxCapacity = 2000, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + freshQueue.rehydrate() + + // Set up dispatcher with working network (online at launch) + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) + val initOptions = InitOptions( + writeKey = "test-key", + ingestionHost = "https://api.example.com", + flushIntervalSeconds = 10 + ) + val dispatcher = Dispatcher( + options = initOptions, + queue = freshQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + // Drain overflow from previous session + val drained = freshQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(3, drained) + assertEquals(1, fakeClient.requests.size) // Direct to network + assertNull(overflowDiskStore.read()) // Cleaned up + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `offline overflow disabled does not write to disk on overflow`() { + val overflowDir = createTempDirectory("metarouter-overflow-disabled").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 3, + eventTTLMs = 0 + ) + + // Overflow NOT enabled (default) + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + repeat(3) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + + smallQueue.flushOverflowBufferToDisk() + + // No overflow disk data — events were dropped normally + assertNull(overflowDiskStore.read()) + assertEquals(5, smallQueue.size()) + + overflowDir.deleteRecursively() + } + + @Test + fun `setOfflineOverflowEnabled toggles overflow behavior`() { + val overflowDir = createTempDirectory("metarouter-overflow-toggle").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + overflowBufferBatchThreshold = 100, + eventTTLMs = 0 + ) + + // Fill queue + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + + // Overflow disabled — events dropped + smallQueue.enqueue(createTestEvent("dropped-1")) + assertEquals(5, smallQueue.size()) + assertEquals(0, smallQueue.offlineOverflowCount()) + + // Enable overflow + smallQueue.setOfflineOverflowEnabled(true) + smallQueue.enqueue(createTestEvent("buffered-1")) + assertEquals(5, smallQueue.size()) + assertTrue(smallQueue.offlineOverflowCount() > 0) + + // Disable overflow again + smallQueue.setOfflineOverflowEnabled(false) + smallQueue.enqueue(createTestEvent("dropped-2")) + assertEquals(5, smallQueue.size()) + + overflowDir.deleteRecursively() + } + private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload { return EnrichedEventPayload( type = EventType.TRACK, diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt index ee46fcd..885caa1 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt @@ -177,6 +177,35 @@ class EventDiskStoreTest { assertFalse(snapshotFile.exists()) } + @Test + fun `custom filename creates separate store`() { + val customStore = EventDiskStore(tempDir, filename = "custom-overflow.v1.json") + val defaultStore = EventDiskStore(tempDir) // Uses default "queue.v1.json" + + val snapshot1 = QueueSnapshot(version = 1, events = listOf(createTestEvent("default-1"))) + val snapshot2 = QueueSnapshot(version = 1, events = listOf(createTestEvent("custom-1"), createTestEvent("custom-2"))) + + defaultStore.write(snapshot1) + customStore.write(snapshot2) + + // Each store reads its own file independently + val defaultLoaded = defaultStore.read() + val customLoaded = customStore.read() + + assertNotNull(defaultLoaded) + assertEquals(1, defaultLoaded!!.events.size) + assertEquals("default-1", defaultLoaded.events[0].messageId) + + assertNotNull(customLoaded) + assertEquals(2, customLoaded!!.events.size) + assertEquals("custom-1", customLoaded.events[0].messageId) + + // Delete one doesn't affect the other + customStore.delete() + assertNull(customStore.read()) + assertNotNull(defaultStore.read()) + } + private fun createTestEvent(messageId: String): EnrichedEventPayload { return EnrichedEventPayload( type = EventType.TRACK, From e29e1a229d67778f67d5b6e5afc99c2030b1f169 Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Fri, 10 Apr 2026 15:54:59 -0600 Subject: [PATCH 2/6] fix: remove add to project --- .github/workflows/add-to-project.yml | 14 -------------- 1 file changed, 14 deletions(-) delete mode 100644 .github/workflows/add-to-project.yml diff --git a/.github/workflows/add-to-project.yml b/.github/workflows/add-to-project.yml deleted file mode 100644 index 0607710..0000000 --- a/.github/workflows/add-to-project.yml +++ /dev/null @@ -1,14 +0,0 @@ -name: Add PR to Mobile SDK Board - -on: - pull_request: - types: [opened] - -jobs: - add-to-project: - runs-on: ubuntu-latest - steps: - - uses: actions/add-to-project@v1 - with: - project-url: https://github.com/orgs/metarouterio/projects/1 - github-token: ${{ secrets.ADD_TO_PROJECT_PAT }} From c7f24ce4e0999b3d0037c2a27ecc3eacbc283932 Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Wed, 15 Apr 2026 17:40:45 -0600 Subject: [PATCH 3/6] fix: improving flush performance --- .../analytics/MetaRouterAnalyticsClient.kt | 20 +- .../analytics/dispatcher/Dispatcher.kt | 74 ++- .../analytics/queue/EventQueueInterface.kt | 9 + .../analytics/queue/PersistableEventQueue.kt | 222 ++++---- .../MetaRouterAnalyticsClientTest.kt | 10 +- .../analytics/dispatcher/DispatcherTest.kt | 19 +- .../queue/PersistableEventQueueTest.kt | 484 +++++++++++++++--- 7 files changed, 613 insertions(+), 225 deletions(-) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt index bb5e196..92155d6 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt @@ -182,22 +182,18 @@ class MetaRouterAnalyticsClient private constructor( networkMonitor = DebouncedNetworkMonitor(rawMonitor, scope) networkMonitor.start { connected -> if (connected) { - persistableEventQueue?.setOfflineOverflowEnabled(false) circuitBreaker.reset() dispatcher.resumeFromOffline() - // Two independent flush paths: - scope.launch { dispatcher.flush() } // (1) memory queue -> network - scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } // (2) disk -> network directly + scope.launch { dispatcher.flush() } + scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } } else { dispatcher.pauseForOffline() - persistableEventQueue?.setOfflineOverflowEnabled(true) } } - // If starting offline, pause dispatcher and enable overflow immediately + // If starting offline, pause dispatcher immediately if (!networkMonitor.isConnected) { dispatcher.pauseForOffline() - persistableEventQueue?.setOfflineOverflowEnabled(true) } // If online at launch and overflow exists from previous session, drain directly to network @@ -207,6 +203,15 @@ class MetaRouterAnalyticsClient private constructor( startEventProcessor() + // After each successful flush, drain any overflow disk events + persistableEventQueue?.let { pQueue -> + dispatcher.onFlushComplete = { + if (networkMonitor.isConnected) { + scope.launch { pQueue.drainDiskOverflowToNetwork(dispatcher) } + } + } + } + dispatcher.start() // Register best-effort terminate flush via ComponentCallbacks2 @@ -431,7 +436,6 @@ class MetaRouterAnalyticsClient private constructor( } flush() persistableEventQueue?.flushToDisk() - persistableEventQueue?.flushOverflowBufferToDisk() dispatcher.pause() } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt index af656a9..3712d98 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt @@ -60,6 +60,12 @@ class Dispatcher( */ var onFatalConfigError: ((Int) -> Unit)? = null + /** + * Callback invoked after a successful flush cycle completes. + * Used to trigger overflow disk drain when events spill to disk while online. + */ + var onFlushComplete: (() -> Unit)? = null + // ===== Lifecycle ===== /** @@ -169,6 +175,9 @@ class Dispatcher( try { processUntilEmpty() + if (!networkPaused) { + onFlushComplete?.invoke() + } } finally { isFlushing.set(false) } @@ -198,10 +207,10 @@ class Dispatcher( * Used by PersistableEventQueue.drainDiskOverflowToNetwork() to flush overflow * events from disk without loading them into the memory queue. * - * @return true if the batch was sent successfully, false on any failure + * @return NetworkResponse on HTTP completion, null on encoding or network error */ - suspend fun sendBatchDirect(events: List): Boolean { - if (events.isEmpty()) return true + suspend fun sendBatchDirect(events: List): NetworkResponse? { + if (events.isEmpty()) return NetworkResponse(200, emptyMap(), null) val sentAt = getIso8601Timestamp() val batch = events.map { it.copy(sentAt = sentAt) } @@ -213,7 +222,7 @@ class Dispatcher( body = json.encodeToString(payload).toByteArray(Charsets.UTF_8) } catch (e: Exception) { Logger.error("Failed to encode direct batch of ${batch.size} events: ${e.message}") - return false + return null } val headers = mutableMapOf() @@ -221,17 +230,18 @@ class Dispatcher( headers["Trace"] = "true" } + Logger.log("Making API call to: $url (overflow drain, ${batch.size} events)") + return try { - val response = networkClient.postJson( + networkClient.postJson( url = url, body = body, timeoutMs = config.timeoutMs, additionalHeaders = if (headers.isNotEmpty()) headers else null ) - response.statusCode in 200..299 } catch (e: Exception) { Logger.warn("Direct batch send failed: ${e.message}") - false + null } } @@ -253,7 +263,11 @@ class Dispatcher( private suspend fun processUntilEmpty() { if (networkPaused) { - Logger.log("Dispatcher skipping flush — device is offline") + if (queue.flushToOfflineStorage()) { + Logger.log("Flushed queue to offline storage") + } else { + Logger.log("Dispatcher skipping flush — device is offline") + } return } @@ -326,8 +340,8 @@ class Dispatcher( response: NetworkResponse, batch: List ): FlushAction { - return when (response.statusCode) { - in 200..299 -> { + return when (ResponseCategory.from(response.statusCode)) { + ResponseCategory.SUCCESS -> { consecutiveRetries.set(0) circuitBreaker.onSuccess() val current = maxBatchSize.get() @@ -338,7 +352,7 @@ class Dispatcher( Logger.log("API call successful") FlushAction.Continue } - in 500..599, 408 -> { + ResponseCategory.SERVER_ERROR -> { consecutiveRetries.incrementAndGet() circuitBreaker.onFailure() queue.requeueToFront(batch) @@ -347,7 +361,7 @@ class Dispatcher( Logger.warn("Server error ${response.statusCode}, will retry ${batch.size} event(s) in ${waitMs}ms (circuit: ${circuitBreaker.getState()}, retry #${consecutiveRetries.get()})") FlushAction.RetryAfter(waitMs) } - 429 -> { + ResponseCategory.RATE_LIMITED -> { consecutiveRetries.incrementAndGet() circuitBreaker.onFailure() queue.requeueToFront(batch) @@ -357,7 +371,7 @@ class Dispatcher( Logger.warn("Rate limited (429), will retry ${batch.size} event(s) in ${waitMs}ms (circuit: ${circuitBreaker.getState()}, retry #${consecutiveRetries.get()})") FlushAction.RetryAfter(waitMs) } - 413 -> { + ResponseCategory.PAYLOAD_TOO_LARGE -> { circuitBreaker.onNonRetryable() val currentSize = maxBatchSize.get() if (currentSize > 1) { @@ -371,23 +385,18 @@ class Dispatcher( FlushAction.Stop } } - 401, 403, 404 -> { + ResponseCategory.FATAL_CONFIG -> { Logger.error("Fatal configuration error: ${response.statusCode} - clearing queue and stopping") queue.clear() stop() onFatalConfigError?.invoke(response.statusCode) FlushAction.Stop } - in 400..499 -> { + ResponseCategory.CLIENT_ERROR -> { circuitBreaker.onNonRetryable() Logger.warn("Client error ${response.statusCode} - dropping batch of ${batch.size} events") FlushAction.Continue } - else -> { - circuitBreaker.onNonRetryable() - Logger.warn("Unexpected status ${response.statusCode} - continuing") - FlushAction.Continue - } } } @@ -422,6 +431,31 @@ private sealed class FlushAction { data object Stop : FlushAction() } +/** + * Classifies an HTTP status code into a response category. + * Shared by the main dispatcher flush loop and the overflow drain. + */ +internal enum class ResponseCategory { + SUCCESS, + SERVER_ERROR, + RATE_LIMITED, + PAYLOAD_TOO_LARGE, + FATAL_CONFIG, + CLIENT_ERROR; + + companion object { + fun from(statusCode: Int): ResponseCategory = when (statusCode) { + in 200..299 -> SUCCESS + in 500..599, 408 -> SERVER_ERROR + 429 -> RATE_LIMITED + 413 -> PAYLOAD_TOO_LARGE + 401, 403, 404 -> FATAL_CONFIG + in 400..499 -> CLIENT_ERROR + else -> CLIENT_ERROR // Unknown status codes are non-retryable (drop, don't retry) + } + } +} + data class DispatcherDebugInfo( val isRunning: Boolean, val maxBatchSize: Int, diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt index 454cd60..01597cf 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/EventQueueInterface.kt @@ -12,4 +12,13 @@ interface EventQueueInterface { fun drain(max: Int): List fun requeueToFront(events: List) fun clear() + + /** + * Flush all events to offline storage. Called by the dispatcher when + * a flush is triggered but the device is offline. + * Default implementation is a no-op (returns false). + * + * @return true if events were flushed to offline storage + */ + fun flushToOfflineStorage(): Boolean = false } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt index 8b9a0e6..7a1eb96 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt @@ -1,6 +1,7 @@ package com.metarouter.analytics.queue import com.metarouter.analytics.dispatcher.Dispatcher +import com.metarouter.analytics.dispatcher.ResponseCategory import com.metarouter.analytics.storage.EventDiskStore import com.metarouter.analytics.storage.QueueSnapshot import com.metarouter.analytics.types.EnrichedEventPayload @@ -24,6 +25,12 @@ import java.util.concurrent.atomic.AtomicBoolean * Capacity model: * - Single shared cap: [maxCapacity] events OR [maxCapacityBytes] total, whichever is hit first * - Flush threshold: [flushThresholdEvents] events OR [flushThresholdBytes], whichever is hit first + * + * Offline overflow model: + * - When offline, flush triggers write the entire memory queue to the overflow disk store + * instead of sending to the network. Same triggers, swapped destination. + * - When the queue hits capacity (online or offline), it flushes to overflow disk rather than dropping. + * - On online transition, overflow events drain from disk directly to network. */ class PersistableEventQueue( private val maxCapacity: Int = 2000, @@ -33,8 +40,7 @@ class PersistableEventQueue( private val maxCapacityBytes: Long = 5L * 1024 * 1024, private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000, private val maxOfflineDiskEvents: Int = 10000, - private val overflowDiskStore: EventDiskStore? = null, - private val overflowBufferBatchThreshold: Int = 100 + private val overflowDiskStore: EventDiskStore? = null ) : EventQueueInterface { companion object { @@ -64,16 +70,14 @@ class PersistableEventQueue( // Cache of per-event byte sizes (parallel to memoryQueue order) private val eventSizes = ArrayDeque() - // Offline overflow state - @Volatile - private var offlineOverflowEnabled = false - private val overflowBuffer = ArrayDeque() + // Guards against concurrent drainDiskOverflowToNetwork invocations + private val isDraining = AtomicBoolean(false) @Synchronized override fun size(): Int = memoryQueue.size /** - * Enqueue an event to memory. No disk I/O. + * Enqueue an event to memory. No disk I/O unless offline and at capacity. * Drops oldest events if event count or byte capacity is exceeded. */ @Synchronized @@ -85,14 +89,10 @@ class PersistableEventQueue( dropOldest("byte capacity") } - // Drop oldest if at event count capacity + // At event count capacity: flush to overflow disk if available, otherwise drop oldest if (memoryQueue.size >= maxCapacity) { - if (offlineOverflowEnabled && overflowDiskStore != null) { - // Overflow to buffer instead of dropping - val oldest = memoryQueue.removeFirst() - val size = eventSizes.removeFirst() - estimatedBytes -= size - bufferOverflow(oldest) + if (overflowDiskStore != null) { + flushToOverflowDiskInternal() } else { dropOldest("event count capacity") } @@ -150,7 +150,7 @@ class PersistableEventQueue( } /** - * Clear all events from memory and delete the disk snapshot. + * Clear all events from memory and delete disk snapshots. */ @Synchronized override fun clear() { @@ -158,7 +158,6 @@ class PersistableEventQueue( memoryQueue.clear() eventSizes.clear() estimatedBytes = 0L - overflowBuffer.clear() diskStore.delete() overflowDiskStore?.delete() if (count > 0) { @@ -166,6 +165,18 @@ class PersistableEventQueue( } } + /** + * Flush all events to offline overflow disk storage. + * Called by the dispatcher when a flush is triggered but the device is offline. + * Drains the entire memory queue and appends to the overflow disk store. + */ + @Synchronized + override fun flushToOfflineStorage(): Boolean { + if (overflowDiskStore == null || memoryQueue.isEmpty()) return false + flushToOverflowDiskInternal() + return true + } + /** * Flush current memory state to disk as a full overwrite snapshot. * Skips the write and deletes any existing snapshot if the queue is empty. @@ -261,65 +272,6 @@ class PersistableEventQueue( // ===== Offline Overflow ===== - /** - * Enable or disable offline overflow mode. When enabled and memory queue is full, - * oldest events overflow to buffer (then disk) instead of being dropped. - */ - @Synchronized - fun setOfflineOverflowEnabled(enabled: Boolean) { - offlineOverflowEnabled = enabled - } - - /** - * Buffer an evicted event for batched disk write. - * Flushes buffer to disk when batch threshold reached. - * Must be called within a synchronized block. - */ - private fun bufferOverflow(event: EnrichedEventPayload) { - overflowBuffer.addLast(event) - if (overflowBuffer.size >= overflowBufferBatchThreshold) { - flushOverflowBufferToDiskInternal() - } - } - - /** - * Flush the in-memory overflow buffer to the overflow disk store. - * Called when buffer reaches batch threshold, on app backgrounding, - * or on offline->online transition before draining. - */ - @Synchronized - fun flushOverflowBufferToDisk() { - flushOverflowBufferToDiskInternal() - } - - /** - * Internal implementation — assumes lock is already held or called within synchronized block. - */ - private fun flushOverflowBufferToDiskInternal() { - val store = overflowDiskStore ?: return - if (overflowBuffer.isEmpty()) return - - val batch = overflowBuffer.toList() - overflowBuffer.clear() - - val existing = try { - store.read()?.events ?: emptyList() - } catch (e: Exception) { - Logger.warn("Failed to read overflow disk store: ${e.message}") - emptyList() - } - - var combined = existing + batch - if (combined.size > maxOfflineDiskEvents) { - val dropCount = combined.size - maxOfflineDiskEvents - combined = combined.drop(dropCount) - Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") - } - - store.write(QueueSnapshot(version = 1, events = combined)) - Logger.log("Flushed ${batch.size} overflow events to disk (total on disk: ${combined.size})") - } - /** * Drain overflow events directly from disk to network in batches. * Does NOT load events into the memory queue — sends via dispatcher's direct path. @@ -329,15 +281,12 @@ class PersistableEventQueue( */ suspend fun drainDiskOverflowToNetwork(dispatcher: Dispatcher): Int { val store = overflowDiskStore ?: return 0 - - // First, flush any remaining buffer to disk - synchronized(this) { - flushOverflowBufferToDiskInternal() - } + if (!isDraining.compareAndSet(false, true)) return 0 var totalDrained = 0 + var currentBatchSize = 100 - while (true) { + try { while (true) { val snapshot = store.read() ?: return totalDrained val now = System.currentTimeMillis() val events = if (eventTTLMs > 0) { @@ -352,40 +301,105 @@ class PersistableEventQueue( } // Take a batch from the front (oldest first) - val batchSize = minOf(100, events.size) - val batch = events.take(batchSize) - val remaining = events.drop(batchSize) - - // Send batch directly to network via dispatcher's HTTP path - val success = dispatcher.sendBatchDirect(batch) - if (success) { - totalDrained += batch.size - if (remaining.isEmpty()) { + val batch = events.take(minOf(currentBatchSize, events.size)) + val remaining = events.drop(batch.size) + + val response = dispatcher.sendBatchDirect(batch) + + if (response == null) { + Logger.warn("Offline overflow drain paused — network error ($totalDrained drained so far)") + return totalDrained + } + + when (ResponseCategory.from(response.statusCode)) { + ResponseCategory.SUCCESS -> { + totalDrained += batch.size + if (currentBatchSize < 100) { + currentBatchSize = minOf(currentBatchSize * 2, 100) + } + if (remaining.isEmpty()) { + store.delete() + Logger.log("Offline overflow disk drain complete ($totalDrained events)") + return totalDrained + } else { + store.write(QueueSnapshot(version = 1, events = remaining)) + } + } + ResponseCategory.PAYLOAD_TOO_LARGE -> { + if (currentBatchSize > 1) { + currentBatchSize = maxOf(1, currentBatchSize / 2) + Logger.warn("Overflow drain: payload too large (413) — reduced batch to $currentBatchSize") + } else { + Logger.warn("Overflow drain: dropping oversized event at batchSize=1 (messageId: ${batch.first().messageId})") + if (remaining.isEmpty()) { + store.delete() + } else { + store.write(QueueSnapshot(version = 1, events = remaining)) + } + } + } + ResponseCategory.FATAL_CONFIG -> { + Logger.error("Overflow drain: fatal config error ${response.statusCode} — deleting overflow store") store.delete() - Logger.log("Offline overflow disk drain complete ($totalDrained events)") return totalDrained - } else { - store.write(QueueSnapshot(version = 1, events = remaining)) } - } else { - // Network failed — stop draining, will retry on next online transition - Logger.warn("Offline overflow drain paused — network send failed ($totalDrained drained so far)") - return totalDrained + ResponseCategory.CLIENT_ERROR -> { + Logger.warn("Overflow drain: client error ${response.statusCode} — dropping batch of ${batch.size}") + if (remaining.isEmpty()) { + store.delete() + return totalDrained + } else { + store.write(QueueSnapshot(version = 1, events = remaining)) + } + } + ResponseCategory.SERVER_ERROR, ResponseCategory.RATE_LIMITED -> { + Logger.warn("Offline overflow drain paused — ${response.statusCode} ($totalDrained drained so far)") + return totalDrained + } } + } } finally { + isDraining.set(false) } } + // ===== Internal ===== + /** - * Get count of events in the offline overflow buffer + disk store. + * Flush all events from memory queue to the overflow disk store. + * Appends to existing overflow events on disk. Enforces maxOfflineDiskEvents cap. + * Resets the memory queue to empty. + * Must be called within a synchronized block. + * + * Note: performs disk I/O (read + write) while holding the synchronized lock. + * This blocks enqueue/drain during the write. Acceptable because local disk I/O + * is fast (sub-ms for typical event sizes), but worth revisiting if profiling + * shows contention under extreme throughput. */ - @Synchronized - fun offlineOverflowCount(): Int { - val diskCount = try { - overflowDiskStore?.read()?.events?.size ?: 0 + private fun flushToOverflowDiskInternal() { + val store = overflowDiskStore ?: return + if (memoryQueue.isEmpty()) return + + val batch = memoryQueue.toList() + memoryQueue.clear() + eventSizes.clear() + estimatedBytes = 0L + + val existing = try { + store.read()?.events ?: emptyList() } catch (e: Exception) { - 0 + Logger.warn("Failed to read overflow disk store: ${e.message}") + emptyList() + } + + var combined = existing + batch + if (combined.size > maxOfflineDiskEvents) { + val dropCount = combined.size - maxOfflineDiskEvents + combined = combined.drop(dropCount) + Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") } - return overflowBuffer.size + diskCount + + store.write(QueueSnapshot(version = 1, events = combined)) + Logger.log("Flushed ${batch.size} events to offline disk (total on disk: ${combined.size})") } private fun dropOldest(reason: String) { diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt index 4780a43..0cc9d9e 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/MetaRouterAnalyticsClientTest.kt @@ -420,7 +420,7 @@ class MetaRouterAnalyticsClientTest { // ===== Queue Overflow ===== @Test - fun `queue overflow drops oldest events`() = runBlocking { + fun `queue overflow flushes to disk instead of dropping`() = runBlocking { val smallOptions = options.copy(maxQueueEvents = 5) val client = MetaRouterAnalyticsClient.initialize(context, smallOptions) @@ -428,13 +428,13 @@ class MetaRouterAnalyticsClientTest { repeat(10) { i -> client.track("Event $i") } - // Wait until queue has processed and is at capacity (5 events) - awaitCondition { (client.getDebugInfo()["queueLength"] as Int) == 5 } + // Wait until all events have been processed through the channel + awaitCondition { (client.getDebugInfo()["queueLength"] as Int) > 0 } val debugInfo = client.getDebugInfo() - // Queue should have exactly 5 events (oldest dropped) - assertEquals(5, debugInfo["queueLength"]) + // Queue should not exceed maxQueueEvents (overflow goes to disk) + assertTrue((debugInfo["queueLength"] as Int) <= 5) } // ===== Flush (Stub) ===== diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt index 7409d2c..644f142 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/dispatcher/DispatcherTest.kt @@ -1053,46 +1053,49 @@ class DispatcherTest { // ===== sendBatchDirect Tests ===== @Test - fun `sendBatchDirect returns true on success`() = runTest { + fun `sendBatchDirect returns 200 response on success`() = runTest { val dispatcher = createDispatcher() networkClient.nextResponse = NetworkResponse(200, emptyMap(), null) val events = listOf(createEvent("direct-1"), createEvent("direct-2")) val result = dispatcher.sendBatchDirect(events) - assertTrue(result) + assertNotNull(result) + assertEquals(200, result!!.statusCode) assertEquals(1, networkClient.requests.size) } @Test - fun `sendBatchDirect returns false on server error`() = runTest { + fun `sendBatchDirect returns 500 response on server error`() = runTest { val dispatcher = createDispatcher() networkClient.nextResponse = NetworkResponse(500, emptyMap(), null) val events = listOf(createEvent("direct-1")) val result = dispatcher.sendBatchDirect(events) - assertFalse(result) + assertNotNull(result) + assertEquals(500, result!!.statusCode) } @Test - fun `sendBatchDirect returns false on network error`() = runTest { + fun `sendBatchDirect returns null on network error`() = runTest { val dispatcher = createDispatcher() networkClient.nextException = java.io.IOException("Connection refused") val events = listOf(createEvent("direct-1")) val result = dispatcher.sendBatchDirect(events) - assertFalse(result) + assertNull(result) } @Test - fun `sendBatchDirect returns true for empty list`() = runTest { + fun `sendBatchDirect returns 200 for empty list`() = runTest { val dispatcher = createDispatcher() val result = dispatcher.sendBatchDirect(emptyList()) - assertTrue(result) + assertNotNull(result) + assertEquals(200, result!!.statusCode) assertEquals(0, networkClient.requests.size) // No request made } diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt index 5482a75..f3353cd 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt @@ -591,7 +591,7 @@ class PersistableEventQueueTest { // ===== Offline Overflow ===== @Test - fun `memory queue overflow while offline buffers then flushes to disk`() { + fun `capacity overflow flushes entire queue to disk`() { val overflowDir = createTempDirectory("metarouter-overflow-test").toFile() val overflowDiskStore = EventDiskStore(overflowDir) @@ -600,43 +600,37 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 3, // Low threshold for testing eventTTLMs = 0 ) - smallQueue.setOfflineOverflowEnabled(true) - // Fill queue to capacity repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } assertEquals(5, smallQueue.size()) - // Enqueue more — should overflow to buffer, then to disk at threshold=3 - repeat(4) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } - - // Memory queue should still be at max capacity - assertEquals(5, smallQueue.size()) + // Enqueue one more — should flush all 5 to disk, then add the new event + smallQueue.enqueue(createTestEvent("overflow-0")) - // Flush remaining buffer to disk - smallQueue.flushOverflowBufferToDisk() + // Memory queue should have just the new event + assertEquals(1, smallQueue.size()) - // Verify overflow events are on disk - // Queue was: fill-0..fill-4. When overflow-0 comes in, fill-0 is evicted to buffer. - // overflow-1 → fill-1 evicted. overflow-2 → fill-2 evicted (buffer hits threshold=3, auto-flush). - // overflow-3 → fill-3 evicted. Explicit flush writes remaining buffer. + // All 5 original events should be on overflow disk val snapshot = overflowDiskStore.read() assertNotNull(snapshot) - assertEquals(4, snapshot!!.events.size) + assertEquals(5, snapshot!!.events.size) assertEquals("fill-0", snapshot.events[0].messageId) - assertEquals("fill-1", snapshot.events[1].messageId) - assertEquals("fill-2", snapshot.events[2].messageId) - assertEquals("fill-3", snapshot.events[3].messageId) + assertEquals("fill-4", snapshot.events[4].messageId) + + // Drain memory to verify contents + val drained = smallQueue.drain(10) + assertEquals(1, drained.size) + assertEquals("overflow-0", drained[0].messageId) overflowDir.deleteRecursively() } @Test - fun `overflow writes are batched not per-event`() { - val overflowDir = createTempDirectory("metarouter-overflow-batch").toFile() + fun `flushToOfflineStorage drains entire queue to disk`() { + val overflowDir = createTempDirectory("metarouter-overflow-flush").toFile() val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( @@ -644,26 +638,55 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 10, // Won't trigger during this test eventTTLMs = 0 ) - smallQueue.setOfflineOverflowEnabled(true) + // Fill queue with 3 events + repeat(3) { i -> smallQueue.enqueue(createTestEvent("event-$i")) } + assertEquals(3, smallQueue.size()) - // Fill queue - repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + // Flush to offline storage (simulates dispatcher offline flush) + val flushed = smallQueue.flushToOfflineStorage() + assertTrue(flushed) - // Overflow 8 events — batch threshold is 10, so no auto-flush to disk - repeat(8) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + // Memory queue should be empty + assertEquals(0, smallQueue.size()) - // No disk write should have happened yet (buffer < threshold) - assertNull(overflowDiskStore.read()) + // Events should be on overflow disk + val snapshot = overflowDiskStore.read() + assertNotNull(snapshot) + assertEquals(3, snapshot!!.events.size) - // Explicit flush writes all at once - smallQueue.flushOverflowBufferToDisk() + overflowDir.deleteRecursively() + } + + @Test + fun `flushToOfflineStorage appends to existing overflow disk data`() { + val overflowDir = createTempDirectory("metarouter-overflow-append").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // Pre-populate overflow disk with 3 events + val existing = (1..3).map { createTestEvent("existing-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = existing)) + + // Add 2 events to memory and flush + repeat(2) { i -> smallQueue.enqueue(createTestEvent("new-$i")) } + smallQueue.flushToOfflineStorage() + + // Disk should have all 5 val snapshot = overflowDiskStore.read() assertNotNull(snapshot) - assertEquals(8, snapshot!!.events.size) + assertEquals(5, snapshot!!.events.size) + assertEquals("existing-1", snapshot.events[0].messageId) + assertEquals("new-1", snapshot.events[4].messageId) overflowDir.deleteRecursively() } @@ -676,24 +699,23 @@ class PersistableEventQueueTest { val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, - maxOfflineDiskEvents = 10, // Cap at 10 + maxOfflineDiskEvents = 8, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 5, eventTTLMs = 0 ) - smallQueue.setOfflineOverflowEnabled(true) - - // Fill queue - repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + // Fill and overflow multiple times to exceed cap + // Round 1: fill 5, overflow triggers flush of 5 to disk + repeat(5) { i -> smallQueue.enqueue(createTestEvent("batch1-$i")) } + smallQueue.flushToOfflineStorage() - // Overflow 15 events (exceeds maxOfflineDiskEvents=10) - repeat(15) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } - smallQueue.flushOverflowBufferToDisk() + // Round 2: fill 5 more, flush again — total would be 10, cap is 8 + repeat(5) { i -> smallQueue.enqueue(createTestEvent("batch2-$i")) } + smallQueue.flushToOfflineStorage() val snapshot = overflowDiskStore.read() assertNotNull(snapshot) - assertTrue("Should cap at maxOfflineDiskEvents", snapshot!!.events.size <= 10) + assertTrue("Should cap at maxOfflineDiskEvents", snapshot!!.events.size <= 8) overflowDir.deleteRecursively() } @@ -714,7 +736,6 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 100, eventTTLMs = 0 ) @@ -742,10 +763,8 @@ class PersistableEventQueueTest { val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(5, drained) - assertEquals(1, fakeClient.requests.size) // Batch of 5 events - assertNull(overflowDiskStore.read()) // Disk file deleted - - // Memory queue should be untouched (0 events — drain goes directly to network) + assertEquals(1, fakeClient.requests.size) + assertNull(overflowDiskStore.read()) assertEquals(0, overflowQueue.size()) overflowDir.deleteRecursively() @@ -768,7 +787,6 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 100, eventTTLMs = 0 ) @@ -796,7 +814,6 @@ class PersistableEventQueueTest { val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(0, drained) - // Events should still be on disk val remaining = overflowDiskStore.read() assertNotNull(remaining) assertEquals(5, remaining!!.events.size) @@ -818,7 +835,6 @@ class PersistableEventQueueTest { eventTTLMs = 0 ) - // Write overflow data overflowDiskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("overflow-1")))) overflowQueue.enqueue(createTestEvent("memory-1")) @@ -874,20 +890,36 @@ class PersistableEventQueueTest { config = DispatcherConfig() ) - // Drain overflow from previous session val drained = freshQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(3, drained) - assertEquals(1, fakeClient.requests.size) // Direct to network - assertNull(overflowDiskStore.read()) // Cleaned up + assertEquals(1, fakeClient.requests.size) + assertNull(overflowDiskStore.read()) overflowDir.deleteRecursively() unmockkStatic(Log::class) } @Test - fun `offline overflow disabled does not write to disk on overflow`() { - val overflowDir = createTempDirectory("metarouter-overflow-disabled").toFile() + fun `no overflow disk store drops events normally at capacity`() { + // No overflowDiskStore — events are dropped when queue is full + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + overflowDiskStore = null, + eventTTLMs = 0 + ) + + repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + repeat(3) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } + + // Events were dropped, queue stays at capacity + assertEquals(5, smallQueue.size()) + } + + @Test + fun `capacity overflow always flushes to disk when overflow store exists`() { + val overflowDir = createTempDirectory("metarouter-overflow-always").toFile() val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( @@ -895,25 +927,23 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 3, eventTTLMs = 0 ) - // Overflow NOT enabled (default) + // Fill queue repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } - repeat(3) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } - - smallQueue.flushOverflowBufferToDisk() - // No overflow disk data — events were dropped normally - assertNull(overflowDiskStore.read()) - assertEquals(5, smallQueue.size()) + // Next enqueue at capacity flushes entire queue to disk + smallQueue.enqueue(createTestEvent("triggers-flush")) + assertEquals(1, smallQueue.size()) + assertNotNull(overflowDiskStore.read()) + assertEquals(5, overflowDiskStore.read()!!.events.size) overflowDir.deleteRecursively() } @Test - fun `setOfflineOverflowEnabled toggles overflow behavior`() { + fun `flushToOfflineStorage drains queue when overflow store exists`() { val overflowDir = createTempDirectory("metarouter-overflow-toggle").toFile() val overflowDiskStore = EventDiskStore(overflowDir) @@ -922,30 +952,324 @@ class PersistableEventQueueTest { diskStore = diskStore, maxOfflineDiskEvents = 100, overflowDiskStore = overflowDiskStore, - overflowBufferBatchThreshold = 100, eventTTLMs = 0 ) - // Fill queue - repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + repeat(3) { i -> smallQueue.enqueue(createTestEvent("event-$i")) } - // Overflow disabled — events dropped - smallQueue.enqueue(createTestEvent("dropped-1")) - assertEquals(5, smallQueue.size()) - assertEquals(0, smallQueue.offlineOverflowCount()) + assertTrue(smallQueue.flushToOfflineStorage()) + assertEquals(0, smallQueue.size()) + assertNotNull(overflowDiskStore.read()) + assertEquals(3, overflowDiskStore.read()!!.events.size) - // Enable overflow - smallQueue.setOfflineOverflowEnabled(true) - smallQueue.enqueue(createTestEvent("buffered-1")) - assertEquals(5, smallQueue.size()) - assertTrue(smallQueue.offlineOverflowCount() > 0) + overflowDir.deleteRecursively() + } - // Disable overflow again - smallQueue.setOfflineOverflowEnabled(false) - smallQueue.enqueue(createTestEvent("dropped-2")) - assertEquals(5, smallQueue.size()) + @Test + fun `flushToOfflineStorage returns false without overflow store`() { + // Queue without overflow store + val noOverflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + overflowDiskStore = null, + eventTTLMs = 0 + ) + + repeat(3) { i -> noOverflowQueue.enqueue(createTestEvent("event-$i")) } + + assertFalse(noOverflowQueue.flushToOfflineStorage()) + assertEquals(3, noOverflowQueue.size()) + } + + // ===== Drain Response Handling ===== + + @Test + fun `drain halves batch on 413 and retries`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-413").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // Pre-populate overflow disk with 4 events + val events = (1..4).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + // First call: 413, second call (halved batch): 200, third call: 200 + fakeClient.enqueueResponses( + NetworkResponse(413, emptyMap(), null), + NetworkResponse(200, emptyMap(), null), + NetworkResponse(200, emptyMap(), null) + ) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(4, drained) + assertNull(overflowDiskStore.read()) overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drain drops oversized event at batchSize 1 on 413`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-413-drop").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // 2 events — first will be too large, second should still send + val events = listOf(createTestEvent("big-event"), createTestEvent("normal-event")) + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + // 413 responses until batch is 1, then 413 again (drop), then 200 for remaining + fakeClient.enqueueResponses( + NetworkResponse(413, emptyMap(), null), // batch=100 -> halve + NetworkResponse(413, emptyMap(), null), // batch=50 -> halve (and so on until 1) + NetworkResponse(413, emptyMap(), null), + NetworkResponse(413, emptyMap(), null), + NetworkResponse(413, emptyMap(), null), + NetworkResponse(413, emptyMap(), null), + NetworkResponse(413, emptyMap(), null), // batch=1 -> drop big-event + NetworkResponse(200, emptyMap(), null) // normal-event succeeds + ) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + // Only normal-event was successfully sent + assertEquals(1, drained) + assertNull(overflowDiskStore.read()) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drain deletes overflow store on fatal config error`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-fatal").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + val events = (1..5).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(401, emptyMap(), null) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(0, drained) + assertNull(overflowDiskStore.read()) // Overflow store deleted on fatal error + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drain drops batch on client error and continues`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-400").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // Write 150 events so drain takes 2 batches (100 + 50) + val events = (1..150).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + // First batch: 400 (drop), second batch: 200 (success) + fakeClient.enqueueResponses( + NetworkResponse(400, emptyMap(), null), + NetworkResponse(200, emptyMap(), null) + ) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + // Only second batch of 50 counted as drained + assertEquals(50, drained) + assertNull(overflowDiskStore.read()) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drain pauses on 429 rate limit`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-429").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + val events = (1..5).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(429, emptyMap(), null) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + assertEquals(0, drained) + // Events should still be on disk + val remaining = overflowDiskStore.read() + assertNotNull(remaining) + assertEquals(5, remaining!!.events.size) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `drain filters expired events by TTL`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-ttl").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val oneDayMs = 24L * 60 * 60 * 1000 + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = oneDayMs // 1 day TTL + ) + + // Mix of expired and fresh events + val expiredTimestamp = formatTimestamp(System.currentTimeMillis() - 2 * oneDayMs) // 2 days ago + val freshTimestamp = formatTimestamp(System.currentTimeMillis() - 1000) // 1 second ago + val events = listOf( + createTestEvent("expired-1", timestamp = expiredTimestamp), + createTestEvent("expired-2", timestamp = expiredTimestamp), + createTestEvent("fresh-1", timestamp = freshTimestamp), + createTestEvent("fresh-2", timestamp = freshTimestamp) + ) + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) + + // Only the 2 fresh events should have been sent + assertEquals(2, drained) + assertNull(overflowDiskStore.read()) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) } private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload { From 4203ff893eee9a8fba42c90b41be37421b77b9db Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Thu, 16 Apr 2026 11:01:57 -0600 Subject: [PATCH 4/6] fix: hardening queue --- .../analytics/MetaRouterAnalyticsClient.kt | 4 +- .../analytics/dispatcher/Dispatcher.kt | 5 + .../analytics/queue/PersistableEventQueue.kt | 214 ++++++++++++------ .../queue/PersistableEventQueueTest.kt | 51 +++++ 4 files changed, 198 insertions(+), 76 deletions(-) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt index 92155d6..ca8c9fa 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt @@ -197,7 +197,7 @@ class MetaRouterAnalyticsClient private constructor( } // If online at launch and overflow exists from previous session, drain directly to network - if (networkMonitor.isConnected) { + if (networkMonitor.isConnected && persistableEventQueue?.hasOverflowData() == true) { scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } } @@ -206,7 +206,7 @@ class MetaRouterAnalyticsClient private constructor( // After each successful flush, drain any overflow disk events persistableEventQueue?.let { pQueue -> dispatcher.onFlushComplete = { - if (networkMonitor.isConnected) { + if (networkMonitor.isConnected && pQueue.hasOverflowData()) { scope.launch { pQueue.drainDiskOverflowToNetwork(dispatcher) } } } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt index 3712d98..783b72f 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/dispatcher/Dispatcher.kt @@ -207,6 +207,11 @@ class Dispatcher( * Used by PersistableEventQueue.drainDiskOverflowToNetwork() to flush overflow * events from disk without loading them into the memory queue. * + * Note: intentionally does not update the circuit breaker or retry counters. + * The overflow drain is an independent pipeline — it stops on first failure and + * retries on the next online transition or flush cycle, not via the dispatcher's + * retry/backoff machinery. + * * @return NetworkResponse on HTTP completion, null on encoding or network error */ suspend fun sendBatchDirect(events: List): NetworkResponse? { diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt index 7a1eb96..ee33650 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt @@ -52,6 +52,9 @@ class PersistableEventQueue( } } + /** Write a checkpoint every N successful drain batches for crash safety. */ + private const val DRAIN_CHECKPOINT_BATCHES = 10 + /** * Reset the rehydration flag so a subsequent initialize() will rehydrate from disk. * Called during SDK reset and in tests. @@ -73,20 +76,41 @@ class PersistableEventQueue( // Guards against concurrent drainDiskOverflowToNetwork invocations private val isDraining = AtomicBoolean(false) + // Lightweight flag: true when overflow data exists on disk. + // Set when events are flushed to overflow disk, cleared when drain empties the store. + // Avoids unnecessary disk reads in onFlushComplete when no overflow exists. + // Initialized by checking if overflow file exists from a previous session. + @Volatile + private var hasOverflowData = try { overflowDiskStore?.read() != null } catch (_: Exception) { false } + + // Lock for overflow disk store access — shared between synchronized queue methods + // (which hold `this` monitor) and the unsynchronized drain coroutine. + // Queue methods call flushToOverflowDiskInternal under `synchronized(this)` which + // also acquires overflowLock. Drain acquires only overflowLock (not `this`), so + // there is no deadlock risk — the lock order is always: this -> overflowLock. + private val overflowLock = Any() + @Synchronized override fun size(): Int = memoryQueue.size /** - * Enqueue an event to memory. No disk I/O unless offline and at capacity. - * Drops oldest events if event count or byte capacity is exceeded. + * Enqueue an event to memory. + * When at capacity (event count or byte size), flushes to overflow disk if available, + * otherwise drops oldest events. */ @Synchronized override fun enqueue(event: EnrichedEventPayload) { val eventSize = estimateEventSize(event) - // Drop oldest while over byte capacity (before adding) - while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { - dropOldest("byte capacity") + // Over byte capacity: flush to overflow disk if available, otherwise drop oldest + if (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { + if (overflowDiskStore != null) { + flushToOverflowDiskInternal() + } else { + while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { + dropOldest("byte capacity") + } + } } // At event count capacity: flush to overflow disk if available, otherwise drop oldest @@ -159,7 +183,9 @@ class PersistableEventQueue( eventSizes.clear() estimatedBytes = 0L diskStore.delete() - overflowDiskStore?.delete() + synchronized(overflowLock) { + deleteOverflowStore() + } if (count > 0) { Logger.log("Cleared $count events from queue and deleted disk snapshots") } @@ -270,6 +296,12 @@ class PersistableEventQueue( @Synchronized fun estimatedSizeBytes(): Long = estimatedBytes + /** + * Check if there is overflow data on disk worth draining. + * Cheap check — no disk I/O, just reads a volatile flag. + */ + fun hasOverflowData(): Boolean = hasOverflowData + // ===== Offline Overflow ===== /** @@ -277,6 +309,11 @@ class PersistableEventQueue( * Does NOT load events into the memory queue — sends via dispatcher's direct path. * Called on offline->online transition as an independent flush pipeline. * + * Reads the entire overflow file once into memory, then drains batches from the + * in-memory list. Writes a checkpoint to disk every [DRAIN_CHECKPOINT_BATCHES] + * successful batches for crash safety. On failure, writes remaining events back + * to disk and returns. + * * @return number of events successfully drained */ suspend fun drainDiskOverflowToNetwork(dispatcher: Dispatcher): Int { @@ -286,82 +323,102 @@ class PersistableEventQueue( var totalDrained = 0 var currentBatchSize = 100 - try { while (true) { - val snapshot = store.read() ?: return totalDrained + try { + // Read entire file once and delete — O(n) instead of O(n²) + val allEvents = synchronized(overflowLock) { + val snapshot = store.read() ?: return 0 + deleteOverflowStore() + snapshot.events + } + + // Filter expired events val now = System.currentTimeMillis() val events = if (eventTTLMs > 0) { - snapshot.events.filter { !isExpired(it, now) } + allEvents.filter { !isExpired(it, now) } } else { - snapshot.events + allEvents } - if (events.isEmpty()) { - store.delete() - return totalDrained - } + if (events.isEmpty()) return 0 - // Take a batch from the front (oldest first) - val batch = events.take(minOf(currentBatchSize, events.size)) - val remaining = events.drop(batch.size) + // Track position via index — avoids O(n) removeAt(0) on each batch + var offset = 0 + var batchesSinceCheckpoint = 0 + fun remaining() = events.subList(offset, events.size) - val response = dispatcher.sendBatchDirect(batch) + while (offset < events.size) { + val batchEnd = minOf(offset + currentBatchSize, events.size) + val batch = events.subList(offset, batchEnd) - if (response == null) { - Logger.warn("Offline overflow drain paused — network error ($totalDrained drained so far)") - return totalDrained - } + val response = dispatcher.sendBatchDirect(batch) - when (ResponseCategory.from(response.statusCode)) { - ResponseCategory.SUCCESS -> { - totalDrained += batch.size - if (currentBatchSize < 100) { - currentBatchSize = minOf(currentBatchSize * 2, 100) - } - if (remaining.isEmpty()) { - store.delete() - Logger.log("Offline overflow disk drain complete ($totalDrained events)") - return totalDrained - } else { - store.write(QueueSnapshot(version = 1, events = remaining)) - } + if (response == null) { + writeOverflowCheckpoint(store, remaining()) + Logger.warn("Offline overflow drain paused — network error ($totalDrained drained so far)") + return totalDrained } - ResponseCategory.PAYLOAD_TOO_LARGE -> { - if (currentBatchSize > 1) { - currentBatchSize = maxOf(1, currentBatchSize / 2) - Logger.warn("Overflow drain: payload too large (413) — reduced batch to $currentBatchSize") - } else { - Logger.warn("Overflow drain: dropping oversized event at batchSize=1 (messageId: ${batch.first().messageId})") - if (remaining.isEmpty()) { - store.delete() + + when (ResponseCategory.from(response.statusCode)) { + ResponseCategory.SUCCESS -> { + offset += batch.size + totalDrained += batch.size + batchesSinceCheckpoint++ + if (currentBatchSize < 100) { + currentBatchSize = minOf(currentBatchSize * 2, 100) + } + // Periodic checkpoint for crash safety + if (batchesSinceCheckpoint >= DRAIN_CHECKPOINT_BATCHES && offset < events.size) { + writeOverflowCheckpoint(store, remaining()) + batchesSinceCheckpoint = 0 + } + } + ResponseCategory.PAYLOAD_TOO_LARGE -> { + if (currentBatchSize > 1) { + currentBatchSize = maxOf(1, currentBatchSize / 2) + Logger.warn("Overflow drain: payload too large (413) — reduced batch to $currentBatchSize") } else { - store.write(QueueSnapshot(version = 1, events = remaining)) + Logger.warn("Overflow drain: dropping oversized event at batchSize=1 (messageId: ${batch.first().messageId})") + offset++ } } - } - ResponseCategory.FATAL_CONFIG -> { - Logger.error("Overflow drain: fatal config error ${response.statusCode} — deleting overflow store") - store.delete() - return totalDrained - } - ResponseCategory.CLIENT_ERROR -> { - Logger.warn("Overflow drain: client error ${response.statusCode} — dropping batch of ${batch.size}") - if (remaining.isEmpty()) { - store.delete() + ResponseCategory.FATAL_CONFIG -> { + Logger.error("Overflow drain: fatal config error ${response.statusCode} — discarding ${events.size - offset} overflow events") + return totalDrained + } + ResponseCategory.CLIENT_ERROR -> { + Logger.warn("Overflow drain: client error ${response.statusCode} — dropping batch of ${batch.size}") + offset += batch.size + } + ResponseCategory.SERVER_ERROR, ResponseCategory.RATE_LIMITED -> { + writeOverflowCheckpoint(store, remaining()) + Logger.warn("Offline overflow drain paused — ${response.statusCode} ($totalDrained drained so far)") return totalDrained - } else { - store.write(QueueSnapshot(version = 1, events = remaining)) } - } - ResponseCategory.SERVER_ERROR, ResponseCategory.RATE_LIMITED -> { - Logger.warn("Offline overflow drain paused — ${response.statusCode} ($totalDrained drained so far)") - return totalDrained } } - } } finally { + + // All events drained successfully + synchronized(overflowLock) { deleteOverflowStore() } + Logger.log("Offline overflow disk drain complete ($totalDrained events)") + return totalDrained + } finally { isDraining.set(false) } } + /** + * Write remaining events back to the overflow disk store as a checkpoint. + * Used during drain for crash safety and on failure to preserve unsent events. + */ + private fun writeOverflowCheckpoint(store: EventDiskStore, events: List) { + if (events.isEmpty()) return + synchronized(overflowLock) { + store.write(QueueSnapshot(version = 1, events = events)) + hasOverflowData = true + } + } + + // ===== Internal ===== /** @@ -384,22 +441,31 @@ class PersistableEventQueue( eventSizes.clear() estimatedBytes = 0L - val existing = try { - store.read()?.events ?: emptyList() - } catch (e: Exception) { - Logger.warn("Failed to read overflow disk store: ${e.message}") - emptyList() - } + synchronized(overflowLock) { + val existing = try { + store.read()?.events ?: emptyList() + } catch (e: Exception) { + Logger.warn("Failed to read overflow disk store: ${e.message}") + emptyList() + } + + var combined = existing + batch + if (combined.size > maxOfflineDiskEvents) { + val dropCount = combined.size - maxOfflineDiskEvents + combined = combined.drop(dropCount) + Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") + } - var combined = existing + batch - if (combined.size > maxOfflineDiskEvents) { - val dropCount = combined.size - maxOfflineDiskEvents - combined = combined.drop(dropCount) - Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") + store.write(QueueSnapshot(version = 1, events = combined)) + hasOverflowData = true + Logger.log("Flushed ${batch.size} events to offline disk (total on disk: ${combined.size})") } + } - store.write(QueueSnapshot(version = 1, events = combined)) - Logger.log("Flushed ${batch.size} events to offline disk (total on disk: ${combined.size})") + /** Delete the overflow disk file and clear the hasOverflowData flag. */ + private fun deleteOverflowStore() { + overflowDiskStore?.delete() + hasOverflowData = false } private fun dropOldest(reason: String) { diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt index f3353cd..e00c405 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt @@ -13,6 +13,7 @@ import com.metarouter.analytics.types.* import io.mockk.every import io.mockk.mockkStatic import io.mockk.unmockkStatic +import kotlinx.coroutines.async import kotlinx.coroutines.test.runTest import org.junit.After import org.junit.Assert.* @@ -1272,6 +1273,56 @@ class PersistableEventQueueTest { unmockkStatic(Log::class) } + @Test + fun `concurrent drain calls do not duplicate sends`() = runTest { + mockkStatic(Log::class) + every { Log.d(any(), any()) } returns 0 + every { Log.w(any(), any()) } returns 0 + every { Log.e(any(), any()) } returns 0 + every { Log.e(any(), any(), any()) } returns 0 + + val overflowDir = createTempDirectory("metarouter-overflow-concurrent").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val overflowQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + val events = (1..10).map { createTestEvent("overflow-$it") } + overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + + val fakeClient = FakeNetworkClient() + fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) + val dispatcher = Dispatcher( + options = InitOptions(writeKey = "test-key", ingestionHost = "https://api.example.com", flushIntervalSeconds = 10), + queue = overflowQueue, + networkClient = fakeClient, + circuitBreaker = CircuitBreaker(), + scope = this, + config = DispatcherConfig() + ) + + // Launch two concurrent drain calls + val drain1 = async { overflowQueue.drainDiskOverflowToNetwork(dispatcher) } + val drain2 = async { overflowQueue.drainDiskOverflowToNetwork(dispatcher) } + + val result1 = drain1.await() + val result2 = drain2.await() + + // One should have drained all 10, the other should have returned 0 (guard) + val totalDrained = result1 + result2 + assertEquals(10, totalDrained) + assertTrue("One drain should return 0", result1 == 0 || result2 == 0) + assertNull(overflowDiskStore.read()) + + overflowDir.deleteRecursively() + unmockkStatic(Log::class) + } + private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload { return EnrichedEventPayload( type = EventType.TRACK, From 923e23636867f54f4fff258d175873c2c75fdd56 Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Thu, 16 Apr 2026 11:43:10 -0600 Subject: [PATCH 5/6] fix: hardening reset logic --- .../analytics/MetaRouterAnalyticsClient.kt | 12 ++++ .../analytics/queue/PersistableEventQueue.kt | 21 ++++++- .../queue/PersistableEventQueueTest.kt | 62 +++++++++++++++++++ 3 files changed, 92 insertions(+), 3 deletions(-) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt index ca8c9fa..9f8f67e 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt @@ -196,6 +196,18 @@ class MetaRouterAnalyticsClient private constructor( dispatcher.pauseForOffline() } + // If online at launch, drain rehydrated memory queue events to network BEFORE + // the proxy replays any buffered user calls. Awaited synchronously (not launched) + // so that events queued in AnalyticsProxy.pendingCalls don't race with the flush + // and push rehydrated events into overflow disk at capacity. + if (networkMonitor.isConnected && eventQueue.size() > 0) { + try { + dispatcher.flush() + } catch (e: Exception) { + Logger.warn("Initial rehydrate flush failed: ${e.message}") + } + } + // If online at launch and overflow exists from previous session, drain directly to network if (networkMonitor.isConnected && persistableEventQueue?.hasOverflowData() == true) { scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt index ee33650..6bd830e 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt @@ -142,14 +142,29 @@ class PersistableEventQueue( } /** - * Requeue events to front. Updates byte size tracking. + * Requeue events to front (for retry after a failed send). + * When at capacity, flushes existing memory queue to overflow disk if available, + * otherwise drops newest events to make room. The requeued events always take + * priority at the front since they were drained first (older) and a send + * already started for them. */ @Synchronized override fun requeueToFront(events: List) { + // If adding these events would overflow capacity and we have an overflow store, + // flush the current memory queue to disk first. The newer events go to disk; + // the requeued (older) events take their place at the front. + if (overflowDiskStore != null && + (memoryQueue.size + events.size > maxCapacity || + estimatedBytes + events.sumOf { estimateEventSize(it) } > maxCapacityBytes) && + memoryQueue.isNotEmpty()) { + flushToOverflowDiskInternal() + } + events.asReversed().forEach { event -> val eventSize = estimateEventSize(event) - // Drop newest while over byte capacity + // Edge case: requeue batch alone exceeds byte capacity. + // Drop newest to make room (no overflow store available, or batch too big). while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { val removedSize = eventSizes.removeLastOrNull() ?: 0L memoryQueue.removeLastOrNull()?.let { @@ -158,7 +173,7 @@ class PersistableEventQueue( } } - // Drop newest if at event count capacity + // Edge case: requeue batch alone exceeds event count capacity. if (memoryQueue.size >= maxCapacity) { val removedSize = eventSizes.removeLastOrNull() ?: 0L memoryQueue.removeLastOrNull()?.let { diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt index e00c405..b03e0ad 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt @@ -1323,6 +1323,68 @@ class PersistableEventQueueTest { unmockkStatic(Log::class) } + @Test + fun `requeueToFront flushes memory queue to overflow disk when at capacity`() { + val overflowDir = createTempDirectory("metarouter-requeue-overflow").toFile() + val overflowDiskStore = EventDiskStore(overflowDir) + + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + maxOfflineDiskEvents = 100, + overflowDiskStore = overflowDiskStore, + eventTTLMs = 0 + ) + + // Simulate: memory queue has 3 new events (enqueued after a drain) + val newEvents = (1..3).map { createTestEvent("new-$it") } + newEvents.forEach { smallQueue.enqueue(it) } + assertEquals(3, smallQueue.size()) + + // Now requeue 5 older events (e.g., a failed send retry). + // Total would be 8, exceeding capacity of 5. + val requeued = (1..5).map { createTestEvent("older-$it") } + smallQueue.requeueToFront(requeued) + + // Memory queue should have the 5 requeued events at the front (no drops) + assertEquals(5, smallQueue.size()) + val drained = smallQueue.drain(10) + assertEquals(5, drained.size) + assertEquals("older-1", drained[0].messageId) + assertEquals("older-5", drained[4].messageId) + + // The 3 new events should be on overflow disk + val snapshot = overflowDiskStore.read() + assertNotNull(snapshot) + assertEquals(3, snapshot!!.events.size) + assertEquals("new-1", snapshot.events[0].messageId) + assertEquals("new-3", snapshot.events[2].messageId) + + overflowDir.deleteRecursively() + } + + @Test + fun `requeueToFront drops newest when no overflow store at capacity`() { + // No overflow store — should still drop (backward compat) + val smallQueue = PersistableEventQueue( + maxCapacity = 5, + diskStore = diskStore, + overflowDiskStore = null, + eventTTLMs = 0 + ) + + val newEvents = (1..3).map { createTestEvent("new-$it") } + newEvents.forEach { smallQueue.enqueue(it) } + + val requeued = (1..5).map { createTestEvent("older-$it") } + smallQueue.requeueToFront(requeued) + + // Memory queue should have the 5 requeued events — new ones were dropped + assertEquals(5, smallQueue.size()) + val drained = smallQueue.drain(10) + assertEquals("older-1", drained[0].messageId) + } + private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload { return EnrichedEventPayload( type = EventType.TRACK, From 934af103a9efe2f3cd9d2fb3567417cf1d4eadda Mon Sep 17 00:00:00 2001 From: Christopher Houdlette Date: Thu, 16 Apr 2026 12:16:57 -0600 Subject: [PATCH 6/6] fix: removing duplicated queue --- .../analytics/MetaRouterAnalyticsClient.kt | 19 +- .../analytics/queue/PersistableEventQueue.kt | 288 ++++------ .../analytics/storage/EventDiskStore.kt | 16 +- .../queue/PersistableEventQueueTest.kt | 528 +++--------------- .../analytics/storage/EventDiskStoreTest.kt | 29 - 5 files changed, 185 insertions(+), 695 deletions(-) diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt index 9f8f67e..c04cfe0 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/MetaRouterAnalyticsClient.kt @@ -150,12 +150,10 @@ class MetaRouterAnalyticsClient private constructor( eventQueue = injectedEventQueue } else { val diskStore = injectedDiskStore ?: EventDiskStore.create(context) - val overflowDiskStore = EventDiskStore.create(context, filename = "offline-overflow.v1.json") val pQueue = PersistableEventQueue( maxCapacity = options.maxQueueEvents, diskStore = diskStore, - maxOfflineDiskEvents = options.maxOfflineDiskEvents, - overflowDiskStore = overflowDiskStore + maxOfflineDiskEvents = options.maxOfflineDiskEvents ) pQueue.rehydrate() eventQueue = pQueue @@ -196,19 +194,8 @@ class MetaRouterAnalyticsClient private constructor( dispatcher.pauseForOffline() } - // If online at launch, drain rehydrated memory queue events to network BEFORE - // the proxy replays any buffered user calls. Awaited synchronously (not launched) - // so that events queued in AnalyticsProxy.pendingCalls don't race with the flush - // and push rehydrated events into overflow disk at capacity. - if (networkMonitor.isConnected && eventQueue.size() > 0) { - try { - dispatcher.flush() - } catch (e: Exception) { - Logger.warn("Initial rehydrate flush failed: ${e.message}") - } - } - - // If online at launch and overflow exists from previous session, drain directly to network + // If online at launch and unsent events exist on disk from a previous session, + // drain them directly to the network. if (networkMonitor.isConnected && persistableEventQueue?.hasOverflowData() == true) { scope.launch { persistableEventQueue?.drainDiskOverflowToNetwork(dispatcher) } } diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt index 6bd830e..a81d09f 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/queue/PersistableEventQueue.kt @@ -14,23 +14,23 @@ import java.util.TimeZone import java.util.concurrent.atomic.AtomicBoolean /** - * Event queue with disk persistence as a safety net. - * Uses composition: wraps an internal memory queue and adds byte tracking + disk I/O. + * Event queue with a single disk store for persistence and offline overflow. + * Memory queue is the hot buffer; disk is the overflow/crash-safety backing. * - * Write-behind model: - * - enqueue() and drain() are memory-only (no disk I/O) - * - flushToDisk() writes a full snapshot of current memory state - * - rehydrate() loads events from disk once per process lifetime + * Model: + * - enqueue() and drain() are memory-only (no disk I/O) in the common case. + * - When memory queue hits capacity (event count OR byte size), the entire queue + * flushes to disk and memory resets to empty. Same triggers swap destination + * when the device is offline (see [flushToOfflineStorage]). + * - flushToDisk() appends current memory queue to disk (crash-safety snapshot). + * - drainDiskOverflowToNetwork() sends disk events directly to the network via + * Dispatcher.sendBatchDirect, bypassing the memory queue. * - * Capacity model: - * - Single shared cap: [maxCapacity] events OR [maxCapacityBytes] total, whichever is hit first - * - Flush threshold: [flushThresholdEvents] events OR [flushThresholdBytes], whichever is hit first - * - * Offline overflow model: - * - When offline, flush triggers write the entire memory queue to the overflow disk store - * instead of sending to the network. Same triggers, swapped destination. - * - When the queue hits capacity (online or offline), it flushes to overflow disk rather than dropping. - * - On online transition, overflow events drain from disk directly to network. + * Concurrency: + * - All memory-queue operations are guarded by `synchronized(this)` via @Synchronized. + * - Disk operations happen under a separate [diskLock] so the suspend drain can + * release it during network calls without blocking enqueue/drain of the memory queue. + * - Lock order is always `this` -> `diskLock` (never the reverse). */ class PersistableEventQueue( private val maxCapacity: Int = 2000, @@ -39,8 +39,7 @@ class PersistableEventQueue( private val flushThresholdBytes: Long = 2L * 1024 * 1024, private val maxCapacityBytes: Long = 5L * 1024 * 1024, private val eventTTLMs: Long = 7L * 24 * 60 * 60 * 1000, - private val maxOfflineDiskEvents: Int = 10000, - private val overflowDiskStore: EventDiskStore? = null + private val maxOfflineDiskEvents: Int = 10000 ) : EventQueueInterface { companion object { @@ -76,50 +75,36 @@ class PersistableEventQueue( // Guards against concurrent drainDiskOverflowToNetwork invocations private val isDraining = AtomicBoolean(false) - // Lightweight flag: true when overflow data exists on disk. - // Set when events are flushed to overflow disk, cleared when drain empties the store. - // Avoids unnecessary disk reads in onFlushComplete when no overflow exists. - // Initialized by checking if overflow file exists from a previous session. + // True when unsent events exist on disk. Cheap volatile flag used to gate drain attempts. + // Initialized from file existence (not a full file parse). @Volatile - private var hasOverflowData = try { overflowDiskStore?.read() != null } catch (_: Exception) { false } + private var hasOverflowData = try { diskStore.exists() } catch (_: Exception) { false } - // Lock for overflow disk store access — shared between synchronized queue methods - // (which hold `this` monitor) and the unsynchronized drain coroutine. - // Queue methods call flushToOverflowDiskInternal under `synchronized(this)` which - // also acquires overflowLock. Drain acquires only overflowLock (not `this`), so - // there is no deadlock risk — the lock order is always: this -> overflowLock. - private val overflowLock = Any() + // Lock for disk store access. The suspend drain releases this lock during network calls, + // so it can't hold the class monitor. Memory queue methods acquire `this` first and + // `diskLock` second (when they touch disk). Drain acquires only `diskLock`. No deadlock. + private val diskLock = Any() @Synchronized override fun size(): Int = memoryQueue.size /** * Enqueue an event to memory. - * When at capacity (event count or byte size), flushes to overflow disk if available, - * otherwise drops oldest events. + * When at capacity (event count or byte size), flushes the entire memory queue + * to disk. Events are only dropped if capacity is hit and disk operations fail. */ @Synchronized override fun enqueue(event: EnrichedEventPayload) { val eventSize = estimateEventSize(event) - // Over byte capacity: flush to overflow disk if available, otherwise drop oldest + // Over byte capacity: flush memory queue to disk if (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { - if (overflowDiskStore != null) { - flushToOverflowDiskInternal() - } else { - while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { - dropOldest("byte capacity") - } - } + flushMemoryToDiskInternal() } - // At event count capacity: flush to overflow disk if available, otherwise drop oldest + // At event count capacity: flush memory queue to disk if (memoryQueue.size >= maxCapacity) { - if (overflowDiskStore != null) { - flushToOverflowDiskInternal() - } else { - dropOldest("event count capacity") - } + flushMemoryToDiskInternal() } memoryQueue.addLast(event) @@ -143,43 +128,30 @@ class PersistableEventQueue( /** * Requeue events to front (for retry after a failed send). - * When at capacity, flushes existing memory queue to overflow disk if available, - * otherwise drops newest events to make room. The requeued events always take - * priority at the front since they were drained first (older) and a send - * already started for them. + * When at capacity, flushes existing memory queue to disk to make room. The requeued + * (older) events take priority at the front since they were drained first and a send + * already started for them. If the requeue batch itself exceeds capacity mid-loop, + * we flush again so no requeued events get dropped. */ @Synchronized override fun requeueToFront(events: List) { - // If adding these events would overflow capacity and we have an overflow store, - // flush the current memory queue to disk first. The newer events go to disk; - // the requeued (older) events take their place at the front. - if (overflowDiskStore != null && - (memoryQueue.size + events.size > maxCapacity || + // Initial flush: if adding these events would exceed capacity, flush current memory first. + if ((memoryQueue.size + events.size > maxCapacity || estimatedBytes + events.sumOf { estimateEventSize(it) } > maxCapacityBytes) && memoryQueue.isNotEmpty()) { - flushToOverflowDiskInternal() + flushMemoryToDiskInternal() } events.asReversed().forEach { event -> val eventSize = estimateEventSize(event) - // Edge case: requeue batch alone exceeds byte capacity. - // Drop newest to make room (no overflow store available, or batch too big). - while (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { - val removedSize = eventSizes.removeLastOrNull() ?: 0L - memoryQueue.removeLastOrNull()?.let { - estimatedBytes -= removedSize - Logger.warn("Queue byte capacity reached during requeue - dropped newest event (messageId: ${it.messageId})") - } + // Mid-loop capacity: the requeue batch itself exceeds capacity after earlier iterations + // filled the queue. Flush to disk instead of dropping requeued events. + if (estimatedBytes + eventSize > maxCapacityBytes && memoryQueue.isNotEmpty()) { + flushMemoryToDiskInternal() } - - // Edge case: requeue batch alone exceeds event count capacity. if (memoryQueue.size >= maxCapacity) { - val removedSize = eventSizes.removeLastOrNull() ?: 0L - memoryQueue.removeLastOrNull()?.let { - estimatedBytes -= removedSize - Logger.warn("Queue event capacity reached during requeue - dropped newest event (messageId: ${it.messageId})") - } + flushMemoryToDiskInternal() } memoryQueue.addFirst(event) @@ -189,7 +161,7 @@ class PersistableEventQueue( } /** - * Clear all events from memory and delete disk snapshots. + * Clear all events from memory and delete the disk file. */ @Synchronized override fun clear() { @@ -197,46 +169,40 @@ class PersistableEventQueue( memoryQueue.clear() eventSizes.clear() estimatedBytes = 0L - diskStore.delete() - synchronized(overflowLock) { - deleteOverflowStore() + synchronized(diskLock) { + deleteDiskStore() } if (count > 0) { - Logger.log("Cleared $count events from queue and deleted disk snapshots") + Logger.log("Cleared $count events from queue and deleted disk snapshot") } } /** - * Flush all events to offline overflow disk storage. - * Called by the dispatcher when a flush is triggered but the device is offline. - * Drains the entire memory queue and appends to the overflow disk store. + * Flush memory queue to disk when the dispatcher triggers a flush while offline. + * Returns true if events were flushed to disk. */ @Synchronized override fun flushToOfflineStorage(): Boolean { - if (overflowDiskStore == null || memoryQueue.isEmpty()) return false - flushToOverflowDiskInternal() + if (memoryQueue.isEmpty()) return false + flushMemoryToDiskInternal() return true } /** - * Flush current memory state to disk as a full overwrite snapshot. - * Skips the write and deletes any existing snapshot if the queue is empty. + * Best-effort crash-safety flush: appends current memory queue to disk. + * Called on app background / onTrimMemory. */ @Synchronized fun flushToDisk() { - if (memoryQueue.isEmpty()) { - diskStore.delete() - return - } - val events = memoryQueue.toList() - val snapshot = QueueSnapshot(version = 1, events = events) - diskStore.write(snapshot) + if (memoryQueue.isEmpty()) return + flushMemoryToDiskInternal() } /** - * Rehydrate events from disk into memory. Only executes once per process. - * Loaded events count toward capacity. Excess events are trimmed (keep newest). - * Deletes the disk file after successful load. + * Called once per process on startup. With the consolidated disk store, events + * persisted across sessions stay on disk and drain to the network via + * [drainDiskOverflowToNetwork]. Memory queue starts empty — no loading needed. + * [hasOverflowData] reflects whether disk has unsent events (set in constructor). */ @Synchronized fun rehydrate() { @@ -244,57 +210,9 @@ class PersistableEventQueue( Logger.log("Rehydration already completed this process - skipping") return } - - val snapshot = diskStore.read() ?: return - - var events = snapshot.events - if (events.isEmpty()) { - diskStore.delete() - return - } - - // Drop events older than TTL (keep newest) - if (eventTTLMs > 0) { - val now = System.currentTimeMillis() - val beforeTTL = events.size - events = events.filter { !isExpired(it, now) } - val dropped = beforeTTL - events.size - if (dropped > 0) { - Logger.warn("Rehydration: dropped $dropped event(s) older than ${eventTTLMs / (24 * 60 * 60 * 1000)}d TTL") - } - } - - // Trim to event count capacity (keep newest) - if (events.size > maxCapacity) { - Logger.warn("Disk snapshot has ${events.size} events, trimming to $maxCapacity (keeping newest)") - events = events.drop(events.size - maxCapacity) - } - - // Trim by byte capacity (keep newest) - var totalBytes = 0L - val sizedEvents = events.map { it to estimateEventSize(it) } - val fittingEvents = mutableListOf>() - - for ((event, size) in sizedEvents.asReversed()) { - if (totalBytes + size > maxCapacityBytes) break - totalBytes += size - fittingEvents.add(event to size) - } - fittingEvents.reverse() - - if (fittingEvents.size < events.size) { - Logger.warn("Disk snapshot exceeds byte capacity, trimmed from ${events.size} to ${fittingEvents.size} events") - } - - // Load into memory - for ((event, size) in fittingEvents) { - memoryQueue.addLast(event) - eventSizes.addLast(size) - estimatedBytes += size + if (hasOverflowData) { + Logger.log("Unsent events present on disk — will drain on next online transition") } - - Logger.log("Rehydrated ${fittingEvents.size} events from disk") - diskStore.delete() } /** @@ -312,27 +230,27 @@ class PersistableEventQueue( fun estimatedSizeBytes(): Long = estimatedBytes /** - * Check if there is overflow data on disk worth draining. - * Cheap check — no disk I/O, just reads a volatile flag. + * Check if there is unsent data on disk worth draining. + * Cheap check — reads a volatile flag, no disk I/O. */ fun hasOverflowData(): Boolean = hasOverflowData // ===== Offline Overflow ===== /** - * Drain overflow events directly from disk to network in batches. - * Does NOT load events into the memory queue — sends via dispatcher's direct path. - * Called on offline->online transition as an independent flush pipeline. + * Drain disk events directly to the network in batches via [Dispatcher.sendBatchDirect]. + * Does NOT load events into the memory queue. Called on offline->online transitions + * and after successful online flushes via onFlushComplete. * - * Reads the entire overflow file once into memory, then drains batches from the - * in-memory list. Writes a checkpoint to disk every [DRAIN_CHECKPOINT_BATCHES] - * successful batches for crash safety. On failure, writes remaining events back - * to disk and returns. + * Reads the entire disk file once, deletes it, then iterates batches in-memory. + * Writes a checkpoint every [DRAIN_CHECKPOINT_BATCHES] successful batches for crash safety. + * On failure, writes remaining events back to disk. * * @return number of events successfully drained */ suspend fun drainDiskOverflowToNetwork(dispatcher: Dispatcher): Int { - val store = overflowDiskStore ?: return 0 + // Bail early if offline — no point burning network call timeouts. + if (dispatcher.isNetworkPaused()) return 0 if (!isDraining.compareAndSet(false, true)) return 0 var totalDrained = 0 @@ -340,9 +258,9 @@ class PersistableEventQueue( try { // Read entire file once and delete — O(n) instead of O(n²) - val allEvents = synchronized(overflowLock) { - val snapshot = store.read() ?: return 0 - deleteOverflowStore() + val allEvents = synchronized(diskLock) { + val snapshot = diskStore.read() ?: return 0 + deleteDiskStore() snapshot.events } @@ -368,7 +286,7 @@ class PersistableEventQueue( val response = dispatcher.sendBatchDirect(batch) if (response == null) { - writeOverflowCheckpoint(store, remaining()) + writeCheckpoint(remaining()) Logger.warn("Offline overflow drain paused — network error ($totalDrained drained so far)") return totalDrained } @@ -383,7 +301,7 @@ class PersistableEventQueue( } // Periodic checkpoint for crash safety if (batchesSinceCheckpoint >= DRAIN_CHECKPOINT_BATCHES && offset < events.size) { - writeOverflowCheckpoint(store, remaining()) + writeCheckpoint(remaining()) batchesSinceCheckpoint = 0 } } @@ -398,6 +316,9 @@ class PersistableEventQueue( } ResponseCategory.FATAL_CONFIG -> { Logger.error("Overflow drain: fatal config error ${response.statusCode} — discarding ${events.size - offset} overflow events") + // Notify the dispatcher so the main pipeline also shuts down. Matches + // the main flush path's handling of auth/config failures. + dispatcher.onFatalConfigError?.invoke(response.statusCode) return totalDrained } ResponseCategory.CLIENT_ERROR -> { @@ -405,7 +326,7 @@ class PersistableEventQueue( offset += batch.size } ResponseCategory.SERVER_ERROR, ResponseCategory.RATE_LIMITED -> { - writeOverflowCheckpoint(store, remaining()) + writeCheckpoint(remaining()) Logger.warn("Offline overflow drain paused — ${response.statusCode} ($totalDrained drained so far)") return totalDrained } @@ -413,7 +334,7 @@ class PersistableEventQueue( } // All events drained successfully - synchronized(overflowLock) { deleteOverflowStore() } + synchronized(diskLock) { deleteDiskStore() } Logger.log("Offline overflow disk drain complete ($totalDrained events)") return totalDrained } finally { @@ -422,13 +343,13 @@ class PersistableEventQueue( } /** - * Write remaining events back to the overflow disk store as a checkpoint. + * Write remaining events back to the disk store as a checkpoint. * Used during drain for crash safety and on failure to preserve unsent events. */ - private fun writeOverflowCheckpoint(store: EventDiskStore, events: List) { + private fun writeCheckpoint(events: List) { if (events.isEmpty()) return - synchronized(overflowLock) { - store.write(QueueSnapshot(version = 1, events = events)) + synchronized(diskLock) { + diskStore.write(QueueSnapshot(version = 1, events = events)) hasOverflowData = true } } @@ -437,18 +358,15 @@ class PersistableEventQueue( // ===== Internal ===== /** - * Flush all events from memory queue to the overflow disk store. - * Appends to existing overflow events on disk. Enforces maxOfflineDiskEvents cap. - * Resets the memory queue to empty. - * Must be called within a synchronized block. + * Flush all events from memory queue to disk. Appends to any existing disk data, + * enforces [maxOfflineDiskEvents] cap, and resets the memory queue. + * Must be called while holding `synchronized(this)`. * - * Note: performs disk I/O (read + write) while holding the synchronized lock. - * This blocks enqueue/drain during the write. Acceptable because local disk I/O - * is fast (sub-ms for typical event sizes), but worth revisiting if profiling - * shows contention under extreme throughput. + * Note: performs disk I/O (read + write) while holding both `this` and `diskLock`. + * Local disk I/O is sub-millisecond for typical event sizes, so this is acceptable + * even on the enqueue hot path. Revisit if profiling shows contention. */ - private fun flushToOverflowDiskInternal() { - val store = overflowDiskStore ?: return + private fun flushMemoryToDiskInternal() { if (memoryQueue.isEmpty()) return val batch = memoryQueue.toList() @@ -456,11 +374,11 @@ class PersistableEventQueue( eventSizes.clear() estimatedBytes = 0L - synchronized(overflowLock) { + synchronized(diskLock) { val existing = try { - store.read()?.events ?: emptyList() + diskStore.read()?.events ?: emptyList() } catch (e: Exception) { - Logger.warn("Failed to read overflow disk store: ${e.message}") + Logger.warn("Failed to read disk store during flush: ${e.message}") emptyList() } @@ -468,29 +386,21 @@ class PersistableEventQueue( if (combined.size > maxOfflineDiskEvents) { val dropCount = combined.size - maxOfflineDiskEvents combined = combined.drop(dropCount) - Logger.warn("Offline overflow disk cap reached — dropped $dropCount oldest events") + Logger.warn("Offline disk cap reached — dropped $dropCount oldest events") } - store.write(QueueSnapshot(version = 1, events = combined)) + diskStore.write(QueueSnapshot(version = 1, events = combined)) hasOverflowData = true - Logger.log("Flushed ${batch.size} events to offline disk (total on disk: ${combined.size})") + Logger.log("Flushed ${batch.size} events to disk (total on disk: ${combined.size})") } } - /** Delete the overflow disk file and clear the hasOverflowData flag. */ - private fun deleteOverflowStore() { - overflowDiskStore?.delete() + /** Delete the disk file and clear the hasOverflowData flag. */ + private fun deleteDiskStore() { + diskStore.delete() hasOverflowData = false } - private fun dropOldest(reason: String) { - memoryQueue.removeFirstOrNull()?.let { dropped -> - val size = eventSizes.removeFirstOrNull() ?: 0L - estimatedBytes -= size - Logger.warn("Queue $reason reached - dropped oldest event (messageId: ${dropped.messageId})") - } - } - private fun estimateEventSize(event: EnrichedEventPayload): Long { return try { json.encodeToString(event).length.toLong() diff --git a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt index 805616d..91cdaf4 100644 --- a/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt +++ b/metarouter-sdk/src/main/java/com/metarouter/analytics/storage/EventDiskStore.kt @@ -24,7 +24,7 @@ import java.io.File * * Thread safety: callers must synchronize externally (PersistableEventQueue does this). */ -class EventDiskStore(private val baseDir: File, private val filename: String = SNAPSHOT_FILENAME) { +class EventDiskStore(private val baseDir: File) { companion object { private const val DIRECTORY_NAME = "metarouter" @@ -35,11 +35,8 @@ class EventDiskStore(private val baseDir: File, private val filename: String = S /** * Production factory: uses Context.noBackupFilesDir as base directory. * Falls back to Context.filesDir, then system temp directory. - * - * @param filename Override the default snapshot filename. Used for separate - * overflow storage (e.g., "offline-overflow.v1.json"). */ - fun create(context: Context, filename: String = SNAPSHOT_FILENAME): EventDiskStore { + fun create(context: Context): EventDiskStore { val baseDir = listOf( { context.noBackupFilesDir }, { context.filesDir } @@ -54,7 +51,7 @@ class EventDiskStore(private val baseDir: File, private val filename: String = S null } } ?: File(System.getProperty("java.io.tmpdir", "/tmp"), "metarouter-fallback") - return EventDiskStore(baseDir, filename) + return EventDiskStore(baseDir) } } @@ -67,10 +64,13 @@ class EventDiskStore(private val baseDir: File, private val filename: String = S get() = File(baseDir, "$DIRECTORY_NAME/$SUBDIRECTORY_NAME") private val snapshotFile: File - get() = File(snapshotDir, filename) + get() = File(snapshotDir, SNAPSHOT_FILENAME) private val tmpFile: File - get() = File(snapshotDir, "$filename.tmp") + get() = File(snapshotDir, "$SNAPSHOT_FILENAME.tmp") + + /** Cheap check — does a snapshot file exist on disk? */ + fun exists(): Boolean = snapshotFile.exists() /** * Write a snapshot to disk. Uses atomic write-to-tmp-then-rename. diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt index b03e0ad..ff3ba14 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/queue/PersistableEventQueueTest.kt @@ -69,7 +69,7 @@ class PersistableEventQueueTest { } @Test - fun `enqueue drops oldest when event count capacity exceeded`() { + fun `enqueue flushes to disk when event count capacity exceeded`() { val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, @@ -82,13 +82,15 @@ class PersistableEventQueueTest { smallQueue.enqueue(createTestEvent("msg-$i")) } - assertEquals(5, smallQueue.size()) - val drained = smallQueue.drain(5) - assertEquals("msg-1", drained[0].messageId) + // Memory queue has just the latest event; others went to disk + assertEquals(1, smallQueue.size()) + val snapshot = diskStore.read() + assertNotNull(snapshot) + assertEquals(5, snapshot!!.events.size) } @Test - fun `enqueue drops oldest when byte capacity exceeded`() { + fun `enqueue flushes to disk when byte capacity exceeded`() { val tinyQueue = PersistableEventQueue( maxCapacity = 2000, diskStore = diskStore, @@ -101,8 +103,9 @@ class PersistableEventQueueTest { tinyQueue.enqueue(createTestEvent("msg-$i")) } - assertTrue(tinyQueue.size() < 10) + // Events that exceeded byte cap got flushed to disk assertTrue(tinyQueue.estimatedSizeBytes() <= 500) + assertTrue(tinyQueue.hasOverflowData()) } // ===== Drain (memory-only) ===== @@ -134,13 +137,14 @@ class PersistableEventQueueTest { // ===== Flush to Disk ===== @Test - fun `flushToDisk writes current memory state to disk`() { + fun `flushToDisk writes memory state to disk and clears memory`() { repeat(3) { i -> queue.enqueue(createTestEvent("msg-$i")) } queue.flushToDisk() + assertEquals(0, queue.size()) val snapshot = diskStore.read() assertNotNull(snapshot) assertEquals(3, snapshot!!.events.size) @@ -148,35 +152,37 @@ class PersistableEventQueueTest { } @Test - fun `flushToDisk fully overwrites previous snapshot`() { + fun `flushToDisk appends memory to existing disk contents and clears memory`() { queue.enqueue(createTestEvent("msg-1")) queue.flushToDisk() + // Memory queue cleared, msg-1 is on disk + assertEquals(0, queue.size()) - queue.drain(1) queue.enqueue(createTestEvent("msg-2")) queue.enqueue(createTestEvent("msg-3")) queue.flushToDisk() val snapshot = diskStore.read() assertNotNull(snapshot) - assertEquals(2, snapshot!!.events.size) - assertEquals("msg-2", snapshot.events[0].messageId) - assertEquals("msg-3", snapshot.events[1].messageId) + assertEquals(3, snapshot!!.events.size) + assertEquals("msg-1", snapshot.events[0].messageId) + assertEquals("msg-2", snapshot.events[1].messageId) + assertEquals("msg-3", snapshot.events[2].messageId) } @Test - fun `flushToDisk with empty queue skips write and deletes existing snapshot`() { + fun `flushToDisk with empty queue is a no-op`() { // Write a snapshot with events first queue.enqueue(createTestEvent("msg-1")) queue.flushToDisk() - assertNotNull(diskStore.read()) // snapshot exists + assertNotNull(diskStore.read()) - // Drain all events, then flush empty queue - queue.drain(1) + // Memory is now empty; flushing again shouldn't delete or modify disk queue.flushToDisk() - // Existing snapshot should be deleted, not overwritten with empty - assertNull(diskStore.read()) + val snapshot = diskStore.read() + assertNotNull(snapshot) + assertEquals(1, snapshot!!.events.size) } @Test @@ -223,172 +229,41 @@ class PersistableEventQueueTest { } // ===== Rehydration ===== + // With consolidated disk store, events stay on disk and drain to network. + // Memory queue starts empty on rehydrate. TTL filtering happens in drain, not rehydrate. @Test - fun `rehydrate loads events from disk into memory`() { + fun `rehydrate sets hasOverflowData when disk file exists`() { val events = listOf(createTestEvent("disk-1"), createTestEvent("disk-2")) - val snapshot = QueueSnapshot(version = 1, events = events) - diskStore.write(snapshot) - - queue.rehydrate() - - assertEquals(2, queue.size()) - val drained = queue.drain(10) - assertEquals("disk-1", drained[0].messageId) - assertEquals("disk-2", drained[1].messageId) - } - - @Test - fun `rehydrate only happens once per process`() { - val events = listOf(createTestEvent("disk-1")) diskStore.write(QueueSnapshot(version = 1, events = events)) - queue.rehydrate() - assertEquals(1, queue.size()) - - queue.drain(1) - val queue2 = PersistableEventQueue( + val freshQueue = PersistableEventQueue( maxCapacity = 2000, diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024 + eventTTLMs = 0 ) + freshQueue.rehydrate() - diskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("disk-2")))) - - queue2.rehydrate() - assertEquals(0, queue2.size()) + // Memory queue starts empty; disk file preserved for drain + assertEquals(0, freshQueue.size()) + assertTrue(freshQueue.hasOverflowData()) + assertNotNull(diskStore.read()) } @Test - fun `rehydrate with no disk file does nothing`() { + fun `rehydrate with no disk file sets hasOverflowData false`() { queue.rehydrate() assertEquals(0, queue.size()) + assertFalse(queue.hasOverflowData()) } @Test - fun `rehydrate drops oldest events if disk exceeds capacity`() { - val smallCapQueue = PersistableEventQueue( - maxCapacity = 3, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 0 - ) - - val events = (1..5).map { createTestEvent("msg-$it") } - diskStore.write(QueueSnapshot(version = 1, events = events)) - - smallCapQueue.rehydrate() - - assertEquals(3, smallCapQueue.size()) - val drained = smallCapQueue.drain(10) - assertEquals("msg-3", drained[0].messageId) - assertEquals("msg-4", drained[1].messageId) - assertEquals("msg-5", drained[2].messageId) - } - - @Test - fun `rehydrate deletes disk file after loading`() { - val events = listOf(createTestEvent("disk-1")) - diskStore.write(QueueSnapshot(version = 1, events = events)) - + fun `rehydrate only sets rehydrated flag once per process`() { + PersistableEventQueue.resetRehydrationFlag() queue.rehydrate() - - assertNull(diskStore.read()) - } - - // ===== Event TTL ===== - - @Test - fun `rehydrate drops events older than TTL`() { - val ttlQueue = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 7L * 24 * 60 * 60 * 1000 // 7 days - ) - - val now = System.currentTimeMillis() - val eightDaysAgo = formatTimestamp(now - 8L * 24 * 60 * 60 * 1000) - val twoDaysAgo = formatTimestamp(now - 2L * 24 * 60 * 60 * 1000) - val recent = formatTimestamp(now - 60_000) // 1 minute ago - - val events = listOf( - createTestEvent("old-1", timestamp = eightDaysAgo), - createTestEvent("mid-1", timestamp = twoDaysAgo), - createTestEvent("new-1", timestamp = recent) - ) - diskStore.write(QueueSnapshot(version = 1, events = events)) - - ttlQueue.rehydrate() - - assertEquals(2, ttlQueue.size()) - val drained = ttlQueue.drain(10) - assertEquals("mid-1", drained[0].messageId) - assertEquals("new-1", drained[1].messageId) - } - - @Test - fun `rehydrate TTL filter runs before capacity trim`() { - val ttlQueue = PersistableEventQueue( - maxCapacity = 2, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 7L * 24 * 60 * 60 * 1000 - ) - - val now = System.currentTimeMillis() - val eightDaysAgo = formatTimestamp(now - 8L * 24 * 60 * 60 * 1000) - val twoDaysAgo = formatTimestamp(now - 2L * 24 * 60 * 60 * 1000) - val oneDayAgo = formatTimestamp(now - 1L * 24 * 60 * 60 * 1000) - val recent = formatTimestamp(now - 60_000) - - // 4 events: 2 expired, 2 valid. maxCapacity = 2. - // TTL filter removes the 2 expired. 2 valid remain. Capacity is satisfied. - val events = listOf( - createTestEvent("expired-1", timestamp = eightDaysAgo), - createTestEvent("expired-2", timestamp = eightDaysAgo), - createTestEvent("valid-1", timestamp = twoDaysAgo), - createTestEvent("valid-2", timestamp = oneDayAgo) - ) - diskStore.write(QueueSnapshot(version = 1, events = events)) - - ttlQueue.rehydrate() - - assertEquals(2, ttlQueue.size()) - val drained = ttlQueue.drain(10) - assertEquals("valid-1", drained[0].messageId) - assertEquals("valid-2", drained[1].messageId) - } - - @Test - fun `rehydrate with unparseable timestamp keeps event`() { - val ttlQueue = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 7L * 24 * 60 * 60 * 1000 - ) - - val events = listOf( - createTestEvent("bad-ts", timestamp = "not-a-timestamp"), - createTestEvent("good-ts", timestamp = formatTimestamp(System.currentTimeMillis() - 60_000)) - ) - diskStore.write(QueueSnapshot(version = 1, events = events)) - - ttlQueue.rehydrate() - - // Unparseable timestamp should be kept (fail-open), not dropped - assertEquals(2, ttlQueue.size()) + // Second call is a no-op (guarded by hasRehydrated flag) + queue.rehydrate() + // No assertions — just confirming it doesn't throw } // ===== Clear ===== @@ -462,37 +337,7 @@ class PersistableEventQueueTest { // ===== End-to-end Scenarios ===== @Test - fun `full lifecycle - enqueue, background flush, process restart, rehydrate`() { - // Simulate first session - repeat(10) { i -> - queue.enqueue(createTestEvent("session1-msg-$i")) - } - assertEquals(10, queue.size()) - - // App backgrounds -> flush to disk - queue.flushToDisk() - - // Simulate process restart: create new queue with same disk store - PersistableEventQueue.resetRehydrationFlag() - val queue2 = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 0 - ) - - queue2.rehydrate() - - assertEquals(10, queue2.size()) - val drained = queue2.drain(10) - assertEquals("session1-msg-0", drained[0].messageId) - assertEquals("session1-msg-9", drained[9].messageId) - } - - @Test - fun `events drained before background are not in snapshot`() { + fun `events drained before background are not flushed to disk`() { repeat(5) { i -> queue.enqueue(createTestEvent("msg-$i")) } @@ -521,86 +366,15 @@ class PersistableEventQueueTest { assertFalse(queue.shouldFlushToDisk()) } - @Test - fun `multiple flush-rehydrate cycles preserve data correctly`() { - // Cycle 1: enqueue and flush - queue.enqueue(createTestEvent("cycle1-a")) - queue.enqueue(createTestEvent("cycle1-b")) - queue.flushToDisk() - - // Cycle 2: restart, rehydrate, add more, flush again - PersistableEventQueue.resetRehydrationFlag() - val queue2 = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 0 - ) - queue2.rehydrate() - queue2.enqueue(createTestEvent("cycle2-c")) - queue2.flushToDisk() - - // Cycle 3: restart, rehydrate - PersistableEventQueue.resetRehydrationFlag() - val queue3 = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 0 - ) - queue3.rehydrate() - - assertEquals(3, queue3.size()) - val drained = queue3.drain(10) - assertEquals("cycle1-a", drained[0].messageId) - assertEquals("cycle1-b", drained[1].messageId) - assertEquals("cycle2-c", drained[2].messageId) - } - - @Test - fun `flush overwrites previous state - no duplicates`() { - queue.enqueue(createTestEvent("batch1-a")) - queue.enqueue(createTestEvent("batch1-b")) - queue.flushToDisk() - - // Drain all, add new events - queue.drain(10) - queue.enqueue(createTestEvent("batch2-a")) - queue.flushToDisk() - - PersistableEventQueue.resetRehydrationFlag() - val queue2 = PersistableEventQueue( - maxCapacity = 2000, - diskStore = diskStore, - flushThresholdEvents = 500, - flushThresholdBytes = 2 * 1024 * 1024, - maxCapacityBytes = 5 * 1024 * 1024, - eventTTLMs = 0 - ) - queue2.rehydrate() - - // Should only have batch2 events — full overwrite, no duplicates - assertEquals(1, queue2.size()) - val drained = queue2.drain(10) - assertEquals("batch2-a", drained[0].messageId) - } - // ===== Offline Overflow ===== @Test fun `capacity overflow flushes entire queue to disk`() { - val overflowDir = createTempDirectory("metarouter-overflow-test").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) @@ -615,7 +389,7 @@ class PersistableEventQueueTest { assertEquals(1, smallQueue.size()) // All 5 original events should be on overflow disk - val snapshot = overflowDiskStore.read() + val snapshot = diskStore.read() assertNotNull(snapshot) assertEquals(5, snapshot!!.events.size) assertEquals("fill-0", snapshot.events[0].messageId) @@ -626,19 +400,15 @@ class PersistableEventQueueTest { assertEquals(1, drained.size) assertEquals("overflow-0", drained[0].messageId) - overflowDir.deleteRecursively() } @Test fun `flushToOfflineStorage drains entire queue to disk`() { - val overflowDir = createTempDirectory("metarouter-overflow-flush").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) @@ -654,54 +424,46 @@ class PersistableEventQueueTest { assertEquals(0, smallQueue.size()) // Events should be on overflow disk - val snapshot = overflowDiskStore.read() + val snapshot = diskStore.read() assertNotNull(snapshot) assertEquals(3, snapshot!!.events.size) - overflowDir.deleteRecursively() } @Test fun `flushToOfflineStorage appends to existing overflow disk data`() { - val overflowDir = createTempDirectory("metarouter-overflow-append").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // Pre-populate overflow disk with 3 events val existing = (1..3).map { createTestEvent("existing-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = existing)) + diskStore.write(QueueSnapshot(version = 1, events = existing)) // Add 2 events to memory and flush repeat(2) { i -> smallQueue.enqueue(createTestEvent("new-$i")) } smallQueue.flushToOfflineStorage() // Disk should have all 5 - val snapshot = overflowDiskStore.read() + val snapshot = diskStore.read() assertNotNull(snapshot) assertEquals(5, snapshot!!.events.size) assertEquals("existing-1", snapshot.events[0].messageId) assertEquals("new-1", snapshot.events[4].messageId) - overflowDir.deleteRecursively() } @Test fun `offline disk overflow respects maxOfflineDiskEvents cap`() { - val overflowDir = createTempDirectory("metarouter-overflow-cap").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 8, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) @@ -714,11 +476,10 @@ class PersistableEventQueueTest { repeat(5) { i -> smallQueue.enqueue(createTestEvent("batch2-$i")) } smallQueue.flushToOfflineStorage() - val snapshot = overflowDiskStore.read() + val snapshot = diskStore.read() assertNotNull(snapshot) assertTrue("Should cap at maxOfflineDiskEvents", snapshot!!.events.size <= 8) - overflowDir.deleteRecursively() } @Test @@ -729,20 +490,17 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-drain").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // Pre-populate overflow disk val events = (1..5).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) // Set up dispatcher with fake network val fakeClient = FakeNetworkClient() @@ -765,10 +523,9 @@ class PersistableEventQueueTest { assertEquals(5, drained) assertEquals(1, fakeClient.requests.size) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) assertEquals(0, overflowQueue.size()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -780,20 +537,17 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-fail").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // Pre-populate overflow disk val events = (1..5).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) // Set up dispatcher that fails val fakeClient = FakeNetworkClient() @@ -815,37 +569,32 @@ class PersistableEventQueueTest { val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(0, drained) - val remaining = overflowDiskStore.read() + val remaining = diskStore.read() assertNotNull(remaining) assertEquals(5, remaining!!.events.size) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @Test fun `clear deletes overflow disk file`() { - val overflowDir = createTempDirectory("metarouter-overflow-clear").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) - overflowDiskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("overflow-1")))) + diskStore.write(QueueSnapshot(version = 1, events = listOf(createTestEvent("overflow-1")))) overflowQueue.enqueue(createTestEvent("memory-1")) overflowQueue.clear() assertEquals(0, overflowQueue.size()) assertNull(diskStore.read()) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() } @Test @@ -856,12 +605,10 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-relaunch").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) // Simulate previous session: overflow events were written to disk val previousEvents = (1..3).map { createTestEvent("prev-session-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = previousEvents)) + diskStore.write(QueueSnapshot(version = 1, events = previousEvents)) // New session — create fresh queue (simulating app relaunch) PersistableEventQueue.resetRehydrationFlag() @@ -869,7 +616,6 @@ class PersistableEventQueueTest { maxCapacity = 2000, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) freshQueue.rehydrate() @@ -895,91 +641,22 @@ class PersistableEventQueueTest { assertEquals(3, drained) assertEquals(1, fakeClient.requests.size) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @Test - fun `no overflow disk store drops events normally at capacity`() { - // No overflowDiskStore — events are dropped when queue is full + fun `flushToOfflineStorage returns false when memory queue is empty`() { val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, - overflowDiskStore = null, eventTTLMs = 0 ) - repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } - repeat(3) { i -> smallQueue.enqueue(createTestEvent("overflow-$i")) } - - // Events were dropped, queue stays at capacity - assertEquals(5, smallQueue.size()) - } - - @Test - fun `capacity overflow always flushes to disk when overflow store exists`() { - val overflowDir = createTempDirectory("metarouter-overflow-always").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) - - val smallQueue = PersistableEventQueue( - maxCapacity = 5, - diskStore = diskStore, - maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, - eventTTLMs = 0 - ) - - // Fill queue - repeat(5) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } - - // Next enqueue at capacity flushes entire queue to disk - smallQueue.enqueue(createTestEvent("triggers-flush")) - assertEquals(1, smallQueue.size()) - assertNotNull(overflowDiskStore.read()) - assertEquals(5, overflowDiskStore.read()!!.events.size) - - overflowDir.deleteRecursively() - } - - @Test - fun `flushToOfflineStorage drains queue when overflow store exists`() { - val overflowDir = createTempDirectory("metarouter-overflow-toggle").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) - - val smallQueue = PersistableEventQueue( - maxCapacity = 5, - diskStore = diskStore, - maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, - eventTTLMs = 0 - ) - - repeat(3) { i -> smallQueue.enqueue(createTestEvent("event-$i")) } - - assertTrue(smallQueue.flushToOfflineStorage()) + // No events in memory queue + assertFalse(smallQueue.flushToOfflineStorage()) assertEquals(0, smallQueue.size()) - assertNotNull(overflowDiskStore.read()) - assertEquals(3, overflowDiskStore.read()!!.events.size) - - overflowDir.deleteRecursively() - } - - @Test - fun `flushToOfflineStorage returns false without overflow store`() { - // Queue without overflow store - val noOverflowQueue = PersistableEventQueue( - maxCapacity = 5, - diskStore = diskStore, - overflowDiskStore = null, - eventTTLMs = 0 - ) - - repeat(3) { i -> noOverflowQueue.enqueue(createTestEvent("event-$i")) } - - assertFalse(noOverflowQueue.flushToOfflineStorage()) - assertEquals(3, noOverflowQueue.size()) } // ===== Drain Response Handling ===== @@ -992,20 +669,17 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-413").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // Pre-populate overflow disk with 4 events val events = (1..4).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() // First call: 413, second call (halved batch): 200, third call: 200 @@ -1026,9 +700,8 @@ class PersistableEventQueueTest { val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(4, drained) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1040,20 +713,17 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-413-drop").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // 2 events — first will be too large, second should still send val events = listOf(createTestEvent("big-event"), createTestEvent("normal-event")) - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() // 413 responses until batch is 1, then 413 again (drop), then 200 for remaining @@ -1080,9 +750,8 @@ class PersistableEventQueueTest { // Only normal-event was successfully sent assertEquals(1, drained) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1094,19 +763,16 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-fatal").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) val events = (1..5).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() fakeClient.nextResponse = NetworkResponse(401, emptyMap(), null) @@ -1122,9 +788,8 @@ class PersistableEventQueueTest { val drained = overflowQueue.drainDiskOverflowToNetwork(dispatcher) assertEquals(0, drained) - assertNull(overflowDiskStore.read()) // Overflow store deleted on fatal error + assertNull(diskStore.read()) // Overflow store deleted on fatal error - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1136,20 +801,17 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-400").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) // Write 150 events so drain takes 2 batches (100 + 50) val events = (1..150).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() // First batch: 400 (drop), second batch: 200 (success) @@ -1170,9 +832,8 @@ class PersistableEventQueueTest { // Only second batch of 50 counted as drained assertEquals(50, drained) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1184,19 +845,16 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-429").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) val events = (1..5).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() fakeClient.nextResponse = NetworkResponse(429, emptyMap(), null) @@ -1213,11 +871,10 @@ class PersistableEventQueueTest { assertEquals(0, drained) // Events should still be on disk - val remaining = overflowDiskStore.read() + val remaining = diskStore.read() assertNotNull(remaining) assertEquals(5, remaining!!.events.size) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1229,15 +886,12 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-ttl").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val oneDayMs = 24L * 60 * 60 * 1000 val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = oneDayMs // 1 day TTL ) @@ -1250,7 +904,7 @@ class PersistableEventQueueTest { createTestEvent("fresh-1", timestamp = freshTimestamp), createTestEvent("fresh-2", timestamp = freshTimestamp) ) - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) @@ -1267,9 +921,8 @@ class PersistableEventQueueTest { // Only the 2 fresh events should have been sent assertEquals(2, drained) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @@ -1281,19 +934,16 @@ class PersistableEventQueueTest { every { Log.e(any(), any()) } returns 0 every { Log.e(any(), any(), any()) } returns 0 - val overflowDir = createTempDirectory("metarouter-overflow-concurrent").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) val overflowQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) val events = (1..10).map { createTestEvent("overflow-$it") } - overflowDiskStore.write(QueueSnapshot(version = 1, events = events)) + diskStore.write(QueueSnapshot(version = 1, events = events)) val fakeClient = FakeNetworkClient() fakeClient.nextResponse = NetworkResponse(200, emptyMap(), null) @@ -1317,22 +967,17 @@ class PersistableEventQueueTest { val totalDrained = result1 + result2 assertEquals(10, totalDrained) assertTrue("One drain should return 0", result1 == 0 || result2 == 0) - assertNull(overflowDiskStore.read()) + assertNull(diskStore.read()) - overflowDir.deleteRecursively() unmockkStatic(Log::class) } @Test - fun `requeueToFront flushes memory queue to overflow disk when at capacity`() { - val overflowDir = createTempDirectory("metarouter-requeue-overflow").toFile() - val overflowDiskStore = EventDiskStore(overflowDir) - + fun `requeueToFront flushes memory queue to disk when at capacity`() { val smallQueue = PersistableEventQueue( maxCapacity = 5, diskStore = diskStore, maxOfflineDiskEvents = 100, - overflowDiskStore = overflowDiskStore, eventTTLMs = 0 ) @@ -1353,37 +998,14 @@ class PersistableEventQueueTest { assertEquals("older-1", drained[0].messageId) assertEquals("older-5", drained[4].messageId) - // The 3 new events should be on overflow disk - val snapshot = overflowDiskStore.read() + // The 3 new events should be on disk + val snapshot = diskStore.read() assertNotNull(snapshot) assertEquals(3, snapshot!!.events.size) assertEquals("new-1", snapshot.events[0].messageId) assertEquals("new-3", snapshot.events[2].messageId) - - overflowDir.deleteRecursively() } - @Test - fun `requeueToFront drops newest when no overflow store at capacity`() { - // No overflow store — should still drop (backward compat) - val smallQueue = PersistableEventQueue( - maxCapacity = 5, - diskStore = diskStore, - overflowDiskStore = null, - eventTTLMs = 0 - ) - - val newEvents = (1..3).map { createTestEvent("new-$it") } - newEvents.forEach { smallQueue.enqueue(it) } - - val requeued = (1..5).map { createTestEvent("older-$it") } - smallQueue.requeueToFront(requeued) - - // Memory queue should have the 5 requeued events — new ones were dropped - assertEquals(5, smallQueue.size()) - val drained = smallQueue.drain(10) - assertEquals("older-1", drained[0].messageId) - } private fun createTestEvent(messageId: String, timestamp: String = "2026-01-01T00:00:00.000Z"): EnrichedEventPayload { return EnrichedEventPayload( diff --git a/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt b/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt index 885caa1..ee46fcd 100644 --- a/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt +++ b/metarouter-sdk/src/test/java/com/metarouter/analytics/storage/EventDiskStoreTest.kt @@ -177,35 +177,6 @@ class EventDiskStoreTest { assertFalse(snapshotFile.exists()) } - @Test - fun `custom filename creates separate store`() { - val customStore = EventDiskStore(tempDir, filename = "custom-overflow.v1.json") - val defaultStore = EventDiskStore(tempDir) // Uses default "queue.v1.json" - - val snapshot1 = QueueSnapshot(version = 1, events = listOf(createTestEvent("default-1"))) - val snapshot2 = QueueSnapshot(version = 1, events = listOf(createTestEvent("custom-1"), createTestEvent("custom-2"))) - - defaultStore.write(snapshot1) - customStore.write(snapshot2) - - // Each store reads its own file independently - val defaultLoaded = defaultStore.read() - val customLoaded = customStore.read() - - assertNotNull(defaultLoaded) - assertEquals(1, defaultLoaded!!.events.size) - assertEquals("default-1", defaultLoaded.events[0].messageId) - - assertNotNull(customLoaded) - assertEquals(2, customLoaded!!.events.size) - assertEquals("custom-1", customLoaded.events[0].messageId) - - // Delete one doesn't affect the other - customStore.delete() - assertNull(customStore.read()) - assertNotNull(defaultStore.read()) - } - private fun createTestEvent(messageId: String): EnrichedEventPayload { return EnrichedEventPayload( type = EventType.TRACK,