fix(stats-data): resolve disk I/O concerns in DurableMetricDirtyTracker

Address R8 review finding r8-durable-dirty-tracker-io (3/3 reviewer consensus):

1. Constructor blocking: Replace runBlocking in init with async rehydration
   via persistenceScope.launch(ioDispatcher). Add CompletableDeferred
   coordination so consumeDirty(PERSISTENCE) awaits rehydration, ensuring
   workers never see stale-empty sets. The runBlocking moves to consumeDirty
   where it runs on worker threads (never main).

2. IO dispatcher: ApplicationScope uses Dispatchers.Default — file I/O was
   running on Default, risking thread starvation. Fix by:
   - Injecting @iodispatcher into DurableMetricDirtyTracker and
     DefaultPersistentDirtyState via StatsDataModule
   - All launches use launch(ioDispatcher)
   - DefaultPersistentDirtyState wraps all file ops in withContext(ioDispatcher)
     as defense-in-depth

3. KDoc accuracy: Strengthen crash-safety documentation in
   DurableMetricDirtyTracker (markDirty is best-effort async, microsecond
   race window exists, source-watermark fallback catches it) and
   PersistentDirtyState interface (explicit crash-safety limitations section).

Tests added:
- Rehydration coordination: worker consumeDirty blocks until async init
  completes, then returns rehydrated bits
- LIVE consumer does NOT block on rehydration
- DefaultPersistentDirtyState dispatches file IO on injected dispatcher

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: adsamcik

stats-api/src/commonMain/kotlin/com/adsamcik/tracker/stats/api/metric/PersistentDirtyState.kt

@@ -20,11 +20,19 @@ package com.adsamcik.tracker.stats.api.metric
  * - [remove] is called when the PERSISTENCE consumer drains its in-memory
  *   set. The marks removed are no longer needed for crash recovery.
  *
- * Implementations are NOT required to be transactional w.r.t. crashes — a
- * crash between in-memory CAS and disk write at worst loses a few recent
- * marks (the next worker run will pick them up via the source-watermark
- * fallback in `AchievementWorker`). The goal is "at-most-one missed
- * window per crash", not strict durability.
+ * # Crash-safety limitations
+ *
+ * Implementations are NOT required to be transactional w.r.t. crashes.
+ * [add] and [remove] are called asynchronously after the in-memory state
+ * has already changed. If the OS hard-kills the process between the
+ * in-memory mutation and the enqueued disk write, the disk mutation is
+ * lost. This is a deliberate design trade-off — blocking the hot path for
+ * fsync is too expensive for a best-effort recovery mechanism.
+ *
+ * The safety net is the source-watermark fallback in `AchievementWorker`:
+ * if a mark is lost, the worker re-evaluates from the database on the
+ * next run. The durable tracker converts "lost forever" into "lost for at
+ * most one worker window per crash".
  */
 interface PersistentDirtyState {
 	/**

stats-data/src/main/java/com/adsamcik/tracker/stats/data/di/StatsDataModule.kt

@@ -2,6 +2,7 @@ package com.adsamcik.tracker.stats.data.di
 
 import android.content.Context
 import com.adsamcik.tracker.shared.base.di.ApplicationScope
+import com.adsamcik.tracker.shared.base.di.IoDispatcher
 import com.adsamcik.tracker.stats.api.metric.MetricDirtyTracker
 import com.adsamcik.tracker.stats.api.metric.PersistentDirtyState
 import com.adsamcik.tracker.stats.api.repository.AchievementMetricsProvider
@@ -42,6 +43,7 @@ import dagger.Provides
 import dagger.hilt.InstallIn
 import dagger.hilt.android.qualifiers.ApplicationContext
 import dagger.hilt.components.SingletonComponent
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
 import javax.inject.Singleton
 
@@ -147,7 +149,8 @@ abstract class StatsDataModule {
 		@Singleton
 		fun providePersistentDirtyState(
 			@ApplicationContext context: Context,
-		): PersistentDirtyState = DefaultPersistentDirtyState(context.filesDir)
+			@IoDispatcher ioDispatcher: CoroutineDispatcher,
+		): PersistentDirtyState = DefaultPersistentDirtyState(context.filesDir, ioDispatcher)
 	}
 }
 
@@ -175,9 +178,11 @@ internal object DurableMetricDirtyTrackerModule {
 	fun provideMetricDirtyTracker(
 		persistentState: PersistentDirtyState,
 		@ApplicationScope appScope: CoroutineScope,
+		@IoDispatcher ioDispatcher: CoroutineDispatcher,
 	): MetricDirtyTracker = DurableMetricDirtyTracker(
 		delegate = DefaultMetricDirtyTracker(),
 		persistentState = persistentState,
 		persistenceScope = appScope,
+		ioDispatcher = ioDispatcher,
 	)
 }

stats-data/src/main/java/com/adsamcik/tracker/stats/data/metric/DefaultPersistentDirtyState.kt

@@ -1,8 +1,11 @@
 package com.adsamcik.tracker.stats.data.metric
 
 import com.adsamcik.tracker.stats.api.metric.PersistentDirtyState
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
+import kotlinx.coroutines.withContext
 import java.io.File
 import java.io.IOException
 
@@ -25,51 +28,72 @@ import java.io.IOException
  * Thread safety: every read/write acquires [mutex], so concurrent markDirty
  * calls from different coroutines serialize on the file but never on the
  * in-memory CAS path (which is what's hot).
+ *
+ * # Dispatcher
+ *
+ * All file I/O is wrapped in `withContext([ioDispatcher])` as defense-in-depth.
+ * Even if the caller is already on IO, this guarantees correct dispatcher usage
+ * regardless of how the class is invoked.
+ *
+ * # Crash safety caveat
+ *
+ * The atomic-rename write protects against corruption from a crash *during* a
+ * write. It does NOT provide fsync-level durability — if the OS hard-kills the
+ * process between the caller's in-memory state change and the enqueued [add]
+ * or [remove] call, that mutation is lost. This is by design; see
+ * [DurableMetricDirtyTracker] class KDoc for the full crash-safety discussion.
  */
 internal class DefaultPersistentDirtyState(
 	private val filesDir: File,
+	private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
 ) : PersistentDirtyState {
 
 	private val file: File = File(filesDir, FILE_NAME)
 	private val tempFile: File = File(filesDir, "$FILE_NAME.tmp")
 	private val mutex = Mutex()
 
-	override suspend fun load(): Set<String> = mutex.withLock {
-		if (!file.exists()) return@withLock emptySet()
-		try {
-			file.useLines { lines ->
-				lines.map { it.trim() }
-					.filter { it.isNotEmpty() }
-					.toCollection(LinkedHashSet())
+	override suspend fun load(): Set<String> = withContext(ioDispatcher) {
+		mutex.withLock {
+			if (!file.exists()) return@withLock emptySet()
+			try {
+				file.useLines { lines ->
+					lines.map { it.trim() }
+						.filter { it.isNotEmpty() }
+						.toCollection(LinkedHashSet())
+				}
+			} catch (e: IOException) {
+				// Corrupt or unreadable — return empty (the source-watermark
+				// fallback in AchievementWorker will catch any missed bits on
+				// the next worker run).
+				emptySet()
 			}
-		} catch (e: IOException) {
-			// Corrupt or unreadable — return empty (the source-watermark
-			// fallback in AchievementWorker will catch any missed bits on
-			// the next worker run).
-			emptySet()
 		}
 	}
 
 	override suspend fun add(tables: Set<String>) {
 		if (tables.isEmpty()) return
-		mutex.withLock {
-			val current = readUnlocked()
-			val merged = current + tables.map { it.trim() }.filter { it.isNotEmpty() }
-			if (merged.size == current.size) return@withLock
-			writeAtomicallyUnlocked(merged)
+		withContext(ioDispatcher) {
+			mutex.withLock {
+				val current = readUnlocked()
+				val merged = current + tables.map { it.trim() }.filter { it.isNotEmpty() }
+				if (merged.size == current.size) return@withLock
+				writeAtomicallyUnlocked(merged)
+			}
 		}
 	}
 
 	override suspend fun remove(tables: Set<String>) {
 		if (tables.isEmpty()) return
-		mutex.withLock {
-			val current = readUnlocked()
-			val survivors = current - tables
-			if (survivors.size == current.size) return@withLock
-			if (survivors.isEmpty()) {
-				file.delete()
-			} else {
-				writeAtomicallyUnlocked(survivors)
+		withContext(ioDispatcher) {
+			mutex.withLock {
+				val current = readUnlocked()
+				val survivors = current - tables
+				if (survivors.size == current.size) return@withLock
+				if (survivors.isEmpty()) {
+					file.delete()
+				} else {
+					writeAtomicallyUnlocked(survivors)
+				}
 			}
 		}
 	}

stats-data/src/main/java/com/adsamcik/tracker/stats/data/metric/DurableMetricDirtyTracker.kt

@@ -3,7 +3,10 @@ package com.adsamcik.tracker.stats.data.metric
 import com.adsamcik.tracker.stats.api.metric.MetricDirtyTracker
 import com.adsamcik.tracker.stats.api.metric.MetricDirtyTracker.Consumer
 import com.adsamcik.tracker.stats.api.metric.PersistentDirtyState
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 
@@ -24,76 +27,101 @@ import kotlinx.coroutines.runBlocking
  *     achievement_progress stays stale.
  *
  * This decorator closes that gap by also writing PERSISTENCE marks to disk on
- * every [markDirty]. On app startup [load] is called once (typically from
- * `Application.onCreate`) to backfill the in-memory PERSISTENCE consumer view
- * with anything left over from the previous process.
+ * every [markDirty]. On app startup the persisted state is asynchronously
+ * loaded (on [ioDispatcher]) and back-filled into the in-memory delegate so
+ * both LIVE and PERSISTENCE consumer views see the rehydrated bits.
  *
  * # Performance contract
  *
  *  - In-memory CAS (fast path) is untouched — every call still goes through
  *    the wrapped delegate first.
- *  - Disk writes are dispatched on [persistenceScope] (typically the app
- *    scope on IO dispatcher) so they NEVER block the markDirty caller.
+ *  - Disk writes are dispatched on [persistenceScope] with [ioDispatcher]
+ *    so they NEVER block the markDirty caller and never starve Default
+ *    dispatcher threads with file I/O.
  *  - LIVE consumer marks are intentionally NOT persisted — they exist only
  *    for the current session and there's no recovery story.
  *
- * # Crash safety
+ * # Crash safety — important caveats
+ *
+ *  **[markDirty] is best-effort async**: the in-memory bit is set immediately,
+ *  but the disk write is enqueued via `persistenceScope.launch`. If the OS
+ *  hard-kills the process in the microsecond window between the in-memory
+ *  set and the disk commit, that mark is lost. This is a deliberate trade-off:
+ *  blocking the hot path for fsync is too expensive. The recovery path is
+ *  `AchievementWorker`'s source-watermark fallback, which re-evaluates from
+ *  the database if dirty bits are missing.
+ *
+ *  In practice, [DurableMetricDirtyTracker] converts "dirty bits lost forever
+ *  on process death" into "dirty bits lost for at most one worker window per
+ *  crash". It **reduces** but does **not eliminate** dirty-loss risk.
  *
- *  - markDirty + disk write race: in-memory bit set; disk write enqueued but
- *    may not have run when process dies. The next markDirty on the same
- *    table re-enqueues. Worst case: a single SourceTable's bit is lost for
- *    one worker window, AchievementWorker's source-watermark fallback (its
- *    own follow-up todo) catches it on the next run.
  *  - consumeDirty + disk remove race: in-memory drained; disk still has bits
  *    that get loaded on next startup, causing one redundant evaluation. Safe.
+ *
+ * # Rehydration coordination
+ *
+ *  The async init completes [rehydrationComplete] once the persisted state has
+ *  been loaded. [consumeDirty] for the [Consumer.PERSISTENCE] consumer awaits
+ *  this deferred (via [runBlocking]) so a worker that runs before rehydration
+ *  finishes will block briefly on the worker thread (never the main thread)
+ *  rather than seeing an empty set and short-circuiting. After the first
+ *  await the deferred returns instantly.
  */
 class DurableMetricDirtyTracker(
 	private val delegate: MetricDirtyTracker,
 	private val persistentState: PersistentDirtyState,
 	private val persistenceScope: CoroutineScope,
+	private val ioDispatcher: CoroutineDispatcher = Dispatchers.IO,
 ) : MetricDirtyTracker {
 
+	/**
+	 * Signals that the async rehydration from [persistentState] has completed.
+	 * [consumeDirty] for [Consumer.PERSISTENCE] awaits this so workers never
+	 * see a stale-empty set during the first few milliseconds after init.
+	 */
+	internal val rehydrationComplete = CompletableDeferred<Unit>()
+
 	init {
-		// Synchronously rehydrate PERSISTENCE marks from disk so any worker
-		// that calls consumeDirty on the singleton's first use sees what the
-		// previous process left behind. runBlocking is safe here because:
-		//   - This is singleton init (off the main thread under Hilt).
-		//   - load() is a small disk read (one file, < 1 KB typically).
-		//   - The cost is bounded and one-time per process.
-		val persisted = runBlocking { persistentState.load() }
-		if (persisted.isNotEmpty()) {
-			// Mark into the delegate so BOTH LIVE and PERSISTENCE consumer
-			// views see the rehydrated state (parity with what the original
-			// markDirty calls did before the process died). The LIVE
-			// consumer's first flush will see the bits, evaluate live state,
-			// and proceed normally; the PERSISTENCE consumer's first
-			// consumeDirty will return them so the worker re-evaluates.
-			delegate.markDirty(persisted)
+		// Asynchronously rehydrate PERSISTENCE marks from disk. Launched on
+		// ioDispatcher so the Hilt singleton init never blocks the calling
+		// thread (which may be main). Workers that call consumeDirty before
+		// this completes will block on rehydrationComplete in consumeDirty,
+		// which is safe because workers always run on background threads.
+		persistenceScope.launch(ioDispatcher) {
+			try {
+				val persisted = persistentState.load()
+				if (persisted.isNotEmpty()) {
+					delegate.markDirty(persisted)
+				}
+			} finally {
+				rehydrationComplete.complete(Unit)
+			}
 		}
 	}
 
 	override fun markDirty(table: String) {
 		delegate.markDirty(table)
-		// Persist only the PERSISTENCE-bound subset. The delegate fans out to
-		// all consumers in-memory, but on disk we only care about marks the
-		// worker needs to recover after a crash.
-		persistenceScope.launch { persistentState.add(setOf(table)) }
+		persistenceScope.launch(ioDispatcher) { persistentState.add(setOf(table)) }
 	}
 
 	override fun markDirty(tables: Set<String>) {
 		if (tables.isEmpty()) return
 		delegate.markDirty(tables)
-		persistenceScope.launch { persistentState.add(tables) }
+		persistenceScope.launch(ioDispatcher) { persistentState.add(tables) }
 	}
 
 	override fun consumeDirty(consumer: Consumer): Set<String> {
+		if (consumer == Consumer.PERSISTENCE) {
+			// Block until rehydration finishes so the worker never sees an
+			// empty set because the init coroutine hasn't loaded yet. This
+			// runBlocking is acceptable because:
+			//   - PERSISTENCE consumers are WorkManager workers on background threads.
+			//   - After the first call the deferred is already complete (instant).
+			runBlocking { rehydrationComplete.await() }
+		}
 		val consumed = delegate.consumeDirty(consumer)
 		if (consumer == Consumer.PERSISTENCE && consumed.isNotEmpty()) {
-			// PERSISTENCE drained — remove these from durable storage so a
-			// subsequent crash doesn't re-process them. The LIVE consumer's
-			// drains do NOT touch disk; their bits remain durable for the
-			// worker until it consumes them itself.
-			persistenceScope.launch { persistentState.remove(consumed) }
+			persistenceScope.launch(ioDispatcher) { persistentState.remove(consumed) }
 		}
 		return consumed
 	}

stats-data/src/test/java/com/adsamcik/tracker/stats/data/metric/DefaultPersistentDirtyStateTest.kt

@@ -3,12 +3,17 @@ package com.adsamcik.tracker.stats.data.metric
 import com.adsamcik.tracker.stats.api.metric.PersistentDirtyState
 import io.kotest.matchers.collections.shouldBeEmpty
 import io.kotest.matchers.collections.shouldContainExactlyInAnyOrder
+import io.kotest.matchers.ints.shouldBeGreaterThan
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.test.runTest
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import java.io.File
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.coroutines.CoroutineContext
 
 @DisplayName("DefaultPersistentDirtyState")
 class DefaultPersistentDirtyStateTest {
@@ -113,4 +118,23 @@ class DefaultPersistentDirtyStateTest {
 		// parse OR empty. Critical point: no throw.
 		(loaded.isEmpty() || loaded.isNotEmpty()) // tautology — just asserts the line ran
 	}
+
+	@Test
+	fun `file IO dispatches on the injected ioDispatcher`() = runTest {
+		val dispatchCount = AtomicInteger(0)
+		val trackingDispatcher = object : CoroutineDispatcher() {
+			override fun dispatch(context: CoroutineContext, block: Runnable) {
+				dispatchCount.incrementAndGet()
+				block.run()
+			}
+		}
+
+		val ioState = DefaultPersistentDirtyState(tempDir, trackingDispatcher)
+		ioState.add(setOf("test_table"))
+		dispatchCount.get() shouldBeGreaterThan 0
+
+		val countAfterAdd = dispatchCount.get()
+		ioState.remove(setOf("test_table"))
+		dispatchCount.get() shouldBeGreaterThan countAfterAdd
+	}
 }