Prefer Connect libp2p watchless mode with WatchService fallback

core/src/main/java/com/minekube/connect/register/WatcherRegister.java

@@ -33,6 +33,7 @@
 import com.minekube.connect.api.logger.ConnectLogger;
 import com.minekube.connect.network.netty.LocalSession;
 import com.minekube.connect.tunnel.Tunneler;
+import com.minekube.connect.tunnel.p2p.Libp2pEndpoint;
 import com.minekube.connect.util.Utils;
 import com.minekube.connect.util.backoff.BackOff;
 import com.minekube.connect.util.backoff.ExponentialBackOff;
@@ -41,7 +42,6 @@
 import com.minekube.connect.watch.WatchBootstrap;
 import com.minekube.connect.watch.WatchClient;
 import com.minekube.connect.watch.Watcher;
-import com.minekube.connect.tunnel.p2p.Libp2pEndpoint;
 import io.netty.util.concurrent.DefaultThreadFactory;
 import java.io.IOException;
 import java.time.Duration;
@@ -66,6 +66,8 @@ public class WatcherRegister {
     // volatile: written from injection thread (start/stop) and read from the
     // scheduler thread (retry) and OkHttp dispatcher (WatcherImpl callbacks).
     private volatile WebSocket ws;
+    private volatile WatcherImpl watcher;
+    private volatile boolean legacyWatchEnabled;
     private ExponentialBackOff backOffPolicy;
     private final AtomicBoolean started = new AtomicBoolean();
 
@@ -78,6 +80,7 @@ public class WatcherRegister {
     @Inject
     public void start() {
         if (started.compareAndSet(false, true)) {
+            legacyWatchEnabled = true;
             scheduler = Executors.newSingleThreadScheduledExecutor(
                     new DefaultThreadFactory("connect-watcher-scheduler", true));
             backOffPolicy = new ExponentialBackOff.Builder()
@@ -101,6 +104,7 @@ public void stop() {
             return;
         }
         logger.info("Stopped watching for sessions");
+        legacyWatchEnabled = false;
         if (retryFuture != null) {
             retryFuture.cancel(false);
             retryFuture = null;
@@ -110,13 +114,17 @@ public void stop() {
             scheduler = null;
         }
         if (ws != null) {
+            if (watcher != null) {
+                watcher.ignoreTerminalEvents();
+            }
             ws.close(1000, "watcher stopped");
             ws = null;
         }
+        watcher = null;
     }
 
     private void retry() {
-        if (!started.get()) {
+        if (!started.get() || !legacyWatchEnabled) {
             return;
         }
         if (retryFuture != null) {
@@ -141,26 +149,81 @@ private void retry() {
         logger.info("Trying to reconnect in {}...",
                 Utils.humanReadableFormat(Duration.ofMillis(millis)));
         retryFuture = s.schedule(() -> {
-            if (started.get()) {
+            if (started.get() && legacyWatchEnabled) {
                 watch();
             }
         }, millis, TimeUnit.MILLISECONDS);
     }
 
     private void watch() {
+        if (!started.get() || !legacyWatchEnabled) {
+            return;
+        }
         if (ws != null) {
+            if (watcher != null) {
+                watcher.ignoreTerminalEvents();
+            }
             ws.close(1000, "watcher is reconnecting");
         }
-        ws = watchClient.watch(new WatcherImpl());
+        watcher = new WatcherImpl();
+        ws = watchClient.watch(watcher);
+    }
+
+    private void enterWatchlessMode() {
+        if (!started.get()) {
+            return;
+        }
+        legacyWatchEnabled = false;
+        if (retryFuture != null) {
+            retryFuture.cancel(false);
+            retryFuture = null;
+        }
+        WatcherImpl currentWatcher = watcher;
+        if (currentWatcher != null) {
+            currentWatcher.ignoreTerminalEvents();
+        }
+        WebSocket currentWebSocket = ws;
+        ws = null;
+        watcher = null;
+        logger.info("Connect libp2p watchless endpoint mode enabled");
+        if (currentWebSocket != null) {
+            currentWebSocket.close(1000, "watchless endpoint mode enabled");
+        }
+    }
+
+    private void enterLegacyWatchFallback() {
+        if (!started.get()) {
+            return;
+        }
+        if (legacyWatchEnabled && ws != null) {
+            return;
+        }
+        legacyWatchEnabled = true;
+        resetBackOff();
+        logger.info("Connect libp2p watchless endpoint fallback enabled WatchService");
+        watch();
     }
 
     private class WatcherImpl implements Watcher {
+        private volatile boolean ignoreTerminalEvents;
 
         @Override
         public void onOpen(WatchBootstrap bootstrap) {
             logger.translatedInfo("connect.watch.started");
             if (bootstrap.hasLibp2p()) {
-                libp2pEndpoint.start(bootstrap.libp2pEdgeAddrs(), bootstrap.libp2pRelayAddrs());
+                if (bootstrap.supportsWatchlessLibp2p()) {
+                    libp2pEndpoint.start(
+                            bootstrap.libp2pEdgeAddrs(),
+                            bootstrap.libp2pRelayAddrs(),
+                            true,
+                            WatcherRegister.this::enterWatchlessMode,
+                            WatcherRegister.this::enterLegacyWatchFallback);
+                } else {
+                    libp2pEndpoint.start(
+                            bootstrap.libp2pEdgeAddrs(),
+                            bootstrap.libp2pRelayAddrs(),
+                            false);
+                }
             }
             startResetBackOffTimer();
         }
@@ -204,6 +267,9 @@ public void onProposal(SessionProposal proposal) {
 
         @Override
         public void onError(Throwable t) {
+            if (shouldIgnoreTerminalEvent()) {
+                return;
+            }
             logger.error("Connection error with WatchService: " +
                             t + (
                             t.getCause() == null ? ""
@@ -216,6 +282,9 @@ public void onError(Throwable t) {
 
         @Override
         public void onCompleted() {
+            if (shouldIgnoreTerminalEvent()) {
+                return;
+            }
             cancelResetBackOffTimer();
             retry();
         }
@@ -226,7 +295,7 @@ void startResetBackOffTimer() {
             cancelResetBackOffTimer();
             // Snapshot: a late onOpen after stop() can land here with scheduler == null.
             ScheduledExecutorService s = scheduler;
-            if (s == null || !started.get()) {
+            if (s == null || !started.get() || !legacyWatchEnabled || ignoreTerminalEvents) {
                 return;
             }
             resetBackOffFuture = s.schedule(() -> {
@@ -242,5 +311,14 @@ void cancelResetBackOffTimer() {
                 resetBackOffFuture = null;
             }
         }
+
+        void ignoreTerminalEvents() {
+            ignoreTerminalEvents = true;
+            cancelResetBackOffTimer();
+        }
+
+        private boolean shouldIgnoreTerminalEvent() {
+            return ignoreTerminalEvents || !legacyWatchEnabled || !started.get();
+        }
     }
 }

core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java

@@ -15,9 +15,7 @@
  * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- * THE SOFTWARE.
+ * DEALINGS IN THE SOFTWARE.
  *
  * @author Minekube
  * @link https://github.com/minekube/connect-java
@@ -38,13 +36,14 @@
 import java.nio.file.Path;
 import java.util.List;
 
-public final class Libp2pEndpoint {
+public class Libp2pEndpoint {
     static final String REGISTER_PROTOCOL_ID = "/minekube/connect/register/1.0.0";
 
     private final ConnectLogger logger;
     private Object runtime;
     private Method startMethod;
     private Method startBootstrapMethod;
+    private Method startWatchlessMethod;
     private Method stopMethod;
 
     @Inject
@@ -80,10 +79,14 @@ public Libp2pEndpoint(
                     platformInjector,
                     api);
             this.startMethod = runtimeClass.getDeclaredMethod("start");
-            this.startBootstrapMethod = runtimeClass.getDeclaredMethod("start", List.class, List.class);
+            this.startBootstrapMethod = runtimeClass.getDeclaredMethod(
+                    "start", List.class, List.class, boolean.class);
+            this.startWatchlessMethod = runtimeClass.getDeclaredMethod(
+                    "start", List.class, List.class, boolean.class, Runnable.class, Runnable.class);
             this.stopMethod = runtimeClass.getDeclaredMethod("stop");
             this.startMethod.setAccessible(true);
             this.startBootstrapMethod.setAccessible(true);
+            this.startWatchlessMethod.setAccessible(true);
             this.stopMethod.setAccessible(true);
         } catch (Exception | LinkageError e) {
             this.runtime = null;
@@ -108,12 +111,25 @@ public synchronized void start() {
         }
     }
 
-    public synchronized void start(List<String> registerAddrs, List<String> relayAddrs) {
+    public synchronized void start(List<String> registerAddrs, List<String> relayAddrs, boolean watchless) {
+        start(registerAddrs, relayAddrs, watchless, null, null);
+    }
+
+    public synchronized void start(
+            List<String> registerAddrs,
+            List<String> relayAddrs,
+            boolean watchless,
+            Runnable watchlessReady,
+            Runnable watchFallback) {
         if (runtime == null) {
             return;
         }
         try {
-            startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs);
+            if (watchlessReady == null && watchFallback == null) {
+                startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs, watchless);
+            } else {
+                startWatchlessMethod.invoke(runtime, registerAddrs, relayAddrs, watchless, watchlessReady, watchFallback);
+            }
         } catch (InvocationTargetException e) {
             stop();
             Throwable cause = e.getCause() == null ? e : e.getCause();

core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java

@@ -38,21 +38,29 @@ final class Libp2pEndpointConfig {
     static final String LISTEN_ADDR_ENV = "CONNECT_LIBP2P_LISTEN_ADDR";
     static final String ADVERTISE_ADDRS_ENV = "CONNECT_LIBP2P_ADVERTISE_ADDRS";
     static final String RELAY_ADDRS_ENV = "CONNECT_LIBP2P_RELAY_ADDRS";
+    private static final List<String> LEGACY_WATCH_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status"));
+    private static final List<String> WATCHLESS_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status", "watchless"));
 
     private final List<String> registerAddrs;
     private final List<String> listenAddrs;
     private final List<String> advertiseAddrs;
     private final List<String> relayAddrs;
+    private final List<String> capabilities;
+    private final boolean watchless;
 
     private Libp2pEndpointConfig(
             List<String> registerAddrs,
             List<String> listenAddrs,
             List<String> advertiseAddrs,
-            List<String> relayAddrs) {
+            List<String> relayAddrs,
+            List<String> capabilities,
+            boolean watchless) {
         this.registerAddrs = registerAddrs;
         this.listenAddrs = listenAddrs;
         this.advertiseAddrs = advertiseAddrs;
         this.relayAddrs = relayAddrs;
+        this.capabilities = capabilities;
+        this.watchless = watchless;
     }
 
     static Libp2pEndpointConfig fromEnvironment(Map<String, String> env) {
@@ -65,19 +73,30 @@ static Libp2pEndpointConfig fromEnvironment(Map<String, String> env) {
                 registerAddrs,
                 listenAddrs,
                 split(env.get(ADVERTISE_ADDRS_ENV)),
-                split(env.get(RELAY_ADDRS_ENV)));
+                split(env.get(RELAY_ADDRS_ENV)),
+                LEGACY_WATCH_CAPABILITIES,
+                false);
     }
 
     static Libp2pEndpointConfig fromSystemEnvironment() {
         return fromEnvironment(System.getenv());
     }
 
     static Libp2pEndpointConfig fromBootstrap(List<String> registerAddrs, List<String> relayAddrs) {
+        return fromBootstrap(registerAddrs, relayAddrs, false);
+    }
+
+    static Libp2pEndpointConfig fromBootstrap(
+            List<String> registerAddrs,
+            List<String> relayAddrs,
+            boolean watchless) {
         return new Libp2pEndpointConfig(
                 immutableCopy(registerAddrs),
                 Collections.singletonList("/ip4/127.0.0.1/tcp/0"),
                 Collections.emptyList(),
-                immutableCopy(relayAddrs));
+                immutableCopy(relayAddrs),
+                watchless ? WATCHLESS_CAPABILITIES : LEGACY_WATCH_CAPABILITIES,
+                watchless);
     }
 
     boolean enabled() {
@@ -100,6 +119,14 @@ List<String> relayAddrs() {
         return relayAddrs;
     }
 
+    List<String> capabilities() {
+        return capabilities;
+    }
+
+    boolean watchless() {
+        return watchless;
+    }
+
     List<String> edgePeerIds() {
         List<String> peerIds = new ArrayList<>();
         Set<String> seen = new HashSet<>();