Keep live observations in sync with entity removals

client/README.md

@@ -54,6 +54,12 @@ complete, which recomposes the Compose code that reads it. Creating an
 observation therefore never blocks the UI thread, even when the server is
 unreachable.
 
+Observations remove entities when the server reports that they were archived,
+deleted, or no longer match the subscription filter. A single-entity observation
+asynchronously rereads its query after a removal and uses the first remaining
+match, or `null` or its supplied default when none remain. A later matching
+update makes the entity available again.
+
 A new observation carries the `DataObservationStatus.Refreshing` status while
 its initial read is in progress. If the connection is already known to be
 unavailable, the observation is instead returned in

client/build.gradle.kts

@@ -1,5 +1,5 @@
 /*
- * Copyright 2025, TeamDev. All rights reserved.
+ * Copyright 2026, TeamDev. All rights reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -24,16 +24,32 @@
  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  */
 
+import com.google.protobuf.gradle.protobuf
+import com.google.protobuf.gradle.protoc
 import io.spine.internal.dependency.Kotlin
 import io.spine.internal.dependency.KotlinX
 import io.spine.internal.dependency.Material3
+import io.spine.internal.dependency.Protobuf
 import io.spine.internal.dependency.Spine
 
 plugins {
     id("io.spine.tools.gradle.bootstrap")
+    id("com.google.protobuf")
     id("org.jetbrains.compose") version "1.5.12"
 }
 
+// Generates entity fixtures for the client protocol tests.
+spine {
+    assembleModel()
+    enableJava()
+}
+
+protobuf {
+    protoc {
+        artifact = Protobuf.compiler
+    }
+}
+
 dependencies {
     implementation(Kotlin.reflect)
     implementation(KotlinX.Coroutines.swing)

client/src/main/kotlin/io/spine/chords/client/Client.kt

@@ -67,6 +67,7 @@ public interface Client {
     /**
      * Reads the list of entities with the [entityClass] class and returns an
      * observation that maintains an up-to-date list.
+     * Archived and deleted entities are removed when the server reports them.
      *
      * This function returns without waiting for the server. The observation is
      * returned with an empty list, and the list read from the server appears in
@@ -97,7 +98,8 @@ public interface Client {
      * Reads all entities of type [entityClass] that match the given
      * [queryFilter]. Then sets up observation to receive future updates to the
      * entities, filtering the observed updates using the provided
-     * [observeFilter].
+     * [observeFilter]. Entities are removed when they are archived, deleted,
+     * or stop matching [observeFilter]. A later matching update adds them again.
      *
      * This function returns without waiting for the server. The observation is
      * returned with an empty list, and the list read from the server appears in
@@ -138,6 +140,8 @@ public interface Client {
      *   or [observeFilter], the returned observation gets the first matching
      *   value.
      * - If no entries match the specified criteria, the value is `null`.
+     * - An archive, deletion, or removal from [observeFilter] triggers an asynchronous
+     *   reread. The value becomes the first remaining match, or `null` if none remain.
      *
      * This function returns without waiting for the server. The observation is
      * returned with a `null` value, and the value read from the server appears
@@ -173,6 +177,8 @@ public interface Client {
      *
      * This overload guarantees a non-null value by using [defaultValue] when no
      * entity matches. If several entities match, the first one is used.
+     * An archive, deletion, or removal from [observeFilter] triggers an asynchronous
+     * reread. The value becomes the first remaining match, or [defaultValue] if none remain.
      *
      * This function returns without waiting for the server. The observation is
      * returned with [defaultValue], and the value read from the server appears

client/src/main/kotlin/io/spine/chords/client/DataObservation.kt

@@ -88,10 +88,11 @@ public class DataObservation<out T> internal constructor(
      */
     private val read: () -> T,
     /**
-     * Creates a server subscription and registers its update and failure callbacks.
+     * Registers state updates, query invalidations, and failures for a server subscription.
      */
     private val subscribe: (
         onUpdate: ((T) -> T) -> Unit,
+        onInvalidated: () -> Unit,
         onError: (Throwable) -> Unit
     ) -> ObservationSubscription,
     /**
@@ -109,7 +110,11 @@ public class DataObservation<out T> internal constructor(
     /**
      * Requests a delayed retry when a stream fails on a connected channel.
      */
-    private val onRecoveryNeeded: (DataObservation<*>) -> Unit = {}
+    private val onRecoveryNeeded: (DataObservation<*>) -> Unit = {},
+    /**
+     * Schedules a reread only while the invalidated subscription generation is current.
+     */
+    private val onRefreshNeeded: (DataObservation<*>, Long) -> Unit = { _, _ -> }
 ) : State<T> {
 
     /**
@@ -177,12 +182,26 @@ public class DataObservation<out T> internal constructor(
      * from this function. Coroutine cancellation is propagated to the caller
      * without being converted into an observation failure.
      */
+    public suspend fun refresh() {
+        refresh(expectedGeneration = null)
+    }
+
+    /**
+     * Rereads an invalidated query unless a lifecycle change has superseded its subscription.
+     */
+    internal suspend fun refreshIfCurrent(expectedGeneration: Long) {
+        refresh(expectedGeneration = expectedGeneration)
+    }
+
+    /**
+     * Serializes refreshes and optionally limits a request to its originating subscription.
+     */
     @Suppress(
         "ReturnCount" /* Each failed or stale recovery phase must stop immediately. */
     )
-    public suspend fun refresh() {
+    private suspend fun refresh(expectedGeneration: Long?) {
         refreshMutex.withLock {
-            val refreshGeneration = beginRefresh() ?: return
+            val refreshGeneration = beginRefresh(expectedGeneration) ?: return
             val pendingUpdates = PendingUpdates<T>()
             try {
                 val result = withContext(requestContext) {
@@ -252,6 +271,7 @@ public class DataObservation<out T> internal constructor(
                         update
                     )
                 },
+                { onRefreshNeeded(this, refreshGeneration) },
                 { error ->
                     bufferOrHandleFailure(
                         refreshGeneration,
@@ -379,12 +399,13 @@ public class DataObservation<out T> internal constructor(
 
     /**
      * Starts a new generation and detaches the previous subscription.
+     * An invalidation must still belong to [expectedGeneration] when it starts its refresh.
      */
-    private fun beginRefresh(): Long? {
+    private fun beginRefresh(expectedGeneration: Long?): Long? {
         val previousSubscription: ObservationSubscription?
         val refreshGeneration: Long
         synchronized(stateLock) {
-            if (cancelled) {
+            if (cancelled || (expectedGeneration != null && generation != expectedGeneration)) {
                 return null
             }
             generation++
@@ -623,7 +644,7 @@ internal fun <T, U> createDataObservation(
 ): DataObservation<T> = DataObservation(
     initialValue,
     read,
-    { onUpdate, onError ->
+    { onUpdate, _, onError ->
         subscribe(
             { update ->
                 onUpdate { value -> applyUpdate(value, update) }

client/src/main/kotlin/io/spine/chords/client/DataObservationScope.kt

@@ -33,8 +33,8 @@ import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers.IO
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 
@@ -316,14 +316,19 @@ internal class DataObservationScope(
 
     /**
      * Refreshes the given [observation] in this coroutine scope.
+     * An invalidation supplies [expectedGeneration] to skip superseded subscriptions.
      *
      * @return The job that performs the refresh. It completes without running
      *   the refresh if this scope has been closed meanwhile, because [close]
      *   cancels the scope.
      */
-    private fun refresh(observation: DataObservation<*>): Job =
+    fun refresh(observation: DataObservation<*>, expectedGeneration: Long? = null): Job =
         coroutineScope.launch {
-            observation.refresh()
+            if (expectedGeneration == null) {
+                observation.refresh()
+            } else {
+                observation.refreshIfCurrent(expectedGeneration)
+            }
         }
 
     /**