nik/signal-cli
1
0
Fork 0
mirror of https://github.com/AsamK/signal-cli.git synced 2026-10-26 08:49:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix race between stopping and starting the receive thread

Browse Source 
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>
This commit is contained in:
iddo 2026-10-08 15:40:25 +00:00
parent 2a9a4b1f3c
commit f532a99199
2 changed files with 25 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
lib/src/main/java/org/asamk/signal/manager/helper/ReceiveHelper.java
View File

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

19
lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java
View File

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