Add async initial connect for cursor SF sender

initial_connect_retry already had a SYNC mode that retried on the user
thread up to reconnect_max_duration_millis, but that contradicts SF's
core promise of decoupling the producer from network state: at startup
the application thread is blocked on a wire that may never come up.

This change extends initial_connect_retry with a third value, async,
which:

- Returns from Sender.fromConfig immediately. The producer thread can
  call at()/atNow()/flush() right away; rows accumulate in the cursor
  SF engine while the I/O thread runs the connect retry loop in the
  background.
- Reuses the existing per-outage retry/backoff/cap/auth-terminal
  machinery from CursorWebSocketSendLoop. fail() is refactored into a
  shared connectLoop(initial, phase) helper used by both the in-flight
  reconnect path (phase="reconnect") and the new attemptInitialConnect
  path (phase="initial connect"). The I/O loop's first iteration now
  drives initial connect when the constructor was handed a null client.
- Surfaces terminal failures (auth/upgrade reject, budget exhaustion)
  through the existing SenderError dispatcher rather than throwing
  from the constructor. close() still rethrows the latched terminal so
  users without a custom error_handler still see the failure.

The new InitialConnectMode enum (OFF, SYNC, ASYNC) replaces the
internal boolean. The legacy initialConnectRetry(boolean) builder
method is preserved as a back-compat shim that maps false to OFF and
true to SYNC, and the config string parser accepts off/false, on/true/
sync, and async as values for initial_connect_retry.

Connectivity classification:

When the connect-retry budget exhausts, the SenderError now tags the
failure by what was actually observed on the wire so users can tell a
config typo apart from a transient blip:

- never-connected-budget-exhausted: ... -- never reached the server
  (check addr/port/firewall) -- when the I/O loop has not once
  installed a live, upgraded WebSocket. Most likely a typo, wrong
  port, firewall block, or server not yet deployed.
- connection-lost-budget-exhausted: ... -- server unreachable since
  last connect (transient) -- when the loop did connect at least
  once and the wire dropped after.

A sticky volatile boolean hasEverConnected on the I/O loop tracks the
distinction; CursorWebSocketSendLoop.hasEverConnected() and
QwpWebSocketSender.wasEverConnected() expose it for handlers that want
to branch programmatically without parsing the message string. The
flag is set in the constructor for SYNC/OFF modes (which are handed a
live client) and inside swapClient on every successful connect for
ASYNC and reconnect paths.

Auth/upgrade rejects keep the existing ws-upgrade-failed message and
SECURITY_ERROR category -- the failure cause is already self-describing
and disambiguation is unnecessary.

Tests:

InitialConnectAsyncTest covers six cases:
- fromConfig returns immediately in async mode with no server reachable;
- buffered rows are delivered once a late-arriving server starts;
- never-connected budget exhaustion uses the never-connected tag and
  wasEverConnected() returns false;
- 401 upgrade reject delivers SECURITY_ERROR via the inbox short of the
  cap;
- connect-then-disconnect budget exhaustion uses the connection-lost
  tag and wasEverConnected() remains true;
- OFF/SYNC modes report wasEverConnected()==true the moment the sender
  is visible to the caller.

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

Author: bluestreak01

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

@@ -569,6 +569,36 @@ enum SfDurability {
         APPEND
     }
 
+    /**
+     * Initial-connect behavior for the WebSocket cursor SF transport.
+     * <ul>
+     *   <li>{@link #OFF} — single attempt on the user thread; a startup
+     *       failure throws immediately. Default; correct for fail-fast
+     *       deployments where a misconfigured host should not stall app
+     *       startup.</li>
+     *   <li>{@link #SYNC} — same retry loop the in-flight reconnect path
+     *       uses, but it runs on the user thread inside {@code fromConfig}.
+     *       Blocks up to {@code reconnect_max_duration_millis}. Auth/upgrade
+     *       failures stay terminal. Useful when the server is expected to
+     *       come up shortly after the producer and the producer is willing
+     *       to wait.</li>
+     *   <li>{@link #ASYNC} — {@code fromConfig} returns immediately with an
+     *       unconnected sender; the I/O thread runs the same retry loop in
+     *       the background. The user thread can call {@code at()} /
+     *       {@code flush()} immediately; rows accumulate in the cursor SF
+     *       engine until the wire is up. A connect-budget exhaustion or a
+     *       terminal upgrade failure is delivered to the async error inbox
+     *       as a {@link io.questdb.client.SenderError} (no synchronous
+     *       throw on the user call site). Wire {@code error_handler=...}
+     *       to observe these.</li>
+     * </ul>
+     */
+    enum InitialConnectMode {
+        OFF,
+        SYNC,
+        ASYNC
+    }
+
     final class LineSenderBuilder {
         private static final int AUTO_FLUSH_DISABLED = 0;
         private static final int DEFAULT_AUTO_FLUSH_INTERVAL_MILLIS = 1_000;
@@ -694,11 +724,11 @@ public int getTimeout() {
         private long reconnectMaxDurationMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private long reconnectInitialBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private long reconnectMaxBackoffMillis = PARAMETER_NOT_SET_EXPLICITLY;
-        // When true, the initial connect goes through the same
-        // backoff/cap/auth-terminal retry path as reconnect. Default false:
-        // a misconfigured host or down server should fail fast at startup,
-        // not after the cap. Auth failures stay terminal even with retry on.
-        private boolean initialConnectRetry = false;
+        // Drives the initial-connect strategy. OFF is fail-fast (default).
+        // SYNC retries on the user thread up to the reconnect cap. ASYNC
+        // returns immediately and lets the I/O thread retry in the
+        // background, surfacing terminal failures via the error inbox.
+        private InitialConnectMode initialConnectMode = InitialConnectMode.OFF;
         // Per-append deadline for SF appendBlocking spin-then-throw. Used to
         // be a hardcoded 30s constant; expose so tight-SLA users can lower
         // and offline-tolerant users can raise.
@@ -1112,7 +1142,7 @@ public Sender build() {
                             actualReconnectMaxDurationMillis,
                             actualReconnectInitialBackoffMillis,
                             actualReconnectMaxBackoffMillis,
-                            initialConnectRetry,
+                            initialConnectMode,
                             errorHandler,
                             actualErrorInboxCapacity
                     );
@@ -1966,12 +1996,34 @@ public LineSenderBuilder reconnectMaxBackoffMillis(long millis) {
          * sit retrying for 5 minutes. Set true if your deployment expects
          * the server to come up shortly after the sender. Auth failures
          * (HTTP 401/403/non-101) stay terminal in either mode.
+         * <p>
+         * For non-blocking startup (the producer thread returns immediately
+         * and the I/O thread retries in the background), use
+         * {@link #initialConnectMode(InitialConnectMode)} with
+         * {@link InitialConnectMode#ASYNC}.
          */
         public LineSenderBuilder initialConnectRetry(boolean enabled) {
             if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
                 throw new LineSenderException("initial_connect_retry is only supported for WebSocket transport");
             }
-            this.initialConnectRetry = enabled;
+            this.initialConnectMode = enabled ? InitialConnectMode.SYNC : InitialConnectMode.OFF;
+            return this;
+        }
+
+        /**
+         * Three-way control over initial-connect behavior — see
+         * {@link InitialConnectMode} for the value semantics. WebSocket
+         * transport only. Replaces {@link #initialConnectRetry(boolean)}
+         * for users who want the {@link InitialConnectMode#ASYNC} mode.
+         */
+        public LineSenderBuilder initialConnectMode(InitialConnectMode mode) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("initial_connect_mode is only supported for WebSocket transport");
+            }
+            if (mode == null) {
+                throw new LineSenderException("initial_connect_mode cannot be null");
+            }
+            this.initialConnectMode = mode;
             return this;
         }
 
@@ -2594,12 +2646,15 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                         throw new LineSenderException("initial_connect_retry is only supported for WebSocket transport");
                     }
                     pos = getValue(configurationString, pos, sink, "initial_connect_retry");
-                    if (Chars.equalsIgnoreCase("on", sink) || Chars.equalsIgnoreCase("true", sink)) {
-                        initialConnectRetry(true);
+                    if (Chars.equalsIgnoreCase("on", sink) || Chars.equalsIgnoreCase("true", sink)
+                            || Chars.equalsIgnoreCase("sync", sink)) {
+                        initialConnectMode(InitialConnectMode.SYNC);
                     } else if (Chars.equalsIgnoreCase("off", sink) || Chars.equalsIgnoreCase("false", sink)) {
-                        initialConnectRetry(false);
+                        initialConnectMode(InitialConnectMode.OFF);
+                    } else if (Chars.equalsIgnoreCase("async", sink)) {
+                        initialConnectMode(InitialConnectMode.ASYNC);
                     } else {
-                        throw new LineSenderException("invalid initial_connect_retry [value=").put(sink).put(", allowed-values=[on, off, true, false]]");
+                        throw new LineSenderException("invalid initial_connect_retry [value=").put(sink).put(", allowed-values=[on, off, true, false, sync, async]]");
                     }
                 } else if (Chars.equals("sf_append_deadline_millis", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {

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

@@ -182,10 +182,15 @@ public class QwpWebSocketSender implements Sender {
             CursorWebSocketSendLoop.DEFAULT_RECONNECT_INITIAL_BACKOFF_MILLIS;
     private long reconnectMaxBackoffMillis =
             CursorWebSocketSendLoop.DEFAULT_RECONNECT_MAX_BACKOFF_MILLIS;
-    // false → startup connect failure is immediately terminal (default).
-    // true  → startup connect goes through the same retry-with-backoff
+    // OFF   → startup connect failure is immediately terminal (default).
+    // SYNC  → startup connect goes through the same retry-with-backoff
     //         loop as in-flight reconnect; auth failures still terminal.
-    private boolean initialConnectRetry = false;
+    // ASYNC → user thread does not connect at all. The I/O thread runs
+    //         the same retry loop in the background; terminal failures
+    //         (auth/upgrade reject, budget exhaustion) are delivered
+    //         to the SenderError dispatcher rather than thrown from the
+    //         constructor.
+    private Sender.InitialConnectMode initialConnectMode = Sender.InitialConnectMode.OFF;
     // Orphan-slot drainer pool. Non-null only when the builder requested
     // drain_orphans=true AND we have a slot path to scan against. Closed
     // alongside the cursor send loop in close().
@@ -361,14 +366,15 @@ public static QwpWebSocketSender connect(
                 maxSchemasPerConnection, requestDurableAck, cursorEngine,
                 closeFlushTimeoutMillis, reconnectMaxDurationMillis,
                 reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
-                false);
+                Sender.InitialConnectMode.OFF);
     }
 
     /**
-     * Master connect overload — also accepts {@code initialConnectRetry}.
-     * When true, the initial connect goes through the same retry loop as
-     * in-flight reconnect (backoff + cap + auth-terminal). When false
-     * (default), a startup connect failure is immediately terminal.
+     * Master connect overload — also accepts {@code initialConnectMode}.
+     * See {@link Sender.InitialConnectMode} for the value semantics:
+     * {@code OFF} fails fast (default), {@code SYNC} retries on the user
+     * thread up to the reconnect cap, {@code ASYNC} returns immediately
+     * and lets the I/O thread retry in the background.
      */
     public static QwpWebSocketSender connect(
             String host,
@@ -386,14 +392,14 @@ public static QwpWebSocketSender connect(
             long reconnectMaxDurationMillis,
             long reconnectInitialBackoffMillis,
             long reconnectMaxBackoffMillis,
-            boolean initialConnectRetry
+            Sender.InitialConnectMode initialConnectMode
     ) {
         return connect(host, port, tlsConfig, autoFlushRows, autoFlushBytes,
                 autoFlushIntervalNanos, inFlightWindowSize, authorizationHeader,
                 maxSchemasPerConnection, requestDurableAck, cursorEngine,
                 closeFlushTimeoutMillis, reconnectMaxDurationMillis,
                 reconnectInitialBackoffMillis, reconnectMaxBackoffMillis,
-                initialConnectRetry, null, SenderErrorDispatcher.DEFAULT_CAPACITY);
+                initialConnectMode, null, SenderErrorDispatcher.DEFAULT_CAPACITY);
     }
 
     /**
@@ -417,7 +423,7 @@ public static QwpWebSocketSender connect(
             long reconnectMaxDurationMillis,
             long reconnectInitialBackoffMillis,
             long reconnectMaxBackoffMillis,
-            boolean initialConnectRetry,
+            Sender.InitialConnectMode initialConnectMode,
             SenderErrorHandler errorHandler,
             int errorInboxCapacity
     ) {
@@ -432,7 +438,9 @@ public static QwpWebSocketSender connect(
             sender.reconnectMaxDurationMillis = reconnectMaxDurationMillis;
             sender.reconnectInitialBackoffMillis = reconnectInitialBackoffMillis;
             sender.reconnectMaxBackoffMillis = reconnectMaxBackoffMillis;
-            sender.initialConnectRetry = initialConnectRetry;
+            sender.initialConnectMode = initialConnectMode == null
+                    ? Sender.InitialConnectMode.OFF
+                    : initialConnectMode;
             if (errorHandler != null) {
                 sender.setErrorHandler(errorHandler);
             }
@@ -1584,6 +1592,20 @@ public QwpWebSocketSender uuidColumn(CharSequence columnName, long lo, long hi)
         return this;
     }
 
+    /**
+     * True iff this sender has at least once installed a live (connected
+     * + upgraded) WebSocket. Sticky — once true, stays true even after a
+     * subsequent disconnect. Lets a {@link SenderErrorHandler}
+     * disambiguate a "never reached the server" budget exhaustion (likely
+     * a config typo or firewall block) from a "lost connection after we
+     * were up" failure (likely transient). Returns {@code false} if no
+     * I/O loop is running.
+     */
+    public boolean wasEverConnected() {
+        CursorWebSocketSendLoop l = cursorSendLoop;
+        return l != null && l.hasEverConnected();
+    }
+
     private void atMicros(long timestampMicros) {
         // Add designated timestamp column (empty name for designated timestamp)
         // Use cached reference to avoid hashmap lookup per row
@@ -1679,15 +1701,30 @@ private void ensureConnected() {
         if (cursorEngine == null) {
             throw new LineSenderException("cursor engine must be attached before connect");
         }
-        if (initialConnectRetry) {
-            client = CursorWebSocketSendLoop.connectWithRetry(
-                    this::buildAndConnect,
-                    reconnectMaxDurationMillis,
-                    reconnectInitialBackoffMillis,
-                    reconnectMaxBackoffMillis,
-                    "initial connect");
-        } else {
-            client = buildAndConnect();
+        switch (initialConnectMode) {
+            case SYNC:
+                client = CursorWebSocketSendLoop.connectWithRetry(
+                        this::buildAndConnect,
+                        reconnectMaxDurationMillis,
+                        reconnectInitialBackoffMillis,
+                        reconnectMaxBackoffMillis,
+                        "initial connect");
+                break;
+            case ASYNC:
+                // Defer the actual connect to the I/O thread. The user thread
+                // returns immediately; rows accumulate in the cursor SF engine.
+                // Encoder stays at its default (V1 — the only supported wire
+                // version today). When v2+ ships, frames written before the
+                // first successful connect will commit to V1 because cursor
+                // segments are immutable. Auth/upgrade rejects and budget
+                // exhaustion are surfaced via the error inbox by the I/O
+                // thread, not thrown here.
+                client = null;
+                break;
+            case OFF:
+            default:
+                client = buildAndConnect();
+                break;
         }
 
         try {
@@ -1708,13 +1745,27 @@ private void ensureConnected() {
             cursorSendLoop.setErrorDispatcher(errorDispatcher);
             cursorSendLoop.start();
         } catch (Throwable t) {
-            client.close();
-            client = null;
+            if (client != null) {
+                client.close();
+                client = null;
+            }
             throw new LineSenderException(
                     "Failed to start cursor I/O thread for " + host + ":" + port, t);
         }
 
-        encoder.setVersion((byte) client.getServerQwpVersion());
+        if (client != null) {
+            encoder.setVersion((byte) client.getServerQwpVersion());
+            LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}, qwpVersion={}]",
+                    host, port, inFlightWindowSize, client.getServerQwpVersion());
+        } else {
+            // Async mode: I/O thread will drive the connect. Encoder uses
+            // its default version (V1). Schema state still gets reset for
+            // consistency with the sync path; the post-connect replay path
+            // does not need a producer-side reset signal because every
+            // cursor frame is self-sufficient.
+            LOG.info("Async initial connect deferred to I/O thread [host={}, port={}, windowSize={}]",
+                    host, port, inFlightWindowSize);
+        }
         // Server starts fresh on each connection — discard any schema IDs
         // retained from prior state. Cursor frames are self-sufficient (every
         // frame carries full schema + full symbol-dict delta from id 0), so
@@ -1723,8 +1774,6 @@ private void ensureConnected() {
         connectionError.set(null);
 
         connected = true;
-        LOG.info("Connected to WebSocket [host={}, port={}, windowSize={}, qwpVersion={}]",
-                host, port, inFlightWindowSize, client.getServerQwpVersion());
     }
 
     /**