feat(jdbc): accept transactionMode in the JDBC URL and connection properties

README.md

@@ -78,6 +78,26 @@ public class Example {
 }
 ```
 
+### Connection URL parameters
+
+The JDBC URL accepts these query parameters. The same keys work as `Properties` passed to
+`DriverManager.getConnection`; a query parameter wins over the matching property.
+
+| Parameter | Values | Default | Notes |
+| --- | --- | --- | --- |
+| `authToken` | token | none | Sent as a bearer token. `jwt` and `password` are accepted aliases. |
+| `busyTimeout` | milliseconds | `5000` | How long a request retries while the server reports busy. |
+| `connectTimeout` | milliseconds | `10000` | HTTP connect and request timeout. |
+| `encoding` | `json`, `protobuf` | `json` | `protobuf` requires a `ws://` or `wss://` URL. |
+| `transactionMode` | `deferred`, `immediate`, `exclusive` | `deferred` | Mode used by the next transaction begin. Case-insensitive; an unknown value fails at connect. |
+| `tls`, `ssl` | `true` | unset | Upgrades `http` to `https` and `ws` to `wss`. |
+
+`Connection.setReadOnly` is the JDBC hint and does not reject writes. A connection is read-only
+only when the server enforces it. With sqld, start the server with `--auth-jwt-key-file` and pass a
+JWT whose `a` claim is `ro` as `authToken`; sqld then refuses writes on that connection. sqld
+v0.24.33 refuses `PRAGMA query_only`, so there is no session-level setting that achieves the same
+thing.
+
 ## Development & Building
 
 ### Prerequisites

libsql-client/src/main/java/io/github/magnusp/libsql/client/LibsqlHttpClient.java

@@ -100,7 +100,13 @@ public StmtResult executeOnStream(Stmt stmt) {
         StreamResult result;
         synchronized (streamLock) {
             PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.Execute(stmt)));
-            PipelineRespBody resp = sendPipeline(req, baseUrl);
+            PipelineRespBody resp;
+            try {
+                resp = sendPipeline(req, baseUrl);
+            } catch (LibsqlException e) {
+                clearStreamOnAuthFailure(e);
+                throw e;
+            }
             updateStreamState(resp);
             result = firstResult(resp);
         }
@@ -135,7 +141,13 @@ public BatchResult executeBatchOnStream(Batch batch) {
         StreamResult result;
         synchronized (streamLock) {
             PipelineReqBody req = new PipelineReqBody(baton, List.of(new StreamRequest.BatchReq(batch)));
-            PipelineRespBody resp = sendPipeline(req, baseUrl);
+            PipelineRespBody resp;
+            try {
+                resp = sendPipeline(req, baseUrl);
+            } catch (LibsqlException e) {
+                clearStreamOnAuthFailure(e);
+                throw e;
+            }
             updateStreamState(resp);
             result = firstResult(resp);
         }
@@ -196,6 +208,17 @@ private void updateStreamState(PipelineRespBody resp) {
         }
     }
 
+    /**
+     * The server invalidates the baton when it refuses a request, so the next statement must open a
+     * fresh stream instead of replaying a dead one. Callers hold {@link #streamLock}.
+     */
+    private void clearStreamOnAuthFailure(LibsqlException e) {
+        if (e.getHttpStatus() == 401 || e.getHttpStatus() == 403) {
+            baton = null;
+            baseUrl = null;
+        }
+    }
+
     public PipelineRespBody sendPipeline(PipelineReqBody reqBody, URI targetUri) {
         return sendPipelineWithRetry(reqBody, targetUri != null ? targetUri : pipelineUri);
     }
@@ -239,7 +262,7 @@ private PipelineRespBody sendPipelineWithRetry(PipelineReqBody reqBody, URI uri)
                     } catch (Exception ignored) {
                     }
                     if (err != null) {
-                        throw new LibsqlException(err);
+                        throw new LibsqlException(err.message(), err.code(), status);
                     }
                     throw new LibsqlException("HTTP error " + status + ": " + new String(response.body()), null, status);
                 }

libsql-client/src/test/java/io/github/magnusp/libsql/client/LibsqlHttpClientAuthTest.java

@@ -0,0 +1,102 @@
+package io.github.magnusp.libsql.client;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpHandler;
+import com.sun.net.httpserver.HttpServer;
+import io.github.magnusp.libsql.hrana.codec.Stmt;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class LibsqlHttpClientAuthTest {
+
+    private static final byte[] FORBIDDEN =
+            "{\"error\":\"Authorization forbidden: nope\"}".getBytes(StandardCharsets.UTF_8);
+
+    private static final byte[] STREAM_STARTED =
+            ("{\"baton\":\"baton-1\",\"results\":[{\"type\":\"ok\",\"response\":"
+                            + "{\"type\":\"execute\",\"result\":{\"cols\":[],\"rows\":[],"
+                            + "\"affected_row_count\":0,\"last_insert_rowid\":null}}}]}")
+                    .getBytes(StandardCharsets.UTF_8);
+
+    @Test
+    void forbiddenResponseSurfacesTheServerMessageAndStatus() throws Exception {
+        try (Server server = new Server(exchange -> respond(exchange, 403, FORBIDDEN))) {
+            LibsqlHttpClient client = server.client();
+
+            assertThatThrownBy(() -> client.executeOneShot(new Stmt("SELECT 1")))
+                    .isInstanceOfSatisfying(LibsqlException.class, e -> {
+                        assertThat(e.getMessage()).isEqualTo("Authorization forbidden: nope");
+                        assertThat(e.getHttpStatus()).isEqualTo(403);
+                    });
+        }
+    }
+
+    @Test
+    void forbiddenResponseOnAStreamDropsTheBaton() throws Exception {
+        AtomicInteger request = new AtomicInteger();
+        try (Server server = new Server(exchange -> {
+            if (request.getAndIncrement() == 1) {
+                respond(exchange, 403, FORBIDDEN);
+            } else {
+                respond(exchange, 200, STREAM_STARTED);
+            }
+        })) {
+            LibsqlHttpClient client = server.client();
+            client.executeOnStream(new Stmt("SELECT 1"));
+            assertThat(client.hasOpenStream()).isTrue();
+
+            assertThatThrownBy(() -> client.executeOnStream(new Stmt("INSERT INTO t VALUES (1)")))
+                    .isInstanceOfSatisfying(
+                            LibsqlException.class,
+                            e -> assertThat(e.getHttpStatus()).isEqualTo(403));
+            assertThat(client.hasOpenStream()).isFalse();
+
+            client.executeOnStream(new Stmt("SELECT 2"));
+            assertThat(server.bodies.get(2)).doesNotContain("baton-1");
+        }
+    }
+
+    private static void respond(HttpExchange exchange, int status, byte[] body) throws IOException {
+        exchange.sendResponseHeaders(status, body.length);
+        try (OutputStream out = exchange.getResponseBody()) {
+            out.write(body);
+        }
+    }
+
+    private static final class Server implements AutoCloseable {
+
+        private final HttpServer server;
+        private final List<String> bodies = new CopyOnWriteArrayList<>();
+
+        Server(HttpHandler handler) throws IOException {
+            server = HttpServer.create(new InetSocketAddress(InetAddress.getLoopbackAddress(), 0), 0);
+            server.createContext("/v3/pipeline", exchange -> {
+                bodies.add(new String(exchange.getRequestBody().readAllBytes(), StandardCharsets.UTF_8));
+                handler.handle(exchange);
+            });
+            server.start();
+        }
+
+        LibsqlHttpClient client() {
+            LibsqlClientConfig config = new LibsqlClientConfig(
+                    "http://127.0.0.1:" + server.getAddress().getPort(), null);
+            return new LibsqlHttpClient(config);
+        }
+
+        @Override
+        public void close() {
+            server.stop(0);
+        }
+    }
+}

libsql-hrana-codec/src/main/java/io/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodec.java

@@ -570,17 +570,26 @@ private DescribeCol parseDescribeCol(FastJsonParser p) throws IOException {
 
     private HranaError parseError(FastJsonParser p) throws IOException {
         p.expectObjectStart();
-        String message = "";
+        String message = null;
         String code = null;
         while (!p.checkObjectEnd()) {
             String field = p.nextFieldName();
             switch (field) {
                 case "message" -> message = p.nextString();
+                // A transport-level refusal carries only an "error" string, not a Hrana Error.
+                case "error" -> {
+                    String text = p.peekString() ? p.nextString() : null;
+                    if (text == null) {
+                        p.skipValue();
+                    } else if (message == null) {
+                        message = text;
+                    }
+                }
                 case "code" -> code = p.nextStringOrNull();
                 default -> p.skipValue();
             }
         }
-        return new HranaError(message, code);
+        return new HranaError(message != null ? message : "", code);
     }
 
     // -------------------------------------------------------------

libsql-hrana-codec/src/main/java/io/github/magnusp/libsql/hrana/codec/JacksonHranaJsonCodec.java

@@ -521,17 +521,27 @@ private HranaError readError(JsonParser p) throws IOException {
         if (p.currentToken() != JsonToken.START_OBJECT) {
             throw new IOException("Expected START_OBJECT for Error");
         }
-        String message = "";
+        String message = null;
         String code = null;
         while (p.nextToken() != JsonToken.END_OBJECT) {
             String field = p.currentName();
             p.nextToken();
             switch (field) {
                 case "message" -> message = p.getText();
+                // A transport-level refusal carries only an "error" string, not a Hrana Error.
+                case "error" -> {
+                    if (p.currentToken() == JsonToken.VALUE_STRING) {
+                        if (message == null) {
+                            message = p.getText();
+                        }
+                    } else {
+                        p.skipChildren();
+                    }
+                }
                 case "code" -> code = p.getText();
                 default -> p.skipChildren();
             }
         }
-        return new HranaError(message, code);
+        return new HranaError(message != null ? message : "", code);
     }
 }

libsql-hrana-codec/src/test/java/io/github/magnusp/libsql/hrana/codec/DirectHranaJsonCodecTest.java

@@ -2,7 +2,9 @@
 
 import org.junit.jupiter.api.Test;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
 
@@ -48,4 +50,24 @@ void testSerializeSupplementaryCharactersAsUtf8() throws IOException {
         assertThat(json).doesNotContain("?");
     }
 
+    @Test
+    void testDeserializeErrorFallsBackToTheErrorField() throws IOException {
+        HranaError error = codec.deserializeError(errorBody("{\"error\":\"Authorization forbidden: nope\"}"));
+
+        assertThat(error.message()).isEqualTo("Authorization forbidden: nope");
+        assertThat(error.code()).isNull();
+    }
+
+    @Test
+    void testDeserializeErrorPrefersTheMessageField() throws IOException {
+        HranaError error = codec.deserializeError(
+                errorBody("{\"message\":\"constraint failed\",\"error\":\"nope\",\"code\":\"SQLITE_CONSTRAINT\"}"));
+
+        assertThat(error.message()).isEqualTo("constraint failed");
+        assertThat(error.code()).isEqualTo("SQLITE_CONSTRAINT");
+    }
+
+    private static InputStream errorBody(String json) {
+        return new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8));
+    }
 }