feat: fix session attach stream abort issues and update version to 6.0.0-alpha.22

Signed-off-by: Vladislav Polyakov <[email protected]>

Author: polRk

.changeset/chubby-jeans-cheat.md

@@ -0,0 +1,5 @@
+---
+'@ydbjs/query': patch
+---
+
+Fix issues when aborting session attach stream

.changeset/pre.json

@@ -14,6 +14,7 @@
   "changesets": [
     "angry-bags-fail",
     "beige-groups-tell",
+    "chubby-jeans-cheat",
     "cold-donkeys-wash",
     "common-facts-kneel",
     "cuddly-cougars-cough",

packages/query/CHANGELOG.md

@@ -1,5 +1,19 @@
 # @ydbjs/query
 
+## 6.0.0-alpha.22
+
+### Patch Changes
+
+- Fix issues when aborting session attach stream
+
+## 6.0.0-alpha.21
+
+### Patch Changes
+
+- Fix cjs bundle issues
+- Updated dependencies
+  - @ydbjs/value@6.0.0-alpha.11
+
 ## 6.0.0-alpha.20
 
 ### Patch Changes

packages/query/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@ydbjs/query",
-	"version": "6.0.0-alpha.20",
+	"version": "6.0.0-alpha.22",
 	"description": "High-level, type-safe YQL query and transaction client for YDB. Supports tagged template syntax, parameter binding, transactions, and statistics.",
 	"license": "Apache-2.0",
 	"type": "module",
@@ -13,7 +13,8 @@
 		"build:esm": "tsc --project tsconfig.json --outDir ./dist/esm",
 		"test": "vitest --run",
 		"test:watch": "vitest --watch",
-		"attw": "attw --pack"
+		"attw": "attw --pack",
+		"prepublishOnly": "npm run clean && npm run build:cjs && npm run build:esm && npm run attw"
 	},
 	"exports": {
 		".": {
@@ -42,7 +43,7 @@
 		"@ydbjs/core": "6.0.0-alpha.16",
 		"@ydbjs/error": "6.0.0-alpha.7",
 		"@ydbjs/retry": "6.0.0-alpha.10",
-		"@ydbjs/value": "6.0.0-alpha.10",
+		"@ydbjs/value": "6.0.0-alpha.11",
 		"debug": "^4.4.0",
 		"nice-grpc": "^2.1.12"
 	},

packages/query/src/index.ts

@@ -1,7 +1,7 @@
 import type { Abortable } from 'node:events'
 
 import { StatusIds_StatusCode } from '@ydbjs/api/operation'
-import { QueryServiceDefinition, type SessionState } from '@ydbjs/api/query'
+import { QueryServiceDefinition } from '@ydbjs/api/query'
 import type { Driver } from '@ydbjs/core'
 import { CommitError, YDBError } from '@ydbjs/error'
 import { defaultRetryConfig, isRetryableError, retry } from '@ydbjs/retry'
@@ -149,21 +149,10 @@ export function query(driver: Driver): QueryClient {
 
 			client = driver.createClient(QueryServiceDefinition, sessionResponse.nodeId)
 
-			let attachSession = Promise.withResolvers<SessionState>()
-				; (async (stream: AsyncIterable<SessionState>) => {
-					try {
-						for await (let state of stream) {
-							signal.throwIfAborted()
-							attachSession.resolve(state)
-						}
-					} catch (err) {
-						attachSession.reject(err)
-					}
-				})(client.attachSession({ sessionId: store.sessionId }, { signal }))
-
-			let attachSessionResult = await attachSession.promise
-			if (attachSessionResult.status !== StatusIds_StatusCode.SUCCESS) {
-				throw new YDBError(attachSessionResult.status, attachSessionResult.issues)
+			let attachSession = client.attachSession({ sessionId: store.sessionId })[Symbol.asyncIterator]()
+			let attachSessionResult = await attachSession.next()
+			if (attachSessionResult.value.status !== StatusIds_StatusCode.SUCCESS) {
+				throw new YDBError(attachSessionResult.value.status, attachSessionResult.value.issues)
 			}
 
 			let beginTransactionResult = await client.beginTransaction({

packages/query/src/query.ts

@@ -6,7 +6,6 @@ import {
 	ExecMode,
 	QueryServiceDefinition,
 	type QueryStats,
-	type SessionState,
 	StatsMode,
 	Syntax,
 } from '@ydbjs/api/query'
@@ -132,21 +131,10 @@ export class Query<T extends any[] = unknown[]> extends EventEmitter<QueryEventM
 				client = this.#driver.createClient(QueryServiceDefinition, nodeId)
 				this.#cleanup.push(async () => await client.deleteSession({ sessionId }))
 
-				let attachSession = Promise.withResolvers<SessionState>()
-					; (async (stream: AsyncIterable<SessionState>) => {
-						try {
-							for await (let state of stream) {
-								signal.throwIfAborted()
-								attachSession.resolve(state)
-							}
-						} catch (err) {
-							attachSession.reject(err)
-						}
-					})(client.attachSession({ sessionId }, { signal }))
-
-				let attachSessionResult = await attachSession.promise
-				if (attachSessionResult.status !== StatusIds_StatusCode.SUCCESS) {
-					throw new YDBError(attachSessionResult.status, attachSessionResult.issues)
+				let attachSession = client.attachSession({ sessionId })[Symbol.asyncIterator]()
+				let attachSessionResult = await attachSession.next()
+				if (attachSessionResult.value.status !== StatusIds_StatusCode.SUCCESS) {
+					throw new YDBError(attachSessionResult.value.status, attachSessionResult.value.issues)
 				}
 			}