From 8894da0b8e05a50df7c820f56f0b6728604b2fce Mon Sep 17 00:00:00 2001 From: wappensc-developer <230031610+wappensc-developer@users.noreply.github.com> Date: Tue, 20 Oct 2026 09:25:49 +0200 Subject: [PATCH] Deliver the envelope in hand to the last receive handler before it is removed removeReceiveHandler removed the handler before it stopped the receive thread. An envelope the thread was handling then was already cached and acknowledged to the server; it reached no handler, and the cache dropped it once handled, so it was lost. While removing the last handler stops the thread, the handler stays in drainingHandlers: it still gets that envelope, but doesn't keep the thread receiving. Co-Authored-By: Claude Opus 5.5 --- .../signal/manager/internal/ManagerImpl.java | 21 ++++++++++-- .../manager/internal/ReceiveThreadTest.java | 34 +++++++++++++++++++ 2 files changed, 52 insertions(+), 3 deletions(-) diff --git a/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java b/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java index 8658f365..78c8eefe 100644 --- a/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java +++ b/lib/src/main/java/org/asamk/signal/manager/internal/ManagerImpl.java @@ -174,6 +174,9 @@ public class ManagerImpl implements Manager { private boolean isReceivingSynchronous; private final Set weakHandlers = new HashSet<>(); private final Set messageHandlers = new HashSet<>(); + // The last handler while its removal stops the receive thread: it still gets the envelope the thread is + // handling, but doesn't keep the thread receiving. + private final Set drainingHandlers = new HashSet<>(); private final List closedListeners = new ArrayList<>(); private final List addressChangedListeners = new ArrayList<>(); private final CompositeDisposable disposable = new CompositeDisposable(); @@ -1541,7 +1544,7 @@ public class ManagerImpl implements Manager { private void passReceivedMessageToHandlers(MessageEnvelope envelope, Throwable e) { synchronized (messageHandlers) { - Stream.concat(messageHandlers.stream(), weakHandlers.stream()).forEach(h -> { + Stream.of(messageHandlers, weakHandlers, drainingHandlers).flatMap(Set::stream).forEach(h -> { try { h.handleMessage(envelope, e); } catch (Throwable ex) { @@ -1556,14 +1559,26 @@ public class ManagerImpl implements Manager { final Thread thread; synchronized (messageHandlers) { weakHandlers.remove(handler); - messageHandlers.remove(handler); + final var removed = messageHandlers.remove(handler); if (!messageHandlers.isEmpty() || receiveThread == null || isReceivingSynchronous) { return; } thread = receiveThread; + // The receive thread may be handling an envelope that is already cached and acknowledged to the server, + // and the cache drops it once it's handled. So the last handler keeps getting envelopes until the thread + // has stopped, otherwise that envelope would reach no handler and be lost. + if (removed) { + drainingHandlers.add(handler); + } } - stopReceiveThread(thread); + try { + stopReceiveThread(thread); + } finally { + synchronized (messageHandlers) { + drainingHandlers.remove(handler); + } + } } private void stopReceiveThread(final Thread thread) { diff --git a/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java b/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java index 63142cc1..8a456416 100644 --- a/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java +++ b/lib/src/test/java/org/asamk/signal/manager/internal/ReceiveThreadTest.java @@ -23,8 +23,10 @@ import java.nio.file.Path; import java.security.Security; import java.time.Duration; import java.util.Optional; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BooleanSupplier; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @Timeout(60) @@ -142,6 +144,28 @@ class ReceiveThreadTest { assertTrue(receiveHelper.await(() -> receiveHelper.startedThreads == 2 && receiveHelper.maxRunningLoops == 1)); } + @Test + void lastHandlerStillGetsTheMessageInHandWhenItIsRemoved() throws Exception { + final var received = new AtomicInteger(); + final var handler = new Manager.ReceiveMessageHandler() { + @Override + public void handleMessage(final MessageEnvelope envelope, final Throwable e) { + received.incrementAndGet(); + } + }; + + manager.addReceiveHandler(handler); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 1)); + + // Like an envelope that is already cached and acknowledged, and still being handled when the last + // handler goes: the loop passes it on before it stops + receiveHelper.setPassOnOneWhenStopped(true); + manager.removeReceiveHandler(handler); + + assertEquals(1, received.get()); + assertTrue(receiveHelper.await(() -> receiveHelper.runningLoops == 0 && receiveHelper.startedThreads == 1)); + } + private static Manager.ReceiveMessageHandler newHandler() { // An anonymous class instead of a lambda, so every handler is a separate instance return new Manager.ReceiveMessageHandler() { @@ -161,6 +185,7 @@ class ReceiveThreadTest { private final Object lock = new Object(); private boolean stopRequested; private boolean holdStoppingLoops; + private boolean passOnOneWhenStopped; private int stopRequests; private int startedThreads; private int startedLoops; @@ -213,6 +238,9 @@ class ReceiveThreadTest { maxRunningLoops = Math.max(maxRunningLoops, runningLoops); lock.notifyAll(); if (await(() -> stopRequested)) { + if (passOnOneWhenStopped) { + handler.handleMessage(null, null); + } stoppingLoops++; lock.notifyAll(); await(() -> !holdStoppingLoops); @@ -223,6 +251,12 @@ class ReceiveThreadTest { } } + void setPassOnOneWhenStopped(final boolean passOn) { + synchronized (lock) { + passOnOneWhenStopped = passOn; + } + } + void setHoldStoppingLoops(final boolean hold) { synchronized (lock) { holdStoppingLoops = hold;