fix(ilp): close schema-reset race window with connection-generation tag

Pre-fix: flushPendingRows checked schemaResetNeeded once, at the top of
the encode pass. If the I/O thread completed a reconnect AFTER that
check but BEFORE encoder.finishMessage, the encoded bytes carried
stale schema-id refs into the previous connection's id space. Those
bytes then went through segmentLog.append (persisted to SF) and out to
the new server, which rejected them. Pre-C4: silent unbounded
reconnect-replay loop. Post-C4: terminal failure with no self-heal —
the user has to manually clear the SF dir to recover.

Fix (option C from the previous review): connection-generation tag on
each encoded batch.

- New volatile long QwpWebSocketSender.connectionGeneration, bumped
  by performReconnect AFTER schemaResetNeeded is flipped. Order is
  load-bearing: a reader that observes the new generation also sees
  the new schemaResetNeeded (volatile happens-before within the
  writer thread).

- flushPendingRows now wraps the encode in a retry loop:
    long genBefore = connectionGeneration;        // read FIRST
    if (schemaResetNeeded) reset;
    encode...
    if (connectionGeneration != genBefore) discard + retry;

  Re-encoding is cheap because the source rows in QwpTableBuffer are
  not reset until AFTER sealAndSwapBuffer (line 1830 in this commit).
  encoder.beginMessage internally calls buffer.reset(), so the
  discard step is implicit.

- Bounded at MAX_SCHEMA_RACE_RETRIES = 10. Reconnects firing faster
  than a single encode is pathological and surfaces as
  LineSenderException to the user rather than a silent infinite
  loop. countNonEmptyTables extracted as a small helper so the
  retry loop reads cleanly.

Together with C4 (server-error responses are now terminal) the
schema-reset race goes from "silent data corruption + infinite loop"
to "no poisoned batch ever reaches SF in the first place".

Tests:
- testGenerationBumpBetweenBatchesTriggersSchemaReset (SfIntegrationTest):
  reflectively bumps connectionGeneration + sets schemaResetNeeded
  between batches; asserts the next batch carries a fresh schema
  definition (frame size >= the first batch).
- testSchemaResetRaceUnderConcurrentBumps (SfIntegrationTest, 30s
  timeout): spawns a bumper thread that flips schemaResetNeeded +
  bumps generation on a 50us cadence while the main thread flushes
  200 batches in a tight loop. Asserts every batch either ships
  successfully OR the bounded MAX_SCHEMA_RACE_RETRIES fail-fast
  trips — never a silent escape, never an unexpected exception.

Mid-encode injection without instrumentation is timing-sensitive;
the stress test is a smoke check that the retry loop does not crash
under load. End-to-end "poisoned bytes never reach the server"
verification would need a strict QWP-wire-format-validating server
test handler — left as a future test.

Full suite: 1983 tests pass, zero regressions.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: bluestreak01

core/src/main/java/io/questdb/client/cutlass/qwp/client/QwpWebSocketSender.java

@@ -181,6 +181,18 @@ public class QwpWebSocketSender implements Sender {
     // thread on the next flushPendingRows so the next batch re-publishes schemas
     // the new server doesn't yet know about.
     private volatile boolean schemaResetNeeded;
+    // Monotonic counter bumped by performReconnect AFTER schemaResetNeeded is
+    // flipped. The user thread reads this at the very top of flushPendingRows
+    // (before checking schemaResetNeeded), then again after finishMessage; if
+    // the value changed, a reconnect happened mid-encode and the encoded bytes
+    // may carry stale schema refs the new server doesn't know — discard and
+    // re-encode. This closes the schema-reset race window where the previous
+    // code could persist a poisoned batch into SF that would replay-loop
+    // forever (or — post C4 fix — surface as a hard terminal failure with no
+    // self-heal). Single writer (the I/O thread) so plain volatile ++ is
+    // safe; readers establish happens-before via the volatile read pair.
+    private volatile long connectionGeneration;
+    private static final int MAX_SCHEMA_RACE_RETRIES = 10;
 
     private QwpWebSocketSender(
             String host,
@@ -1574,9 +1586,15 @@ private WebSocketClient performReconnect() throws Exception {
         client.upgrade(WRITE_PATH, authorizationHeader);
         encoder.setVersion((byte) client.getServerQwpVersion());
         // Tell the user thread to reset schema-id state on its next encode pass.
-        // Safe to set from here because the user thread reads this flag only at
-        // batch boundaries (top of flushPendingRows), not mid-encode.
+        // Order is load-bearing: schemaResetNeeded BEFORE connectionGeneration
+        // bump. The user thread reads connectionGeneration first, then
+        // schemaResetNeeded; volatile happens-before guarantees that a user
+        // thread observing the new generation also observes the new
+        // schemaResetNeeded. The post-encode generation re-read catches the
+        // race where the user thread already passed the schemaResetNeeded
+        // check before this write landed.
         schemaResetNeeded = true;
+        connectionGeneration++;
         return client;
     }
 
@@ -1668,33 +1686,13 @@ private void flushPendingRows() {
             return;
         }
 
-        // SF reconnect happened on the I/O thread; the new server has no memory
-        // of our previous schema-id assignments. Reset before encoding so the
-        // next batch carries full schema definitions, not just refs.
-        if (schemaResetNeeded) {
-            schemaResetNeeded = false;
-            resetSchemaStateForNewConnection();
-        }
-
-        // Invalidate cached column references -- table buffers will be reset below
+        // Invalidate cached column references -- table buffers will be reset
+        // below (or on retry).
         cachedTimestampColumn = null;
         cachedTimestampNanosColumn = null;
 
         ObjList<CharSequence> keys = tableBuffers.keys();
-
-        // Count non-empty tables for the message header
-        int tableCount = 0;
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer != null && tableBuffer.getRowCount() > 0) {
-                tableCount++;
-            }
-        }
-
+        int tableCount = countNonEmptyTables(keys);
         if (tableCount == 0) {
             pendingBytes = 0;
             pendingRowCount = 0;
@@ -1706,42 +1704,90 @@ private void flushPendingRows() {
             LOG.debug("Flushing pending rows [count={}, tables={}]", pendingRowCount, tableCount);
         }
 
-        // Ensure activeBuffer is ready for writing
-        // It might be in RECYCLED state if previous batch was sent but we didn't swap yet
+        // Schema-reset race retry loop. The encoded message carries either
+        // full schema definitions or schema-id refs. If the I/O thread
+        // performs a reconnect mid-encode (schemaResetNeeded flips while
+        // we're emitting refs), the resulting bytes would be poisoned —
+        // the new server has no memory of those ids and would reject the
+        // batch. Connection-generation tagging closes the window: we read
+        // connectionGeneration BEFORE checking schemaResetNeeded, encode,
+        // re-read after finishMessage. If the value changed, a reconnect
+        // happened mid-encode; discard the encoded bytes (still in the
+        // encoder, not yet copied into activeBuffer; table buffers haven't
+        // been reset() yet) and retry from the top. Bounded at
+        // MAX_SCHEMA_RACE_RETRIES to surface a hard failure if reconnects
+        // are pathologically frequent.
         ensureActiveBufferReady();
-
-        // Encode all non-empty tables into a single QWP v1 message
         int batchMaxSchemaId = maxSentSchemaId;
-        encoder.beginMessage(tableCount, globalSymbolDictionary, maxSentSymbolId, currentBatchMaxSymbolId);
-        for (int i = 0, n = keys.size(); i < n; i++) {
-            CharSequence tableName = keys.getQuick(i);
-            if (tableName == null) {
-                continue;
-            }
-            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
-            if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
-                continue;
-            }
+        int messageSize;
+        QwpBufferWriter buffer;
+        int retries = 0;
+        while (true) {
+            long genBefore = connectionGeneration;
+            if (schemaResetNeeded) {
+                schemaResetNeeded = false;
+                resetSchemaStateForNewConnection();
+                // resetSchemaStateForNewConnection wipes nextSchemaId and
+                // every table's cached schema id; recompute batchMaxSchemaId
+                // against the fresh maxSentSchemaId.
+                batchMaxSchemaId = maxSentSchemaId;
+            }
+
+            // Encode all non-empty tables into a single QWP v1 message.
+            // beginMessage calls buffer.reset(), so this is safe to invoke
+            // on every retry without any explicit cleanup.
+            int currBatchMaxSchemaId = batchMaxSchemaId;
+            encoder.beginMessage(tableCount, globalSymbolDictionary, maxSentSymbolId, currentBatchMaxSymbolId);
+            for (int i = 0, n = keys.size(); i < n; i++) {
+                CharSequence tableName = keys.getQuick(i);
+                if (tableName == null) {
+                    continue;
+                }
+                QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
+                if (tableBuffer == null || tableBuffer.getRowCount() == 0) {
+                    continue;
+                }
 
-            if (tableBuffer.getSchemaId() < 0) {
-                if (nextSchemaId >= maxSchemasPerConnection) {
-                    throw new LineSenderException("maximum schemas per connection exceeded")
-                            .put("[maxSchemasPerConnection=").put(maxSchemasPerConnection).put(']');
+                if (tableBuffer.getSchemaId() < 0) {
+                    if (nextSchemaId >= maxSchemasPerConnection) {
+                        throw new LineSenderException("maximum schemas per connection exceeded")
+                                .put("[maxSchemasPerConnection=").put(maxSchemasPerConnection).put(']');
+                    }
+                    tableBuffer.setSchemaId(nextSchemaId++);
                 }
-                tableBuffer.setSchemaId(nextSchemaId++);
+                currBatchMaxSchemaId = Math.max(currBatchMaxSchemaId, tableBuffer.getSchemaId());
+                boolean useSchemaRef = tableBuffer.getSchemaId() <= maxSentSchemaId;
+
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("Encoding table [name={}, rows={}, maxSentSymbolId={}, batchMaxId={}, useSchemaRef={}]", tableName, tableBuffer.getRowCount(), maxSentSymbolId, currentBatchMaxSymbolId, useSchemaRef);
+                }
+
+                encoder.addTable(tableBuffer, useSchemaRef);
             }
-            batchMaxSchemaId = Math.max(batchMaxSchemaId, tableBuffer.getSchemaId());
-            boolean useSchemaRef = tableBuffer.getSchemaId() <= maxSentSchemaId;
+            messageSize = encoder.finishMessage();
+            buffer = encoder.getBuffer();
 
+            // Race detection: did connectionGeneration advance during the
+            // encode? If yes, the bytes we just produced may carry stale
+            // schema refs.
+            if (connectionGeneration == genBefore) {
+                batchMaxSchemaId = currBatchMaxSchemaId;
+                break;
+            }
+            retries++;
+            if (retries >= MAX_SCHEMA_RACE_RETRIES) {
+                throw new LineSenderException(
+                        "schema-reset race exceeded retry limit [" + MAX_SCHEMA_RACE_RETRIES
+                                + "] — reconnects are firing faster than the user thread "
+                                + "can encode a single batch");
+            }
+            // Discard and loop. Table buffers were not reset (that happens
+            // only after sealAndSwapBuffer below); the source rows are
+            // intact for the next attempt.
             if (LOG.isDebugEnabled()) {
-                LOG.debug("Encoding table [name={}, rows={}, maxSentSymbolId={}, batchMaxId={}, useSchemaRef={}]", tableName, tableBuffer.getRowCount(), maxSentSymbolId, currentBatchMaxSymbolId, useSchemaRef);
+                LOG.debug("Schema-reset race detected mid-encode; retrying [attempt={}]", retries);
             }
-
-            encoder.addTable(tableBuffer, useSchemaRef);
         }
-        int messageSize = encoder.finishMessage();
-
-        QwpBufferWriter buffer = encoder.getBuffer();
 
         // Copy the single multi-table message to the microbatch buffer and seal
         activeBuffer.ensureCapacity(messageSize);
@@ -1904,6 +1950,21 @@ private long getPendingBytes() {
         return pendingBytes;
     }
 
+    private int countNonEmptyTables(ObjList<CharSequence> keys) {
+        int tableCount = 0;
+        for (int i = 0, n = keys.size(); i < n; i++) {
+            CharSequence tableName = keys.getQuick(i);
+            if (tableName == null) {
+                continue;
+            }
+            QwpTableBuffer tableBuffer = tableBuffers.get(tableName);
+            if (tableBuffer != null && tableBuffer.getRowCount() > 0) {
+                tableCount++;
+            }
+        }
+        return tableCount;
+    }
+
     private void resetSchemaStateForNewConnection() {
         maxSentSchemaId = -1;
         nextSchemaId = 0;

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/sf/SfIntegrationTest.java

@@ -40,6 +40,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.nio.charset.StandardCharsets;
@@ -527,6 +528,162 @@ public void testMultiTableSurvivesReconnect() throws Exception {
                 handler.frameCount() >= 5);
     }
 
+    /**
+     * Schema-reset race protection — between-batches case.
+     * <p>
+     * After a (real or simulated) reconnect, {@code connectionGeneration} is
+     * bumped and {@code schemaResetNeeded} flips. The next user-thread
+     * {@code flushPendingRows} must observe the bump, reset schema state,
+     * and emit a fresh batch — server receives a frame carrying full
+     * schema definitions, not stale refs into the previous connection's
+     * id space. This covers the simple "reconnect happened, then user
+     * flushes" path.
+     */
+    @Test
+    public void testGenerationBumpBetweenBatchesTriggersSchemaReset() throws Exception {
+        int port = TEST_PORT + 90;
+        CapturingAckHandler handler = new CapturingAckHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue("server start", server.awaitStart(5, TimeUnit.SECONDS));
+
+            try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
+                    "localhost", port,
+                    1, // autoFlushRows = 1 → each atNow ships one batch
+                    QwpWebSocketSender.DEFAULT_AUTO_FLUSH_BYTES,
+                    QwpWebSocketSender.DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
+                    8)) {
+                // First batch: server sees a fresh schema definition.
+                sender.table("foo").longColumn("v", 1L).atNow();
+                sender.flush();
+                long deadline = System.currentTimeMillis() + 5_000;
+                while (System.currentTimeMillis() < deadline && handler.frames.size() < 1) {
+                    Thread.sleep(20);
+                }
+                Assert.assertEquals(1, handler.frames.size());
+                int firstBatchSize = handler.frames.get(0).length;
+
+                // Simulate a reconnect: flip schemaResetNeeded and bump
+                // connectionGeneration via reflection. Closes the loop
+                // without going through the network — we're testing the
+                // user-thread side of the contract here.
+                forceSchemaResetAndBumpGeneration(sender);
+
+                // Second batch: must carry a full schema definition again,
+                // not a ref. Frame should be at least as large as the
+                // first (definition is strictly heavier than a ref).
+                sender.table("foo").longColumn("v", 2L).atNow();
+                sender.flush();
+                deadline = System.currentTimeMillis() + 5_000;
+                while (System.currentTimeMillis() < deadline && handler.frames.size() < 2) {
+                    Thread.sleep(20);
+                }
+                Assert.assertEquals(2, handler.frames.size());
+                int secondBatchSize = handler.frames.get(1).length;
+                Assert.assertTrue(
+                        "post-reset batch must carry a fresh schema definition; "
+                                + "first=" + firstBatchSize + " bytes, second=" + secondBatchSize
+                                + " bytes (a ref-only batch would be strictly smaller)",
+                        secondBatchSize >= firstBatchSize);
+            }
+        }
+    }
+
+    /**
+     * Schema-reset race protection — concurrent stress.
+     * <p>
+     * Spawn a thread that bumps {@code connectionGeneration} as fast as
+     * it can while the main thread flushes batches in a tight loop. Any
+     * landing of a bump during {@code flushPendingRows}' encode window
+     * must be caught by the post-encode generation re-read and re-driven
+     * through the retry loop. The test passes as long as no exception
+     * escapes flush() (other than the bounded MAX_SCHEMA_RACE_RETRIES
+     * fail-fast, which we tolerate at the very upper end of bumper rates).
+     */
+    @Test(timeout = 30_000)
+    public void testSchemaResetRaceUnderConcurrentBumps() throws Exception {
+        int port = TEST_PORT + 91;
+        CapturingAckHandler handler = new CapturingAckHandler();
+        try (TestWebSocketServer server = new TestWebSocketServer(port, handler)) {
+            server.start();
+            Assert.assertTrue("server start", server.awaitStart(5, TimeUnit.SECONDS));
+
+            try (QwpWebSocketSender sender = QwpWebSocketSender.createForTesting(
+                    "localhost", port,
+                    1,
+                    QwpWebSocketSender.DEFAULT_AUTO_FLUSH_BYTES,
+                    QwpWebSocketSender.DEFAULT_AUTO_FLUSH_INTERVAL_NANOS,
+                    8)) {
+                Field genField = QwpWebSocketSender.class.getDeclaredField("connectionGeneration");
+                genField.setAccessible(true);
+                Field resetField = QwpWebSocketSender.class.getDeclaredField("schemaResetNeeded");
+                resetField.setAccessible(true);
+
+                final int batches = 200;
+                final java.util.concurrent.atomic.AtomicBoolean stopBumper = new java.util.concurrent.atomic.AtomicBoolean(false);
+                final java.util.concurrent.atomic.AtomicLong bumpCount = new java.util.concurrent.atomic.AtomicLong(0);
+                Thread bumper = new Thread(() -> {
+                    try {
+                        while (!stopBumper.get()) {
+                            // Throttled: pause so most bumps land between
+                            // batches; a few will land mid-encode and
+                            // exercise the retry path.
+                            Thread.sleep(0, 50_000); // 50 microseconds
+                            resetField.setBoolean(sender, true);
+                            genField.setLong(sender, genField.getLong(sender) + 1);
+                            bumpCount.incrementAndGet();
+                        }
+                    } catch (Exception ignored) {
+                    }
+                }, "schema-race-bumper");
+                bumper.setDaemon(true);
+                bumper.start();
+
+                try {
+                    int sent = 0;
+                    LineSenderException maxRetryError = null;
+                    for (int i = 0; i < batches; i++) {
+                        try {
+                            sender.table("foo").longColumn("v", (long) i).atNow();
+                            sender.flush();
+                            sent++;
+                        } catch (LineSenderException e) {
+                            // The only acceptable exception is the
+                            // bounded retry-limit fail-fast — bumper is
+                            // running flat-out so it can occasionally
+                            // win 10 races back-to-back.
+                            if (e.getMessage() != null
+                                    && e.getMessage().contains("schema-reset race exceeded retry limit")) {
+                                maxRetryError = e;
+                                break;
+                            }
+                            throw e;
+                        }
+                    }
+                    Assert.assertTrue(
+                            "bumper must have fired at least once; bumps=" + bumpCount.get(),
+                            bumpCount.get() > 0);
+                    Assert.assertTrue(
+                            "either every batch shipped or the retry-limit fail-fast tripped; "
+                                    + "sent=" + sent + ", maxRetryError=" + maxRetryError,
+                            sent == batches || maxRetryError != null);
+                } finally {
+                    stopBumper.set(true);
+                    bumper.join(5_000);
+                }
+            }
+        }
+    }
+
+    private static void forceSchemaResetAndBumpGeneration(QwpWebSocketSender sender) throws Exception {
+        Field genField = QwpWebSocketSender.class.getDeclaredField("connectionGeneration");
+        genField.setAccessible(true);
+        Field resetField = QwpWebSocketSender.class.getDeclaredField("schemaResetNeeded");
+        resetField.setAccessible(true);
+        resetField.setBoolean(sender, true);
+        genField.setLong(sender, genField.getLong(sender) + 1);
+    }
+
     /**
      * sf_fsync_on_flush=off (default): the user's flush() must NOT call
      * segmentLog.fsync(). Pre-fix the docs claimed an fsync happened on