refactor: 下沉设置 transform 原子更新

app/src/main/java/io/legado/app/data/repository/BookExportSettingsRepository.kt

@@ -21,7 +21,11 @@ class BookExportSettingsRepository : BookExportSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (BookExportSettings) -> BookExportSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, BookExportSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toBookExportSettings,
+            toPrefMap = BookExportSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 

app/src/main/java/io/legado/app/data/repository/ChangeSourceSettingsRepository.kt

@@ -21,7 +21,11 @@ class ChangeSourceSettingsRepository : ChangeSourceSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (ChangeSourceSettings) -> ChangeSourceSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, ChangeSourceSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toChangeSourceSettings,
+            toPrefMap = ChangeSourceSettings::toPrefMap,
+            transform = transform,
+        )
     }
 
     override suspend fun setMigrationOptions(options: ChangeSourceMigrationOptions) {

app/src/main/java/io/legado/app/data/repository/FeatureSettingsRepositories.kt

@@ -201,8 +201,10 @@ class DownloadCacheSettingsRepository : DownloadCacheSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (DownloadCacheSettings) -> DownloadCacheSettings) {
-        AppConfigStore.putAllAndAwait(
-            currentSettings.diffPrefMap(transform, DownloadCacheSettings::toPrefMap)
+        AppConfigStore.atomicUpdateAndAwait(
+            read = Preferences::toDownloadCacheSettings,
+            toPrefMap = DownloadCacheSettings::toPrefMap,
+            transform = transform,
         )
     }
 }
@@ -216,7 +218,11 @@ class CoverSettingsRepository : CoverSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (CoverSettings) -> CoverSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, CoverSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toCoverSettings,
+            toPrefMap = CoverSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 
@@ -229,7 +235,11 @@ class LabSettingsRepository : LabSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (LabSettings) -> LabSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, LabSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toLabSettings,
+            toPrefMap = LabSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 
@@ -242,7 +252,11 @@ class TranslationSettingsRepository : TranslationSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (TranslationSettings) -> TranslationSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, TranslationSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toTranslationSettings,
+            toPrefMap = TranslationSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 
@@ -255,7 +269,11 @@ class BackupSettingsRepository : BackupSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (BackupSettings) -> BackupSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, BackupSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toBackupSettings,
+            toPrefMap = BackupSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 

app/src/main/java/io/legado/app/data/repository/ImportBookSettingsRepository.kt

@@ -21,7 +21,11 @@ class ImportBookSettingsRepository : ImportBookSettingsGateway {
         .distinctUntilChanged()
 
     override suspend fun update(transform: (ImportBookSettings) -> ImportBookSettings) {
-        AppConfigStore.putAll(currentSettings.diffPrefMap(transform, ImportBookSettings::toPrefMap))
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toImportBookSettings,
+            toPrefMap = ImportBookSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 

app/src/main/java/io/legado/app/data/repository/MangaSettingsRepository.kt

@@ -11,33 +11,22 @@ import io.legado.app.help.config.compatDsString
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.distinctUntilChanged
 import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.sync.Mutex
-import kotlinx.coroutines.sync.withLock
 
-class MangaSettingsRepository internal constructor(
-    private val preferences: () -> Preferences,
-    preferencesFlow: Flow<Preferences>,
-    private val putAll: (Map<String, Any?>) -> Unit,
-) : MangaSettingsGateway {
-    constructor() : this(
-        preferences = { AppConfigStore.preferences },
-        preferencesFlow = AppConfigStore.preferencesFlow,
-        putAll = AppConfigStore::putAll,
-    )
-
-    private val updateMutex = Mutex()
+class MangaSettingsRepository : MangaSettingsGateway {
 
     override val currentSettings: MangaSettings
-        get() = preferences().toMangaSettings()
+        get() = AppConfigStore.preferences.toMangaSettings()
 
-    override val settings: Flow<MangaSettings> = preferencesFlow
+    override val settings: Flow<MangaSettings> = AppConfigStore.preferencesFlow
         .map { it.toMangaSettings() }
         .distinctUntilChanged()
 
     override suspend fun update(transform: (MangaSettings) -> MangaSettings) {
-        updateMutex.withLock {
-            putAll(currentSettings.diffPrefMap(transform, MangaSettings::toPrefMap))
-        }
+        AppConfigStore.atomicUpdate(
+            read = Preferences::toMangaSettings,
+            toPrefMap = MangaSettings::toPrefMap,
+            transform = transform,
+        )
     }
 }
 

app/src/main/java/io/legado/app/help/config/AppConfigStore.kt

@@ -106,6 +106,26 @@ object AppConfigStore {
     suspend fun putAllAndAwait(values: Map<String, Any?>) =
         requireCore().putAllAndAwait(values)
 
+    /**
+     * 基于当前生效快照原子更新设置。
+     * [transform] 必须是纯计算、快速且不挂起，不得在其中执行 IO 或回调外部状态。
+     */
+    fun <T> atomicUpdate(
+        read: (Preferences) -> T,
+        toPrefMap: (T) -> Map<String, Any?>,
+        transform: (T) -> T,
+    ) = requireCore().atomicUpdate(read, toPrefMap, transform)
+
+    /**
+     * 原子更新设置并等待 DataStore edit 完成。
+     * [transform] 必须是纯计算、快速且不挂起。
+     */
+    suspend fun <T> atomicUpdateAndAwait(
+        read: (Preferences) -> T,
+        toPrefMap: (T) -> Map<String, Any?>,
+        transform: (T) -> T,
+    ) = requireCore().atomicUpdateAndAwait(read, toPrefMap, transform)
+
     // 订阅：替代 SP 监听器的统一入口，值不存在时发 null
     fun observeString(key: String): Flow<String?> = observe { it.compatDsString(key) }
     fun observeInt(key: String): Flow<Int?> = observe { it.compatDsInt(key) }
@@ -177,6 +197,50 @@ internal class PendingOverlayCore(
         completion.await().getOrThrow()
     }
 
+    /**
+     * 读取 overlay、计算 transform 与差量入队均在同一临界区内完成。
+     * transform 必须是纯计算、快速且不挂起。
+     */
+    fun <T> atomicUpdate(
+        read: (Preferences) -> T,
+        toPrefMap: (T) -> Map<String, Any?>,
+        transform: (T) -> T,
+    ) {
+        synchronized(lock) {
+            val values = atomicDiffLocked(read, toPrefMap, transform)
+            if (values.isNotEmpty()) enqueueAllLocked(values, completion = null)
+        }
+    }
+
+    /** [transform] 必须是纯计算、快速且不挂起。 */
+    suspend fun <T> atomicUpdateAndAwait(
+        read: (Preferences) -> T,
+        toPrefMap: (T) -> Map<String, Any?>,
+        transform: (T) -> T,
+    ) {
+        val completion = synchronized(lock) {
+            val values = atomicDiffLocked(read, toPrefMap, transform)
+            if (values.isEmpty()) {
+                null
+            } else {
+                CompletableDeferred<Result<Unit>>().also {
+                    enqueueAllLocked(values, completion = it)
+                }
+            }
+        }
+        completion?.await()?.getOrThrow()
+    }
+
+    private fun <T> atomicDiffLocked(
+        read: (Preferences) -> T,
+        toPrefMap: (T) -> Map<String, Any?>,
+        transform: (T) -> T,
+    ): Map<String, Any?> {
+        val current = read(_preferences.value)
+        val previous = toPrefMap(current)
+        return toPrefMap(transform(current)).filter { (key, value) -> previous[key] != value }
+    }
+
     private fun enqueueAll(
         values: Map<String, Any?>,
         completion: CompletableDeferred<Result<Unit>>?,
@@ -186,27 +250,35 @@ internal class PendingOverlayCore(
             return
         }
         synchronized(lock) {
-            val writes = values.mapValues { PendingWrite(it.value) }
-            pending.putAll(writes)
-            rebuild()
-            launchWrite {
-                runCatching { persistAll(values) }
-                    .onSuccess { completion?.complete(Result.success(Unit)) }
-                    .onFailure { e ->
-                        synchronized(lock) {
-                            var removed = false
-                            writes.forEach { (key, write) ->
-                                if (pending[key] === write) {
-                                    pending.remove(key)
-                                    removed = true
-                                }
+            enqueueAllLocked(values, completion)
+        }
+    }
+
+    /** 调用方必须已持有 [lock]。 */
+    private fun enqueueAllLocked(
+        values: Map<String, Any?>,
+        completion: CompletableDeferred<Result<Unit>>?,
+    ) {
+        val writes = values.mapValues { PendingWrite(it.value) }
+        pending.putAll(writes)
+        rebuild()
+        launchWrite {
+            runCatching { persistAll(values) }
+                .onSuccess { completion?.complete(Result.success(Unit)) }
+                .onFailure { e ->
+                    synchronized(lock) {
+                        var removed = false
+                        writes.forEach { (key, write) ->
+                            if (pending[key] === write) {
+                                pending.remove(key)
+                                removed = true
                             }
-                            if (removed) rebuild()
-                            LogUtils.e("AppConfigStore", "批量保存设置失败: keys=${values.keys}\n${e.stackTraceStr}")
                         }
-                        completion?.complete(Result.failure(e))
+                        if (removed) rebuild()
+                        LogUtils.e("AppConfigStore", "批量保存设置失败: keys=${values.keys}\n${e.stackTraceStr}")
                     }
-            }
+                    completion?.complete(Result.failure(e))
+                }
         }
     }
 

app/src/test/java/io/legado/app/data/repository/MangaSettingsMappingTest.kt

@@ -8,18 +8,17 @@ import io.legado.app.domain.gateway.MangaSettingsGateway
 import io.legado.app.domain.model.settings.MangaSettings
 import io.legado.app.help.config.PendingOverlayCore
 import io.legado.app.help.config.setPrefValue
-import kotlinx.coroutines.CompletableDeferred
-import kotlinx.coroutines.CoroutineStart
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.asStateFlow
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertFalse
 import org.junit.Assert.assertTrue
 import org.junit.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
 
 class MangaSettingsMappingTest {
 
@@ -102,7 +101,7 @@ class MangaSettingsMappingTest {
     }
 
     @Test
-    fun `并发启用墨水屏与灰度时完整 transform 串行且保持互斥`() = runBlocking {
+    fun `并发启用墨水屏与灰度时完整 transform 串行且保持互斥`() {
         val initial = MangaSettings()
         val core = PendingOverlayCore(
             initial = initial.expectedPrefMap().toPreferences(),
@@ -110,38 +109,42 @@ class MangaSettingsMappingTest {
             persist = { _, _ -> error("不会执行落盘") },
             persistAll = { error("不会执行落盘") },
         )
-        val repository = MangaSettingsRepository(
-            preferences = { core.preferencesFlow.value },
-            preferencesFlow = core.preferencesFlow,
-            putAll = core::putAll,
-        )
-        val firstEntered = CompletableDeferred<Unit>()
-        val releaseFirst = CompletableDeferred<Unit>()
-        val first = launch(Dispatchers.Default) {
-            repository.update {
-                firstEntered.complete(Unit)
-                runBlocking { releaseFirst.await() }
+        val firstEntered = CountDownLatch(1)
+        val releaseFirst = CountDownLatch(1)
+        val first = thread {
+            core.atomicUpdate(
+                read = Preferences::toMangaSettings,
+                toPrefMap = MangaSettings::toPrefMap,
+            ) {
+                firstEntered.countDown()
+                releaseFirst.await()
                 it.copy(enableEInk = true, enableGray = false)
             }
         }
         firstEntered.await()
 
-        val secondEntered = CompletableDeferred<Unit>()
-        val second = launch(start = CoroutineStart.UNDISPATCHED) {
-            repository.update {
-                secondEntered.complete(Unit)
+        val secondStarted = CountDownLatch(1)
+        val secondEntered = CountDownLatch(1)
+        val second = thread {
+            secondStarted.countDown()
+            core.atomicUpdate(
+                read = Preferences::toMangaSettings,
+                toPrefMap = MangaSettings::toPrefMap,
+            ) {
+                secondEntered.countDown()
                 it.copy(enableEInk = false, enableGray = true)
             }
         }
-        val secondRacedWithFirst = secondEntered.isCompleted
-        releaseFirst.complete(Unit)
+        secondStarted.await()
+        val secondRacedWithFirst = secondEntered.await(200, TimeUnit.MILLISECONDS)
+        releaseFirst.countDown()
         first.join()
         second.join()
 
         assertFalse(secondRacedWithFirst)
         assertEquals(
             MangaSettings(enableEInk = false, enableGray = true),
-            repository.currentSettings,
+            core.preferencesFlow.value.toMangaSettings(),
         )
     }
 

app/src/test/java/io/legado/app/help/config/PendingOverlayCoreTest.kt

@@ -1,15 +1,20 @@
 package io.legado.app.help.config
 
 import androidx.datastore.preferences.core.Preferences
+import androidx.datastore.preferences.core.booleanPreferencesKey
 import androidx.datastore.preferences.core.intPreferencesKey
 import androidx.datastore.preferences.core.mutablePreferencesOf
 import androidx.datastore.preferences.core.stringPreferencesKey
-import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.async
+import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNull
 import org.junit.Test
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import kotlin.concurrent.thread
 
 /**
  * 覆盖 PR-1 合入门禁要求的 pending overlay 三个竞态时序：
@@ -17,6 +22,11 @@ import org.junit.Test
  */
 class PendingOverlayCoreTest {
 
+    private data class CoupledSettings(
+        val eInk: Boolean = false,
+        val gray: Boolean = false,
+    )
+
     /** 模拟的 DataStore 落盘状态 + 可手动执行的写队列 */
     private class Harness(initial: Preferences) {
         var dsState: Preferences = initial
@@ -254,4 +264,112 @@ class PendingOverlayCoreTest {
         assertEquals("disk full", result.await().exceptionOrNull()?.message)
         assertEquals(1, core.preferencesFlow.value.compatDsInt("k"))
     }
+
+    @Test
+    fun `原子 transform 串行读取 overlay 并保持跨字段不变式`() {
+        val core = PendingOverlayCore(
+            initial = CoupledSettings().toPreferences(),
+            launchWrite = {},
+            persist = { _, _ -> error("不会执行落盘") },
+            persistAll = { error("不会执行落盘") },
+        )
+        val firstEntered = CountDownLatch(1)
+        val releaseFirst = CountDownLatch(1)
+        val first = thread {
+            core.atomicUpdate(
+                read = { it.toCoupledSettings() },
+                toPrefMap = { it.toPrefMap() },
+            ) {
+                firstEntered.countDown()
+                releaseFirst.await()
+                it.copy(eInk = true, gray = false)
+            }
+        }
+        firstEntered.await()
+
+        val secondStarted = CountDownLatch(1)
+        val secondEntered = CountDownLatch(1)
+        val second = thread {
+            secondStarted.countDown()
+            core.atomicUpdate(
+                read = { it.toCoupledSettings() },
+                toPrefMap = { it.toPrefMap() },
+            ) {
+                secondEntered.countDown()
+                it.copy(eInk = false, gray = true)
+            }
+        }
+        secondStarted.await()
+        val secondRacedWithFirst = secondEntered.await(200, TimeUnit.MILLISECONDS)
+        releaseFirst.countDown()
+        first.join()
+        second.join()
+
+        assertFalse(secondRacedWithFirst)
+        assertEquals(
+            CoupledSettings(eInk = false, gray = true),
+            core.preferencesFlow.value.toCoupledSettings(),
+        )
+    }
+
+    @Test
+    fun `等待原子 transform 落盘失败会回滚并传给调用方`() = runBlocking {
+        val initial = mutablePreferencesOf(intPreferencesKey("k") to 1)
+        val writeQueue = ArrayDeque<suspend () -> Unit>()
+        val core = PendingOverlayCore(
+            initial = initial,
+            launchWrite = { writeQueue += it },
+            persist = { _, _ -> error("unused") },
+            persistAll = { throw java.io.IOException("atomic disk full") },
+        )
+
+        val result = async(start = CoroutineStart.UNDISPATCHED) {
+            runCatching {
+                core.atomicUpdateAndAwait<Int>(
+                    read = { preferences -> preferences.compatDsInt("k") ?: 0 },
+                    toPrefMap = { value -> mapOf("k" to value) },
+                    transform = { it + 1 },
+                )
+            }
+        }
+        assertEquals(2, core.preferencesFlow.value.compatDsInt("k"))
+
+        writeQueue.removeFirst().invoke()
+
+        assertEquals("atomic disk full", result.await().exceptionOrNull()?.message)
+        assertEquals(1, core.preferencesFlow.value.compatDsInt("k"))
+    }
+
+    @Test
+    fun `等待原子 transform 无差量时不入队并立即返回`() = runBlocking {
+        val core = PendingOverlayCore(
+            initial = mutablePreferencesOf(intPreferencesKey("k") to 1),
+            launchWrite = { error("无差量不应入队") },
+            persist = { _, _ -> error("unused") },
+            persistAll = { error("unused") },
+        )
+
+        core.atomicUpdateAndAwait(
+            read = { preferences -> preferences.compatDsInt("k") ?: 0 },
+            toPrefMap = { value -> mapOf("k" to value) },
+            transform = { it },
+        )
+
+        assertEquals(1, core.preferencesFlow.value.compatDsInt("k"))
+    }
+
+    private fun Preferences.toCoupledSettings(): CoupledSettings = CoupledSettings(
+        eInk = compatDsBoolean("eInk") ?: false,
+        gray = compatDsBoolean("gray") ?: false,
+    )
+
+    private fun CoupledSettings.toPrefMap(): Map<String, Any?> = mapOf(
+        "eInk" to eInk,
+        "gray" to gray,
+    )
+
+    private fun CoupledSettings.toPreferences(): Preferences = mutablePreferencesOf(
+        booleanPreferencesKey("eInk") to eInk,
+        booleanPreferencesKey("gray") to gray,
+    )
 }