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 <noreply@anthropic.com>

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());

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 final ResendTiming resendTiming = new ResendTiming();
-    private static final double alphaSmooth = 0.125D;
-    private static final double betaSmooth = 0.25D;
-    private double smoothedRtt;
-    private double smoothedRttVar;
-    private double currentRto;
-    //
 
     // Ping
     private List<Double> 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() &&
+            while (getState() != ClientConnectionState.DISCONNECTED) {
-                    getState() != ClientConnectionState.DISCONNECTED) {
                 try {
-                    if (getState() == ClientConnectionState.DISCONNECTED)
+                    response.waitForAcknowledgement(resendTiming.timeoutMs(response.getResends()));
-                        throw new IllegalStateException("no longer connected");
-
-                    // Wait one cycle for an ACK
-                    response.waitForAcknowledgement(1000L);
                     break;
                 } catch (TimeoutException e) {
-                    if (response.willResend())
+                    if (!response.willResend() || response.getResends() >= ResendTiming.MAX_RESENDS
-                        response.resend();
+                            || System.nanoTime() - response.getFirstSent()
-                    else
+                                    >= TimeUnit.MILLISECONDS.toNanos(ResendTiming.GIVE_UP_MS))
-                        throw e;
+                        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()
-                                + " (" +
+                                + " id=" + packet.getHeader().getPacketId()
-                                "transformation=" + transformation.getClass() +
+                                + " generation=" + networkPacket.getHeader().getGeneration()
-                                ", state=" + getState() +
+                                + " (" + transformation.getClass().getSimpleName() + ", " + getState() + ")",
-                                ", id=" + packet.getHeader().getPacketId() +
-                                ", generation=" + networkPacket.getHeader().getGeneration(),
                         e
                 );
             }
@@ -813,13 +811,9 @@ public abstract class AbstractTeamspeakClientSocket
                     }
                 }
 
-                if (smoothedRtt < 0)
+                // An acknowledgement of a resent packet can't tell which copy it answers.
-                    smoothedRtt = sampleRtt;
+                if (response.getResends() == 0)
-                else
+                    resendTiming.addSample(TimeUnit.NANOSECONDS.toMillis(rttNano));
-                    smoothedRtt = (long)((1 - alphaSmooth) * smoothedRtt + alphaSmooth * sampleRtt);
-
-                smoothedRttVar = (long)((1 - betaSmooth) * smoothedRttVar + betaSmooth * Math.abs(sampleRtt - smoothedRtt));
-                currentRto = smoothedRtt + Math.max(0D, 4 * smoothedRttVar);
             }
 
             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<Packet> future = new CompletableFuture<>();
         private final Map<Integer, PacketResponse> responsibleQueue;
-        private long lastSent = 0L;
+        private final long firstSent = System.nanoTime();
-        private int tries = 1, maxTries;
+        private long lastSent = firstSent;
-        private boolean willResend;
+        private int resends;
+        private final boolean willResend;
 
         public PacketResponse(NetworkPacket sentPacket,
                               Map<Integer, PacketResponse> responsibleQueue,
-                              int maxTries) {
+                              boolean willResend) {
             this.sentPacket = sentPacket;
             this.responsibleQueue = responsibleQueue;
-            this.maxTries = maxTries;
+            this.willResend = willResend;
-            this.lastSent = System.nanoTime();
-            this.willResend = maxTries > 0;
         }
 
         public NetworkPacket getPacket() {
@@ -1057,17 +1051,8 @@ public abstract class AbstractTeamspeakClientSocket
 
         public void resend() throws IOException {
             AbstractTeamspeakClientSocket.this.writeNetworkPacket(sentPacket);
-
+            setLastSent(System.nanoTime());
-            long t = System.nanoTime();
+            resends++;
-            long before = lastSent;
-            long elapsed = t - before;
-
-            setLastSent(t);
-
-            double elapsedSeconds = elapsed / 1_000_000_000D;
-            currentRto += elapsedSeconds;
-
-            tries ++;
         }
 
         /**
@@ -1100,12 +1085,12 @@ public abstract class AbstractTeamspeakClientSocket
             return willResend;
         }
 
-        public int getMaxTries() {
+        public int getResends() {
-            return maxTries;
+            return resends;
         }
 
-        public int getRetries() {
+        public long getFirstSent() {
-            return tries;
+            return firstSent;
         }
 
         public long getLastSent() {

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) {
+                    // The server confirms by letting us go, and after that acknowledges nothing: if the
-                        writePacket(new PacketBody2Command(
+                    // confirmation is lost, waiting for an ACK would only keep us resending until the give-up.
-                                ProtocolRole.CLIENT,
+                    writePacketAsync(clientDisconnect(reason));
-                                new SingleCommand("clientdisconnect", ProtocolRole.CLIENT))
+                    try {
-                        );
+                        waitForState(ClientConnectionState.DISCONNECTED, 5000L);
-                    } else {
+                    } catch (TimeoutException ignored) {
-                        writePacket(new PacketBody2Command(
-                                ProtocolRole.CLIENT,
-                                new SingleCommand(
-                                        "clientdisconnect", ProtocolRole.CLIENT,
-                                        new CommandSingleParameter("reasonid", "8"),
-                                        new CommandSingleParameter("reasonmsg", reason)
-                                ))
-                        );
                     }
-
-                    waitForState(ClientConnectionState.DISCONNECTED, 30000L);
-
                     break;
             }
         } finally {

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;
+    }
+}