diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java index 3318d4d..4bdd882 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannel.java @@ -22,6 +22,7 @@ import java.nio.ByteBuffer; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicBoolean; public abstract class NetherNetChannel extends AbstractChannel { private static final InternalLogger log = InternalLoggerFactory.getInstance(NetherNetChannel.class); @@ -35,10 +36,38 @@ public abstract class NetherNetChannel extends AbstractChannel { protected RTCDataChannel reliableChannel; protected RTCDataChannel unreliableChannel; + // Mirrors the reliable channel's OPEN state, updated from the data channel + // observer. isActive() is called on every write and formerly crossed JNI + // into a blocking native call for it; the cached flag makes it free. + private volatile boolean reliableOpen; + + // Reusable outbound fragment buffer. Confined to the event loop (doWrite), + // and safe to reuse across sends because the native layer copies the data + // out before sendAsync returns. + private ByteBuffer outboundScratch; + protected final Queue pendingWrites = new ConcurrentLinkedQueue<>(); + private final AtomicBoolean channelActiveFired = new AtomicBoolean(); protected volatile boolean open = true; + // Reassembly state for inbound fragmented messages. assemblyBuf is written on + // the WebRTC callback thread (the data channel observer) and released on the + // event loop (doClose), so every access is guarded by assemblyLock together + // with the assemblyClosed flag, which orders the release strictly after the + // observer is detached and prevents a use-after-free. + private final Object assemblyLock = new Object(); + private ByteBuf assemblyBuf; + private boolean assemblyClosed; + private int currentSegmentCount = -1; + + // Maximum outbound SCTP message size in bytes. writeInternal must never + // fragment beyond what the remote advertised it can receive (the + // a=max-message-size attribute in its SDP); subclasses set this from the + // negotiated description. Defaults to the conservative MAX_SCTP_MESSAGE_SIZE + // until negotiation supplies the real value. + protected volatile int maxOutboundMessageSize = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE; + protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketAddress local) { super(parent); this.remoteAddress = remote; @@ -47,12 +76,25 @@ protected NetherNetChannel(Channel parent, InetSocketAddress remote, InetSocketA public void setDataChannels(RTCDataChannel reliable, RTCDataChannel unreliable) { this.reliableChannel = reliable; + // Seed the cached state once; the observer keeps it current from here. + this.reliableOpen = reliable.getState() == RTCDataChannelState.OPEN; + // The unreliable channel is intentionally stored but never observed or + // written to: this is a reliable-only implementation, which gives the + // ordered, lossless stream a Bedrock translation proxy needs. If it is + // ever wired up it must stay single-message only, because an unreliable + // channel can drop or reorder and would strand a multi-fragment message + // mid-reassembly: observe it but never run reassembly on it (every + // inbound message must arrive complete, header 0), and on send drop + // anything too large to fit one message rather than fragmenting it. this.unreliableChannel = unreliable; - RTCDataChannelObserver observer = new RTCDataChannelObserver() { - private final ByteBuf assemblyBuf = config.getAllocator().buffer(); - private int currentSegmentCount = -1; + synchronized (assemblyLock) { + if (assemblyBuf == null && !assemblyClosed) { + assemblyBuf = config.getAllocator().buffer(); + } + } + RTCDataChannelObserver observer = new RTCDataChannelObserver() { @Override public void onBufferedAmountChange(long previousAmount) { } @@ -64,45 +106,53 @@ public void onStateChange() { @Override public void onMessage(RTCDataChannelBuffer buffer) { + // The ByteBuffer wraps native memory that is only valid for + // the duration of this callback, so every path below must copy + // exactly once into a netty buffer before handing off. ByteBuffer data = buffer.data; if (!data.hasRemaining()) return; int segments = data.get() & 0xFF; - if (currentSegmentCount == -1) { - currentSegmentCount = segments; - } else { - if (segments != currentSegmentCount - 1) { - assemblyBuf.clear(); - currentSegmentCount = -1; + synchronized (assemblyLock) { + if (assemblyClosed || assemblyBuf == null) { return; } - currentSegmentCount = segments; - } - - if (data.hasRemaining()) { - byte[] payload = new byte[data.remaining()]; - data.get(payload); - assemblyBuf.writeBytes(payload); - } - if (segments == 0) { - try { - if (assemblyBuf.isReadable()) { - ByteBuf packet = assemblyBuf.copy(); - assemblyBuf.skipBytes(assemblyBuf.readableBytes()); + // Fast path: complete single-segment message (the common + // case) skips the assembly buffer entirely. + if (currentSegmentCount == -1 && segments == 0) { + if (data.hasRemaining()) { + deliver(copyToBuf(data)); + } + return; + } - eventLoop().execute(() -> { - pipeline().fireChannelRead(packet); - pipeline().fireChannelReadComplete(); - }); + if (currentSegmentCount == -1) { + currentSegmentCount = segments; + } else { + if (segments != currentSegmentCount - 1) { + assemblyBuf.clear(); + currentSegmentCount = -1; + return; } - } catch (Exception e) { - log.error("Error processing packet", e); - } finally { - assemblyBuf.clear(); + currentSegmentCount = segments; + } + + if (data.hasRemaining()) { + assemblyBuf.writeBytes(data); + } + + if (segments == 0) { currentSegmentCount = -1; + if (assemblyBuf.isReadable()) { + // Hand the assembled buffer off to the pipeline and + // start a fresh one instead of copying it again. + ByteBuf packet = assemblyBuf; + assemblyBuf = config.getAllocator().buffer(); + deliver(packet); + } } } } @@ -115,17 +165,78 @@ public void onMessage(RTCDataChannelBuffer buffer) { } } + /** + * Copies the remaining bytes of the given callback-scoped buffer into a + * freshly allocated netty buffer. + */ + private ByteBuf copyToBuf(ByteBuffer data) { + ByteBuf packet = config.getAllocator().buffer(data.remaining()); + packet.writeBytes(data); + return packet; + } + + /** + * Fires a fully reassembled packet down the pipeline on the event loop. + * Releases the packet if the event loop rejects the task (shutdown race) + * so the buffer cannot leak. + */ + private void deliver(ByteBuf packet) { + try { + eventLoop().execute(() -> { + fireChannelActiveIfReady(); + pipeline().fireChannelRead(packet); + pipeline().fireChannelReadComplete(); + }); + } catch (Exception e) { + packet.release(); + } + } + + /** + * Sets the maximum outbound SCTP message size, in bytes, for this channel. + * Should be the {@code a=max-message-size} the remote peer advertised in its + * SDP. Values too small to leave room for the fragment header are ignored. + * + * @param size the negotiated maximum message size + */ + public void setMaxOutboundMessageSize(int size) { + if (size > 1) { + this.maxOutboundMessageSize = size; + } + } + private void onDataChannelStateChange() { + RTCDataChannel channel = this.reliableChannel; + if (channel == null) { + return; + } + // One native call per state transition; every other isActive() check + // reads the cached flag. + RTCDataChannelState state = channel.getState(); + reliableOpen = state == RTCDataChannelState.OPEN; + if (isActive()) { - if (!pendingWrites.isEmpty()) { - pipeline().fireChannelWritabilityChanged(); - unsafe().flush(); - } - } else if (reliableChannel.getState() == RTCDataChannelState.CLOSED) { + fireChannelActiveIfReady(); + } else if (state == RTCDataChannelState.CLOSED) { close(); } } + protected void fireChannelActiveIfReady() { + if (!isRegistered() || !isActive()) { + return; + } + + if (channelActiveFired.compareAndSet(false, true)) { + pipeline().fireChannelActive(); + } + + if (!pendingWrites.isEmpty()) { + pipeline().fireChannelWritabilityChanged(); + unsafe().flush(); + } + } + @Override protected void doWrite(ChannelOutboundBuffer in) throws Exception { if (!isActive()) { @@ -160,29 +271,46 @@ private void writeInternal(Object msg) { ByteBuf payload = (ByteBuf) msg; - ByteBuf framed = payload.retainedDuplicate(); - - int totalLength = framed.readableBytes(); - int maxPayload = NetherNetConstants.MAX_SCTP_MESSAGE_SIZE - 1; + int maxPayload = maxOutboundMessageSize - 1; + int totalLength = payload.readableBytes(); int segments = (totalLength / maxPayload); if (totalLength % maxPayload != 0) segments++; + // Each fragment carries a one-byte countdown header, so a message can + // span at most 256 fragments. The negotiated max-message-size keeps this + // ceiling far out of reach, but guard it anyway: without the check an + // oversized message would wrap the header byte and silently corrupt the + // stream instead of failing. + if (segments > 256) { + pipeline().fireExceptionCaught(new IllegalStateException( + "Outbound message of " + totalLength + " bytes exceeds the maximum " + + (256 * maxPayload) + " bytes addressable by the fragment header")); + return; + } + + ByteBuf framed = payload.retainedDuplicate(); try { + // One reusable direct buffer for every fragment of every message: + // sendAsync copies the bytes out before returning, so the scratch + // can be refilled immediately, and unlike the blocking send it + // never stalls this event loop on the native network thread. + ByteBuffer chunk = outboundScratch(1 + maxPayload); + int offset = 0; for (int i = 0; i < segments; i++) { int remaining = segments - 1 - i; int chunkSize = Math.min(maxPayload, framed.readableBytes() - offset); - ByteBuffer chunk = ByteBuffer.allocateDirect(1 + chunkSize); + chunk.clear(); + chunk.limit(1 + chunkSize); chunk.put((byte) remaining); framed.getBytes(offset, chunk); - chunk.position(chunk.limit()); chunk.flip(); - reliableChannel.send(new RTCDataChannelBuffer(chunk, true)); + reliableChannel.sendAsync(new RTCDataChannelBuffer(chunk, true)); offset += chunkSize; } } catch (Exception e) { @@ -192,8 +320,25 @@ private void writeInternal(Object msg) { } } + /** + * Returns the reusable outbound fragment buffer, growing it when the + * negotiated max message size demands more. Only used from doWrite on the + * event loop. + */ + private ByteBuffer outboundScratch(int capacity) { + ByteBuffer scratch = this.outboundScratch; + if (scratch == null || scratch.capacity() < capacity) { + scratch = ByteBuffer.allocateDirect(capacity); + this.outboundScratch = scratch; + } + return scratch; + } + @Override protected void doRegister() throws Exception { + if (isActive()) { + channelActiveFired.set(true); + } } @Override @@ -213,6 +358,8 @@ protected void doDisconnect() throws Exception { @Override protected void doClose() throws Exception { this.open = false; + this.reliableOpen = false; + this.outboundScratch = null; if (reliableChannel != null) { reliableChannel.unregisterObserver(); @@ -226,6 +373,18 @@ protected void doClose() throws Exception { peerConnection.close(); } + // Release the reassembly buffer now that the observer is unregistered. + // Guarded by assemblyLock with assemblyClosed so it cannot race, or be + // re-allocated by, an in-flight onMessage on the WebRTC callback thread. + synchronized (assemblyLock) { + assemblyClosed = true; + if (assemblyBuf != null) { + assemblyBuf.release(); + assemblyBuf = null; + } + currentSegmentCount = -1; + } + Object msg; while ((msg = pendingWrites.poll()) != null) { ReferenceCountUtil.release(msg); @@ -263,11 +422,13 @@ public boolean isOpen() { @Override public boolean isActive() { - return isOpen() && this.reliableChannel != null && this.reliableChannel.getState() == RTCDataChannelState.OPEN; + // Hot path: called around every write. Reads the observer-maintained + // flag instead of a blocking JNI call into the native network thread. + return isOpen() && this.reliableChannel != null && this.reliableOpen; } @Override public ChannelMetadata metadata() { return METADATA; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java index 82c3cc0..8860b54 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetChannelFactory.java @@ -6,6 +6,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelFactory; +import java.util.List; import java.util.function.Supplier; public class NetherNetChannelFactory implements ChannelFactory { @@ -32,6 +33,23 @@ public static ChannelFactory server(PeerConnectionFactor return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factory, signaling)); } + /** + * Creates a NetherNet Server Channel Factory backed by a pool of + * PeerConnectionFactory instances. Each native factory carries exactly one + * network, worker, and signaling thread shared by every peer connection it + * creates, so a pool multiplies the threads available to the data plane: + * incoming connections are assigned round robin across the pool. + * + * @param factories The PeerConnectionFactory pool. Must not be empty. The + * resulting server channel takes ownership and disposes + * every factory on close. + * @param signaling The NetherNetServerSignaling instance for signaling. + * @return A ChannelFactory for NetherNetServerChannel. + */ + public static ChannelFactory server(List factories, NetherNetServerSignaling signaling) { + return new NetherNetChannelFactory<>(() -> new NetherNetServerChannel(factories, signaling)); + } + /** * Creates a NetherNet Client Channel Factory. * diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java index 8749323..6c5c05a 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetClientChannel.java @@ -18,6 +18,7 @@ import dev.kastle.webrtc.RTCIceCandidate; import dev.kastle.webrtc.RTCIceServer; import dev.kastle.webrtc.RTCOfferOptions; +import dev.kastle.webrtc.RTCPeerConnection; import dev.kastle.webrtc.RTCPeerConnectionState; import dev.kastle.webrtc.RTCSdpType; import dev.kastle.webrtc.RTCSessionDescription; @@ -32,6 +33,7 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; import java.util.List; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -53,6 +55,20 @@ public class NetherNetClientChannel extends NetherNetChannel { private int retryCount = 0; + // Monotonic attempt marker, bumped on every handshake retry. Async engine + // callbacks belonging to a previous attempt (offer creation, description + // observers, data channel state changes) capture their generation and + // bail once a retry has moved past them, so a delayed stale callback can + // no longer mutate the replacement attempt's state. + private volatile int attemptGeneration; + + // Event loop confined. The synchronous native addIceCandidate rejects + // candidates applied while the peer connection has no remote description + // yet, so candidates arriving before the CONNECT_RESPONSE answer has been + // applied are buffered and drained once it succeeds. + private boolean remoteDescriptionSet; + private List pendingRemoteCandidates = new ArrayList<>(); + /** * Creates a NetherNetClientChannel with a new PeerConnectionFactory. * @@ -189,6 +205,7 @@ private void resetAndRetryHandshake() { } retryCount++; + attemptGeneration++; if (peerConnection != null) { peerConnection.close(); @@ -197,6 +214,8 @@ private void resetAndRetryHandshake() { signaling.removeSignalHandler(this.connectionId); this.cycleConnectionId(); + remoteDescriptionSet = false; + pendingRemoteCandidates = new ArrayList<>(); startHandshake(); } @@ -215,17 +234,27 @@ private void initWebRTC(List iceServers) { } } - peerConnection = factory.createPeerConnection(rtcConfig, new PeerConnectionObserver() { + final int gen = attemptGeneration; + final long attemptConnectionId = this.connectionId; + + RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, new PeerConnectionObserver() { @Override public void onIceCandidate(RTCIceCandidate candidate) { + if (gen != attemptGeneration) { + return; + } try { signaling.sendSignal( - targetNetworkId, - NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) + targetNetworkId, + NetherNetConstants.buildSignalCandidateAdd(attemptConnectionId, candidate.sdp) ); } catch (Exception e) { log.error("Failed to send ICE candidate", e); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } } @@ -234,7 +263,11 @@ public void onConnectionChange(RTCPeerConnectionState state) { if (state == RTCPeerConnectionState.FAILED) { // Fast fail trigger: retry immediately instead of waiting for timeout log.warn("PeerConnection entered FAILED state, resetting and retrying handshake."); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } else { log.trace("PeerConnection state changed to {}", state); } @@ -242,27 +275,36 @@ public void onConnectionChange(RTCPeerConnectionState state) { @Override public void onDataChannel(RTCDataChannel dataChannel) { } }); + this.peerConnection = pc; - setupDataChannels(); + setupDataChannels(pc, gen); } private void createAndSendOffer() { - if (peerConnection == null) return; - peerConnection.createOffer(new RTCOfferOptions(), new CreateSessionDescriptionObserver() { + final RTCPeerConnection pc = this.peerConnection; + final int gen = attemptGeneration; + final long attemptConnectionId = this.connectionId; + if (pc == null) return; + pc.createOffer(new RTCOfferOptions(), new CreateSessionDescriptionObserver() { @Override public void onSuccess(RTCSessionDescription description) { - if (peerConnection == null) return; - peerConnection.setLocalDescription(description, new SetSessionDescriptionObserver() { + if (gen != attemptGeneration) return; + pc.setLocalDescription(description, new SetSessionDescriptionObserver() { @Override public void onSuccess() { + if (gen != attemptGeneration) return; try { signaling.sendSignal( - targetNetworkId, - NetherNetConstants.buildSignalConnectRequest(connectionId, description.sdp) + targetNetworkId, + NetherNetConstants.buildSignalConnectRequest(attemptConnectionId, description.sdp) ); } catch (Exception e) { log.error("Failed to send Connect Request", e); - eventLoop().execute(() -> resetAndRetryHandshake()); + eventLoop().execute(() -> { + if (gen == attemptGeneration) { + resetAndRetryHandshake(); + } + }); } } @Override public void onFailure(String error) { /* Retry handled by timeout */ } @@ -280,8 +322,9 @@ private void handleSignal(String signal) { String data = parts.length > 2 ? parts[2] : ""; // Verify this signal belongs to the current attempt + final long signalId; try { - long signalId = Long.parseUnsignedLong(idStr); + signalId = Long.parseUnsignedLong(idStr); if (signalId != this.connectionId) { log.debug("Ignored stale signal for ID {}", idStr); return; @@ -291,18 +334,47 @@ private void handleSignal(String signal) { } eventLoop().execute(() -> { + // Re-validate on the event loop: a retry may have cycled the + // connection id between the check above (signaling thread) and + // this task running. Inside the task the id, generation, and + // peer connection mutate together, so passing this check means + // everything read below belongs to the current attempt. + if (signalId != this.connectionId) { + log.debug("Ignored stale signal for ID {} (attempt retried)", idStr); + return; + } if (peerConnection == null) return; if (!isOpen() || handshakeComplete) return; switch (type) { case NetherNetConstants.RTC_NEGOTIATION_CONNECT_RESPONSE -> { + // Fragment outbound data no larger than the remote advertised + // it can receive (a=max-message-size in its answer). + setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(data, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); + final int gen = attemptGeneration; peerConnection.setRemoteDescription(new RTCSessionDescription(RTCSdpType.ANSWER, data), new SetSessionDescriptionObserver() { - @Override public void onSuccess() {} + @Override public void onSuccess() { + // Apply candidates that arrived before the answer + // finished applying, in arrival order. + eventLoop().execute(() -> { + if (gen != attemptGeneration) return; + remoteDescriptionSet = true; + List drained = pendingRemoteCandidates; + pendingRemoteCandidates = new ArrayList<>(); + for (String candidate : drained) { + applyRemoteCandidate(candidate); + } + }); + } @Override public void onFailure(String e) { /* Retry handled by timeout */ } }); } case NetherNetConstants.RTC_NEGOTIATION_CANDIDATE_ADD -> { - peerConnection.addIceCandidate(new RTCIceCandidate("0", 0, data)); + if (remoteDescriptionSet) { + applyRemoteCandidate(data); + } else { + pendingRemoteCandidates.add(data); + } } case NetherNetConstants.RTC_NEGOTIATION_CONNECT_ERROR -> { log.error("Received SIGNAL_CONNECT_ERROR for {}.", Long.toUnsignedString(this.connectionId)); @@ -318,7 +390,7 @@ private void handleSignal(String signal) { }); } - private void setupDataChannels() { + private void setupDataChannels(RTCPeerConnection pc, int gen) { RTCDataChannelInit reliableInit = new RTCDataChannelInit(); reliableInit.ordered = true; reliableInit.protocol = NetherNetConstants.RELIABLE_CHANNEL_LABEL; @@ -327,14 +399,17 @@ private void setupDataChannels() { unreliableInit.ordered = false; unreliableInit.maxRetransmits = 0; - RTCDataChannel reliable = peerConnection.createDataChannel(NetherNetConstants.RELIABLE_CHANNEL_LABEL, reliableInit); - RTCDataChannel unreliable = peerConnection.createDataChannel(NetherNetConstants.UNRELIABLE_CHANNEL_LABEL, unreliableInit); + RTCDataChannel reliable = pc.createDataChannel(NetherNetConstants.RELIABLE_CHANNEL_LABEL, reliableInit); + RTCDataChannel unreliable = pc.createDataChannel(NetherNetConstants.UNRELIABLE_CHANNEL_LABEL, unreliableInit); reliable.registerObserver(new RTCDataChannelObserver() { @Override public void onStateChange() { if (reliable.getState() == RTCDataChannelState.OPEN) { eventLoop().execute(() -> { + if (gen != attemptGeneration) { + return; + } if (!handshakeComplete) { log.debug("NetherNet Connection Established!"); handshakeComplete = true; @@ -348,7 +423,7 @@ public void onStateChange() { if (connectPromise != null && !connectPromise.isDone()) { connectPromise.trySuccess(); } - pipeline().fireChannelActive(); + fireChannelActiveIfReady(); } }); } @@ -360,8 +435,25 @@ public void onStateChange() { }); } + /** + * Applies a remote ICE candidate to the current peer connection. Runs on + * the event loop, either directly from a CANDIDATE_ADD signal or from the + * drain after the remote description applies. + */ + private void applyRemoteCandidate(String candidateSdp) { + RTCPeerConnection pc = this.peerConnection; + if (pc == null) { + return; + } + try { + pc.addIceCandidate(new RTCIceCandidate("0", 0, candidateSdp)); + } catch (Exception e) { + log.debug("Failed to apply ICE candidate (connection likely closed): {}", e.toString()); + } + } + private long cycleConnectionId() { this.connectionId = ThreadLocalRandom.current().nextLong(1, Long.MAX_VALUE); return this.connectionId; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java index dcc6a0f..3df9546 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetConstants.java @@ -177,4 +177,34 @@ public static String buildSignalConnectResponse(long connectionId, String sdp) { public static String buildSignalCandidateAdd(long connectionId, String candidateSdp) { return RTC_NEGOTIATION_CANDIDATE_ADD + " " + Long.toUnsignedString(connectionId) + " " + candidateSdp; } + + /** + * Parses the {@code a=max-message-size} attribute from an SDP description. + * This is the maximum SCTP user message size, in bytes, that the describing + * endpoint is willing to receive, so outbound fragmentation must never + * exceed the value advertised by the remote peer. + * + * @param sdp the SDP description to scan, may be null + * @param fallback the value to return when the attribute is absent or invalid + * @return the advertised maximum message size, or {@code fallback} if not found + */ + public static int parseMaxMessageSize(String sdp, int fallback) { + if (sdp == null) { + return fallback; + } + for (String line : sdp.split("\\r?\\n")) { + String trimmed = line.trim(); + if (trimmed.startsWith("a=max-message-size:")) { + try { + int value = Integer.parseInt(trimmed.substring("a=max-message-size:".length()).trim()); + if (value > 1) { + return value; + } + } catch (NumberFormatException ignored) { + // Fall through to the fallback for a malformed attribute. + } + } + } + return fallback; + } } \ No newline at end of file diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java index 1eef7f7..f66e09b 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/NetherNetServerChannel.java @@ -28,7 +28,9 @@ import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.ArrayList; import java.util.List; +import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; public class NetherNetServerChannel extends AbstractServerChannel { @@ -36,15 +38,19 @@ public class NetherNetServerChannel extends AbstractServerChannel { private static final ChannelMetadata METADATA = new ChannelMetadata(false, 16); private final DefaultNetherServerChannelConfig config; - private final PeerConnectionFactory factory; + private final List factories; private final NetherNetServerSignaling signaling; - + + // Round robin cursor into factories. Only touched from this channel's + // event loop (establishConnection), so no synchronization is needed. + private int nextFactory; + private InetSocketAddress localAddress; private volatile boolean open = true; /** * Creates a NetherNetServerChannel with a new PeerConnectionFactory. - * + * * @param signaling The NetherNetServerSignaling instance for signaling. */ public NetherNetServerChannel(NetherNetServerSignaling signaling) { @@ -53,12 +59,30 @@ public NetherNetServerChannel(NetherNetServerSignaling signaling) { /** * Creates a NetherNetServerChannel. - * + * * @param factory The PeerConnectionFactory to use for creating peer connections. Should be reused where possible. * @param signaling The NetherNetServerSignaling instance for signaling. */ public NetherNetServerChannel(PeerConnectionFactory factory, NetherNetServerSignaling signaling) { - this.factory = factory; + this(List.of(factory), signaling); + } + + /** + * Creates a NetherNetServerChannel backed by a pool of PeerConnectionFactory + * instances. Each native factory runs one network, worker, and signaling + * thread shared by all its peer connections; connections are assigned to + * factories round robin, spreading the DTLS and SCTP load of many players + * across the pool instead of serializing it on a single network thread. + * + * @param factories The PeerConnectionFactory pool, at least one. This + * channel takes ownership and disposes each on close. + * @param signaling The NetherNetServerSignaling instance for signaling. + */ + public NetherNetServerChannel(List factories, NetherNetServerSignaling signaling) { + if (factories.isEmpty()) { + throw new IllegalArgumentException("factories must not be empty"); + } + this.factories = List.copyOf(factories); this.signaling = signaling; this.config = new DefaultNetherServerChannelConfig(this); } @@ -67,7 +91,7 @@ public NetherNetServerChannel(PeerConnectionFactory factory, NetherNetServerSign protected void doBind(SocketAddress localAddress) throws Exception { if (!(localAddress instanceof InetSocketAddress)) throw new IllegalArgumentException("Unsupported address type"); this.localAddress = (InetSocketAddress) localAddress; - + this.signaling.setNewConnectionHandler((connectionId, remoteNetworkId, offerSdp) -> { acceptConnection(connectionId, offerSdp, remoteNetworkId); }); @@ -75,44 +99,151 @@ protected void doBind(SocketAddress localAddress) throws Exception { this.signaling.bind(localAddress); } + /** + * Accepts an incoming connection offer. Runs on the signaling I/O thread: + * only the signal handler registration happens here (a cheap map put, so + * candidates arriving right behind the offer are never dropped), then all + * WebRTC work hops onto this server channel's event loop. Native peer + * connection calls block on the WebRTC signaling thread and must never + * stall the signaling socket's thread, whose keepalives are what hold the + * connection to the signaling service open. + */ public void acceptConnection(long connectionId, String offerSdp, String remoteNetworkId) { - RTCConfiguration rtcConfig = new RTCConfiguration(); - rtcConfig.portAllocatorConfig = this.config.getOption(NetherChannelOption.NETHER_PORT_ALLOCATOR_CONFIG); - rtcConfig.bundlePolicy = RTCBundlePolicy.MAX_BUNDLE; - - // Inject ICE servers if the signaling implementation supports it - List iceServers = this.signaling.getIceServers(); - if (iceServers != null && !iceServers.isEmpty()) { - log.trace("Injecting {} ICE Servers into PeerConnection for {}", iceServers.size(), Long.toUnsignedString(connectionId)); - for (IceServerInfo info : iceServers) { - RTCIceServer iceServer = new RTCIceServer(); - iceServer.urls = info.urls(); - iceServer.username = info.username(); - iceServer.password = info.password(); - rtcConfig.iceServers.add(iceServer); + PendingConnection pending = new PendingConnection(connectionId); + signaling.setSignalHandler(connectionId, signal -> eventLoop().execute(() -> pending.handleSignal(signal))); + eventLoop().execute(() -> establishConnection(pending, connectionId, offerSdp, remoteNetworkId)); + } + + /** + * Builds the peer connection for an accepted offer. Runs on this server + * channel's event loop; incoming signals for the connection hop onto the + * same loop, so everything here is single-threaded and ordered. + */ + private void establishConnection(PendingConnection pending, long connectionId, String offerSdp, String remoteNetworkId) { + try { + RTCConfiguration rtcConfig = new RTCConfiguration(); + rtcConfig.portAllocatorConfig = this.config.getOption(NetherChannelOption.NETHER_PORT_ALLOCATOR_CONFIG); + rtcConfig.bundlePolicy = RTCBundlePolicy.MAX_BUNDLE; + + // Inject ICE servers if the signaling implementation supports it + List iceServers = this.signaling.getIceServers(); + if (iceServers != null && !iceServers.isEmpty()) { + log.trace("Injecting {} ICE Servers into PeerConnection for {}", iceServers.size(), Long.toUnsignedString(connectionId)); + for (IceServerInfo info : iceServers) { + RTCIceServer iceServer = new RTCIceServer(); + iceServer.urls = info.urls(); + iceServer.username = info.username(); + iceServer.password = info.password(); + rtcConfig.iceServers.add(iceServer); + } } + + ServerPeerConnectionObserver observer = new ServerPeerConnectionObserver(connectionId, remoteNetworkId); + PeerConnectionFactory factory = factories.get(nextFactory); + nextFactory = (nextFactory + 1) % factories.size(); + RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, observer); + + NetherNetChildChannel child = new NetherNetChildChannel(this, pc, generatePlaceholderAddress(), localAddress); + // Fragment outbound data no larger than the client advertised it can + // receive (a=max-message-size in its offer), falling back to the + // conservative default when the client does not advertise one. + child.setMaxOutboundMessageSize(NetherNetConstants.parseMaxMessageSize(offerSdp, NetherNetConstants.MAX_SCTP_MESSAGE_SIZE)); + observer.setChildChannel(child); + + child.closeFuture().addListener(future -> signaling.removeSignalHandler(connectionId)); + + int handshakeTimeoutSeconds = this.config.getOption(NetherChannelOption.NETHER_SERVER_RTC_HANDSHAKE_TIMEOUT_SECONDS); + ScheduledFuture timeoutTask = eventLoop().schedule(() -> { + if (!child.isActive()) { + log.warn("Connection {} timed out during handshake ({}s)", Long.toUnsignedString(connectionId), handshakeTimeoutSeconds); + child.close(); + pc.close(); + } + }, handshakeTimeoutSeconds, TimeUnit.SECONDS); + observer.setHandshakeTimeout(timeoutTask); + + // Handle Offer + pc.setRemoteDescription(new RTCSessionDescription(RTCSdpType.OFFER, offerSdp), new SetSessionDescriptionObserver() { + @Override + public void onSuccess() { + log.trace("Remote description set for {}", Long.toUnsignedString(connectionId)); + pc.createAnswer(new RTCAnswerOptions(), new CreateSessionDescriptionObserver() { + @Override + public void onSuccess(RTCSessionDescription description) { + pc.setLocalDescription(description, new SetSessionDescriptionObserver() { + @Override + public void onSuccess() { + log.trace("Sending Answer SDP for {}", Long.toUnsignedString(connectionId)); + try { + signaling.sendSignal( + remoteNetworkId, + NetherNetConstants.buildSignalConnectResponse(connectionId, description.sdp) + ); + } catch (Exception e) { + // Signaling dropped mid-handshake; the client cannot + // receive the answer, so let the handshake timeout + // reap this connection. + log.warn("Failed to send answer for {} (signaling unavailable): {}", + Long.toUnsignedString(connectionId), e.getMessage()); + return; + } + pipeline().fireChannelRead(child); + } + @Override public void onFailure(String error) { log.error("SetLocalDesc failed: {}", error); } + }); + } + @Override public void onFailure(String error) { log.error("CreateAnswer failed: {}", error); } + }); + } + @Override public void onFailure(String error) { log.error("SetRemoteDesc failed: {}", error); } + }); + + // Publish the peer connection to the signal handler and drain any + // candidates that arrived while it was being created. + pending.attach(pc, child); + } catch (Exception e) { + log.error("Failed to establish connection {}: {}", Long.toUnsignedString(connectionId), e.getMessage(), e); + signaling.removeSignalHandler(connectionId); } + } - ServerPeerConnectionObserver observer = new ServerPeerConnectionObserver(connectionId, remoteNetworkId); - RTCPeerConnection pc = factory.createPeerConnection(rtcConfig, observer); + /** + * Per-connection signal state. All methods run on this server channel's + * event loop, so no synchronization is needed. Signals that arrive between + * the offer and the peer connection becoming available are queued and + * drained by attach, preserving arrival order. + */ + private final class PendingConnection { + private final long connectionId; + private RTCPeerConnection pc; + private NetherNetChildChannel child; + private List queued = new ArrayList<>(); - NetherNetChildChannel child = new NetherNetChildChannel(this, pc, new InetSocketAddress(0), localAddress); - observer.setChildChannel(child); + PendingConnection(long connectionId) { + this.connectionId = connectionId; + } - child.closeFuture().addListener(future -> signaling.removeSignalHandler(connectionId)); + void attach(RTCPeerConnection pc, NetherNetChildChannel child) { + this.pc = pc; + this.child = child; + List pendingSignals = this.queued; + this.queued = null; + for (String signal : pendingSignals) { + apply(signal); + } + } - int handshakeTimeoutSeconds = this.config.getOption(NetherChannelOption.NETHER_SERVER_RTC_HANDSHAKE_TIMEOUT_SECONDS); - ScheduledFuture timeoutTask = eventLoop().schedule(() -> { - if (!child.isActive()) { - log.warn("Connection {} timed out during handshake ({}s)", Long.toUnsignedString(connectionId), handshakeTimeoutSeconds); - child.close(); - pc.close(); + void handleSignal(String signal) { + if (pc == null) { + if (queued != null) { + queued.add(signal); + } + return; } - }, handshakeTimeoutSeconds, TimeUnit.SECONDS); - observer.setHandshakeTimeout(timeoutTask); - - // Register Signal Handler - signaling.setSignalHandler(connectionId, (signal) -> { + apply(signal); + } + + private void apply(String signal) { String[] parts = signal.split(" ", 3); if (parts.length < 3) return; String type = parts[0]; @@ -132,34 +263,7 @@ public void acceptConnection(long connectionId, String offerSdp, String remoteNe child.close(); } } - }); - - // Handle Offer - pc.setRemoteDescription(new RTCSessionDescription(RTCSdpType.OFFER, offerSdp), new SetSessionDescriptionObserver() { - @Override - public void onSuccess() { - log.trace("Remote description set for {}", Long.toUnsignedString(connectionId)); - pc.createAnswer(new RTCAnswerOptions(), new CreateSessionDescriptionObserver() { - @Override - public void onSuccess(RTCSessionDescription description) { - pc.setLocalDescription(description, new SetSessionDescriptionObserver() { - @Override - public void onSuccess() { - log.trace("Sending Answer SDP for {}", Long.toUnsignedString(connectionId)); - signaling.sendSignal( - remoteNetworkId, - NetherNetConstants.buildSignalConnectResponse(connectionId, description.sdp) - ); - pipeline().fireChannelRead(child); - } - @Override public void onFailure(String error) { log.error("SetLocalDesc failed: {}", error); } - }); - } - @Override public void onFailure(String error) { log.error("CreateAnswer failed: {}", error); } - }); - } - @Override public void onFailure(String error) { log.error("SetRemoteDesc failed: {}", error); } - }); + } } /** @@ -169,12 +273,17 @@ private class ServerPeerConnectionObserver implements PeerConnectionObserver { private final long connectionId; private final String remoteNetworkId; private NetherNetChildChannel child; - + private RTCDataChannel reliable; private RTCDataChannel unreliable; + private boolean dataChannelsSet; private ScheduledFuture handshakeTimeout; + // Real remote address from ICE nomination. Buffered here when the + // callback fires before the child channel is attached. + private volatile InetSocketAddress pendingRemoteAddress; + public ServerPeerConnectionObserver(long connectionId, String remoteNetworkId) { this.connectionId = connectionId; this.remoteNetworkId = remoteNetworkId; @@ -186,19 +295,57 @@ public void setHandshakeTimeout(ScheduledFuture handshakeTimeout) { public void setChildChannel(NetherNetChildChannel child) { this.child = child; + InetSocketAddress pending = this.pendingRemoteAddress; + if (pending != null) { + child.remoteAddress = pending; + } checkDataChannels(); } + @Override + public void onSelectedCandidatePairChanged(String remoteAddress, int remotePort, String candidateType) { + // ICE nominated a pair. This fires before DTLS and the data + // channels open, so the channel carries its real remote address + // before it activates and anything downstream reads it. Until + // then the channel holds its unique random placeholder, which + // also covers the case of this callback never firing. For a + // relayed connection the address is the TURN relay, which is the + // peer actually connected to us and the intended value. A later + // re-nomination simply overwrites. + try { + InetSocketAddress resolved = new InetSocketAddress(remoteAddress, remotePort); + NetherNetChildChannel target = this.child; + if (target != null) { + target.remoteAddress = resolved; + } else { + this.pendingRemoteAddress = resolved; + } + log.debug("Resolved remote address for {}: {}:{} (type: {})", + Long.toUnsignedString(this.connectionId), remoteAddress, remotePort, candidateType); + } catch (Exception e) { + log.debug("Failed to set remote address for {}: {}", + Long.toUnsignedString(this.connectionId), e.getMessage()); + } + } + @Override public void onIceCandidate(RTCIceCandidate candidate) { if (log.isTraceEnabled()) { - log.trace("Generated ICE Candidate for {}: {} (Type: {})", + log.trace("Generated ICE Candidate for {}: {} (Type: {})", Long.toUnsignedString(this.connectionId), candidate.sdp, extractCandidateType(candidate.sdp)); } - signaling.sendSignal( - remoteNetworkId, - NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) - ); + try { + signaling.sendSignal( + remoteNetworkId, + NetherNetConstants.buildSignalCandidateAdd(connectionId, candidate.sdp) + ); + } catch (Exception e) { + // Signaling dropped mid-handshake. Established connections don't + // signal candidates, so only this in-flight handshake is affected; + // the handshake timeout cleans it up if it cannot complete. + log.debug("Failed to signal ICE candidate for {} (signaling unavailable): {}", + Long.toUnsignedString(this.connectionId), e.getMessage()); + } } private String extractCandidateType(String sdp) { @@ -226,40 +373,71 @@ public void onConnectionChange(RTCPeerConnectionState state) { public void onDataChannel(RTCDataChannel dataChannel) { String label = dataChannel.getLabel(); log.debug("Received Data Channel: {}", label); - + if (NetherNetConstants.RELIABLE_CHANNEL_LABEL.equals(label)) { this.reliable = dataChannel; } else if (NetherNetConstants.UNRELIABLE_CHANNEL_LABEL.equals(label)) { this.unreliable = dataChannel; } - + checkDataChannels(); } - + private void checkDataChannels() { - if (child != null && reliable != null && unreliable != null) { + if (child != null && reliable != null && unreliable != null && !dataChannelsSet) { + dataChannelsSet = true; if (handshakeTimeout != null) { handshakeTimeout.cancel(false); } log.debug("Data Channels established for {}", Long.toUnsignedString(this.connectionId)); child.setDataChannels(reliable, unreliable); - - if (child.pipeline() != null) { - child.pipeline().fireChannelActive(); - } + child.fireChannelActiveIfReady(); } } + + // + // private void activateOnce(NetherNetChildChannel child, java.util.concurrent.atomic.AtomicBoolean activated) { + // if (!activated.compareAndSet(false, true)) return; + // child.eventLoop().execute(() -> { + // if (child.isOpen() && child.pipeline() != null) { + // child.pipeline().fireChannelActive(); + // } + // }); + // } + } + + /** + * Generates a unique placeholder address in the 10.x.x.x range for a new + * Nethernet connection. The 10.0.0.0/8 range is private (RFC 1918) and + * will not collide with real public client addresses. + */ + private static InetSocketAddress generatePlaceholderAddress() { + ThreadLocalRandom r = ThreadLocalRandom.current(); + String ip = "10." + (r.nextInt(1, 256)) + "." + (r.nextInt(256)) + "." + (r.nextInt(1, 256)); + return new InetSocketAddress(ip, 0); } @Override protected void doClose() throws Exception { this.open = false; - + try { signaling.close(); } finally { - factory.dispose(); + Exception failure = null; + for (PeerConnectionFactory factory : factories) { + try { + factory.dispose(); + } catch (Exception e) { + // Keep disposing the rest; rethrow the first failure after. + if (failure == null) failure = e; + log.warn("Failed to dispose PeerConnectionFactory: {}", e.getMessage()); + } + } + if (failure != null) { + throw failure; + } } } @@ -275,24 +453,24 @@ protected SocketAddress localAddress0() { @Override protected boolean isCompatible(EventLoop loop) { - return true; + return true; } @Override public ChannelConfig config() { return config; } - - @Override - public boolean isOpen() { + + @Override + public boolean isOpen() { return this.open; } - - @Override - public boolean isActive() { + + @Override + public boolean isActive() { return isOpen() && localAddress0() != null; } - - @Override - public ChannelMetadata metadata() { - return METADATA; + + @Override + public ChannelMetadata metadata() { + return METADATA; } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java index dde7b09..6e46e08 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/AbstractNetherNetXboxSignaling.java @@ -8,6 +8,7 @@ import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.channel.Channel; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelPipeline; @@ -17,6 +18,7 @@ import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.HttpClientCodec; import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.websocketx.PingWebSocketFrame; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory; @@ -25,6 +27,8 @@ import io.netty.handler.codec.http.websocketx.WebSocketVersion; import io.netty.handler.ssl.SslContext; import io.netty.handler.ssl.SslContextBuilder; +import io.netty.util.AttributeKey; +import io.netty.util.concurrent.ScheduledFuture; import io.netty.util.internal.logging.InternalLogger; import io.netty.util.internal.logging.InternalLoggerFactory; @@ -38,20 +42,37 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; -public abstract class AbstractNetherNetXboxSignaling extends SimpleChannelInboundHandler +public abstract class AbstractNetherNetXboxSignaling extends SimpleChannelInboundHandler implements NetherNetClientSignaling, NetherNetServerSignaling { - + + /** Time to wait for the connect plus credential exchange before giving up. */ + private static final long CONNECT_TIMEOUT_SECONDS = 20; + /** Interval for WebSocket protocol-level pings. The RFC obliges the server to + * answer with a pong, so these guarantee inbound traffic on a healthy socket + * and make {@link #isChannelAlive(long)} reliable even on idle servers. */ + private static final long WS_PING_INTERVAL_SECONDS = 15; + + /** Recurring tasks tied to one WebSocket channel's lifetime, cancelled in + * channelInactive. Without this, every reconnect would leak the previous + * channel's ping loops on the shared event loop. */ + private static final AttributeKey>> CHANNEL_TASKS = + AttributeKey.valueOf("nethernet-signaling-channel-tasks"); + protected final InternalLogger log = InternalLoggerFactory.getInstance(getClass()); - protected final String xboxToken; + protected volatile String xboxToken; protected final String localNetworkId; protected final URI uri; protected final EventLoopGroup eventLoopGroup; - - protected Channel channel; + + protected volatile Channel channel; protected CompletableFuture> connectFuture; protected volatile List iceServers = new ArrayList<>(); + protected volatile long lastMessageReceivedAt; + private volatile boolean closed; protected final Map handlers = new ConcurrentHashMap<>(); protected NetherNetServerSignaling.NewConnectionHandler newConnectionHandler; @@ -76,28 +97,60 @@ public synchronized CompletableFuture> connect(SocketAddress @Override public void bind(SocketAddress localAddress) throws ConnectException { - try { - connectInternal().join(); - } catch (Exception e) { - Throwable cause = e.getCause() != null ? e.getCause() : e; - close(); - if (cause instanceof ConnectException) throw (ConnectException) cause; - ConnectException ce = new ConnectException("Failed to connect to Xbox Signaling: " + cause.getMessage()); - ce.initCause(cause); - throw ce; + joinConnect(connectInternal()); + } + + /** + * Reconnects the WebSocket in place using a fresh authorization token. The + * signaling instance, its registered handlers, and everything built on top + * of it (server channel, peer connection factory, live peer connections) + * are untouched; only the socket to the signaling service is replaced. + * + * Must not be called from this signaling instance's own event loop; it + * blocks until the new socket has completed its credential exchange. + * + * @param freshToken the authorization token to connect with + * @throws ConnectException if the reconnect fails; the instance remains + * reconnectable and the call may be retried + */ + public void reconnect(String freshToken) throws ConnectException { + CompletableFuture> future; + synchronized (this) { + if (closed) { + throw new ConnectException("Signaling has been closed"); + } + this.xboxToken = freshToken; + Channel old = this.channel; + this.channel = null; + CompletableFuture> pending = this.connectFuture; + this.connectFuture = null; + if (pending != null && !pending.isDone()) { + pending.completeExceptionally(new ClosedChannelException()); + } + if (old != null) { + old.close(); + } + future = connectInternal(); } + joinConnect(future); } protected synchronized CompletableFuture> connectInternal() { + if (closed) { + CompletableFuture> failed = new CompletableFuture<>(); + failed.completeExceptionally(new ClosedChannelException()); + return failed; + } if (connectFuture != null) return connectFuture; - connectFuture = new CompletableFuture<>(); - connectFuture.thenAccept(servers -> this.iceServers = servers); - + CompletableFuture> future = new CompletableFuture<>(); + connectFuture = future; + future.thenAccept(servers -> this.iceServers = servers); + try { SslContext sslCtx = SslContextBuilder.forClient().build(); WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker( - uri, WebSocketVersion.V13, null, false, + uri, WebSocketVersion.V13, null, false, new DefaultHttpHeaders() .add("Authorization", xboxToken) .add("User-Agent", NetherNetConstants.SIGNALING_USER_AGENT) @@ -114,35 +167,162 @@ protected void initChannel(SocketChannel ch) { ChannelPipeline p = ch.pipeline(); p.addLast(sslCtx.newHandler(ch.alloc(), uri.getHost(), 443)); p.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192)); - p.addLast("ws-handshake", new WebSocketClientProtocolHandler(handshaker)); - p.addLast("ws-aggregator", new WebSocketFrameAggregator(16 * 1024)); // Allow 16KB aggregations + // dropPongFrames=false: pongs must reach channelRead so they + // refresh lastMessageReceivedAt for liveness detection. + p.addLast("ws-handshake", new WebSocketClientProtocolHandler(handshaker, true, false)); + p.addLast("ws-aggregator", new WebSocketFrameAggregator(128 * 1024)); p.addLast("handler", AbstractNetherNetXboxSignaling.this); } }); - this.channel = b.connect(uri.getHost(), 443).sync().channel(); + // Asynchronous on purpose: connectInternal may be invoked while + // holding this signaling's lock or (via channelInactive paths) on + // the event loop itself, where a sync() would deadlock. Callers + // that need to block use joinConnect on the returned future. + ChannelFuture connect = b.connect(uri.getHost(), 443); + this.channel = connect.channel(); + connect.addListener(f -> { + if (!f.isSuccess() && !future.isDone()) { + future.completeExceptionally(f.cause()); + } + }); } catch (Exception e) { Throwable cause = e.getCause() != null ? e.getCause() : e; - if (connectFuture != null) connectFuture.completeExceptionally(cause); + if (!future.isDone()) future.completeExceptionally(cause); + } + return future; + } + + /** + * Blocks until the given connect attempt has completed its credential + * exchange, translating failures (including a stalled handshake) into a + * ConnectException after cleaning up the half-open channel. + */ + protected void joinConnect(CompletableFuture> future) throws ConnectException { + try { + future.orTimeout(CONNECT_TIMEOUT_SECONDS, TimeUnit.SECONDS).join(); + } catch (Exception e) { + Throwable cause = e.getCause() != null ? e.getCause() : e; + abortConnect(); + if (cause instanceof ConnectException) throw (ConnectException) cause; + ConnectException ce = new ConnectException("Failed to connect to Xbox Signaling: " + cause.getMessage()); + ce.initCause(cause); + throw ce; + } + } + + private synchronized void abortConnect() { + Channel c = this.channel; + this.channel = null; + CompletableFuture> pending = this.connectFuture; + this.connectFuture = null; + if (pending != null && !pending.isDone()) { + pending.completeExceptionally(new ClosedChannelException()); + } + if (c != null) { + c.close(); } - return connectFuture; } @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt == WebSocketClientProtocolHandler.ClientHandshakeStateEvent.HANDSHAKE_COMPLETE) { log.debug("{} WebSocket Connected", getClass().getSimpleName()); + lastMessageReceivedAt = System.currentTimeMillis(); + scheduleRecurring(ctx, "ws-ping", () -> ctx.writeAndFlush(new PingWebSocketFrame()), + WS_PING_INTERVAL_SECONDS, WS_PING_INTERVAL_SECONDS); onConnected(ctx); } else { super.userEventTriggered(ctx, evt); } } + @Override + public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + // Track liveness: every inbound frame counts, including pong frames + // answering our protocol-level pings. Used by isChannelAlive(long) + // to detect silent half-closed TCP where channel.isActive() lies. + lastMessageReceivedAt = System.currentTimeMillis(); + super.channelRead(ctx, msg); + } + + /** + * Schedules a recurring task bound to the given channel's lifetime. The + * task is skipped while the channel is inactive, survives its own + * exceptions (scheduleAtFixedRate would otherwise cancel silently on the + * first throw), and is cancelled when the channel goes inactive. + */ + protected void scheduleRecurring(ChannelHandlerContext ctx, String name, Runnable task, + long initialDelaySeconds, long periodSeconds) { + ScheduledFuture future = ctx.executor().scheduleAtFixedRate(() -> { + try { + if (!ctx.channel().isActive()) { + return; + } + task.run(); + } catch (Throwable t) { + log.warn("{} task threw; loop continues: {}", name, t.getMessage()); + } + }, initialDelaySeconds, periodSeconds, TimeUnit.SECONDS); + + CopyOnWriteArrayList> tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + if (tasks == null) { + ctx.channel().attr(CHANNEL_TASKS).setIfAbsent(new CopyOnWriteArrayList<>()); + tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + } + tasks.add(future); + } + + /** + * @return true if the signaling WebSocket channel is active. Does NOT + * detect silent half-closed TCP (Netty can report active on a + * dead socket). For stricter checking use {@link #isChannelAlive(long)} + * or compare {@link #getMillisSinceLastMessage()} against your + * own threshold. + */ + public boolean isChannelAlive() { + Channel c = this.channel; + return c != null && c.isActive(); + } + + /** + * @param maxSilenceMillis max tolerated time since last received frame. + * The recurring protocol-level ping guarantees a + * pong at least every {@value #WS_PING_INTERVAL_SECONDS} + * seconds on a healthy socket, so thresholds of + * two to three times that are safe even when no + * signaling traffic is flowing. + * @return true if the channel is active and has received a frame within + * the given window. + */ + public boolean isChannelAlive(long maxSilenceMillis) { + if (!isChannelAlive()) return false; + long silence = getMillisSinceLastMessage(); + return silence >= 0 && silence <= maxSilenceMillis; + } + + /** + * @return milliseconds since the last received frame, or -1 if no frame + * has been received yet. + */ + public long getMillisSinceLastMessage() { + if (lastMessageReceivedAt == 0) return -1; + return System.currentTimeMillis() - lastMessageReceivedAt; + } + /** - * Called when the WebSocket handshake is complete. + * Called when the WebSocket handshake is complete. */ protected abstract void onConnected(ChannelHandlerContext ctx); + /** + * Called when a WebSocket channel of this signaling instance goes + * inactive, before the base class finishes its own cleanup. Subclasses + * use this to fail state tied to the dead socket (e.g. pending requests). + */ + protected void onChannelInactive(ChannelHandlerContext ctx) { + } + @Override public List getIceServers() { return this.iceServers; @@ -167,7 +347,7 @@ public void setSignalHandler(long connectionId, SignalHandler handler) { public void removeSignalHandler(long connectionId) { this.handlers.remove(connectionId); } - + @Override public void setAdvertisementData(PongData pongData) { // No-op for Xbox Signaling. @@ -175,8 +355,10 @@ public void setAdvertisementData(PongData pongData) { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { - if (connectFuture != null && !connectFuture.isDone()) { - connectFuture.completeExceptionally(cause); + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(cause); + } } log.error("Signaling Exception: {}", cause.getMessage(), cause); ctx.close(); @@ -184,19 +366,45 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { + CopyOnWriteArrayList> tasks = ctx.channel().attr(CHANNEL_TASKS).get(); + if (tasks != null) { + for (ScheduledFuture task : tasks) { + task.cancel(false); + } + tasks.clear(); + } synchronized (this) { - if (connectFuture != null && !connectFuture.isDone()) { - connectFuture.completeExceptionally(new ClosedChannelException()); + // Only clear state if the channel going inactive is still the + // current one. During a reconnect the old channel's inactive event + // arrives after the replacement is already installed and must not + // tear down the new connection's state. + if (ctx.channel() == this.channel) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(new ClosedChannelException()); + } + connectFuture = null; + this.channel = null; } - connectFuture = null; - this.channel = null; } + onChannelInactive(ctx); super.channelInactive(ctx); } @Override public void close() { - if (channel != null) channel.close(); + Channel c; + synchronized (this) { + closed = true; + c = this.channel; + this.channel = null; + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.completeExceptionally(new ClosedChannelException()); + } + connectFuture = null; + } + if (c != null) { + c.close(); + } eventLoopGroup.shutdownGracefully(); } @@ -207,7 +415,7 @@ protected void dispatchSignalToPipeline(String sender, String rawMsg) { if (parts.length < 2) return; long connectionId = Long.parseUnsignedLong(parts[1]); - + SignalHandler handler = handlers.get(connectionId); if (handler != null) { handler.onSignal(rawMsg); @@ -236,19 +444,19 @@ protected List parseTurnServers(JsonObject json) { for (JsonElement el : servers) { JsonObject server = el.getAsJsonObject(); List urls = new ArrayList<>(); - + JsonArray urlsArray = null; if (server.has("Urls")) urlsArray = server.getAsJsonArray("Urls"); else if (server.has("urls")) urlsArray = server.getAsJsonArray("urls"); if (urlsArray != null) { urlsArray.forEach(u -> urls.add(u.getAsString())); - + IceServerInfo.Builder info = new IceServerInfo.Builder().setUrls(urls); - + if (server.has("Username")) info.setUsername(server.get("Username").getAsString()); else if (server.has("username")) info.setUsername(server.get("username").getAsString()); - + if (server.has("Password")) info.setPassword(server.get("Password").getAsString()); else if (server.has("password")) info.setPassword(server.get("password").getAsString()); else if (server.has("Credential")) info.setPassword(server.get("Credential").getAsString()); @@ -264,4 +472,20 @@ protected List parseTurnServers(JsonObject json) { log.debug("Successfully parsed {} ICE servers.", result.size()); return result; } + + /** + * Stores freshly received TURN/ICE servers and completes the pending + * connect future if this was the credential exchange of a new socket. + * Called by subclasses whenever the service supplies credentials, both + * during connect and on later refreshes, so peer connections created at + * any point get unexpired TURN credentials. + */ + protected void updateIceServers(List servers) { + this.iceServers = servers; + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) { + connectFuture.complete(servers); + } + } + } } diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java index 4345a77..49b346e 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxRpcSignaling.java @@ -7,26 +7,48 @@ import com.google.gson.JsonObject; import com.google.gson.JsonParser; import dev.kastle.netty.channel.nethernet.NetherNetConstants; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandler.Sharable; import io.netty.channel.ChannelHandlerContext; import io.netty.handler.codec.http.websocketx.TextWebSocketFrame; import java.net.URI; import java.nio.channels.ClosedChannelException; -import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; +@Sharable public class NetherNetXboxRpcSignaling extends AbstractNetherNetXboxSignaling { private static final Gson gson = new GsonBuilder().serializeNulls().create(); - private final Map> pendingRequests = new ConcurrentHashMap<>(); + + /** Interval for proactively re-fetching TURN credentials so peer + * connections created on a long-lived socket never receive expired ones. */ + private static final long TURN_REFRESH_INTERVAL_SECONDS = 30 * 60; + + /** + * An in-flight JSON-RPC request, tagged with the WebSocket channel it was + * written to so that a dying channel only fails its own requests. During + * a reconnect the old channel's inactive event must not fail requests + * already sent on the replacement socket. + */ + private static final class PendingRequest { + final CompletableFuture future; + final Channel channel; + + PendingRequest(CompletableFuture future, Channel channel) { + this.future = future; + this.channel = channel; + } + } + + private final Map pendingRequests = new ConcurrentHashMap<>(); /** * Creates a NetherNetXboxRpcSignaling instance. - * + * * @param networkId The Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -36,7 +58,7 @@ public NetherNetXboxRpcSignaling(String networkId, String xboxToken) { /** * Creates a NetherNetXboxRpcSignaling instance. - * + * * @param localNetworkId The local Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -46,7 +68,7 @@ public NetherNetXboxRpcSignaling(long localNetworkId, String xboxToken) { /** * Creates a NetherNetXboxRpcSignaling instance with a random local Network ID. - * + * * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ public NetherNetXboxRpcSignaling(String xboxToken) { @@ -55,24 +77,47 @@ public NetherNetXboxRpcSignaling(String xboxToken) { @Override protected void onConnected(ChannelHandlerContext ctx) { - ctx.executor().scheduleAtFixedRate(() -> { - if (channel != null && channel.isActive()) { - sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_PING, new JsonObject()); - } - }, 30, 50, TimeUnit.SECONDS); + scheduleRecurring(ctx, "rpc-ping", () -> + sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_PING, new JsonObject()), 30, 50); + scheduleRecurring(ctx, "turn-refresh", this::refreshTurnCredentials, + TURN_REFRESH_INTERVAL_SECONDS, TURN_REFRESH_INTERVAL_SECONDS); + + refreshTurnCredentials(); + } + + /** + * Fetches TURN credentials and applies them via updateIceServers, which + * also completes the connect future during the initial exchange. On later + * refreshes a failure only logs; the previous credentials stay in place. + */ + private void refreshTurnCredentials() { sendJsonRpcRequest(NetherNetConstants.XBOX_RPC_METHOD_TURN_AUTH, new JsonObject()) - .thenAccept(response -> { - List servers = parseTurnServers(response); - if (connectFuture != null && !connectFuture.isDone()) connectFuture.complete(servers); - }) + .thenAccept(response -> updateIceServers(parseTurnServers(response))) .exceptionally(t -> { log.error("Failed to fetch TURN credentials", t); - if (connectFuture != null && !connectFuture.isDone()) connectFuture.completeExceptionally(t); + synchronized (this) { + if (connectFuture != null && !connectFuture.isDone()) connectFuture.completeExceptionally(t); + } return null; }); } + @Override + protected void onChannelInactive(ChannelHandlerContext ctx) { + // Fail everything that was waiting on a reply over the dead socket so + // callers see a prompt error instead of a future that never completes. + // Only requests written to THIS channel: during a reconnect the old + // channel's inactive event must not fail the new socket's requests. + pendingRequests.entrySet().removeIf(entry -> { + if (entry.getValue().channel == ctx.channel()) { + entry.getValue().future.completeExceptionally(new ClosedChannelException()); + return true; + } + return false; + }); + } + @Override protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) { String text = frame.text(); @@ -92,13 +137,14 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) private void handleResponse(JsonObject json) { if (!json.has("id") || json.get("id").isJsonNull()) return; String id = json.get("id").getAsString(); - CompletableFuture future = pendingRequests.remove(id); - + PendingRequest pending = pendingRequests.remove(id); + CompletableFuture future = pending != null ? pending.future : null; + if (future != null) { if (json.has("error") && !json.get("error").isJsonNull()) { JsonObject error = json.getAsJsonObject("error"); String msg = error.has("message") ? error.get("message").getAsString() : error.toString(); - + boolean isNotFound = msg.contains("Player not registered"); if (!isNotFound && error.has("data") && error.get("data").isJsonObject()) { JsonObject data = error.getAsJsonObject("data"); @@ -161,6 +207,7 @@ private void processIncomingMessage(JsonObject msgObj) { @Override public void sendSignal(String targetNetworkId, String data) { + var channel = this.channel; if (channel == null || !channel.isActive()) throw new IllegalStateException("Signaling channel is not active"); JsonObject innerParams = new JsonObject(); @@ -192,9 +239,10 @@ private CompletableFuture sendJsonRpcRequest(String method, JsonObje rpc.addProperty("id", id); CompletableFuture future = new CompletableFuture<>(); - pendingRequests.put(id, future); + var channel = this.channel; if (channel != null && channel.isActive()) { + pendingRequests.put(id, new PendingRequest(future, channel)); channel.writeAndFlush(new TextWebSocketFrame(gson.toJson(rpc))); } else { future.completeExceptionally(new ClosedChannelException()); @@ -207,6 +255,7 @@ private void sendJsonRpcResult(JsonElement id, JsonElement result) { response.add("id", id); response.add("result", result); response.addProperty("jsonrpc", "2.0"); + var channel = this.channel; if (channel != null && channel.isActive()) channel.writeAndFlush(new TextWebSocketFrame(gson.toJson(response))); } -} \ No newline at end of file +} diff --git a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java index 8e00d65..3894c89 100644 --- a/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java +++ b/transport-nethernet/src/main/java/dev/kastle/netty/channel/nethernet/signaling/NetherNetXboxSignaling.java @@ -10,7 +10,6 @@ import java.net.URI; import java.util.concurrent.ThreadLocalRandom; -import java.util.concurrent.TimeUnit; @Sharable public class NetherNetXboxSignaling extends AbstractNetherNetXboxSignaling { @@ -18,7 +17,7 @@ public class NetherNetXboxSignaling extends AbstractNetherNetXboxSignaling { /** * Creates a NetherNetXboxSignaling instance. - * + * * @param networkId The Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -28,7 +27,7 @@ public NetherNetXboxSignaling(String networkId, String xboxToken) { /** * Creates a NetherNetXboxSignaling instance. - * + * * @param localNetworkId The local Network ID to use. * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ @@ -38,7 +37,7 @@ public NetherNetXboxSignaling(long localNetworkId, String xboxToken) { /** * Creates a NetherNetXboxSignaling instance with a random local Network ID. - * + * * @param xboxToken The Minecraft Bedrock Session authorization header ('MCToken ***'). */ public NetherNetXboxSignaling(String xboxToken) { @@ -47,11 +46,16 @@ public NetherNetXboxSignaling(String xboxToken) { @Override protected void onConnected(ChannelHandlerContext ctx) { - ctx.executor().scheduleAtFixedRate(() -> { + scheduleRecurring(ctx, "app-ping", () -> { JsonObject ping = new JsonObject(); - ping.addProperty("Type", 0); - ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))); - }, 5, 5, TimeUnit.SECONDS); + ping.addProperty("Type", 0); + ctx.writeAndFlush(new TextWebSocketFrame(gson.toJson(ping))) + .addListener(future -> { + if (!future.isSuccess()) { + log.warn("Ping write failed: {}", future.cause() != null ? future.cause().getMessage() : "unknown"); + } + }); + }, 5, 5); } @Override @@ -78,11 +82,15 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) } case NetherNetConstants.XBOX_SIGNAL_CREDENTIALS -> { log.trace("Received Credentials"); - if (json.has("Message") && connectFuture != null && !connectFuture.isDone()) { + if (json.has("Message")) { String rawMsg = json.get("Message").getAsString(); JsonObject credentials = JsonParser.parseString(rawMsg).getAsJsonObject(); - - connectFuture.complete(parseTurnServers(credentials)); + + // Applied unconditionally, not just while connecting: the + // service pushes refreshed TURN credentials over the + // lifetime of the socket, and peer connections created + // later must not be handed the expired originals. + updateIceServers(parseTurnServers(credentials)); } } case NetherNetConstants.XBOX_SIGNAL_ACCEPTED, NetherNetConstants.XBOX_SIGNAL_ACK -> log.trace("Signal Ack: {}", text); @@ -95,6 +103,7 @@ protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame frame) @Override public void sendSignal(String targetNetworkId, String data) { + var channel = this.channel; if (channel != null && channel.isActive()) { JsonObject msg = new JsonObject(); msg.addProperty("Type", 1); @@ -105,4 +114,4 @@ public void sendSignal(String targetNetworkId, String data) { throw new IllegalStateException("Attempted to send signal to " + targetNetworkId + " but WebSocket is closed!"); } } -} \ No newline at end of file +}