Fix seqNo renumbering when messages written before session initialization

polRk · polRk · commit a33d8b1d6f9e · 2025-11-07T21:53:23.000+03:00
- Remove lastSeqNo update from _flush() - it was updating before messages were actually sent to server
- Fix renumbering logic in _on_init_response to properly handle messages written before init
- Check if messages in buffer need renumbering by comparing their seqNo with serverLastSeqNo
- Never renumber messages if user provided seqNo (manual mode)
- Update lastSeqNo only when session is initialized or new message is written, not on ACKs
diff --git a/.changeset/remove-write-seqno-return.md b/.changeset/remove-write-seqno-return.md
@@ -2,11 +2,12 @@
 '@ydbjs/topic': minor
 ---
 
-Fix seqNo renumbering bug and simplify TopicWriter API.
+Fix seqNo renumbering bug in both writer implementations and simplify TopicWriter API.
 
 **Bug fix:**
 
 - Fixed issue where messages written before session initialization were not renumbered after receiving `lastSeqNo` from server. Previously, auto-generated seqNo started from 0 and were not updated when server provided actual `lastSeqNo`, causing seqNo conflicts. Now messages are properly renumbered to continue from server's `lastSeqNo + 1`.
+- Fixed in both `writer` (legacy) and `writer2` implementations
 
 **API changes:**
 
diff --git a/packages/topic/src/writer/_flush.ts b/packages/topic/src/writer/_flush.ts
@@ -1,37 +1,37 @@
-import type { StreamWriteMessage_FromClient, StreamWriteMessage_WriteRequest_MessageData } from "@ydbjs/api/topic";
-import type { CompressionCodec } from "../codec.js";
-import type { AsyncPriorityQueue } from "../queue.js";
-import type { TX } from "../tx.js";
-import { _batch_messages } from "./_batch_messages.js";
-import { _emit_write_request } from "./_write_request.js";
-import type { ThroughputSettings } from "./types.js";
+import type { StreamWriteMessage_FromClient, StreamWriteMessage_WriteRequest_MessageData } from '@ydbjs/api/topic'
+import type { CompressionCodec } from '../codec.js'
+import type { AsyncPriorityQueue } from '../queue.js'
+import type { TX } from '../tx.js'
+import { _batch_messages } from './_batch_messages.js'
+import { _emit_write_request } from './_write_request.js'
+import type { ThroughputSettings } from './types.js'
 
 export const _flush = function flush(ctx: {
 	readonly tx?: TX
-	readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>,
-	readonly codec: CompressionCodec, // Codec to use for compression
-	readonly buffer: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages in the buffer
-	readonly inflight: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages that are currently in-flight
-	readonly throughputSettings: ThroughputSettings;
-	updateBufferSize: (bytes: bigint) => void; // Function to update the buffer size
+	readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>
+	readonly codec: CompressionCodec // Codec to use for compression
+	readonly buffer: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages in the buffer
+	readonly inflight: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages that are currently in-flight
+	readonly throughputSettings: ThroughputSettings
+	updateBufferSize: (bytes: bigint) => void // Function to update the buffer size
 }) {
 	if (!ctx.buffer.length) {
-		return; // Nothing to flush
+		return // Nothing to flush
 	}
 
-	let messagesToSend: StreamWriteMessage_WriteRequest_MessageData[] = [];
+	let messagesToSend: StreamWriteMessage_WriteRequest_MessageData[] = []
 
 	while (ctx.inflight.length < ctx.throughputSettings.maxInflightCount) {
-		let message = ctx.buffer.shift();
+		let message = ctx.buffer.shift()
 		if (!message) {
-			break; // No more messages to send
+			break // No more messages to send
 		}
 
-		ctx.inflight.push(message);
-		messagesToSend.push(message);
+		ctx.inflight.push(message)
+		messagesToSend.push(message)
 	}
 
 	for (let batch of _batch_messages(messagesToSend)) {
-		_emit_write_request(ctx, batch); // Emit the write request with the batch of messages
+		_emit_write_request(ctx, batch) // Emit the write request with the batch of messages
 	}
 }
diff --git a/packages/topic/src/writer/_init_reponse.ts b/packages/topic/src/writer/_init_reponse.ts
@@ -1,35 +1,84 @@
-import type { StreamWriteMessage_FromClient, StreamWriteMessage_InitResponse, StreamWriteMessage_WriteRequest_MessageData } from "@ydbjs/api/topic";
-import type { CompressionCodec } from "../codec.js";
-import type { AsyncPriorityQueue } from "../queue.js";
-import type { TX } from "../tx.js";
-import { _flush } from "./_flush.js";
-import type { ThroughputSettings } from "./types.js";
+import type {
+	StreamWriteMessage_FromClient,
+	StreamWriteMessage_InitResponse,
+	StreamWriteMessage_WriteRequest_MessageData,
+} from '@ydbjs/api/topic'
+import type { CompressionCodec } from '../codec.js'
+import type { AsyncPriorityQueue } from '../queue.js'
+import type { TX } from '../tx.js'
+import { _flush } from './_flush.js'
+import type { ThroughputSettings } from './types.js'
 
-export const _on_init_response = function on_init_response(ctx: {
-	readonly tx?: TX
-	readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>,
-	readonly codec: CompressionCodec, // Codec to use for compression
-	readonly buffer: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages in the buffer
-	readonly inflight: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages that are currently in-flight
-	readonly lastSeqNo?: bigint; // The last sequence number acknowledged by the server
-	readonly throughputSettings: ThroughputSettings; // Current throughput settings for the writer
-	updateLastSeqNo: (seqNo: bigint) => void;
-	updateBufferSize: (bytes: bigint) => void; // Function to update the buffer size
-}, input: StreamWriteMessage_InitResponse) {
-	if (!ctx.lastSeqNo) {
-		// Store the last sequence number from the server.
-		ctx.updateLastSeqNo(input.lastSeqNo);
-	}
+export const _on_init_response = function on_init_response(
+	ctx: {
+		readonly tx?: TX
+		readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>
+		readonly codec: CompressionCodec // Codec to use for compression
+		readonly buffer: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages in the buffer
+		readonly inflight: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages that are currently in-flight
+		readonly lastSeqNo?: bigint // The last sequence number acknowledged by the server
+		readonly throughputSettings: ThroughputSettings // Current throughput settings for the writer
+		readonly isSeqNoProvided?: boolean // Whether user provided seqNo (manual mode)
+		updateLastSeqNo: (seqNo: bigint) => void
+		updateBufferSize: (bytes: bigint) => void // Function to update the buffer size
+	},
+	input: StreamWriteMessage_InitResponse
+) {
+	let serverLastSeqNo = input.lastSeqNo || 0n
+	let currentLastSeqNo = ctx.lastSeqNo
+	let isFirstInit = currentLastSeqNo === undefined
+	let lastSeqNoChanged = isFirstInit || currentLastSeqNo !== serverLastSeqNo
 
+	// Return inflight messages to buffer
 	while (ctx.inflight.length > 0) {
-		const message = ctx.inflight.pop();
+		const message = ctx.inflight.pop()
 		if (!message) {
-			continue;
+			continue
 		}
 
-		ctx.buffer.unshift(message);
-		ctx.updateBufferSize(BigInt(message.data.length));
+		ctx.buffer.unshift(message)
+		ctx.updateBufferSize(BigInt(message.data.length))
+	}
+
+	// If this is the first initialization or server provided a new lastSeqNo, and we're in auto seqNo mode,
+	// renumber all messages in buffer to continue from serverLastSeqNo + 1
+	// Always renumber on first init, even if currentLastSeqNo === serverLastSeqNo (messages written before init)
+	// Also renumber if there are messages in buffer that were written before init (their seqNo start from 1, not serverLastSeqNo + 1)
+	let finalLastSeqNo = serverLastSeqNo
+	let shouldRenumber = false
+	// Only renumber in auto mode (when user didn't provide seqNo)
+	if (!ctx.isSeqNoProvided && ctx.buffer.length > 0) {
+		if (isFirstInit) {
+			// First initialization: always renumber messages written before init
+			shouldRenumber = true
+		} else if (lastSeqNoChanged) {
+			// Reconnection: renumber if server's lastSeqNo changed
+			shouldRenumber = true
+		} else if (ctx.buffer.length > 0) {
+			// Check if messages in buffer were written before init (seqNo start from 1, not serverLastSeqNo + 1)
+			// If first message's seqNo is <= serverLastSeqNo, it was written before init and needs renumbering
+			let firstMessageSeqNo = ctx.buffer[0]?.seqNo
+			if (firstMessageSeqNo !== undefined && firstMessageSeqNo <= serverLastSeqNo) {
+				shouldRenumber = true
+			}
+		}
+	}
+
+	if (shouldRenumber) {
+		let nextSeqNo = serverLastSeqNo + 1n
+		// Renumber all messages in buffer sequentially starting from serverLastSeqNo + 1
+		for (let message of ctx.buffer) {
+			message.seqNo = nextSeqNo
+			nextSeqNo++
+		}
+		// Update lastSeqNo to the last renumbered seqNo so flush() returns correct value
+		finalLastSeqNo = nextSeqNo - 1n
+		ctx.updateLastSeqNo(finalLastSeqNo)
+	} else if (lastSeqNoChanged) {
+		// Store the last sequence number from the server if we didn't renumber
+		ctx.updateLastSeqNo(serverLastSeqNo)
 	}
 
-	_flush(ctx); // Flush the buffer to send any pending messages.
+	// Flush the buffer to send any pending messages
+	_flush(ctx)
 }
diff --git a/packages/topic/src/writer/_write.ts b/packages/topic/src/writer/_write.ts
@@ -1,47 +1,59 @@
-import { create } from "@bufbuild/protobuf";
-import { timestampFromDate } from "@bufbuild/protobuf/wkt";
-import { type StreamWriteMessage_WriteRequest_MessageData, StreamWriteMessage_WriteRequest_MessageDataSchema } from "@ydbjs/api/topic";
-import type { CompressionCodec } from "../codec.js";
-import { _flush } from "./_flush.js";
-import { MAX_PAYLOAD_SIZE } from "./constants.js";
+import { create } from '@bufbuild/protobuf'
+import { timestampFromDate } from '@bufbuild/protobuf/wkt'
+import {
+	type StreamWriteMessage_WriteRequest_MessageData,
+	StreamWriteMessage_WriteRequest_MessageDataSchema,
+} from '@ydbjs/api/topic'
+import type { CompressionCodec } from '../codec.js'
+import { _flush } from './_flush.js'
+import { MAX_PAYLOAD_SIZE } from './constants.js'
 
-export function _write(ctx: {
-	readonly codec: CompressionCodec, // Codec to use for compression
-	readonly buffer: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages in the buffer
-	readonly inflight: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages that are currently in-flight
-	readonly lastSeqNo: bigint, // Last sequence number used
-	updateLastSeqNo: (seqNo: bigint) => void;
-	updateBufferSize: (bytes: bigint) => void; // Function to update the buffer size
-}, msg: {
-	data: Uint8Array,
-	seqNo?: bigint,
-	createdAt?: Date,
-	metadataItems?: Record<string, Uint8Array>
-}): bigint {
-	let data = ctx.codec ? ctx.codec.compress(msg.data) : msg.data;
+export function _write(
+	ctx: {
+		readonly codec: CompressionCodec // Codec to use for compression
+		readonly buffer: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages in the buffer
+		readonly inflight: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages that are currently in-flight
+		readonly lastSeqNo: bigint // Last sequence number used
+		updateLastSeqNo: (seqNo: bigint) => void
+		updateBufferSize: (bytes: bigint) => void // Function to update the buffer size
+	},
+	msg: {
+		data: Uint8Array
+		seqNo?: bigint
+		createdAt?: Date
+		metadataItems?: Record<string, Uint8Array>
+	}
+): bigint {
+	let data = ctx.codec ? ctx.codec.compress(msg.data) : msg.data
 
 	// Validate the payload size, it should not exceed MAX_PAYLOAD_SIZE
 	// This is a YDB limitation for single message size.
 	if (data.length > MAX_PAYLOAD_SIZE) {
-		throw new Error(`Payload size exceeds ${Number(MAX_PAYLOAD_SIZE / (1024n * 1024n))}MiB limit.`);
+		throw new Error(`Payload size exceeds ${Number(MAX_PAYLOAD_SIZE / (1024n * 1024n))}MiB limit.`)
 	}
 
-	let seqNo = msg.seqNo ?? ((ctx.lastSeqNo ?? 0n) + 1n);
-	let createdAt = timestampFromDate(msg.createdAt ?? new Date());
-	let metadataItems = Object.entries(msg.metadataItems || {}).map(([key, value]) => ({ key, value }));
-	let uncompressedSize = BigInt(data.length);
+	let seqNo = msg.seqNo ?? (ctx.lastSeqNo ?? 0n) + 1n
+	let createdAt = timestampFromDate(msg.createdAt ?? new Date())
+	let metadataItems = Object.entries(msg.metadataItems || {}).map(([key, value]) => ({ key, value }))
+	let uncompressedSize = BigInt(data.length)
 
 	let message = create(StreamWriteMessage_WriteRequest_MessageDataSchema, {
 		data,
 		seqNo,
 		createdAt,
 		metadataItems,
 		uncompressedSize,
-	});
+	})
+
+	ctx.buffer.push(message) // Store the message in the buffer
+	ctx.updateBufferSize(BigInt(data.length)) // Update the buffer size
 
-	ctx.buffer.push(message); // Store the message in the buffer
-	ctx.updateBufferSize(BigInt(data.length)); // Update the buffer size
-	ctx.updateLastSeqNo(seqNo); // Update the last sequence number
+	// Only update lastSeqNo if session is initialized (lastSeqNo is defined)
+	// For messages written before session initialization, lastSeqNo will be updated
+	// after renumbering in _on_init_response
+	if (ctx.lastSeqNo !== undefined) {
+		ctx.updateLastSeqNo(seqNo)
+	}
 
-	return seqNo;
+	return seqNo
 }
diff --git a/packages/topic/src/writer/_write_response.ts b/packages/topic/src/writer/_write_response.ts
@@ -1,38 +1,45 @@
-import type { StreamWriteMessage_FromClient, StreamWriteMessage_WriteRequest_MessageData, StreamWriteMessage_WriteResponse } from "@ydbjs/api/topic";
-import type { CompressionCodec } from "../codec.js";
-import type { AsyncPriorityQueue } from "../queue.js";
-import type { TX } from "../tx.js";
-import { _flush } from "./_flush.js";
-import type { ThroughputSettings } from "./types.js";
+import type {
+	StreamWriteMessage_FromClient,
+	StreamWriteMessage_WriteRequest_MessageData,
+	StreamWriteMessage_WriteResponse,
+} from '@ydbjs/api/topic'
+import type { CompressionCodec } from '../codec.js'
+import type { AsyncPriorityQueue } from '../queue.js'
+import type { TX } from '../tx.js'
+import { _flush } from './_flush.js'
+import type { ThroughputSettings } from './types.js'
 
-export const _on_write_response = function on_write_response(ctx: {
-	readonly tx?: TX
-	readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>,
-	readonly codec: CompressionCodec, // Codec to use for compression
-	readonly buffer: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages that are currently in-flight
-	readonly inflight: StreamWriteMessage_WriteRequest_MessageData[]; // Array of messages that are currently in-flight
-	readonly throughputSettings: ThroughputSettings; // Current throughput settings for the writer
-	onAck?: (seqNo: bigint, status?: 'skipped' | 'written' | 'writtenInTx') => void // Callback for handling acknowledgments
-	updateBufferSize: (bytes: bigint) => void; // Function to update the buffer size
-}, input: StreamWriteMessage_WriteResponse) {
+export const _on_write_response = function on_write_response(
+	ctx: {
+		readonly tx?: TX
+		readonly queue: AsyncPriorityQueue<StreamWriteMessage_FromClient>
+		readonly codec: CompressionCodec // Codec to use for compression
+		readonly buffer: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages that are currently in-flight
+		readonly inflight: StreamWriteMessage_WriteRequest_MessageData[] // Array of messages that are currently in-flight
+		readonly throughputSettings: ThroughputSettings // Current throughput settings for the writer
+		onAck?: (seqNo: bigint, status?: 'skipped' | 'written' | 'writtenInTx') => void // Callback for handling acknowledgments
+		updateBufferSize: (bytes: bigint) => void // Function to update the buffer size
+	},
+	input: StreamWriteMessage_WriteResponse
+) {
 	// Process each acknowledgment in the response.
 
-	let acks = new Map<bigint, 'skipped' | 'written' | 'writtenInTx'>();
+	let acks = new Map<bigint, 'skipped' | 'written' | 'writtenInTx'>()
 	for (let ack of input.acks) {
-		acks.set(ack.seqNo, ack.messageWriteStatus.case!);
+		acks.set(ack.seqNo, ack.messageWriteStatus.case!)
 	}
 
 	// Acknowledge messages that have been processed.
 	for (let i = ctx.inflight.length - 1; i >= 0; i--) {
-		const message = ctx.inflight[i]!;
+		const message = ctx.inflight[i]!
 		if (acks.has(message.seqNo)) {
-			ctx.onAck?.(message.seqNo, acks.get(message.seqNo));
-			ctx.inflight.splice(i, 1);
+			ctx.onAck?.(message.seqNo, acks.get(message.seqNo))
+			ctx.inflight.splice(i, 1)
 		}
 	}
 
 	// Clear the acknowledgment map.
-	acks.clear();
+	acks.clear()
 
 	// If there are still messages in the buffer, flush them.
 	_flush(ctx)
diff --git a/packages/topic/src/writer/index.ts b/packages/topic/src/writer/index.ts
@@ -205,8 +205,9 @@ export const createTopicWriter = function createTopicWriter(driver: Driver, opti
 									throughputSettings,
 									updateLastSeqNo,
 									updateBufferSize,
+									isSeqNoProvided,
 									...(options.tx && { tx: options.tx }),
-									...(lastSeqNo && { lastSeqNo })
+									...(lastSeqNo && { lastSeqNo }),
 								},
 								chunk.serverMessage.value
 							)
@@ -503,4 +504,4 @@ export const createTopicTxWriter = function createTopicTxWriter(
 }
 
 // Re-export types for compatibility
-export type { TopicTxWriter, TopicWriter, TopicWriterOptions } from "./types.js"
+export type { TopicTxWriter, TopicWriter, TopicWriterOptions } from './types.js'
diff --git a/packages/topic/tests/writer.test.ts b/packages/topic/tests/writer.test.ts
