fix: keep late observations out of subsequent collection buffers (#2471)

## Problem and proof

A late appender can increment its stripe during collection A, then read
`activeGeneration` only after A has completed and collection B has
started.

B already includes that observation in `expectedCount`. If the appender
puts it into B's
buffer, B waits for an observation that will only be replayed after B's
wait ends.
This deterministically causes the five-second collection timeout.

The old late-appender test paused after reading the generation, missing
this window.
Moving the existing injected pause hook before that read, without the
fix, reproduced:

```text
BufferTest.lateAppenderCountedByNextGenerationMustNotBeBufferedAgain
Time elapsed: 5.037 s
ExecutionException: IllegalStateException:
Timed out while waiting for in-flight observations.
```

The production hook remains a no-op; the test forces a scheduling window
that real threads can hit.

This has the same failure signature as the histogram failures observed
in
[#2468](#2468) and
[#2441](#2441).
The reproduction proves this race exists, not which interleaving
occurred in any particular CI run.

## Fix

- Record each stripe's absolute observation count at generation
activation, under `appendLock`.
- Under that same lock, reject buffering an observation whose ticket is
already included in the
current generation's starting count. It completes through the existing
direct observation path.
- Reuse a single cutoff array per Buffer: one additional long per
stripe, plus array overhead,
  with no new per-collection allocation or additional fast-path lock.
- Keep cutoffs independent of the reset-adjusted aggregate observation
count.
- Do not change the five-second timeout, buffer capacity, or public API.

## Regression coverage

Strengthen the late-appender test to force the previously untested
window and capture worker
exceptions through futures, with bounded waits and executor cleanup.

Cover both normal collection and a preceding reset. Each case also
verifies that a genuinely new
observation buffers normally, stays out of the current snapshot, replays
exactly once, and appears
in the following collection.

## Validation

- Before fix: deterministic timeout reproduced as above.
- After fix: `BufferTest`, `HistogramTest`, and `SummaryTest` — 50 tests
passed.
- Temporary local JUnit launcher: both regression cases and the original
`HistogramTest.testObserveMultithreaded`, repeated 50 times with each of
`-XX:ActiveProcessorCount=1`, `2`, and `4` — **450 successful test
executions** on JDK 25.
- `mise run lint:fix` — passed; formatter changes retained.
- `mise run build -- -DskipITs=true` — passed. Docker-backed integration
tests skipped because
  Docker is unavailable locally.
- `mise run test` — full unit-test suite passed.
- `./mvnw verify -pl prometheus-metrics-core -am` — passed, including
coverage checks.
- `git diff --check` — passed.

Kept separate from the benchmark-only PR.


## Current scope after update onto main

The original production fix is already present on `main` through
[#2441](#2441) and
[#2468](#2468). This PR
now contains the remaining merge-update and review hardening: explicit
coverage for both generation handoff windows, fail-closed absolute
cutoffs, clearer test seams, and non-masking executor cleanup
assertions. The historical race reproduction and validation context
above are retained.

---------

Signed-off-by: Gregor Zeitlinger <gregor.zeitlinger@grafana.com>

Author: zeitlinger

prometheus-metrics-core/src/main/java/io/prometheus/metrics/core/metrics/Buffer.java

@@ -62,23 +62,34 @@ private static final class Generation {
   private final Condition bufferSpaceAvailable = appendLock.newCondition();
   private final long maxSpinWaitNanos;
   private final int maxBufferSize;
-  private final Runnable beforeAppendLock;
+  // These hooks are test seams only; production buffers use no-op callbacks.
+  private final Runnable beforeGenerationRead;
+  private final Runnable afterGenerationRead;
 
   Buffer() {
-    this(DEFAULT_MAX_SPIN_WAIT_NANOS, DEFAULT_MAX_BUFFER_SIZE, () -> {});
+    this(DEFAULT_MAX_SPIN_WAIT_NANOS, DEFAULT_MAX_BUFFER_SIZE, () -> {}, () -> {});
   }
 
   Buffer(long maxSpinWaitNanos) {
-    this(maxSpinWaitNanos, DEFAULT_MAX_BUFFER_SIZE, () -> {});
+    this(maxSpinWaitNanos, DEFAULT_MAX_BUFFER_SIZE, () -> {}, () -> {});
   }
 
-  Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeAppendLock) {
+  Buffer(long maxSpinWaitNanos, int maxBufferSize, Runnable beforeGenerationRead) {
+    this(maxSpinWaitNanos, maxBufferSize, beforeGenerationRead, () -> {});
+  }
+
+  Buffer(
+      long maxSpinWaitNanos,
+      int maxBufferSize,
+      Runnable beforeGenerationRead,
+      Runnable afterGenerationRead) {
     if (maxBufferSize <= 0) {
       throw new IllegalArgumentException("maxBufferSize must be positive");
     }
     this.maxSpinWaitNanos = maxSpinWaitNanos;
     this.maxBufferSize = maxBufferSize;
-    this.beforeAppendLock = beforeAppendLock;
+    this.beforeGenerationRead = beforeGenerationRead;
+    this.afterGenerationRead = afterGenerationRead;
     stripedObservationCounts = new AtomicLong[Runtime.getRuntime().availableProcessors()];
     generationStartCounts = new long[stripedObservationCounts.length];
     for (int i = 0; i < stripedObservationCounts.length; i++) {
@@ -87,6 +98,7 @@ private static final class Generation {
   }
 
   boolean append(double value) {
+    // Keep the uncontended hot path small enough for the JIT to inline into observations.
     int stripe = stripeIndex(Thread.currentThread().getId(), stripedObservationCounts.length);
     AtomicLong counter = stripedObservationCounts[stripe];
     long count = counter.incrementAndGet();
@@ -97,9 +109,14 @@ boolean append(double value) {
     if ((count & BUFFER_ACTIVE_BIT) == 0) {
       return false;
     }
+    return appendToActiveGeneration(value, stripe, count);
+  }
+
+  private boolean appendToActiveGeneration(double value, int stripe, long count) {
     // Allow tests to pause between allocating an observation ticket and reading the generation.
-    beforeAppendLock.run();
+    beforeGenerationRead.run();
     Generation generation = activeGeneration;
+    afterGenerationRead.run();
     if (generation == null) {
       return false;
     }
@@ -193,7 +210,7 @@ <T extends DataPointSnapshot> T run(
         long total = 0;
         for (int i = 0; i < stripedObservationCounts.length; i++) {
           long count = stripedObservationCounts[i].getAndAdd(BUFFER_ACTIVE_BIT);
-          generationStartCounts[i] = count;
+          generationStartCounts[i] = count & ~BUFFER_ACTIVE_BIT;
           total += count;
         }
         expectedCount = total - observationCountOffset;

prometheus-metrics-core/src/test/java/io/prometheus/metrics/core/metrics/BufferTest.java

@@ -223,15 +223,21 @@ void interruptedAppenderLeavesBoundedBufferWait() throws InterruptedException {
 
   @Test
   void lateAppenderCountedByNextGenerationMustNotBeBufferedAgain() throws Exception {
-    assertLateAppenderHandoff(false);
+    assertLateAppenderHandoff(false, true);
   }
 
   @Test
   void lateAppenderHandoffUsesAbsoluteStripeCountsAfterReset() throws Exception {
-    assertLateAppenderHandoff(true);
+    assertLateAppenderHandoff(true, true);
   }
 
-  private static void assertLateAppenderHandoff(boolean reset) throws Exception {
+  @Test
+  void lateAppenderAfterGenerationReadMustNotBeBufferedAgain() throws Exception {
+    assertLateAppenderHandoff(false, false);
+  }
+
+  private static void assertLateAppenderHandoff(boolean reset, boolean pauseBeforeGenerationRead)
+      throws Exception {
     CountDownLatch firstSnapshotStarted = new CountDownLatch(1);
     CountDownLatch finishFirstSnapshot = new CountDownLatch(1);
     CountDownLatch observationCounted = new CountDownLatch(1);
@@ -240,16 +246,19 @@ private static void assertLateAppenderHandoff(boolean reset) throws Exception {
     AtomicLong completedObservations = new AtomicLong();
     AtomicLong secondExpectedCount = new AtomicLong();
     AtomicBoolean pauseFirstAppender = new AtomicBoolean(true);
+    Runnable pauseHook =
+        () -> {
+          if (pauseFirstAppender.compareAndSet(true, false)) {
+            observationCounted.countDown();
+            awaitLatch(readGeneration);
+          }
+        };
     Buffer buffer =
         new Buffer(
             TimeUnit.SECONDS.toNanos(5),
             16,
-            () -> {
-              if (pauseFirstAppender.compareAndSet(true, false)) {
-                observationCounted.countDown();
-                awaitLatch(readGeneration);
-              }
-            });
+            pauseBeforeGenerationRead ? pauseHook : () -> {},
+            pauseBeforeGenerationRead ? () -> {} : pauseHook);
     if (reset) {
       assertThat(buffer.append(1.0)).isFalse();
       buffer.observeDirect(completedObservations::incrementAndGet);
@@ -329,8 +338,8 @@ private static void assertLateAppenderHandoff(boolean reset) throws Exception {
       finishFirstSnapshot.countDown();
       readGeneration.countDown();
       executor.shutdownNow();
-      assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue();
     }
+    assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).as("executor terminated").isTrue();
   }
 
   private static void awaitLatch(CountDownLatch latch) {