mirror of
https://github.com/AsamK/signal-cli.git
synced 2026-10-26 08:49:38 +00:00
Fix race between stopping and starting the receive thread (#2142)
* Fix race between stopping and starting the receive thread When the last receive handler was removed, receiveThread was cleared before the thread had stopped. A handler added in that window, e.g. by a client reconnecting to the daemon in on-connection receive mode, started a second receive thread. While the stop request for the old thread was still set, each new thread ended immediately and started the next one. When the old thread ended, it cleared the reference to the newer thread and started yet another one, leaving two receive loops on the same websocket. The unreferenced loop was never stopped and kept receiving and acknowledging messages after all clients had gone. Keep receiveThread set until the receive thread has finished and only let the thread itself clear it. Its end block already starts a new receive thread if handlers were added in the meantime. Clear a stop request right before a new receive loop is started instead of when a loop ends, so it can neither be cleared by another loop nor stop a later one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Add test for adding a receive handler while the receive thread stops The receive loop is replaced by a fake one that can be held while it stops, so the test doesn't connect to the server. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: iddo <917594+iddo@users.noreply.github.com> Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
iddo
GitHub
parent
2a9a4b1f3c
commit
6f2ea80060
@ -42,8 +42,8 @@ public class ReceiveHelper {
|
||||
|
||||
private ReceiveConfig receiveConfig = new ReceiveConfig(false, false, false, false, false);
|
||||
private boolean hasCaughtUpWithOldMessages = false;
|
||||
private boolean isWaitingForMessage = false;
|
||||
private boolean shouldStop = false;
|
||||
private volatile boolean isWaitingForMessage = false;
|
||||
private volatile boolean shouldStop = false;
|
||||
private Callable authenticationFailureListener;
|
||||
private Callable caughtUpWithOldMessagesListener;
|
||||
|
||||
@ -71,6 +71,15 @@ public class ReceiveHelper {
|
||||
return isWaitingForMessage;
|
||||
}
|
||||
|
||||
/**
|
||||
* Must only be called right before a new receive loop is started, while no other receive loop is running.
|
||||
* Ending a receive loop doesn't clear the stop request, so a loop that was asked to stop can't continue
|
||||
* receiving, e.g. by retrying in receiveMessagesContinuously after an IOException.
|
||||
*/
|
||||
public void clearStopRequest() {
|
||||
this.shouldStop = false;
|
||||
}
|
||||
|
||||
public void receiveMessagesContinuously(Manager.ReceiveMessageHandler handler) {
|
||||
while (!shouldStop) {
|
||||
try {
|
||||
@ -111,7 +120,6 @@ public class ReceiveHelper {
|
||||
signalWebSocket.removeKeepAliveToken("receive");
|
||||
signalWebSocket.disconnect();
|
||||
webSocketStateDisposable.dispose();
|
||||
shouldStop = false;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -1510,15 +1510,18 @@ public class ManagerImpl implements Manager {
|
||||
private static final AtomicInteger threadNumber = new AtomicInteger(0);
|
||||
|
||||
private void startReceiveThreadIfRequired() {
|
||||
if (receiveThread != null || isReceivingSynchronous) {
|
||||
if (isReceiveThreadAlive() || isReceivingSynchronous) {
|
||||
return;
|
||||
}
|
||||
context.getReceiveHelper().clearStopRequest();
|
||||
receiveThread = Thread.ofPlatform().name("receive-" + threadNumber.getAndIncrement()).start(() -> {
|
||||
logger.debug("Starting receiving messages");
|
||||
context.getReceiveHelper().receiveMessagesContinuously(this::passReceivedMessageToHandlers);
|
||||
logger.debug("Finished receiving messages");
|
||||
synchronized (messageHandlers) {
|
||||
receiveThread = null;
|
||||
if (receiveThread == Thread.currentThread()) {
|
||||
receiveThread = null;
|
||||
}
|
||||
|
||||
// Check if in the meantime another handler has been registered
|
||||
if (!messageHandlers.isEmpty()) {
|
||||
@ -1529,6 +1532,13 @@ public class ManagerImpl implements Manager {
|
||||
});
|
||||
}
|
||||
|
||||
private boolean isReceiveThreadAlive() {
|
||||
// The receive thread clears receiveThread itself when it has finished, so it stays set while the thread is
|
||||
// still stopping and no second receive thread can be started in the meantime.
|
||||
// A receive thread that died from an uncaught exception never clears it.
|
||||
return receiveThread != null && receiveThread.isAlive();
|
||||
}
|
||||
|
||||
private void passReceivedMessageToHandlers(MessageEnvelope envelope, Throwable e) {
|
||||
synchronized (messageHandlers) {
|
||||
Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> {
|
||||
@ -1551,7 +1561,6 @@ public class ManagerImpl implements Manager {
|
||||
return;
|
||||
}
|
||||
thread = receiveThread;
|
||||
receiveThread = null;
|
||||
}
|
||||
|
||||
stopReceiveThread(thread);
|
||||
@ -1607,11 +1616,12 @@ public class ManagerImpl implements Manager {
|
||||
ReceiveMessageHandler handler
|
||||
) throws IOException, AlreadyReceivingException {
|
||||
synchronized (messageHandlers) {
|
||||
if (isReceiving()) {
|
||||
if (isReceiving() || isReceiveThreadAlive()) {
|
||||
throw new AlreadyReceivingException("Already receiving message.");
|
||||
}
|
||||
isReceivingSynchronous = true;
|
||||
receiveThread = Thread.currentThread();
|
||||
context.getReceiveHelper().clearStopRequest();
|
||||
}
|
||||
try {
|
||||
context.getReceiveHelper().receiveMessages(timeout, maxMessages, (envelope, e) -> {
|
||||
@ -2036,7 +2046,6 @@ public class ManagerImpl implements Manager {
|
||||
weakHandlers.clear();
|
||||
messageHandlers.clear();
|
||||
thread = receiveThread;
|
||||
receiveThread = null;
|
||||
}
|
||||
if (thread != null) {
|
||||
stopReceiveThread(thread);
|
||||
|
||||
@ -0,0 +1,256 @@
|
||||
package org.asamk.signal.manager.internal;
|
||||
|
||||
import org.asamk.signal.manager.Manager;
|
||||
import org.asamk.signal.manager.Settings;
|
||||
import org.asamk.signal.manager.api.MessageEnvelope;
|
||||
import org.asamk.signal.manager.api.ServiceEnvironment;
|
||||
import org.asamk.signal.manager.config.ServiceConfig;
|
||||
import org.asamk.signal.manager.helper.AccountFileUpdater;
|
||||
import org.asamk.signal.manager.helper.Context;
|
||||
import org.asamk.signal.manager.helper.ReceiveHelper;
|
||||
import org.asamk.signal.manager.storage.SignalAccount;
|
||||
import org.asamk.signal.manager.util.KeyUtils;
|
||||
import org.bouncycastle.jce.provider.BouncyCastleProvider;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.junit.jupiter.api.Timeout;
|
||||
import org.junit.jupiter.api.io.TempDir;
|
||||
import org.signal.core.models.ServiceId.ACI;
|
||||
import org.whispersystems.signalservice.api.push.ServiceIdType;
|
||||
|
||||
import java.nio.file.Path;
|
||||
import java.security.Security;
|
||||
import java.time.Duration;
|
||||
import java.util.Optional;
|
||||
import java.util.function.BooleanSupplier;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
@Timeout(60)
|
||||
class ReceiveThreadTest {
|
||||
|
||||
private static final Duration TIMEOUT = Duration.ofSeconds(10);
|
||||
|
||||
@TempDir
|
||||
Path directory;
|
||||
|
||||
private ManagerImpl manager;
|
||||
private FakeReceiveHelper receiveHelper;
|
||||
|
||||
@BeforeEach
|
||||
void setUp() throws Exception {
|
||||
// ManagerImpl.close() creates the push service socket, whose trust store needs BouncyCastle
|
||||
if (Security.getProvider(BouncyCastleProvider.PROVIDER_NAME) == null) {
|
||||
Security.addProvider(new BouncyCastleProvider());
|
||||
}
|
||||
|
||||
final var account = SignalAccount.createLinkedAccount(directory.toFile(),
|
||||
"account",
|
||||
ServiceEnvironment.STAGING,
|
||||
Settings.DEFAULT);
|
||||
account.setProvisioningData(null,
|
||||
ACI.parseOrThrow("11111111-1111-4111-8111-111111111111"),
|
||||
null,
|
||||
"test-password",
|
||||
new byte[]{1},
|
||||
KeyUtils.generateIdentityKeyPair(),
|
||||
null,
|
||||
KeyUtils.createProfileKey(),
|
||||
null,
|
||||
new byte[32],
|
||||
null);
|
||||
account.finishLinking(2, KeyUtils.generatePreKeysForType(account.getAccountData(ServiceIdType.ACI)), null);
|
||||
manager = new ManagerImpl(account,
|
||||
PathConfig.createDefault(directory.toFile()),
|
||||
new AccountFileUpdater() {
|
||||
@Override
|
||||
public void updateAccountIdentifiers(final String number, final ACI aci) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void removeAccount() {
|
||||
}
|
||||
},
|
||||
ServiceConfig.getServiceEnvironmentConfig(ServiceEnvironment.STAGING, "signal-cli-test"),
|
||||
"signal-cli-test");
|
||||
|
||||
// Replace the receive helper, so the receive thread doesn't connect to the server
|
||||
final var contextField = ManagerImpl.class.getDeclaredField("context");
|
||||
contextField.setAccessible(true);
|
||||
final var context = (Context) contextField.get(manager);
|
||||
receiveHelper = new FakeReceiveHelper(context);
|
||||
final var receiveHelperField = Context.class.getDeclaredField("receiveHelper");
|
||||
receiveHelperField.setAccessible(true);
|
||||
receiveHelperField.set(context, receiveHelper);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void tearDown() {
|
||||
receiveHelper.setHoldStoppingLoops(false);
|
||||
manager.close();
|
||||
}
|
||||
|
||||
@Test
|
||||
void handlerAddedWhileReceiveThreadIsStoppingDoesNotStartSecondReceiveThread() throws Exception {
|
||||
final var first = newHandler();
|
||||
final var second = newHandler();
|
||||
|
||||
manager.addReceiveHandler(first);
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 1));
|
||||
|
||||
receiveHelper.setHoldStoppingLoops(true);
|
||||
final var remover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(first));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.stoppingLoops == 1));
|
||||
|
||||
// Like a client that reconnects while the receive thread for its previous connection is still stopping
|
||||
manager.addReceiveHandler(second);
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 1));
|
||||
|
||||
receiveHelper.setHoldStoppingLoops(false);
|
||||
assertTrue(remover.join(TIMEOUT));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.startedLoops == 2 && receiveHelper.runningLoops == 1));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 2 && receiveHelper.maxRunningLoops == 1));
|
||||
|
||||
manager.removeReceiveHandler(second);
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 0));
|
||||
}
|
||||
|
||||
@Test
|
||||
void stopRequestedForStoppingReceiveThreadDoesNotStopNextReceiveThread() throws Exception {
|
||||
final var first = newHandler();
|
||||
final var second = newHandler();
|
||||
final var third = newHandler();
|
||||
|
||||
manager.addReceiveHandler(first);
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 1));
|
||||
|
||||
receiveHelper.setHoldStoppingLoops(true);
|
||||
final var firstRemover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(first));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.stoppingLoops == 1));
|
||||
manager.addReceiveHandler(second);
|
||||
final var secondRemover = Thread.ofPlatform().start(() -> manager.removeReceiveHandler(second));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.stopRequests == 2));
|
||||
|
||||
receiveHelper.setHoldStoppingLoops(false);
|
||||
assertTrue(firstRemover.join(TIMEOUT));
|
||||
assertTrue(secondRemover.join(TIMEOUT));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 0 && receiveHelper.startedThreads == 1));
|
||||
|
||||
manager.addReceiveHandler(third);
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.startedLoops == 2 && receiveHelper.runningLoops == 1));
|
||||
assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 2 && receiveHelper.maxRunningLoops == 1));
|
||||
}
|
||||
|
||||
private static Manager.ReceiveMessageHandler newHandler() {
|
||||
// An anonymous class instead of a lambda, so every handler is a separate instance
|
||||
return new Manager.ReceiveMessageHandler() {
|
||||
@Override
|
||||
public void handleMessage(final MessageEnvelope envelope, final Throwable e) {
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Instead of receiving messages from the server, a receive loop waits until it has been asked to stop.
|
||||
* While holdStoppingLoops is set, a loop that has been asked to stop doesn't return yet, like a loop that
|
||||
* is still busy when the last handler is removed.
|
||||
*/
|
||||
private static class FakeReceiveHelper extends ReceiveHelper {
|
||||
|
||||
private final Object lock = new Object();
|
||||
private boolean stopRequested;
|
||||
private boolean holdStoppingLoops;
|
||||
private int stopRequests;
|
||||
private int startedThreads;
|
||||
private int startedLoops;
|
||||
private int runningLoops;
|
||||
private int maxRunningLoops;
|
||||
private int stoppingLoops;
|
||||
|
||||
FakeReceiveHelper(final Context context) {
|
||||
super(context);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean requestStopReceiveMessages() {
|
||||
super.requestStopReceiveMessages();
|
||||
synchronized (lock) {
|
||||
stopRequested = true;
|
||||
stopRequests++;
|
||||
lock.notifyAll();
|
||||
}
|
||||
// The fake loop isn't blocked in a read that needs to be interrupted
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void clearStopRequest() {
|
||||
super.clearStopRequest();
|
||||
synchronized (lock) {
|
||||
stopRequested = false;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void receiveMessagesContinuously(final Manager.ReceiveMessageHandler handler) {
|
||||
synchronized (lock) {
|
||||
startedThreads++;
|
||||
lock.notifyAll();
|
||||
}
|
||||
super.receiveMessagesContinuously(handler);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void receiveMessages(
|
||||
final Optional<Duration> timeout,
|
||||
final Integer maxMessages,
|
||||
final Manager.ReceiveMessageHandler handler
|
||||
) {
|
||||
synchronized (lock) {
|
||||
startedLoops++;
|
||||
runningLoops++;
|
||||
maxRunningLoops = Math.max(maxRunningLoops, runningLoops);
|
||||
lock.notifyAll();
|
||||
if (await(() -> stopRequested)) {
|
||||
stoppingLoops++;
|
||||
lock.notifyAll();
|
||||
await(() -> !holdStoppingLoops);
|
||||
stoppingLoops--;
|
||||
}
|
||||
runningLoops--;
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
void setHoldStoppingLoops(final boolean hold) {
|
||||
synchronized (lock) {
|
||||
holdStoppingLoops = hold;
|
||||
lock.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Waits until the condition, which is evaluated while holding the lock, is true.
|
||||
* Returns false if it didn't become true within the timeout.
|
||||
*/
|
||||
boolean await(final BooleanSupplier condition) {
|
||||
synchronized (lock) {
|
||||
final var deadline = System.nanoTime() + TIMEOUT.toNanos();
|
||||
while (!condition.getAsBoolean()) {
|
||||
final var remainingMillis = (deadline - System.nanoTime()) / 1_000_000;
|
||||
if (remainingMillis <= 0) {
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
lock.wait(remainingMillis);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
x
Reference in New Issue
Block a user