From 9c9616e51cd9036654fbad7f664c8ac60615f216 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:32:02 +0100 Subject: [PATCH 1/4] Delete the orphan a superseded registration leaves, fence its error Fixes #63 Signed-off-by: Thomas Darimont --- CHANGELOG.md | 3 +- .../receiver/stream/SsfStreamRegistrar.java | 40 +++++++++++---- .../stream/SsfStreamManagementTests.java | 50 +++++++++++++++++++ 3 files changed, 82 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 48784ff..4747e58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,7 +43,8 @@ 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, and + deletes it only if the new registration already uses another one. - 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 diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index 2860a4b..c468c4f 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -282,11 +282,24 @@ else if (this.thread == null) { } else { // 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 new registration adopts the + // stream created on the way if it finds it, so the stream stays + // unless the receiver already has another one: then it is an + // orphan, created after the new registration looked. A new + // registration that has not published yet may still adopt it, + // or create one of its own and leave it behind. + String streamId = registration.stream().streamId(); + String published = this.receiverStream.getStreamId(); + if (registration.created() && this.deleteOnShutdown && published != null + && !published.equals(streamId)) { + logger.debug("A registration that outlived its stop() deletes SSF stream " + streamId + + ", the one that followed uses " + published); + deleteQuietly(streamId); + } + else { + logger.debug("A registration that outlived its stop() leaves SSF stream " + streamId + + " to the one that followed"); + } } } finally { @@ -310,12 +323,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); diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index 3570d1e..f847e0b 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -368,6 +368,56 @@ void registrationThatOutlivesItsStopLeavesTheStreamToTheRegistrationThatFollowed assertThat(transmitter.streams()).isEmpty(); } + @Test + void registrationThatOutlivesItsStopDeletesTheStreamItCreatedAfterTheNextOnePublished() throws Exception { + CountDownLatch discovered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + // the first registration stalls after it looked for streams and found none, and + // does not care for the interrupt of stop() either + SsfHttpClient stuck = (request) -> { + var response = http.execute(request); + if ("GET".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && discovered.getCount() > 0) { + discovered.countDown(); + boolean released = false; + while (!released) { + try { + release.await(); + released = true; + } + catch (InterruptedException ex) { + // keeps waiting + } + } + } + return response; + }; + SsfStreamClient client = new SsfStreamClient(stuck, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.start(); + assertThat(discovered.await(5, TimeUnit.SECONDS)).isTrue(); + + // stop() gives the first registration up, the second finds no stream and creates + // one + registrar.stop(); + registrar.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> this.receiverStream.getStreamId() != null); + String published = this.receiverStream.getStreamId(); + + // the first registration goes on to create a stream of its own, which nobody uses + release.countDown(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.streams().size() == 1); + await().during(Duration.ofMillis(300)).until(() -> transmitter.streams().size() == 1); + assertThat(transmitter.streams().get(0).get("stream_id")).isEqualTo(published); + assertThat(this.receiverStream.getStreamId()).isEqualTo(published); + + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + } + @Test void obtainsNewAccessTokenWhenTheTransmitterRejectsTheCurrentOne() { SsfStreamClient client = new SsfStreamClient(http, this.tokenProvider, From 409b5e5ba988944ef2fb752eff907dc8de69071b Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:44:05 +0100 Subject: [PATCH 2/4] Hand a stopped registration's stream to the one that followed Signed-off-by: Thomas Darimont --- CHANGELOG.md | 5 +- .../receiver/stream/SsfStreamRegistrar.java | 99 +++++++++--- .../stream/SsfStreamManagementTests.java | 144 +++++++++++++++++- .../org/easyssf/test/TestTransmitter.java | 14 +- 4 files changed, 236 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4747e58..4ed959d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,8 +43,9 @@ 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, and - deletes it only if the new registration already uses another one. + 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. `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 diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index c468c4f..3d6d2df 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/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,30 @@ 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 unadoptedStreams = new ArrayList<>(); + /** start() and stop() are idempotent and safe to call from any thread */ private final ReentrantLock lifecycle = new ReentrantLock(); @@ -166,6 +182,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 +244,18 @@ public void stop() { this.lifecycle.lock(); try { Thread thread; + boolean published; + List 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(); } finally { this.publication.unlock(); @@ -234,7 +266,7 @@ 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"); } } @@ -242,8 +274,10 @@ public void stop() { 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 +286,11 @@ public void stop() { logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); } } + for (String streamId : unadopted) { + logger.info("Deleting SSF stream " + streamId + ", created by a registration that was stopped before " + + "it could publish it"); + deleteQuietly(streamId); + } } finally { this.lifecycle.unlock(); @@ -280,26 +319,25 @@ else if (this.thread == null) { deleteQuietly(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 adopts the - // stream created on the way if it finds it, so the stream stays - // unless the receiver already has another one: then it is an - // orphan, created after the new registration looked. A new - // registration that has not published yet may still adopt it, - // or create one of its own and leave it behind. + // 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(); - String published = this.receiverStream.getStreamId(); - if (registration.created() && this.deleteOnShutdown && published != null - && !published.equals(streamId)) { - logger.debug("A registration that outlived its stop() deletes SSF stream " + streamId - + ", the one that followed uses " + published); - deleteQuietly(streamId); - } - else { + 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())) { + logger.info("Deleting SSF stream " + streamId + ", created by a registration that was " + + "stopped before it could publish it; the receiver uses " + + this.receiverStream.getStreamId()); + deleteQuietly(streamId); + } } } finally { @@ -403,7 +441,26 @@ private Registration lookUp() { private void publish(Registration registration) { SsfStreamConfiguration stream = registration.stream(); - this.receiverStream.setConfiguration(stream); + // 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 { + this.receiverStream.setConfiguration(stream); + this.publishedBy = Thread.currentThread(); + // a stream a stopped registration created is adopted, if it is this one, or + // deleted, as the receiver uses this one + for (String unadopted : this.unadoptedStreams) { + if (!unadopted.equals(stream.streamId())) { + logger.info("Deleting SSF stream " + unadopted + ", created by a registration that was stopped " + + "before it could publish it; the receiver uses " + stream.streamId()); + deleteQuietly(unadopted); + } + } + this.unadoptedStreams.clear(); + } + 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 " diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index f847e0b..d04bf38 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -9,6 +9,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import org.easyssf.core.SsfDeliveryMethod; @@ -35,6 +36,12 @@ class SsfStreamManagementTests { private static final TestTransmitter transmitter = new TestTransmitter(); + /** + * the registrations of these tests outlive their stop() on purpose, no need to wait + * long + */ + private static final Duration STOP_TIMEOUT = Duration.ofMillis(200); + private static final SsfHttpClient http = new JdkSsfHttpClient(); private final SsfReceiverStream receiverStream = new SsfReceiverStream(); @@ -341,6 +348,7 @@ void registrationThatOutlivesItsStopLeavesTheStreamToTheRegistrationThatFollowed SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); registrar.setDeleteOnShutdown(true); + registrar.setStopTimeout(STOP_TIMEOUT); registrar.addListener(registered::add); registrar.start(); assertThat(created.await(5, TimeUnit.SECONDS)).isTrue(); @@ -351,8 +359,7 @@ void registrationThatOutlivesItsStopLeavesTheStreamToTheRegistrationThatFollowed // one, which finds the stream the first created and adopts it long stopping = System.nanoTime(); registrar.stop(); - assertThat(Duration.ofNanos(System.nanoTime() - stopping)) - .isGreaterThanOrEqualTo(SsfStreamRegistrar.STOP_TIMEOUT); + assertThat(Duration.ofNanos(System.nanoTime() - stopping)).isGreaterThanOrEqualTo(STOP_TIMEOUT); registrar.start(); await().atMost(Duration.ofSeconds(5)).until(() -> this.receiverStream.getStreamId() != null); assertThat(this.receiverStream.getStreamId()).isEqualTo(streamId); @@ -370,6 +377,7 @@ void registrationThatOutlivesItsStopLeavesTheStreamToTheRegistrationThatFollowed @Test void registrationThatOutlivesItsStopDeletesTheStreamItCreatedAfterTheNextOnePublished() throws Exception { + transmitter.setSeveralStreamsAllowed(true); CountDownLatch discovered = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); // the first registration stalls after it looked for streams and found none, and @@ -397,6 +405,7 @@ void registrationThatOutlivesItsStopDeletesTheStreamItCreatedAfterTheNextOnePubl SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); registrar.setDeleteOnShutdown(true); + registrar.setStopTimeout(STOP_TIMEOUT); registrar.start(); assertThat(discovered.await(5, TimeUnit.SECONDS)).isTrue(); @@ -418,6 +427,137 @@ void registrationThatOutlivesItsStopDeletesTheStreamItCreatedAfterTheNextOnePubl assertThat(transmitter.streams()).isEmpty(); } + @Test + void streamAStoppedRegistrationLeavesIsAdoptedByTheNextOneAndNotMistakenForAnEarlierOne() throws Exception { + // a first registration published a stream and stop() deleted it; the receiver + // still + // remembers that stream, which must not pass for the one the next registration + // uses + AtomicBoolean stallCreate = new AtomicBoolean(); + CountDownLatch created = new CountDownLatch(1); + CountDownLatch releaseCreate = new CountDownLatch(1); + AtomicBoolean stallDiscovery = new AtomicBoolean(); + CountDownLatch discovered = new CountDownLatch(1); + CountDownLatch releaseDiscovery = new CountDownLatch(1); + SsfHttpClient stuck = (request) -> { + var response = http.execute(request); + if ("POST".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && stallCreate.getAndSet(false)) { + created.countDown(); + awaitIgnoringInterrupts(releaseCreate); + } + if ("GET".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && stallDiscovery.getAndSet(false)) { + discovered.countDown(); + awaitIgnoringInterrupts(releaseDiscovery); + } + return response; + }; + SsfStreamClient client = new SsfStreamClient(stuck, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.setStopTimeout(STOP_TIMEOUT); + registrar.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> this.receiverStream.getStreamId() != null); + String earlier = this.receiverStream.getStreamId(); + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + assertThat(this.receiverStream.getStreamId()).isEqualTo(earlier); + + // the second registration creates a stream and stalls; stop() gives it up and the + // third finds that stream, and stalls before it publishes it + stallCreate.set(true); + registrar.start(); + assertThat(created.await(5, TimeUnit.SECONDS)).isTrue(); + String streamId = (String) transmitter.streams().get(0).get("stream_id"); + registrar.stop(); + stallDiscovery.set(true); + registrar.start(); + assertThat(discovered.await(5, TimeUnit.SECONDS)).isTrue(); + + // the second returns now: the third has not published, so the stream is left to + // it, + // not deleted for differing from the stream of the first + releaseCreate.countDown(); + await().during(Duration.ofMillis(300)).until(() -> transmitter.streams().size() == 1); + releaseDiscovery.countDown(); + await().atMost(Duration.ofSeconds(5)).until(() -> streamId.equals(this.receiverStream.getStreamId())); + assertThat(transmitter.streams()).hasSize(1); + assertThat(registrar.getState()).isEqualTo(SsfStreamRegistrar.State.REGISTERED); + + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + } + + @Test + void streamAStoppedRegistrationCreatesAfterTheNextOneLookedIsDeletedOnceTheNextOnePublished() throws Exception { + transmitter.setSeveralStreamsAllowed(true); + // the first registration stalls after it looked and found no stream; the second + // finds none either, creates a stream and stalls before it publishes; the first + // then creates a stream nobody will adopt + CountDownLatch discovered = new CountDownLatch(1); + CountDownLatch releaseDiscovery = new CountDownLatch(1); + AtomicBoolean stallCreate = new AtomicBoolean(); + CountDownLatch created = new CountDownLatch(1); + CountDownLatch releaseCreate = new CountDownLatch(1); + SsfHttpClient stuck = (request) -> { + var response = http.execute(request); + if ("GET".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && discovered.getCount() > 0) { + discovered.countDown(); + awaitIgnoringInterrupts(releaseDiscovery); + } + if ("POST".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && stallCreate.getAndSet(false)) { + created.countDown(); + awaitIgnoringInterrupts(releaseCreate); + } + return response; + }; + SsfStreamClient client = new SsfStreamClient(stuck, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.setStopTimeout(STOP_TIMEOUT); + registrar.start(); + assertThat(discovered.await(5, TimeUnit.SECONDS)).isTrue(); + registrar.stop(); + stallCreate.set(true); + registrar.start(); + assertThat(created.await(5, TimeUnit.SECONDS)).isTrue(); + String published = (String) transmitter.streams().get(0).get("stream_id"); + + // the first creates its stream while the second has not published + releaseDiscovery.countDown(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.streams().size() == 2); + await().during(Duration.ofMillis(300)).until(() -> transmitter.streams().size() == 2); + + // the second publishes its own stream: the other one is an orphan and goes + releaseCreate.countDown(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.streams().size() == 1); + assertThat(transmitter.streams().get(0).get("stream_id")).isEqualTo(published); + assertThat(this.receiverStream.getStreamId()).isEqualTo(published); + + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + } + + private static void awaitIgnoringInterrupts(CountDownLatch latch) { + boolean released = false; + while (!released) { + try { + latch.await(); + released = true; + } + catch (InterruptedException ex) { + // a client that does not care for the interrupt of stop() + } + } + } + @Test void obtainsNewAccessTokenWhenTheTransmitterRejectsTheCurrentOne() { SsfStreamClient client = new SsfStreamClient(http, this.tokenProvider, diff --git a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java index a095049..47f4ecc 100644 --- a/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java +++ b/easyssf-test/src/main/java/org/easyssf/test/TestTransmitter.java @@ -89,6 +89,8 @@ public final class TestTransmitter implements AutoCloseable { private volatile String streamIssuer; + private volatile boolean severalStreamsAllowed; + private volatile String lastUserAgent; private final HttpServer server; @@ -207,7 +209,7 @@ private void handleStreams(HttpExchange exchange, String method, String streamId } } case "POST" -> { - if (!this.streams.isEmpty()) { + if (!this.streams.isEmpty() && !this.severalStreamsAllowed) { respond(exchange, 409, JSONObjectUtils.toJSONString(Map.of("error", "stream_error", "error_description", "Only one stream per receiver is allowed"))); return; @@ -338,6 +340,15 @@ public void expireAccessTokens() { this.validAccessToken = ACCESS_TOKEN + "-" + UUID.randomUUID(); } + /** + * Lets the transmitter create a stream per request. By default it allows one stream + * per receiver and answers a second create request with {@code 409}, as Keycloak + * does. + */ + public void setSeveralStreamsAllowed(boolean severalStreamsAllowed) { + this.severalStreamsAllowed = severalStreamsAllowed; + } + /** * Makes the transmitter create streams with the given issuer instead of its own. */ @@ -465,6 +476,7 @@ public void reset() { this.available = true; this.validAccessToken = ACCESS_TOKEN; this.streamIssuer = null; + this.severalStreamsAllowed = false; } public void publishMetadata(MetadataLocation metadataLocation) { From bcf94d06640e630cfbb88bc74e513faee2d4f6d6 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:49:29 +0100 Subject: [PATCH 3/4] Delete orphaned streams outside the publication lock Signed-off-by: Thomas Darimont --- .../receiver/stream/SsfStreamRegistrar.java | 49 ++++++++++++------- 1 file changed, 31 insertions(+), 18 deletions(-) diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index 3d6d2df..04b5764 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -286,11 +286,7 @@ public void stop() { logger.warn("Could not delete SSF stream " + registeredStreamId + ": " + ex.getMessage()); } } - for (String streamId : unadopted) { - logger.info("Deleting SSF stream " + streamId + ", created by a registration that was stopped before " - + "it could publish it"); - deleteQuietly(streamId); - } + deleteOrphans(unadopted); } finally { this.lifecycle.unlock(); @@ -307,16 +303,19 @@ 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 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 if (registration.created() && this.deleteOnShutdown) { @@ -333,16 +332,14 @@ else if (registration.created() && this.deleteOnShutdown) { + " to the one that followed"); } else if (!streamId.equals(this.receiverStream.getStreamId())) { - logger.info("Deleting SSF stream " + streamId + ", created by a registration that was " - + "stopped before it could publish it; the receiver uses " - + this.receiverStream.getStreamId()); - deleteQuietly(streamId); + orphans.add(streamId); } } } finally { this.publication.unlock(); } + deleteOrphans(orphans); return; } catch (SsfStreamIssuerMismatchException ex) { @@ -397,7 +394,7 @@ else if (!streamId.equals(this.receiverStream.getStreamId())) { */ public SsfStreamConfiguration register() { Registration registration = lookUp(); - publish(registration); + deleteOrphans(publish(registration)); return registration.stream(); } @@ -439,21 +436,24 @@ private Registration lookUp() { } } - 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 publish(Registration registration) { SsfStreamConfiguration stream = registration.stream(); + List 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 { this.receiverStream.setConfiguration(stream); this.publishedBy = Thread.currentThread(); - // a stream a stopped registration created is adopted, if it is this one, or - // deleted, as the receiver uses this one for (String unadopted : this.unadoptedStreams) { if (!unadopted.equals(stream.streamId())) { - logger.info("Deleting SSF stream " + unadopted + ", created by a registration that was stopped " - + "before it could publish it; the receiver uses " + stream.streamId()); - deleteQuietly(unadopted); + orphans.add(unadopted); } } this.unadoptedStreams.clear(); @@ -476,6 +476,19 @@ 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 orphans) { + for (String streamId : orphans) { + logger.info("Deleting SSF stream " + streamId + ", created by a registration that was stopped before it " + + "could publish it"); + deleteQuietly(streamId); + } } private void deleteQuietly(String streamId) { From 12231bade822ec68c16ac8f129b4e667670ed9a1 Mon Sep 17 00:00:00 2001 From: Thomas Darimont Date: Thu, 29 Oct 2026 22:56:29 +0100 Subject: [PATCH 4/4] Keep a registration from adopting a stream being deleted Signed-off-by: Thomas Darimont --- CHANGELOG.md | 5 +- .../receiver/stream/SsfStreamRegistrar.java | 49 ++++++++++++- .../stream/SsfStreamManagementTests.java | 70 +++++++++++++++++++ 3 files changed, 119 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ed959d..940c157 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,8 +44,9 @@ The notable changes of every release. The section of a version is the text of it 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 if it finds - it and deletes it once it publishes another one. `stop()` deletes only a stream published since the - last `stop()`, not again the one that `stop()` deleted before. + 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 diff --git a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java index 04b5764..ea53a8d 100644 --- a/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java +++ b/easyssf-receiver/src/main/java/org/easyssf/receiver/stream/SsfStreamRegistrar.java @@ -75,6 +75,13 @@ public class SsfStreamRegistrar { */ private final List 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 deletingStreams = new HashSet<>(); + /** start() and stop() are idempotent and safe to call from any thread */ private final ReentrantLock lifecycle = new ReentrantLock(); @@ -256,6 +263,7 @@ public void stop() { this.publishedBy = null; unadopted = List.copyOf(this.unadoptedStreams); this.unadoptedStreams.clear(); + this.deletingStreams.addAll(unadopted); } finally { this.publication.unlock(); @@ -335,6 +343,7 @@ else if (!streamId.equals(this.receiverStream.getStreamId())) { orphans.add(streamId); } } + this.deletingStreams.addAll(orphans); } finally { this.publication.unlock(); @@ -409,7 +418,12 @@ private Registration lookUp() { return new Registration(this.streamClient.getStream(this.streamId), "Using", false); } List streams = this.streamClient.getStreams(); - SsfStreamConfiguration existing = streams.stream().filter(this::hasDesiredDelivery).findFirst().orElse(null); + Set 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); } @@ -428,7 +442,7 @@ 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", @@ -449,6 +463,10 @@ private List publish(Registration registration) { // 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) { @@ -457,6 +475,7 @@ private List publish(Registration registration) { } } this.unadoptedStreams.clear(); + this.deletingStreams.addAll(orphans); } finally { this.publication.unlock(); @@ -487,7 +506,31 @@ private void deleteOrphans(List orphans) { for (String streamId : orphans) { logger.info("Deleting SSF stream " + streamId + ", created by a registration that was stopped before it " + "could publish it"); - deleteQuietly(streamId); + 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 deletingStreams() { + this.publication.lock(); + try { + return Set.copyOf(this.deletingStreams); + } + finally { + this.publication.unlock(); } } diff --git a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java index d04bf38..650693f 100644 --- a/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java +++ b/easyssf-receiver/src/test/java/org/easyssf/receiver/stream/SsfStreamManagementTests.java @@ -11,6 +11,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.easyssf.core.SsfDeliveryMethod; import org.easyssf.core.event.SsfEventTypes; @@ -545,6 +546,75 @@ void streamAStoppedRegistrationCreatesAfterTheNextOneLookedIsDeletedOnceTheNextO assertThat(transmitter.streams()).isEmpty(); } + @Test + void streamWhoseDeletionIsUnderWayIsNotAdoptedByTheNextRegistration() throws Exception { + // the first registration stalls after it looked and found no stream; the second + // creates a stream and publishes it; the first then creates an orphan and deletes + // it, + // which the transmitter takes its time for; the registrar is restarted meanwhile + transmitter.setSeveralStreamsAllowed(true); + CountDownLatch discovered = new CountDownLatch(1); + CountDownLatch releaseDiscovery = new CountDownLatch(1); + AtomicReference second = new AtomicReference<>(); + AtomicReference orphan = new AtomicReference<>(); + CountDownLatch deleting = new CountDownLatch(1); + CountDownLatch releaseDelete = new CountDownLatch(1); + SsfHttpClient stuck = (request) -> { + if ("GET".equals(request.method()) && request.uri().getPath().endsWith("/streams") + && discovered.getCount() > 0) { + var response = http.execute(request); + discovered.countDown(); + awaitIgnoringInterrupts(releaseDiscovery); + return response; + } + // the delete of the orphan, the only stream deleted while the second is in + // use + if ("DELETE".equals(request.method()) && second.get() != null + && !request.uri().getQuery().contains(second.get()) && deleting.getCount() > 0) { + orphan.set(request.uri().getQuery().replaceAll(".*stream_id=([^&]*).*", "$1")); + deleting.countDown(); + awaitIgnoringInterrupts(releaseDelete); + } + return http.execute(request); + }; + SsfStreamClient client = new SsfStreamClient(stuck, () -> TestTransmitter.ACCESS_TOKEN, + new SsfTransmitterMetadataResolver(transmitter.issuer(), null, http)); + List registered = new ArrayList<>(); + SsfStreamRegistrar registrar = SsfStreamRegistrar.forManagedStream(client, this.receiverStream, + SsfStreamConfiguration.poll(List.of("CaepSessionRevoked"), null)); + registrar.setDeleteOnShutdown(true); + registrar.setStopTimeout(STOP_TIMEOUT); + registrar.setInitialRetryDelay(Duration.ofMillis(50)); + registrar.addListener(registered::add); + registrar.start(); + assertThat(discovered.await(5, TimeUnit.SECONDS)).isTrue(); + registrar.stop(); + registrar.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> this.receiverStream.getStreamId() != null); + second.set(this.receiverStream.getStreamId()); + + // the first creates its stream, an orphan, and gets stuck deleting it + releaseDiscovery.countDown(); + assertThat(deleting.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(transmitter.streams()).hasSize(2); + + // restarted while the orphan still exists: the third registration finds it and + // must not adopt it, it creates a stream of its own instead + registrar.stop(); + assertThat(transmitter.streams()).hasSize(1); + registrar.start(); + await().atMost(Duration.ofSeconds(5)).until(() -> registered.size() == 2); + String third = this.receiverStream.getStreamId(); + assertThat(third).isNotEqualTo(orphan.get()).isNotEqualTo(second.get()); + + releaseDelete.countDown(); + await().atMost(Duration.ofSeconds(5)).until(() -> transmitter.streams().size() == 1); + assertThat(transmitter.streams().get(0).get("stream_id")).isEqualTo(third); + + registrar.stop(); + assertThat(transmitter.streams()).isEmpty(); + } + private static void awaitIgnoringInterrupts(CountDownLatch latch) { boolean released = false; while (!released) {