Fix OIDC cancelRow crash and SF drainer terminal

The httpTokenProvider work introduced two defects.

cancelRow() could segfault the JVM. With a provider configured,
newRequest() defers the token and leaves the request at the header
stage, so contentStart stays at its -1 sentinel and no row bytes are
buffered. cancelRow() then ran trimContentToLen(0), which set the
write pointer to -1, and the next buffer write faulted in
Unsafe.putByte. cancelRow() now returns early while the token is
pending (nothing is buffered yet), and trimContentToLen refuses the
-1 sentinel as a defensive guard.

The store-and-forward background drainer could terminate a producer on
a transient credential outage. It bounded a token-provider failure by
reconnect_max_duration_millis and then latched a SECURITY_ERROR
terminal, dropping a producer that store-and-forward had promised to
keep alive. A failing provider (IdP unreachable, a silent refresh
failing, an interactive sign-in in progress) is a transient outage
like any other, so the running drainer now retries it indefinitely
with capped backoff, per Invariant B. The foreground/SYNC initial
connect still fails fast, because a connectivity error is only the
caller's problem during initialization. This restores the file's own
field comment, which already declared reconnect_max_duration_millis
"NOT consulted by the background loop".

The crash gets a regression test proven to segfault without the guard.
The drainer tests now assert the sender survives a persistent provider
outage and recovers, replacing a catch-all that asserted nothing about
the terminal.

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

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/http/client/HttpClient.java

@@ -548,6 +548,13 @@ public String toString() {
         }
 
         public void trimContentToLen(int contentLen) {
+            if (contentStart < 0) {
+                // withContent() has not started a content section yet, so contentStart is the -1 sentinel
+                // and contentStart + contentLen would be a negative, invalid write pointer that the next
+                // write would segfault on. Nothing has been written into a content section, so there is
+                // nothing to trim.
+                return;
+            }
             ptr = contentStart + contentLen;
         }
 

core/src/main/java/io/questdb/client/cutlass/line/http/AbstractLineHttpSender.java

@@ -469,6 +469,15 @@ public DirectByteSlice bufferView() {
     @Override
     public void cancelRow() {
         validateNotClosed();
+        if (isTokenPending) {
+            // newRequest() left the request at the header stage with the provider token deferred, so
+            // withContent() has not run and contentStart is still -1 (getContentLength() reads 0): no row
+            // bytes were written, so there is nothing to trim. trimContentToLen(0) would set the write
+            // pointer to contentStart + 0 == -1 and the next buffer write would segfault. Just reset the
+            // row state and leave the token pending for the next row.
+            state = RequestState.EMPTY;
+            return;
+        }
         request.trimContentToLen(rowBookmark);
         state = RequestState.EMPTY;
     }

core/src/main/java/io/questdb/client/cutlass/qwp/client/sf/cursor/CursorWebSocketSendLoop.java

@@ -1138,18 +1138,16 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
         // INVARIANT B: a store-and-forward drainer must NEVER terminate on a
         // wall-clock reconnect budget. A replica-only / all-endpoints-replica
         // window is TRANSIENT -- a replica gets promoted, a primary reappears --
-        // so this background loop retries for as long as it is running, backing
-        // off between attempts. The ONLY terminal conditions are a genuinely
-        // non-retriable upgrade (auth / non-421 upgrade / durable-ack capability
-        // gap), which return directly below, a credential the client cannot
-        // ACQUIRE (QwpCredentialUnavailableException -- see its catch below), or
-        // the sender being stopped. SF exhaustion is surfaced to the PRODUCER as
-        // append backpressure, never here. reconnect_max_duration_millis is
-        // intentionally NOT consulted for a TRANSPORT outage: for those it bounds
-        // only the blocking (non-lazy) initial connect in
-        // QwpWebSocketSender.buildAndConnect, never this background loop. It does
-        // bound an uninterrupted run of credential-acquisition failures, which no
-        // amount of retrying can clear.
+        // and so is a token-provider failure -- the IdP becomes reachable again,
+        // or the user completes an interactive sign-in -- so this background loop
+        // retries all of them for as long as it is running, backing off between
+        // attempts. The ONLY terminal conditions are a genuinely non-retriable
+        // upgrade (auth / non-421 upgrade / durable-ack capability gap), which
+        // return directly below, or the sender being stopped. SF exhaustion is
+        // surfaced to the PRODUCER as append backpressure, never here.
+        // reconnect_max_duration_millis is intentionally NOT consulted anywhere
+        // in this background loop: it bounds only the blocking (non-lazy) initial
+        // connect in QwpWebSocketSender.buildAndConnect, never this loop.
         long backoffMillis = reconnectInitialBackoffMillis;
         if (paceFirstAttemptMillis > 0 && running) {
             // NACK-initiated recycle against a reachable server: pace the
@@ -1170,11 +1168,6 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
         int attempts = 0;
         long lastLogNanos = 0L;
         Throwable lastReconnectError = initial;
-        // Start of the current UNINTERRUPTED run of credential-acquisition failures; 0 when none is in
-        // flight. Any other outcome (a connect, or a transport-class failure) clears it, so only a
-        // sustained inability to obtain a credential burns the budget and terminates -- a token blip
-        // between transport errors still recovers.
-        long credentialFailingSinceNanos = 0L;
         while (running) {
             attempts++;
             totalReconnectAttempts.incrementAndGet();
@@ -1271,52 +1264,26 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                 dispatchError(err);
                 return;
             } catch (QwpCredentialUnavailableException e) {
-                // The token provider threw instead of returning a credential. Unlike a transport outage
-                // (Invariant B), retrying alone cannot clear this: the loop cannot conjure a token the
-                // provider will not hand over. But a silent refresh can fail transiently, so retry while
-                // one still might recover -- bounded by the reconnect budget -- and only terminate once
-                // credential acquisition has failed for that whole uninterrupted window. Terminating is
-                // what keeps a dead credential ("not signed in") from reconnect-looping forever behind
-                // throttled logs: it surfaces the provider's own message to the producer via checkError()
-                // and to the async handler, exactly as a server-side 401/403 does above.
-                long now = System.nanoTime();
-                if (credentialFailingSinceNanos == 0L) {
-                    credentialFailingSinceNanos = now;
-                }
-                // Compare in millis, not nanos: the builder puts no upper bound on the budget, and a
-                // millis-to-nanos multiply of a large one overflows to a negative bound -- which would
-                // terminate on the FIRST credential blip instead of never.
-                if ((now - credentialFailingSinceNanos) / 1_000_000L > reconnectMaxDurationMillis) {
-                    LOG.error("token provider failed for {}ms during {} -- won't retry: {}",
-                            reconnectMaxDurationMillis, phase, e.getMessage());
-                    long fromFsn = engine.ackedFsn() + 1L;
-                    long toFsn = Math.max(fromFsn, engine.publishedFsn());
-                    SenderError err = new SenderError(
-                            SenderError.Category.SECURITY_ERROR,
-                            SenderError.Policy.TERMINAL,
-                            SenderError.NO_STATUS_BYTE,
-                            "token-provider-failed: " + e.getMessage(),
-                            SenderError.NO_MESSAGE_SEQUENCE,
-                            fromFsn,
-                            toFsn,
-                            null,
-                            System.nanoTime()
-                    );
-                    totalServerErrors.incrementAndGet();
-                    recordFatal(new LineSenderServerException(err));
-                    dispatchError(err);
-                    return;
-                }
+                // The token provider threw instead of returning a credential (a failed silent refresh, an
+                // interactive sign-in in progress on another thread, or not signed in yet). In the RUNNING
+                // background drainer this is a TRANSIENT outage like any other under Invariant B: the provider
+                // hands over a token again once the IdP is reachable or the user finishes signing in, and the
+                // un-acked rows stay safe in on-disk SF meanwhile. So retry indefinitely with capped backoff --
+                // NEVER bound by a wall-clock budget and NEVER latch a terminal, which would drop a producer
+                // that store-and-forward promised to keep alive on a recoverable fault. The foreground/SYNC
+                // initial connect still fails fast with the provider's own exception (connectWithRetry, and the
+                // OFF-mode connect in QwpWebSocketSender), because a connectivity error is only the caller's to
+                // see DURING initialization, not after the drainer is running.
                 lastReconnectError = e;
+                long now = System.nanoTime();
                 if (now - lastLogNanos >= RECONNECT_LOG_THROTTLE_NANOS) {
-                    LOG.warn("{} attempt {}: the token provider failed ({}); retrying within the "
-                                    + "credential budget -- if this persists the sender terminates",
+                    LOG.warn("{} attempt {}: the token provider failed ({}); retrying with capped backoff -- "
+                                    + "the sender keeps buffering to SF and recovers once a token is available",
                             phase, attempts, e.getMessage());
                     lastLogNanos = now;
                 }
                 // fall through to the shared capped-backoff block
             } catch (QwpRoleMismatchException | QwpIngressRoleRejectedException e) {
-                credentialFailingSinceNanos = 0L;
                 // Role mismatch: every reachable endpoint role-rejected the
                 // upgrade -- right now they are all replicas / primary-catchup.
                 // This is a TRANSIENT failover window (a replica is promotable),
@@ -1355,7 +1322,6 @@ private void connectLoop(Throwable initial, String phase, long paceFirstAttemptM
                     recordFatal(e);
                     throw (Error) e;
                 }
-                credentialFailingSinceNanos = 0L;
                 lastReconnectError = e;
                 long now = System.nanoTime();
                 if (now - lastLogNanos >= RECONNECT_LOG_THROTTLE_NANOS) {

core/src/test/java/io/questdb/client/test/cutlass/line/LineHttpSenderTokenProviderTest.java

@@ -87,6 +87,49 @@ public void testBuildSucceedsWhenProviderHasNotSignedInYet() throws Exception {
         });
     }
 
+    @Test(timeout = 30_000)
+    public void testCancelRowWithPendingTokenDoesNotCorruptRequest() throws Exception {
+        assertMemoryLeak(() -> {
+            // Regression: with an httpTokenProvider, newRequest() defers the token and leaves the request at the
+            // header stage (withContent() not yet run), so the native contentStart is still the -1 sentinel and
+            // no row bytes are buffered. cancelRow() must be a safe no-op in that window: trimContentToLen(0)
+            // would otherwise set the write pointer to contentStart + 0 == -1, and the next buffer write (the
+            // deferred Authorization header on the following row) would segfault the JVM. The window is entered
+            // after build() and again after every flush (reset() re-arms the pending token); a rejected table
+            // name - validateTableName() runs BEFORE the token is stamped - is a mainstream way to reach a
+            // cancelRow() with the token still pending.
+            try (MockOidcServer server = new MockOidcServer((method, path, body) -> MockOidcServer.json(204, ""))) {
+                AtomicInteger calls = new AtomicInteger();
+                HttpTokenProvider provider = () -> "TOKEN-" + calls.incrementAndGet();
+                try (Sender sender = Sender.builder(Sender.Transport.HTTP)
+                        .address("127.0.0.1:" + server.port())
+                        .protocolVersion(Sender.PROTOCOL_VERSION_V1)
+                        .disableAutoFlush()
+                        .httpTokenProvider(provider)
+                        .build()) {
+                    // (1) cancelRow immediately after build(), token pending, nothing buffered: before the fix the
+                    // write pointer went to -1 and the following row's write segfaulted the JVM
+                    sender.cancelRow();
+                    Assert.assertEquals("cancelRow must not pull the deferred token", 0, calls.get());
+
+                    // the sender is still usable: a real row buffers, flushes, and carries the token to the wire
+                    sender.table("t").longColumn("v", 1L).atNow();
+                    sender.flush();
+
+                    // (2) after a flush the token is pending again; cancelRow in that window must also be a safe
+                    // no-op, and the next row must still send its (rotated) token
+                    sender.cancelRow();
+                    sender.table("t").longColumn("v", 2L).atNow();
+                    sender.flush();
+                }
+                List<String> auth = server.requestAuthHeaders();
+                Assert.assertEquals("both flushes must reach the server", 2, auth.size());
+                Assert.assertEquals("Bearer TOKEN-1", auth.get(0));
+                Assert.assertEquals("Bearer TOKEN-2", auth.get(1));
+            }
+        });
+    }
+
     @Test(timeout = 30_000)
     public void testChangedProviderTokenIsRevalidated() throws Exception {
         assertMemoryLeak(() -> {