#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

@@ -49,6 +49,7 @@ public class IntervalCheck {
 
 	private final String name;
 
+	@Getter
 	private final Map<InstanceId, Instant> lastChecked = new ConcurrentHashMap<>();
 
 	private final Function<InstanceId, Mono<Void>> checkFn;
@@ -85,7 +86,8 @@ 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)
-			.onBackpressureDrop((tick) -> log.debug("Dropped check tick due to 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)

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

@@ -99,6 +99,8 @@ 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);
@@ -149,6 +151,38 @@ void should_not_overflow_when_checks_timeout_randomly() {
 		}
 	}
 
+	@Test
+	void should_not_lose_checks_under_backpressure() {
+		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,
+			Duration.ofMillis(50), Duration.ofSeconds(1));
+
+		List<Long> checkTimes = new CopyOnWriteArrayList<>();
+		doAnswer((invocation) -> {
+			checkTimes.add(System.currentTimeMillis());
+			return Mono.empty();
+		}).when(slowCheckFn).apply(any());
+
+		slowCheck.markAsChecked(INSTANCE_ID);
+		slowCheck.start();
+
+		try {
+			await().atMost(Duration.ofSeconds(2)).until(() -> checkTimes.size() >= 5);
+			// With onBackpressureLatest, we should have processed multiple checks without
+			// drops
+			assertThat(checkTimes).hasSizeGreaterThanOrEqualTo(5);
+		}
+		finally {
+			slowCheck.stop();
+		}
+	}
+
 	@AfterEach
 	void tearDown() {
 		this.intervalCheck.stop();