Handle "Unexpected response type for remote procedure call: Close" on query stream opening (#759)

## Usage and product changes
Fix a rare `InternalError` returned by mistake when a client sends a
query request while the transaction is being closed. Now, an expected
"The transaction is closed and no further operation is allowed." error
is returned instead.

Additionally, wait for specific transaction responses in `rollback`,
`commit`, and `query` to solidify the protocol and ensure that the
server acts as expected.

## Implementation
We
[faced](https://discord.com/channels/665254494820368395/1301180015173435454/1379414371402256454)
this error when running a huge number of queries against a server that
eventually went to sleep. Presumably, this state led to the GRPC stream
interruption, and it "gracefully" stopped without errors. However, the
driver's code did not expect that it could happen right after the stream
opening request, while it should be considered a normal situation.

The source code where `Close` responses are generated:
```
async fn listen_loop(
    mut grpc_source: Streaming<transaction::Server>,
    collector: ResponseCollector,
    shutdown_sink: UnboundedSender<()>,
) {
    loop {
        match grpc_source.next().await {
            Some(Ok(message)) => collector.collect(message).await,
            Some(Err(status)) => break collector.close_with_error(status.into()).await,
            None => break collector.close().await, // <---- here, the stream has ended
        }
    }
    shutdown_sink.send(()).ok();
}

async fn close(self) {
  self.is_open.store(false);
  let mut listeners = std::mem::take(&mut *self.callbacks.write().unwrap());
  for (_, listener) in listeners.drain() {
      listener.finish(Ok(TransactionResponse::Close)); // <--- here, the only `Close` response is sent
  }
  // ....
}
```

Additionally, we explicitly wait for *request's response type* or
*stream closed marker* with their correct processing for each other
transaction request in `transaction_stream`. For this, the commit
behavior of the server has been made more explicit:
typedb/typedb#7484

Author: farost

dependencies/typedb/artifacts.bzl

@@ -25,7 +25,7 @@ def typedb_artifact():
         artifact_name = "typedb-all-{platform}-{version}.{ext}",
         tag_source = deployment["artifact"]["release"]["download"],
         commit_source = deployment["artifact"]["snapshot"]["download"],
-        commit = "16eca5f4238a772532a6824966e2fc78437467dc"
+        commit = "d83f3b2b40c673c30d00b377ce327ac0ff233056"
     )
 
 #def typedb_cloud_artifact():

rust/src/connection/transaction_stream.rs

@@ -35,6 +35,30 @@ use crate::{
     promisify, resolve, Error, QueryOptions, TransactionOptions, TransactionType,
 };
 
+macro_rules! require_transaction_response {
+    ($response:expr, $variant:ident(_)) => {
+        match $response {
+            Ok(TransactionResponse::$variant(inner)) => Ok(inner),
+            other => handle_unexpected_response(other),
+        }
+    };
+
+    ($response:expr, $variant:ident) => {
+        match $response {
+            Ok(TransactionResponse::$variant) => Ok(()),
+            other => handle_unexpected_response(other),
+        }
+    };
+}
+
+fn handle_unexpected_response<T>(response: Result<TransactionResponse>) -> Result<T> {
+    match response {
+        Ok(TransactionResponse::Close) => Err(ConnectionError::TransactionIsClosed.into()),
+        Ok(other) => Err(InternalError::UnexpectedResponseType { response_type: format!("{other:?}") }.into()),
+        Err(err) => Err(err),
+    }
+}
+
 pub(crate) struct TransactionStream {
     type_: TransactionType,
     options: TransactionOptions,
@@ -73,18 +97,14 @@ impl TransactionStream {
     pub(crate) fn commit(self: Pin<Box<Self>>) -> impl Promise<'static, Result> {
         let promise = self.transaction_transmitter.single(TransactionRequest::Commit);
         promisify! {
-            let _this = self;  // move into the promise so the stream isn't dropped until the promise is resolved
-            resolve!(promise).map(|_| {
-                ()
-            }).map_err(|err| {
-                err
-            })
+            let _this = self; // move into the promise so the stream isn't dropped until the promise is resolved
+            require_transaction_response!(resolve!(promise), Commit)
         }
     }
 
     pub(crate) fn rollback(&self) -> impl Promise<'_, Result> {
         let promise = self.single(TransactionRequest::Rollback);
-        promisify! { resolve!(promise).map(|_| ()) }
+        promisify! { require_transaction_response!(resolve!(promise), Rollback) }
     }
 
     pub(crate) fn query(&self, query: &str, options: QueryOptions) -> impl Promise<'static, Result<QueryAnswer>> {
@@ -164,11 +184,9 @@ impl TransactionStream {
     }
 
     fn query_stream(&self, req: QueryRequest) -> Result<impl Stream<Item = Result<QueryResponse>>> {
-        Ok(self.stream(TransactionRequest::Query(req))?.map(|response| match response {
-            Ok(TransactionResponse::Query(res)) => Ok(res),
-            Ok(other) => Err(InternalError::UnexpectedResponseType { response_type: format!("{other:?}") }.into()),
-            Err(err) => Err(err),
-        }))
+        Ok(self
+            .stream(TransactionRequest::Query(req))?
+            .map(|response| require_transaction_response!(response, Query(_))))
     }
 }
 

tool/test/start-community-server.sh

@@ -21,8 +21,7 @@ set -e
 rm -rf typedb-all
 
 bazel run //tool/test:typedb-extractor -- typedb-all
-BAZEL_JAVA_HOME=$(bazel run //tool/test:echo-java-home)
-./typedb-all/typedb server --development-mode.enabled --server.authentication.token_ttl_seconds 15 &
+./typedb-all/typedb server --development-mode.enabled true --server.authentication.token_ttl_seconds 15 &
 
 set +e
 POLL_INTERVAL_SECS=0.5