Merge remote-tracking branch 'origin/main' into vi_egress

# Conflicts:
#	core/src/main/java/io/questdb/client/cutlass/http/client/WebSocketClient.java

Author: bluestreak01

.github/workflows/rebuild_native_libs.yml

@@ -109,11 +109,12 @@ jobs:
           wget -nv https://github.com/Kitware/CMake/releases/download/v3.29.2/cmake-3.29.2-linux-x86_64.tar.gz
           tar -zxf cmake-3.29.2-linux-x86_64.tar.gz
           echo "PATH=`pwd`/cmake-3.29.2-linux-x86_64/bin/:$PATH" >> "$GITHUB_ENV"
-      - name: Install JDK17 (for jni.h)
+      - name: Install GraalVM JDK 25 (for jni.h)
         run: |
-          wget -nv https://github.com/adoptium/temurin17-binaries/releases/download/jdk-17.0.11%2B9/OpenJDK17U-jdk_x64_linux_hotspot_17.0.11_9.tar.gz
-          tar xfz OpenJDK17U-jdk_x64_linux_hotspot_17.0.11_9.tar.gz
-          echo "JAVA_HOME=`pwd`/jdk-17.0.11+9/" >> "$GITHUB_ENV"
+          wget -nv -O graalvm.tar.gz https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_linux-x64_bin.tar.gz
+          mkdir graalvm
+          tar xfz graalvm.tar.gz -C graalvm --strip-components=1
+          echo "JAVA_HOME=`pwd`/graalvm" >> "$GITHUB_ENV"
       - name: Generate Makefiles
         run: |
           cd ./core
@@ -144,11 +145,12 @@ jobs:
         run: |
           yum update -y
           yum install wget nasm zstd -y
-      - name: Install JDK17 (for jni.h)
+      - name: Install GraalVM JDK 25 (for jni.h)
         run: |
-          wget -v --timeout=180 https://api.adoptium.net/v3/binary/version/jdk-17.0.11+9/linux/aarch64/jdk/hotspot/normal/eclipse
-          tar xfvz eclipse
-          echo "JAVA_HOME=`pwd`/jdk-17.0.11+9/" >> "$GITHUB_ENV"
+          wget -v --timeout=180 -O graalvm.tar.gz https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_linux-aarch64_bin.tar.gz
+          mkdir graalvm
+          tar xfz graalvm.tar.gz -C graalvm --strip-components=1
+          echo "JAVA_HOME=`pwd`/graalvm" >> "$GITHUB_ENV"
       - name: CMAKE linux-aarch64
         run: |
           cd ./core
@@ -177,10 +179,16 @@ jobs:
           sudo sysctl -w fs.file-max=500000
           sudo apt-get update -y
           sudo apt-get install -y nasm gcc-mingw-w64 g++-mingw-w64
-      - name: Download windows jni_md.h from JDK 11
+      - name: Install GraalVM JDK 25 (for jni.h)
+        run: |
+          wget -nv -O graalvm.tar.gz https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_linux-x64_bin.tar.gz
+          mkdir graalvm
+          tar xfz graalvm.tar.gz -C graalvm --strip-components=1
+          echo "JAVA_HOME=`pwd`/graalvm" >> "$GITHUB_ENV"
+      - name: Download windows jni_md.h from JDK 25
         run: |
           cd core
-          curl https://raw.githubusercontent.com/AdoptOpenJDK/openjdk-jdk11/master/src/java.base/windows/native/include/jni_md.h > $JAVA_HOME/include/jni_md.h
+          curl https://raw.githubusercontent.com/openjdk/jdk25u/master/src/java.base/windows/native/include/jni_md.h > $JAVA_HOME/include/jni_md.h
       - name: CMake Windows
         run: |
           cd core

ci/run_tests_pipeline.yaml

@@ -23,6 +23,9 @@ stages:
         pool:
           name: "Azure Pipelines"
           vmImage: "ubuntu-latest"
+        variables:
+          jdkArch: "x64"
+          imageName: "ubuntu-latest"
         steps:
           - checkout: self
             fetchDepth: 0
@@ -33,12 +36,7 @@ stages:
               git checkout FETCH_HEAD
             displayName: "Checkout PR source branch"
             condition: eq(variables['Build.Reason'], 'PullRequest')
-          - task: JavaToolInstaller@0
-            displayName: "Install Java 17"
-            inputs:
-              versionSpec: "17"
-              jdkArchitectureOption: "x64"
-              jdkSourceOption: "PreInstalled"
+          - template: setup.yaml
           - bash: mvn -f core/pom.xml javadoc:javadoc -Pjavadoc --batch-mode
             displayName: "Verify Javadoc"
 

ci/setup.yaml

@@ -1,10 +1,48 @@
 steps:
-  - task: JavaToolInstaller@0
-    displayName: "Install Java 17"
-    inputs:
-      versionSpec: "17"
-      jdkArchitectureOption: "$(jdkArch)"
-      jdkSourceOption: "PreInstalled"
+  - bash: |
+      set -eux
+      case "$(jdkArch)" in
+        arm64) ARCH="aarch64" ;;
+        *)     ARCH="x64" ;;
+      esac
+      curl -fsSL -o "$HOME/graalvm.tar.gz" \
+        "https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_linux-${ARCH}_bin.tar.gz"
+      mkdir -p "$HOME/graalvm"
+      tar -xzf "$HOME/graalvm.tar.gz" -C "$HOME/graalvm" --strip-components=1
+      echo "##vso[task.setvariable variable=JAVA_HOME]$HOME/graalvm"
+      echo "##vso[task.prependpath]$HOME/graalvm/bin"
+    displayName: "Install GraalVM JDK 25 (Linux)"
+    condition: eq(variables['Agent.OS'], 'Linux')
+
+  - bash: |
+      set -eux
+      case "$(jdkArch)" in
+        arm64) ARCH="aarch64" ;;
+        *)     ARCH="x64" ;;
+      esac
+      curl -fsSL -o "$HOME/graalvm.tar.gz" \
+        "https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_macos-${ARCH}_bin.tar.gz"
+      mkdir -p "$HOME/graalvm"
+      tar -xzf "$HOME/graalvm.tar.gz" -C "$HOME/graalvm" --strip-components=1
+      echo "##vso[task.setvariable variable=JAVA_HOME]$HOME/graalvm/Contents/Home"
+      echo "##vso[task.prependpath]$HOME/graalvm/Contents/Home/bin"
+    displayName: "Install GraalVM JDK 25 (macOS)"
+    condition: eq(variables['Agent.OS'], 'Darwin')
+
+  - powershell: |
+      $ErrorActionPreference = "Stop"
+      $url = "https://download.oracle.com/graalvm/25/latest/graalvm-jdk-25_windows-x64_bin.zip"
+      $zip = Join-Path $env:USERPROFILE "graalvm.zip"
+      $dest = Join-Path $env:USERPROFILE "graalvm"
+      Invoke-WebRequest -Uri $url -OutFile $zip -UseBasicParsing
+      if (Test-Path $dest) { Remove-Item -Recurse -Force $dest }
+      Expand-Archive -Path $zip -DestinationPath $dest -Force
+      $jdkDir = Get-ChildItem -Path $dest -Directory | Select-Object -First 1
+      $javaHome = $jdkDir.FullName
+      Write-Host "##vso[task.setvariable variable=JAVA_HOME]$javaHome"
+      Write-Host "##vso[task.prependpath]$javaHome\bin"
+    displayName: "Install GraalVM JDK 25 (Windows)"
+    condition: eq(variables['Agent.OS'], 'Windows_NT')
 
   - bash: sudo sysctl -w fs.file-max=500000
     displayName: "Increase file count on Linux"

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

@@ -619,6 +619,7 @@ public int getTimeout() {
         private PrivateKey privateKey;
         private int protocol = PARAMETER_NOT_SET_EXPLICITLY;
         private int protocolVersion = PARAMETER_NOT_SET_EXPLICITLY;
+        private boolean requestDurableAck;
         private int retryTimeoutMillis = PARAMETER_NOT_SET_EXPLICITLY;
         private boolean shouldDestroyPrivKey;
         private boolean tlsEnabled;
@@ -932,7 +933,8 @@ public Sender build() {
                         actualAutoFlushIntervalNanos,
                         actualInFlightWindowSize,
                         wsAuthHeader,
-                        actualMaxSchemasPerConnection
+                        actualMaxSchemasPerConnection,
+                        requestDurableAck
                 );
             }
 
@@ -1483,6 +1485,27 @@ public LineSenderBuilder protocolVersion(int protocolVersion) {
             return this;
         }
 
+        /**
+         * Opts the connection in for STATUS_DURABLE_ACK frames. When enabled,
+         * servers with primary replication will emit per-table durable-upload
+         * watermarks as WAL data reaches the object store.
+         * <p>
+         * This setting is only supported for WebSocket transport.
+         * <p>
+         * Observe durable progress via
+         * {@link QwpWebSocketSender#getHighestDurableSeqTxn(CharSequence)}.
+         *
+         * @param enabled true to request durable ACKs
+         * @return this instance for method chaining
+         */
+        public LineSenderBuilder requestDurableAck(boolean enabled) {
+            if (protocol != PARAMETER_NOT_SET_EXPLICITLY && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
+            }
+            this.requestDurableAck = enabled;
+            return this;
+        }
+
         /**
          * Configures the maximum time the Sender will spend retrying upon receiving a recoverable error from the server.
          * <br>
@@ -1875,6 +1898,18 @@ private LineSenderBuilder fromConfig(CharSequence configurationString) {
                     pos = getValue(configurationString, pos, sink, "in_flight_window");
                     int windowSize = parseIntValue(sink, "in_flight_window");
                     inFlightWindowSize(windowSize);
+                } else if (Chars.equals("request_durable_ack", sink)) {
+                    if (protocol != PROTOCOL_WEBSOCKET) {
+                        throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
+                    }
+                    pos = getValue(configurationString, pos, sink, "request_durable_ack");
+                    if (Chars.equalsIgnoreCase("on", sink)) {
+                        requestDurableAck(true);
+                    } else if (Chars.equalsIgnoreCase("off", sink)) {
+                        requestDurableAck(false);
+                    } else {
+                        throw new LineSenderException("invalid request_durable_ack [value=").put(sink).put(", allowed-values=[on, off]]");
+                    }
                 } else if (Chars.equals("max_schemas_per_connection", sink)) {
                     if (protocol != PROTOCOL_WEBSOCKET) {
                         throw new LineSenderException("max_schemas_per_connection is only supported for WebSocket transport");
@@ -1972,6 +2007,9 @@ private void validateParameters() {
                         .put(", requestedCapacity=").put(bufferCapacity)
                         .put("]");
             }
+            if (requestDurableAck && protocol != PROTOCOL_WEBSOCKET) {
+                throw new LineSenderException("request_durable_ack is only supported for WebSocket transport");
+            }
             if (protocol == PROTOCOL_HTTP) {
                 if (httpClientConfiguration.getMaximumRequestBufferSize() < httpClientConfiguration.getInitialRequestBufferSize()) {
                     throw new LineSenderException("maximum buffer capacity cannot be less than initial buffer capacity ")

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

@@ -117,6 +117,8 @@ public abstract class WebSocketClient implements QuietCloseable {
     // down to its own hard limit.
     private int qwpMaxBatchRows;
     private int qwpMaxVersion = 1;
+    // Opt-in for STATUS_DURABLE_ACK frames; sent as X-QWP-Request-Durable-Ack: true
+    private boolean qwpRequestDurableAck;
     // Receive buffer (native memory)
     private long recvBufPtr;
     private int recvBufSize;
@@ -422,6 +424,16 @@ public void setQwpMaxVersion(int maxVersion) {
         this.qwpMaxVersion = maxVersion;
     }
 
+    /**
+     * Enables the opt-in X-QWP-Request-Durable-Ack upgrade header. When set,
+     * servers with primary replication configured will additionally emit
+     * STATUS_DURABLE_ACK frames as the WAL containing committed client
+     * messages reaches the object store.
+     */
+    public void setQwpRequestDurableAck(boolean enabled) {
+        this.qwpRequestDurableAck = enabled;
+    }
+
     /**
      * Non-blocking attempt to receive a WebSocket frame.
      * Returns immediately if no complete frame is available.
@@ -518,6 +530,9 @@ public void upgrade(CharSequence path, int timeout, CharSequence authorizationHe
             sendBuffer.putAscii(Integer.toString(qwpMaxBatchRows));
             sendBuffer.putAscii("\r\n");
         }
+        if (qwpRequestDurableAck) {
+            sendBuffer.putAscii("X-QWP-Request-Durable-Ack: true\r\n");
+        }
         if (authorizationHeader != null) {
             sendBuffer.putAscii("Authorization: ");
             sendBuffer.putAscii(authorizationHeader);

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

@@ -377,13 +377,28 @@ public Throwable getLastError() {
         return lastError.get();
     }
 
+    /**
+     * Returns the highest batch sequence acknowledged by the server, or -1 if
+     * no acknowledgment has been received yet.
+     */
+    public long getHighestAckedSequence() {
+        return highestAcked;
+    }
+
     /**
      * Returns the maximum window size.
      */
     public int getMaxWindowSize() {
         return maxWindowSize;
     }
 
+    /**
+     * Returns the timeout (ms) applied to blocking window operations.
+     */
+    public long getTimeoutMs() {
+        return timeoutMs;
+    }
+
     /**
      * Returns the total number of batches acknowledged.
      */