diff --git a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java index 6bca70100..8c12fe916 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/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(); + } } } diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java index 275830eee..4a8783138 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java +++ b/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 registerAddrs, List relayAddrs) { + public synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { + start(registerAddrs, relayAddrs, watchless, null, null); + } + + public synchronized void start( + List registerAddrs, + List 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(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java index 7d8dceb09..781260ae4 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java +++ b/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 LEGACY_WATCH_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status")); + private static final List WATCHLESS_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status", "watchless")); private final List registerAddrs; private final List listenAddrs; private final List advertiseAddrs; private final List relayAddrs; + private final List capabilities; + private final boolean watchless; private Libp2pEndpointConfig( List registerAddrs, List listenAddrs, List advertiseAddrs, - List relayAddrs) { + List relayAddrs, + List 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 env) { @@ -65,7 +73,9 @@ static Libp2pEndpointConfig fromEnvironment(Map 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() { @@ -73,11 +83,20 @@ static Libp2pEndpointConfig fromSystemEnvironment() { } static Libp2pEndpointConfig fromBootstrap(List registerAddrs, List relayAddrs) { + return fromBootstrap(registerAddrs, relayAddrs, false); + } + + static Libp2pEndpointConfig fromBootstrap( + List registerAddrs, + List 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 relayAddrs() { return relayAddrs; } + List capabilities() { + return capabilities; + } + + boolean watchless() { + return watchless; + } + List edgePeerIds() { List peerIds = new ArrayList<>(); Set seen = new HashSet<>(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java index 6e6725d59..56653424f 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java @@ -67,6 +67,7 @@ final class Libp2pEndpointRuntime { private static final long STREAM_TIMEOUT_SECONDS = 5; private static final int REGISTER_ATTEMPTS_PER_ADDRESS = 4; private static final int RELAY_RESERVATION_ATTEMPTS_PER_ADDRESS = 4; + private static final int WATCHLESS_STATUS_FAILURE_THRESHOLD = 3; private final Path dataDirectory; private final ConnectConfig connectConfig; @@ -84,6 +85,7 @@ final class Libp2pEndpointRuntime { private ScheduledExecutorService registrationExecutor; private ScheduledExecutorService statusExecutor; private ActiveRegistration activeRegistration; + private Libp2pWatchlessController watchlessController; private List reservedRelayAddrs = Collections.emptyList(); private boolean started; @@ -111,18 +113,31 @@ public synchronized void start() { startWithConfig(libp2pConfig); } - synchronized void start(List registerAddrs, List relayAddrs) { + synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { + start(registerAddrs, relayAddrs, watchless, null, null); + } + + synchronized void start( + List registerAddrs, + List relayAddrs, + boolean watchless, + Runnable watchlessReady, + Runnable watchFallback) { if (started) { return; } Libp2pEndpointConfig environmentConfig = Libp2pEndpointConfig.fromSystemEnvironment(); libp2pConfig = environmentConfig.enabled() ? environmentConfig - : Libp2pEndpointConfig.fromBootstrap(registerAddrs, relayAddrs); - startWithConfig(libp2pConfig); + : Libp2pEndpointConfig.fromBootstrap(registerAddrs, relayAddrs, watchless); + startWithConfig(libp2pConfig, watchlessReady, watchFallback); } private void startWithConfig(Libp2pEndpointConfig config) { + startWithConfig(config, null, null); + } + + private void startWithConfig(Libp2pEndpointConfig config, Runnable watchlessReady, Runnable watchFallback) { if (started) { return; } @@ -130,6 +145,11 @@ private void startWithConfig(Libp2pEndpointConfig config) { if (!libp2pConfig.enabled()) { return; } + watchlessController = new Libp2pWatchlessController( + libp2pConfig.watchless(), + WATCHLESS_STATUS_FAILURE_THRESHOLD, + watchlessReady, + watchFallback); try { identity = EndpointPeerIdentity.loadOrCreate(dataDirectory.resolve("libp2p-identity.key")); host = Libp2pTunnelTransportRuntime.createHost( @@ -164,6 +184,7 @@ public synchronized void stop() { activeRegistration.close(); activeRegistration = null; } + watchlessController = null; reservedRelayAddrs = Collections.emptyList(); if (statusExecutor != null) { statusExecutor.shutdownNow(); @@ -244,7 +265,7 @@ private ActiveRegistration registerOnce(OfflineMode offlineMode, EndpointAuthTyp : connectConfig.getSuperEndpoints(), offlineMode, authType, - Arrays.asList("session", "status"), + libp2pConfig.capabilities(), this::currentCapacity); client = new PeerRegistrationClient(handshake); PeerRegisterResult result = await(client.install( @@ -344,6 +365,8 @@ private void scheduleRegisterRetry(OfflineMode offlineMode, EndpointAuthType aut private void registerAndWatch(OfflineMode offlineMode, EndpointAuthType authType) { ActiveRegistration registration = registerOnce(offlineMode, authType); ActiveRegistration previous; + Libp2pWatchlessController controller; + boolean statusHealthy; synchronized (this) { if (!started) { registration.close(); @@ -353,12 +376,27 @@ private void registerAndWatch(OfflineMode offlineMode, EndpointAuthType authType activeRegistration = registration; logger.info("Connect libp2p endpoint registered: " + registration.result().getEndpointId() + " (" + identity.peerId() + ")"); - startStatusReporter(registration.result()); + statusHealthy = startStatusReporter(registration.result()); + controller = watchlessController; } if (previous != null) { previous.close(); } + if (controller != null) { + controller.registrationCommitted(statusHealthy); + } registration.client().closedFuture().whenComplete((ignored, error) -> { + Libp2pWatchlessController currentController; + synchronized (this) { + if (activeRegistration != registration) { + return; + } + currentController = watchlessController; + stopStatusReporter(); + } + if (currentController != null) { + currentController.registrationClosed(); + } if (error == null) { logger.info("Connect libp2p registration stream closed; reconnecting"); } else if (Libp2pEndpointErrors.isTransientConnectError(error)) { @@ -518,11 +556,8 @@ private static T await(CompletableFuture future, long timeoutSeconds, Str } } - private void startStatusReporter(PeerRegisterResult result) { - if (statusExecutor != null) { - statusExecutor.shutdownNow(); - statusExecutor = null; - } + private boolean startStatusReporter(PeerRegisterResult result) { + stopStatusReporter(); Libp2pStatusReporter reporter = new Libp2pStatusReporter( host, endpointInstanceId, @@ -531,17 +566,34 @@ private void startStatusReporter(PeerRegisterResult result) { result, platformUtils, logger); - reporter.reportSafely(); + boolean initialStatusHealthy = reporter.reportSafely(); statusExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, "connect-libp2p-status-report"); thread.setDaemon(true); return thread; }); statusExecutor.scheduleWithFixedDelay( - reporter::reportSafely, + () -> { + boolean healthy = reporter.reportSafely(); + Libp2pWatchlessController controller; + synchronized (this) { + controller = watchlessController; + } + if (controller != null) { + controller.statusReportResult(healthy); + } + }, Libp2pStatusReporter.REPORT_INTERVAL_SECONDS, Libp2pStatusReporter.REPORT_INTERVAL_SECONDS, TimeUnit.SECONDS); + return initialStatusHealthy; + } + + private void stopStatusReporter() { + if (statusExecutor != null) { + statusExecutor.shutdownNow(); + statusExecutor = null; + } } private static final class ActiveRegistration { diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java index db96c913f..ce2769191 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java @@ -98,9 +98,10 @@ final class Libp2pStatusReporter { this.streamOpener = streamOpener == null ? this::openStatusStream : streamOpener; } - void reportSafely() { + boolean reportSafely() { try { reportOnce(System.currentTimeMillis()); + return true; } catch (RuntimeException e) { if (Libp2pEndpointErrors.isTransientConnectError(e)) { logger.warn("Connect libp2p status report failed; retrying: " @@ -108,6 +109,7 @@ void reportSafely() { } else { logger.error("Failed to report Connect libp2p status", e); } + return false; } } diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java new file mode 100644 index 000000000..515059b1c --- /dev/null +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java @@ -0,0 +1,91 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * 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 + * DEALINGS IN THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import java.util.Objects; + +final class Libp2pWatchlessController { + private static final Runnable NOOP = () -> { + }; + + private final boolean supported; + private final int statusFailureThreshold; + private final Runnable readyCallback; + private final Runnable fallbackCallback; + + private boolean ready; + private boolean registrationActive; + private int statusFailures; + + Libp2pWatchlessController( + boolean supported, + int statusFailureThreshold, + Runnable readyCallback, + Runnable fallbackCallback) { + this.supported = supported; + this.statusFailureThreshold = Math.max(1, statusFailureThreshold); + this.readyCallback = readyCallback == null ? NOOP : Objects.requireNonNull(readyCallback, "readyCallback"); + this.fallbackCallback = fallbackCallback == null ? NOOP : Objects.requireNonNull(fallbackCallback, "fallbackCallback"); + } + + synchronized void registrationCommitted(boolean statusHealthy) { + if (!supported) { + return; + } + registrationActive = true; + statusReportResult(statusHealthy); + } + + synchronized void statusReportResult(boolean healthy) { + if (!supported || !registrationActive) { + return; + } + if (healthy) { + statusFailures = 0; + if (!ready) { + ready = true; + readyCallback.run(); + } + return; + } + statusFailures++; + if (ready && statusFailures >= statusFailureThreshold) { + ready = false; + fallbackCallback.run(); + } + } + + synchronized void registrationClosed() { + if (!supported) { + return; + } + registrationActive = false; + statusFailures = 0; + if (!ready) { + return; + } + ready = false; + fallbackCallback.run(); + } +} diff --git a/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java b/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java index 708c1eff1..a6be64e6f 100644 --- a/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java +++ b/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java @@ -30,25 +30,45 @@ public final class WatchBootstrap { public static final String LIBP2P_EDGE_ADDRS_HEADER = "Connect-Libp2p-Edge-Addrs"; public static final String LIBP2P_RELAY_ADDRS_HEADER = "Connect-Libp2p-Relay-Addrs"; + public static final String LIBP2P_CAPABILITIES_HEADER = "Connect-Libp2p-Capabilities"; + + private static final String CAPABILITY_WATCHLESS = "watchless"; private final List libp2pEdgeAddrs; private final List libp2pRelayAddrs; + private final List libp2pCapabilities; - private WatchBootstrap(List libp2pEdgeAddrs, List libp2pRelayAddrs) { - this.libp2pEdgeAddrs = Collections.unmodifiableList(new ArrayList<>(libp2pEdgeAddrs)); - this.libp2pRelayAddrs = Collections.unmodifiableList(new ArrayList<>(libp2pRelayAddrs)); + private WatchBootstrap( + List libp2pEdgeAddrs, + List libp2pRelayAddrs, + List libp2pCapabilities) { + this.libp2pEdgeAddrs = immutableCopy(libp2pEdgeAddrs); + this.libp2pRelayAddrs = immutableCopy(libp2pRelayAddrs); + this.libp2pCapabilities = immutableCopy(libp2pCapabilities); } public static WatchBootstrap fromHeaders(Headers headers) { - return new WatchBootstrap( + return fromLists( parseHeaderValues(headers.values(LIBP2P_EDGE_ADDRS_HEADER)), - parseHeaderValues(headers.values(LIBP2P_RELAY_ADDRS_HEADER))); + parseHeaderValues(headers.values(LIBP2P_RELAY_ADDRS_HEADER)), + parseHeaderValues(headers.values(LIBP2P_CAPABILITIES_HEADER))); + } + + public static WatchBootstrap fromLists( + List libp2pEdgeAddrs, + List libp2pRelayAddrs, + List libp2pCapabilities) { + return new WatchBootstrap(libp2pEdgeAddrs, libp2pRelayAddrs, libp2pCapabilities); } public boolean hasLibp2p() { return !libp2pEdgeAddrs.isEmpty(); } + public boolean supportsWatchlessLibp2p() { + return libp2pCapabilities.contains(CAPABILITY_WATCHLESS); + } + public List libp2pEdgeAddrs() { return libp2pEdgeAddrs; } @@ -75,4 +95,11 @@ private static List parseHeaderValues(List values) { } return out; } + + private static List immutableCopy(List values) { + if (values == null || values.isEmpty()) { + return Collections.emptyList(); + } + return Collections.unmodifiableList(new ArrayList<>(values)); + } } diff --git a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java index 6fef80016..1331ed989 100644 --- a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java +++ b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java @@ -7,23 +7,32 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; import com.minekube.connect.api.SimpleConnectApi; import com.minekube.connect.api.inject.PlatformInjector; import com.minekube.connect.api.logger.ConnectLogger; +import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import com.minekube.connect.tunnel.Tunneler; import com.minekube.connect.watch.SessionProposal; +import com.minekube.connect.watch.WatchBootstrap; import com.minekube.connect.watch.WatchClient; import com.minekube.connect.watch.Watcher; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.net.InetSocketAddress; +import java.util.List; import java.util.Timer; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; +import org.mockito.ArgumentCaptor; import minekube.connect.v1alpha1.WatchServiceOuterClass.GameProfile; import minekube.connect.v1alpha1.WatchServiceOuterClass.Player; import minekube.connect.v1alpha1.WatchServiceOuterClass.Session; @@ -32,7 +41,7 @@ import okhttp3.WebSocket; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; -import org.mockito.ArgumentCaptor; +import org.mockito.stubbing.Answer; class WatcherRegisterTest { private WatcherRegister register; @@ -126,6 +135,102 @@ void resetBackOffTimerDoesNotThrowWhenSchedulerIsClearedAfterStop() throws Excep assertNull(scheduler(register)); } + @Test + void watchOpenStartsLibp2pEndpointFromBootstrap() throws Exception { + Fixture fixture = newFixture(); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + + verify(fixture.libp2pEndpoint).start( + eq(List.of("/dns4/connect.example/tcp/4001/p2p/edge")), + eq(List.of("/dns4/connect.example/tcp/4001/p2p/edge")), + eq(true), + any(Runnable.class), + any(Runnable.class)); + } + + @Test + void watchlessReadyClosesLegacyWatchSocketAndSuppressesReconnect() throws Exception { + Fixture fixture = newFixture(); + RunnableCaptor callbacks = new RunnableCaptor(); + doAnswer(callbacks.capture()).when(fixture.libp2pEndpoint).start( + any(), + any(), + anyBoolean(), + any(Runnable.class), + any(Runnable.class)); + WebSocket webSocket = mock(WebSocket.class); + when(fixture.watchClient.watch(any(Watcher.class))).thenReturn(webSocket); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + callbacks.ready.run(); + watcher.getValue().onCompleted(); + + verify(webSocket).close(1000, "watchless endpoint mode enabled"); + verifyNoMoreInteractions(fixture.watchClient); + } + + @Test + void watchlessFallbackReopensLegacyWatchSocket() throws Exception { + Fixture fixture = newFixture(); + RunnableCaptor callbacks = new RunnableCaptor(); + doAnswer(callbacks.capture()).when(fixture.libp2pEndpoint).start( + any(), + any(), + anyBoolean(), + any(Runnable.class), + any(Runnable.class)); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + callbacks.ready.run(); + callbacks.fallback.run(); + + verify(fixture.watchClient, times(2)).watch(any(Watcher.class)); + } + + @Test + void oldWatchBootstrapKeepsLegacyWatchSocketPrimary() throws Exception { + Fixture fixture = newFixture(); + WebSocket webSocket = mock(WebSocket.class); + when(fixture.watchClient.watch(any(Watcher.class))).thenReturn(webSocket); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status"))); + + verify(fixture.libp2pEndpoint).start( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + false); + verifyNoMoreInteractions(webSocket); + } + @Test void watcherRegisterDoesNotDeclareTimerFields() { assertNoTimerFields(WatcherRegister.class); @@ -180,10 +285,12 @@ private static Fixture newFixture() throws Exception { inject(register, "platformInjector", mock(PlatformInjector.class)); inject(register, "logger", mock(ConnectLogger.class)); inject(register, "api", mock(SimpleConnectApi.class)); + inject(register, "libp2pEndpoint", mock(Libp2pEndpoint.class)); when(((PlatformInjector) getField(register, "platformInjector")).getServerSocketAddress()) .thenReturn(new InetSocketAddress("127.0.0.1", 25565)); return new Fixture(register, watchClient, (Tunneler) getField(register, "tunneler"), - (PlatformInjector) getField(register, "platformInjector")); + (PlatformInjector) getField(register, "platformInjector"), + (Libp2pEndpoint) getField(register, "libp2pEndpoint")); } private static void inject(WatcherRegister register, String fieldName, Object value) @@ -225,17 +332,33 @@ private static final class Fixture { private final WatchClient watchClient; private final Tunneler tunneler; private final PlatformInjector platformInjector; + private final Libp2pEndpoint libp2pEndpoint; private Fixture( WatcherRegister register, WatchClient watchClient, Tunneler tunneler, - PlatformInjector platformInjector + PlatformInjector platformInjector, + Libp2pEndpoint libp2pEndpoint ) { this.register = register; this.watchClient = watchClient; this.tunneler = tunneler; this.platformInjector = platformInjector; + this.libp2pEndpoint = libp2pEndpoint; + } + } + + private static final class RunnableCaptor { + private Runnable ready; + private Runnable fallback; + + private Answer capture() { + return invocation -> { + ready = invocation.getArgument(3); + fallback = invocation.getArgument(4); + return null; + }; } } } diff --git a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java index 603650c2c..85e017c19 100644 --- a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java +++ b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java @@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.List; import java.util.Map; import org.junit.jupiter.api.Test; @@ -17,6 +18,30 @@ void disabledWhenRegisterAddressIsMissing() { Libp2pEndpointConfig config = Libp2pEndpointConfig.fromEnvironment(Map.of()); assertFalse(config.enabled()); + assertEquals(List.of("session", "status"), config.capabilities()); + } + + @Test + void bootstrapConfigEnablesWatchlessWhenSupportedByEdge() { + Libp2pEndpointConfig config = Libp2pEndpointConfig.fromBootstrap( + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + true); + + assertTrue(config.enabled()); + assertEquals(List.of("/ip4/127.0.0.1/tcp/0"), config.listenAddrs()); + assertEquals(List.of("session", "status", "watchless"), config.capabilities()); + } + + @Test + void bootstrapConfigKeepsLegacyWatchCapabilityWhenWatchlessIsUnsupported() { + Libp2pEndpointConfig config = Libp2pEndpointConfig.fromBootstrap( + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + false); + + assertTrue(config.enabled()); + assertEquals(List.of("session", "status"), config.capabilities()); } @Test diff --git a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java new file mode 100644 index 000000000..ab75de545 --- /dev/null +++ b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java @@ -0,0 +1,89 @@ +package com.minekube.connect.tunnel.p2p; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class Libp2pWatchlessControllerTest { + @Test + void registrationAndHealthyStatusEnterWatchlessOnce() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(true); + + assertEquals(1, ready.count); + assertEquals(0, fallback.count); + } + + @Test + void registrationCloseFallsBackAfterWatchlessReady() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.registrationClosed(); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void repeatedStatusFailuresFallBackAfterThreshold() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(false); + controller.statusReportResult(false); + controller.statusReportResult(false); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void staleHealthyStatusAfterRegistrationCloseDoesNotReenterWatchless() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.registrationClosed(); + controller.statusReportResult(true); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void unsupportedWatchlessDoesNotNotify() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + false, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(false); + controller.registrationClosed(); + + assertEquals(0, ready.count); + assertEquals(0, fallback.count); + } + + private static final class Counter { + private int count; + + private void increment() { + count++; + } + } +} diff --git a/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java b/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java index 5a0012060..819246bd6 100644 --- a/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java +++ b/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java @@ -13,20 +13,34 @@ void parsesLibp2pBootstrapHeaders() { Headers headers = new Headers.Builder() .add(WatchBootstrap.LIBP2P_EDGE_ADDRS_HEADER, " /ip4/127.0.0.1/tcp/4001/p2p/a , /ip4/127.0.0.1/tcp/4001/p2p/b ") .add(WatchBootstrap.LIBP2P_RELAY_ADDRS_HEADER, "/ip4/127.0.0.1/tcp/4001/p2p/a") + .add(WatchBootstrap.LIBP2P_CAPABILITIES_HEADER, "session,status,watchless") .build(); WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(headers); assertTrue(bootstrap.hasLibp2p()); + assertTrue(bootstrap.supportsWatchlessLibp2p()); assertEquals(2, bootstrap.libp2pEdgeAddrs().size()); assertEquals("/ip4/127.0.0.1/tcp/4001/p2p/a", bootstrap.libp2pEdgeAddrs().get(0)); assertEquals("/ip4/127.0.0.1/tcp/4001/p2p/a", bootstrap.libp2pRelayAddrs().get(0)); } + @Test + void missingWatchlessCapabilityKeepsLegacyWatchMode() { + WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(new Headers.Builder() + .add(WatchBootstrap.LIBP2P_EDGE_ADDRS_HEADER, "/ip4/127.0.0.1/tcp/4001/p2p/a") + .add(WatchBootstrap.LIBP2P_CAPABILITIES_HEADER, "session,status") + .build()); + + assertTrue(bootstrap.hasLibp2p()); + assertFalse(bootstrap.supportsWatchlessLibp2p()); + } + @Test void emptyHeadersDisableLibp2pBootstrap() { WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(Headers.of()); assertFalse(bootstrap.hasLibp2p()); + assertFalse(bootstrap.supportsWatchlessLibp2p()); } }