Retry soon while the poll endpoint is not known yet

CHANGELOG.md

@@ -6,6 +6,12 @@ The notable changes of every release. The section of a version is the text of it
 
 ## [Unreleased]
 
+### Fixed
+
+- `SsfPoller` looks again after one second (`setEndpointRetry`) while the poll endpoint of the stream
+  is not known yet, instead of waiting the full interval: an application whose stream was still being
+  registered when the poller started polled for the first time up to an interval later.
+
 ## [0.4.0] - 2026-10-06
 
 One schema change: the column PROCESSED_AT of EASYSSF_PROCESSED_SET becomes

easyssf-receiver-spring-boot-starter/README.md

@@ -207,6 +207,8 @@ easyssf:
   agreed with the transmitter (30 seconds by default); the request waits that long plus a margin. A
   transmitter that answers an empty long poll at once is polled every `poll.interval` nevertheless.
   Whether Keycloak holds poll requests has not been verified; its default remains short polling.
+  Until the stream is registered and its poll endpoint known, the poller looks again every second
+  rather than waiting `poll.interval`, so the first poll follows the registration closely.
   With long polling on, the poller calls the transmitter with the JDK HTTP client even if the
   application has a `RestClient`, because the `RestClient` has no timeout per request
   (`SsfTransmitterFactory.pollHttpClient`); a poller bean of your own, built with

easyssf-receiver/src/main/java/org/easyssf/receiver/poll/SsfPoller.java

@@ -46,6 +46,13 @@
  * instead: the transmitter holds the request until SETs are available or its hold time
  * elapses (section 2.5), and the poller sends the next request as soon as it has handled
  * the response. A backlog is fetched with immediate requests in both modes.
+ *
+ * <p>
+ * Until the poll endpoint of the stream is known, typically because the
+ * {@code SsfStreamRegistrar} is still registering the stream when the poller starts, the
+ * poller looks again every {@link #setEndpointRetry(Duration) endpoint retry} (one
+ * second) rather than waiting the interval: not knowing the endpoint is a condition of
+ * the startup, not a request of the transmitter to slow down.
  */
 public class SsfPoller {
 
@@ -86,6 +93,8 @@ public class SsfPoller {
 
     private Duration initialDelay = Duration.ZERO;
 
+    private Duration endpointRetry = Duration.ofSeconds(1);
+
     private Duration longPollingHold;
 
     private boolean flushOnStop = true;
@@ -181,6 +190,15 @@ public void setInitialDelay(Duration initialDelay) {
      * (section 2.2); the request waits that long plus a margin for the response.
      * {@code null} returns to short polling.
      */
+    /**
+     * How soon the poller looks again while the poll endpoint of the stream is not known
+     * yet, one second by default; never longer than the interval.
+     */
+    public void setEndpointRetry(Duration endpointRetry) {
+        SsfAssert.isTrue(endpointRetry != null && endpointRetry.isPositive(), "endpointRetry must be positive");
+        this.endpointRetry = endpointRetry;
+    }
+
     public void setLongPolling(Duration hold) {
         SsfAssert.isTrue(hold == null || hold.isPositive(), "hold must be positive");
         this.longPollingHold = hold;
@@ -351,7 +369,11 @@ private void run() {
                 long started = System.nanoTime();
                 try {
                     PollResult result = this.polling.tryLock() ? pollLocked() : PollResult.SKIPPED;
-                    if (result.skipped()) {
+                    if (result.endpointUnknown()) {
+                        // the stream is still being registered: look again soon
+                        pause = (this.endpointRetry.compareTo(this.interval) < 0) ? this.endpointRetry : this.interval;
+                    }
+                    else if (result.skipped()) {
                         pause = pauseWhileSkipped();
                     }
                     else if (isLongPolling()) {
@@ -421,7 +443,7 @@ private PollResult poll(boolean hold) {
         URI endpoint = this.endpoint.get();
         if (endpoint == null) {
             logger.debug("Not polling, the poll endpoint of the SSF stream is not known yet");
-            return PollResult.SKIPPED;
+            return PollResult.NO_ENDPOINT;
         }
         if (Instant.now().isBefore(this.pausedUntil)) {
             logger.debug("Not polling, the SSF transmitter asked to wait until " + this.pausedUntil);
@@ -601,10 +623,16 @@ private static Map<String, Object> sets(Map<String, Object> response) {
         return (response.get("sets") instanceof Map<?, ?> sets) ? (Map<String, Object>) sets : Map.of();
     }
 
-    private record PollResult(int fetched, boolean skipped) {
+    private record PollResult(int fetched, boolean skipped, boolean endpointUnknown) {
+
+        PollResult(int fetched, boolean skipped) {
+            this(fetched, skipped, false);
+        }
 
         static final PollResult SKIPPED = new PollResult(0, true);
 
+        static final PollResult NO_ENDPOINT = new PollResult(0, true, true);
+
     }
 
 }

easyssf-receiver/src/test/java/org/easyssf/receiver/poll/SsfPollerTests.java

@@ -4,6 +4,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.easyssf.receiver.event.SsfEventHandler;
 import org.easyssf.receiver.http.JdkSsfHttpClient;
@@ -206,6 +207,26 @@ void doesNotPollWhileThePollEndpointIsUnknown() {
         assertThat(this.handled).isEmpty();
     }
 
+    @Test
+    void pollsSoonAfterTheEndpointBecomesKnownInsteadOfWaitingTheInterval() {
+        AtomicReference<URI> endpoint = new AtomicReference<>();
+        SsfPoller waiting = poller(endpoint::get, TestTransmitter.ACCESS_TOKEN);
+        waiting.setInterval(Duration.ofHours(1));
+        waiting.setEndpointRetry(Duration.ofMillis(100));
+        String jti = queue("session-1");
+        waiting.start();
+        try {
+            // the stream is still being registered: nothing is polled
+            await().during(Duration.ofMillis(300)).untilAsserted(() -> assertThat(this.handled).isEmpty());
+            // registered: the poller notices within the retry, not after the interval
+            endpoint.set(URI.create(transmitter.pollUri()));
+            await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> assertThat(this.handled).containsExactly(jti));
+        }
+        finally {
+            waiting.stop();
+        }
+    }
+
     @Test
     void reportsFailedPollRequest() {
         SsfPoller unauthenticated = poller(() -> URI.create(transmitter.pollUri()), "wrong-token");