Keep the pending-acks gauge alive as long as the poller

CHANGELOG.md

@@ -8,6 +8,10 @@ The notable changes of every release. The section of a version is the text of it
 
 ### Fixed
 
+- 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
+  gauge does too. The JDK 27 job of the CI collected it first.
 - The conformance tests start again on GitHub's current Ubuntu runners: their MongoDB is 9.0.4 now,
   pinned by digest. MongoDB 8.0 and newer exit at startup on a Linux kernel from 6.19 up to 7.0.13,
   and before 9.0.4 they took Ubuntu's 7.0 kernels for 7.0.0 (SERVER-121912, SERVER-131779), so the

easyssf-receiver/src/main/java/org/easyssf/receiver/metrics/SsfReceiverMetrics.java

@@ -110,7 +110,9 @@ default void pollCompleted(String transmitter, Duration duration, boolean succes
 
     /**
      * A poller started and tells how to read the number of acknowledgements and error
-     * reports it still owes its transmitter, for a gauge.
+     * reports it still owes its transmitter, for a gauge. The poller keeps a reference to
+     * {@code pendingAcks} for as long as it lives, so an implementation may hold it
+     * weakly, as Micrometer's gauges do, and the gauge goes away with the poller.
      * @param transmitter the issuer of the transmitter, {@code null} if unknown
      * @param pendingAcks reads the current number
      */

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

@@ -10,6 +10,7 @@
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.IntSupplier;
 import java.util.function.Supplier;
 
 import org.easyssf.core.SsfDeliveryMethod;
@@ -131,6 +132,14 @@ public class SsfPoller {
 
     private final ReentrantLock pausing = new ReentrantLock();
 
+    /**
+     * What {@link SsfReceiverMetrics#pollerStarted} gets to read the pending
+     * acknowledgements: a field, not a fresh method reference per start, as Micrometer
+     * refers to the state of a gauge weakly and reports {@code NaN} once it is collected.
+     * The gauge lives as long as the poller.
+     */
+    private final IntSupplier pendingAckCount = this::getPendingAckCount;
+
     private final Condition wokenUp = this.pausing.newCondition();
 
     /** only read and written while holding {@code pausing}, hence not volatile */
@@ -289,7 +298,7 @@ public void start() {
                 return;
             }
             this.running = true;
-            this.metrics.pollerStarted(this.transmitter, this::getPendingAckCount);
+            this.metrics.pollerStarted(this.transmitter, this.pendingAckCount);
             Thread thread = this.threadFactory.newThread(this::run);
             SsfAssert.notNull(thread, "the thread factory returned no thread");
             this.thread = thread;

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

@@ -1,5 +1,6 @@
 package org.easyssf.receiver.poll;
 
+import java.lang.ref.WeakReference;
 import java.net.URI;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -312,6 +313,9 @@ void pendingAcknowledgementsAreAGauge() {
             .tag("transmitter", transmitter.issuer())
             .gauge()
             .value()).isZero();
+        // Micrometer refers to what a gauge reads weakly: it must survive a collection
+        // as long as the poller does, or it reads NaN (seen on the JDK 27 job of the CI)
+        collectGarbage();
         String jti = queue("session-1");
         this.afterHandling = () -> transmitter.setAvailable(false);
         assertThatIllegalStateException().isThrownBy(this.poller::pollNow);
@@ -393,6 +397,26 @@ void pollNowAsksForAnImmediateAnswerEvenWhenLongPolling() {
         assertThat(transmitter.lastPollRequest()).containsEntry("returnImmediately", true);
     }
 
+    /**
+     * Runs a garbage collection that clears weak references, provably:
+     * {@code System.gc()} is only a hint, so this allocates and asks until a weak
+     * reference of its own is cleared, and fails rather than letting a test pass without
+     * the collection.
+     */
+    private static void collectGarbage() {
+        WeakReference<Object> sentinel = new WeakReference<>(new Object());
+        List<byte[]> garbage = new ArrayList<>();
+        await().atMost(Duration.ofSeconds(10)).untilAsserted(() -> {
+            // small arrays, as G1 allocates large ones outside the young generation
+            for (int i = 0; i < 256; i++) {
+                garbage.add(new byte[64 * 1024]);
+            }
+            garbage.clear();
+            System.gc();
+            assertThat(sentinel.get()).as("no garbage collection could be provoked").isNull();
+        });
+    }
+
     private String queue(String sessionId) {
         var claims = transmitter.setClaims("CaepSessionRevoked", opaque(sessionId)).build();
         transmitter.queueSet(transmitter.signSet(claims));