Fail fast on QWP token-provider failures

Address three review follow-ups in the OIDC device-flow client.

QwpQueryClient resolved the bearer token inside the per-endpoint
upgrade, so a token-provider failure (not signed in, a failed silent
refresh, a rejected token) was caught as a per-endpoint transport
error and reported as "all QWP endpoints unreachable", and the
provider was queried once per endpoint. Resolve the header once
before the endpoint walk in connect() and reconnectViaTracker() and
thread it through connectToEndpoint/runUpgradeWithTimeout, so a
provider failure - which is cluster-wide - propagates directly with
the provider's own message and the provider is queried once per
connect/reconnect. Strengthen testThrowingProviderFailsConnect to
assert the provider's own exception surfaces, not a wrapped
"unreachable" error.

Validate Builder.httpTimeoutMillis: a non-positive value gave an
already-expired read deadline and an unbounded recv(int), so reject
it like Sender.Builder already does. Add a test.

Drop the always-true scopeEncoded null check in tryRefresh: build()
defaults scope to DEFAULT_SCOPE, so append it unconditionally like
runDeviceFlow().

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

@@ -1294,9 +1294,7 @@ private boolean tryRefresh() {
         formSink.putAscii("grant_type=").putAscii(GRANT_TYPE_REFRESH_TOKEN_ENCODED);
         appendParam(formSink, "refresh_token", refreshToken);
         appendEncodedParam(formSink, "client_id", clientIdEncoded);
-        if (scopeEncoded != null) {
-            appendEncodedParam(formSink, "scope", scopeEncoded);
-        }
+        appendEncodedParam(formSink, "scope", scopeEncoded);
         if (audienceEncoded != null) {
             appendEncodedParam(formSink, "audience", audienceEncoded);
         }
@@ -1421,6 +1419,9 @@ public Builder groupsInToken(boolean groupsInToken) {
         }
 
         public Builder httpTimeoutMillis(int httpTimeoutMillis) {
+            if (httpTimeoutMillis <= 0) {
+                throw new OidcAuthException("httpTimeoutMillis must be positive");
+            }
             this.httpTimeoutMillis = httpTimeoutMillis;
             return this;
         }

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

@@ -684,6 +684,11 @@ public void close() {
      * observed so callers can distinguish "no primary available" from "all
      * endpoints unreachable" (the latter surfaces as a plain
      * {@link HttpClientException}).
+     * <p>
+     * A configured token provider is queried once here, before the walk. A
+     * provider failure (not signed in, a failed silent refresh, a rejected
+     * token) is cluster-wide, so it fails fast with the provider's own error
+     * rather than being retried across endpoints as a transport failure.
      */
     public synchronized void connect() {
         if (closedFlag.get()) {
@@ -702,14 +707,20 @@ public synchronized void connect() {
         QwpServerInfo lastObservedMismatch = null;
         QwpIngressRoleRejectedException lastUpgradeRoleReject = null;
         Throwable lastTransportError = null;
+        // Resolve the bearer credential once, before the endpoint walk: a token is cluster-wide, so a
+        // token-provider failure (not signed in, a failed silent refresh, a rejected token) is not a
+        // per-endpoint transport fault. Resolving here lets it propagate as the provider's own error
+        // instead of being folded into "all endpoints unreachable", and avoids re-querying the provider
+        // once per endpoint.
+        String authHeader = resolveAuthorizationHeader();
         while (true) {
             int i = hostTracker.pickNext();
             if (i < 0) {
                 break;
             }
             Endpoint ep = endpoints.get(i);
             try {
-                connectToEndpoint(ep);
+                connectToEndpoint(ep, authHeader);
             } catch (QwpAuthFailedException ae) {
                 cleanupFailedConnect();
                 throw ae;
@@ -1401,7 +1412,7 @@ private void cleanupFailedConnect() {
         currentEndpointIndex = -1;
     }
 
-    private void connectToEndpoint(Endpoint ep) {
+    private void connectToEndpoint(Endpoint ep, String authHeader) {
         if (tlsEnabled) {
             webSocketClient = WebSocketClientFactory.newTlsInstance(
                     new ClientTlsConfiguration(trustStorePath, trustStorePassword, tlsValidationMode));
@@ -1412,7 +1423,7 @@ private void connectToEndpoint(Endpoint ep) {
         webSocketClient.setQwpClientId(clientId != null ? clientId : defaultClientId());
         webSocketClient.setQwpAcceptEncoding(buildAcceptEncodingHeader());
         webSocketClient.setQwpMaxBatchRows(maxBatchRows);
-        runUpgradeWithTimeout(ep);
+        runUpgradeWithTimeout(ep, authHeader);
         negotiatedQwpVersion = webSocketClient.getServerQwpVersion();
         negotiatedZstdLevel = webSocketClient.getServerNegotiatedZstdLevel();
 
@@ -1730,6 +1741,10 @@ private void reconnectViaTracker() {
         QwpServerInfo lastMismatch = null;
         Throwable lastError = null;
         boolean retriedAfterReset = false;
+        // Resolve the bearer credential once per reconnect, before the endpoint walk, for the same
+        // reason as connect(): a provider failure is cluster-wide, so surface it directly rather than
+        // as a per-endpoint transport error retried across every host.
+        String authHeader = resolveAuthorizationHeader();
         while (true) {
             int i = hostTracker.pickNext();
             if (i < 0) {
@@ -1742,7 +1757,7 @@ private void reconnectViaTracker() {
             }
             Endpoint ep = endpoints.get(i);
             try {
-                connectToEndpoint(ep);
+                connectToEndpoint(ep, authHeader);
             } catch (QwpAuthFailedException ae) {
                 cleanupFailedConnect();
                 throw ae;
@@ -1788,12 +1803,11 @@ private void reconnectViaTracker() {
     }
 
     private String resolveAuthorizationHeader() {
-        // With a token provider, re-query it at each upgrade so a reconnect
-        // presents a freshly refreshed token; validateToken rejects a
-        // null/empty/blank return, or one carrying a control or non-ASCII
-        // character, before it reaches the "Bearer " header. A provider that
-        // throws (a failed silent refresh) propagates and fails this connection
-        // attempt, matching the QWP ingress sender.
+        // With a token provider, query it once per connect()/reconnect (the caller resolves before the
+        // endpoint walk) so a reconnect presents a freshly refreshed token; validateToken rejects a
+        // null/empty/blank return, or one carrying a control or non-ASCII character, before it reaches
+        // the "Bearer " header. A provider that throws (a failed silent refresh, or not signed in yet)
+        // propagates out of connect()/reconnect with its own message, matching the QWP ingress sender.
         if (tokenProvider != null) {
             CharSequence token = tokenProvider.getToken();
             HttpTokenProvider.validateToken(token);
@@ -1802,9 +1816,8 @@ private String resolveAuthorizationHeader() {
         return authorizationHeader;
     }
 
-    private void runUpgradeWithTimeout(Endpoint ep) {
+    private void runUpgradeWithTimeout(Endpoint ep, String authHeader) {
         int timeoutMs = (int) Math.min(authTimeoutMs, Integer.MAX_VALUE);
-        String authHeader = resolveAuthorizationHeader();
         try {
             webSocketClient.connect(ep.host, ep.port);
             webSocketClient.upgrade(DEFAULT_ENDPOINT_PATH, timeoutMs, authHeader);

core/src/test/java/io/questdb/client/test/cutlass/auth/OidcDeviceAuthTest.java

@@ -266,6 +266,20 @@ public void testBuilderRejectsMissingRequiredOptions() {
         }
     }
 
+    @Test(timeout = 30_000)
+    public void testBuilderRejectsNonPositiveHttpTimeout() {
+        // every other timing input is clamped; a non-positive HTTP timeout yields an already-expired read
+        // deadline and an unbounded recv(int), so the setter rejects it (matching Sender.Builder)
+        for (int bad : new int[]{0, -1}) {
+            try {
+                OidcDeviceAuth.builder().httpTimeoutMillis(bad);
+                Assert.fail("expected httpTimeoutMillis(" + bad + ") to be rejected");
+            } catch (OidcAuthException e) {
+                Assert.assertTrue(e.getMessage(), e.getMessage().contains("httpTimeoutMillis"));
+            }
+        }
+    }
+
     @Test(timeout = 30_000)
     public void testBuilderRejectsSplitOriginEndpoints() {
         // the token and device authorization endpoints are on different origins; RFC 8628 co-locates them

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

@@ -128,9 +128,9 @@ public void testProviderSynthesizesBearerHeader() {
 
     @Test(timeout = 15_000)
     public void testProviderTokenSentOnRealUpgrade() throws Exception {
-        // drive the REAL connect path (runUpgradeWithTimeout -> resolveAuthorizationHeader), not the test
-        // hook: the upgrade request must carry the freshly pulled "Bearer <token>". The mock answers 404
-        // (not auth-failed, not terminal) so connect() fails fast after the header was already sent.
+        // drive the REAL connect path (connect() -> resolveAuthorizationHeader -> runUpgradeWithTimeout),
+        // not the test hook: the upgrade request must carry the freshly pulled "Bearer <token>". The mock
+        // answers 404 (not auth-failed, not terminal) so connect() fails fast after the header was sent.
         List<String> authHeaders = Collections.synchronizedList(new ArrayList<>());
         ServerSocket listener = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
         int port = listener.getLocalPort();
@@ -213,8 +213,8 @@ public void testSettingBearerTokenThenProviderConflicts() {
     @Test(timeout = 10_000)
     public void testThrowingProviderFailsConnect() throws Exception {
         // a provider that throws must fail the connection attempt on the REAL connect path:
-        // resolveAuthorizationHeader runs inside runUpgradeWithTimeout, before the socket connect, so the
-        // throw aborts the upgrade; connect() exhausts the single endpoint and surfaces the provider failure
+        // resolveAuthorizationHeader runs once before the endpoint walk, so the throw propagates straight
+        // out of connect() as the provider's own error (not wrapped as "all endpoints unreachable")
         try (
                 ServerSocket listener = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
                 QwpQueryClient client = QwpQueryClient.fromConfig(
@@ -227,7 +227,11 @@ public void testThrowingProviderFailsConnect() throws Exception {
                 client.connect();
                 Assert.fail("a throwing provider must fail the connection attempt");
             } catch (RuntimeException expected) {
+                // the provider's own exception propagates directly (the header is resolved before the
+                // endpoint walk), not wrapped as a transport "all endpoints unreachable" error
+                Assert.assertTrue(expected.getClass().getName(), expected instanceof LineSenderException);
                 Assert.assertTrue(expected.getMessage(), expected.getMessage().contains("provider down"));
+                Assert.assertFalse(expected.getMessage(), expected.getMessage().contains("unreachable"));
             }
         }
     }