fix: unhandled exception error catch (#2091)

* fix: unhandled exception error catch

* refactor: test

* lint fix

* test: catch error

* test: for aborted and unavailable error

* test: refactor

* lint fix

* refactor

* fix for begin transaction called twice

---------

Co-authored-by: surbhigarg92 <surbhigarg.92@gmail.com>

Author: alkatrivedi

.gitignore

@@ -12,3 +12,4 @@ system-test/*key.json
 .DS_Store
 package-lock.json
 __pycache__
+.vscode

src/transaction.ts

@@ -712,8 +712,16 @@ export class Snapshot extends EventEmitter {
           this._update(response.metadata!.transaction);
         }
       })
-      .on('error', () => {
-        if (!this.id && this._useInRunner) {
+      .on('error', err => {
+        const isServiceError = err && typeof err === 'object' && 'code' in err;
+        if (
+          !this.id &&
+          this._useInRunner &&
+          !(
+            isServiceError &&
+            (err as grpc.ServiceError).code === grpc.status.ABORTED
+          )
+        ) {
           this.begin();
         }
       });
@@ -1219,8 +1227,16 @@ export class Snapshot extends EventEmitter {
           this._update(response.metadata!.transaction);
         }
       })
-      .on('error', () => {
-        if (!this.id && this._useInRunner) {
+      .on('error', err => {
+        const isServiceError = err && typeof err === 'object' && 'code' in err;
+        if (
+          !this.id &&
+          this._useInRunner &&
+          !(
+            isServiceError &&
+            (err as grpc.ServiceError).code === grpc.status.ABORTED
+          )
+        ) {
           this.begin();
         }
       });
@@ -1437,6 +1453,7 @@ export class Snapshot extends EventEmitter {
       this._waitingRequests.push(() => {
         makeRequest(resumeToken)
           .on('data', chunk => streamProxy.emit('data', chunk))
+          .on('error', err => streamProxy.emit('error', err))
           .on('end', () => streamProxy.emit('end'));
       });
 

test/mockserver/mockspanner.ts

@@ -540,6 +540,7 @@ export class MockSpanner {
             call.request!.transaction.id
           }`;
           if (this.abortedTransactions.has(fullTransactionId)) {
+            call.sendMetadata(new Metadata());
             call.emit(
               'error',
               MockSpanner.createTransactionAbortedError(`${fullTransactionId}`)
@@ -556,6 +557,7 @@ export class MockSpanner {
               call.request!.transaction.begin
             );
             if (txn instanceof Error) {
+              call.sendMetadata(new Metadata());
               call.emit('error', txn);
               call.end();
               return;
@@ -593,6 +595,7 @@ export class MockSpanner {
                   index
                 );
                 if (streamErr) {
+                  call.sendMetadata(new Metadata());
                   call.emit('error', streamErr);
                   break;
                 }
@@ -610,6 +613,7 @@ export class MockSpanner {
                 1
               );
               if (streamErr) {
+                call.sendMetadata(new Metadata());
                 call.emit('error', streamErr);
                 break;
               }