#2897: WIP - ensure the most recent interval tick is always processed, preventing lost checks under overload.

Author: ulischulte

spring-boot-admin-server/src/main/java/de/codecentric/boot/admin/server/services/IntervalCheck.java

@@ -73,10 +73,13 @@ public class IntervalCheck {
 	@Setter
 	private Consumer<Throwable> errorConsumer;
 
+	@Setter
+	private Consumer<Throwable> retryConsumer;
+
 	public IntervalCheck(String name, Function<InstanceId, Mono<Void>> checkFn, Duration interval,
 			Duration minRetention, Duration maxBackoff) {
 		this.name = name;
-		this.errorConsumer = (Throwable throwable) -> log.error("Unexpected error in {}-check", this.name, throwable);
+		this.retryConsumer = (Throwable throwable) -> log.warn("Unexpected error in {}-check", this.name, throwable);
 		this.checkFn = checkFn;
 		this.interval = interval;
 		this.minRetention = minRetention;
@@ -86,22 +89,22 @@ public IntervalCheck(String name, Function<InstanceId, Mono<Void>> checkFn, Dura
 	public void start() {
 		this.scheduler = Schedulers.newSingle(this.name + "-check");
 		this.subscription = Flux.interval(this.interval)
-			// ensure the most recent interval tick is always processed, preventing lost checks under overload.
+			// ensure the most recent interval tick is always processed, preventing
+			// lost checks under overload.
 			.onBackpressureLatest()
 			.doOnSubscribe((s) -> log.debug("Scheduled {}-check every {}", this.name, this.interval))
-			.log(log.getName(), Level.FINEST)
-			.subscribeOn(this.scheduler)
+			.log(log.getName(), Level.FINEST) //
+			.subscribeOn(this.scheduler) //
 			.flatMap((i) -> this.checkAllInstances()) // Allow concurrent check cycles
 														// if previous is slow
 			.retryWhen(createRetrySpec())
-			.subscribe(null, this.errorConsumer);
+			.subscribe(null, (Throwable error) -> log.error("Unexpected error in {}-check", this.name, error));
 	}
 
 	private Retry createRetrySpec() {
-		return Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1)).maxBackoff(maxBackoff).doBeforeRetry((s) -> {
-			log.warn("Unexpected error in {}-check", this.name, s.failure());
-			this.errorConsumer.accept(s.failure());
-		});
+		return Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
+			.maxBackoff(maxBackoff)
+			.doBeforeRetry((s) -> this.retryConsumer.accept(s.failure()));
 	}
 
 	public void markAsChecked(InstanceId instanceId) {

spring-boot-admin-server/src/test/java/de/codecentric/boot/admin/server/services/IntervalCheckTest.java

@@ -18,11 +18,17 @@
 
 import java.time.Duration;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.invocation.InvocationOnMock;
 import reactor.core.publisher.Mono;
 
 import de.codecentric.boot.admin.server.domain.values.InstanceId;
@@ -99,8 +105,6 @@ void should_not_wait_longer_than_maxBackoff() {
 		await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> verify(this.checkFn, atLeast(7)).apply(INSTANCE_ID));
 	}
 
-
-
 	@Test
 	void should_check_after_error() {
 		this.intervalCheck.markAsChecked(INSTANCE_ID);
@@ -126,25 +130,25 @@ void should_not_overflow_when_checks_timeout_randomly() {
 			}
 			else {
 				// Sometimes timeout
-				return Mono.just("slow response").delayElement(CHECK_INTERVAL.plus(Duration.ofMillis(100))).then();
+				return Mono.just("slow response").delayElement(CHECK_INTERVAL.plus(Duration.ofSeconds(1))).then();
 			}
 		}).when(timeoutCheckFn).apply(any());
 
 		IntervalCheck timeoutCheck = new IntervalCheck("overflow-test", timeoutCheckFn, CHECK_INTERVAL, CHECK_INTERVAL,
-				Duration.ofMillis(1000));
+				Duration.ofSeconds(1));
 
-		List<Throwable> errors = new CopyOnWriteArrayList<>();
+		List<Throwable> retryErrors = new CopyOnWriteArrayList<>();
 
-		timeoutCheck.setErrorConsumer(errors::add);
+		timeoutCheck.setRetryConsumer(retryErrors::add);
 		timeoutCheck.markAsChecked(INSTANCE_ID);
 		timeoutCheck.start();
 		try {
 			await().pollDelay(Duration.ofSeconds(5))
-				.until(() -> errors.stream()
+				.until(() -> retryErrors.stream()
 					.noneMatch((Throwable er) -> "OverflowException".equalsIgnoreCase(er.getClass().getSimpleName())));
 
-			assertThat(errors)
-				.noneMatch((Throwable e) -> "OverflowException".equalsIgnoreCase(e.getClass().getSimpleName()));
+			assertThat(retryErrors).noneMatch(
+					(Throwable e) -> "OverflowException".equalsIgnoreCase(e.getCause().getClass().getSimpleName()));
 		}
 		finally {
 			timeoutCheck.stop();
@@ -161,7 +165,7 @@ void should_not_lose_checks_under_backpressure() {
 			.apply(any());
 
 		IntervalCheck slowCheck = new IntervalCheck("backpressure-test", slowCheckFn, CHECK_INTERVAL,
-			Duration.ofMillis(50), Duration.ofSeconds(1));
+				Duration.ofMillis(50), Duration.ofSeconds(1));
 
 		List<Long> checkTimes = new CopyOnWriteArrayList<>();
 		doAnswer((invocation) -> {
@@ -183,9 +187,59 @@ void should_not_lose_checks_under_backpressure() {
 		}
 	}
 
+	@Test
+	void should_not_lose_checks_under_backpressure_latest() {
+		Duration CHECK_INTERVAL = Duration.ofMillis(100);
+
+		@SuppressWarnings("unchecked")
+		Function<InstanceId, Mono<Void>> slowCheckFn = mock(Function.class);
+		doAnswer((invocation) -> Mono.delay(CHECK_INTERVAL.plus(Duration.ofMillis(50))).then()).when(slowCheckFn)
+			.apply(any());
+
+		IntervalCheck slowCheck = new IntervalCheck("backpressure-test", slowCheckFn, CHECK_INTERVAL, CHECK_INTERVAL,
+				Duration.ofSeconds(1));
+
+		// Add multiple instances to increase load and cause drops
+		Set<InstanceId> instanceIds = IntStream.range(0, 50)
+			.mapToObj((i) -> InstanceId.of("Test" + i))
+			.collect(Collectors.toSet());
+
+		instanceIds.forEach(slowCheck::markAsChecked);
+
+		List<Long> checkTimes = new CopyOnWriteArrayList<>();
+		Map<String, List<Long>> checkTimesPerInstance = new ConcurrentHashMap<>();
+
+		doAnswer((invocation) -> {
+			long checkTime = System.currentTimeMillis();
+			String instanceId = instanceIdString(invocation);
+			List<Long> checkTimesList = checkTimesPerInstance.computeIfAbsent(instanceId,
+					(String k) -> new CopyOnWriteArrayList<>());
+			checkTimesList.add(checkTime);
+			checkTimes.add(checkTime);
+			return Mono.empty();
+		}).when(slowCheckFn).apply(any());
+
+		slowCheck.start();
+
+		try {
+			await().atMost(Duration.ofSeconds(2)).until(() -> checkTimes.size() > 50);
+			// With onBackpressureLatest, we should process more checks without drops
+			assertThat(checkTimes).hasSizeGreaterThanOrEqualTo(100);
+			instanceIds.forEach((InstanceId instanceId) -> assertThat(checkTimesPerInstance.get(instanceId.getValue()))
+				.hasSizeGreaterThanOrEqualTo(2)); // Expect more checks
+		}
+		finally {
+			slowCheck.stop();
+		}
+	}
+
 	@AfterEach
 	void tearDown() {
 		this.intervalCheck.stop();
 	}
 
+	private static String instanceIdString(InvocationOnMock invocation) {
+		return invocation.getArguments()[0].toString();
+	}
+
 }