Fix seqNo renumbering bug and simplify TopicWriter API

- Fix bug where messages written before session initialization were not renumbered after receiving lastSeqNo from server
- Remove return value from write() method (now returns void) to simplify API
- Update both legacy writer and writer2 implementations
- Add comprehensive tests for seqNo renumbering logic
- Update documentation and migration guide in changeset

Breaking changes: write() method no longer returns seqNo (minor version bump)

Signed-off-by: Vladislav Polyakov <polRk@ydb.tech>

Author: polRk

.changeset/remove-write-seqno-return.md

@@ -0,0 +1,29 @@
+---
+'@ydbjs/topic': minor
+---
+
+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:**
+
+- `TopicWriter.write()` no longer returns sequence number (now returns `void`) to simplify API and prevent confusion about temporary vs final seqNo values
+
+**Migration guide:**
+
+- If you were storing seqNo from `write()` return value, use `flush()` instead to get final seqNo:
+
+  ```typescript
+  // Before
+  let seqNo = writer.write(data)
+
+  // After
+  writer.write(data)
+  let lastSeqNo = await writer.flush() // Get final seqNo
+  ```
+
+- User-provided seqNo (via `extra.seqNo`) remain final and unchanged - no migration needed for this case.

packages/topic/package.json

@@ -30,9 +30,11 @@
 	"types": "dist/index.d.ts",
 	"exports": {
 		".": "./dist/index.js",
+		"./codec": "./dist/codec.js",
 		"./reader": "./dist/reader/index.js",
 		"./writer": "./dist/writer/index.js",
-		"./writer2": "./dist/writer2/index.js"
+		"./writer2": "./dist/writer2/index.js",
+		"./message": "./dist/message.js"
 	},
 	"engines": {
 		"node": ">=20.19.0",

packages/topic/src/index.ts

@@ -39,6 +39,3 @@ export function topic(driver: Driver): TopicClient {
 		},
 	} as TopicClient
 }
-
-export type { TopicTxReader } from './reader/index.js'
-export type { TopicTxWriter } from './writer/index.js'

packages/topic/src/writer/_init_reponse.ts

@@ -18,16 +18,18 @@ export const _on_init_response = function on_init_response(
 		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
 ) {
-	if (!ctx.lastSeqNo) {
-		// Store the last sequence number from the server.
-		ctx.updateLastSeqNo(input.lastSeqNo)
-	}
+	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()
 		if (!message) {
@@ -38,5 +40,48 @@ export const _on_init_response = function on_init_response(
 		ctx.updateBufferSize(BigInt(message.data.length))
 	}
 
-	_flush(ctx) // Flush the buffer to send any pending messages.
+	// 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 the buffer to send any pending messages
+	_flush(ctx)
 }

packages/topic/src/writer/_write.ts

@@ -37,10 +37,7 @@ export function _write(
 	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,
-		})
+		([key, value]) => ({ key, value })
 	)
 	let uncompressedSize = BigInt(data.length)
 
@@ -54,7 +51,13 @@ export function _write(
 
 	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
 }

packages/topic/src/writer/index.ts

@@ -238,6 +238,7 @@ export const createTopicWriter = function createTopicWriter(
 									throughputSettings,
 									updateLastSeqNo,
 									updateBufferSize,
+									isSeqNoProvided,
 									...(options.tx && { tx: options.tx }),
 									...(lastSeqNo && { lastSeqNo }),
 								},