Fix several bugs with websocket closing

- Correctly send a close websocket frame when a websocket is manually
   closed (status code 1000) or is garbage collected (1001). Computer
   shutdowns still just close the socket, as we don't/can't wait for the
   socket to cleanly shutdown.

 - After a user calls close(), don't mark the websocket as closed until
   we receive the server's close response. This means we now have a
   slightly awkward interstitial state where we can receive messages but
   not send them, but I don't think there's much we can do about that.

Fixes #2096
This commit is contained in:
Jonathan Coates
2026-09-29 20:52:10 +01:00
parent f0cfd0929b
commit 5f69f7acd6
9 changed files with 154 additions and 67 deletions
@@ -4,7 +4,7 @@
package dan200.computercraft.core.apis.http;
import dan200.computercraft.core.util.GlobalCleaner;
import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
import dan200.computercraft.core.util.IoUtil;
import io.netty.channel.ChannelFuture;
import org.jspecify.annotations.Nullable;
@@ -64,21 +64,13 @@ public abstract class Resource<T extends Resource<T>> implements Closeable {
* Note, this may be called multiple times, and so should be thread-safe and
* avoid any major side effects.
*/
@OverridingMethodsMustInvokeSuper
protected void dispose() {
@SuppressWarnings("unchecked")
var thisT = (T) this;
limiter.release(thisT);
}
/**
* Register an object which, when garbage collected, will delete this object.
*
* @param object The object to reference to
*/
protected final void registerCleanable(Object object) {
GlobalCleaner.register(object, this::dispose);
}
@Override
public final void close() {
tryClose();
@@ -12,6 +12,7 @@ import dan200.computercraft.core.apis.http.*;
import dan200.computercraft.core.apis.http.options.Options;
import dan200.computercraft.core.metrics.Metrics;
import dan200.computercraft.core.util.AtomicHelpers;
import dan200.computercraft.core.util.GlobalCleaner;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
@@ -30,12 +31,14 @@ import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Provides functionality to verify and connect to a remote websocket.
*/
public class Websocket extends Resource<Websocket> implements WebsocketClient {
public final class Websocket extends Resource<Websocket> implements WebsocketClient {
private static final Logger LOG = LoggerFactory.getLogger(Websocket.class);
/**
@@ -44,8 +47,14 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
*/
public static final int MAX_MESSAGE_SIZE = 1 << 30;
/**
* The timeout after sending a {@link CloseWebSocketFrame} after which the channel should be force-closed.
*/
private static final long CLOSE_TIMEOUT = 5;
private @Nullable Future<?> executorFuture;
private @Nullable ChannelFuture channelFuture;
private @Nullable ScheduledFuture<?> closeFuture;
private final IAPIEnvironment environment;
private final NetworkUtils network;
@@ -56,6 +65,7 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
private final AtomicInteger inFlight = new AtomicInteger(0);
private final GenericFutureListener<? extends io.netty.util.concurrent.Future<? super Void>> onSend = f -> inFlight.decrementAndGet();
private boolean isClosing = false;
public Websocket(ResourceGroup<Websocket> limiter, IAPIEnvironment environment, NetworkUtils network, String address, URI uri, HttpHeaders headers, int timeout) {
super(limiter);
@@ -102,15 +112,15 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
);
})
.addListener(c -> {
if (!c.isSuccess()) failure(NetworkUtils.toFriendlyError(c.cause()));
if (!c.isSuccess()) handshakeFailure(NetworkUtils.toFriendlyError(c.cause()));
});
// Do an additional check for cancellation
checkClosed();
} catch (HTTPRequestException e) {
failure(NetworkUtils.toFriendlyError(e));
handshakeFailure(NetworkUtils.toFriendlyError(e));
} catch (Exception e) {
failure(NetworkUtils.toFriendlyError(e));
handshakeFailure(NetworkUtils.toFriendlyError(e));
LOG.error(Logging.HTTP_ERROR, "Error in websocket", e);
}
}
@@ -125,17 +135,17 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
var handle = new WebsocketHandle(environment, address, this, headers, options);
GlobalCleaner.register(handle, () -> close(1001, ""));
environment().queueEvent(SUCCESS_EVENT, address, handle);
registerCleanable(handle);
checkClosed();
}
void failure(String message) {
void handshakeFailure(String message) {
if (tryClose()) environment.queueEvent(FAILURE_EVENT, address, message);
}
void close(int status, String reason) {
void serverClose(int status, String reason) {
if (tryClose()) {
environment.queueEvent(CLOSE_EVENT, address,
Strings.isNullOrEmpty(reason) ? null : reason,
@@ -149,6 +159,7 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
executorFuture = closeFuture(executorFuture);
channelFuture = closeChannel(channelFuture);
closeFuture = closeFuture(closeFuture);
}
IAPIEnvironment environment() {
@@ -177,7 +188,7 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
private void sendMessage(WebSocketFrame frame, long size) throws LuaException {
var channel = channel();
if (channel == null) return;
if (channel == null || isClosing) throw new LuaException(CLOSED_ERROR);
// Grow the number of in-flight requests, aborting if we've hit the limit. This is then decremented when the
// promise finishes.
@@ -188,4 +199,22 @@ public class Websocket extends Resource<Websocket> implements WebsocketClient {
environment.observe(Metrics.WEBSOCKET_OUTGOING, size);
channel.writeAndFlush(frame).addListener(onSend);
}
@Override
public void close(int status, String reason) {
// Close is idempotent, so should not fail if the socket is already closed.
var channel = channel();
if (channel == null || isClosing) return;
isClosing = true;
var packet = new CloseWebSocketFrame(status, reason);
environment.observe(Metrics.WEBSOCKET_OUTGOING, packet.content().readableBytes());
channel.writeAndFlush(packet);
// Schedule a callback to force-close the websocket if the server does not respond. This reimplements
// WebSocketProtocolHandler.applyCloseSentTimeout — unfortunately our CustomWebSocketHandshaker means we cannot
// specify a custom forceCloseTimeoutMillis.
closeFuture = channel.eventLoop().schedule(() -> serverClose(1006, ""), CLOSE_TIMEOUT, TimeUnit.SECONDS);
}
}
@@ -6,8 +6,8 @@ package dan200.computercraft.core.apis.http.websocket;
import dan200.computercraft.api.lua.LuaException;
import dan200.computercraft.core.apis.http.HTTPRequestException;
import io.netty.handler.codec.http.websocketx.WebSocketCloseStatus;
import java.io.Closeable;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
@@ -17,12 +17,17 @@ import java.nio.ByteBuffer;
* <p>
* {@link WebsocketHandle} wraps this into a Lua-compatible interface.
*/
public interface WebsocketClient extends Closeable {
public interface WebsocketClient {
String SUCCESS_EVENT = "websocket_success";
String FAILURE_EVENT = "websocket_failure";
String CLOSE_EVENT = "websocket_closed";
String MESSAGE_EVENT = "websocket_message";
/**
* Error message to throw when attempting to use a {@linkplain #isClosed() closed} websocket.
*/
String CLOSED_ERROR = "attempt to use a closed websocket";
/**
* Determine whether this websocket is closed.
*
@@ -32,9 +37,13 @@ public interface WebsocketClient extends Closeable {
/**
* Close this websocket.
*
* @param status The status code for this closure. Must be a {@linkplain WebSocketCloseStatus#isValidStatusCode(int)
* valid status code}.
* @param reason The reason for this closure.
* @see WebSocketCloseStatus
*/
@Override
void close();
void close(int status, String reason);
/**
* Send a text websocket frame.
@@ -31,7 +31,7 @@ import static dan200.computercraft.core.apis.http.websocket.WebsocketClient.MESS
* @cc.module http.Websocket
* @see dan200.computercraft.core.apis.HTTPAPI#websocket On how to open a websocket.
*/
public class WebsocketHandle {
public final class WebsocketHandle {
private static final ThreadLocal<CharsetDecoder> DECODER = ThreadLocal.withInitial(() -> StandardCharsets.UTF_8.newDecoder().onMalformedInput(CodingErrorAction.REPLACE));
private final IAPIEnvironment environment;
@@ -64,8 +64,9 @@ public class WebsocketHandle {
* @cc.changed 1.117.0 Added return value indicating why receiving the message failed.
*/
@LuaFunction
public final MethodResult receive(Optional<Double> timeout) throws LuaException {
checkOpen();
public MethodResult receive(Optional<Double> timeout) throws LuaException {
if (websocket.isClosed()) throw new LuaException(WebsocketClient.CLOSED_ERROR);
var timeoutId = timeout.isPresent()
? environment.startTimer(Math.round(checkFinite(0, timeout.get()) / 0.05))
: -1;
@@ -83,8 +84,8 @@ public class WebsocketHandle {
* @cc.changed 1.81.0 Added argument for binary mode.
*/
@LuaFunction
public final void send(Coerced<ByteBuffer> message, Optional<Boolean> binary) throws LuaException {
checkOpen();
public void send(Coerced<ByteBuffer> message, Optional<Boolean> binary) throws LuaException {
if (websocket.isClosed()) throw new LuaException(WebsocketClient.CLOSED_ERROR);
var text = message.value();
if (options.websocketMessage() != 0 && text.remaining() > options.websocketMessage()) {
@@ -108,8 +109,8 @@ public class WebsocketHandle {
* along it.
*/
@LuaFunction
public final void close() {
websocket.close();
public void close() {
websocket.close(1000, "");
}
/**
@@ -132,14 +133,10 @@ public class WebsocketHandle {
* @since 1.117.0
*/
@LuaFunction
public final Map<String, String> getResponseHeaders() {
public Map<String, String> getResponseHeaders() {
return responseHeaders;
}
private void checkOpen() throws LuaException {
if (websocket.isClosed()) throw new LuaException("attempt to use a closed file");
}
private final class ReceiveCallback implements ILuaCallback {
final MethodResult pull = MethodResult.pullEvent(null, this);
private final IAPIEnvironment environment;
@@ -41,7 +41,7 @@ class WebsocketHandler extends SimpleChannelInboundHandler<Object> {
websocket.success(headers, options);
handshakeComplete = true;
} else if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_TIMEOUT) {
websocket.failure("Timed out");
websocket.handshakeFailure("Timed out");
}
}
@@ -65,7 +65,7 @@ class WebsocketHandler extends SimpleChannelInboundHandler<Object> {
websocket.environment().observe(Metrics.WEBSOCKET_INCOMING, data.length);
websocket.environment().queueEvent(MESSAGE_EVENT, websocket.address(), data, true);
} else if (frame instanceof CloseWebSocketFrame closeFrame) {
websocket.close(closeFrame.statusCode(), closeFrame.reasonText());
websocket.serverClose(closeFrame.statusCode(), closeFrame.reasonText());
}
}
@@ -77,9 +77,9 @@ class WebsocketHandler extends SimpleChannelInboundHandler<Object> {
private void fail(String message) {
if (handshakeComplete) {
websocket.close(-1, message);
websocket.serverClose(-1, message);
} else {
websocket.failure(message);
websocket.handshakeFailure(message);
}
}
}
@@ -12,15 +12,26 @@ import io.netty.channel.nio.NioEventLoopGroup
import io.netty.channel.socket.SocketChannel
import io.netty.channel.socket.nio.NioServerSocketChannel
import io.netty.handler.codec.http.*
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame
import io.netty.handler.codec.http.websocketx.WebSocketFrame
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler
import io.netty.handler.codec.http.websocketx.*
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler.HandshakeComplete
import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketServerCompressionHandler
import java.net.InetSocketAddress
import java.nio.charset.StandardCharsets
class HttpServer(val port: Int, private val workerGroup: EventLoopGroup, private val activeConnections: Set<Channel>) {
class HttpRequest(val headers: HttpHeaders)
private class HttpState(var lastRequest: HttpRequest? = null)
class HttpServer private constructor(
val port: Int,
private val workerGroup: EventLoopGroup,
private val activeConnections: Set<Channel>,
private val state: HttpState,
) {
/** The last HTTP request made to this server */
val lastRequest: HttpRequest
get() = state.lastRequest ?: throw NullPointerException("No request has occurred yet")
/** Stop the server from running */
fun stop() {
workerGroup.shutdownGracefully()
@@ -33,10 +44,17 @@ class HttpServer(val port: Int, private val workerGroup: EventLoopGroup, private
companion object {
/** Runs a small HTTP server to run alongside [TestHttpApi] */
fun runServer(run: (server: HttpServer) -> Unit) {
fun runServer(closeWebsocket: Boolean = true, run: (server: HttpServer) -> Unit) {
val workerGroup: EventLoopGroup = NioEventLoopGroup(2)
val activeConnections = mutableSetOf<Channel>()
try {
val activeConnections = mutableSetOf<Channel>()
val state = HttpState()
val websocketConfig = WebSocketServerProtocolConfig.newBuilder()
.websocketPath("/ws")
.handleCloseFrames(closeWebsocket)
.allowExtensions(true)
.build()
val ch = ServerBootstrap()
.group(workerGroup)
.channel(NioServerSocketChannel::class.java)
@@ -47,16 +65,16 @@ class HttpServer(val port: Int, private val workerGroup: EventLoopGroup, private
p.addLast(HttpServerCodec())
p.addLast(HttpContentCompressor())
p.addLast(HttpObjectAggregator(8192))
p.addLast(HttpServerHandler())
p.addLast(HttpServerHandler(state))
p.addLast(WebSocketServerCompressionHandler())
p.addLast(WebSocketServerProtocolHandler("/ws", null, true))
p.addLast(WebSocketServerProtocolHandler(websocketConfig))
p.addLast(WebSocketFrameHandler(activeConnections))
}
},
).bind(0).sync().channel()
val port = (ch.localAddress() as InetSocketAddress).port
try {
run(HttpServer(port, workerGroup, activeConnections))
run(HttpServer(port, workerGroup, activeConnections, state))
} finally {
ch.close().sync()
}
@@ -70,7 +88,7 @@ class HttpServer(val port: Int, private val workerGroup: EventLoopGroup, private
/**
* A HTTP handler which hosts `/` (a simple static page) and `/ws` (see [WebSocketFrameHandler])
*/
private class HttpServerHandler : SimpleChannelInboundHandler<FullHttpRequest>() {
private class HttpServerHandler(private val state: HttpState) : SimpleChannelInboundHandler<FullHttpRequest>() {
companion object {
private val CONTENT = "Hello, world!".toByteArray(StandardCharsets.UTF_8)
}
@@ -80,6 +98,7 @@ private class HttpServerHandler : SimpleChannelInboundHandler<FullHttpRequest>()
}
override fun channelRead0(ctx: ChannelHandlerContext, request: FullHttpRequest) {
state.lastRequest = HttpRequest(request.headers())
when (request.uri()) {
"/", "/index.html" -> handleIndex(ctx, request)
"/ws" -> handleWebsocket(ctx, request)
@@ -124,12 +143,16 @@ private class HttpServerHandler : SimpleChannelInboundHandler<FullHttpRequest>()
*/
private class WebSocketFrameHandler(private val activeConnections: MutableSet<Channel>) : SimpleChannelInboundHandler<WebSocketFrame>() {
override fun channelRead0(ctx: ChannelHandlerContext, frame: WebSocketFrame) {
if (frame is TextWebSocketFrame) {
// Send the uppercase string back.
val request = frame.text()
ctx.channel().writeAndFlush(TextWebSocketFrame(request.uppercase()))
} else {
throw UnsupportedOperationException("unsupported frame type: ${frame.javaClass.name}")
when (frame) {
is TextWebSocketFrame -> {
// Send the uppercase string back.
val request = frame.text()
ctx.channel().writeAndFlush(TextWebSocketFrame(request.uppercase()))
}
is CloseWebSocketFrame -> {}
else -> throw UnsupportedOperationException("unsupported frame type: ${frame.javaClass.name}")
}
}
@@ -18,14 +18,16 @@ import dan200.computercraft.core.apis.http.request.HttpResponseHandle
import dan200.computercraft.core.apis.http.websocket.WebsocketHandle
import dan200.computercraft.test.core.computer.LuaTaskRunner
import io.netty.buffer.Unpooled
import io.netty.handler.codec.http.HttpHeaderNames
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame
import org.hamcrest.MatcherAssert.assertThat
import org.hamcrest.Matchers.*
import org.junit.jupiter.api.Assertions.assertFalse
import org.junit.jupiter.api.Test
import org.junit.jupiter.api.assertThrows
import java.lang.ref.Reference
import java.util.*
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
class TestHttpApi {
/** Create a [HTTPAPI] which is only permitted to access the current server. */
@@ -73,8 +75,10 @@ class TestHttpApi {
websocket.close()
val closeEvent = pullEventOrTimeout(500.milliseconds, "websocket_closed")
assertThat("No event was queued", closeEvent, equalTo(null))
val closeEvent = pullEvent()
assertThat("Received a close event", closeEvent, array(equalTo("websocket_closed"), equalTo(url), nullValue(), equalTo(1000)))
assertFalse(server.lastRequest.headers.contains(HttpHeaderNames.ORIGIN), "HTTP Headers should not contain Origin")
Reference.reachabilityFence(websocket)
}
@@ -161,4 +165,26 @@ class TestHttpApi {
}
}
}
@Test
fun `Closes websocket if the server doesn't respond`() {
runServer(closeWebsocket = false) { server ->
LuaTaskRunner.runTest(timeout = 10.seconds) {
val url = "ws://127.0.0.1:${server.port}/ws"
val httpApi = addApi(createHttpApi(environment, server.port))
assertThat("http.websocket succeeded", httpApi.websocket(ObjectArguments(url)), array(equalTo(true)))
val connectEvent = pullEvent()
assertThat(connectEvent, array(equalTo("websocket_success"), equalTo(url), isA(WebsocketHandle::class.java)))
val websocket = connectEvent[2] as WebsocketHandle
websocket.close()
val closeEvent = pullEvent()
assertThat("Received a close event", closeEvent, array(equalTo("websocket_closed"), equalTo(url), nullValue(), equalTo(1006)))
Reference.reachabilityFence(websocket)
}
}
}
}
+1 -1
View File
@@ -255,7 +255,7 @@ tasks.verifyClassesMatch {
option("ModLoader", "forge")
}
// We have an extra adapater between SimpleJsonResourceReloadListener.prepare and SimplePreparableReloadListener.prepare.
// We have an extra adapter between SimpleJsonResourceReloadListener.prepare and SimplePreparableReloadListener.prepare.
// Not quite sure why, but doesn't cause issues.
ignoredFiles.add("dan200/computercraft/impl/UpgradeManager.class")
// Forge widens the inner classes of RenderType, so we end up with different InnerClasses attributes. Doesn't cause
@@ -6,6 +6,7 @@ package cc.tweaked.web.http;
import cc.tweaked.web.js.Console;
import com.google.common.base.Strings;
import dan200.computercraft.api.lua.LuaException;
import dan200.computercraft.core.apis.IAPIEnvironment;
import dan200.computercraft.core.apis.http.Resource;
import dan200.computercraft.core.apis.http.ResourceGroup;
@@ -13,6 +14,7 @@ import dan200.computercraft.core.apis.http.options.Action;
import dan200.computercraft.core.apis.http.options.Options;
import dan200.computercraft.core.apis.http.websocket.WebsocketClient;
import dan200.computercraft.core.apis.http.websocket.WebsocketHandle;
import dan200.computercraft.core.util.GlobalCleaner;
import org.jspecify.annotations.Nullable;
import org.teavm.jso.typedarrays.ArrayBuffer;
import org.teavm.jso.typedarrays.Int8Array;
@@ -32,6 +34,7 @@ final class Websocket extends Resource<Websocket> implements WebsocketClient {
private final String address;
private @Nullable WebSocket websocket;
private boolean isClosing = false;
Websocket(ResourceGroup<Websocket> limiter, IAPIEnvironment environment, String address, URI uri) {
super(limiter);
@@ -48,7 +51,7 @@ final class Websocket extends Resource<Websocket> implements WebsocketClient {
client.onOpen(e -> success(Action.ALLOW.toPartial().toOptions()));
client.onError(e -> {
Console.error(e);
failure("Could not connect");
handshakeFailure("Could not connect");
});
client.onMessage(e -> {
if (isClosed()) return;
@@ -59,21 +62,29 @@ final class Websocket extends Resource<Websocket> implements WebsocketClient {
environment.queueEvent("websocket_message", address, e.getDataAsString(), false);
}
});
client.onClose(e -> close(e.getCode(), e.getReason()));
client.onClose(e -> serverClosed(e.getCode(), e.getReason()));
}
@Override
public void sendText(String message) {
if (websocket == null) return;
public void sendText(String message) throws LuaException {
if (websocket == null || isClosing) throw new LuaException(CLOSED_ERROR);
websocket.send(message);
}
@Override
public void sendBinary(ByteBuffer message) {
if (websocket == null) return;
public void sendBinary(ByteBuffer message) throws LuaException {
if (websocket == null || isClosing) throw new LuaException(CLOSED_ERROR);
websocket.send(Int8Array.fromJavaBuffer(message));
}
@Override
public void close(int status, String reason) {
if (websocket == null || isClosing) return;
websocket.close(status, reason);
isClosing = true;
}
@Override
protected void dispose() {
super.dispose();
@@ -88,16 +99,16 @@ final class Websocket extends Resource<Websocket> implements WebsocketClient {
var handle = new WebsocketHandle(environment, address, this, Map.of(), options);
environment.queueEvent(SUCCESS_EVENT, address, handle);
registerCleanable(handle);
GlobalCleaner.register(handle, () -> close(1001, ""));
checkClosed();
}
void failure(String message) {
private void handshakeFailure(String message) {
if (tryClose()) environment.queueEvent(FAILURE_EVENT, address, message);
}
void close(int status, String reason) {
private void serverClosed(int status, String reason) {
if (!tryClose()) return;
environment.queueEvent(CLOSE_EVENT, address, Strings.isNullOrEmpty(reason) ? null : reason, status < 0 ? null : status);