feat(runtime): negotiate bounded control liveness

README.md

@@ -137,6 +137,15 @@ public final class PublishedTunnel {
 `forwardTo()` remains available for existing services that already bind a local
 TCP port.
 
+The control channel uses a negotiated heartbeat deadline. A missed deadline or
+an unexpected control-transport EOF closes tunnel admission immediately, while
+streams already returned by `accept()` and active `forwardTo()` relays keep
+their independent payload sockets and drain normally. This prevents a
+transient or asymmetric control-path outage from interrupting an established
+application session. Explicit tunnel/control closure and protocol revocation
+remain hard lifecycle operations enforced by the engine. Treat stream EOF or
+I/O failure—not `control.done()`—as the payload-lifecycle signal.
+
 ## Private dial
 
 ```java

src/main/java/io/rstream/BytestreamTunnel.java

@@ -5,21 +5,35 @@
 import java.io.OutputStream;
 import java.net.Socket;
 import java.time.Duration;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.Objects;
-import java.util.concurrent.BlockingQueue;
+import java.util.Set;
+import java.util.WeakHashMap;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 
 /** A bytestream tunnel opened on the rstream engine. */
 public final class BytestreamTunnel implements AutoCloseable {
-  private static final Object CLOSED = new Object();
   private final ControlChannel control;
   private final TunnelProperties properties;
-  private final BlockingQueue<Object> streams = new LinkedBlockingQueue<>();
+  private final StreamQueue streams = new StreamQueue();
   private final ExecutorService executor;
+  private final Object lifecycleLock = new Object();
+  private final Set<FutureTask<Void>> forwarders = new HashSet<>();
+  // Application-owned accepted streams must not be retained solely by lifecycle tracking.
+  private final Set<RstreamStream> payloads = Collections.newSetFromMap(new WeakHashMap<>());
+  private volatile boolean hardClosed;
   private volatile boolean closed;
+  private RuntimeException hardCloseError;
 
   BytestreamTunnel(ControlChannel control, TunnelProperties properties, ExecutorService executor) {
     if (properties.id() == null || properties.id().isBlank()) {
@@ -47,15 +61,13 @@ public String forwardingAddress() {
   }
 
   public RstreamStream accept() throws InterruptedException {
-    return accepted(streams.take());
+    return finishAcceptance(streams.take(null));
   }
 
   public RstreamStream accept(Duration timeout) throws InterruptedException {
     Objects.requireNonNull(timeout, "timeout");
     if (timeout.isNegative()) throw new IllegalArgumentException("timeout must not be negative");
-    var item = streams.poll(timeout.toMillis(), TimeUnit.MILLISECONDS);
-    if (item == null) return null;
-    return accepted(item);
+    return finishAcceptance(streams.take(timeout));
   }
 
   public CompletableFuture<RstreamStream> acceptAsync() {
@@ -83,7 +95,7 @@ public CompletableFuture<Void> forwardTo(String host, int port) {
           while (!closed) {
             try {
               var stream = accept();
-              executor.submit(() -> pipeToLocal(stream, host, port));
+              startForwarder(stream, host, port);
             } catch (InterruptedException error) {
               Thread.currentThread().interrupt();
               return;
@@ -105,25 +117,58 @@ public CompletableFuture<Void> serveHttp(RstreamHttpHandler handler, RstreamHttp
   }
 
   boolean deliver(RstreamStream stream) {
-    if (closed) {
-      stream.closeQuietly();
-      return false;
+    synchronized (lifecycleLock) {
+      if (closed) {
+        stream.closeQuietly();
+        return false;
+      }
+      payloads.add(stream);
     }
-    streams.offer(stream);
-    return true;
+    return streams.offer(stream);
   }
 
   void onClose(Throwable error) {
-    if (closed) return;
-    closed = true;
-    if (error != null) streams.offer(error);
-    streams.offer(CLOSED);
+    onClose(error, false);
+  }
+
+  void onClose(Throwable error, boolean preserveForwarders) {
+    var closeError =
+        error instanceof RuntimeException runtimeError
+            ? runtimeError
+            : new RstreamException("Tunnel closed.", "ERR_RSTREAM_TUNNEL_CLOSED", error);
+    var firstClose = false;
+    Set<FutureTask<Void>> canceled = Set.of();
+    Set<RstreamStream> closedPayloads = Set.of();
+    synchronized (lifecycleLock) {
+      if (!closed) {
+        closed = true;
+        firstClose = true;
+      }
+      if (!preserveForwarders && !hardClosed) {
+        hardClosed = true;
+        hardCloseError = closeError;
+        canceled = Set.copyOf(forwarders);
+        closedPayloads = Set.copyOf(payloads);
+        payloads.clear();
+      }
+    }
+    if (firstClose) streams.close(closeError);
+    closedPayloads.forEach(RstreamStream::closeQuietly);
+    canceled.forEach(task -> task.cancel(true));
   }
 
   @Override
   public void close() {
-    if (closed) return;
-    control.closeTunnel(id());
+    if (closed) {
+      onClose(null, false);
+      return;
+    }
+    try {
+      control.closeTunnel(id());
+    } catch (RuntimeException error) {
+      if (closed) onClose(error, false);
+      throw error;
+    }
   }
 
   public CompletableFuture<Void> closeAsync() {
@@ -149,17 +194,6 @@ static String formatForwardingAddress(TunnelProperties properties) {
         "Invalid tunnel properties: no host, name, or ID.", "ERR_RSTREAM_INVALID_TUNNEL");
   }
 
-  private static RstreamStream accepted(Object item) {
-    if (item == CLOSED) {
-      throw new RstreamException("Tunnel closed.", "ERR_RSTREAM_TUNNEL_CLOSED");
-    }
-    if (item instanceof RuntimeException error) throw error;
-    if (item instanceof Throwable error) {
-      throw new RstreamException("Tunnel closed.", "ERR_RSTREAM_TUNNEL_CLOSED", error);
-    }
-    return (RstreamStream) item;
-  }
-
   private static String publishedHost(TunnelProperties properties) {
     if (properties.hostname() != null && !properties.hostname().isBlank()) {
       var port = properties.port() == null ? 443 : properties.port();
@@ -208,6 +242,57 @@ private static void pipeToLocal(RstreamStream stream, String host, int port) {
     }
   }
 
+  private void startForwarder(RstreamStream stream, String host, int port) {
+    var task =
+        new FutureTask<Void>(() -> pipeToLocal(stream, host, port), null) {
+          @Override
+          protected void done() {
+            forwarderDone(this);
+          }
+        };
+    synchronized (lifecycleLock) {
+      if (hardClosed) {
+        stream.closeQuietly();
+        return;
+      }
+      forwarders.add(task);
+    }
+    try {
+      executor.execute(task);
+    } catch (RejectedExecutionException error) {
+      synchronized (lifecycleLock) {
+        forwarders.remove(task);
+      }
+      stream.closeQuietly();
+      throw new RstreamException(
+          "Failed to schedule tunnel forwarding.", "ERR_RSTREAM_FORWARD", error);
+    }
+  }
+
+  private RstreamStream finishAcceptance(RstreamStream stream) {
+    if (stream == null) return null;
+    RuntimeException closeError;
+    synchronized (lifecycleLock) {
+      closeError = hardClosed ? hardCloseError : null;
+    }
+    if (closeError == null) return stream;
+    stream.closeQuietly();
+    throw closeError;
+  }
+
+  private void forwarderDone(FutureTask<Void> task) {
+    synchronized (lifecycleLock) {
+      forwarders.remove(task);
+    }
+    try {
+      task.get();
+    } catch (CancellationException ignored) {
+    } catch (InterruptedException error) {
+      Thread.currentThread().interrupt();
+    } catch (ExecutionException ignored) {
+    }
+  }
+
   private static void copy(InputStream input, OutputStream output) {
     try {
       input.transferTo(output);
@@ -222,4 +307,66 @@ private static void shutdownOutput(Socket socket) {
     } catch (IOException ignored) {
     }
   }
+
+  private static final class StreamQueue {
+    private final ReentrantLock lock = new ReentrantLock();
+    private final Condition available = lock.newCondition();
+    private final ArrayDeque<RstreamStream> queued = new ArrayDeque<>();
+    private RuntimeException closeError;
+
+    RstreamStream take(Duration timeout) throws InterruptedException {
+      var remaining = timeout == null ? 0 : timeoutNanos(timeout);
+      lock.lockInterruptibly();
+      try {
+        while (queued.isEmpty()) {
+          if (closeError != null) throw closeError;
+          if (timeout == null) available.await();
+          else if (remaining <= 0) return null;
+          else remaining = available.awaitNanos(remaining);
+        }
+        return queued.removeFirst();
+      } finally {
+        lock.unlock();
+      }
+    }
+
+    boolean offer(RstreamStream stream) {
+      var accepted = false;
+      lock.lock();
+      try {
+        if (closeError == null) {
+          queued.addLast(stream);
+          available.signal();
+          accepted = true;
+        }
+      } finally {
+        lock.unlock();
+      }
+      if (!accepted) stream.closeQuietly();
+      return accepted;
+    }
+
+    void close(RuntimeException error) {
+      ArrayList<RstreamStream> abandoned;
+      lock.lock();
+      try {
+        if (closeError != null) return;
+        closeError = error;
+        abandoned = new ArrayList<>(queued);
+        queued.clear();
+        available.signalAll();
+      } finally {
+        lock.unlock();
+      }
+      abandoned.forEach(RstreamStream::closeQuietly);
+    }
+
+    private static long timeoutNanos(Duration timeout) {
+      try {
+        return timeout.toNanos();
+      } catch (ArithmeticException ignored) {
+        return Long.MAX_VALUE;
+      }
+    }
+  }
 }

src/main/java/io/rstream/ClientOptions.java

@@ -33,6 +33,13 @@ public record ClientOptions(
     if (operationTimeout.isZero() || operationTimeout.isNegative()) {
       throw new IllegalArgumentException("operationTimeout must be positive");
     }
+    if (heartbeat
+        && (heartbeatInterval.compareTo(Duration.ofSeconds(1)) < 0
+            || heartbeatInterval.compareTo(Duration.ofMinutes(5)) > 0
+            || heartbeatInterval.toNanosPart() % 1_000_000 != 0)) {
+      throw new IllegalArgumentException(
+          "heartbeatInterval must be between 1 second and 5 minutes with millisecond precision");
+    }
   }
 
   public static Builder builder() {