Fix WS per-endpoint token pull and getToken doc

QwpWebSocketSender.buildAndConnect resolved the Authorization header
inside its per-endpoint walk, so a token provider was queried once per
endpoint per reconnect round. A throwing provider (a failed silent
refresh, or not signed in) was then caught as a per-endpoint transport
error, retried across every endpoint, and surfaced as "all endpoints
unreachable" - masking the real auth failure and re-hammering the token
endpoint with the same dead credential.

Resolve the header once per (re)connect round, before the endpoint walk,
and reuse it across a failover (a token is cluster-wide). A provider
throw now propagates to connectWithRetry, which retries it within the
reconnect budget - the documented streaming model - and surfaces the
provider's own message. A close that races the round aborts before the
possibly blocking token pull. This mirrors QwpQueryClient, which
likewise resolves the credential once before its walk.

Add a WebSocketTokenProviderTest case that proves the provider is
queried once (not per endpoint) and its error surfaces instead of
"all endpoints unreachable"; it fails against the pre-fix code.

Also correct the OidcDeviceAuth.getToken() javadoc: the silent refresh
is not bounded by httpTimeoutMillis end to end. That timeout bounds the
send, response wait and body parse, but the preceding connection phase
(DNS, TCP connect, TLS handshake) is bounded by the OS, so an
unreachable token endpoint can stall the refresh for the OS TCP-connect
timeout (~2 minutes), not 30s. The class already documented this for
the token-store lock; the getToken() contract now matches.

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

Author: glasstiger

core/src/main/java/io/questdb/client/cutlass/auth/OidcDeviceAuth.java

@@ -87,8 +87,9 @@
  * lifetime (up to 30 minutes), so a concurrent {@link #signIn()} or {@link #clearCache()} blocks
  * behind it - but {@link #getToken()} never waits behind an interactive sign-in: it fails fast with an
  * {@link OidcAuthException} rather than stall a request/flush path (a needed silent refresh still runs,
- * bounded by {@link Builder#httpTimeoutMillis(int)} plus, with a coordinating {@link TokenStore}, a brief
- * cross-process lock wait - see {@link #getToken()}). To abort a waiting sign-in, call
+ * each HTTP round-trip phase bounded by {@link Builder#httpTimeoutMillis(int)} - though an unreachable
+ * endpoint's connect is bounded by the OS, not by it - plus, with a coordinating {@link TokenStore}, a brief
+ * cross-process lock wait; see {@link #getToken()}). To abort a waiting sign-in, call
  * {@link #close()} from another thread; it signals the flow to stop, which then fails with an
  * {@link OidcAuthException} rather than polling until the device code expires. Cancellation is seen
  * between polls (within ~100ms while waiting out an interval); a poll already in flight is not
@@ -468,12 +469,16 @@ public String getAuthorizationHeaderValue() {
      * It does not wait behind an interactive {@link #signIn()} running on another thread (which would stall
      * the flush for the whole device-code lifetime): if such a sign-in holds the lock it fails fast, and the
      * caller should retry once the sign-in completes. It is not, however, instantaneous - when the cached
-     * token has expired it makes one synchronous refresh round-trip to the token endpoint, bounded by
-     * {@link Builder#httpTimeoutMillis(int)} (30s by default); when a {@link TokenStore} coordinates the
-     * refresh across processes it may first wait briefly to acquire the store's per-identity lock (a few
-     * seconds at most for {@link FileTokenStore}, then it proceeds without the lock) before that round-trip.
-     * That is the "quick silent refresh" the {@code HttpTokenProvider} contract permits on the flush path,
-     * not an unbounded interactive wait.
+     * token has expired it makes one synchronous refresh round-trip to the token endpoint (and, with a
+     * coordinating {@link TokenStore}, may first wait briefly to acquire the store's per-identity lock - a few
+     * seconds at most for {@link FileTokenStore}, then it proceeds without the lock - before that round-trip).
+     * The send, response wait and body parse of that round-trip are each bounded by
+     * {@link Builder#httpTimeoutMillis(int)} (30s by default); the connection phase that precedes them - DNS
+     * resolution, the TCP connect and the TLS handshake - is bounded by the OS, not by httpTimeoutMillis, so an
+     * unreachable (black-holed) token endpoint can stall this refresh for the OS TCP-connect timeout (commonly
+     * ~2 minutes on Linux) rather than 30s. That is the "quick silent refresh" the {@code HttpTokenProvider}
+     * contract permits on the flush path, not an unbounded interactive wait - but a producer sizing backpressure
+     * against this call should expect that OS-bounded connect stall, not a hard 30s cap.
      *
      * @return a non-null, non-empty token
      * @throws OidcAuthException if no token has been obtained yet, if the cached token expired and could

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

@@ -139,8 +139,9 @@ public class QwpWebSocketSender implements Sender {
     // Yields the Authorization header value presented on each WebSocket upgrade. A constant for a
     // fixed token or Basic credential; for an httpTokenProvider it pulls a freshly refreshed token,
     // so the initial connect and every reconnect re-handshake carry the current token. May be null
-    // when no auth is configured. Evaluated inside buildAndConnect's per-endpoint try, so a throwing
-    // provider (e.g. a failed silent refresh) is handled as a connect failure rather than escaping.
+    // when no auth is configured. Evaluated once per (re)connect round in buildAndConnect, before the
+    // endpoint walk (not once per endpoint); a throwing provider propagates to the connectWithRetry
+    // reconnect wrapper, which retries it within the reconnect budget and surfaces the provider's message.
     private final Supplier<String> authorizationHeaderSupplier;
     private final int autoFlushBytes;
     private final long autoFlushIntervalNanos;
@@ -2433,6 +2434,22 @@ private synchronized WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
         HttpClientException terminalUpgradeError = null;
         QwpIngressRoleRejectedException lastRoleReject = null;
         Endpoint lastEndpoint = null;
+        // Honor a close/stop that raced this (re)connect before doing any work - a token pull can make a
+        // blocking network call - mirroring the per-endpoint check at the top of the walk below.
+        if (cursorSendLoop == null ? closed : !cursorSendLoop.isRunning()) {
+            throw new LineSenderException("sender closed during connect");
+        }
+        // Resolve the Authorization header ONCE per (re)connect round, before the endpoint walk. For an
+        // httpTokenProvider this queries the provider a single time - a fresh token per handshake round, as the
+        // javadoc documents - NOT once per endpoint: a token-provider failure is cluster-wide (a failed silent
+        // refresh, or not signed in), not a per-endpoint transport fault, so re-querying it per endpoint would
+        // hammer the token endpoint with the same dead credential and mislabel the failure as "all endpoints
+        // unreachable". A throw here propagates to the connectWithRetry reconnect wrapper, which retries it
+        // within the reconnect budget like any other connect failure and surfaces the provider's own message,
+        // so a transient failed refresh recovers and only a persistent one terminates the sender (this
+        // transport's documented reconnect model). Mirrors QwpQueryClient, which likewise resolves the
+        // credential once before its endpoint walk.
+        final String authHeader = authorizationHeaderSupplier == null ? null : authorizationHeaderSupplier.get();
         while (true) {
             if (cursorSendLoop == null ? closed : !cursorSendLoop.isRunning()) {
                 throw new LineSenderException("sender closed during connect");
@@ -2451,11 +2468,9 @@ private synchronized WebSocketClient buildAndConnect(ReconnectSupplier ctx) {
                 newClient.setQwpRequestDurableAck(requestDurableAck);
                 newClient.connect(ep.host, ep.port);
                 int upgradeTimeoutMs = (int) Math.min(authTimeoutMs, Integer.MAX_VALUE);
-                // Pull the current Authorization header for this handshake. For an httpTokenProvider
-                // this re-queries the provider, so a reconnect presents a freshly refreshed token. A
-                // provider that throws here (a failed silent refresh) is caught below as a connect
-                // failure for this endpoint and retried within the reconnect budget.
-                String authHeader = authorizationHeaderSupplier == null ? null : authorizationHeaderSupplier.get();
+                // Present the header resolved once above for this handshake. On a failover to a later endpoint
+                // the same round's token is reused (a token is cluster-wide), so the provider is queried once
+                // per reconnect round, not once per endpoint.
                 newClient.upgrade(WRITE_PATH, upgradeTimeoutMs, authHeader);
             } catch (HttpClientException e) {
                 HttpClientException classified = QwpUpgradeFailures.classify(newClient, ep.host, ep.port, e);

core/src/test/java/io/questdb/client/test/cutlass/qwp/client/WebSocketTokenProviderTest.java

@@ -25,6 +25,7 @@
 package io.questdb.client.test.cutlass.qwp.client;
 
 import io.questdb.client.Sender;
+import io.questdb.client.cutlass.auth.OidcAuthException;
 import io.questdb.client.test.cutlass.qwp.websocket.TestWebSocketServer;
 import org.junit.Assert;
 import org.junit.Test;
@@ -135,6 +136,47 @@ public void testStaticTokenStillSuppliedOverWebSocket() throws Exception {
         });
     }
 
+    @Test
+    public void testThrowingProviderResolvedOncePerConnectRound() throws Exception {
+        assertMemoryLeak(() -> {
+            // A token-provider failure (a failed silent refresh, or not signed in) is cluster-wide, not a
+            // per-endpoint transport fault. The credential is resolved once before the endpoint walk, so the
+            // provider is queried exactly once per connect round even across a multi-endpoint failover, and the
+            // provider's own error reaches the caller instead of being masked as "all endpoints unreachable".
+            AtomicInteger calls = new AtomicInteger();
+            try (TestWebSocketServer server = new TestWebSocketServer(new AckHandler())) {
+                int port = server.getPort();
+                server.start();
+                Assert.assertTrue(server.awaitStart(5, TimeUnit.SECONDS));
+
+                // two endpoints at the same reachable server (distinct host strings, so not rejected as
+                // duplicates) - a pre-fix per-endpoint pull would query the provider twice for one connect
+                try {
+                    Sender.builder(Sender.Transport.WEBSOCKET)
+                            .address("localhost:" + port)
+                            .address("127.0.0.1:" + port)
+                            .httpTokenProvider(() -> {
+                                calls.incrementAndGet();
+                                throw new OidcAuthException("no token has been obtained yet; call signIn()");
+                            })
+                            .build();
+                    Assert.fail("expected build() to fail when the token provider throws");
+                } catch (OidcAuthException e) {
+                    // the provider's own error surfaces directly, not wrapped as a transport failure
+                    String msg = e.getMessage();
+                    Assert.assertTrue("expected the provider's message, got: " + msg,
+                            msg.contains("no token has been obtained yet"));
+                    Assert.assertFalse("a provider failure must not be mislabeled as unreachable, got: " + msg,
+                            msg.contains("unreachable"));
+                } catch (Exception e) {
+                    Assert.fail("expected the provider's OidcAuthException to surface, got: " + e);
+                }
+                // queried once per connect round, not once per endpoint (pre-fix this would be 2)
+                Assert.assertEquals(1, calls.get());
+            }
+        });
+    }
+
     @Test
     public void testUsernamePasswordStillSuppliedOverWebSocket() throws Exception {
         assertMemoryLeak(() -> {