fix: Prevent watch stream from emitting events after close. #1471

tom-andersen · 2023-11-08T19:53:02Z

b/288270811

tom-andersen · 2023-11-08T19:58:25Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/BulkWriter.java

@@ -916,7 +916,7 @@ private void sendBatchLocked(final BulkCommitBatch batch, final boolean flush) {
              bulkWriterExecutor);
    } else {
      long delayMs = rateLimiter.getNextRequestDelayMs(batch.getMutationsSize());
-      logger.log(Level.FINE, String.format("Backing off for %d seconds", delayMs / 1000));
+      logger.log(Level.FINE, () -> String.format("Backing off for %d seconds", delayMs / 1000));


This isn't directly related to PR. But since I was adding logging...

I noticed this log was building a string, even if log level is not fine. A lambda allows lazy evaluation, such that production environments don't suffer work when logging is set to a more coarse level.

tom-andersen · 2023-11-08T20:06:19Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/WatchTest.java

@@ -84,6 +86,10 @@

 @RunWith(MockitoJUnitRunner.class)
 public class WatchTest {
+
+  @Rule
+  public Timeout timeout = new Timeout(1, TimeUnit.SECONDS);


While working through problem, I introduced bug that made unit tests hang. To make sure tests completed, I added timeout to fail test if they don't run within 1 second.

wu-hui

Please also look into if we can add unit tests for the new stream class.

wu-hui · 2023-11-09T16:29:03Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SuppressibleBidiStream.java

+import java.util.function.Function;
+import java.util.logging.Logger;
+
+final class SuppressibleBidiStream<RequestT, ResponseT>


Please add a class level comment to explain what this class does and why it is neccessary.

wu-hui · 2023-11-09T16:30:03Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SuppressibleBidiStream.java

+
+  private final ClientStream<ListenRequest> stream;
+  private final BidiStreamObserver<RequestT, ResponseT> delegate;
+  private boolean silence = false;


Since the class is now call Suppressbile.., this should be isSuppressed?

wu-hui · 2023-11-09T16:30:29Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SuppressibleBidiStream.java

+    stream.closeSend();
+  }
+
+  public void closeAndSilence() {


Same here, use closeAndSuppress instead.

wu-hui · 2023-11-09T16:30:46Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SuppressibleBidiStream.java

+  @Override
+  public void onReady(ClientStream<RequestT> stream) {
+    if (silence) {
+      LOGGER.info(() -> String.format("Silenced: %s", stream));


Same for the log message.

wu-hui · 2023-11-09T17:03:10Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/Watch.java

-        if (listenResponse.getFilter().getCount() != currentSize()) {
+        int filterCount = listenResponse.getFilter().getCount();
+        int currentSize = currentSize();
+        if (filterCount != currentSize) {


Inlining currentSize() seems better IMO?

currentSize() hides a lot of computation:

private int currentSize() { ChangeSet changeSet = extractChanges(Timestamp.now()); return documentSet.size() + changeSet.adds.size() - changeSet.deletes.size(); }

private ChangeSet extractChanges(Timestamp readTime) { ChangeSet changeSet = new ChangeSet(); for (Entry<ResourcePath, Document> change : changeMap.entrySet()) { if (change.getValue() == null) { if (documentSet.contains(change.getKey())) { changeSet.deletes.add(documentSet.getDocument(change.getKey())); } continue; } QueryDocumentSnapshot snapshot = QueryDocumentSnapshot.fromDocument(firestore, readTime, change.getValue()); if (documentSet.contains(change.getKey())) { changeSet.updates.add(snapshot); } else { changeSet.adds.add(snapshot); } } return changeSet; }

Sorry, I meant if (filterCount != currentSize())

wu-hui · 2023-11-09T17:56:35Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITQueryWatchTest.java

+      listener.eventsCountDownLatch.awaitInitialEvents();
+      listener
+          .assertionsForLastEvent()
+          .noError()


Add more comments explaining why we have this test and what would happen if SuppressiableStream is not used.

tom-andersen · 2023-11-09T21:09:05Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/it/ITQueryWatchTest.java

+
+      // A race condition will sometimes throw an error if the SuppressibleBidiStream does not
+      // silence the old stream. This can be caused by `Preconditions.checkState(stream == null)`
+      // in Watch class.


Explanation added here.

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

wu-hui

Some minor nits, but LGTM in general, thanks!

wu-hui · 2023-11-13T22:03:06Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/SilenceableBidiStream.java

+ * allowed to send `onError` or `onComplete` since that would signal the downstream that the stream
+ * is finished.
+ *
+ * @param <RequestT>


Don't think these @params are useful here.

wu-hui · 2023-11-13T22:04:04Z

google-cloud-firestore/src/main/java/com/google/cloud/firestore/Watch.java

-        if (listenResponse.getFilter().getCount() != currentSize()) {
+        int filterCount = listenResponse.getFilter().getCount();
+        int currentSize = currentSize();
+        if (filterCount != currentSize) {


Sorry, I meant if (filterCount != currentSize())

wu-hui · 2023-11-13T22:04:52Z

google-cloud-firestore/src/test/java/com/google/cloud/firestore/SilenceableBidiStreamTest.java

+
+  @Mock ClientStream<Integer> mockClientStream;
+
+  SilenceableBidiStream<Integer, String> sut;


what does sut stand for here?

System Under Test

Prevent watch stream from emitting events after close.

b9262b6

product-auto-label bot added size: l Pull request size is large. api: firestore Issues related to the googleapis/java-firestore API. labels Nov 8, 2023

tom-andersen commented Nov 8, 2023

View reviewed changes

Cleanup

ea3ebde

tom-andersen changed the title ~~Prevent watch stream from emitting events after close.~~ fix: Prevent watch stream from emitting events after close. Nov 8, 2023

tom-andersen added 4 commits November 8, 2023 15:17

Copyright

ce2a75f

Make classes final and hide visibility.

d115f87

Add comments

477526d

Hand roll spy

590a5b0

tom-andersen added the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Nov 8, 2023

yoshi-kokoro removed the kokoro:force-run Add this label to force Kokoro to re-run the tests. label Nov 8, 2023

tom-andersen marked this pull request as ready for review November 9, 2023 14:07

tom-andersen requested a review from a team as a code owner November 9, 2023 14:07

tom-andersen assigned wu-hui Nov 9, 2023

tom-andersen requested a review from wu-hui November 9, 2023 14:07

wu-hui reviewed Nov 9, 2023

View reviewed changes

Add unit test, rename class, add comments.

ac426fb

tom-andersen commented Nov 9, 2023

View reviewed changes

🦉 Updates from OwlBot post-processor

abb7e24

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

gcf-owl-bot bot requested a review from a team as a code owner November 9, 2023 21:09

tom-andersen requested a review from wu-hui November 9, 2023 21:09

tom-andersen and others added 3 commits November 9, 2023 16:20

Copyright

31db7ae

Comment

b022e4b

🦉 Updates from OwlBot post-processor

ad515dc

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

wu-hui approved these changes Nov 13, 2023

View reviewed changes

Comments

dbe6a99

tom-andersen merged commit ee3f8c0 into main Nov 14, 2023
21 of 22 checks passed

tom-andersen deleted the tomandersen/FixWatchFilterMismatchRetryLogic branch November 14, 2023 17:15

release-please bot mentioned this pull request Nov 14, 2023

chore(main): release 3.15.5 #1474

Merged

release-please bot mentioned this pull request May 17, 2024

chore(main): release 3.21.4 #1689

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: Prevent watch stream from emitting events after close. #1471

fix: Prevent watch stream from emitting events after close. #1471

tom-andersen commented Nov 8, 2023 •

edited

tom-andersen Nov 8, 2023

tom-andersen Nov 8, 2023

wu-hui left a comment

wu-hui Nov 9, 2023

wu-hui Nov 9, 2023

wu-hui Nov 9, 2023

wu-hui Nov 9, 2023

wu-hui Nov 9, 2023

tom-andersen Nov 9, 2023

wu-hui Nov 13, 2023

wu-hui Nov 9, 2023

tom-andersen Nov 9, 2023

wu-hui left a comment

wu-hui Nov 13, 2023

wu-hui Nov 13, 2023

wu-hui Nov 13, 2023

tom-andersen Nov 14, 2023


		@Mock ClientStream<Integer> mockClientStream;

		SilenceableBidiStream<Integer, String> sut;

fix: Prevent watch stream from emitting events after close. #1471

fix: Prevent watch stream from emitting events after close. #1471

Conversation

tom-andersen commented Nov 8, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wu-hui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wu-hui left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tom-andersen commented Nov 8, 2023 •

edited