fix: Find out again which API the server offers once it answers a path of the v2 API with 404 or 405, for example after a rollback.

README.md

@@ -64,7 +64,7 @@ client.verifyApiToken();
 
 ### Talking to Servers With and Without the v2 API
 
-The client works with servers that offer the v2 API, and with servers that do not, such as EventSourcingDB 1.2.0. Creating a client sends no request. The first request of a client asks the server whether it offers the v2 API, and the client remembers the answer: from then on, it uses the v2 API if the server offers it, and the v1 API otherwise. If asking fails, e.g. because the server can not be reached, the first request fails, and the next one asks again.
+The client works with servers that offer the v2 API, and with servers that do not, such as EventSourcingDB 1.2.0. Creating a client sends no request. The first request of a client asks the server whether it offers the v2 API, and the client remembers the answer: from then on, it uses the v2 API if the server offers it, and the v1 API otherwise. If asking fails, e.g. because the server can not be reached, the first request fails, and the next one asks again. If the server answers a path of the v2 API with 404 or 405, e.g. after it was rolled back to a version without it, the client asks again on the next request.
 
 The v2 API reads with the HTTP method `QUERY` ([RFC 10008](https://www.rfc-editor.org/rfc/rfc10008)), which carries a body, as `POST` does, but changes nothing. If a proxy, a load balancer, or a firewall sits in front of the database, it has to let `QUERY` through. The client does not fall back to `POST`.
 

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Client.java

@@ -5,6 +5,7 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.Stream;
 import org.jspecify.annotations.Nullable;
@@ -18,8 +19,9 @@
  * <p>The client talks to every supported version of EventSourcingDB, whether it offers the v2 API or not. Creating a
  * client sends no request. The first request of a client asks the server whether it offers the v2 API, and the client
  * uses the v2 API from then on if it does, and the v1 API otherwise. If asking fails, e.g. because the server can not
- * be reached, that first request fails, and the next request asks again. The v2 API reads with the HTTP method
- * {@code QUERY}, so a proxy in front of the database has to let it through.
+ * be reached, that first request fails, and the next request asks again. If the server answers a path of the v2 API
+ * with 404 or 405, e.g. after it was rolled back to a version without it, the next request asks again, too. The v2 API
+ * reads with the HTTP method {@code QUERY}, so a proxy in front of the database has to let it through.
  *
  * <p>Every method that sends a request throws a {@link DbApiException} if the server reports an error, e.g. a
  * precondition that does not hold, and an {@link InvalidServerHeaderException} if the server is not an
@@ -104,10 +106,27 @@ Transport chosenTransport() {
     }
 
     // Learns which API the server offers, which only the first request of a
-    // client asks the server for.
-    private Transport transport(String action) {
+    // client asks the server for, and sends the request through it. How the
+    // request fails may tell that the server does not offer that API any
+    // longer.
+    private <T extends @Nullable Object> T send(String action, Function<Transport, T> request) {
         Http.throwIfInterrupted(action);
-        return TransportChooser.await(action, transports.transport());
+        var choice = TransportChooser.await(action, transports.transport());
+
+        try {
+            return request.apply(choice.transport());
+        } catch (RuntimeException ex) {
+            transports.forgetIfTheV2ApiIsMissing(choice, ex);
+            throw ex;
+        }
+    }
+
+    // Sends a request that returns nothing, as send does.
+    private void run(String action, Consumer<Transport> request) {
+        send(action, transport -> {
+            request.accept(transport);
+            return null;
+        });
     }
 
     private <T> Stream<T> stream(String action, Function<Transport, StreamSpec<T>> specs) {
@@ -121,7 +140,7 @@ private <T> Stream<T> stream(String action, Function<Transport, StreamSpec<T>> s
      * @throws EventSourcingDbException if the instance is not reachable, or does not confirm the ping
      */
     public void ping() {
-        transport("ping").ping();
+        run("ping", Transport::ping);
     }
 
     /**
@@ -131,7 +150,7 @@ public void ping() {
      * @throws EventSourcingDbException if the instance does not confirm the API token
      */
     public void verifyApiToken() {
-        transport("verify API token").verifyApiToken();
+        run("verify API token", Transport::verifyApiToken);
     }
 
     /**
@@ -185,7 +204,7 @@ public List<Event> writeEvents(List<EventCandidate> events, List<Precondition> p
             }
         }
 
-        return transport("write events").writeEvents(body);
+        return send("write events", transport -> transport.writeEvents(body));
     }
 
     private static Map<String, Object> toRequest(Precondition precondition) {
@@ -362,7 +381,7 @@ public Stream<EventType> readEventTypes() {
      * @throws DbApiException with the code {@link ErrorCode#EVENT_TYPE_UNKNOWN} if the event type does not exist
      */
     public EventType readEventType(String eventType) {
-        return transport("read event type").readEventType(eventType);
+        return send("read event type", transport -> transport.readEventType(eventType));
     }
 
     /**
@@ -378,7 +397,7 @@ public EventType readEventType(String eventType) {
      */
     public void registerEventSchema(String eventType, Map<String, ?> schema) {
         var schemaNode = Json.MAPPER.valueToTree(schema);
-        transport("register event schema").registerEventSchema(eventType, schemaNode);
+        run("register event schema", transport -> transport.registerEventSchema(eventType, schemaNode));
     }
 
     /**

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/Http.java

@@ -73,7 +73,7 @@ private static HttpResponse<InputStream> check(
             String action, HttpResponse<InputStream> response, Duration stallTimeout, CodeMapping codes) {
         if (!isFromEventSourcingDb(response)) {
             close(response.body());
-            throw new InvalidServerHeaderException();
+            throw new InvalidServerHeaderException(response.statusCode());
         }
 
         if (response.statusCode() != 200) {

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/InvalidServerHeaderException.java

@@ -8,7 +8,17 @@
 public final class InvalidServerHeaderException extends EventSourcingDbException {
     private static final long serialVersionUID = 1L;
 
-    InvalidServerHeaderException() {
+    /** The HTTP status code the server answered with. */
+    private final int status;
+
+    InvalidServerHeaderException(int status) {
         super("server must be EventSourcingDB");
+        this.status = status;
+    }
+
+    // Lets the client tell from the status code whether the server offers the
+    // v2 API, which a 404 or a 405 tells whatever server sent it.
+    int status() {
+        return status;
     }
 }

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/LineStream.java

@@ -121,7 +121,8 @@ private void end(BufferedReader reader) {
     }
 
     // Learns which API the server offers, and sends the request of the
-    // stream through it.
+    // stream through it. How the server answers the request may tell that it
+    // does not offer that API any longer.
     private Opened<T> open() {
         throwIfClosed();
         Http.throwIfInterrupted(action);
@@ -133,12 +134,18 @@ private Opened<T> open() {
         InputStream body;
         try {
             throwIfClosed();
-            spec = specs.apply(TransportChooser.await(action, transport));
+            var choice = TransportChooser.await(action, transport);
+            spec = specs.apply(choice.transport());
 
             var response = requests.sendForStream(spec.request());
             pending = response;
-            body = Http.await(action, response, requests.stallTimeout(), spec.codes())
-                    .body();
+            try {
+                body = Http.await(action, response, requests.stallTimeout(), spec.codes())
+                        .body();
+            } catch (RuntimeException ex) {
+                transports.forgetIfTheV2ApiIsMissing(choice, ex);
+                throw ex;
+            }
         } catch (RuntimeException ex) {
             throwIfClosed();
             throw ex;

eventsourcingdb/src/main/java/io/thenativeweb/eventsourcingdb/TransportChooser.java

@@ -26,6 +26,10 @@
 // the request that asked fails, and the next request asks again. Requests
 // that are made while the server is being asked wait for its answer rather
 // than asking again, and fail along with it, each with its own action.
+//
+// A server that answers a request through the v2 API with 404 or 405 does not
+// offer it any longer, e.g. because it was rolled back to a version without
+// it, so the client forgets the answer, and the next request asks again.
 final class TransportChooser {
     private static final String PROBE_PATH = "/api/v2/system/ping";
 
@@ -34,19 +38,28 @@ final class TransportChooser {
     private final V2Transport v2;
 
     private final Object lock = new Object();
-    private @Nullable Transport chosen;
-    private @Nullable CompletableFuture<Transport> probe;
+    private @Nullable Choice chosen;
+    private @Nullable CompletableFuture<Choice> probe;
+    private long generation;
+
+    // A choice is the transport that a probe found out, along with the
+    // generation of that answer, which counts the answers that probes stored.
+    // The generation tells an answer apart from a later one that names the
+    // same transport, so that a request can tell which answer it was sent
+    // under.
+    record Choice(Transport transport, long generation) {}
 
     TransportChooser(Requests requests, V1Transport v1, V2Transport v2) {
         this.requests = requests;
         this.v1 = v1;
         this.v2 = v2;
     }
 
-    // Returns the transport, once the server has told which API it offers.
-    // The future is the caller's own, so that canceling it ends only the wait
-    // of the caller, not the probe, which other requests may wait for, too.
-    CompletableFuture<Transport> transport() {
+    // Returns the choice of the transport, once the server has told which API
+    // it offers. The future is the caller's own, so that canceling it ends
+    // only the wait of the caller, not the probe, which other requests may
+    // wait for, too.
+    CompletableFuture<Choice> transport() {
         synchronized (lock) {
             var chosen = this.chosen;
             if (chosen != null) {
@@ -65,14 +78,63 @@ CompletableFuture<Transport> transport() {
     @Nullable
     Transport chosen() {
         synchronized (lock) {
-            return chosen;
+            var chosen = this.chosen;
+            return chosen != null ? chosen.transport() : null;
+        }
+    }
+
+    // Forgets the API once the server answered a request through the v2 API
+    // with 404 or 405, while the request fails as it does anyway. As for the
+    // probe, that holds whatever server sent the answer, so also for one that
+    // lacks the Server header: forgetting only makes the next request ask
+    // again, and the probe decides. A write fails that way with an
+    // OutcomeUnknownException, whose cause is the failure. Through the v1
+    // API, a 404 may be an error of its own, e.g. an event type that does not
+    // exist, and an error that the v2 API reports in its answer has the status
+    // code 200, so neither tells anything about the API.
+    //
+    // Only the answer that the request was sent under is forgotten, which
+    // its generation tells, so that a request answered late leaves alone the
+    // probe that another one started meanwhile, which other requests may wait
+    // for, and the answer that such a probe stored, even if it is the v2 API
+    // again.
+    void forgetIfTheV2ApiIsMissing(Choice choice, RuntimeException failure) {
+        var answer = failure instanceof OutcomeUnknownException ? failure.getCause() : failure;
+        if (choice.transport() != v2 || !isAnsweredWithoutTheV2Api(answer)) {
+            return;
         }
+
+        synchronized (lock) {
+            var chosen = this.chosen;
+            if (chosen != null && chosen.generation() == choice.generation()) {
+                this.chosen = null;
+            }
+        }
+    }
+
+    // Tells whether a request failed because the server answered it with a
+    // status code that says that it does not offer the v2 API.
+    private static boolean isAnsweredWithoutTheV2Api(@Nullable Throwable answer) {
+        if (answer instanceof DbApiException exception) {
+            return isV2ApiMissing(exception.status());
+        }
+        if (answer instanceof InvalidServerHeaderException exception) {
+            return isV2ApiMissing(exception.status());
+        }
+
+        return false;
+    }
+
+    // A server answers the paths of the v2 API with 404 or 405 if it does not
+    // offer the v2 API, as the comment on the class tells.
+    private static boolean isV2ApiMissing(int status) {
+        return status == 404 || status == 405;
     }
 
     // Waits for the transport. A failed probe fails the request with an
     // exception of its own, so that each request that waited for the probe
     // names its own action.
-    static Transport await(String action, CompletableFuture<Transport> transport) {
+    static Choice await(String action, CompletableFuture<Choice> transport) {
         try {
             return transport.get();
         } catch (InterruptedException ex) {
@@ -88,7 +150,7 @@ static Transport await(String action, CompletableFuture<Transport> transport) {
 
     // The answer to the probe is small, so the stall timeout bounds the probe
     // as a whole, which spares it a reader of its own.
-    private CompletableFuture<Transport> startProbe() {
+    private CompletableFuture<Choice> startProbe() {
         var response = requests.sendAsync(requests.get(PROBE_PATH), HttpResponse.BodyHandlers.ofString());
 
         var stallTimeout = requests.stallTimeout();
@@ -98,7 +160,7 @@ private CompletableFuture<Transport> startProbe() {
             response.cancel(true);
         });
 
-        var result = new CompletableFuture<Transport>();
+        var result = new CompletableFuture<Choice>();
         probe = result;
         response.whenComplete((answer, failure) -> {
             stallTimer.cancel(false);
@@ -108,12 +170,13 @@ private CompletableFuture<Transport> startProbe() {
         return result;
     }
 
-    // Keeps the transport that the probe found out, or forgets the probe if
-    // it failed, so that the next request probes again. Either way, a probe
-    // is running no longer. A probe starts only while no other is running, so
-    // the probe that completes is the one that is kept.
+    // Keeps the transport that the probe found out, as an answer of a
+    // generation of its own, or forgets the probe if it failed, so that the
+    // next request probes again. Either way, a probe is running no longer. A
+    // probe starts only while no other is running, so the probe that
+    // completes is the one that is kept.
     private void complete(
-            CompletableFuture<Transport> result,
+            CompletableFuture<Choice> result,
             @Nullable HttpResponse<String> answer,
             @Nullable Throwable failure,
             boolean hasStalled) {
@@ -128,11 +191,14 @@ private void complete(
             return;
         }
 
+        Choice choice;
         synchronized (lock) {
-            chosen = transport;
+            generation++;
+            choice = new Choice(transport, generation);
+            chosen = choice;
             probe = null;
         }
-        result.complete(transport);
+        result.complete(choice);
     }
 
     private Transport interpret(
@@ -145,11 +211,11 @@ private Transport interpret(
         // proxy that passes on the v1 API only answers the probe itself. The
         // requests to the v1 API check that the server is an EventSourcingDB.
         var status = answer.statusCode();
-        if (status == 404 || status == 405) {
+        if (isV2ApiMissing(status)) {
             return v1;
         }
         if (!Http.isFromEventSourcingDb(answer)) {
-            throw new ProbeFailure(action -> new InvalidServerHeaderException());
+            throw new ProbeFailure(action -> new InvalidServerHeaderException(status));
         }
         if (status != 200) {
             var body = answer.body().getBytes(UTF_8);