This repository has been archived by the owner on Oct 9, 2023. It is now read-only.
Implement and test configurable transaction timeouts #197
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -40,6 +40,14 @@ export class BlockingQueue<T> { | |
| return this._promises.shift(); | ||
| } | ||
|
|
||
| drain(): Promise<T>[] { | ||
| const values = []; | ||
| for (let i = 0; i < this._promises.length; i++) { | ||
| values.push(this._promises.shift()); | ||
| } | ||
| return values; | ||
| } | ||
|
|
||
| private addPromise(): void { | ||
| this._promises.push(new Promise(resolve => { | ||
| this._resolvers.push(resolve); | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,22 +19,24 @@ | |
| * under the License. | ||
| */ | ||
|
|
||
| import {Transaction} from "typedb-protocol/common/transaction_pb"; | ||
| import {ConceptManager} from "../api/concept/ConceptManager"; | ||
| import {TypeDBOptions} from "../api/connection/TypeDBOptions"; | ||
| import {TransactionType, TypeDBTransaction} from "../api/connection/TypeDBTransaction"; | ||
| import {LogicManager} from "../api/logic/LogicManager"; | ||
| import {QueryManager} from "../api/query/QueryManager"; | ||
| import {ErrorMessage} from "../common/errors/ErrorMessage"; | ||
| import {TypeDBClientError} from "../common/errors/TypeDBClientError"; | ||
| import {RequestBuilder} from "../common/rpc/RequestBuilder"; | ||
| import {Stream} from "../common/util/Stream"; | ||
| import {ConceptManagerImpl} from "../concept/ConceptManagerImpl"; | ||
| import {LogicManagerImpl} from "../logic/LogicManagerImpl"; | ||
| import {QueryManagerImpl} from "../query/QueryManagerImpl"; | ||
| import {BidirectionalStream} from "../stream/BidirectionalStream"; | ||
| import {TypeDBSessionImpl} from "./TypeDBSessionImpl"; | ||
| import assert = require("assert"); | ||
|
Comment on lines
+22
to
+37
There was a problem hiding this comment. We've never used Unlike in Java, there is no clean way to turn off assertions in NodeJS. (Methods typically used include hot module replacement - literally stripping the statements out of the source code before running it - and modifying the I think we're better off replacing it with
flyingsilverfin marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| import TRANSACTION_CLOSED = ErrorMessage.Client.TRANSACTION_CLOSED; | ||
| import TRANSACTION_CLOSED_WITH_ERRORS = ErrorMessage.Client.TRANSACTION_CLOSED_WITH_ERRORS; | ||
|
|
||
| export class TypeDBTransactionImpl implements TypeDBTransaction.Extended { | ||
| private readonly _session: TypeDBSessionImpl; | ||
|
|
@@ -105,13 +107,20 @@ export class TypeDBTransactionImpl implements TypeDBTransaction.Extended { | |
| } | ||
|
|
||
| public async rpcExecute(request: Transaction.Req, batch?: boolean): Promise<Transaction.Res> { | ||
| if (!this.isOpen()) this.throwTransactionClosed() | ||
| const useBatch = batch !== false; | ||
| return this._bidirectionalStream.single(request, useBatch); | ||
| } | ||
|
|
||
| public rpcStream(request: Transaction.Req): Stream<Transaction.ResPart> { | ||
| if (!this.isOpen()) this.throwTransactionClosed(); | ||
| return this._bidirectionalStream.stream(request); | ||
| } | ||
|
|
||
| private throwTransactionClosed(): void { | ||
| assert(!this.isOpen()); | ||
| const errors = this._bidirectionalStream.getErrors(); | ||
| if (errors.length == 0) throw new TypeDBClientError(TRANSACTION_CLOSED); | ||
| else throw new TypeDBClientError(TRANSACTION_CLOSED_WITH_ERRORS.message(errors)); | ||
| } | ||
|
There was a problem hiding this comment. we can differentiate between a transaction closed with errors and one without errors |
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -29,7 +29,7 @@ def vaticle_typedb_artifacts(): | |
| artifact_name = "typedb-server-{platform}-{version}.{ext}", | ||
| tag_source = deployment["artifact.release"], | ||
| commit_source = deployment["artifact.snapshot"], | ||
| commit = "2367157bdd898e474198726d0ef5446372a73314", | ||
|
There was a problem hiding this comment. update core artifact with transaction timeout |
||
| ) | ||
|
|
||
| def vaticle_typedb_cluster_artifacts(): | ||
|
|
@@ -39,5 +39,5 @@ def vaticle_typedb_cluster_artifacts(): | |
| artifact_name = "typedb-cluster-all-{platform}-{version}.{ext}", | ||
| tag_source = deployment_private["artifact.release"], | ||
| commit_source = deployment_private["artifact.snapshot"], | ||
| commit = "f6d9eae58c83c165d932c0433a3a0d858e206fab" | ||
| ) | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -32,12 +32,12 @@ def vaticle_typedb_common(): | |
| git_repository( | ||
| name = "vaticle_typedb_common", | ||
| remote = "https://github.com/vaticle/typedb-common", | ||
| commit = "a436f7c9b64060d7fed718a3be60895c15bd893f" # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_common | ||
| ) | ||
|
|
||
| def vaticle_typedb_behaviour(): | ||
| git_repository( | ||
| name = "vaticle_typedb_behaviour", | ||
| remote = "https://github.com/vaticle/typedb-behaviour", | ||
| commit = "d92d840dd40cc8393936d6f6042820ebcd3a9cef", # sync-marker: do not remove this comment, this is used for sync-dependencies by @vaticle_typedb_behaviour | ||
|
Comment on lines
32
to
+42
There was a problem hiding this comment. bump dependencies |
||
| ) | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,7 +25,7 @@ | |
| "dependencies": { | ||
| "@grpc/grpc-js": "1.3.2", | ||
| "google-protobuf": "3.14.0", | ||
| "typedb-protocol": "https://repo.vaticle.com/repository/npm-snapshot/typedb-protocol/-/typedb-protocol-0.0.0-4ce4e1339995c8982914e9fd3177273b693609b7.tgz", | ||
| "uuid": "8.3.2" | ||
| }, | ||
| "devDependencies": { | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -127,4 +127,8 @@ export class BidirectionalStream { | |
| if (!queue) throw new TypeDBClientError(UNKNOWN_REQUEST_ID.message(requestId)); | ||
| queue.put(res); | ||
| } | ||
|
|
||
| getErrors(): (Error|string)[] { | ||
| return (this._responseCollector.getErrors()).concat(this._responsePartCollector.getErrors()); | ||
| } | ||
|
Comment on lines
+131
to
+133
There was a problem hiding this comment. retrieve errors from the different collectors |
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,9 +19,9 @@ | |
| * under the License. | ||
| */ | ||
|
|
||
| import {ErrorMessage} from "../common/errors/ErrorMessage"; | ||
| import {TypeDBClientError} from "../common/errors/TypeDBClientError"; | ||
| import {BlockingQueue} from "../common/util/BlockingQueue"; | ||
|
|
||
| export class ResponseCollector<T> { | ||
|
|
||
|
|
@@ -43,18 +43,27 @@ export class ResponseCollector<T> { | |
|
|
||
| close(error?: Error | string) { | ||
| Object.values(this._collectors).forEach(collector => collector.close(error)); | ||
| } | ||
|
|
||
| getErrors(): (Error | string)[] { | ||
| const errors: (Error | string)[] = []; | ||
| for (const requestId in this._collectors) { | ||
| const error = this._collectors[requestId].getError(); | ||
| if (error) errors.push(error); | ||
| delete this._collectors[requestId]; | ||
| } | ||
| return errors; | ||
| } | ||
|
Comment on lines
+48
to
+56
There was a problem hiding this comment. get errors from each collector, and delete the collector after |
||
| } | ||
|
|
||
| export namespace ResponseCollector { | ||
|
|
||
| import TRANSACTION_CLOSED = ErrorMessage.Client.TRANSACTION_CLOSED; | ||
| import ILLEGAL_STATE = ErrorMessage.Internal.ILLEGAL_STATE; | ||
|
|
||
| export class ResponseQueue<T> { | ||
| private readonly _queue: BlockingQueue<QueueElement>; | ||
| private _error: string | Error = null; | ||
|
|
||
| constructor() { | ||
| this._queue = new BlockingQueue<QueueElement>() | ||
|
|
@@ -63,23 +72,23 @@ export namespace ResponseCollector { | |
| async take(): Promise<T> { | ||
| const element = await this._queue.take(); | ||
| if (element.isResponse()) return (element as Response<T>).value; | ||
| else if (element.isDone() && !this._error) throw new TypeDBClientError(TRANSACTION_CLOSED); | ||
| else if (element.isDone() && this._error) throw new TypeDBClientError(this._error); | ||
| else throw new TypeDBClientError(ILLEGAL_STATE); | ||
|
There was a problem hiding this comment. if it's a |
||
| } | ||
|
|
||
| put(element: T): void { | ||
| this._queue.add(new Response(element)); | ||
| } | ||
|
|
||
| close(error?: Error | string): void { | ||
| this._error = error; | ||
| this._queue.add(new Done()); | ||
|
Comment on lines
+86
to
+87
There was a problem hiding this comment. major change: when we receive an error, we immediate put the |
||
| } | ||
|
|
||
| getError(): string | Error { | ||
| return this._error; | ||
| } | ||
| } | ||
|
|
||
| class QueueElement { | ||
|
|
@@ -113,19 +122,9 @@ export namespace ResponseCollector { | |
| } | ||
|
|
||
| class Done extends QueueElement { | ||
|
|
||
| constructor() { | ||
| super(); | ||
|
Comment on lines
-118
to
-128
There was a problem hiding this comment. keeping an error inside the |
||
| } | ||
|
|
||
| isDone(): boolean { | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
new transaction option: millis before timeout