Improve recovery and usage of threads

Schedule reconnection attempts instead of waiting and blocking the
thread.
Use platform threads for the environment scheduler.
Use a common event loop to record topology.

Virtual threads do not behave as expected, especially for long-running
tasks (e.g. polling a blocking queue for new tasks to process), which
makes recovery fails. They will be used only for short-lived task, like
polling a future to get the settlement.

Author: acogoluegnes

ci/start-cluster.sh

@@ -17,8 +17,8 @@ wait_for_message rabbitmq0 "completed with"
 
 docker exec rabbitmq0 rabbitmqctl await_online_nodes 3
 
-docker exec rabbitmq0 rabbitmqctl enable_feature_flag khepri_db
-docker exec rabbitmq1 rabbitmqctl enable_feature_flag khepri_db
-docker exec rabbitmq2 rabbitmqctl enable_feature_flag khepri_db
+docker exec rabbitmq0 rabbitmqctl enable_feature_flag --experimental khepri_db
+docker exec rabbitmq1 rabbitmqctl enable_feature_flag --experimental khepri_db
+docker exec rabbitmq2 rabbitmqctl enable_feature_flag --experimental khepri_db
 
 docker exec rabbitmq0 rabbitmqctl cluster_status

pom.xml

@@ -346,7 +346,6 @@
                     <argLine>${test-arguments}</argLine>
                     <systemPropertyVariables>
                         <net.bytebuddy.experimental>true</net.bytebuddy.experimental>
-                        <rabbitmqctl.bin>DOCKER:rabbitmq</rabbitmqctl.bin>
                     </systemPropertyVariables>
                 </configuration>
             </plugin>

src/main/java/com/rabbitmq/client/amqp/BackOffDelayPolicy.java

@@ -25,21 +25,85 @@ public interface BackOffDelayPolicy {
 
   Duration delay(int recoveryAttempt);
 
+  static BackOffDelayPolicy fixedWithInitialDelay(Duration initialDelay, Duration delay) {
+    return new FixedWithInitialDelayBackOffPolicy(initialDelay, delay);
+  }
+
+  static BackOffDelayPolicy fixedWithInitialDelay(
+      Duration initialDelay, Duration delay, Duration timeout) {
+    return new FixedWithInitialDelayAndTimeoutBackOffPolicy(initialDelay, delay, timeout);
+  }
+
   static BackOffDelayPolicy fixed(Duration delay) {
-    return new FixedBackOffDelayPolicy(delay);
+    return new FixedWithInitialDelayBackOffPolicy(delay, delay);
   }
 
-  class FixedBackOffDelayPolicy implements BackOffDelayPolicy {
+  final class FixedWithInitialDelayBackOffPolicy implements BackOffDelayPolicy {
 
+    private final Duration initialDelay;
     private final Duration delay;
 
-    private FixedBackOffDelayPolicy(Duration delay) {
+    private FixedWithInitialDelayBackOffPolicy(Duration initialDelay, Duration delay) {
+      this.initialDelay = initialDelay;
       this.delay = delay;
     }
 
     @Override
     public Duration delay(int recoveryAttempt) {
-      return this.delay;
+      return recoveryAttempt == 0 ? initialDelay : delay;
+    }
+
+    @Override
+    public String toString() {
+      return "FixedWithInitialDelayBackOffPolicy{"
+          + "initialDelay="
+          + initialDelay
+          + ", delay="
+          + delay
+          + '}';
+    }
+  }
+
+  final class FixedWithInitialDelayAndTimeoutBackOffPolicy implements BackOffDelayPolicy {
+
+    private final int attemptLimitBeforeTimeout;
+    private final BackOffDelayPolicy delegate;
+
+    private FixedWithInitialDelayAndTimeoutBackOffPolicy(
+        Duration initialDelay, Duration delay, Duration timeout) {
+      this(fixedWithInitialDelay(initialDelay, delay), timeout);
+    }
+
+    private FixedWithInitialDelayAndTimeoutBackOffPolicy(
+        BackOffDelayPolicy policy, Duration timeout) {
+      if (timeout.toMillis() < policy.delay(0).toMillis()) {
+        throw new IllegalArgumentException("Timeout must be longer than initial delay");
+      }
+      this.delegate = policy;
+      // best effort, assume FixedWithInitialDelay-ish policy
+      Duration initialDelay = policy.delay(0);
+      Duration delay = policy.delay(1);
+      long timeoutWithInitialDelay = timeout.toMillis() - initialDelay.toMillis();
+      this.attemptLimitBeforeTimeout = (int) (timeoutWithInitialDelay / delay.toMillis()) + 1;
+    }
+
+    @Override
+    public Duration delay(int recoveryAttempt) {
+      if (recoveryAttempt >= attemptLimitBeforeTimeout) {
+        return TIMEOUT;
+      } else {
+        return delegate.delay(recoveryAttempt);
+      }
+    }
+
+    @Override
+    public String toString() {
+      return "FixedWithInitialDelayAndTimeoutBackOffPolicy{"
+          + "attemptLimitBeforeTimeout="
+          + attemptLimitBeforeTimeout
+          + ", delegate="
+          + delegate
+          + '}';
     }
   }
 }