QWP egress token provider and OIDC API rename

Add QwpQueryClient.withBearerTokenProvider(HttpTokenProvider): the
egress query client now accepts an on-demand token provider, the same
HttpTokenProvider the ingress Sender uses, so OidcDeviceAuth::getToken
plugs into both. runUpgradeWithTimeout resolves the Authorization
header at every WebSocket upgrade, so the initial connect and each
failover reconnect present a freshly refreshed token; a throwing
provider fails that connection attempt, matching the ingress sender.
The setter is mutually exclusive with withBearerToken/withBasicAuth
and validates each token before it reaches the header.

Rename the OidcDeviceAuth methods so the safe, non-blocking accessor
takes the plain name and the interactive step is explicit:
  getToken() is now signIn() (interactive sign-in, may block);
  getTokenSilently() is now getToken() (cached/refresh, never prompts).
getAuthorizationHeaderValue() keeps its blocking behavior by calling
signIn(). Update the tests, examples, and doc references across the
client. The Python client is left unchanged for now.

Print the egress timestamp column as a formatted Instant in
OIDCAuthExample instead of the raw microsecond long.

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

Author: glasstiger

core/src/main/java/io/questdb/client/HttpTokenProvider.java

@@ -29,7 +29,7 @@
 
 /**
  * Supplies an HTTP authentication token to a {@link Sender} on demand, so a provider returning a
- * freshly refreshed token - e.g. {@code OidcDeviceAuth::getTokenSilently} - keeps a long-lived sender
+ * freshly refreshed token - e.g. {@code OidcDeviceAuth::getToken} - keeps a long-lived sender
  * authenticated as the token rotates, without rebuilding it. Over HTTP the sender calls
  * {@link #getToken()} as it builds each request; over WebSocket it calls it once per connection
  * handshake, on the initial connect and again on every reconnect.

core/src/main/java/io/questdb/client/Sender.java

@@ -2018,7 +2018,7 @@ public LineSenderBuilder httpToken(String token) {
         /**
          * Supplies the HTTP authentication token from a provider queried as the sender builds each request,
          * instead of a fixed {@link #httpToken(String) token} captured once, so a long-lived sender follows
-         * token refreshes - e.g. an OIDC device-flow token: {@code .httpTokenProvider(auth::getTokenSilently)}.
+         * token refreshes - e.g. an OIDC device-flow token: {@code .httpTokenProvider(auth::getToken)}.
          * <br>
          * Over HTTP the provider is not called at build time: the first call happens when the first row is
          * started, then once per flush. Over WebSocket the initial connection handshake runs during

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

@@ -65,7 +65,7 @@
  * Typical use, discovering everything from the QuestDB server:
  * <pre>{@code
  * try (OidcDeviceAuth auth = OidcDeviceAuth.fromQuestDB("https://questdb.example.com:9000")) {
- *     String token = auth.getToken(); // signs in on first use, then caches and refreshes
+ *     String token = auth.signIn(); // signs in on first use, then caches and refreshes
  *     // ... use token as an HTTP Bearer header or a PG-wire _sso password ...
  * }
  * }</pre>
@@ -79,11 +79,11 @@
  *         .groupsInToken(true)
  *         .build();
  * }</pre>
- * {@link #getToken()} serves a cached token while valid, silently refreshes when a refresh token
+ * {@link #signIn()} serves a cached token while valid, silently refreshes when a refresh token
  * exists, otherwise re-runs the interactive flow. An instance lock serializes calls, so two
  * sign-ins never start at once. A sign-in waiting for the user holds that lock for the device code
- * lifetime (up to 30 minutes), so a concurrent {@link #getToken()} or {@link #clearCache()} blocks
- * behind it - but {@link #getTokenSilently()} never waits: it fails fast with an
+ * lifetime (up to 30 minutes), so a concurrent {@link #signIn()} or {@link #clearCache()} blocks
+ * behind it - but {@link #getToken()} never waits: it fails fast with an
  * {@link OidcAuthException} so a request/flush path never stalls. 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
@@ -153,8 +153,8 @@ public class OidcDeviceAuth implements QuietCloseable {
     private final StringSink formSink = new StringSink();
     private final boolean groupsInToken;
     private final int httpTimeoutMillis;
-    // serializes getToken()/getTokenSilently()/clearCache()/close(); getToken() holds it for the whole
-    // interactive flow, getTokenSilently() uses tryLock so the flush path never stalls behind a sign-in
+    // serializes signIn()/getToken()/clearCache()/close(); signIn() holds it for the whole
+    // interactive flow, getToken() uses tryLock so the flush path never stalls behind a sign-in
     private final ReentrantLock lock = new ReentrantLock();
     private final DeviceCodePrompt prompt;
     private final StringSink responseStatus = new StringSink();
@@ -351,7 +351,7 @@ public static OidcDeviceAuth fromQuestDB(String questdbUrl, DiscoveryOptions opt
     }
 
     /**
-     * Drops any cached token so the next {@link #getToken()} starts a fresh interactive sign-in.
+     * Drops any cached token so the next {@link #signIn()} starts a fresh interactive sign-in.
      */
     public void clearCache() {
         lock.lock();
@@ -368,7 +368,7 @@ public void clearCache() {
     }
 
     /**
-     * Frees the network connections and native buffers this instance holds. If a {@link #getToken()}
+     * Frees the network connections and native buffers this instance holds. If a {@link #signIn()}
      * sign-in is in flight on another thread, signals it to stop so it fails with an
      * {@link OidcAuthException} instead of polling until the device code expires. The signal is observed
      * between polls (within ~100ms while waiting out a poll interval); a poll request already in flight
@@ -378,12 +378,12 @@ public void clearCache() {
      * {@link DeviceCodePrompt} that blocks in {@code promptUser} - for example the default
      * {@link DeviceCodePrompt#openBrowser()} prompt while it hands the verification URL to the OS browser,
      * which is not bounded by the HTTP timeout: the flow holds the lock across that one-off prompt, so a
-     * racing {@code close()} waits it out too. Idempotent. After close, {@link #getToken()} and
-     * {@link #clearCache()} throw.
+     * racing {@code close()} waits it out too. Idempotent. After close, {@link #signIn()},
+     * {@link #getToken()} and {@link #clearCache()} throw.
      */
     @Override
     public void close() {
-        // flag cancellation before taking the lock: getToken() holds it for the whole flow, so signal the
+        // flag cancellation before taking the lock: signIn() holds it for the whole flow, so signal the
         // in-flight sign-in to stop via a lock-free volatile write, then acquire the lock - released by the
         // cancelled flow once it observes the flag (between polls, or after an in-flight poll returns) - and
         // free the native resources. close() never frees while a flow holds the lock, so no use-after-free
@@ -399,11 +399,11 @@ public void close() {
     }
 
     /**
-     * @return {@code "Bearer " + getToken()}, ready to use as the value of an HTTP
+     * @return {@code "Bearer " + signIn()}, ready to use as the value of an HTTP
      * {@code Authorization} header.
      */
     public String getAuthorizationHeaderValue() {
-        return "Bearer " + getToken();
+        return "Bearer " + signIn();
     }
 
     /**
@@ -415,11 +415,11 @@ public String getAuthorizationHeaderValue() {
      * @throws OidcAuthException if the interactive flow fails, times out, or the identity provider
      *                           does not return the expected token
      */
-    public String getToken() {
+    public String signIn() {
         lock.lock();
         try {
             throwIfClosed();
-            // only the kind of token getToken() actually serves counts as a cache hit; a grant that
+            // only the kind of token signIn() actually serves counts as a cache hit; a grant that
             // returned the other kind (access token when the server wants the id token, or vice versa)
             // leaves the served token null, so re-run the flow rather than report the unusable grant as
             // valid and have selectToken() throw on this and every later call
@@ -440,13 +440,13 @@ public String getToken() {
     }
 
     /**
-     * Like {@link #getToken()} but never starts the interactive device flow, never prompts, and never waits
+     * Like {@link #signIn()} but never starts the interactive device flow, never prompts, and never waits
      * on interactive input: returns the cached token while valid, silently refreshes when a refresh token is
      * available, otherwise throws. Designed for the request/flush path of a long-lived client, for example
-     * {@code Sender.builder(...).httpTokenProvider(auth::getTokenSilently)}, where an interactive prompt is
-     * inappropriate. Call {@link #getToken()} once to sign in first.
+     * {@code Sender.builder(...).httpTokenProvider(auth::getToken)}, where an interactive prompt is
+     * inappropriate. Call {@link #signIn()} once to sign in first.
      * <p>
-     * It does not wait behind an interactive {@link #getToken()} running on another thread (which would stall
+     * 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
@@ -458,9 +458,9 @@ public String getToken() {
      *                           not be refreshed without an interactive sign-in, or if a sign-in or
      *                           refresh is already in progress on another thread
      */
-    public String getTokenSilently() {
+    public String getToken() {
         throwIfClosed();
-        // never wait on the flush path: getToken()'s sign-in holds the lock for the whole device-code
+        // never wait on the flush path: signIn()'s sign-in holds the lock for the whole device-code
         // lifetime (up to 30 minutes), so tryLock and fail fast if held. A sign-in in progress means there
         // is no token to serve yet, so the caller gets a prompt exception to retry rather than a stalled
         // flush
@@ -477,9 +477,9 @@ public String getTokenSilently() {
                 if (refreshToken != null && tryRefresh()) {
                     return selectToken();
                 }
-                throw new OidcAuthException("the cached token expired and could not be refreshed without an interactive sign-in; call getToken() to sign in again");
+                throw new OidcAuthException("the cached token expired and could not be refreshed without an interactive sign-in; call signIn() to sign in again");
             }
-            throw new OidcAuthException("no token has been obtained yet; call getToken() to sign in before using getTokenSilently()");
+            throw new OidcAuthException("no token has been obtained yet; call signIn() to sign in before using getToken()");
         } finally {
             lock.unlock();
         }

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

@@ -412,7 +412,7 @@ public static AbstractLineHttpSender createLineSender(
         if (httpTokenProvider != null) {
             // The constructor already built the initial request without a token. Defer the first
             // getToken() off this build path to the first row (table()), so a provider that signs in
-            // lazily - e.g. OidcDeviceAuth::getTokenSilently - can be wired before sign-in completes,
+            // lazily - e.g. OidcDeviceAuth::getToken - can be wired before sign-in completes,
             // and the token pull stays on the use/flush path the provider documents.
             sender.httpTokenProvider = httpTokenProvider;
             sender.isTokenPending = true;
@@ -816,7 +816,7 @@ private boolean rowAdded() {
     private void stampTokenIfPending() {
         if (isTokenPending) {
             // The construct/flush path deferred the token so a lazily-signing-in provider (e.g.
-            // OidcDeviceAuth::getTokenSilently) could be wired before sign-in completed, and so a provider
+            // OidcDeviceAuth::getToken) could be wired before sign-in completed, and so a provider
             // failure never strikes after a successful send. The caller is now starting the first row, so
             // rebuild the still-empty request to carry the token before any row data goes in. Clear the
             // flag only after newRequest(true) succeeds: a pull that throws (not signed in yet, or a failed

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

@@ -25,6 +25,7 @@
 package io.questdb.client.cutlass.qwp.client;
 
 import io.questdb.client.ClientTlsConfiguration;
+import io.questdb.client.HttpTokenProvider;
 import io.questdb.client.cutlass.http.client.HttpClientException;
 import io.questdb.client.cutlass.http.client.WebSocketClient;
 import io.questdb.client.cutlass.http.client.WebSocketClientFactory;
@@ -285,6 +286,11 @@ public class QwpQueryClient implements QuietCloseable {
     private boolean tlsEnabled;
     // Only meaningful when tlsEnabled. Default is full validation against the JVM's trust store.
     private int tlsValidationMode = ClientTlsConfiguration.TLS_VALIDATION_MODE_FULL;
+    // Supplies a fresh Bearer token at each WebSocket upgrade (the initial
+    // connect and every failover reconnect), so a long-lived client follows
+    // token rotation. Mutually exclusive with the fixed authorizationHeader
+    // synthesized by withBearerToken/withBasicAuth; null when unset.
+    private HttpTokenProvider tokenProvider;
     private char[] trustStorePassword;
     private String trustStorePath;
     private volatile WebSocketClient webSocketClient;
@@ -838,11 +844,12 @@ public int getCompressionLevelForTest() {
     /**
      * Test-only hook: the synthesized {@code Authorization} header value
      * ({@code Basic ...} or {@code Bearer ...}), or null when no credentials
-     * were configured.
+     * were configured. When a token provider is configured, queries it and
+     * validates the returned token, exactly as a real upgrade would.
      */
     @TestOnly
     public String getAuthorizationHeaderForTest() {
-        return authorizationHeader;
+        return resolveAuthorizationHeader();
     }
 
     /**
@@ -1003,6 +1010,9 @@ public QwpQueryClient withAuthTimeout(long authTimeoutMs) {
      */
     public QwpQueryClient withBasicAuth(String username, String password) {
         checkPreConnect("withBasicAuth");
+        if (tokenProvider != null) {
+            throw new IllegalStateException("withBasicAuth cannot be combined with withBearerTokenProvider");
+        }
         if (username == null || password == null) {
             throw new IllegalArgumentException("username and password must not be null");
         }
@@ -1020,13 +1030,46 @@ public QwpQueryClient withBasicAuth(String username, String password) {
      */
     public QwpQueryClient withBearerToken(String token) {
         checkPreConnect("withBearerToken");
+        if (tokenProvider != null) {
+            throw new IllegalStateException("withBearerToken cannot be combined with withBearerTokenProvider");
+        }
         if (token == null) {
             throw new IllegalArgumentException("token must not be null");
         }
         this.authorizationHeader = "Bearer " + token;
         return this;
     }
 
+    /**
+     * Configures HTTP Bearer authentication with a token supplied on demand by
+     * {@code provider}, instead of the fixed token captured once by
+     * {@link #withBearerToken(String)}. The provider is queried for a fresh
+     * token at every WebSocket upgrade -- the initial {@link #connect()} and
+     * each failover reconnect -- so a long-lived client keeps working as the
+     * token rotates (for example an OIDC device-flow token:
+     * {@code .withBearerTokenProvider(auth::getToken)}).
+     * <p>
+     * {@link HttpTokenProvider#getToken()} runs on the connect and reconnect
+     * paths, so it must return promptly and must not block on interactive
+     * input; a quick silent refresh is fine. Each returned token is validated
+     * ({@link HttpTokenProvider#validateToken(CharSequence)}) before it is sent,
+     * and a provider that throws fails that connection attempt. Mutually
+     * exclusive with {@link #withBearerToken(String)} and
+     * {@link #withBasicAuth(String, String)}. Must be called before
+     * {@link #connect}.
+     */
+    public QwpQueryClient withBearerTokenProvider(HttpTokenProvider provider) {
+        checkPreConnect("withBearerTokenProvider");
+        if (provider == null) {
+            throw new IllegalArgumentException("provider must not be null");
+        }
+        if (authorizationHeader != null) {
+            throw new IllegalStateException("withBearerTokenProvider cannot be combined with withBearerToken or withBasicAuth");
+        }
+        this.tokenProvider = provider;
+        return this;
+    }
+
     /**
      * Overrides the default I/O buffer pool depth (4). Larger pools let the
      * I/O thread decode further ahead of the consumer at the cost of memory;
@@ -1744,11 +1787,27 @@ private void reconnectViaTracker() {
                         + ", lastError=" + (lastError == null ? "<none>" : lastError.getMessage()) + ']');
     }
 
+    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.
+        if (tokenProvider != null) {
+            CharSequence token = tokenProvider.getToken();
+            HttpTokenProvider.validateToken(token);
+            return "Bearer " + token;
+        }
+        return authorizationHeader;
+    }
+
     private void runUpgradeWithTimeout(Endpoint ep) {
         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, authorizationHeader);
+            webSocketClient.upgrade(DEFAULT_ENDPOINT_PATH, timeoutMs, authHeader);
         } catch (HttpClientException ex) {
             if (ex.isTimeout()) {
                 HttpClientException timeout = new HttpClientException("WebSocket upgrade to ")