Merge branch '181-query-publisher-crash' into 'dev'

QueryPublisher: wait for the query to finish on close #181

See merge request objectbox/objectbox-java!159

Author: greenrobot-team

CHANGELOG.md

@@ -7,6 +7,7 @@ For more insights into what changed in the ObjectBox C++ core, [check the Object
 ## 4.3.1 - in development
 
 - Requires at least Kotlin compiler and standard library 1.7.
+- Data Observers: closing a Query now waits on a running publisher to finish its query, preventing a VM crash. [#1147](https://github.com/objectbox/objectbox-java/issues/1147)
 
 ## 4.3.0 - 2025-05-13
 

objectbox-java/src/main/java/io/objectbox/query/InternalAccess.java

@@ -29,4 +29,12 @@ public static <T> void nativeFindFirst(Query<T> query, long cursorHandle) {
         query.nativeFindFirst(query.handle, cursorHandle);
     }
 
+    /**
+     * See {@link QueryPublisher#LOG_STATES}.
+     */
+    @Internal
+    public static void queryPublisherLogStates() {
+        QueryPublisher.LOG_STATES = true;
+    }
+
 }

objectbox-java/src/main/java/io/objectbox/query/Query.java

@@ -175,6 +175,7 @@ protected void finalize() throws Throwable {
      * Calling any other methods of this afterwards will throw an {@link IllegalStateException}.
      */
     public synchronized void close() {
+        publisher.stopAndAwait();  // Ensure it is done so that the query is not used anymore
         if (handle != 0) {
             // Closeable recommendation: mark as "closed" before nativeDestroy could throw.
             long handleCopy = handle;
@@ -939,6 +940,7 @@ public long remove() {
      * See {@link SubscriptionBuilder#observer(DataObserver)} for additional details.
      *
      * @return A {@link SubscriptionBuilder} to build a subscription.
+     * @see #publish()
      */
     public SubscriptionBuilder<List<T>> subscribe() {
         checkOpen();
@@ -958,11 +960,15 @@ public SubscriptionBuilder<List<T>> subscribe(DataSubscriptionList dataSubscript
     }
 
     /**
-     * Publishes the current data to all subscribed @{@link DataObserver}s.
-     * This is useful triggering observers when new parameters have been set.
-     * Note, that setParameter methods will NOT be propagated to observers.
+     * Manually schedules publishing the current results of this query to all {@link #subscribe() subscribed}
+     * {@link DataObserver observers}, even if the underlying Boxes have not changed.
+     * <p>
+     * This is useful to publish new results after changing parameters of this query which would otherwise not trigger
+     * publishing of new results.
      */
     public void publish() {
+        // Do open check to not silently fail (publisher runnable would just not get scheduled if query is closed)
+        checkOpen();
         publisher.publish();
     }
 

objectbox-java/src/main/java/io/objectbox/query/QueryPublisher.java

@@ -35,20 +35,32 @@
 import io.objectbox.reactive.SubscriptionBuilder;
 
 /**
- * A {@link DataPublisher} that subscribes to an ObjectClassPublisher if there is at least one observer.
- * Publishing is requested if the ObjectClassPublisher reports changes, a subscription is
- * {@link SubscriptionBuilder#observer(DataObserver) observed} or {@link Query#publish()} is called.
- * For publishing the query is re-run and the result delivered to the current observers.
- * Results are published on a single thread, one at a time, in the order publishing was requested.
+ * A {@link DataPublisher} that {@link BoxStore#subscribe(Class) subscribes to the Box} of its associated {@link Query}
+ * while there is at least one observer (see {@link #subscribe(DataObserver, Object)} and
+ * {@link #unsubscribe(DataObserver, Object)}).
+ * <p>
+ * Publishing is requested if the Box reports changes, a subscription is
+ * {@link SubscriptionBuilder#observer(DataObserver) observed} (if {@link #publishSingle(DataObserver, Object)} is
+ * called) or {@link Query#publish()} (calls {@link #publish()}) is called.
+ * <p>
+ * For publishing the query is re-run and the result data is delivered to the current observers.
+ * <p>
+ * Data is passed to observers on a single thread ({@link BoxStore#internalScheduleThread(Runnable)}), one at a time, in
+ * the order observers were added.
  */
 @Internal
 class QueryPublisher<T> implements DataPublisher<List<T>>, Runnable {
 
+    /**
+     * If enabled, logs states of the publisher runnable. Useful to debug a query subscription.
+     */
+    public static boolean LOG_STATES = false;
     private final Query<T> query;
     private final Box<T> box;
     private final Set<DataObserver<List<T>>> observers = new CopyOnWriteArraySet<>();
     private final Deque<DataObserver<List<T>>> publishQueue = new ArrayDeque<>();
     private volatile boolean publisherRunning = false;
+    private volatile boolean publisherStopped = false;
 
     private static class SubscribedObservers<T> implements DataObserver<List<T>> {
         @Override
@@ -106,6 +118,10 @@ void publish() {
      */
     private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
         synchronized (publishQueue) {
+            // Check after obtaining the lock as the publisher may have been stopped while waiting on the lock
+            if (publisherStopped) {
+                return;
+            }
             publishQueue.add(observer);
             if (!publisherRunning) {
                 publisherRunning = true;
@@ -114,6 +130,31 @@ private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
         }
     }
 
+    /**
+     * Marks this publisher as stopped and if it is currently running waits on it to complete.
+     * <p>
+     * After calling this, this publisher will no longer run, even if observers subscribe or publishing is requested.
+     */
+    void stopAndAwait() {
+        publisherStopped = true;
+        // Doing wait/notify waiting here; could also use the Future from BoxStore.internalScheduleThread() instead.
+        // The latter would require another member though, which seems redundant.
+        synchronized (this) {
+            while (publisherRunning) {
+                try {
+                    this.wait();
+                } catch (InterruptedException e) {
+                    if (publisherRunning) {
+                        // When called by Query.close() throwing here will leak the query. But not throwing would allow
+                        // close() to proceed in destroying the native query while it may still be active (run() of this
+                        // is at the query.find() call), which would trigger a VM crash.
+                        throw new RuntimeException("Interrupted while waiting for publisher to finish", e);
+                    }
+                }
+            }
+        }
+    }
+
     /**
      * Processes publish requests for this query on a single thread to prevent
      * older query results getting delivered after newer query results.
@@ -122,9 +163,11 @@ private void queueObserverAndScheduleRun(DataObserver<List<T>> observer) {
      */
     @Override
     public void run() {
+        log("started");
         try {
-            while (true) {
+            while (!publisherStopped) {
                 // Get all queued observer(s), stop processing if none.
+                log("checking for observers");
                 List<DataObserver<List<T>>> singlePublishObservers = new ArrayList<>();
                 boolean notifySubscribedObservers = false;
                 synchronized (publishQueue) {
@@ -143,9 +186,12 @@ public void run() {
                 }
 
                 // Query.
+                log("running query");
+                if (publisherStopped) break;  // Check again to avoid running the query if possible
                 List<T> result = query.find();
 
                 // Notify observer(s).
+                log("notifying observers");
                 for (DataObserver<List<T>> observer : singlePublishObservers) {
                     observer.onData(result);
                 }
@@ -158,8 +204,12 @@ public void run() {
                 }
             }
         } finally {
+            log("stopped");
             // Re-set if wrapped code throws, otherwise this publisher can no longer publish.
             publisherRunning = false;
+            synchronized (this) {
+                this.notifyAll();
+            }
         }
     }
 
@@ -172,4 +222,8 @@ public synchronized void unsubscribe(DataObserver<List<T>> observer, @Nullable O
         }
     }
 
+    private static void log(String message) {
+        if (LOG_STATES) System.out.println("QueryPublisher: " + message);
+    }
+
 }

tests/objectbox-java-test/src/test/java/io/objectbox/AbstractObjectBoxTest.java

@@ -45,6 +45,7 @@
 import io.objectbox.model.ExternalPropertyType;
 import io.objectbox.model.PropertyFlags;
 import io.objectbox.model.PropertyType;
+import io.objectbox.query.InternalAccess;
 
 
 import static org.junit.Assert.assertEquals;
@@ -92,6 +93,7 @@ static void printProcessId() {
     public void setUp() throws IOException {
         Cursor.TRACK_CREATION_STACK = true;
         Transaction.TRACK_CREATION_STACK = true;
+        InternalAccess.queryPublisherLogStates();
 
         // Note: is logged, so create before logging.
         boxStoreDir = prepareTempDir("object-store-test");

tests/objectbox-java-test/src/test/java/io/objectbox/query/QueryObserverTest.java

@@ -235,6 +235,60 @@ public void transform_inOrderOfPublish() {
         assertEquals(2, (int) placing.get(1));
     }
 
+    @Test
+    public void queryCloseWaitsOnPublisher() throws InterruptedException {
+        CountDownLatch beforeBlockPublisher = new CountDownLatch(1);
+        CountDownLatch blockPublisher = new CountDownLatch(1);
+        CountDownLatch beforeQueryClose = new CountDownLatch(1);
+        CountDownLatch afterQueryClose = new CountDownLatch(1);
+
+        AtomicBoolean publisherBlocked = new AtomicBoolean(false);
+        AtomicBoolean waitedBeforeQueryClose = new AtomicBoolean(false);
+
+        new Thread(() -> {
+            Query<TestEntity> query = box.query().build();
+            query.subscribe()
+                    .onlyChanges() // prevent initial publish call
+                    .observer(data -> {
+                beforeBlockPublisher.countDown();
+                try {
+                    publisherBlocked.set(blockPublisher.await(1, TimeUnit.SECONDS));
+                } catch (InterruptedException e) {
+                    throw new RuntimeException("Observer was interrupted while waiting", e);
+                }
+            });
+
+            // Trigger the query publisher, prepare so it runs its loop, incl. the query, at least twice
+            // and block it from completing the first loop using the observer.
+            query.publish();
+            query.publish();
+
+            try {
+                waitedBeforeQueryClose.set(beforeQueryClose.await(1, TimeUnit.SECONDS));
+            } catch (InterruptedException e) {
+                throw new RuntimeException("Thread was interrupted while waiting before closing query", e);
+            }
+            query.close();
+            afterQueryClose.countDown();
+        }).start();
+
+        // Wait for observer to block the publisher
+        assertTrue(beforeBlockPublisher.await(1, TimeUnit.SECONDS));
+        // Start closing the query
+        beforeQueryClose.countDown();
+
+        // While the publisher is blocked, the query close call should block
+        assertFalse(afterQueryClose.await(100, TimeUnit.MILLISECONDS));
+
+        // After the publisher is unblocked and can stop, the query close call should complete
+        blockPublisher.countDown();
+        assertTrue(afterQueryClose.await(100, TimeUnit.MILLISECONDS));
+
+        // Verify latches were triggered due to reaching 0, not due to timeout
+        assertTrue(publisherBlocked.get());
+        assertTrue(waitedBeforeQueryClose.get());
+    }
+
     private void putTestEntitiesScalars() {
         putTestEntities(10, null, 2000);
     }