The probe enters port 2010 on the tunnel server and must come back out here; echoing its + * nonce proves the round trip went through this client's forward and not through a listener some + * other, dead session left behind. The host is in {@code .invalid}, which never resolves, so no + * real request can match it. + */ +public class ForwardProbeHandler extends Handler.Wrapper { + public static final String PROBE_HOST = "reverse-forward-probe.testingbot.invalid"; + public static final String PROBE_HEADER = "X-TestingBot-Probe"; + + @Override + public boolean handle(Request request, Response response, Callback callback) throws Exception { + if (!PROBE_HOST.equalsIgnoreCase(Request.getServerName(request))) { + return super.handle(request, response, callback); + } + String nonce = request.getHeaders().get(PROBE_HEADER); + response.setStatus(HttpStatus.NO_CONTENT_204); + if (nonce != null) { + response.getHeaders().put(PROBE_HEADER, nonce); + } + response.getHeaders().put(HttpHeader.CONNECTION, "close"); + callback.succeeded(); + return true; + } +} diff --git a/src/main/java/com/testingbot/tunnel/HttpProxy.java b/src/main/java/com/testingbot/tunnel/HttpProxy.java index cf1a729..98ec8bd 100644 --- a/src/main/java/com/testingbot/tunnel/HttpProxy.java +++ b/src/main/java/com/testingbot/tunnel/HttpProxy.java @@ -229,7 +229,11 @@ public void onClosed(org.eclipse.jetty.io.Connection connection) { // GracefulHandler tracks in-flight requests so stop() can drain them; it must sit // outermost to see every request. GracefulHandler gracefulHandler = new GracefulHandler(); - gracefulHandler.setHandler(logHandler); + // Before the log handler: the reverse-forward probe arrives every few seconds and is + // not traffic anyone wants to see in the HTTP log or the request metrics. + ForwardProbeHandler probeHandler = new ForwardProbeHandler(); + probeHandler.setHandler(logHandler); + gracefulHandler.setHandler(probeHandler); if (app.isDebugMode()) { // Jetty 12 handlers own a Callback that must be completed exactly once. This diff --git a/src/main/java/ssh/CustomConnectionMonitor.java b/src/main/java/ssh/CustomConnectionMonitor.java index 80402b6..6d38ded 100644 --- a/src/main/java/ssh/CustomConnectionMonitor.java +++ b/src/main/java/ssh/CustomConnectionMonitor.java @@ -26,6 +26,16 @@ public class CustomConnectionMonitor { static final long CURRENT_RETRY_DELAY = 5000; static final int MAX_RETRIES = 30; + /** + * The first reconnect goes out after this long, not after the full retry delay: the session + * has already been gone for the keepalive's ~15 seconds, and every second more is a second + * of failing tests. + */ + static final long FIRST_RETRY_DELAY = 1000; + /** Between re-requests of the reverse forward on a session that is up. */ + static final long FORWARD_RETRY_DELAY = 2000; + /** About two minutes of re-requests before the tunnel is rebuilt. */ + static final int MAX_FORWARD_RETRIES = 60; private final ReconnectableTunnel tunnel; private final ReconnectHost host; @@ -34,6 +44,7 @@ public class CustomConnectionMonitor { private final AtomicBoolean retrying = new AtomicBoolean(false); private int retryAttempts = 0; + private int forwardRetryAttempts = 0; private final Object lifecycleLock = new Object(); private volatile boolean cancelled; @@ -67,7 +78,7 @@ public void connectionLost(Throwable reason) { // Only the first loss starts a retry cycle; further reports while one is already in // flight would otherwise stack up timers all racing to reconnect the same tunnel. if (retrying.compareAndSet(false, true)) { - scheduleRetry(); + scheduleRetry(this::attemptReconnect, Math.min(FIRST_RETRY_DELAY, retryDelayMs)); } } } @@ -77,9 +88,13 @@ private void scheduleRetry() { } private void scheduleRetry(Runnable task) { + scheduleRetry(task, retryDelayMs); + } + + private void scheduleRetry(Runnable task, long delayMs) { synchronized (lifecycleLock) { if (!isStopped()) { - scheduler.scheduleOnce("Reconnect-" + tunnel.getConnectionId(), task, retryDelayMs); + scheduler.scheduleOnce("Reconnect-" + tunnel.getConnectionId(), task, delayMs); } } } @@ -171,14 +186,19 @@ private void finishReconnect() { return; } tunnel.createPortForwarding(); + checkReconnected(); + } + + /** Ready when the session and the forwards are up; otherwise schedules what is missing. */ + private void checkReconnected() { synchronized (lifecycleLock) { if (isStopped()) { tunnel.stop(); return; } - if (!tunnel.isAuthenticated() || !tunnel.isForwardingEstablished()) { + if (!tunnel.isAuthenticated()) { host.setReady(false); - LOG.log(Level.WARNING, "SSH reconnected but port forwarding failed; retrying setup"); + LOG.log(Level.WARNING, "SSH dropped again while setting up port forwarding; retrying"); if (retryAttempts >= MAX_RETRIES) { giveUpAndRebuild(); } else { @@ -186,6 +206,28 @@ private void finishReconnect() { } return; } + if (!tunnel.isForwardingEstablished()) { + // The session is up; what failed is the forward. On a tunnel server this is + // nearly always 2010 still held by the session this one replaces, which the + // server releases once it notices that session is gone. Tearing the healthy + // session down and re-dialling it every five seconds -- what this did -- made + // each attempt slower and burned the retry budget, so a stale listener that + // lived longer than ~3 minutes ended in a rebuild. Ask again on this session, + // often, and rebuild only if the port never comes free. + host.setReady(false); + forwardRetryAttempts += 1; + LOG.log(Level.WARNING, String.format( + "[%s] SSH reconnected but port forwarding failed; requesting it again (%d)", + tunnel.getConnectionId(), forwardRetryAttempts)); + if (forwardRetryAttempts >= MAX_FORWARD_RETRIES) { + forwardRetryAttempts = 0; + giveUpAndRebuild(); + } else { + scheduleRetry(this::retryForwarding, FORWARD_RETRY_DELAY); + } + return; + } + forwardRetryAttempts = 0; retrying.set(false); scheduler.cancel(); host.setReady(true); @@ -196,6 +238,19 @@ private void finishReconnect() { } } + /** Re-requests the forwards on the current session; reconnects if the session went away. */ + private void retryForwarding() { + if (isStopped()) { + return; + } + if (!tunnel.isAuthenticated()) { + attemptReconnect(); + return; + } + tunnel.restartPortForwarding(); + checkReconnected(); + } + /** * Restarts the local proxy without touching the SSH session. * diff --git a/src/main/java/ssh/ReconnectableTunnel.java b/src/main/java/ssh/ReconnectableTunnel.java index 56079ae..f9dd05c 100644 --- a/src/main/java/ssh/ReconnectableTunnel.java +++ b/src/main/java/ssh/ReconnectableTunnel.java @@ -28,4 +28,9 @@ public interface ReconnectableTunnel { default boolean isForwardingEstablished() { return false; } + + /** Drops whatever forwards are registered and requests them again, on the same session. */ + default void restartPortForwarding() { + createPortForwarding(); + } } diff --git a/src/main/java/ssh/SSHTunnel.java b/src/main/java/ssh/SSHTunnel.java index cd2bddc..8d9e86c 100644 --- a/src/main/java/ssh/SSHTunnel.java +++ b/src/main/java/ssh/SSHTunnel.java @@ -1,5 +1,6 @@ package ssh; +import com.jcraft.jsch.ChannelDirectTCPIP; import com.jcraft.jsch.JSch; import com.jcraft.jsch.ProxySOCKS5; import com.testingbot.tunnel.proxy.ProxySpec; @@ -7,10 +8,15 @@ import com.jcraft.jsch.Session; import com.jcraft.jsch.JSchException; import com.testingbot.tunnel.App; +import com.testingbot.tunnel.ForwardProbeHandler; import com.testingbot.tunnel.TunnelMetrics; import io.prometheus.client.Histogram; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.Locale; import java.util.Timer; import java.util.TimerTask; import java.util.UUID; @@ -34,6 +40,9 @@ public class SSHTunnel implements ReconnectableTunnel { private final AtomicBoolean shuttingDown = new AtomicBoolean(false); private final CustomConnectionMonitor connectionMonitor; private volatile boolean portForwardingEstablished = false; + /** Set once a reverse-forward probe has succeeded on this session: the server permits it. */ + private volatile boolean probeSupported; + private int probeFailures; /** Long enough for a slow network and a proxy hop, short enough to fail rather than hang. */ static final int DEFAULT_CONNECT_TIMEOUT_MS = 30_000; @@ -47,6 +56,28 @@ public class SSHTunnel implements ReconnectableTunnel { /** The port the tunnel server accepts browser traffic on and forwards back to us. */ static final int DEFAULT_REMOTE_PROXY_PORT = 2010; + /** + * How quickly a dead session is noticed: JSch sends a keepalive after this long without a + * packet from the server, and drops the session after {@link #SERVER_ALIVE_COUNT_MAX} + * unanswered ones. + * + *
Without it nothing read-side ever timed out. The 30-second keepalive below only writes, + * so after a NAT change or a laptop switching networks the session looked connected until the + * operating system gave up retransmitting -- about 15 minutes on Linux -- and all that time + * the tunnel server kept port 2010 for a client that was gone. + */ + static final int SERVER_ALIVE_INTERVAL_MS = 5_000; + static final int SERVER_ALIVE_COUNT_MAX = 3; + + /** How often the connection and forwarding monitors look. */ + static final long CONNECTION_CHECK_INTERVAL_MS = 2_000; + static final long FORWARDING_CHECK_INTERVAL_MS = 5_000; + + /** How long one end-to-end probe of the reverse forward may take. */ + static final int REVERSE_PROBE_TIMEOUT_MS = 5_000; + /** Consecutive failed probes before the reverse forward is re-registered. */ + static final int REVERSE_PROBE_FAILURES_BEFORE_REPAIR = 2; + private final int sshPort; private final String hubHost; private final int remoteProxyPort; @@ -178,9 +209,13 @@ public final void connect() throws Exception { /* Now connect */ long startTime = System.currentTimeMillis(); session = jsch.getSession(app.getClientKey(), server, sshPort); + probeSupported = false; + probeFailures = 0; session.setPassword(app.getClientSecret()); applyHostKeyVerification(session); applyUpstreamProxy(session); + session.setServerAliveInterval(SERVER_ALIVE_INTERVAL_MS); + session.setServerAliveCountMax(SERVER_ALIVE_COUNT_MAX); // Bounded, because connect() covers the TCP dial, the proxy's CONNECT and the SSH // banner exchange: a peer that accepts the socket and then says nothing -- a // silently dropping firewall, a proxy holding the connection -- otherwise left this @@ -219,9 +254,11 @@ public final void connect() throws Exception { keepAliveTimer = new Timer("KeepAlive-" + connectionId); keepAliveTimer.schedule(new KeepAliveTask(), 30000, 30000); connectionMonitorTimer = new Timer("ConnectionMonitor-" + connectionId); - connectionMonitorTimer.schedule(new ConnectionMonitorTask(), 10000, 10000); + connectionMonitorTimer.schedule(new ConnectionMonitorTask(), + CONNECTION_CHECK_INTERVAL_MS, CONNECTION_CHECK_INTERVAL_MS); portForwardingMonitorTimer = new Timer("PortForwardingMonitor-" + connectionId); - portForwardingMonitorTimer.schedule(new PortForwardingMonitorTask(), 15000, 15000); + portForwardingMonitorTimer.schedule(new PortForwardingMonitorTask(), + FORWARDING_CHECK_INTERVAL_MS, FORWARDING_CHECK_INTERVAL_MS); } } @@ -334,6 +371,74 @@ public boolean verifyReverseForwardDelivery() { } } + /** + * Sends a request into port 2010 on the tunnel server and checks it comes back out of this + * client's local proxy. + * + *
{@link #verifyReverseForwardDelivery()} only proves the local end. What fails in + * practice is the server end: the forward was never registered because an older session of + * the same account still held 2010, or the listener on 2010 belongs to a session whose client + * is gone. In both cases the tunnel looked healthy here while every device request through + * it failed. This opens a direct-tcpip channel to the server's own 127.0.0.1:2010 -- exactly + * where squid and the devices connect -- and sends a request only our proxy answers with this + * nonce. + * + * @return true when the probe made the round trip through this client's reverse forward + */ + boolean probeReverseForward(int timeoutMs) { + Session current = session; + if (current == null || !current.isConnected()) { + return false; + } + String nonce = UUID.randomUUID().toString(); + ChannelDirectTCPIP channel = null; + try { + channel = (ChannelDirectTCPIP) current.openChannel("direct-tcpip"); + channel.setHost("127.0.0.1"); + channel.setPort(remoteProxyPort); + InputStream in = channel.getInputStream(); + OutputStream out = channel.getOutputStream(); + channel.connect(timeoutMs); + out.write(("GET http://" + ForwardProbeHandler.PROBE_HOST + "/ HTTP/1.1\r\n" + + "Host: " + ForwardProbeHandler.PROBE_HOST + "\r\n" + + ForwardProbeHandler.PROBE_HEADER + ": " + nonce + "\r\n" + + "Connection: close\r\n\r\n").getBytes(StandardCharsets.US_ASCII)); + out.flush(); + String expected = ForwardProbeHandler.PROBE_HEADER.toLowerCase(Locale.ROOT) + ": " + nonce; + StringBuilder answer = new StringBuilder(); + byte[] buffer = new byte[1024]; + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline && answer.length() < 8192) { + if (in.available() > 0) { + int n = in.read(buffer); + if (n < 0) { + break; + } + answer.append(new String(buffer, 0, n, StandardCharsets.ISO_8859_1)); + if (answer.toString().toLowerCase(Locale.ROOT).contains(expected)) { + return true; + } + } else if (channel.isClosed()) { + break; + } else { + Thread.sleep(20); + } + } + return false; + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return false; + } catch (Exception failed) { + // The channel could not be opened: nothing listens on 2010, or the server does not + // permit the probe. Which of the two is decided by the caller. + return false; + } finally { + if (channel != null) { + channel.disconnect(); + } + } + } + /** The live JSch session; package-private so tests can inspect the real forwarding tables. */ Session getSession() { return session; @@ -506,6 +611,12 @@ public void run() { String.format("[%s] Reverse port forwarding delivery to %s:%d restored", connectionId, REVERSE_FORWARD_HOST, app.getJettyPort())); default -> { } } + + // A reconnect in flight owns the forwards; repairing them in parallel would + // race its own attempt. + if (!connectionMonitor.isRetrying()) { + checkRemoteForward(); + } } } catch (Exception ex) { Logger.getLogger(SSHTunnel.class.getName()).log(Level.WARNING, @@ -514,7 +625,51 @@ public void run() { } } - private void restartPortForwarding() { + /** + * Keeps port 2010 on the tunnel server pointing at this client. + * + *
A reverse forward that could not be registered -- typically because an older session of + * the same account still held 2010 -- was never asked for again while the SSH session stayed + * up, so the tunnel stayed up, not ready, and carrying nothing until the client was + * restarted. Now it is re-requested on every check until the server lets it go. A forward + * that is registered but no longer delivers (the probe fails twice in a row) is re-registered + * the same way. + */ + private void checkRemoteForward() { + if (!portForwardingEstablished) { + Logger.getLogger(SSHTunnel.class.getName()).log(Level.WARNING, String.format( + "[%s] Port forwarding is not established; requesting it again", connectionId)); + restartPortForwarding(); + if (portForwardingEstablished) { + app.reverseForwardRestored(); + } + return; + } + if (probeReverseForward(REVERSE_PROBE_TIMEOUT_MS)) { + probeSupported = true; + probeFailures = 0; + return; + } + if (!probeSupported) { + // Never succeeded on this session: the server does not allow the probe (an older + // image's PermitOpen), so a failure says nothing about the forward. + return; + } + probeFailures += 1; + if (probeFailures >= REVERSE_PROBE_FAILURES_BEFORE_REPAIR) { + probeFailures = 0; + Logger.getLogger(SSHTunnel.class.getName()).log(Level.SEVERE, String.format( + "[%s] Port %d on the tunnel server no longer reaches this client; re-registering it", + connectionId, remoteProxyPort)); + restartPortForwarding(); + if (portForwardingEstablished) { + app.reverseForwardRestored(); + } + } + } + + @Override + public synchronized void restartPortForwarding() { try { Logger.getLogger(SSHTunnel.class.getName()).log(Level.INFO, String.format("[%s] Restarting port forwarding", connectionId)); diff --git a/src/test/java/ssh/CustomConnectionMonitorTest.java b/src/test/java/ssh/CustomConnectionMonitorTest.java index 6f3343d..643e07f 100644 --- a/src/test/java/ssh/CustomConnectionMonitorTest.java +++ b/src/test/java/ssh/CustomConnectionMonitorTest.java @@ -23,11 +23,13 @@ private static final class ManualScheduler implements Scheduler { private Runnable pending; private int scheduled; private int cancels; + private long lastDelayMs = -1; @Override public void scheduleOnce(String name, Runnable task, long delayMs) { pending = task; scheduled++; + lastDelayMs = delayMs; } @Override @@ -103,6 +105,11 @@ public boolean isForwardingEstablished() { public void createPortForwarding() { calls.add("createPortForwarding"); } + + @Override + public void restartPortForwarding() { + calls.add("restartPortForwarding"); + } } private static final class FakeHost implements ReconnectHost { @@ -216,7 +223,8 @@ void permanentlyRefusedForwardingEventuallyRebuilds() { tunnel.authenticated = true; tunnel.forwardingEstablished = false; monitor.connectionLost(new RuntimeException("drop")); - for (int i = 0; i < CustomConnectionMonitor.MAX_RETRIES; i++) { + scheduler.fire(); // the reconnect, whose forward request is the first of them + for (int i = 1; i < CustomConnectionMonitor.MAX_FORWARD_RETRIES; i++) { scheduler.fire(); } assertThat(host.calls).contains("rebuildTunnel"); @@ -224,6 +232,47 @@ void permanentlyRefusedForwardingEventuallyRebuilds() { assertThat(com.testingbot.tunnel.TunnelMetrics.isTunnelUp()).isFalse(); } + @Test + void aRefusedForwardIsRequestedAgainOnTheSameSession() { + // 2010 still held by the session this one replaces: the SSH session is fine, only the + // forward failed. It used to be torn down and re-dialled every five seconds. + tunnel.authenticated = true; + tunnel.forwardingEstablished = false; + monitor.connectionLost(new RuntimeException("drop")); + scheduler.fire(); + assertThat(tunnel.calls).containsExactly("stop", "connect", "createPortForwarding"); + assertThat(scheduler.lastDelayMs).isEqualTo(CustomConnectionMonitor.FORWARD_RETRY_DELAY); + + scheduler.fire(); + scheduler.fire(); + assertThat(tunnel.calls).containsExactly("stop", "connect", "createPortForwarding", + "restartPortForwarding", "restartPortForwarding"); + assertThat(monitor.isRetrying()).isTrue(); + + tunnel.forwardingEstablished = true; // the server let go of 2010 + scheduler.fire(); + assertThat(monitor.isRetrying()).isFalse(); + assertThat(com.testingbot.tunnel.TunnelMetrics.isTunnelUp()).isTrue(); + assertThat(host.calls).doesNotContain("rebuildTunnel"); + } + + @Test + void aSessionThatDropsWhileTheForwardIsRetriedIsReconnected() { + tunnel.authenticated = true; + tunnel.forwardingEstablished = false; + monitor.connectionLost(new RuntimeException("drop")); + scheduler.fire(); + tunnel.authenticated = false; + scheduler.fire(); + assertThat(tunnel.calls).endsWith("stop", "connect"); + } + + @Test + void theFirstReconnectIsNotDelayedByTheFullRetryInterval() { + monitor.connectionLost(new RuntimeException("drop")); + assertThat(scheduler.lastDelayMs).isLessThanOrEqualTo(CustomConnectionMonitor.FIRST_RETRY_DELAY); + } + @Test void aDropStopsTheLocalProxyAndSchedulesARetry() { monitor.connectionLost(new RuntimeException("connection reset")); diff --git a/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java b/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java index b98f49e..943e56b 100644 --- a/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java +++ b/src/test/java/ssh/SSHTunnelEmbeddedServerTest.java @@ -2,6 +2,7 @@ import com.testingbot.tunnel.TestPorts; import com.testingbot.tunnel.App; +import com.testingbot.tunnel.ForwardProbeHandler; import org.apache.sshd.common.util.net.SshdSocketAddress; import org.apache.sshd.server.SshServer; import org.apache.sshd.server.forward.AcceptAllForwardingFilter; @@ -259,6 +260,100 @@ void reverseForwardingIsRegisteredOnTheServer() throws Exception { .anyMatch(forward -> forward.contains("2010")); } + /* ------------------------------------------- end-to-end probe of the reverse forward */ + + /** Stands in for the local proxy's ForwardProbeHandler: echoes the probe header back. */ + private ServerSocket probeAnsweringProxy() throws IOException { + ServerSocket server = new ServerSocket(0, 50, InetAddress.getLoopbackAddress()); + pool.submit(() -> { + while (!server.isClosed()) { + // the reader goes with the socket, after the response is written + try (Socket socket = server.accept(); + BufferedReader in = new BufferedReader(new InputStreamReader( + socket.getInputStream(), StandardCharsets.UTF_8))) { + String nonce = null; + String line; + while ((line = in.readLine()) != null && !line.isEmpty()) { + if (line.regionMatches(true, 0, ForwardProbeHandler.PROBE_HEADER + ":", 0, + ForwardProbeHandler.PROBE_HEADER.length() + 1)) { + nonce = line.substring(line.indexOf(':') + 1).trim(); + } + } + OutputStream out = socket.getOutputStream(); + out.write(("HTTP/1.1 204 No Content\r\n" + ForwardProbeHandler.PROBE_HEADER + + ": " + nonce + "\r\nConnection: close\r\n\r\n") + .getBytes(StandardCharsets.UTF_8)); + out.flush(); + } catch (IOException closed) { + return; + } + } + }); + return server; + } + + /** Counts the tunnel telling the app its reverse forward came back. */ + private static final class RestoreCountingApp extends App { + private final java.util.concurrent.atomic.AtomicInteger restored = + new java.util.concurrent.atomic.AtomicInteger(); + + @Override + public void reverseForwardRestored() { + restored.incrementAndGet(); + } + } + + @Test + void theProbeMakesTheRoundTripThroughTheServersReverseForward() throws Exception { + hub = serverAnswering("HUB"); + localProxy = probeAnsweringProxy(); + App app = app(); + app.setHubPort(hub.getLocalPort()); + app.setJettyPort(localProxy.getLocalPort()); + tunnel = new SSHTunnel(app, "127.0.0.1", sshd.getPort(), "127.0.0.1", findFreePort()); + tunnel.createPortForwarding(); + + assertThat(tunnel.probeReverseForward(5_000)).isTrue(); + } + + @Test + void theProbeFailsWhenTheServerHasNoReverseForwardForUs() throws Exception { + localProxy = probeAnsweringProxy(); + App app = app(); + app.setJettyPort(localProxy.getLocalPort()); + tunnel = new SSHTunnel(app, "127.0.0.1", sshd.getPort(), "127.0.0.1", findFreePort()); + + assertThat(tunnel.probeReverseForward(2_000)).isFalse(); + } + + @Test + void aReverseForwardTheServerRefusedIsRequestedAgainByTheMonitor() throws Exception { + // The production failure: an older session of the same account still holds 2010 on the + // tunnel server, so the forward is refused. It used to be asked for exactly once. + hub = serverAnswering("HUB"); + localProxy = probeAnsweringProxy(); + RestoreCountingApp app = new RestoreCountingApp(); + app.setClientKey(USER); + app.setClientSecret(PASSWORD); + app.setSeleniumPort(findFreePort()); + app.setHubPort(hub.getLocalPort()); + app.setJettyPort(localProxy.getLocalPort()); + int remotePort = findFreePort(); + tunnel = new SSHTunnel(app, "127.0.0.1", sshd.getPort(), "127.0.0.1", remotePort); + + try (ServerSocket heldByAStaleSession = new ServerSocket(remotePort, 50, + InetAddress.getLoopbackAddress())) { + tunnel.createPortForwarding(); + assertThat(tunnel.isForwardingEstablished()).isFalse(); + } + + tunnel.new PortForwardingMonitorTask().run(); + + assertThat(tunnel.isForwardingEstablished()).isTrue(); + assertThat(tunnel.probeReverseForward(5_000)).isTrue(); + assertThat(app.restored.get()).isEqualTo(1); + } + /* ----------------------------------------------------- reverse delivery probe */ @Test