diff --git a/README.md b/README.md index 0cbb69d8..d81bf404 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,10 @@ Environment variables: the Connect edge can challenge the endpoint to sign equivalent `/p2p-circuit/p2p/` addresses that are better for other edge proxies to dial, such as private per-machine relay addresses. +- `CONNECT_WATCH_HEALTH_ADDR`: optional `host:port` address for an HTTP watcher + health endpoint. `GET /healthz` returns `200` after the watcher opens and + while the negotiated watchless endpoint remains ready; it returns `503` + before opening, while reconnecting, and after stop, error, or completion. ## Working setups diff --git a/core/src/main/java/com/minekube/connect/ConnectPlatform.java b/core/src/main/java/com/minekube/connect/ConnectPlatform.java index c84281ca..7e589c5a 100644 --- a/core/src/main/java/com/minekube/connect/ConnectPlatform.java +++ b/core/src/main/java/com/minekube/connect/ConnectPlatform.java @@ -41,6 +41,7 @@ import com.minekube.connect.inject.CommonPlatformInjector; import com.minekube.connect.module.ConfigLoadedModule; import com.minekube.connect.module.PostInitializeModule; +import com.minekube.connect.register.WatchHealthServer; import com.minekube.connect.register.WatcherRegister; import com.minekube.connect.tunnel.Tunneler; import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; @@ -141,6 +142,7 @@ public boolean disable() { guice.getInstance(Libp2pEndpoint.class).stop(); } catch (ConfigurationException ignored) { } + guice.getInstance(WatchHealthServer.class).stop(); guice.getInstance(WatcherRegister.class).stop(); guice.getInstance(Tunneler.class).close(); guice.getInstance(CommonPlatformInjector.class).shutdown(); diff --git a/core/src/main/java/com/minekube/connect/module/WatcherModule.java b/core/src/main/java/com/minekube/connect/module/WatcherModule.java index d94920e7..d2a9f68b 100644 --- a/core/src/main/java/com/minekube/connect/module/WatcherModule.java +++ b/core/src/main/java/com/minekube/connect/module/WatcherModule.java @@ -26,6 +26,7 @@ package com.minekube.connect.module; import com.google.inject.AbstractModule; +import com.minekube.connect.register.WatchHealthServer; import com.minekube.connect.register.WatcherRegister; public class WatcherModule extends AbstractModule { @@ -33,5 +34,6 @@ public class WatcherModule extends AbstractModule { @Override protected void configure() { bind(WatcherRegister.class).asEagerSingleton(); + bind(WatchHealthServer.class).asEagerSingleton(); } } diff --git a/core/src/main/java/com/minekube/connect/register/WatchHealthServer.java b/core/src/main/java/com/minekube/connect/register/WatchHealthServer.java new file mode 100644 index 00000000..a9609029 --- /dev/null +++ b/core/src/main/java/com/minekube/connect/register/WatchHealthServer.java @@ -0,0 +1,129 @@ +/* + * Copyright (c) 2026 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 + * 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. + */ + +package com.minekube.connect.register; + +import com.google.inject.Inject; +import com.minekube.connect.api.logger.ConnectLogger; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.function.BooleanSupplier; + +public class WatchHealthServer { + private static final String HEALTH_ADDR_ENV = "CONNECT_WATCH_HEALTH_ADDR"; + private static final String HEALTH_PATH = "/healthz"; + + private final HttpServer server; + private final ExecutorService executor; + private final String baseUrl; + + @Inject + public WatchHealthServer(WatcherRegister watcherRegister, ConnectLogger logger) throws IOException { + this(System.getenv(HEALTH_ADDR_ENV), watcherRegister::isHealthy, logger); + } + + WatchHealthServer(String address, BooleanSupplier healthy, ConnectLogger logger) throws IOException { + if (address == null || address.trim().isEmpty()) { + server = null; + executor = null; + baseUrl = null; + return; + } + + InetSocketAddress socketAddress = parseAddress(address); + server = HttpServer.create(socketAddress, 0); + executor = Executors.newSingleThreadExecutor(new DaemonThreadFactory()); + baseUrl = "http://" + address; + + server.createContext("/", exchange -> handle(exchange, healthy)); + server.setExecutor(executor); + server.start(); + logger.info("Connect watcher health endpoint listening on " + address); + } + + public void stop() { + if (server != null) { + server.stop(0); + } + if (executor != null) { + executor.shutdownNow(); + } + } + + String url(String path) { + return baseUrl + path; + } + + private static void handle(HttpExchange exchange, BooleanSupplier healthy) throws IOException { + try { + if (!"GET".equals(exchange.getRequestMethod())) { + write(exchange, 405, "method not allowed\n"); + return; + } + if (!HEALTH_PATH.equals(exchange.getRequestURI().getPath())) { + write(exchange, 404, "not found\n"); + return; + } + + boolean isHealthy = healthy.getAsBoolean(); + write(exchange, isHealthy ? 200 : 503, isHealthy ? "ok\n" : "unhealthy\n"); + } finally { + exchange.close(); + } + } + + private static void write(HttpExchange exchange, int statusCode, String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.sendResponseHeaders(statusCode, bytes.length); + try (OutputStream response = exchange.getResponseBody()) { + response.write(bytes); + } + } + + private static InetSocketAddress parseAddress(String address) { + int separator = address.lastIndexOf(':'); + if (separator <= 0 || separator == address.length() - 1) { + throw new IllegalArgumentException( + HEALTH_ADDR_ENV + " must be in host:port form, for example 127.0.0.1:8087"); + } + + String host = address.substring(0, separator); + int port = Integer.parseInt(address.substring(separator + 1)); + return new InetSocketAddress(host, port); + } + + private static final class DaemonThreadFactory implements ThreadFactory { + @Override + public Thread newThread(Runnable runnable) { + Thread thread = new Thread(runnable, "connect-watch-health"); + thread.setDaemon(true); + return thread; + } + } +} 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 8c12fe91..eff9c395 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java @@ -70,6 +70,7 @@ public class WatcherRegister { private volatile boolean legacyWatchEnabled; private ExponentialBackOff backOffPolicy; private final AtomicBoolean started = new AtomicBoolean(); + private final AtomicBoolean healthy = new AtomicBoolean(); // Lazily created in start() so a stop()/start() cycle reuses cleanly, // and so the daemon thread isn't allocated if start() is never called. @@ -78,7 +79,7 @@ public class WatcherRegister { private volatile ScheduledFuture retryFuture; @Inject - public void start() { + public synchronized void start() { if (started.compareAndSet(false, true)) { legacyWatchEnabled = true; scheduler = Executors.newSingleThreadScheduledExecutor( @@ -97,12 +98,21 @@ public void resetBackOff() { backOffPolicy.reset(); } - public void stop() { + public boolean isHealthy() { + return healthy.get(); + } + + public synchronized void stop() { // Gate the whole teardown so a concurrent stop() races safely. // A stop() before start() is a no-op (started is already false). if (!started.compareAndSet(true, false)) { return; } + WatcherImpl currentWatcher = watcher; + if (currentWatcher != null) { + currentWatcher.ignoreTerminalEvents(); + } + healthy.set(false); logger.info("Stopped watching for sessions"); legacyWatchEnabled = false; if (retryFuture != null) { @@ -114,9 +124,6 @@ public void stop() { scheduler = null; } if (ws != null) { - if (watcher != null) { - watcher.ignoreTerminalEvents(); - } ws.close(1000, "watcher stopped"); ws = null; } @@ -155,21 +162,23 @@ private void retry() { }, millis, TimeUnit.MILLISECONDS); } - private void watch() { + private synchronized void watch() { if (!started.get() || !legacyWatchEnabled) { return; } + WatcherImpl currentWatcher = watcher; + if (currentWatcher != null) { + currentWatcher.ignoreTerminalEvents(); + } + healthy.set(false); if (ws != null) { - if (watcher != null) { - watcher.ignoreTerminalEvents(); - } ws.close(1000, "watcher is reconnecting"); } watcher = new WatcherImpl(); ws = watchClient.watch(watcher); } - private void enterWatchlessMode() { + private synchronized void enterWatchlessMode() { if (!started.get()) { return; } @@ -182,6 +191,7 @@ private void enterWatchlessMode() { if (currentWatcher != null) { currentWatcher.ignoreTerminalEvents(); } + healthy.set(true); WebSocket currentWebSocket = ws; ws = null; watcher = null; @@ -191,7 +201,7 @@ private void enterWatchlessMode() { } } - private void enterLegacyWatchFallback() { + private synchronized void enterLegacyWatchFallback() { if (!started.get()) { return; } @@ -209,6 +219,9 @@ private class WatcherImpl implements Watcher { @Override public void onOpen(WatchBootstrap bootstrap) { + if (!acceptOpen()) { + return; + } logger.translatedInfo("connect.watch.started"); if (bootstrap.hasLibp2p()) { if (bootstrap.supportsWatchlessLibp2p()) { @@ -267,7 +280,7 @@ public void onProposal(SessionProposal proposal) { @Override public void onError(Throwable t) { - if (shouldIgnoreTerminalEvent()) { + if (!acceptTerminalEvent()) { return; } logger.error("Connection error with WatchService: " + @@ -282,7 +295,7 @@ public void onError(Throwable t) { @Override public void onCompleted() { - if (shouldIgnoreTerminalEvent()) { + if (!acceptTerminalEvent()) { return; } cancelResetBackOffTimer(); @@ -313,8 +326,30 @@ void cancelResetBackOffTimer() { } void ignoreTerminalEvents() { - ignoreTerminalEvents = true; - cancelResetBackOffTimer(); + synchronized (WatcherRegister.this) { + ignoreTerminalEvents = true; + cancelResetBackOffTimer(); + } + } + + private boolean acceptOpen() { + synchronized (WatcherRegister.this) { + if (shouldIgnoreTerminalEvent()) { + return false; + } + healthy.set(true); + return true; + } + } + + private boolean acceptTerminalEvent() { + synchronized (WatcherRegister.this) { + if (shouldIgnoreTerminalEvent()) { + return false; + } + healthy.set(false); + return true; + } } private boolean shouldIgnoreTerminalEvent() { diff --git a/core/src/test/java/com/minekube/connect/module/CommonModuleTest.java b/core/src/test/java/com/minekube/connect/module/CommonModuleTest.java index 8d029c12..6c3c2f33 100644 --- a/core/src/test/java/com/minekube/connect/module/CommonModuleTest.java +++ b/core/src/test/java/com/minekube/connect/module/CommonModuleTest.java @@ -67,7 +67,6 @@ void connectTokenIsPersistedForAllConnectClients() throws Exception { void watchHttpClientKeepsConnectHeadersAndUsesWebSocketLiveness() throws Exception { CommonModule module = new CommonModule(tempDir); PlatformUtils platformUtils = platformUtils(); - OkHttpClient connectClient = module.connectOkHttpClient( module.defaultOkHttpClient(), platformUtils, @@ -79,6 +78,7 @@ void watchHttpClientKeepsConnectHeadersAndUsesWebSocketLiveness() throws Excepti assertEquals(0, watchClient.readTimeoutMillis()); assertEquals(30_000, watchClient.pingIntervalMillis()); + assertEquals(0, connectClient.pingIntervalMillis()); try (MockWebServer server = new MockWebServer()) { server.enqueue(new MockResponse().setBody("ok")); diff --git a/core/src/test/java/com/minekube/connect/register/WatchHealthServerTest.java b/core/src/test/java/com/minekube/connect/register/WatchHealthServerTest.java new file mode 100644 index 00000000..f4d0ca4c --- /dev/null +++ b/core/src/test/java/com/minekube/connect/register/WatchHealthServerTest.java @@ -0,0 +1,226 @@ +package com.minekube.connect.register; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +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.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.Tunneler; +import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; +import com.minekube.connect.watch.WatchBootstrap; +import com.minekube.connect.watch.WatchClient; +import com.minekube.connect.watch.Watcher; +import java.io.IOException; +import java.io.InputStream; +import java.lang.reflect.Field; +import java.net.HttpURLConnection; +import java.net.InetAddress; +import java.net.URI; +import java.net.ServerSocket; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import org.mockito.ArgumentCaptor; +import okhttp3.WebSocket; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +class WatchHealthServerTest { + private WatchHealthServer server; + + @AfterEach + void stopServer() { + if (server != null) { + server.stop(); + } + } + + @Test + void healthzReturnsOkWhenWatcherIsHealthy() throws Exception { + AtomicBoolean healthy = new AtomicBoolean(true); + server = new WatchHealthServer(loopbackAddress(), healthy::get, mock(ConnectLogger.class)); + + assertEquals(200, responseCode(server.url("/healthz"))); + } + + @Test + void healthzReturnsUnavailableWhenWatcherIsUnhealthy() throws Exception { + AtomicBoolean healthy = new AtomicBoolean(false); + server = new WatchHealthServer(loopbackAddress(), healthy::get, mock(ConnectLogger.class)); + + assertEquals(503, responseCode(server.url("/healthz"))); + } + + @Test + void healthzReflectsWatcherHealthChanges() throws Exception { + AtomicBoolean healthy = new AtomicBoolean(false); + server = new WatchHealthServer(loopbackAddress(), healthy::get, mock(ConnectLogger.class)); + + assertEquals(503, responseCode(server.url("/healthz"))); + + healthy.set(true); + + assertEquals(200, responseCode(server.url("/healthz"))); + } + + @Test + void unknownPathReturnsNotFound() throws Exception { + AtomicBoolean healthy = new AtomicBoolean(true); + server = new WatchHealthServer(loopbackAddress(), healthy::get, mock(ConnectLogger.class)); + + assertEquals(404, responseCode(server.url("/missing"))); + } + + @Test + void publicHealthEndpointTracksAcceptedWatcherLifecycle() throws Exception { + WatcherFixture fixture = newWatcherFixture(); + WatcherRegister register = fixture.register; + server = new WatchHealthServer(loopbackAddress(), register::isHealthy, mock(ConnectLogger.class)); + + assertResponse("before-start", response(server.url("/healthz")), 503, "unhealthy\n"); + register.start(); + assertResponse("connecting", response(server.url("/healthz")), 503, "unhealthy\n"); + + ArgumentCaptor watchers = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watchers.capture()); + watchers.getValue().onOpen(emptyBootstrap()); + assertResponse("watch-open", response(server.url("/healthz")), 200, "ok\n"); + + watchers.getValue().onCompleted(); + assertResponse("watch-completed", response(server.url("/healthz")), 503, "unhealthy\n"); + + register.stop(); + register.start(); + verify(fixture.watchClient, times(2)).watch(watchers.capture()); + Watcher restartedWatcher = watchers.getAllValues().get(2); + restartedWatcher.onOpen(emptyBootstrap()); + assertResponse("restarted-open", response(server.url("/healthz")), 200, "ok\n"); + + register.stop(); + assertResponse("stopped", response(server.url("/healthz")), 503, "unhealthy\n"); + } + + @Test + void publicHealthEndpointStaysHealthyWhenWatchlessModeIgnoresTerminalEvent() throws Exception { + WatcherFixture fixture = newWatcherFixture(); + CapturedCallbacks callbacks = new CapturedCallbacks(); + doAnswer(invocation -> { + callbacks.ready = invocation.getArgument(3); + return null; + }).when(fixture.libp2pEndpoint).start( + any(), any(), anyBoolean(), any(Runnable.class), any(Runnable.class)); + WatcherRegister register = fixture.register; + server = new WatchHealthServer(loopbackAddress(), register::isHealthy, mock(ConnectLogger.class)); + 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"))); + assertResponse("watchless-negotiated", response(server.url("/healthz")), 200, "ok\n"); + + callbacks.ready.run(); + watcher.getValue().onCompleted(); + + assertResponse("watchless-terminal-ignored", response(server.url("/healthz")), 200, "ok\n"); + register.stop(); + } + + private static String loopbackAddress() throws IOException { + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + return "127.0.0.1:" + socket.getLocalPort(); + } + } + + private static int responseCode(String url) throws IOException { + HttpURLConnection connection = (HttpURLConnection) URI.create(url).toURL().openConnection(); + connection.setRequestMethod("GET"); + connection.setConnectTimeout(1000); + connection.setReadTimeout(1000); + try { + return connection.getResponseCode(); + } finally { + connection.disconnect(); + } + } + + private static Response response(String url) throws IOException { + HttpURLConnection connection = (HttpURLConnection) URI.create(url).toURL().openConnection(); + connection.setRequestMethod("GET"); + connection.setConnectTimeout(1000); + connection.setReadTimeout(1000); + try { + int status = connection.getResponseCode(); + InputStream stream = status >= 400 ? connection.getErrorStream() : connection.getInputStream(); + String body = new String(stream.readAllBytes(), StandardCharsets.UTF_8); + return new Response(status, body); + } finally { + connection.disconnect(); + } + } + + private static void assertResponse(String state, Response response, int status, String body) { + System.out.printf("%-28s GET /healthz -> %d %s", state, response.status, response.body); + assertEquals(status, response.status); + assertEquals(body, response.body); + } + + private static WatchBootstrap emptyBootstrap() { + return WatchBootstrap.fromLists(List.of(), List.of(), List.of()); + } + + private static WatcherFixture newWatcherFixture() throws Exception { + WatcherRegister register = new WatcherRegister(); + WatchClient watchClient = mock(WatchClient.class); + when(watchClient.watch(any(Watcher.class))).thenReturn(mock(WebSocket.class)); + Libp2pEndpoint libp2pEndpoint = mock(Libp2pEndpoint.class); + inject(register, "watchClient", watchClient); + inject(register, "tunneler", mock(Tunneler.class)); + inject(register, "platformInjector", mock(PlatformInjector.class)); + inject(register, "logger", mock(ConnectLogger.class)); + inject(register, "api", mock(SimpleConnectApi.class)); + inject(register, "libp2pEndpoint", libp2pEndpoint); + return new WatcherFixture(register, watchClient, libp2pEndpoint); + } + + private static void inject(WatcherRegister register, String fieldName, Object value) throws Exception { + Field field = WatcherRegister.class.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(register, value); + } + + private static final class Response { + private final int status; + private final String body; + + private Response(int status, String body) { + this.status = status; + this.body = body; + } + } + + private static final class WatcherFixture { + private final WatcherRegister register; + private final WatchClient watchClient; + private final Libp2pEndpoint libp2pEndpoint; + + private WatcherFixture( + WatcherRegister register, WatchClient watchClient, Libp2pEndpoint libp2pEndpoint) { + this.register = register; + this.watchClient = watchClient; + this.libp2pEndpoint = libp2pEndpoint; + } + } + + private static final class CapturedCallbacks { + private Runnable ready; + } +} 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 1331ed98..29cfe199 100644 --- a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java +++ b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java @@ -131,7 +131,7 @@ void resetBackOffTimerDoesNotThrowWhenSchedulerIsClearedAfterStop() throws Excep register.stop(); - assertDoesNotThrow(() -> watcher.getValue().onOpen()); + assertDoesNotThrow(() -> watcher.getValue().onOpen(emptyBootstrap())); assertNull(scheduler(register)); } @@ -172,18 +172,44 @@ void watchlessReadyClosesLegacyWatchSocketAndSuppressesReconnect() throws Except 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"))); + assertTrue(register.isHealthy()); callbacks.ready.run(); watcher.getValue().onCompleted(); + assertTrue(register.isHealthy()); verify(webSocket).close(1000, "watchless endpoint mode enabled"); verifyNoMoreInteractions(fixture.watchClient); } + @Test + void watchlessReadyRestoresHealthAfterTerminalEventWinsRace() 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"))); + + watcher.getValue().onCompleted(); + callbacks.ready.run(); + + assertTrue(register.isHealthy()); + } + @Test void watchlessFallbackReopensLegacyWatchSocket() throws Exception { Fixture fixture = newFixture(); @@ -231,6 +257,80 @@ void oldWatchBootstrapKeepsLegacyWatchSocketPrimary() throws Exception { verifyNoMoreInteractions(webSocket); } + @Test + void watcherHealthIsUnhealthyUntilWatchSocketOpens() throws Exception { + Fixture fixture = newFixture(); + register = fixture.register; + + assertFalse(register.isHealthy()); + + register.start(); + + assertFalse(register.isHealthy()); + + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + watcher.getValue().onOpen(emptyBootstrap()); + + assertTrue(register.isHealthy()); + } + + @Test + void watcherHealthBecomesUnhealthyWhenWatchEnds() 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(emptyBootstrap()); + + watcher.getValue().onCompleted(); + + assertFalse(register.isHealthy()); + } + + @Test + void watcherHealthBecomesUnhealthyWhenWatchErrors() 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(emptyBootstrap()); + + watcher.getValue().onError(new RuntimeException("watch failed")); + + assertFalse(register.isHealthy()); + } + + @Test + void watcherHealthBecomesUnhealthyWhenStopped() 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(emptyBootstrap()); + + register.stop(); + + assertFalse(register.isHealthy()); + } + + @Test + void lateOpenAfterStopDoesNotMarkWatcherHealthy() throws Exception { + Fixture fixture = newFixture(); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + register.stop(); + watcher.getValue().onOpen(emptyBootstrap()); + + assertFalse(register.isHealthy()); + } + @Test void watcherRegisterDoesNotDeclareTimerFields() { assertNoTimerFields(WatcherRegister.class); @@ -275,6 +375,10 @@ private static WatcherRegister newRegister() throws Exception { return newFixture().register; } + private static WatchBootstrap emptyBootstrap() { + return WatchBootstrap.fromLists(List.of(), List.of(), List.of()); + } + private static Fixture newFixture() throws Exception { WatcherRegister register = new WatcherRegister(); WatchClient watchClient = mock(WatchClient.class); diff --git a/velocity/build.gradle.kts b/velocity/build.gradle.kts index 2ee9f39c..11f27eae 100644 --- a/velocity/build.gradle.kts +++ b/velocity/build.gradle.kts @@ -15,6 +15,8 @@ dependencies { testImplementation("org.junit.jupiter:junit-jupiter:5.10.5") testImplementation("io.netty", "netty-transport", Versions.nettyVersion) + testImplementation("org.mockito:mockito-core:4.11.0") + testRuntimeOnly("com.velocitypowered", "velocity-api", velocityVersion) testRuntimeOnly("org.junit.platform:junit-platform-launcher") } diff --git a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java index 4174740b..86cc5a65 100644 --- a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java +++ b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java @@ -37,6 +37,7 @@ import com.minekube.connect.util.ReflectionUtils; import com.velocitypowered.api.event.Subscribe; import com.velocitypowered.api.event.proxy.ProxyInitializeEvent; +import com.velocitypowered.api.event.proxy.ProxyShutdownEvent; import com.velocitypowered.api.plugin.annotation.DataDirectory; import java.nio.file.Path; @@ -70,4 +71,9 @@ public void onProxyInitialization(ProxyInitializeEvent event) { new WatcherModule() ); } + + @Subscribe + public void onProxyShutdown(ProxyShutdownEvent event) { + platform.disable(); + } } diff --git a/velocity/src/test/java/com/minekube/connect/VelocityPluginTest.java b/velocity/src/test/java/com/minekube/connect/VelocityPluginTest.java new file mode 100644 index 00000000..1f2d42ec --- /dev/null +++ b/velocity/src/test/java/com/minekube/connect/VelocityPluginTest.java @@ -0,0 +1,37 @@ +package com.minekube.connect; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.inject.Injector; +import com.google.inject.Module; +import com.minekube.connect.api.logger.ConnectLogger; +import com.velocitypowered.api.event.Subscribe; +import com.velocitypowered.api.event.proxy.ProxyShutdownEvent; +import java.lang.reflect.Method; +import java.nio.file.Path; +import org.junit.jupiter.api.Test; + +class VelocityPluginTest { + @Test + void proxyShutdownDisablesConnectPlatform() throws Exception { + ConnectPlatform platform = mock(ConnectPlatform.class); + Injector parentInjector = mock(Injector.class); + Injector childInjector = mock(Injector.class); + when(parentInjector.createChildInjector(any(Module[].class))).thenReturn(childInjector); + when(childInjector.getInstance(ConnectPlatform.class)).thenReturn(platform); + when(childInjector.getInstance(ConnectLogger.class)).thenReturn(mock(ConnectLogger.class)); + VelocityPlugin plugin = new VelocityPlugin(Path.of("."), parentInjector); + + Method shutdownHandler = assertDoesNotThrow(() -> + VelocityPlugin.class.getMethod("onProxyShutdown", ProxyShutdownEvent.class)); + assertNotNull(shutdownHandler.getAnnotation(Subscribe.class)); + shutdownHandler.invoke(plugin, new ProxyShutdownEvent()); + + verify(platform).disable(); + } +}