hold silent payment notifications arriving before the subscribe response is recorded, discarding those of a replaced subscription

### src/main/java/com/sparrowwallet/sparrow/net/ElectrumServer.java
@@ -2557,15 +2557,15 @@ public static void holdSilentPaymentSubscription(Wallet wallet, SilentPaymentSca
                 SilentPaymentsSubscription response = electrumServerRpc.subscribeSilentPayments(getTransport(), wallet, scanPrivHex, spendPubHex, neededStart, NO_LABELS);
                 cache.lock();
                 try {
-                    cache.setServerStart(response.start_height);
+                    postSilentPaymentsNotified(spAddress, cache.setServerStart(response.start_height, TcpTransport.getLastResponseSequence()));
                 } finally {
                     cache.unlock();
                 }
             } catch(Exception e) {
                 cache.lock();
                 try {
                     if(rollbackSnapshot != null && cache.hasMultipleHolders()) {
-                        cache.restoreFromSnapshot(rollbackSnapshot);
+                        postSilentPaymentsNotified(spAddress, cache.restoreFromSnapshot(rollbackSnapshot));
                     } else {
                         cache.cancel();
                     }
@@ -2580,6 +2580,22 @@ public static void holdSilentPaymentSubscription(Wallet wallet, SilentPaymentSca
         }
     }
 
+    /**
+     * Posts the events for notifications the cache has applied. Called while its lock is still held, so that events
+     * reach the application thread in the order the notifications were applied rather than the order their posting
+     * threads happen to be scheduled in, which a replay racing a live notification would otherwise invert. Posting
+     * only enqueues, so it cannot block on the application thread. Shared by the live notification path and the
+     * replay of notifications held while the subscribe response was still in flight.
+     */
+    static void postSilentPaymentsNotified(String spAddress, List<SilentPaymentsScanCache.Notified> notifiedList) {
+        for(SilentPaymentsScanCache.Notified notified : notifiedList) {
+            Platform.runLater(() -> EventManager.get().post(new SilentPaymentsScanProgressEvent(spAddress, notified.progress())));
+            if(notified.historyUpdated()) {
+                Platform.runLater(() -> EventManager.get().post(new SilentPaymentsHistoryUpdatedEvent(spAddress)));
+            }
+        }
+    }
+
     private static boolean needsWiderCoverage(int neededStart, int serverStart) {
         boolean neededIsTimestamp = neededStart >= Transaction.MAX_BLOCK_LOCKTIME;
         boolean serverIsTimestamp = serverStart >= Transaction.MAX_BLOCK_LOCKTIME;

### src/main/java/com/sparrowwallet/sparrow/net/SilentPaymentsScanCache.java
@@ -1,18 +1,35 @@
 package com.sparrowwallet.sparrow.net;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 
 class SilentPaymentsScanCache {
+    private static final Logger log = LoggerFactory.getLogger(SilentPaymentsScanCache.class);
+
     private enum State { SCANNING, COMPLETED, CANCELLED }
 
+    //The specification has the server send the subscribe response before any notification for it, so only a stalled
+    //subscribe RPC can hold more than the one or two this leaves room for. The completion notification is the last to
+    //arrive and so the first that a cap would lose, which would leave the scan waiting for a completion that can no
+    //longer come, so reaching this cancels the scan instead of dropping anything
+    private static final int MAX_PENDING_NOTIFICATIONS = 1000;
+
     private Integer serverStart;
     private int refCount;
     private State state = State.SCANNING;
     private final List<SilentPaymentsTx> entries = new ArrayList<>();
 
+    //The subscribe response naming the canonical start height is recorded only once the RPC returns, while the read
+    //thread that delivered it is free to dispatch a notification for the same subscription first. One arriving that
+    //early is held here until setServerStart can say whether it belongs to the subscription being established
+    private final List<PendingNotification> pendingNotifications = new ArrayList<>();
+
     private final ReentrantLock lock = new ReentrantLock();
     private final Condition subscriptionComplete = lock.newCondition();
     private final Condition scanComplete = lock.newCondition();
@@ -51,6 +68,7 @@ void cancel() {
         assert lock.isHeldByCurrentThread();
         if(state == State.SCANNING) {
             state = State.CANCELLED;
+            pendingNotifications.clear();
             subscriptionComplete.signalAll();
             scanComplete.signalAll();
         }
@@ -90,6 +108,7 @@ void restartScan() {
         assert lock.isHeldByCurrentThread();
         serverStart = null;
         entries.clear();
+        pendingNotifications.clear();
         state = State.SCANNING;
     }
 
@@ -98,10 +117,85 @@ Integer getServerStart() {
         return serverStart;
     }
 
-    void setServerStart(int height) {
+    /**
+     * Records the canonical start height the subscribe response named, and applies the notifications held while it was
+     * unknown that this subscription produced: those naming the same height that the server wrote after the response
+     * establishing it, at or above the given sequence. Returns what the caller should post once it has released the lock.
+     */
+    List<Notified> setServerStart(int height, long responseSequence) {
         assert lock.isHeldByCurrentThread();
         serverStart = height;
         subscriptionComplete.signalAll();
+
+        return applyPendingNotifications(responseSequence);
+    }
+
+    /**
+     * Applies a notification to this cache, or holds it where the subscribe response naming the canonical start height
+     * has not been recorded yet. Returns what the caller should post once it has released the lock, which is nothing
+     * for a notification that was held or that names the start height of a subscription this cache has moved on from.
+     */
+    List<Notified> applyOrHold(int startHeight, long responseSequence, double progress, List<SilentPaymentsTx> history) {
+        assert lock.isHeldByCurrentThread();
+        //Tested against CANCELLED rather than SCANNING because a completed scan still applies the history deltas that
+        //follow it. A cancelled one applies nothing, and holding for it would refill the list cancelling just emptied
+        if(state == State.CANCELLED) {
+            return Collections.emptyList();
+        }
+
+        if(serverStart == null) {
+            if(pendingNotifications.size() >= MAX_PENDING_NOTIFICATIONS) {
+                log.warn("Cancelling silent payments scan: " + pendingNotifications.size() + " notifications held while the subscribe response is outstanding");
+                cancel();
+                return Collections.emptyList();
+            }
+
+            pendingNotifications.add(new PendingNotification(startHeight, responseSequence, progress, history));
+            return Collections.emptyList();
+        }
+
+        if(startHeight != serverStart) {
+            return Collections.emptyList();
+        }
+
+        return List.of(apply(progress, history));
+    }
+
+    /**
+     * Applies the held notifications the subscription now established produced, which are those naming its start
+     * height that the server wrote after the response establishing it. The rest are from the subscription it replaced:
+     * a server replaces a subscription for the same keys silently, so a re-subscribe at an unchanged start height
+     * leaves the two indistinguishable by anything the notification itself carries.
+     */
+    private List<Notified> applyPendingNotifications(long responseSequence) {
+        assert lock.isHeldByCurrentThread();
+        if(serverStart == null) {
+            //Still not established, so the held notifications cannot be matched yet and must keep waiting
+            return Collections.emptyList();
+        }
+
+        List<Notified> notified = new ArrayList<>();
+        for(PendingNotification pending : pendingNotifications) {
+            if(pending.startHeight() == serverStart && pending.responseSequence() >= responseSequence) {
+                notified.add(apply(pending.progress(), pending.history()));
+            }
+        }
+        pendingNotifications.clear();
+
+        return notified;
+    }
+
+    private Notified apply(double progress, List<SilentPaymentsTx> history) {
+        assert lock.isHeldByCurrentThread();
+        entries.addAll(history);
+
+        boolean justCompleted = false;
+        if(progress >= 1.0 && state == State.SCANNING) {
+            complete();
+            justCompleted = true;
+        }
+
+        return new Notified(progress, progress >= 1.0 && !justCompleted && !history.isEmpty());
     }
 
     int incrementRefCount() {
@@ -119,11 +213,6 @@ boolean hasMultipleHolders() {
         return refCount > 1;
     }
 
-    void addEntries(List<SilentPaymentsTx> newEntries) {
-        assert lock.isHeldByCurrentThread();
-        entries.addAll(newEntries);
-    }
-
     List<SilentPaymentsTx> snapshotEntries() {
         assert lock.isHeldByCurrentThread();
         return new ArrayList<>(entries);
@@ -144,14 +233,14 @@ Snapshot captureSnapshot() {
      * waiters whose conditions may have become re-evaluable. Used by the widening-failure recovery path
      * to restore an in-progress scan when the widening RPC fails but other holders still depend on the cache.
      */
-    void restoreFromSnapshot(Snapshot snapshot) {
+    List<Notified> restoreFromSnapshot(Snapshot snapshot) {
         assert lock.isHeldByCurrentThread();
         //If the cache was cancelled between captureSnapshot and now (e.g., a server disconnect ran
         //cancelSilentPaymentScans during the widening RPC), preserve the cancellation rather than
         //resurrecting a CANCELLED cache to its pre-widening state. Cancel already signalled both
         //conditions, so no further signal is needed here.
         if(state == State.CANCELLED) {
-            return;
+            return Collections.emptyList();
         }
         state = snapshot.state;
         serverStart = snapshot.serverStart;
@@ -161,6 +250,20 @@ void restoreFromSnapshot(Snapshot snapshot) {
         //scanComplete waiters whose state-condition was unchanged during the widening don't need a signal,
         //but signalling is harmless (they re-check isScanning() and re-await if still scanning).
         subscriptionComplete.signalAll();
+
+        //A notification held while the widening RPC was in flight belongs to the subscription just restored, that RPC
+        //having established nothing, so there is no response for the server to have written them before
+        return applyPendingNotifications(0L);
+    }
+
+    private record PendingNotification(int startHeight, long responseSequence, double progress, List<SilentPaymentsTx> history) {
+    }
+
+    /**
+     * What the caller should post for an applied notification once it has released the lock, kept separate so that
+     * this cache has no dependency on the event bus.
+     */
+    record Notified(double progress, boolean historyUpdated) {
     }
 
     static final class Snapshot {

### src/main/java/com/sparrowwallet/sparrow/net/SubscriptionService.java
@@ -6,8 +6,6 @@
 import com.github.arteam.simplejsonrpc.core.annotation.JsonRpcService;
 import com.sparrowwallet.sparrow.EventManager;
 import com.sparrowwallet.sparrow.event.NewBlockEvent;
-import com.sparrowwallet.sparrow.event.SilentPaymentsHistoryUpdatedEvent;
-import com.sparrowwallet.sparrow.event.SilentPaymentsScanProgressEvent;
 import com.sparrowwallet.sparrow.event.WalletNodeHistoryChangedEvent;
 import javafx.application.Platform;
 import org.slf4j.Logger;
@@ -59,27 +57,14 @@ public void silentPaymentsUpdate(@JsonRpcParam("subscription") final SilentPayme
             return;
         }
 
-        boolean justCompleted = false;
+        //A notification can reach the read thread before the subscribe response has been recorded, so the cache decides
+        //whether to apply it now, hold it until the canonical start height is known, or drop it as being from a prior subscribe
         cache.lock();
         try {
-            //Stale-notification filter: filter out notifications from a prior subscribe
-            Integer canonical = cache.getServerStart();
-            if(canonical == null || subscription.start_height != canonical) {
-                return;
-            }
-            cache.addEntries(history);
-            if(progress >= 1.0 && cache.isScanning()) {
-                cache.complete();
-                justCompleted = true;
-            }
+            ElectrumServer.postSilentPaymentsNotified(silentPaymentAddress,
+                    cache.applyOrHold(subscription.start_height, TcpTransport.getDeliveredResponses(), progress, history));
         } finally {
             cache.unlock();
         }
-
-        Platform.runLater(() -> EventManager.get().post(new SilentPaymentsScanProgressEvent(silentPaymentAddress, progress)));
-
-        if(progress >= 1.0 && !justCompleted && !history.isEmpty()) {
-            Platform.runLater(() -> EventManager.get().post(new SilentPaymentsHistoryUpdatedEvent(silentPaymentAddress)));
-        }
     }
 }

### src/main/java/com/sparrowwallet/sparrow/net/TcpTransport.java
@@ -24,6 +24,7 @@
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.regex.Matcher;
@@ -42,6 +43,9 @@ public class TcpTransport implements CloseableTransport, TimeoutCounter {
     private static final Pattern ID_PATTERN = Pattern.compile("\"id\"\\s*:\\s*(\\d+)");
     private static final JsonFactory JSON_FACTORY = new JsonFactory();
 
+    private static final AtomicLong deliveredResponses = new AtomicLong(); //Counts the responses delivered to callers ordering notifications with responses
+    private static final ThreadLocal<Long> lastResponseSequence = new ThreadLocal<>(); //Sequence of the response this thread's last RPC received
+
     protected final HostAndPort server;
     protected final SocketFactory socketFactory;
     protected final int[] readTimeouts;
@@ -191,6 +195,7 @@ private String readResponse() throws IOException {
                 throw new IOException("Transport closed");
             }
 
+            lastResponseSequence.set(deliveredResponses.get());
             reading = true;
 
             readingCondition.signal();
@@ -266,6 +271,7 @@ private void deliverResponse(String received) throws InterruptedException {
         readLock.lock();
         try {
             response = received;
+            deliveredResponses.incrementAndGet();
             reading = false;
             readingCondition.signal();
             while(!reading && running) {
@@ -367,6 +373,15 @@ public int getTimeoutCount() {
         return readTimeoutIndex;
     }
 
+    static long getDeliveredResponses() {
+        return deliveredResponses.get();
+    }
+
+    static long getLastResponseSequence() {
+        Long sequence = lastResponseSequence.get();
+        return sequence == null ? 0L : sequence;
+    }
+
     private static boolean isNotification(String json) {
         try(JsonParser parser = JSON_FACTORY.createParser(json)) {
             if(parser.nextToken() != JsonToken.START_OBJECT) {

### src/test/java/com/sparrowwallet/sparrow/net/SilentPaymentsScanCacheTest.java
@@ -0,0 +1,264 @@
+package com.sparrowwallet.sparrow.net;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class SilentPaymentsScanCacheTest {
+    private static final int SERVER_START = 800000;
+
+    //The place in the delivery order of the subscribe response establishing the subscription under test
+    private static final long RESPONSE_SEQUENCE = 10L;
+
+    private List<SilentPaymentsTx> history(String txid) {
+        return List.of(new SilentPaymentsTx(800001, txid, "00"));
+    }
+
+    private List<SilentPaymentsScanCache.Notified> applyOrHold(SilentPaymentsScanCache cache, int startHeight, double progress, List<SilentPaymentsTx> history) {
+        return applyOrHold(cache, startHeight, RESPONSE_SEQUENCE, progress, history);
+    }
+
+    private List<SilentPaymentsScanCache.Notified> applyOrHold(SilentPaymentsScanCache cache, int startHeight, long responseSequence, double progress, List<SilentPaymentsTx> history) {
+        cache.lock();
+        try {
+            return cache.applyOrHold(startHeight, responseSequence, progress, history);
+        } finally {
+            cache.unlock();
+        }
+    }
+
+    private List<SilentPaymentsScanCache.Notified> setServerStart(SilentPaymentsScanCache cache, int height) {
+        return setServerStart(cache, height, RESPONSE_SEQUENCE);
+    }
+
+    private List<SilentPaymentsScanCache.Notified> setServerStart(SilentPaymentsScanCache cache, int height, long responseSequence) {
+        cache.lock();
+        try {
+            return cache.setServerStart(height, responseSequence);
+        } finally {
+            cache.unlock();
+        }
+    }
+
+    private List<SilentPaymentsTx> entries(SilentPaymentsScanCache cache) {
+        cache.lock();
+        try {
+            return cache.snapshotEntries();
+        } finally {
+            cache.unlock();
+        }
+    }
+
+    @Test
+    public void testNotificationBeforeSubscribeResponseIsApplied() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+
+        //The read thread delivering the subscribe response is free to dispatch the first notification for it before
+        //the caller has parsed that response and recorded the start height it names
+        assertTrue(applyOrHold(cache, SERVER_START, 0.5, history("aa")).isEmpty(), "Nothing can be posted for a notification that is only held");
+        assertTrue(entries(cache).isEmpty());
+
+        List<SilentPaymentsScanCache.Notified> replayed = setServerStart(cache, SERVER_START);
+
+        assertEquals(1, entries(cache).size(), "The held notification must be applied once the start height is known");
+        assertEquals("aa", entries(cache).getFirst().tx_hash);
+        assertEquals(1, replayed.size());
+        assertEquals(0.5, replayed.getFirst().progress());
+        assertTrue(cache.isScanning());
+    }
+
+    @Test
+    public void testHeldCompletionCompletesTheScan() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+
+        //A dropped completion is what leaves getSilentPaymentHistory waiting on an untimed await until a reconnect
+        applyOrHold(cache, SERVER_START, 1.0, history("aa"));
+        assertTrue(cache.isScanning());
+
+        List<SilentPaymentsScanCache.Notified> replayed = setServerStart(cache, SERVER_START);
+
+        assertTrue(cache.isCompleted(), "A held completion must complete the scan when it is applied");
+        assertEquals(1, replayed.size());
+        assertFalse(replayed.getFirst().historyUpdated(), "The completion that completes the scan needs no history event of its own");
+    }
+
+    @Test
+    public void testHeldNotificationsAppliedInOrder() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+
+        applyOrHold(cache, SERVER_START, 0.5, history("aa"));
+        applyOrHold(cache, SERVER_START, 1.0, history("bb"));
+        setServerStart(cache, SERVER_START);
+
+        assertEquals(List.of("aa", "bb"), entries(cache).stream().map(tx -> tx.tx_hash).toList());
+        assertTrue(cache.isCompleted());
+    }
+
+    @Test
+    public void testHeldNotificationFromPriorSubscribeIsDiscarded() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+
+        applyOrHold(cache, SERVER_START, 1.0, history("aa"));
+        List<SilentPaymentsScanCache.Notified> replayed = setServerStart(cache, 700000);
+
+        assertTrue(entries(cache).isEmpty(), "A notification naming another subscription's start height must not be applied");
+        assertTrue(replayed.isEmpty());
+        assertTrue(cache.isScanning());
+    }
+
+    @Test
+    public void testLiveNotificationIsAppliedAndFiltered() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        setServerStart(cache, SERVER_START);
+
+        List<SilentPaymentsScanCache.Notified> notified = applyOrHold(cache, SERVER_START, 0.5, history("aa"));
+        assertEquals(1, notified.size());
+        assertEquals(1, entries(cache).size());
+
+        assertTrue(applyOrHold(cache, 700000, 1.0, history("bb")).isEmpty(), "A notification from a prior subscribe must still be dropped");
+        assertEquals(1, entries(cache).size());
+        assertTrue(cache.isScanning());
+    }
+
+    @Test
+    public void testLaterHistoryOnCompletedScanIsReported() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        setServerStart(cache, SERVER_START);
+        applyOrHold(cache, SERVER_START, 1.0, Collections.emptyList());
+        assertTrue(cache.isCompleted());
+
+        List<SilentPaymentsScanCache.Notified> notified = applyOrHold(cache, SERVER_START, 1.0, history("aa"));
+
+        assertEquals(1, notified.size());
+        assertTrue(notified.getFirst().historyUpdated(), "History arriving after the scan completed must be reported");
+    }
+
+    @Test
+    public void testCancelDiscardsHeldNotifications() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        applyOrHold(cache, SERVER_START, 1.0, history("aa"));
+
+        cache.lock();
+        try {
+            cache.cancel();
+        } finally {
+            cache.unlock();
+        }
+
+        assertTrue(setServerStart(cache, SERVER_START).isEmpty(), "A cancelled cache must not apply what it was holding");
+        assertTrue(entries(cache).isEmpty());
+    }
+
+    @Test
+    public void testRestartScanDiscardsHeldNotifications() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        setServerStart(cache, SERVER_START);
+
+        cache.lock();
+        try {
+            cache.restartScan();
+        } finally {
+            cache.unlock();
+        }
+
+        applyOrHold(cache, SERVER_START, 1.0, history("aa"));
+
+        cache.lock();
+        try {
+            cache.restartScan();
+        } finally {
+            cache.unlock();
+        }
+
+        assertTrue(setServerStart(cache, SERVER_START).isEmpty(), "A restarted scan must not apply what the prior one was holding");
+        assertTrue(entries(cache).isEmpty());
+    }
+
+    @Test
+    public void testOverflowCancelsTheScan() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        for(int i = 0; i < 1000; i++) {
+            applyOrHold(cache, SERVER_START, 0.5, history("aa"));
+        }
+        assertTrue(cache.isScanning());
+
+        applyOrHold(cache, SERVER_START, 0.5, history("aa"));
+
+        //Dropping to stay within the cap would lose the completion, which arrives last, and leave the scan waiting for one
+        assertTrue(cache.isCancelled(), "A cache that can hold no more must fail the scan rather than drop what it cannot hold");
+        assertTrue(setServerStart(cache, SERVER_START).isEmpty());
+        assertTrue(entries(cache).isEmpty());
+    }
+
+    @Test
+    public void testNotificationWrittenBeforeTheSubscribeResponseIsDiscarded() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+
+        //A subscription replaced by a re-subscribe at an unchanged start height can only be told from its replacement
+        //by where the server wrote its notifications relative to the response establishing that replacement
+        applyOrHold(cache, SERVER_START, RESPONSE_SEQUENCE - 1, 1.0, history("stale"));
+        applyOrHold(cache, SERVER_START, RESPONSE_SEQUENCE, 0.5, history("fresh"));
+
+        List<SilentPaymentsScanCache.Notified> replayed = setServerStart(cache, SERVER_START);
+
+        assertEquals(List.of("fresh"), entries(cache).stream().map(tx -> tx.tx_hash).toList(), "Only what the replacing subscription produced may be applied");
+        assertEquals(1, replayed.size());
+        assertTrue(cache.isScanning(), "A completion from the replaced subscription must not complete this scan");
+    }
+
+    @Test
+    public void testWideningFailureAppliesNotificationsHeldForTheRestoredSubscription() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        setServerStart(cache, SERVER_START);
+        applyOrHold(cache, SERVER_START, 1.0, history("scanned"));
+        assertTrue(cache.isCompleted());
+
+        //Widening captures the established scan and clears it before re-subscribing at an earlier start
+        SilentPaymentsScanCache.Snapshot snapshot;
+        cache.lock();
+        try {
+            snapshot = cache.captureSnapshot();
+            cache.restartScan();
+        } finally {
+            cache.unlock();
+        }
+
+        //A live update for the subscription still in place arrives while the widening RPC is outstanding, stamped
+        //below any response sequence, since the widening establishes no response for the server to have written it before
+        applyOrHold(cache, SERVER_START, RESPONSE_SEQUENCE - 1, 1.0, history("delta"));
+
+        List<SilentPaymentsScanCache.Notified> replayed;
+        cache.lock();
+        try {
+            replayed = cache.restoreFromSnapshot(snapshot);
+        } finally {
+            cache.unlock();
+        }
+
+        assertEquals(List.of("scanned", "delta"), entries(cache).stream().map(tx -> tx.tx_hash).toList(), "A held update belongs to the subscription restored");
+        assertTrue(cache.isCompleted());
+        assertEquals(1, replayed.size());
+        assertTrue(replayed.getFirst().historyUpdated(), "History arriving after the restored scan completed must be reported");
+    }
+
+    @Test
+    public void testCancelledCacheHoldsNothingFurther() {
+        SilentPaymentsScanCache cache = new SilentPaymentsScanCache();
+        cache.lock();
+        try {
+            cache.cancel();
+        } finally {
+            cache.unlock();
+        }
+
+        //A server still streaming into a cancelled cache would otherwise refill the list cancelling just emptied
+        assertTrue(applyOrHold(cache, SERVER_START, 1.0, history("aa")).isEmpty());
+        assertTrue(setServerStart(cache, SERVER_START).isEmpty(), "A cancelled cache must not hold what it will never apply");
+        assertTrue(entries(cache).isEmpty());
+    }
+}