Fix topic reader/writer disconnecting after Discovery.listEndpoints

When the driver refreshed its endpoint pool on a periodic discovery
round,
it closed and recreated gRPC channels for all known nodes. This caused
active
bidirectional streams (topic reader / writer) to receive a `CANCELLED`
gRPC
status, which was not treated as a retryable error — so the streams
terminated
instead of reconnecting.

Author: polRk

.changeset/fix-topic-stream-reconnect.md

@@ -0,0 +1,27 @@
+---
+"@ydbjs/retry": minor
+"@ydbjs/topic": patch
+---
+
+Fix topic reader/writer disconnecting after Discovery.listEndpoints
+
+When the driver refreshed its endpoint pool on a periodic discovery round,
+it closed and recreated gRPC channels for all known nodes. This caused active
+bidirectional streams (topic reader / writer) to receive a `CANCELLED` gRPC
+status, which was not treated as a retryable error — so the streams terminated
+instead of reconnecting.
+
+Changes:
+
+- **`@ydbjs/retry`**: added `isRetryableStreamError` and `defaultStreamRetryConfig`.
+  Long-lived streaming RPCs should reconnect on `CANCELLED` and `UNAVAILABLE`
+  in addition to the errors already handled by `isRetryableError`, because for
+  streams those codes indicate a transport interruption rather than a semantic
+  cancellation.
+
+- **`@ydbjs/topic`**: reader (`_consume_stream`) and both writers (`writer`,
+  `writer2`) now use `isRetryableStreamError` / `defaultStreamRetryConfig` so
+  they transparently reconnect after a discovery-triggered channel replacement.
+  Fixed a zombie-reader bug where `read()` would block forever if the retry
+  budget was exhausted: the reader is now destroyed on unrecoverable stream
+  errors so pending `read()` calls are unblocked immediately.

.oxlintrc.json

@@ -12,7 +12,14 @@
 		"unicorn/require-post-message-target-origin": "off",
 		"yoda": "deny",
 		"vitest/consistent-test-it": ["error", { "fn": "test" }],
-		"vitest/max-nested-describe": ["error", 0]
+		"vitest/max-nested-describe": ["error", 0],
+		"vitest/valid-expect": [
+			"error",
+			{
+				"maxArgs": 3,
+				"minArgs": 1
+			}
+		]
 	},
 	"plugins": ["node", "import", "vitest", "promise", "typescript"],
 	"ignorePatterns": ["**/node_modules/**", "**/dist/**"],

package-lock.json

@@ -19,7 +19,7 @@
 		"build": "turbo run build --output-logs=errors-only",
 		"attw": "turbo run attw",
 		"test": "vitest --run",
-		"lint": "oxlint",
+		"lint": "oxlint --fix",
 		"format": "prettier . --ignore-unknown --write",
 		"version": "npx changeset version && npm install",
 		"publish": "npx changeset publish",

package.json

@@ -8,7 +8,7 @@ import { ClientError, Status } from 'nice-grpc'
 
 import type { RetryConfig } from './config.js'
 import type { RetryContext } from './context.js'
-import { exponential, fixed } from './strategy.js'
+import { type RetryStrategy, exponential, fixed } from './strategy.js'
 
 export * from './config.js'
 export * from './context.js'
@@ -137,6 +137,29 @@ export function isRetryableError(error: unknown, idempotent = false): boolean {
 	return false
 }
 
+/**
+ * Determines whether an error from a long-lived gRPC stream should trigger
+ * a reconnect attempt.
+ *
+ * Streaming RPCs differ from unary calls: a CANCELLED or UNAVAILABLE status
+ * means the transport was interrupted (e.g. the server restarted, the
+ * connection pool was refreshed after a discovery round), not that the
+ * *operation* was semantically cancelled by the caller.  We therefore always
+ * reconnect on those codes, in addition to the errors handled by
+ * {@link isRetryableError}.
+ */
+export function isRetryableStreamError(error: unknown): boolean {
+	if (error instanceof ClientError) {
+		return (
+			error.code === Status.CANCELLED ||
+			error.code === Status.UNAVAILABLE ||
+			isRetryableError(error, true)
+		)
+	}
+
+	return isRetryableError(error, false)
+}
+
 export const defaultRetryConfig: RetryConfig = {
 	retry: isRetryableError,
 	budget: Infinity,
@@ -179,3 +202,26 @@ export const defaultRetryConfig: RetryConfig = {
 		return exponential(10)(ctx, cfg)
 	},
 }
+
+/**
+ * Default retry configuration for long-lived gRPC streaming connections
+ * (topic reader / writer).
+ *
+ * Extends {@link defaultRetryConfig} with reconnect logic for transient
+ * transport errors ({@link isRetryableStreamError}).
+ */
+export const defaultStreamRetryConfig: RetryConfig = {
+	...defaultRetryConfig,
+	retry: isRetryableStreamError,
+	strategy: (ctx, cfg) => {
+		if (
+			ctx.error instanceof ClientError &&
+			(ctx.error.code === Status.CANCELLED ||
+				ctx.error.code === Status.UNAVAILABLE)
+		) {
+			return exponential(10)(ctx, cfg)
+		}
+
+		return (defaultRetryConfig.strategy as RetryStrategy)(ctx, cfg)
+	},
+}

packages/retry/src/index.ts

@@ -3,7 +3,7 @@ import { StatusIds_StatusCode } from '@ydbjs/api/operation'
 import { TopicServiceDefinition } from '@ydbjs/api/topic'
 import { loggers } from '@ydbjs/debug'
 import { YDBError } from '@ydbjs/error'
-import { defaultRetryConfig, retry } from '@ydbjs/retry'
+import { defaultStreamRetryConfig, retry } from '@ydbjs/retry'
 
 import { _send_init_request } from './_init_request.js'
 import { _on_init_response } from './_init_response.js'
@@ -26,7 +26,7 @@ export let _consume_stream = async function consume_stream(
 	let signal = state.controller.signal
 	await state.driver.ready(signal)
 
-	await retry({ ...defaultRetryConfig, signal }, async (signal) => {
+	await retry({ ...defaultStreamRetryConfig, signal }, async (signal) => {
 		state.outgoingQueue.close()
 		state.outgoingQueue.reset()
 

packages/topic/src/reader/_consume_stream.ts

@@ -67,11 +67,13 @@ export const createTopicReader = function createTopicReader(
 		} catch (error) {
 			if (!state.controller.signal.aborted) {
 				dbg.log('error occurred while streaming: %O', error)
+				// Stream died unexpectedly (retry budget exhausted or non-retryable
+				// error). Destroy the reader so that any pending read() calls are
+				// unblocked rather than polling forever.
+				destroy(error)
 			}
 		} finally {
 			dbg.log('stream closed')
-			// Don't call destroy here - it's already being called by close/destroy
-			// and calling it again would be a no-op anyway due to disposed check
 		}
 	})()
 

packages/topic/src/reader/index.ts

@@ -10,7 +10,7 @@ import {
 import type { Driver } from '@ydbjs/core'
 import { loggers } from '@ydbjs/debug'
 import { YDBError } from '@ydbjs/error'
-import { isRetryableError, retry } from '@ydbjs/retry'
+import { isRetryableStreamError, retry } from '@ydbjs/retry'
 import { backoff, combine, jitter } from '@ydbjs/retry/strategy'
 
 import { type CompressionCodec, defaultCodecMap } from '../codec.js'
@@ -160,7 +160,7 @@ export const createTopicWriter = function createTopicWriter(
 
 		let retryConfig = options.retryConfig?.(signal)
 		retryConfig ??= {
-			retry: isRetryableError,
+			retry: isRetryableStreamError,
 			signal: signal,
 			budget: Infinity,
 			strategy: combine(backoff(50, 5000), jitter(50)),

packages/topic/src/writer/index.ts

@@ -33,7 +33,7 @@ import {
 	TransactionIdentitySchema,
 } from '@ydbjs/api/topic'
 import type { Driver } from '@ydbjs/core'
-import { isRetryableError } from '@ydbjs/retry'
+import { isRetryableStreamError } from '@ydbjs/retry'
 import { assign, enqueueActions, sendTo, setup } from 'xstate'
 import { defaultCodecMap } from '../codec.js'
 import { WriterStream, type WriterStreamReceiveEvent } from './stream.js'
@@ -791,14 +791,14 @@ let writerMachineFactory = setup({
 		},
 		retryableError: ({ context }) => {
 			if (context.lastError) {
-				return isRetryableError(context.lastError, true)
+				return isRetryableStreamError(context.lastError)
 			}
 
 			return false
 		},
 		nonRetryableError: ({ context }) => {
 			if (context.lastError) {
-				return !isRetryableError(context.lastError, true)
+				return !isRetryableStreamError(context.lastError)
 			}
 
 			return false