feat(core): expose watcher health endpoint

README.md

@@ -32,6 +32,10 @@ Environment variables:
   the Connect edge can challenge the endpoint to sign equivalent
   `/p2p-circuit/p2p/<endpoint-peer-id>` 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
 

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();

core/src/main/java/com/minekube/connect/module/WatcherModule.java

@@ -26,12 +26,14 @@
 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 {
 
     @Override
     protected void configure() {
         bind(WatcherRegister.class).asEagerSingleton();
+        bind(WatchHealthServer.class).asEagerSingleton();
     }
 }

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;
+        }
+    }
+}

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() {

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"));