Skip to content

Stop HealthPoller from polling a closed client after cancellation #81

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
atoivonen merged 1 commit into from
Jul 2, 2026
Merged

Stop HealthPoller from polling a closed client after cancellation #81

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions build.sbt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ lazy val `rsocket-consul-java` = (project in file("."))
"org.junit.platform" % "junit-platform-launcher" % "1.14.4" % Test,
"net.aichler" % "jupiter-interface" % JupiterKeys.jupiterVersion.value % Test,
"org.mockito" % "mockito-core" % "5.23.0" % Test,
"io.projectreactor" % "reactor-test" % "3.4.41" % Test,
"org.slf4j" % "slf4j-nop" % "2.0.17" % Test,
)
},
Expand Down
64 changes: 40 additions & 24 deletions src/main/java/com/dwolla/rsocket/consul/HealthPoller.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@
import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

Expand All @@ -19,6 +21,7 @@

public class HealthPoller implements AutoCloseable {
private final int StartingIndex = 0;
private final Duration RetryDelay = Duration.ofSeconds(5);

private final String healthUrl = "%s/v1/health/service/%s?passing=true&index=%d&wait=1m";
private final Gson gson = new Gson();
Expand All @@ -35,32 +38,45 @@ public HealthPoller(HttpClient client, String consulHost) {

public Flux<Set<Address>> start(String serviceName) {
return Flux.create(emitter -> {
Disposable.Swap backoff = Disposables.swap();

@bpholt bpholt Jul 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: could this be final?

emitter.onDispose(backoff);

@bpholt bpholt Jul 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just to confirm my understanding: if an issue comes up while long polling Consul, the retry delay will be registered with this backoff. Normally it will be retried after the delay; however, if the overall Flux is closed before the delay completes, the retry will also be canceled (via a call to backoff.dispose() inside onDispose).

Sanity check: what happens if multiple retries are necessary? It will just keep updating backup with new delays?

@atoivonen atoivonen Jul 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Each poll failure registers a 5s retry via backoff.update(...). If the Flux is cancelled or terminated early, emitter.onDispose(backoff) handles disposing of it so the pending retry never fires.

Great question regarding the multiple retries. Yeah, it will keep updating backoff with the newest delay. A new delay is only scheduled after the previous one fires and fails again, so we will never have more than one pending delay at a time. Disposable.Swap holds a single slot, .update() safely clears out the old (already completed) delay and stores the new one. backoff is always tracking just that one in-flight retry in case it needs to be cancelled.


Recursive<Function<Integer, CompletableFuture<Integer>>> r = new Recursive<>();
r.func =
index ->
client
.get(String.format(healthUrl, consulHost, serviceName, index))
.thenApply(
res -> {
int nextIdx = getNextIdxFrom(res);

if (nextIdx > index) {
lastResponse = getAddressesFrom(res.getBody());
emitter.next(lastResponse);
}

return nextIdx;
})
.whenComplete(
(nextId, th) -> {
if (th != null) {
logger.warn("Got an exception while long polling Consul.", th);
Mono.delay(Duration.ofSeconds(5))
.subscribe(i -> r.func.apply(StartingIndex));
} else {
r.func.apply(nextId);
}
});
index -> {
if (emitter.isCancelled()) {
return CompletableFuture.completedFuture(index);

@bpholt bpholt Jul 1, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we return a completed future containing index and not e.g. a failed future with a CancelationException?

@atoivonen atoivonen Jul 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I went with a completed future because the returned future is never actually used. The recursion is driven by the whenComplete callback calling r.func.apply(...), and all three call discard the returned CompletableFuture. It really only exists to satisfy the Function<Integer, CompletableFuture<Integer>> signature.

It felt like the most straightforward no-op: basically cleanly stopping because the subscription is gone.

@bpholt bpholt Jul 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Gotcha, that makes sense.

I think it might be more clear to use CompletableFuture.failedFuture(new CancellationException()) instead just to avoid any doubt that the future is not used for anything in the normal code path, but it's not a huge deal either way.

}

return client
.get(String.format(healthUrl, consulHost, serviceName, index))
.thenApply(
res -> {
int nextIdx = getNextIdxFrom(res);

if (nextIdx > index) {
lastResponse = getAddressesFrom(res.getBody());
emitter.next(lastResponse);
}

return nextIdx;
})
.whenComplete(
(nextId, th) -> {
if (emitter.isCancelled()) {
return;
}

if (th != null) {
logger.warn("Got an exception while long polling Consul.", th);
backoff.update(
Mono.delay(RetryDelay)
.subscribe(i -> r.func.apply(StartingIndex)));
} else {
r.func.apply(nextId);
}
});
};

r.func.apply(StartingIndex);
});
Expand Down
61 changes: 61 additions & 0 deletions src/test/java/com/dwolla/rsocket/HealthPollerShould.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@
import com.dwolla.rsocket.consul.SimpleResponse;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.test.scheduler.VirtualTimeScheduler;

import java.time.Duration;
import java.util.*;
Expand All @@ -14,7 +16,11 @@
import java.util.stream.Collectors;

import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

class HealthPollerShould {
Expand Down Expand Up @@ -171,4 +177,59 @@ void startOverAtIndexZeroAfterFiveSecondsIfAnExceptionOccurs() throws ExecutionE
assertTrue(output.contains(host1));
assertTrue(output.contains(host2));
}

@Test
void stopPollingOnceTheSubscriptionIsCancelled() {
final String firstUrl =
CONSUL_HOST + "/v1/health/service/" + SERVICE_NAME + "?passing=true&index=0&wait=1m";
final String secondUrl =
CONSUL_HOST
+ "/v1/health/service/"
+ SERVICE_NAME
+ "?passing=true&index="
+ firstIndex
+ "&wait=1m";
final SimpleResponse res = new SimpleResponse(JSON_RES, new ArrayList<>(firstHeaders.entrySet()));
final CompletableFuture<SimpleResponse> firstFuture = new CompletableFuture<>();

when(client.get(firstUrl)).thenReturn(firstFuture);

// The initial request is issued synchronously on subscribe.
final Disposable subscription = poller.start(SERVICE_NAME).subscribe();

// Cancel before the in-flight request resolves. Completing it now runs the
// dependent stages synchronously, so whenComplete observes the cancellation
// and returns instead of rescheduling the next poll.
subscription.dispose();
firstFuture.complete(res);

verify(client, times(1)).get(anyString());
verify(client, never()).get(secondUrl);
}

@Test
void notRestartAfterTheBackoffWhenCancelledDuringTheDelay() {
final VirtualTimeScheduler scheduler = VirtualTimeScheduler.getOrSet();
try {
final String firstUrl =
CONSUL_HOST + "/v1/health/service/" + SERVICE_NAME + "?passing=true&index=0&wait=1m";
final CompletableFuture<SimpleResponse> failedFuture = new CompletableFuture<>();

when(client.get(firstUrl)).thenReturn(failedFuture);

final Disposable subscription = poller.start(SERVICE_NAME).subscribe();

// Exception schedules the 5s backoff on the virtual scheduler.
failedFuture.completeExceptionally(new RuntimeException("Something happened"));

// Cancel during the backoff window, then advance past when it would have fired.
subscription.dispose();
scheduler.advanceTimeBy(Duration.ofSeconds(6));

// The loop stays dead: only the initial index=0 request was ever issued.
verify(client, times(1)).get(anyString());
} finally {
VirtualTimeScheduler.reset();
}
}
}
Loading