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 9648971..0f253db 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 @@ -72,6 +72,13 @@ class PersistableEventQueue( // Cache of per-event byte sizes (parallel to memoryQueue order) private val eventSizes = ArrayDeque() + // Holding pen for events that arrived while the memory queue was at capacity + // AND the capacity-triggered disk flush failed. These events are *not* counted + // against [maxCapacity]/[maxCapacityBytes] — they are JS-memory-owned until a + // later flush succeeds and writes them through to disk. Capped at [maxDiskEvents] + // so a sustained disk failure can't balloon memory indefinitely. + private val pendingOverflow = ArrayDeque() + // Guards against concurrent drainDiskOverflowToNetwork invocations private val isDraining = AtomicBoolean(false) @@ -118,11 +125,37 @@ class PersistableEventQueue( flushMemoryToDiskInternal() } + // If flush failed, memory is still over capacity. Route the new event to + // pendingOverflow so it survives until a later flush succeeds — never + // drop a synchronously-enqueued event because a disk write is failing. + if (estimatedBytes + eventSize > maxCapacityBytes || memoryQueue.size >= maxCapacity) { + parkPendingOverflow(event) + return + } + memoryQueue.addLast(event) eventSizes.addLast(eventSize) estimatedBytes += eventSize } + /** + * Park an event in the pending-overflow buffer when disk flush is failing. + * Capped at [maxDiskEvents]; if the cap is reached, the new event is dropped + * (loudly) because memory + pending + disk would otherwise grow without bound. + */ + private fun parkPendingOverflow(event: EnrichedEventPayload) { + if (pendingOverflow.size >= maxDiskEvents) { + Logger.error( + "Disk flush failing and pendingOverflow cap ($maxDiskEvents) reached — dropping event ${event.messageId}" + ) + return + } + pendingOverflow.addLast(event) + Logger.warn( + "Disk flush failed — parked event in pendingOverflow (pending=${pendingOverflow.size})" + ) + } + /** * Drain events from memory. No disk I/O. Updates byte size tracking. */ @@ -185,14 +218,15 @@ class PersistableEventQueue( } /** - * Clear all events from memory and delete the disk file. + * Clear all events from memory (including pendingOverflow) and delete the disk file. */ @Synchronized override fun clear() { - val count = memoryQueue.size + val count = memoryQueue.size + pendingOverflow.size memoryQueue.clear() eventSizes.clear() estimatedBytes = 0L + pendingOverflow.clear() synchronized(diskLock) { deleteDiskStore() } @@ -209,7 +243,7 @@ class PersistableEventQueue( @Synchronized override fun flushToOfflineStorage(): Boolean { if (maxDiskEvents == 0) return false - if (memoryQueue.isEmpty()) return false + if (memoryQueue.isEmpty() && pendingOverflow.isEmpty()) return false flushMemoryToDiskInternal() return true } @@ -221,7 +255,7 @@ class PersistableEventQueue( @Synchronized fun flushToDisk() { if (maxDiskEvents == 0) return - if (memoryQueue.isEmpty()) return + if (memoryQueue.isEmpty() && pendingOverflow.isEmpty()) return flushMemoryToDiskInternal() } @@ -243,11 +277,14 @@ class PersistableEventQueue( } /** - * Check if the flush threshold has been reached (either event count or byte size). + * Check if the flush threshold has been reached (either event count or byte size), + * or if there are pendingOverflow events waiting for a retry flush. */ @Synchronized fun shouldFlushToDisk(): Boolean { - return memoryQueue.size >= flushThresholdEvents || estimatedBytes >= flushThresholdBytes + return memoryQueue.size >= flushThresholdEvents || + estimatedBytes >= flushThresholdBytes || + pendingOverflow.isNotEmpty() } /** @@ -286,7 +323,13 @@ class PersistableEventQueue( try { // Read entire file once and delete — O(n) instead of O(n²) val allEvents = synchronized(diskLock) { - val snapshot = diskStore.read() ?: return 0 + val snapshot = try { + diskStore.read() + } catch (e: Exception) { + // Corrupt snapshot already deleted by the store; nothing to drain this cycle. + Logger.warn("Failed to read disk store during drain: ${e.message}") + null + } ?: return 0 deleteDiskStore() snapshot.events } @@ -372,12 +415,18 @@ class PersistableEventQueue( /** * Write remaining events back to the disk store as a checkpoint. * Used during drain for crash safety and on failure to preserve unsent events. + * Failure is logged but not propagated — the drain loop will re-attempt on the + * next cycle with the events still held in local memory. */ private fun writeCheckpoint(events: List) { if (events.isEmpty()) return synchronized(diskLock) { - diskStore.write(QueueSnapshot(version = 1, events = events)) - hasOverflowData = true + try { + diskStore.write(QueueSnapshot(version = 1, events = events)) + hasOverflowData = true + } catch (e: Exception) { + Logger.error("Checkpoint write failed during drain: ${e.message}") + } } } @@ -385,21 +434,23 @@ class PersistableEventQueue( // ===== Internal ===== /** - * Flush all events from memory queue to disk. Appends to any existing disk data, - * enforces [maxDiskEvents] cap, and resets the memory queue. + * Flush memory queue AND pendingOverflow to disk in a single combined write. * Must be called while holding `synchronized(this)`. * + * Behavior: + * - Reads existing snapshot (treating read errors as empty + logged). + * - Combines existing + memoryQueue + pendingOverflow, applies [maxDiskEvents] cap. + * - Writes the combined snapshot. On success, clears memoryQueue, eventSizes, + * estimatedBytes, and pendingOverflow, and sets [hasOverflowData]. + * - On write failure, leaves all in-memory state intact so the events can be + * retried on the next flush. No data is dropped. + * * 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 flushMemoryToDiskInternal() { - if (memoryQueue.isEmpty()) return - - val batch = memoryQueue.toList() - memoryQueue.clear() - eventSizes.clear() - estimatedBytes = 0L + if (memoryQueue.isEmpty() && pendingOverflow.isEmpty()) return synchronized(diskLock) { val existing = try { @@ -409,16 +460,33 @@ class PersistableEventQueue( emptyList() } - var combined = existing + batch + var combined = existing + memoryQueue.toList() + pendingOverflow.toList() if (combined.size > maxDiskEvents) { val dropCount = combined.size - maxDiskEvents combined = combined.drop(dropCount) Logger.warn("Offline disk cap reached — dropped $dropCount oldest events") } - diskStore.write(QueueSnapshot(version = 1, events = combined)) - hasOverflowData = true - Logger.log("Flushed ${batch.size} events to disk (total on disk: ${combined.size})") + val wroteOK = try { + diskStore.write(QueueSnapshot(version = 1, events = combined)) + true + } catch (e: Exception) { + Logger.error( + "Disk write failed during flush — memory (${memoryQueue.size}) and pendingOverflow (${pendingOverflow.size}) retained for retry: ${e.message}" + ) + false + } + + if (wroteOK) { + val flushedMemory = memoryQueue.size + val flushedPending = pendingOverflow.size + memoryQueue.clear() + eventSizes.clear() + estimatedBytes = 0L + pendingOverflow.clear() + hasOverflowData = true + Logger.log("Flushed ${flushedMemory + flushedPending} events to disk (total on disk: ${combined.size})") + } } } @@ -460,10 +528,18 @@ class PersistableEventQueue( } } - /** Delete the disk file and clear the hasOverflowData flag. */ + /** + * Delete the disk file and clear the hasOverflowData flag on success. + * Best-effort: a real delete failure is logged but not propagated — the flag + * stays true and the next drain/flush cycle will converge on actual state. + */ private fun deleteDiskStore() { - diskStore.delete() - hasOverflowData = false + try { + diskStore.delete() + hasOverflowData = false + } catch (e: Exception) { + Logger.warn("Failed to delete disk store: ${e.message}") + } } private fun estimateEventSize(event: EnrichedEventPayload): Long { 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 91cdaf4..4bdb7c7 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 @@ -9,6 +9,7 @@ import kotlinx.serialization.json.jsonArray import kotlinx.serialization.json.jsonObject import kotlinx.serialization.json.jsonPrimitive import java.io.File +import java.io.IOException /** * Handles reading/writing queue snapshot files to disk. @@ -17,10 +18,13 @@ import java.io.File * Uses atomic write-then-rename to prevent partial writes on crash. * * Error handling: - * - Corrupt/unparseable files: returns null, deletes file, logs warning - * - Unrecognized schema version: returns null, deletes file, logs warning + * - Missing file on read: returns null (no log, no throw) + * - Corrupt / unparseable snapshot, unrecognized version, missing version field: + * deletes the offending file, then throws IOException so callers can distinguish + * real failure from "nothing on disk" * - Individual event deserialization failure: skips that event, keeps valid events, logs warning - * - Missing file: returns null (no log) + * - write failures: throws IOException after cleaning up the tmp file + * - delete on absent file: no-op; delete on real I/O failure: throws IOException * * Thread safety: callers must synchronize externally (PersistableEventQueue does this). */ @@ -75,7 +79,11 @@ class EventDiskStore(private val baseDir: File) { /** * Write a snapshot to disk. Uses atomic write-to-tmp-then-rename. * Overwrites any existing snapshot completely. + * + * Throws IOException if the write is not durable. Callers are expected to + * treat a throw as "batch still owned by the caller" so events can be retried. */ + @Throws(IOException::class) fun write(snapshot: QueueSnapshot) { try { snapshotDir.mkdirs() @@ -89,45 +97,47 @@ class EventDiskStore(private val baseDir: File) { Logger.log("Queue snapshot written to disk (${snapshot.events.size} events)") } catch (e: Exception) { Logger.error("Failed to write queue snapshot to disk: ${e.message}") - tmpFile.delete() + try { tmpFile.delete() } catch (_: Exception) {} + throw if (e is IOException) e else IOException("Failed to write queue snapshot", e) } } /** * Read the snapshot from disk with resilient per-event deserialization. * - * Returns null if: file missing, corrupted JSON structure, unrecognized version. - * Individual events that fail deserialization are skipped (not fatal). - * Deletes corrupt files and files with unrecognized versions. + * Returns null ONLY when the snapshot file is absent — that is the "nothing + * persisted" signal. All other failure modes (I/O error, corrupt JSON, + * unrecognized version, missing version field) delete the offending file + * and throw IOException, so callers can distinguish "no prior work" from + * "real error worth surfacing." + * + * Individual events that fail deserialization are still tolerated (skipped, + * not fatal) — a best-effort within an otherwise-valid envelope. */ + @Throws(IOException::class) fun read(): QueueSnapshot? { if (!snapshotFile.exists()) return null - return try { - val data = snapshotFile.readText(Charsets.UTF_8) + val data = try { + snapshotFile.readText(Charsets.UTF_8) + } catch (e: Exception) { + Logger.warn("Failed to read queue snapshot from disk: ${e.message}") + throw if (e is IOException) e else IOException("Failed to read queue snapshot", e) + } + val parsed = try { // Parse as raw JSON first to extract version and handle events individually val jsonObject = json.parseToJsonElement(data).jsonObject val version = jsonObject["version"]?.jsonPrimitive?.int - ?: run { - Logger.warn("Queue snapshot missing version field") - delete() - return null - } + ?: throw IOException("Queue snapshot missing version field") if (version != CURRENT_VERSION) { - Logger.warn("Skipping queue snapshot with unrecognized version: $version (expected $CURRENT_VERSION)") - delete() - return null + throw IOException("Queue snapshot has unrecognized version $version (expected $CURRENT_VERSION)") } val eventsArray = jsonObject["events"]?.jsonArray - ?: run { - Logger.warn("Queue snapshot missing events array") - delete() - return null - } + ?: throw IOException("Queue snapshot missing events array") // Resilient per-event deserialization: skip individual corrupt events var skipped = 0 @@ -147,21 +157,25 @@ class EventDiskStore(private val baseDir: File) { QueueSnapshot(version = version, events = events) } catch (e: Exception) { - Logger.warn("Failed to read queue snapshot from disk (corrupted?): ${e.message}") - delete() - null + Logger.warn("Queue snapshot unreadable, deleting: ${e.message}") + try { delete() } catch (_: Exception) {} + throw if (e is IOException) e else IOException("Queue snapshot unreadable", e) } + return parsed } /** - * Delete the snapshot file from disk. + * Delete the snapshot file from disk. No-op when absent; throws on real + * I/O failure so callers can tell "already gone" from "filesystem said no." + * The tmp file (if any) is cleaned up best-effort — it's never user-visible. */ + @Throws(IOException::class) fun delete() { - try { - snapshotFile.delete() - tmpFile.delete() - } catch (e: Exception) { - Logger.warn("Failed to delete queue snapshot: ${e.message}") + if (snapshotFile.exists() && !snapshotFile.delete()) { + throw IOException("Failed to delete queue snapshot: $snapshotFile") + } + if (tmpFile.exists()) { + try { tmpFile.delete() } catch (_: Exception) {} } } 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 c6c090a..1117bab 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 @@ -1007,6 +1007,152 @@ class PersistableEventQueueTest { } + // ===== Pending Overflow (write-failure retention) ===== + + @Test + fun `pendingOverflow preserves events when disk write fails at capacity`() { + 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 + + // Poison the disk path: place a regular file where metarouter/disk-queue dir should be. + val poisonedBase = createTempDirectory("metarouter-pq-poisoned").toFile() + val metarouterDir = File(poisonedBase, "metarouter") + metarouterDir.mkdirs() + val blocker = File(metarouterDir, "disk-queue") + blocker.writeText("not a directory") + + val poisonedStore = EventDiskStore(poisonedBase) + val smallQueue = PersistableEventQueue( + maxCapacity = 3, + diskStore = poisonedStore, + maxDiskEvents = 100, + eventTTLMs = 0 + ) + + // Fill memory to capacity + repeat(3) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + assertEquals(3, smallQueue.size()) + + // Overflow: these two events trigger capacity-flushes which fail. + // They must land in pendingOverflow instead of being dropped. + smallQueue.enqueue(createTestEvent("overflow-0")) + smallQueue.enqueue(createTestEvent("overflow-1")) + + // Memory remains at capacity because flush failed and we did not evict. + assertEquals(3, smallQueue.size()) + + // Remove poison and drain 1 from memory so a subsequent flush has room to append. + blocker.delete() + smallQueue.drain(1) + + // Now a successful flush should write the remaining memory AND pendingOverflow + // in a single combined batch. + smallQueue.flushToDisk() + + val snapshot = poisonedStore.read() + assertNotNull(snapshot) + val ids = snapshot!!.events.map { it.messageId } + assertEquals(4, ids.size) + assertTrue( + "expected memory + pendingOverflow events on disk after successful flush, got $ids", + ids.containsAll(listOf("fill-1", "fill-2", "overflow-0", "overflow-1")) + ) + + poisonedBase.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `pendingOverflow retained across repeated failures`() { + 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 poisonedBase = createTempDirectory("metarouter-pq-poisoned2").toFile() + val metarouterDir = File(poisonedBase, "metarouter") + metarouterDir.mkdirs() + val blocker = File(metarouterDir, "disk-queue") + blocker.writeText("not a directory") + + val poisonedStore = EventDiskStore(poisonedBase) + val smallQueue = PersistableEventQueue( + maxCapacity = 2, + diskStore = poisonedStore, + maxDiskEvents = 100, + eventTTLMs = 0 + ) + + // Fill to cap and keep trying. Every enqueue past cap hits a failing disk. + repeat(2) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + smallQueue.enqueue(createTestEvent("overflow-0")) + smallQueue.enqueue(createTestEvent("overflow-1")) + smallQueue.enqueue(createTestEvent("overflow-2")) + + // Memory still pinned at cap across the failure streak — we did not silently evict. + assertEquals(2, smallQueue.size()) + + // Recovery: drain memory dry, then flush. + blocker.delete() + smallQueue.drain(2) + smallQueue.flushToDisk() + + val snapshot = poisonedStore.read() + assertNotNull(snapshot) + assertEquals( + "all three overflow events must survive the failure streak", + listOf("overflow-0", "overflow-1", "overflow-2"), + snapshot!!.events.map { it.messageId } + ) + + poisonedBase.deleteRecursively() + unmockkStatic(Log::class) + } + + @Test + fun `clear wipes pendingOverflow too`() { + 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 poisonedBase = createTempDirectory("metarouter-pq-poisoned3").toFile() + val metarouterDir = File(poisonedBase, "metarouter") + metarouterDir.mkdirs() + val blocker = File(metarouterDir, "disk-queue") + blocker.writeText("not a directory") + + val poisonedStore = EventDiskStore(poisonedBase) + val smallQueue = PersistableEventQueue( + maxCapacity = 2, + diskStore = poisonedStore, + maxDiskEvents = 100, + eventTTLMs = 0 + ) + + repeat(2) { i -> smallQueue.enqueue(createTestEvent("fill-$i")) } + smallQueue.enqueue(createTestEvent("parked")) // lands in pendingOverflow + + // Clear must drop both memory and pendingOverflow. After clear + unpoison, + // a flush with no memory events should be a true no-op. + smallQueue.clear() + blocker.delete() + smallQueue.flushToDisk() + + assertNull( + "flushToDisk after clear() must not write anything — pendingOverflow was cleared too", + poisonedStore.read() + ) + + poisonedBase.deleteRecursively() + unmockkStatic(Log::class) + } + // ===== maxDiskEvents = 0 (in-memory-only ring buffer) ===== @Test 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..4cf5e77 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 @@ -73,30 +73,32 @@ class EventDiskStoreTest { assertEquals(1, loaded!!.events.size) } - @Test - fun `read returns null for corrupted file and deletes it`() { + @Test(expected = Exception::class) + fun `read throws on corrupted file and still deletes it`() { val dir = File(tempDir, "metarouter/disk-queue") dir.mkdirs() val corruptFile = File(dir, "queue.v1.json") corruptFile.writeText("not valid json {{{") - val loaded = store.read() - - assertNull(loaded) - assertFalse(corruptFile.exists()) + try { + store.read() + } finally { + assertFalse("corrupt file must be deleted even though read threw", corruptFile.exists()) + } } - @Test - fun `read returns null for unrecognized schema version and deletes file`() { + @Test(expected = Exception::class) + fun `read throws on unrecognized schema version and still deletes file`() { val dir = File(tempDir, "metarouter/disk-queue") dir.mkdirs() val versionFile = File(dir, "queue.v1.json") versionFile.writeText("""{"version":99,"events":[]}""") - val loaded = store.read() - - assertNull(loaded) - assertFalse(versionFile.exists()) + try { + store.read() + } finally { + assertFalse("version-mismatched file must be deleted even though read threw", versionFile.exists()) + } } @Test @@ -164,17 +166,50 @@ class EventDiskStoreTest { store.delete() } - @Test - fun `read returns null when version field is missing`() { + @Test(expected = Exception::class) + fun `read throws when version field is missing and still deletes file`() { val dir = File(tempDir, "metarouter/disk-queue") dir.mkdirs() val snapshotFile = File(dir, "queue.v1.json") snapshotFile.writeText("""{"events":[]}""") - val loaded = store.read() + try { + store.read() + } finally { + assertFalse("version-less file must be deleted even though read threw", snapshotFile.exists()) + } + } - assertNull(loaded) - assertFalse(snapshotFile.exists()) + @Test + fun `read returns null only when file is absent`() { + // File absent → null (no log, no throw). Contract pinned so callers can + // distinguish "no prior snapshot" from "real I/O failure". + assertNull(store.read()) + } + + @Test(expected = Exception::class) + fun `write throws on real I-O failure`() { + // Poison the snapshot directory path with a regular file so mkdirs() and + // the subsequent tmp-file write both fail at the filesystem level. + val parent = File(tempDir, "metarouter") + parent.mkdirs() + val blocker = File(parent, "disk-queue") + blocker.writeText("not a directory") + + val snapshot = QueueSnapshot(version = 1, events = listOf(createTestEvent("msg-1"))) + store.write(snapshot) + } + + @Test(expected = Exception::class) + fun `read throws on real I-O failure`() { + // Place a directory where the snapshot file should be. snapshotFile.exists() + // returns true, but readText() on a directory throws at the JVM layer. + val snapshotDir = File(tempDir, "metarouter/disk-queue") + snapshotDir.mkdirs() + val dirWhereFileShouldBe = File(snapshotDir, "queue.v1.json") + dirWhereFileShouldBe.mkdirs() + + store.read() } private fun createTestEvent(messageId: String): EnrichedEventPayload {