diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 05b5e3555..e3895248a 100644 --- a/gradle/libs.versions.toml +++ b/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" diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/MutableStore.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/MutableStore.kt index 8018451c5..cf3eb1007 100644 --- a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/MutableStore.kt +++ b/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 : Read.StreamWithConflictResolution, diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreAdapterContext.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreAdapterContext.kt new file mode 100644 index 000000000..b0a74e518 --- /dev/null +++ b/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 +} + +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 withMutableStoreAdapter( + store: Any, + key: Any, + block: suspend () -> T, +): T { + val parent = currentCoroutineContext()[MutableStoreAdapterContext] + return withContext(MutableStoreAdapterContext(store, key, parent)) { block() } +} diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreKeyState.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/MutableStoreKeyState.kt new file mode 100644 index 000000000..2512d8931 --- /dev/null +++ b/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 { + val localMutex = Mutex() + val remoteMutex = Mutex() + val pending = ArrayDeque>() +} + +internal class PendingStoreWrite( + val request: StoreWriteRequest, +) { + var acknowledged: UpdaterResult.Success? = null +} + +internal class MutableStoreSyncSnapshot( + val value: Output, + val entries: List>, +) + +/** Called only with localMutex held; distinct admissions retain identity even for a reused request. */ +internal fun acknowledgeSnapshot( + state: MutableStoreKeyState, + snapshot: MutableStoreSyncSnapshot, + result: UpdaterResult.Success, +): List> { + val completed = ArrayList>(snapshot.entries.size) + for (entry in snapshot.entries) { + if (entry.acknowledged == null && state.pending.remove(entry)) { + entry.acknowledged = result + completed.add(entry) + } + } + return completed +} diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealMutableStore.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealMutableStore.kt index 83f7af843..1c396096b 100644 --- a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealMutableStore.kt +++ b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealMutableStore.kt @@ -1,12 +1,10 @@ -@file:Suppress("UNCHECKED_CAST") - package org.mobilenativefoundation.store.store5.impl +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOf -import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock import org.mobilenativefoundation.store.core5.ExperimentalStoreApi @@ -20,9 +18,6 @@ import org.mobilenativefoundation.store.store5.StoreWriteRequest import org.mobilenativefoundation.store.store5.StoreWriteResponse import org.mobilenativefoundation.store.store5.Updater import org.mobilenativefoundation.store.store5.UpdaterResult -import org.mobilenativefoundation.store.store5.impl.extensions.now -import org.mobilenativefoundation.store.store5.internal.concurrent.ThreadSafety -import org.mobilenativefoundation.store.store5.internal.definition.WriteRequestQueue import org.mobilenativefoundation.store.store5.internal.result.EagerConflictResolutionResult import org.mobilenativefoundation.store.store5.internal.result.StoreDelegateWriteResult @@ -32,316 +27,218 @@ internal class RealMutableStore, private val bookkeeper: Bookkeeper?, private val logger: Logger = DefaultLogger(), -) : MutableStore, Clear.Key by delegate, Clear.All by delegate { + private val beforeAcknowledgementCommit: suspend () -> Unit = {}, +) : MutableStore, Clear.All { private val storeLock = Mutex() - private val keyToWriteRequestQueue = mutableMapOf>() - private val keyToThreadSafety = mutableMapOf() + private val states = mutableMapOf>() override fun stream(request: StoreReadRequest): Flow> = flow { - // Ensure we are ready for this key. - safeInitStore(request.key) - - // Try to eagerly resolve conflicts before pulling from network. - when (val eagerConflictResolutionResult = tryEagerlyResolveConflicts(request.key)) { - // TODO(#678): Many use cases will not want to pull immediately after failing to push local changes. - // We should enable configuration of conflict resolution strategies, such as logging, retrying, canceling. - - is EagerConflictResolutionResult.Error.Exception -> { - logger.error(eagerConflictResolutionResult.error.toString()) - } - - is EagerConflictResolutionResult.Error.Message -> { - logger.error(eagerConflictResolutionResult.message) - } - - is EagerConflictResolutionResult.Success.ConflictsResolved -> { - logger.debug(eagerConflictResolutionResult.value.toString()) - } - - EagerConflictResolutionResult.Success.NoConflicts -> { - logger.debug("No conflicts.") - } + checkMutableStoreEntry(this@RealMutableStore, request.key) + val state = stateFor(request.key) + // TODO(#678): Allow configuring whether a failed push should prevent a subsequent fetch. + when (val result = tryEagerlyResolveConflicts(request.key, state)) { + is EagerConflictResolutionResult.Error.Exception -> logger.error(result.error.toString()) + is EagerConflictResolutionResult.Error.Message -> logger.error(result.message) + is EagerConflictResolutionResult.Success.ConflictsResolved -> logger.debug(result.value.toString()) + EagerConflictResolutionResult.Success.NoConflicts -> logger.debug("No conflicts.") } - - // Now, we can just delegate to the underlying stream. - delegate.stream(request).collect { storeReadResponse -> emit(storeReadResponse) } + delegate.stream(request).collect { emit(it) } } @ExperimentalStoreApi override fun stream(requestStream: Flow>): Flow = flow { - // Each incoming write request is enqueued. - // Then we try to update the network and delegate. - - requestStream - .onEach { writeRequest -> - // Prepare per-key data structures. - safeInitStore(writeRequest.key) - - // Enqueue the new write request. - addWriteRequestToQueue(writeRequest) - } - .collect { writeRequest -> - val storeWriteResponse = - try { - // Always write to local first. - // Only proceed to network if local write succeeded. - when (val delegateWriteResult = delegate.write(writeRequest.key, writeRequest.value)) { - is StoreDelegateWriteResult.Error.Exception -> { - StoreWriteResponse.Error.Exception(delegateWriteResult.error) - } - is StoreDelegateWriteResult.Error.Message -> { - StoreWriteResponse.Error.Message(delegateWriteResult.error) - } - is StoreDelegateWriteResult.Success -> { - // Try to sync to network. - when (val updaterResult = tryUpdateServer(writeRequest)) { - is UpdaterResult.Error.Exception -> StoreWriteResponse.Error.Exception(updaterResult.error) - is UpdaterResult.Error.Message -> StoreWriteResponse.Error.Message(updaterResult.message) - is UpdaterResult.Success.Typed<*> -> { - val typedValue = updaterResult.value as? Response - if (typedValue == null) { - StoreWriteResponse.Success.Untyped(updaterResult.value) - } else { - StoreWriteResponse.Success.Typed(updaterResult.value) - } - } - is UpdaterResult.Success.Untyped -> StoreWriteResponse.Success.Untyped(updaterResult.value) - } - } - } - } catch (throwable: Throwable) { - StoreWriteResponse.Error.Exception(throwable) - } - emit(storeWriteResponse) - } + requestStream.collect { emit(writeRequest(it)) } } @ExperimentalStoreApi override suspend fun write(request: StoreWriteRequest): StoreWriteResponse = stream(flowOf(request)).first() - private suspend fun tryUpdateServer(request: StoreWriteRequest): UpdaterResult { - val updaterResult = postLatest(request.key) - - if (updaterResult is UpdaterResult.Success) { - // We successfully synced to network, can now clear out any stale writes. - updateWriteRequestQueue( - key = request.key, - created = request.created, - updaterResult = updaterResult, - ) - bookkeeper?.clear(request.key) - } else { - // Could not sync, need to record a failed timestamp. - bookkeeper?.setLastFailedSync(request.key) - } - - return updaterResult + override suspend fun clear(key: Key) { + checkMutableStoreEntry(this, key) + delegate.clear(key) } - /** - * Post the very latest write for [key] to the network using [updater]. - */ - private suspend fun postLatest(key: Key): UpdaterResult { - // The "latest" is the last item in the queue for this key. - val writer = getLatestWriteRequest(key) + @ExperimentalStoreApi + override suspend fun clear() { + checkMutableStoreEntry(this) + delegate.clear() + } - return when (val updaterResult = updater.post(key, writer.value)) { - is UpdaterResult.Error.Exception -> UpdaterResult.Error.Exception(updaterResult.error) - is UpdaterResult.Error.Message -> UpdaterResult.Error.Message(updaterResult.message) - is UpdaterResult.Success.Untyped -> UpdaterResult.Success.Untyped(updaterResult.value) - is UpdaterResult.Success.Typed<*> -> { - val typedValue = updaterResult.value as? Response - if (typedValue == null) { - UpdaterResult.Success.Untyped(updaterResult.value) - } else { - UpdaterResult.Success.Typed(updaterResult.value) + private suspend fun writeRequest(request: StoreWriteRequest): StoreWriteResponse = + try { + checkMutableStoreEntry(this, request.key) + val state = stateFor(request.key) + val entry = PendingStoreWrite(request) + val localResult = + state.localMutex.withLock { + withMutableStoreAdapter(this, request.key) { + delegate.write(request.key, request.value).also { result -> + // Admit inside the adapter context before returning through prompt cancellation. + if (result is StoreDelegateWriteResult.Success) state.pending.add(entry) + } + } + } + when (localResult) { + is StoreDelegateWriteResult.Error.Exception -> { + if (localResult.error is CancellationException) throw localResult.error + StoreWriteResponse.Error.Exception(localResult.error) } + is StoreDelegateWriteResult.Error.Message -> StoreWriteResponse.Error.Message(localResult.error) + is StoreDelegateWriteResult.Success -> requireNotNull(synchronize(request.key, state, entry)).toWriteResponse() } + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + StoreWriteResponse.Error.Exception(error) } - } /** - * Remove or keep queue items after a successful network sync. + * A writer releases localMutex before waiting here. The only nested per-key lock order is + * remoteMutex then localMutex; local persistence can therefore proceed while the updater waits. */ - private suspend fun updateWriteRequestQueue( + private suspend fun synchronize( key: Key, - created: Long, - updaterResult: UpdaterResult.Success, - ) { - val nextWriteRequestQueue = - withWriteRequestQueueLock(key) { - val remaining = ArrayDeque>() - - for (writeRequest in this) { - if (writeRequest.created <= created) { - // Mark each relevant request as succeeded. - updater.onCompletion?.onSuccess?.invoke(updaterResult) + state: MutableStoreKeyState, + entry: PendingStoreWrite? = null, + ): UpdaterResult? { + if (entry == null && bookkeeper == null) return null + var completed = emptyList>() + var completedResult: UpdaterResult.Success? = null + try { + return state.remoteMutex.withLock remote@{ + val acknowledged = state.localMutex.withLock { entry?.acknowledged } + if (acknowledged != null) return@remote acknowledged + + val snapshot = + state.localMutex.withLock { + if (entry != null) { + MutableStoreSyncSnapshot(state.pending.last().request.value, state.pending.toList()) + } else { + withMutableStoreAdapter(this, key) { + val failed = bookkeeper?.getLastFailedSync(key) + if (failed == null && state.pending.isEmpty()) return@withMutableStoreAdapter null + // Direct SourceOfTruth updates may be newer than the last pending request. + val latest = delegate.latestOrNull(key) ?: return@withMutableStoreAdapter null + MutableStoreSyncSnapshot(latest, state.pending.toList()) + } + } + } ?: return@remote null - val storeWriteResponse = - when (updaterResult) { - is UpdaterResult.Success.Typed<*> -> { - val typedValue = updaterResult.value as? Response - if (typedValue == null) { - StoreWriteResponse.Success.Untyped(updaterResult.value) - } else { - StoreWriteResponse.Success.Typed(updaterResult.value) - } + val result = post(key, snapshot.value) + when (result) { + is UpdaterResult.Success -> { + beforeAcknowledgementCommit() + state.localMutex.withLock { + completed = acknowledgeSnapshot(state, snapshot, result) + completedResult = result + // Keep the empty check and clear together so a new admission cannot slip in. + if (state.pending.isEmpty()) { + tryBookkeeping("clear", key) { bookkeeper?.clear(key) } + } + } + } + is UpdaterResult.Error -> { + state.localMutex.withLock { + tryBookkeeping("setLastFailedSync", key) { + if (bookkeeper?.setLastFailedSync(key) == false) { + logger.error("Bookkeeper.setLastFailedSync returned false for key=$key.") } - - is UpdaterResult.Success.Untyped -> StoreWriteResponse.Success.Untyped(updaterResult.value) } - - // Notify each on-completion callback. - writeRequest.onCompletions?.forEach { onStoreWriteCompletion -> - onStoreWriteCompletion.onSuccess(storeWriteResponse) } - } else { - // Keep requests that happened after created. - remaining.add(writeRequest) } } - remaining + result } - - // Update the in-memory map outside the queue's mutex. - storeLock.withLock { - keyToWriteRequestQueue[key] = nextWriteRequestQueue + } finally { + // The committed batch survives a canceled clear. Callbacks run after both locks release. + completedResult?.let { deliverCallbacks(completed, it) } } } - /** - * Locks the queue for [key] and invokes [block]. - */ - private suspend fun withWriteRequestQueueLock( + private suspend fun post( key: Key, - block: suspend WriteRequestQueue.() -> Result, - ): Result { - // Acquire the ThreadSafety object for this key without holding storeLock. - val threadSafety = getThreadSafety(key) - - // Exclusively lock the queue's own mutex. The block both reads and structurally mutates the - // per-key ArrayDeque (add / iterate-and-rebuild), so callers must mutually exclude each other. - // A shared/reader lock here would allow a concurrent add() during iteration, corrupting the - // deque's backing array — a ConcurrentModificationException on the JVM and an EXC_BAD_ACCESS - // on Kotlin/Native. - return threadSafety.writeRequests.mutex.withLock { - val queue = getQueue(key) - queue.block() - } - } - - private suspend fun getLatestWriteRequest(key: Key): StoreWriteRequest { - val threadSafety = getThreadSafety(key) - threadSafety.writeRequests.mutex.lock() - return try { - val queue = getQueue(key) - require(queue.isNotEmpty()) { - "No writes found for key=$key." + value: Output, + ): UpdaterResult = + try { + withMutableStoreAdapter(this, key) { updater.post(key, value) }.also { result -> + if (result is UpdaterResult.Error.Exception && result.error is CancellationException) throw result.error } - queue.last() - } finally { - threadSafety.writeRequests.mutex.unlock() + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + UpdaterResult.Error.Exception(error) } - } - /** - * Checks if we have un-synced writes or a recorded failed sync for [key]. - */ - private suspend fun conflictsMightExist(key: Key): Boolean { - val failed = bookkeeper?.getLastFailedSync(key) - return (failed != null) || !writeRequestsQueueIsEmpty(key) + private suspend fun tryBookkeeping( + operation: String, + key: Key, + block: suspend () -> Unit, + ) { + try { + withMutableStoreAdapter(this, key) { block() } + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + logger.error("Bookkeeper.$operation failed for key=$key.", error) + } } - private fun writeRequestsQueueIsEmpty(key: Key): Boolean = keyToWriteRequestQueue[key].isNullOrEmpty() - - private suspend fun addWriteRequestToQueue(writeRequest: StoreWriteRequest) = - withWriteRequestQueueLock(writeRequest.key) { - add(writeRequest) + private suspend fun tryEagerlyResolveConflicts( + key: Key, + state: MutableStoreKeyState, + ): EagerConflictResolutionResult = + try { + when (val result = synchronize(key, state)) { + null -> EagerConflictResolutionResult.Success.NoConflicts + is UpdaterResult.Error.Exception -> EagerConflictResolutionResult.Error.Exception(result.error) + is UpdaterResult.Error.Message -> EagerConflictResolutionResult.Error.Message(result.message) + is UpdaterResult.Success -> EagerConflictResolutionResult.Success.ConflictsResolved(result) + } + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Throwable) { + EagerConflictResolutionResult.Error.Exception(error) } - private suspend fun tryEagerlyResolveConflicts(key: Key): EagerConflictResolutionResult { - // Acquire the ThreadSafety object for this key without holding storeLock. - val threadSafety = getThreadSafety(key) - - // Lock just long enough to check if conflicts exist. - val (latestValue, conflictsExist) = - threadSafety.readCompletions.mutex.withLock { - val latestValue = delegate.latestOrNull(key) - val conflictsExist = latestValue != null && bookkeeper != null && conflictsMightExist(key) - latestValue to conflictsExist - } + private fun deliverCallbacks( + entries: List>, + result: UpdaterResult.Success, + ) { + var cancellation: CancellationException? = null - return if (!conflictsExist || latestValue == null) { - EagerConflictResolutionResult.Success.NoConflicts - } else { + fun invokeCallback(callback: () -> Unit) { try { - val updaterResult = - updater.post(key, latestValue).also { updaterResult -> - if (updaterResult is UpdaterResult.Success) { - // If it succeeds, we want to remove stale requests and clear the bookkeeper. - updateWriteRequestQueue(key = key, created = now(), updaterResult = updaterResult) - - bookkeeper?.clear(key) - } - } - - when (updaterResult) { - is UpdaterResult.Error.Exception -> { - EagerConflictResolutionResult.Error.Exception(updaterResult.error) - } - - is UpdaterResult.Error.Message -> { - EagerConflictResolutionResult.Error.Message(updaterResult.message) - } - - is UpdaterResult.Success -> { - EagerConflictResolutionResult.Success.ConflictsResolved(updaterResult) - } - } + callback() + } catch (error: CancellationException) { + if (cancellation == null) cancellation = error } catch (error: Throwable) { - EagerConflictResolutionResult.Error.Exception(error) + logger.error("MutableStore success callback failed.", error) } } - } - - /** - * Ensures that [keyToThreadSafety] and [keyToWriteRequestQueue] have entries for [key]. - * We only hold [storeLock] while touching these two maps, then release it immediately. - */ - private suspend fun safeInitStore(key: Key) { - storeLock.withLock { - if (keyToThreadSafety[key] == null) { - keyToThreadSafety[key] = ThreadSafety() - } - if (keyToWriteRequestQueue[key] == null) { - keyToWriteRequestQueue[key] = ArrayDeque() - } + val response = result.toSuccessResponse() + for (entry in entries) { + updater.onCompletion?.onSuccess?.let { callback -> invokeCallback { callback(result) } } + entry.request.onCompletions?.forEach { callback -> invokeCallback { callback.onSuccess(response) } } } + cancellation?.let { throw it } } - /** - * Retrieves the [ThreadSafety] object for [key] without reinitializing it, since [safeInitStore] handles creation. - * We do a quick [storeLock] read then release it without nesting per-key locks inside [storeLock]. - */ - private suspend fun getThreadSafety(key: Key): ThreadSafety { - return storeLock.withLock { - requireNotNull(keyToThreadSafety[key]) { - "ThreadSafety not initialized for key=$key." - } + private fun UpdaterResult.toWriteResponse(): StoreWriteResponse = + when (this) { + is UpdaterResult.Error.Exception -> StoreWriteResponse.Error.Exception(error) + is UpdaterResult.Error.Message -> StoreWriteResponse.Error.Message(message) + is UpdaterResult.Success -> toSuccessResponse() } - } - /** - * Helper to retrieve the queue for [key] without re-initialization logic. - */ - private suspend fun getQueue(key: Key): WriteRequestQueue { - return storeLock.withLock { - requireNotNull(keyToWriteRequestQueue[key]) { - "No write request queue found for key=$key." - } + private fun UpdaterResult.Success.toSuccessResponse(): StoreWriteResponse.Success = + when (this) { + is UpdaterResult.Success.Typed<*> -> StoreWriteResponse.Success.Typed(value) + is UpdaterResult.Success.Untyped -> StoreWriteResponse.Success.Untyped(value) } - } + + private suspend fun stateFor(key: Key): MutableStoreKeyState = + storeLock.withLock { states.getOrPut(key) { MutableStoreKeyState() } } } diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealStore.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealStore.kt index 4e9a0b16a..fffff06a7 100644 --- a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealStore.kt +++ b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/RealStore.kt @@ -16,6 +16,7 @@ package org.mobilenativefoundation.store.store5.impl import co.touchlab.kermit.Logger +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow @@ -354,6 +355,8 @@ internal class RealStore( memCache?.put(key, value) StoreDelegateWriteResult.Success } + } catch (cancellation: CancellationException) { + throw cancellation } catch (error: Throwable) { StoreDelegateWriteResult.Error.Exception(error) } diff --git a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/SourceOfTruthWithBarrier.kt b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/SourceOfTruthWithBarrier.kt index d791757a7..f9926a25f 100644 --- a/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/SourceOfTruthWithBarrier.kt +++ b/store/src/commonMain/kotlin/org/mobilenativefoundation/store/store5/impl/SourceOfTruthWithBarrier.kt @@ -18,6 +18,7 @@ package org.mobilenativefoundation.store.store5.impl import kotlinx.atomicfu.atomic import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.catch @@ -26,6 +27,7 @@ import kotlinx.coroutines.flow.flatMapLatest import kotlinx.coroutines.flow.flow import kotlinx.coroutines.flow.flowOf import kotlinx.coroutines.flow.onStart +import kotlinx.coroutines.withContext import org.mobilenativefoundation.store.store5.Converter import org.mobilenativefoundation.store.store5.SourceOfTruth import org.mobilenativefoundation.store.store5.StoreReadResponse @@ -157,43 +159,34 @@ internal class SourceOfTruthWithBarrier throw CancellationException("writer cancelled") } + + assertFailsWith { source.write(1, "cancelled") } + assertTrue(currentCoroutineContext().isActive) + assertNull(persister.read(1)) + assertEquals(0, source.barrierCount()) + + persister.preWriteCallback = null + source.write(1, "recovered") + assertEquals("recovered", persister.read(1)) + } + + @Test + fun cancellingSuspendedWriterReopensBarrierForActiveReader() = + testScope.runTest { + val entered = CompletableDeferred() + val release = CompletableDeferred() + persister.preWriteCallback = { _, value -> + if (value == "cancelled") { + entered.complete(Unit) + release.await() + } + value + } + val collected = mutableListOf>() + val reader = launch { source.reader(1, CompletableDeferred(Unit)).collect { collected.add(it) } } + advanceUntilIdle() + val writer = launch { source.write(1, "cancelled") } + entered.await() + advanceUntilIdle() + writer.cancelAndJoin() + advanceUntilIdle() + assertTrue(reader.isActive) + assertTrue(writer.isCancelled) + assertEquals(2, collected.size) + assertNull(collected.last().dataOrNull()) + + source.write(1, "recovered") + advanceUntilIdle() + assertEquals("recovered", collected.last().dataOrNull()) + reader.cancelAndJoin() + assertEquals(0, source.barrierCount()) + } + private val testScope = TestScope() private val persister = InMemoryPersister() private val delegate: SourceOfTruth = diff --git a/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreAcknowledgementTest.kt b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreAcknowledgementTest.kt new file mode 100644 index 000000000..6e446c56c --- /dev/null +++ b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreAcknowledgementTest.kt @@ -0,0 +1,472 @@ +@file:OptIn( + kotlinx.coroutines.ExperimentalCoroutinesApi::class, + org.mobilenativefoundation.store.core5.ExperimentalStoreApi::class, +) + +package org.mobilenativefoundation.store.store5.mutablestore + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.mobilenativefoundation.store.store5.SourceOfTruth +import org.mobilenativefoundation.store.store5.StoreReadRequest +import org.mobilenativefoundation.store.store5.StoreReadResponse +import org.mobilenativefoundation.store.store5.StoreWriteResponse +import org.mobilenativefoundation.store.store5.UpdaterResult +import org.mobilenativefoundation.store.store5.mutablestore.util.MutableStoreRaceFixture +import org.mobilenativefoundation.store.store5.mutablestore.util.RaceGate +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertIs +import kotlin.test.assertNotNull +import kotlin.test.assertNull + +class MutableStoreAcknowledgementTest { + @Test + fun newerWriteSurvivesAcknowledgementCommit() = + runTest { + val acknowledgement = RaceGate() + val bLocalReturn = RaceGate() + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> UpdaterResult.Success.Typed("ack:$value") }, + beforeLocalReturn = { _, value -> + if (value == "B") bLocalReturn.pause() + }, + beforeAcknowledgementCommit = { acknowledgement.pause() }, + ) + + val a = async { fixture.store.write(fixture.request("A", 1L)) } + acknowledgement.entered.await() + val b = async { fixture.store.write(fixture.request("B", 2L)) } + bLocalReturn.entered.await() + assertEquals("B", fixture.local.value["key"]) + + acknowledgement.open() + runCurrent() + // Fixed acknowledgment can be waiting for B's local transaction here. + bLocalReturn.open() + + assertIs(a.await()) + assertIs(b.await()) + assertEquals(listOf("key" to "A", "key" to "B"), fixture.posted) + assertEquals("B", fixture.remote["key"]) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:A"), + "B" to StoreWriteResponse.Success.Typed("ack:B"), + ), + fixture.successes, + ) + } + + @Test + fun olderEagerSuccessDoesNotAcknowledgeNewerFailedWrite() = + runTest { + val eagerA = RaceGate() + var aPosts = 0 + var failB = true + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + when { + value == "A" && ++aPosts == 1 -> UpdaterResult.Error.Message("offline:A") + value == "A" -> { + eagerA.pause() + UpdaterResult.Success.Typed("ack:A") + } + failB -> UpdaterResult.Error.Message("offline:B") + else -> UpdaterResult.Success.Typed("ack:B") + } + }, + ) + assertIs(fixture.store.write(fixture.request("A", 1L))) + val reader = + async { + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + } + eagerA.entered.await() + val b = async { fixture.store.write(fixture.request("B", 2L)) } + runCurrent() + assertEquals("B", fixture.local.value["key"]) + assertEquals(0, fixture.successes.count { it.first == "B" }) + + // B's remote attempt waits for A after the fix. Do not await B yet. + eagerA.open() + reader.await() + assertEquals(StoreWriteResponse.Error.Message("offline:B"), b.await()) + assertEquals(0, fixture.successes.count { it.first == "B" }) + assertNotNull(fixture.bookkeeper.getLastFailedSync("key")) + + failB = false + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + assertEquals("B", fixture.remote["key"]) + assertEquals( + listOf>( + "B" to StoreWriteResponse.Success.Typed("ack:B"), + ), + fixture.successes.filter { it.first == "B" }, + ) + assertNull(fixture.bookkeeper.getLastFailedSync("key")) + } + + @Test + fun failedLocalWriteIsNeverPostedOrAcknowledged() = + runTest { + val sequential = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> UpdaterResult.Success.Typed("ack:$value") }, + beforeLocalWrite = { _, value -> + if (value == "B") error("local:B") + }, + ) + + val failedB = + assertIs( + sequential.store.write(sequential.request("B", 1L)), + ) + val writeException = assertIs(failedB.error) + assertEquals("local:B", assertIs(writeException.cause).message) + assertIs(sequential.store.write(sequential.request("A", 2L))) + assertEquals(listOf("key" to "A"), sequential.localWrites) + assertEquals(listOf("key" to "A"), sequential.posted) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:A"), + ), + sequential.successes, + ) + + val aLocalReturn = RaceGate() + val overlapping = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> UpdaterResult.Success.Typed("ack:$value") }, + beforeLocalWrite = { _, value -> + if (value == "B") error("local:B") + }, + beforeLocalReturn = { _, value -> + if (value == "A") aLocalReturn.pause() + }, + ) + val a = async { overlapping.store.write(overlapping.request("A", 3L)) } + aLocalReturn.entered.await() + val b = async { overlapping.store.write(overlapping.request("B", 4L)) } + runCurrent() + assertEquals(listOf("key" to "A"), overlapping.localAttempts) + + aLocalReturn.open() + assertIs(a.await()) + val overlappingFailedB = assertIs(b.await()) + assertIs(overlappingFailedB.error) + assertEquals(listOf("key" to "A"), overlapping.localWrites) + assertEquals(listOf("key" to "A"), overlapping.posted) + assertEquals(0, overlapping.successes.count { it.first == "B" }) + } + + @Test + fun acknowledgementIgnoresCreatedOrder() = + runTest { + val scope = this + + suspend fun verifyOrder( + aCreated: Long, + bCreated: Long, + ) { + var attempt = 0 + val fixture = + MutableStoreRaceFixture( + scope = scope, + post = { _, value -> + if (++attempt == 1) { + UpdaterResult.Error.Message("offline:$value") + } else { + UpdaterResult.Success.Typed("ack:$value") + } + }, + ) + + assertIs(fixture.store.write(fixture.request("A", aCreated))) + assertEquals( + StoreWriteResponse.Success.Typed("ack:B"), + fixture.store.write(fixture.request("B", bCreated)), + ) + assertEquals(listOf("key" to "A", "key" to "B"), fixture.posted) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:B"), + "B" to StoreWriteResponse.Success.Typed("ack:B"), + ), + fixture.successes, + ) + assertNull(fixture.bookkeeper.getLastFailedSync("key")) + } + + verifyOrder(Long.MAX_VALUE, Long.MIN_VALUE) + verifyOrder(7L, 7L) + } + + @Test + fun reusedRequestHasDistinctAdmissions() = + runTest { + var attempt = 0 + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + if (++attempt == 1) { + UpdaterResult.Error.Message("offline:$value") + } else { + UpdaterResult.Success.Typed("ack:$value") + } + }, + ) + val request = fixture.request("A", 1L, id = "same") + + assertIs(fixture.store.write(request)) + assertEquals(StoreWriteResponse.Success.Typed("ack:A"), fixture.store.write(request)) + + assertEquals(listOf("key" to "A", "key" to "A"), fixture.localWrites) + assertEquals(listOf("key" to "A", "key" to "A"), fixture.posted) + assertEquals( + listOf>( + "same" to StoreWriteResponse.Success.Typed("ack:A"), + "same" to StoreWriteResponse.Success.Typed("ack:A"), + ), + fixture.successes, + ) + assertEquals(2, fixture.updaterSuccesses.size) + } + + @Test + fun coalescedWaitingCallerReturnsSavedSuccess() = + runTest { + val aPost = RaceGate() + val bLocalReturn = RaceGate() + val cPersisted = CompletableDeferred() + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + if (value == "A") aPost.pause() + UpdaterResult.Success.Typed("ack:$value") + }, + beforeLocalReturn = { _, value -> + when (value) { + "B" -> bLocalReturn.pause() + "C" -> cPersisted.complete(Unit) + } + }, + ) + + val a = async { fixture.store.write(fixture.request("A", 1L)) } + aPost.entered.await() + val b = async { fixture.store.write(fixture.request("B", 2L)) } + bLocalReturn.entered.await() + val c = async { fixture.store.write(fixture.request("C", 3L)) } + runCurrent() + assertEquals(listOf("key" to "A", "key" to "B"), fixture.localWrites) + + bLocalReturn.open() + cPersisted.await() + assertEquals(listOf("key" to "A"), fixture.posted) + aPost.open() + + assertEquals(StoreWriteResponse.Success.Typed("ack:A"), a.await()) + assertEquals(StoreWriteResponse.Success.Typed("ack:C"), b.await()) + assertEquals(StoreWriteResponse.Success.Typed("ack:C"), c.await()) + assertEquals(listOf("key" to "A", "key" to "C"), fixture.posted) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:A"), + "B" to StoreWriteResponse.Success.Typed("ack:C"), + "C" to StoreWriteResponse.Success.Typed("ack:C"), + ), + fixture.successes, + ) + } + + @Test + fun sameKeyPostsSerializeWithoutBlockingLocalWrites() = + runTest { + val aPost = RaceGate() + val bPersisted = CompletableDeferred() + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + if (value == "A") aPost.pause() + UpdaterResult.Success.Typed("ack:$value") + }, + beforeLocalReturn = { _, value -> + if (value == "B") bPersisted.complete(Unit) + }, + ) + + val a = async { fixture.store.write(fixture.request("A", 1L)) } + aPost.entered.await() + val b = async { fixture.store.write(fixture.request("B", 2L)) } + bPersisted.await() + + assertEquals("B", fixture.local.value["key"]) + assertEquals(listOf("key" to "A"), fixture.posted) + assertEquals(1, fixture.maximumActivePosts["key"]) + + aPost.open() + assertIs(a.await()) + assertIs(b.await()) + assertEquals(listOf("key" to "A", "key" to "B"), fixture.posted) + assertEquals(1, fixture.maximumActivePosts["key"]) + } + + @Test + fun eagerRetryRechecksAfterExplicitSuccess() = + runTest { + val explicitA = RaceGate() + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + explicitA.pause() + UpdaterResult.Success.Typed("ack:$value") + }, + ) + + val writer = async { fixture.store.write(fixture.request("A", 1L)) } + explicitA.entered.await() + val reader = + async { + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + } + runCurrent() + assertEquals(listOf("key" to "A"), fixture.posted) + + explicitA.open() + assertIs(writer.await()) + reader.await() + assertEquals(listOf("key" to "A"), fixture.posted) + } + + @Test + fun differentKeysMakeProgressIndependently() = + runTest { + val kPost = RaceGate() + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { key, value -> + if (key == "K") kPost.pause() + UpdaterResult.Success.Typed("ack:$value") + }, + ) + + val k = async { fixture.store.write(fixture.request("value-K", 1L, key = "K")) } + kPost.entered.await() + val jResponse = fixture.store.write(fixture.request("value-J", 2L, key = "J")) + + assertEquals(StoreWriteResponse.Success.Typed("ack:value-J"), jResponse) + assertEquals("value-J", fixture.local.value["J"]) + assertEquals("value-J", fixture.remote["J"]) + assertEquals(listOf("K" to "value-K", "J" to "value-J"), fixture.posted) + + kPost.open() + assertIs(k.await()) + assertEquals("value-K", fixture.remote["K"]) + } + + @Test + fun markerOnlyRetryUsesLocalValue() = + runTest { + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> UpdaterResult.Success.Typed("ack:$value") }, + ) + fixture.local.value = mapOf("key" to "A") + fixture.bookkeeper.setLastFailedSync("key", 1L) + + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + + assertEquals(listOf("key" to "A"), fixture.posted) + assertEquals("A", fixture.remote["key"]) + assertNull(fixture.bookkeeper.getLastFailedSync("key")) + assertEquals(emptyList(), fixture.successes) + assertEquals(emptyList(), fixture.updaterSuccesses) + } + + @Test + fun eagerRetryUsesLatestLocalValueWithPendingWrites() = + runTest { + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + if (value == "A") { + UpdaterResult.Error.Message("offline:A") + } else { + UpdaterResult.Success.Typed("ack:$value") + } + }, + ) + assertIs(fixture.store.write(fixture.request("A", 1L))) + fixture.local.value = mapOf("key" to "B") + fixture.cache.invalidate("key") + + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + + assertEquals(listOf("key" to "A", "key" to "B"), fixture.posted) + assertEquals("B", fixture.remote["key"]) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:B"), + ), + fixture.successes, + ) + assertNull(fixture.bookkeeper.getLastFailedSync("key")) + } + + @Test + fun noBookkeeperPreservesExistingEagerBehavior() = + runTest { + val fixture = + MutableStoreRaceFixture( + scope = this, + post = { _, value -> + if (value == "A") { + UpdaterResult.Error.Message("offline:A") + } else { + UpdaterResult.Success.Typed("ack:$value") + } + }, + withBookkeeper = false, + ) + assertIs(fixture.store.write(fixture.request("A", 1L))) + + fixture.store.stream(StoreReadRequest.localOnly("key")) + .first { it is StoreReadResponse.Data } + assertEquals(listOf("key" to "A"), fixture.posted) + + assertEquals( + StoreWriteResponse.Success.Typed("ack:B"), + fixture.store.write(fixture.request("B", 2L)), + ) + assertEquals(listOf("key" to "A", "key" to "B"), fixture.posted) + assertEquals( + listOf>( + "A" to StoreWriteResponse.Success.Typed("ack:B"), + "B" to StoreWriteResponse.Success.Typed("ack:B"), + ), + fixture.successes, + ) + } +} diff --git a/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreConcurrencyTest.kt b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreConcurrencyTest.kt index 2a2b643ca..3a5de6531 100644 --- a/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreConcurrencyTest.kt +++ b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreConcurrencyTest.kt @@ -3,6 +3,7 @@ package org.mobilenativefoundation.store.store5.mutablestore import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.async import kotlinx.coroutines.awaitAll @@ -27,11 +28,10 @@ import kotlin.test.assertTrue /** * Regression test for a data race in [RealMutableStore]'s per-key write-request queue. * - * The queue is a non-thread-safe `ArrayDeque`. Mutating access goes through - * `withWriteRequestQueueLock`, which historically guarded it with a shared/reader lock that lets - * multiple holders run concurrently. As a result two operations on the same key could run at once: - * `addWriteRequestToQueue` doing `add(...)` while `updateWriteRequestQueue` iterates the same deque - * (`for (writeRequest in this)`). A structural `add` during iteration corrupts the backing array. + * The queue is a non-thread-safe `ArrayDeque`. Historically, a shared/reader lock allowed + * admission to mutate it during acknowledgment iteration, corrupting its backing array. + * A later queue-replacement race could also lose admitted writes. This workload requires every + * write to complete successfully while exercising admission and acknowledgment across threads. * * On Kotlin/Native this surfaces as `EXC_BAD_ACCESS` (a hard process crash). On the JVM the deque's * fail-fast iterator throws `ConcurrentModificationException`, which `RealMutableStore` catches and @@ -43,9 +43,10 @@ import kotlin.test.assertTrue */ @OptIn(ExperimentalCoroutinesApi::class, ExperimentalStoreApi::class) class MutableStoreConcurrencyTest { - private fun newMutableStore(): RealMutableStore { + private fun newMutableStore(scope: CoroutineScope): RealMutableStore { val delegate: RealStore = testStore( + scope = scope, fetcher = TestFetcher(), sourceOfTruth = null, converter = TestConverter(), @@ -63,21 +64,21 @@ class MutableStoreConcurrencyTest { @Test fun sequentialWritesToSameKey_allSucceed() = runTest { - val mutableStore = newMutableStore() + val mutableStore = newMutableStore(backgroundScope) val key = "key" val responses = (1..500).map { i -> mutableStore.write(StoreWriteRequest.of(key = key, value = i)) } - val failures = responses.filterIsInstance() + val failures = responses.filterNot { it is StoreWriteResponse.Success } assertTrue( failures.isEmpty(), "Baseline sequential writes should all succeed, but ${failures.size} failed" + - (failures.firstOrNull()?.let { ", first error = ${it.error}" } ?: ""), + (failures.firstOrNull()?.let { ", first error = $it" } ?: ""), ) } @Test fun concurrentWritesToSameKey_doNotCorruptWriteQueue() = runTest { - val mutableStore = newMutableStore() + val mutableStore = newMutableStore(backgroundScope) val key = "key" val concurrentWriters = 64 val rounds = 50 @@ -96,30 +97,11 @@ class MutableStoreConcurrencyTest { .awaitAll() } - // A corrupted ArrayDeque surfaces as a memory-safety symptom: ConcurrentModificationException, - // NullPointerException, or IndexOutOfBoundsException on the JVM (EXC_BAD_ACCESS aborts the - // process on Native, so reaching this assertion at all already proves no native crash). - // NOTE: concurrent writes to the SAME key can still legitimately fail with - // IllegalArgumentException("No writes found ...") — a separate, pre-existing logical race - // where one write drains another's queue entry. That is not memory corruption and is out of - // scope for this fix, so it is tolerated here. - val corruption = - responses - .filterIsInstance() - .filter { response -> - when (response.error) { - is ConcurrentModificationException, - is NullPointerException, - is IndexOutOfBoundsException, - -> true - else -> false - } - } + val failures = responses.filterNot { it is StoreWriteResponse.Success } assertTrue( - corruption.isEmpty(), - "Write-queue memory corruption in round $round: ${corruption.size}/${responses.size} " + - "writes hit a corruption-class error" + - (corruption.firstOrNull()?.let { ", first = ${it.error}" } ?: ""), + failures.isEmpty(), + "Write failure in round $round: ${failures.size}/${responses.size} writes failed" + + (failures.firstOrNull()?.let { ", first = $it" } ?: ""), ) } } diff --git a/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreLifecycleTest.kt b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreLifecycleTest.kt new file mode 100644 index 000000000..02b3d983e --- /dev/null +++ b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/MutableStoreLifecycleTest.kt @@ -0,0 +1,633 @@ +@file:OptIn(kotlinx.coroutines.ExperimentalCoroutinesApi::class, org.mobilenativefoundation.store.core5.ExperimentalStoreApi::class) + +package org.mobilenativefoundation.store.store5.mutablestore + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineStart +import kotlinx.coroutines.async +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.emitAll +import kotlinx.coroutines.flow.first +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runCurrent +import kotlinx.coroutines.test.runTest +import org.mobilenativefoundation.store.cache5.CacheBuilder +import org.mobilenativefoundation.store.store5.Bookkeeper +import org.mobilenativefoundation.store.store5.Fetcher +import org.mobilenativefoundation.store.store5.OnUpdaterCompletion +import org.mobilenativefoundation.store.store5.SourceOfTruth +import org.mobilenativefoundation.store.store5.StoreReadRequest +import org.mobilenativefoundation.store.store5.StoreWriteRequest +import org.mobilenativefoundation.store.store5.StoreWriteResponse +import org.mobilenativefoundation.store.store5.Updater +import org.mobilenativefoundation.store.store5.UpdaterResult +import org.mobilenativefoundation.store.store5.impl.OnStoreWriteCompletion +import org.mobilenativefoundation.store.store5.impl.RealMutableStore +import org.mobilenativefoundation.store.store5.mutablestore.util.TestConverter +import org.mobilenativefoundation.store.store5.mutablestore.util.TestLogger +import org.mobilenativefoundation.store.store5.mutablestore.util.TestValidator +import org.mobilenativefoundation.store.store5.mutablestore.util.testStore +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertIs +import kotlin.test.assertNull +import kotlin.test.assertTrue + +class MutableStoreLifecycleTest { + @Test + fun cancellingWhileWaitingForLocalMutexHasNoEffects() = + runTest { + val fixture = LifecycleFixture(this) + val gate = LifecycleGate() + fixture.beforeWrite = { _, value -> if (value == "A") gate.pause() } + val first = async { fixture.write("A") } + gate.entered.await() + val cancelled = async { fixture.write("B") } + runCurrent() + cancelled.cancelAndJoin() + assertTrue(fixture.persisted.isEmpty()) + assertTrue(fixture.posted.isEmpty()) + gate.open() + assertIs(first.await()) + assertIs(fixture.write("C")) + assertEquals(listOf("A", "C"), fixture.persisted.map { it.second }) + assertEquals(listOf("A", "C"), fixture.completed) + } + + @Test + fun cancellingInsideLocalWriterDoesNotAdmitAndKeyRecovers() = + runTest { + val fixture = LifecycleFixture(this) + val gate = LifecycleGate() + fixture.beforeWrite = { _, value -> if (value == "A") gate.pause() } + val cancelled = async { fixture.write("A") } + gate.entered.await() + cancelled.cancelAndJoin() + assertTrue(fixture.posted.isEmpty()) + assertTrue(fixture.completed.isEmpty()) + assertIs(fixture.write("B")) + assertEquals(listOf("B"), fixture.persisted.map { it.second }) + assertEquals(listOf("B"), fixture.completed) + } + + @Test + fun cancellingAfterAdmissionRetainsTheWaitingWrite() = + runTest { + val fixture = LifecycleFixture(this) + val gate = LifecycleGate() + fixture.post = { _, value -> + if (value == "A") gate.pause() + UpdaterResult.Success.Typed(value) + } + val first = async { fixture.write("A") } + gate.entered.await() + val cancelled = async { fixture.write("B") } + runCurrent() + assertEquals("B", fixture.local.value["k"]) + cancelled.cancelAndJoin() + gate.open() + first.await() + fixture.write("C") + assertEquals(listOf("A", "C"), fixture.posted.map { it.second }) + assertEquals(listOf("A", "B", "C"), fixture.completed) + } + + @Test + fun cancellingUpdaterRetainsPendingWorkAndExistingMarker() = + runTest { + val fixture = LifecycleFixture(this) + fixture.marker = 7L + val gate = LifecycleGate() + fixture.post = { _, _ -> + gate.pause() + UpdaterResult.Success.Typed("unused") + } + val cancelled = async { fixture.write("A") } + gate.entered.await() + cancelled.cancelAndJoin() + assertEquals(7L, fixture.marker) + assertEquals(0, fixture.clears) + assertTrue(fixture.completed.isEmpty()) + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + fixture.write("B") + assertEquals(listOf("A", "B"), fixture.completed) + assertNull(fixture.marker) + } + + @Test + fun cancellingBeforeAcknowledgementLeavesBatchRetryable() = + runTest { + val gate = LifecycleGate() + var pause = true + val fixture = LifecycleFixture(this, beforeAcknowledgement = { if (pause) gate.pause() }) + val cancelled = async { fixture.write("A") } + gate.entered.await() + cancelled.cancelAndJoin() + assertTrue(fixture.completed.isEmpty()) + assertEquals(0, fixture.clears) + pause = false + fixture.write("B") + assertEquals(listOf("A", "B"), fixture.completed) + assertEquals(listOf("A", "B"), fixture.posted.map { it.second }) + } + + @Test + fun cancellingClearPreservesEarnedCallbacksAndWaitingCallersCachedSuccess() = + runTest { + val fixture = LifecycleFixture(this) + val postGate = LifecycleGate() + val clearGate = LifecycleGate() + fixture.post = { _, value -> + if (value == "A") { + postGate.pause() + UpdaterResult.Error.Message("retry") + } else { + UpdaterResult.Success.Typed("saved") + } + } + fixture.beforeClear = { clearGate.pause() } + val first = async { fixture.write("A") } + postGate.entered.await() + val cancelled = async { fixture.write("B") } + val waiting = async { fixture.write("C") } + runCurrent() + postGate.open() + clearGate.entered.await() + assertIs(first.await()) + cancelled.cancelAndJoin() + assertTrue(cancelled.isCancelled) + assertEquals(StoreWriteResponse.Success.Typed("saved"), waiting.await()) + assertEquals(listOf("A", "B", "C"), fixture.completed) + assertEquals(listOf("A", "C"), fixture.posted.map { it.second }) + fixture.beforeClear = {} + assertIs(fixture.write("D")) + } + + @Test + fun thrownUpdaterExceptionPreservesItsCauseAndPendingWork() = + runTest { + val fixture = LifecycleFixture(this) + val failure = IllegalArgumentException("updater failed") + fixture.post = { _, _ -> throw failure } + val response = fixture.write("A") + val returnedError = assertIs(response).error + assertIs(returnedError) + assertEquals(failure.message, returnedError.message) + assertTrue(generateSequence(returnedError) { it.cause }.any { it === failure }) + assertTrue(fixture.completed.isEmpty()) + assertTrue(fixture.marker != null) + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + assertIs(fixture.write("B")) + assertEquals(listOf("A", "B"), fixture.completed) + assertNull(fixture.marker) + } + + @Test + fun thrownAndReturnedUpdaterCancellationPropagateAndRemainRetryable() = + runTest { + for (returned in listOf(false, true)) { + val fixture = LifecycleFixture(this) + fixture.post = { _, _ -> + val cancellation = CancellationException("updater") + if (returned) UpdaterResult.Error.Exception(cancellation) else throw cancellation + } + assertFailsWith { fixture.write("A") } + assertTrue(fixture.completed.isEmpty()) + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + fixture.write("B") + assertEquals(listOf("A", "B"), fixture.completed) + } + } + + @Test + fun activeJobLocalCancellationPropagatesWithoutAdmission() = + runTest { + val fixture = LifecycleFixture(this) + fixture.beforeWrite = { _, _ -> throw CancellationException("local") } + assertFailsWith { fixture.write("A") } + assertTrue(fixture.posted.isEmpty()) + fixture.beforeWrite = { _, _ -> } + fixture.write("B") + assertEquals(listOf("B"), fixture.completed) + } + + @Test + fun updaterRejectsRecursiveWriteReadAndBothClearsBeforeEffects() = + runTest { + for (operation in listOf("write", "read", "clearKey", "clearAll")) { + val fixture = LifecycleFixture(this) + var checked = false + fixture.post = { _, value -> + fixture.assertRejected(operation) + checked = true + UpdaterResult.Success.Typed(value) + } + assertIs(fixture.write("A")) + assertTrue(checked) + assertEquals(listOf("k" to "A"), fixture.persisted) + assertEquals(1, fixture.posted.size) + assertEquals(0, fixture.deletes) + } + } + + @Test + fun sourceWriterAndLatestReaderRejectRecursiveWritesBeforeEffects() = + runTest { + val fixture = LifecycleFixture(this) + var writerChecks = 0 + fixture.beforeWrite = { _, _ -> + fixture.assertRejected("write") + writerChecks++ + } + fixture.write("A") + assertEquals(1, writerChecks) + assertEquals(listOf("k" to "A"), fixture.persisted) + fixture.beforeWrite = { _, _ -> } + fixture.marker = 1L + fixture.cache.invalidate("k") + var readerChecks = 0 + fixture.beforeRead = { + fixture.beforeRead = {} + fixture.assertRejected("write") + readerChecks++ + } + fixture.read() + assertTrue(readerChecks > 0) + assertEquals(listOf("k" to "A"), fixture.persisted) + assertEquals(listOf("A", "A"), fixture.posted.map { it.second }) + } + + @Test + fun bookkeeperAdaptersRejectRecursiveOperationsBeforeEffects() = + runTest { + for (adapter in listOf("get", "set", "clear")) { + val fixture = LifecycleFixture(this) + var checks = 0 + val check: suspend () -> Unit = { + fixture.assertRejected("write") + fixture.assertRejected("read") + fixture.assertRejected("clearKey") + fixture.assertRejected("clearAll") + checks++ + } + when (adapter) { + "get" -> fixture.beforeGet = check + "set" -> { + fixture.beforeSet = check + fixture.post = { _, _ -> UpdaterResult.Error.Message("failed") } + } + "clear" -> fixture.beforeClear = check + } + fixture.write("A") + if (adapter == "get") fixture.read() + assertTrue(checks > 0, adapter) + assertEquals(listOf("k" to "A"), fixture.persisted) + assertEquals(1, fixture.posted.size) + assertEquals(0, fixture.deletes) + } + } + + @Test + fun inheritedChildAndDifferentKeyCycleCannotReenterAncestorKey() = + runTest { + val fixture = LifecycleFixture(this) + var childChecked = false + var cycleChecked = false + fixture.post = { key, value -> + if (key == "k") { + coroutineScope { + async { + fixture.assertRejected("write") + childChecked = true + }.await() + } + assertIs(fixture.write("J", "j")) + } else { + fixture.assertRejected("write") + cycleChecked = true + } + UpdaterResult.Success.Typed(value) + } + assertIs(fixture.write("A")) + assertTrue(childChecked) + assertTrue(cycleChecked) + assertEquals(listOf("k" to "A", "j" to "J"), fixture.persisted) + assertEquals(2, fixture.posted.size) + } + + @Test + fun anotherStoreWithSameKeyIsAllowed() = + runTest { + val outer = LifecycleFixture(this) + val other = LifecycleFixture(this) + outer.post = { _, value -> + assertIs(other.write("nested")) + UpdaterResult.Success.Typed(value) + } + outer.write("A") + assertEquals(listOf("k" to "nested"), other.persisted) + } + + @Test + fun requestAndUpdaterCallbacksCanCompleteSameKeyWriteBeforeReturning() = + runTest { + for (updaterCallback in listOf(false, true)) { + val fixture = LifecycleFixture(this) + var reentered = false + var provedImmediateCompletion = false + val callback: () -> Unit = { + if (!reentered) { + reentered = true + var nestedResponse: StoreWriteResponse? = null + val nested = async(start = CoroutineStart.UNDISPATCHED) { nestedResponse = fixture.write("nested") } + assertTrue(nested.isCompleted, "Nested write must complete before callback returns") + assertIs(nestedResponse) + assertEquals("nested", fixture.local.value["k"]) + provedImmediateCompletion = true + } + } + if (updaterCallback) fixture.onUpdaterSuccess = callback + val callbacks = if (updaterCallback) emptyList() else listOf(callback) + assertIs(fixture.write("A", callbacks = callbacks)) + assertTrue(reentered) + assertTrue(provedImmediateCompletion) + assertEquals(2, fixture.posted.size) + } + } + + @Test + fun ordinaryCallbackFailuresDoNotChangeSuccessOrSkipRemainingCallbacks() = + runTest { + val fixture = LifecycleFixture(this) + val events = mutableListOf() + fixture.onUpdaterSuccess = { + events.add("updater") + error("updater callback") + } + val response = + fixture.write( + "A", + callbacks = + listOf({ + events.add("first") + error("request callback") + }, { events.add("last") }), + ) + assertIs(response) + assertEquals(listOf("updater", "first", "last"), events) + assertEquals(listOf("A"), fixture.completed) + assertEquals("A", fixture.remote["k"]) + assertTrue(fixture.logger.errorLogs.size >= 2) + fixture.read() + assertEquals(1, fixture.posted.size) + } + + @Test + fun callbackCancellationFinishesEarnedBatchBeforePropagating() = + runTest { + for (updaterCallback in listOf(false, true)) { + val fixture = LifecycleFixture(this) + val events = mutableListOf() + fixture.post = { _, _ -> UpdaterResult.Error.Message("retry") } + fixture.write( + "A", + callbacks = + listOf({ + events.add("first") + throw CancellationException("request callback") + }, { events.add("last") }), + ) + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + fixture.onUpdaterSuccess = { + events.add("updater") + if (updaterCallback) throw CancellationException("updater callback") + } + assertFailsWith { + fixture.write("B", callbacks = listOf({ events.add("lastB") })) + } + assertEquals(listOf("updater", "first", "last", "updater", "lastB"), events) + assertEquals(listOf("A", "B"), fixture.completed) + assertEquals("B", fixture.remote["k"]) + fixture.onUpdaterSuccess = {} + fixture.read() + assertEquals(2, fixture.posted.size) + assertIs(fixture.write("C")) + } + } + + @Test + fun bookkeepingFalseAndOrdinaryFailuresPreserveOutcomesAndRecoverability() = + runTest { + for (failure in listOf("setFalse", "setThrow", "clearFalse", "clearThrow", "getThrow")) { + val fixture = LifecycleFixture(this) + val failedUpdate = UpdaterResult.Error.Message("original failure") + when (failure) { + "setFalse" -> { + fixture.setResult = false + fixture.post = { _, _ -> failedUpdate } + } + "setThrow" -> { + fixture.beforeSet = { error("set failure") } + fixture.post = { _, _ -> failedUpdate } + } + "clearFalse" -> fixture.clearResult = false + "clearThrow" -> fixture.beforeClear = { error("clear failure") } + "getThrow" -> fixture.beforeGet = { error("get failure") } + } + val response = fixture.write("A") + if (failure.startsWith("set")) { + assertEquals(StoreWriteResponse.Error.Message("original failure"), response) + assertTrue(fixture.completed.isEmpty()) + } else { + assertIs(response) + assertEquals(listOf("A"), fixture.completed) + } + if (failure == "getThrow") fixture.read() + if (failure != "clearFalse") assertTrue(fixture.logger.errorLogs.isNotEmpty(), failure) + fixture.beforeGet = {} + fixture.beforeSet = {} + fixture.beforeClear = {} + fixture.setResult = true + fixture.clearResult = true + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + assertIs(fixture.write("B")) + assertEquals(listOf("A", "B"), fixture.completed) + assertTrue(fixture.failedCallbacks.isEmpty()) + } + } + + @Test + fun cancellingEagerPostPreservesPendingCallbacksAndMarker() = + runTest { + val fixture = LifecycleFixture(this) + fixture.post = { _, _ -> UpdaterResult.Error.Message("retry") } + fixture.write("A") + val marker = fixture.marker + val gate = LifecycleGate() + fixture.post = { _, value -> + gate.pause() + UpdaterResult.Success.Typed(value) + } + val reader = async { fixture.read() } + gate.entered.await() + reader.cancelAndJoin() + assertEquals(marker, fixture.marker) + assertTrue(fixture.completed.isEmpty()) + assertEquals(0, fixture.clears) + fixture.post = { _, value -> UpdaterResult.Success.Typed(value) } + fixture.read() + assertEquals(listOf("A"), fixture.completed) + assertNull(fixture.marker) + } + + @Test + fun eagerBookkeeperCancellationIsNotSwallowed() = + runTest { + val fixture = LifecycleFixture(this) + fixture.write("A") + fixture.beforeGet = { throw CancellationException("lookup") } + assertFailsWith { fixture.read() } + fixture.beforeGet = {} + fixture.read() + assertEquals(1, fixture.posted.size) + } +} + +private class LifecycleGate { + val entered = CompletableDeferred() + private val released = CompletableDeferred() + + suspend fun pause() { + entered.complete(Unit) + released.await() + } + + fun open() { + released.complete(Unit) + } +} + +private class LifecycleFixture(scope: TestScope, beforeAcknowledgement: suspend () -> Unit = {}) { + val local = MutableStateFlow>(emptyMap()) + val persisted = mutableListOf>() + val posted = mutableListOf>() + val remote = mutableMapOf() + val completed = mutableListOf() + val failedCallbacks = mutableListOf() + val logger = TestLogger() + val cache = CacheBuilder().build() + var marker: Long? = null + var clears = 0 + var deletes = 0 + var setResult = true + var clearResult = true + var beforeWrite: suspend (String, String) -> Unit = { _, _ -> } + var beforeRead: suspend () -> Unit = {} + var beforeGet: suspend () -> Unit = {} + var beforeSet: suspend () -> Unit = {} + var beforeClear: suspend () -> Unit = {} + var post: suspend (String, String) -> UpdaterResult = { _, value -> UpdaterResult.Success.Typed(value) } + var onUpdaterSuccess: () -> Unit = {} + val store = + RealMutableStore( + delegate = + testStore( + dispatcher = StandardTestDispatcher(scope.testScheduler), + scope = scope.backgroundScope, + fetcher = Fetcher.of { _: String -> error("Unexpected fetch") }, + sourceOfTruth = + SourceOfTruth.of( + reader = { key: String -> + flow { + beforeRead() + emitAll(local.map { it[key] }) + } + }, + writer = { key: String, value: String -> + beforeWrite(key, value) + local.value = local.value + (key to value) + persisted.add(key to value) + }, + delete = { _: String -> deletes++ }, + deleteAll = { deletes++ }, + ), + converter = TestConverter(), + validator = TestValidator(), + memoryCache = cache, + ), + updater = + Updater.by( + post = { key, value -> + posted.add(key to value) + post(key, value).also { result -> if (result is UpdaterResult.Success) remote[key] = value } + }, + onCompletion = OnUpdaterCompletion(onSuccess = { onUpdaterSuccess() }, onFailure = {}), + ), + bookkeeper = + object : Bookkeeper { + override suspend fun getLastFailedSync(key: String): Long? { + beforeGet() + return marker + } + + override suspend fun setLastFailedSync(key: String, timestamp: Long): Boolean { + beforeSet() + if (setResult) marker = timestamp + return setResult + } + + override suspend fun clear(key: String): Boolean { + clears++ + beforeClear() + if (clearResult) marker = null + return clearResult + } + + override suspend fun clearAll(): Boolean { + marker = null + return true + } + }, + logger = logger, + beforeAcknowledgementCommit = beforeAcknowledgement, + ) + + suspend fun write(value: String, key: String = "k", callbacks: List<() -> Unit> = emptyList()): StoreWriteResponse = + store.write( + StoreWriteRequest.of( + key = key, + value = value, + onCompletions = + listOf(OnStoreWriteCompletion(onSuccess = { completed.add(value) }, onFailure = { failedCallbacks.add(it) })) + + callbacks.map { callback -> OnStoreWriteCompletion(onSuccess = { callback() }, onFailure = {}) }, + ), + ) + + suspend fun read() = store.stream(StoreReadRequest.localOnly("k")).first() + + suspend fun assertRejected(operation: String) { + val error = + when (operation) { + "write" -> { + try { + val response = write("recursive") + assertIs(response).error + } catch (error: IllegalStateException) { + error + } + } + "read" -> assertFailsWith { read() } + "clearKey" -> assertFailsWith { store.clear("k") } + else -> assertFailsWith { store.clear() } + } + assertIs(error) + assertTrue(error.message.orEmpty().contains("Recursive")) + } +} diff --git a/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/util/MutableStoreRaceFixture.kt b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/util/MutableStoreRaceFixture.kt new file mode 100644 index 000000000..58d78511d --- /dev/null +++ b/store/src/commonTest/kotlin/org/mobilenativefoundation/store/store5/mutablestore/util/MutableStoreRaceFixture.kt @@ -0,0 +1,128 @@ +@file:OptIn( + kotlinx.coroutines.ExperimentalCoroutinesApi::class, + org.mobilenativefoundation.store.core5.ExperimentalStoreApi::class, +) + +package org.mobilenativefoundation.store.store5.mutablestore.util + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.test.StandardTestDispatcher +import kotlinx.coroutines.test.TestScope +import org.mobilenativefoundation.store.cache5.CacheBuilder +import org.mobilenativefoundation.store.store5.Bookkeeper +import org.mobilenativefoundation.store.store5.Fetcher +import org.mobilenativefoundation.store.store5.OnUpdaterCompletion +import org.mobilenativefoundation.store.store5.SourceOfTruth +import org.mobilenativefoundation.store.store5.StoreWriteRequest +import org.mobilenativefoundation.store.store5.StoreWriteResponse +import org.mobilenativefoundation.store.store5.Updater +import org.mobilenativefoundation.store.store5.UpdaterResult +import org.mobilenativefoundation.store.store5.impl.OnStoreWriteCompletion +import org.mobilenativefoundation.store.store5.impl.RealMutableStore + +internal class RaceGate { + val entered = CompletableDeferred() + val released = CompletableDeferred() + + suspend fun pause() { + entered.complete(Unit) + released.await() + } + + fun open() { + released.complete(Unit) + } +} + +internal class MutableStoreRaceFixture( + scope: TestScope, + post: suspend (String, String) -> UpdaterResult, + beforeLocalWrite: suspend (String, String) -> Unit = { _, _ -> }, + beforeLocalReturn: suspend (String, String) -> Unit = { _, _ -> }, + beforeAcknowledgementCommit: suspend () -> Unit = {}, + val bookkeeper: Bookkeeper = TestInMemoryBookkeeper(), + withBookkeeper: Boolean = true, +) { + val local = MutableStateFlow>(emptyMap()) + val localAttempts = mutableListOf>() + val localWrites = mutableListOf>() + val posted = mutableListOf>() + val activePosts = mutableMapOf() + val maximumActivePosts = mutableMapOf() + val remote = mutableMapOf() + val successes = mutableListOf>() + val failures = mutableListOf>() + val updaterSuccesses = mutableListOf() + val updaterFailures = mutableListOf() + val logger = TestLogger() + val cache = CacheBuilder().build() + + val store = + RealMutableStore( + delegate = + testStore( + dispatcher = StandardTestDispatcher(scope.testScheduler), + scope = scope.backgroundScope, + fetcher = Fetcher.of { _: String -> error("Unexpected fetch") }, + sourceOfTruth = + SourceOfTruth.of( + reader = { key: String -> local.map { it[key] } }, + writer = { key: String, value: String -> + localAttempts.add(key to value) + beforeLocalWrite(key, value) + local.value = local.value + (key to value) + localWrites.add(key to value) + beforeLocalReturn(key, value) + }, + ), + converter = TestConverter(), + validator = TestValidator(), + memoryCache = cache, + ), + updater = + Updater.by( + post = { key, value -> + posted.add(key to value) + val active = (activePosts[key] ?: 0) + 1 + activePosts[key] = active + maximumActivePosts[key] = maxOf(maximumActivePosts[key] ?: 0, active) + try { + post(key, value).also { result -> + if (result is UpdaterResult.Success) remote[key] = value + } + } finally { + activePosts[key] = (activePosts[key] ?: 1) - 1 + } + }, + onCompletion = + OnUpdaterCompletion( + onSuccess = { updaterSuccesses.add(it) }, + onFailure = { updaterFailures.add(it) }, + ), + ), + bookkeeper = bookkeeper.takeIf { withBookkeeper }, + logger = logger, + beforeAcknowledgementCommit = beforeAcknowledgementCommit, + ) + + fun request( + value: String, + created: Long, + id: String = value, + key: String = "key", + ): StoreWriteRequest = + StoreWriteRequest.of( + key = key, + value = value, + created = created, + onCompletions = + listOf( + OnStoreWriteCompletion( + onSuccess = { successes.add(id to it) }, + onFailure = { failures.add(id to it) }, + ), + ), + ) +}