fix(store): synchronize mutable writes by admitted batch

gradle/libs.versions.toml

@@ -26,7 +26,7 @@ kmmBridge = "1.2.1"
 ktlint = "1.8.0"
 kover = "0.9.9"
 #noinspection UnusedVersionCatalogEntry
-store = "5.1.0-alpha11"
+store = "5.1.0-alpha12"
 truth = "1.4.5"
 turbine = "1.2.1"
 binary-compatibility-validator = "0.18.1"

store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/MutableStore.kt

@@ -2,6 +2,21 @@ package org.mobilenativefoundation.store.store5
 
 import org.mobilenativefoundation.store.core5.ExperimentalStoreApi
 
+/**
+ * Persists each write locally before admitting it for server synchronization. Pending writes for
+ * one key may be coalesced: posting the latest admitted value can complete earlier writes in the
+ * same batch. Updater calls are serialized per key within this instance, while newer writes can
+ * persist locally during an earlier updater call.
+ *
+ * Success callbacks run after internal synchronization locks are released and may reenter the
+ * store. Participating storage, updater, and bookkeeping adapters must not recursively read,
+ * write, or clear the same store and key. This restriction follows inherited coroutine context;
+ * detached work that discards that context cannot be detected.
+ *
+ * Admitted pending writes remain in this instance's memory until acknowledged, even if their
+ * caller is cancelled. This state is not a durable outbox and does not guarantee recovery
+ * across process death. A Bookkeeper is required for eager synchronization during reads.
+ */
 @ExperimentalStoreApi
 interface MutableStore<Key : Any, Output : Any> :
     Read.StreamWithConflictResolution<Key, Output>,

store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreAdapterContext.kt

@@ -0,0 +1,33 @@
+package org.mobilenativefoundation.store.store5.impl
+
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.withContext
+import kotlin.coroutines.AbstractCoroutineContextElement
+import kotlin.coroutines.CoroutineContext
+
+internal class MutableStoreAdapterContext(
+    val store: Any,
+    val storeKey: Any,
+    val parent: MutableStoreAdapterContext?,
+) : AbstractCoroutineContextElement(Key) {
+    companion object Key : CoroutineContext.Key<MutableStoreAdapterContext>
+}
+
+internal suspend fun checkMutableStoreEntry(store: Any, key: Any? = null) {
+    var frame = currentCoroutineContext()[MutableStoreAdapterContext]
+    while (frame != null) {
+        check(frame.store !== store || (key != null && frame.storeKey != key)) {
+            "Recursive MutableStore adapter operation for key=$key."
+        }
+        frame = frame.parent
+    }
+}
+
+internal suspend fun <T> withMutableStoreAdapter(
+    store: Any,
+    key: Any,
+    block: suspend () -> T,
+): T {
+    val parent = currentCoroutineContext()[MutableStoreAdapterContext]
+    return withContext(MutableStoreAdapterContext(store, key, parent)) { block() }
+}

store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreKeyState.kt

@@ -0,0 +1,38 @@
+package org.mobilenativefoundation.store.store5.impl
+
+import kotlinx.coroutines.sync.Mutex
+import org.mobilenativefoundation.store.store5.StoreWriteRequest
+import org.mobilenativefoundation.store.store5.UpdaterResult
+
+internal class MutableStoreKeyState<Key : Any, Output : Any> {
+    val localMutex = Mutex()
+    val remoteMutex = Mutex()
+    val pending = ArrayDeque<PendingStoreWrite<Key, Output>>()
+}
+
+internal class PendingStoreWrite<Key : Any, Output : Any>(
+    val request: StoreWriteRequest<Key, Output, *>,
+) {
+    var acknowledged: UpdaterResult.Success? = null
+}
+
+internal class MutableStoreSyncSnapshot<Key : Any, Output : Any>(
+    val value: Output,
+    val entries: List<PendingStoreWrite<Key, Output>>,
+)
+
+/** Called only with localMutex held; distinct admissions retain identity even for a reused request. */
+internal fun <Key : Any, Output : Any> acknowledgeSnapshot(
+    state: MutableStoreKeyState<Key, Output>,
+    snapshot: MutableStoreSyncSnapshot<Key, Output>,
+    result: UpdaterResult.Success,
+): List<PendingStoreWrite<Key, Output>> {
+    val completed = ArrayList<PendingStoreWrite<Key, Output>>(snapshot.entries.size)
+    for (entry in snapshot.entries) {
+        if (entry.acknowledged == null && state.pending.remove(entry)) {
+            entry.acknowledged = result
+            completed.add(entry)
+        }
+    }
+    return completed
+}