From 2018c1c59347ec6655fb0a4d30ddf12f50c9a914 Mon Sep 17 00:00:00 2001 From: ericek111 Date: Thu, 24 Sep 2026 04:58:21 +0000 Subject: [PATCH] Resend like the official client; keep the handshake consistent under loss - Resend unacknowledged packets on the official client's schedule (mean RTT + 3 sigma + 50 ms over the last 20 samples, x1.4 per resend) and give up after 18 s, instead of a fixed 1 s x 30. - Resend INIT1 packets until the server answers; they are never ACKed. - Answer each server Init1 step once: a duplicated step-3 reply sent a second clientinitiv with a new alpha, leaving client and server with different keys (MAC check failures). - Disconnect without blocking on an ACK the server may never send. - Fail waitForState promptly when the connection drops, and name the packet in decrypt errors. Co-Authored-By: Claude Opus 5.5 --- .../client/LocalClientHandlerConnecting.java | 58 ++++++++++ .../client/AbstractTeamspeakClientSocket.java | 103 ++++++++---------- .../client/LocalTeamspeakClientSocket.java | 39 ++++--- .../protocol/socket/client/ResendTiming.java | 45 ++++++++ 4 files changed, 166 insertions(+), 79 deletions(-) create mode 100644 src/main/java/com/github/manevolent/ts3j/protocol/socket/client/ResendTiming.java diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java index bc312d2..f8cfd53 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/packet/handler/client/LocalClientHandlerConnecting.java @@ -22,6 +22,9 @@ import java.nio.ByteBuffer; import java.nio.charset.Charset; import java.util.Base64; import java.util.Random; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class LocalClientHandlerConnecting extends LocalClientHandler { @@ -30,12 +33,19 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { private byte[] randomBytes; private byte[] alphaBytes; + /** The last Init1 we sent; resent until the server answers it. */ + private PacketBody8Init1 pendingInit1; + /** The highest Init1 step the server has answered with; anything up to it again is a duplicate. */ + private int serverStep; + private ScheduledExecutorService init1Resender; + public LocalClientHandlerConnecting(LocalTeamspeakClientSocket client) { super(client); } @Override public void onAssigned() throws IOException, TimeoutException { + serverStep = 0; PacketBody8Init1 packet = new PacketBody8Init1(ProtocolRole.CLIENT); // Initialize connection @@ -53,10 +63,53 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { //getClient().writePacket(new PacketBody2Command(ProtocolRole.CLIENT, createInitIv())); } + @Override + public void onUnassigning() { + stopResendingInit1(); + } + + /** + * Sends an Init1 and keeps resending the same packet until the server answers: Init1 has no + * acknowledgements, so without this one lost datagram stalls the handshake for good. + */ private void sendInit1(PacketBody8Init1 packet) throws IOException, TimeoutException { packet.setVersion(INIT1_VERSION); + synchronized (this) { + pendingInit1 = packet; + if (init1Resender == null) { + init1Resender = Executors.newSingleThreadScheduledExecutor(r -> { + Thread thread = new Thread(r, "TS3J/Init1Resend"); + thread.setDaemon(true); + return thread; + }); + } + } getClient().writePacket(packet); + scheduleInit1Resend(packet, 0); + } + + private synchronized void scheduleInit1Resend(PacketBody8Init1 packet, int resends) { + if (pendingInit1 != packet || init1Resender == null) return; + init1Resender.schedule(() -> { + synchronized (this) { + if (pendingInit1 != packet) return; + } + try { + getClient().writePacket(packet); + } catch (Exception e) { + Ts3Debugging.debug("Problem resending Init1", e); + } + scheduleInit1Resend(packet, resends + 1); + }, getClient().getResendTimeout(resends), TimeUnit.MILLISECONDS); + } + + private synchronized void stopResendingInit1() { + pendingInit1 = null; + if (init1Resender != null) { + init1Resender.shutdownNow(); + init1Resender = null; + } } @Override @@ -65,6 +118,10 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { PacketBody8Init1 init1 = (PacketBody8Init1) packet.getBody(); Ts3Debugging.debug("Handle Init1 step " + init1.getStep().getNumber()); + // Answering a duplicate would redo the handshake step, and a second clientinitiv would + // come with a new alpha, leaving us with a different key than the server. + if (init1.getStep().getNumber() <= serverStep) return; + serverStep = init1.getStep().getNumber(); PacketBody8Init1.Step step; switch (init1.getStep().getNumber()) { @@ -138,6 +195,7 @@ public class LocalClientHandlerConnecting extends LocalClientHandler { sendInit1(new PacketBody8Init1(getClient().getRole().getOut(), step)); } else if (packet.getBody() instanceof PacketBody2Command) { + stopResendingInit1(); SingleCommand command = ((PacketBody2Command) packet.getBody()).parse().simplifyOne(); Ts3Debugging.debug(command.build()); diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java index db8b4c6..ba95c9b 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/AbstractTeamspeakClientSocket.java @@ -84,13 +84,7 @@ public abstract class AbstractTeamspeakClientSocket private long lastResponse; private long lastPing; - // RTT - private static final double alphaSmooth = 0.125D; - private static final double betaSmooth = 0.25D; - private double smoothedRtt; - private double smoothedRttVar; - private double currentRto; - // + private final ResendTiming resendTiming = new ResendTiming(); // Ping private List pings = new LinkedList<>(); @@ -251,15 +245,23 @@ public abstract class AbstractTeamspeakClientSocket return connectionState; } + /** + * Waits for the connection to reach a state. + * + * @throws IOException if the connection is closed before it gets there + */ public void waitForState(ClientConnectionState state, long wait) - throws InterruptedException, TimeoutException { + throws InterruptedException, TimeoutException, IOException { synchronized (connectionStateLock) { long start = System.currentTimeMillis(); - while (connectionState != state && System.currentTimeMillis() - start < wait) { + while (connectionState != state && connectionState != ClientConnectionState.DISCONNECTED + && System.currentTimeMillis() - start < wait) { connectionStateLock.wait(Math.max(0L, wait - (System.currentTimeMillis() - start))); } + if (connectionState != state && connectionState == ClientConnectionState.DISCONNECTED) + throw new IOException("disconnected while waiting for " + state.name() + " state"); if (connectionState != state) throw new TimeoutException("timeout waiting for " + state.name() + " state"); } @@ -518,7 +520,7 @@ public abstract class AbstractTeamspeakClientSocket response = new PacketResponse( networkPacket, responsibleQueue, - networkPacket.getHeader().getType().canResend() ? 30 : 0 + networkPacket.getHeader().getType().canResend() ) ); } else @@ -539,34 +541,32 @@ public abstract class AbstractTeamspeakClientSocket return willWaitForAck ? response : null; } + /** How long to wait for an answer to a packet that has been resent the given number of times. */ + public long getResendTimeout(int resends) { + return resendTiming.timeoutMs(resends); + } + /** * Waits for a packet's acknowledgement, resending it until it arrives. */ private void awaitAcknowledgement(PacketResponse response) throws IOException, TimeoutException { if (response != null) { - while (response.getRetries() < response.getMaxTries() && - getState() != ClientConnectionState.DISCONNECTED) { + while (getState() != ClientConnectionState.DISCONNECTED) { try { - if (getState() == ClientConnectionState.DISCONNECTED) - throw new IllegalStateException("no longer connected"); - - // Wait one cycle for an ACK - response.waitForAcknowledgement(1000L); + response.waitForAcknowledgement(resendTiming.timeoutMs(response.getResends())); break; } catch (TimeoutException e) { - if (response.willResend()) - response.resend(); - else - throw e; + if (!response.willResend() || response.getResends() >= ResendTiming.MAX_RESENDS + || System.nanoTime() - response.getFirstSent() + >= TimeUnit.MILLISECONDS.toNanos(ResendTiming.GIVE_UP_MS)) + throw new TimeoutException("send"); + response.resend(); } catch (CancellationException e) { throw e; } catch (Exception e) { throw new RuntimeException(e); } } - - if (response.getRetries() >= response.getMaxTries()) - throw new TimeoutException("send"); } } @@ -609,11 +609,9 @@ public abstract class AbstractTeamspeakClientSocket } catch (InvalidCipherTextException e) { throw new IOException( "failed to decrypt " + networkPacket.getHeader().getType().name() - + " (" + - "transformation=" + transformation.getClass() + - ", state=" + getState() + - ", id=" + packet.getHeader().getPacketId() + - ", generation=" + networkPacket.getHeader().getGeneration(), + + " id=" + packet.getHeader().getPacketId() + + " generation=" + networkPacket.getHeader().getGeneration() + + " (" + transformation.getClass().getSimpleName() + ", " + getState() + ")", e ); } @@ -813,13 +811,9 @@ public abstract class AbstractTeamspeakClientSocket } } - if (smoothedRtt < 0) - smoothedRtt = sampleRtt; - else - smoothedRtt = (long)((1 - alphaSmooth) * smoothedRtt + alphaSmooth * sampleRtt); - - smoothedRttVar = (long)((1 - betaSmooth) * smoothedRttVar + betaSmooth * Math.abs(sampleRtt - smoothedRtt)); - currentRto = smoothedRtt + Math.max(0D, 4 * smoothedRttVar); + // An acknowledgement of a resent packet can't tell which copy it answers. + if (response.getResends() == 0) + resendTiming.addSample(TimeUnit.NANOSECONDS.toMillis(rttNano)); } if (response != null) @@ -1005,7 +999,8 @@ public abstract class AbstractTeamspeakClientSocket } } catch (Throwable e) { Ts3Debugging.debug("Problem reading packet", e); - getExceptionHandler().accept(e); + // A disconnect interrupts whatever the handler was waiting on; that is not an error. + if (isReading()) getExceptionHandler().accept(e); } } } @@ -1037,18 +1032,17 @@ public abstract class AbstractTeamspeakClientSocket private final NetworkPacket sentPacket; private final CompletableFuture future = new CompletableFuture<>(); private final Map responsibleQueue; - private long lastSent = 0L; - private int tries = 1, maxTries; - private boolean willResend; + private final long firstSent = System.nanoTime(); + private long lastSent = firstSent; + private int resends; + private final boolean willResend; public PacketResponse(NetworkPacket sentPacket, Map responsibleQueue, - int maxTries) { + boolean willResend) { this.sentPacket = sentPacket; this.responsibleQueue = responsibleQueue; - this.maxTries = maxTries; - this.lastSent = System.nanoTime(); - this.willResend = maxTries > 0; + this.willResend = willResend; } public NetworkPacket getPacket() { @@ -1057,17 +1051,8 @@ public abstract class AbstractTeamspeakClientSocket public void resend() throws IOException { AbstractTeamspeakClientSocket.this.writeNetworkPacket(sentPacket); - - long t = System.nanoTime(); - long before = lastSent; - long elapsed = t - before; - - setLastSent(t); - - double elapsedSeconds = elapsed / 1_000_000_000D; - currentRto += elapsedSeconds; - - tries ++; + setLastSent(System.nanoTime()); + resends++; } /** @@ -1100,12 +1085,12 @@ public abstract class AbstractTeamspeakClientSocket return willResend; } - public int getMaxTries() { - return maxTries; + public int getResends() { + return resends; } - public int getRetries() { - return tries; + public long getFirstSent() { + return firstSent; } public long getLastSent() { diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java index d11041c..f7b91e7 100644 --- a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/LocalTeamspeakClientSocket.java @@ -247,7 +247,7 @@ public class LocalTeamspeakClientSocket ); } - throw new IOException("disconnected"); + throw new EOFException("disconnected"); } @Override @@ -1291,34 +1291,33 @@ public class LocalTeamspeakClientSocket disconnect(null); } + private static PacketBody2Command clientDisconnect(String reason) { + SingleCommand command = new SingleCommand("clientdisconnect", ProtocolRole.CLIENT); + if (reason != null) { + command.add(new CommandSingleParameter("reasonid", "8")); + command.add(new CommandSingleParameter("reasonmsg", reason)); + } + return new PacketBody2Command(ProtocolRole.CLIENT, command); + } + public void disconnect(String reason) throws IOException, TimeoutException, ExecutionException, InterruptedException { try { switch (getState()) { - case CONNECTING: case RETRIEVING_DATA: - waitForState(ClientConnectionState.CONNECTED, 10000L); + // Still loading, possibly stuck: say goodbye, but don't wait on a server that may not answer. + writePacketAsync(clientDisconnect(reason)); + break; case CONNECTED: - if (reason == null) { - writePacket(new PacketBody2Command( - ProtocolRole.CLIENT, - new SingleCommand("clientdisconnect", ProtocolRole.CLIENT)) - ); - } else { - writePacket(new PacketBody2Command( - ProtocolRole.CLIENT, - new SingleCommand( - "clientdisconnect", ProtocolRole.CLIENT, - new CommandSingleParameter("reasonid", "8"), - new CommandSingleParameter("reasonmsg", reason) - )) - ); + // The server confirms by letting us go, and after that acknowledges nothing: if the + // confirmation is lost, waiting for an ACK would only keep us resending until the give-up. + writePacketAsync(clientDisconnect(reason)); + try { + waitForState(ClientConnectionState.DISCONNECTED, 5000L); + } catch (TimeoutException ignored) { } - - waitForState(ClientConnectionState.DISCONNECTED, 30000L); - break; } } finally { diff --git a/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/ResendTiming.java b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/ResendTiming.java new file mode 100644 index 0000000..124c119 --- /dev/null +++ b/src/main/java/com/github/manevolent/ts3j/protocol/socket/client/ResendTiming.java @@ -0,0 +1,45 @@ +package com.github.manevolent.ts3j.protocol.socket.client; + +/** + * When to resend an unacknowledged packet and when to give up on it, as the official client + * (3.6.2) does it: the timeout is the mean round trip plus three standard deviations plus 50 ms + * over the last 20 samples (a flat 500 ms until there are 6), growing by 1.4x per resend. + */ +final class ResendTiming { + /** How long a packet may go unacknowledged before the connection counts as lost. */ + static final long GIVE_UP_MS = 18_000L; + /** Resends after which the connection is dropped regardless of time. */ + static final int MAX_RESENDS = 255; + + private static final int WINDOW = 20; + private static final int MIN_SAMPLES = 6; + private static final int INITIAL_MS = 500; + + private final long[] samples = new long[WINDOW]; + private int count; + private int next; + + /** Adds the round trip of a packet that was acknowledged without having been resent. */ + synchronized void addSample(long rttMs) { + samples[next] = rttMs; + next = (next + 1) % WINDOW; + if (count < WINDOW) count++; + } + + /** How long to wait for an acknowledgement after the given number of resends. */ + synchronized long timeoutMs(int resends) { + double base = INITIAL_MS; + if (count >= MIN_SAMPLES) { + double sum = 0, sumSquares = 0; + for (int i = 0; i < count; i++) { + sum += samples[i]; + sumSquares += (double) samples[i] * samples[i]; + } + double mean = sum / count; + double deviation = Math.sqrt(Math.max(0, (sumSquares - mean * sum) / count)); + base = (int) (mean + 3 * deviation + 50); + } + int timeout = (int) (base * Math.pow(1.4, resends)); + return timeout < 15_001 ? timeout : 19_000; + } +}