Delete the orphan a superseded registration leaves, fence its error

CHANGELOG.md

@@ -43,7 +43,10 @@ The notable changes of every release. The section of a version is the text of it
 - `SsfStreamRegistrar.start()` is idempotent and `stop()` waits for the registration thread to end
   before deleting the stream. A registration that completes after `stop()` publishes nothing and
   deletes a stream it created on the way if the registrar deletes on shutdown; if the registrar was
-  started again meanwhile, it leaves that stream to the new registration, which adopts it.
+  started again meanwhile, it leaves that stream to the new registration, which adopts it if it finds
+  it and deletes it once it publishes another one; a stream whose deletion is under way is not adopted.
+  `stop()` deletes only a stream published since the last `stop()`, not again the one that `stop()`
+  deleted before.
 - The `easyssf.receiver.poll.pending-acks` gauge no longer turns `NaN` after a garbage collection.
   Micrometer refers to what a gauge reads weakly, and the poller handed it a method reference
   nothing else held on to; the poller now keeps that reference for as long as it lives, so the

easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java

@@ -1,6 +1,7 @@
 package org.easyssf.receiver.stream;
 
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.List;
@@ -38,9 +39,6 @@ public class SsfStreamRegistrar {
 
     private static final Duration MAX_RETRY_DELAY = Duration.ofSeconds(30);
 
-    /** how long stop() waits for a registration in flight */
-    static final Duration STOP_TIMEOUT = Duration.ofSeconds(5);
-
     private final SsfStreamClient streamClient;
 
     private final SsfReceiverStream receiverStream;
@@ -53,12 +51,37 @@ public class SsfStreamRegistrar {
 
     private Duration initialRetryDelay = Duration.ofSeconds(1);
 
+    /** how long stop() waits for a registration in flight; the tests shorten it */
+    private Duration stopTimeout = Duration.ofSeconds(5);
+
     /**
      * The thread of the current registration; a registration that finds another thread
      * here, or none, was stopped and publishes nothing
      */
     private volatile Thread thread;
 
+    /**
+     * The thread that published the stream the receiver has, {@code null} once stop()
+     * ran: tells a registration that outlived its stop() whether the one that followed
+     * has published yet. Guarded by {@link #publication}.
+     */
+    private Thread publishedBy;
+
+    /**
+     * Streams created by registrations that outlived their stop() while another
+     * registration had already started, and so could not publish them. The registration
+     * that follows adopts such a stream if it finds it, and deletes it once it publishes
+     * another; stop() deletes what is left. Guarded by {@link #publication}.
+     */
+    private final List<String> unadoptedStreams = new ArrayList<>();
+
+    /**
+     * Streams whose deletion is under way, marked under {@link #publication} before the
+     * lock is released for the call: a registration does not adopt one of them, a slow
+     * delete would otherwise remove the stream a registration that followed uses.
+     */
+    private final Set<String> deletingStreams = new HashSet<>();
+
     /** start() and stop() are idempotent and safe to call from any thread */
     private final ReentrantLock lifecycle = new ReentrantLock();
 
@@ -166,6 +189,16 @@ public void addListener(Listener listener) {
         this.listeners.add(listener);
     }
 
+    /**
+     * How long {@link #stop()} waits for a registration in flight, 5 seconds by default.
+     * A registration that takes longer, because its HTTP client does not answer the
+     * interrupt, is left to finish on its own and publishes nothing.
+     */
+    void setStopTimeout(Duration stopTimeout) {
+        SsfAssert.isTrue(stopTimeout != null && stopTimeout.isPositive(), "stopTimeout must be positive");
+        this.stopTimeout = stopTimeout;
+    }
+
     public void setInitialRetryDelay(Duration initialRetryDelay) {
         SsfAssert.isTrue(initialRetryDelay != null && initialRetryDelay.isPositive(),
                 "initialRetryDelay must be positive");
@@ -218,12 +251,19 @@ public void stop() {
         this.lifecycle.lock();
         try {
             Thread thread;
+            boolean published;
+            List<String> unadopted;
             // a publication in progress completes first, so that the stream it publishes
             // is the one deleted below; none starts afterwards
             this.publication.lock();
             try {
                 thread = this.thread;
                 this.thread = null;
+                published = this.publishedBy != null;
+                this.publishedBy = null;
+                unadopted = List.copyOf(this.unadoptedStreams);
+                this.unadoptedStreams.clear();
+                this.deletingStreams.addAll(unadopted);
             }
             finally {
                 this.publication.unlock();
@@ -234,16 +274,18 @@ public void stop() {
                     // a registration in flight ends with the interrupted call; one that
                     // completes regardless sees it is no longer current and publishes
                     // nothing
-                    if (!thread.join(STOP_TIMEOUT)) {
+                    if (!thread.join(this.stopTimeout)) {
                         logger.warn("The registration of the SSF stream is still in progress");
                     }
                 }
                 catch (InterruptedException ex) {
                     Thread.currentThread().interrupt();
                 }
             }
+            // the receiver keeps the stream after stop(), so only one published since the
+            // last stop() is deleted, not one a previous stop() deleted already
             String registeredStreamId = this.receiverStream.getStreamId();
-            if (this.deleteOnShutdown && this.desiredStream != null && registeredStreamId != null) {
+            if (this.deleteOnShutdown && this.desiredStream != null && published && registeredStreamId != null) {
                 try {
                     this.streamClient.deleteStream(registeredStreamId);
                     logger.info("Deleted SSF stream " + registeredStreamId);
@@ -252,6 +294,7 @@ public void stop() {
                     logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage());
                 }
             }
+            deleteOrphans(unadopted);
         }
         finally {
             this.lifecycle.unlock();
@@ -268,30 +311,44 @@ private void registerWithRetries() {
         while (!thread.isInterrupted()) {
             try {
                 Registration registration = lookUp();
+                // decided under the lock, deleted after it: a slow transmitter must not
+                // hold up stop()
+                List<String> orphans = new ArrayList<>();
                 this.publication.lock();
                 try {
                     if (this.thread == thread) {
-                        publish(registration);
+                        orphans = publish(registration);
                     }
                     else if (this.thread == null) {
                         // stopped while the transmitter was called: the result is not
                         // published, and a stream created on the way is not left behind
                         if (registration.created() && this.deleteOnShutdown) {
-                            deleteQuietly(registration.stream().streamId());
+                            orphans.add(registration.stream().streamId());
                         }
                     }
-                    else {
+                    else if (registration.created() && this.deleteOnShutdown) {
                         // stopped and started again while the transmitter was called,
-                        // because stop() gave up waiting: the new registration has found
-                        // the stream created on the way, or will, so it stays; the next
-                        // stop() deletes the stream the receiver has then
-                        logger.debug("A registration that outlived its stop() leaves SSF stream "
-                                + registration.stream().streamId() + " to the one that followed");
+                        // because stop() gave up waiting: the registration that followed
+                        // adopts the stream created on the way if it finds it, so it is
+                        // not deleted on the spot. Once the other has published, the
+                        // stream it uses is known: any other is an orphan.
+                        String streamId = registration.stream().streamId();
+                        if (this.publishedBy != this.thread) {
+                            // the other has not published yet, it decides on publishing
+                            this.unadoptedStreams.add(streamId);
+                            logger.debug("A registration that outlived its stop() leaves SSF stream " + streamId
+                                    + " to the one that followed");
+                        }
+                        else if (!streamId.equals(this.receiverStream.getStreamId())) {
+                            orphans.add(streamId);
+                        }
                     }
+                    this.deletingStreams.addAll(orphans);
                 }
                 finally {
                     this.publication.unlock();
                 }
+                deleteOrphans(orphans);
                 return;
             }
             catch (SsfStreamIssuerMismatchException ex) {
@@ -310,12 +367,19 @@ else if (this.thread == null) {
                 return;
             }
             catch (RuntimeException ex) {
-                if (this.thread != thread) {
-                    // stopped meanwhile, possibly started again: not this registration's
-                    // error to report
-                    return;
+                // the check and the report go together under the lock, so that a
+                // registration stopped meanwhile, possibly replaced, reports no error of
+                // its own over the one that followed
+                this.publication.lock();
+                try {
+                    if (this.thread != thread) {
+                        return;
+                    }
+                    this.lastError = ex.getMessage();
+                }
+                finally {
+                    this.publication.unlock();
                 }
-                this.lastError = ex.getMessage();
                 logger.warn("Could not set up the SSF stream, trying again in " + delay.toSeconds() + "s: "
                         + ex.getMessage());
                 logger.debug("Cause of the failed SSF stream setup", ex);
@@ -339,7 +403,7 @@ else if (this.thread == null) {
      */
     public SsfStreamConfiguration register() {
         Registration registration = lookUp();
-        publish(registration);
+        deleteOrphans(publish(registration));
         return registration.stream();
     }
 
@@ -354,7 +418,12 @@ private Registration lookUp() {
             return new Registration(this.streamClient.getStream(this.streamId), "Using", false);
         }
         List<SsfStreamConfiguration> streams = this.streamClient.getStreams();
-        SsfStreamConfiguration existing = streams.stream().filter(this::hasDesiredDelivery).findFirst().orElse(null);
+        Set<String> deleting = deletingStreams();
+        SsfStreamConfiguration existing = streams.stream()
+            .filter((stream) -> !deleting.contains(stream.streamId()))
+            .filter(this::hasDesiredDelivery)
+            .findFirst()
+            .orElse(null);
         if (existing != null && hasDesiredEvents(existing)) {
             return new Registration(existing, "Using existing", false);
         }
@@ -373,17 +442,44 @@ private Registration lookUp() {
         catch (SsfStreamException ex) {
             // the transmitter allows a single stream per receiver and it is configured
             // differently
-            if (ex.getStatusCode() != 409 || streams.size() != 1) {
+            if (ex.getStatusCode() != 409 || streams.size() != 1 || deleting.contains(streams.get(0).streamId())) {
                 throw ex;
             }
             return new Registration(this.streamClient.updateStream(streams.get(0).streamId(), changes()), "Updated",
                     false);
         }
     }
 
-    private void publish(Registration registration) {
+    /**
+     * Publishes the stream and tells the listeners.
+     * @return the streams of stopped registrations the caller deletes, once it no longer
+     * holds the {@link #publication} lock: those not adopted, as the receiver uses this
+     * one
+     */
+    private List<String> publish(Registration registration) {
         SsfStreamConfiguration stream = registration.stream();
-        this.receiverStream.setConfiguration(stream);
+        List<String> orphans = new ArrayList<>();
+        // under the lock for a direct call of register() as well; it is reentrant for the
+        // registration thread, which holds it already
+        this.publication.lock();
+        try {
+            if (this.deletingStreams.contains(stream.streamId())) {
+                // found by the look-up before a delete marked it: looked up again
+                throw new SsfStreamException("SSF stream " + stream.streamId() + " is being deleted", 0, null);
+            }
+            this.receiverStream.setConfiguration(stream);
+            this.publishedBy = Thread.currentThread();
+            for (String unadopted : this.unadoptedStreams) {
+                if (!unadopted.equals(stream.streamId())) {
+                    orphans.add(unadopted);
+                }
+            }
+            this.unadoptedStreams.clear();
+            this.deletingStreams.addAll(orphans);
+        }
+        finally {
+            this.publication.unlock();
+        }
         logger.info(registration.action() + " SSF stream " + stream.streamId() + " (delivery " + stream.deliveryMethod()
                 + " " + Objects.toString(stream.deliveryEndpointUrl(), "") + ", events "
                 + stream.eventsDelivered().stream().map(SsfEventTypes::aliasOf).toList() + ", audience "
@@ -399,6 +495,43 @@ private void publish(Registration registration) {
                 logger.debug("Cause of the failed listener", ex);
             }
         }
+        return orphans;
+    }
+
+    /**
+     * Deletes streams that registrations created and the receiver does not use; called
+     * without the {@link #publication} lock, as the transmitter may be slow.
+     */
+    private void deleteOrphans(List<String> orphans) {
+        for (String streamId : orphans) {
+            logger.info("Deleting SSF stream " + streamId + ", created by a registration that was stopped before it "
+                    + "could publish it");
+            try {
+                deleteQuietly(streamId);
+            }
+            finally {
+                this.publication.lock();
+                try {
+                    this.deletingStreams.remove(streamId);
+                }
+                finally {
+                    this.publication.unlock();
+                }
+            }
+        }
+    }
+
+    /**
+     * @return the streams a delete is under way for, to leave them out of a look-up
+     */
+    private Set<String> deletingStreams() {
+        this.publication.lock();
+        try {
+            return Set.copyOf(this.deletingStreams);
+        }
+        finally {
+            this.publication.unlock();
+        }
     }
 
     private void deleteQuietly(String streamId) {