fix(core): clear the buffer active bit when the snapshot supplier throws

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

@@ -95,43 +95,47 @@ <T extends DataPointSnapshot> T run(
         expectedCount += observationCount.getAndAdd(bufferActiveBit);
       }
 
-      while (!complete.apply(expectedCount)) {
-        // Wait until all in-flight threads have added their observations to the histogram /
-        // summary.
-        // we can't use a condition here, because the other thread doesn't have a lock as it's on
-        // the fast path.
-        Thread.yield();
-      }
-      result = createResult.get();
-
-      // Signal that the buffer is inactive.
-      long expectedBufferSize = 0;
-      if (reset) {
-        for (AtomicLong observationCount : stripedObservationCounts) {
-          expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit;
-        }
-        reset = false;
-      } else {
-        for (AtomicLong observationCount : stripedObservationCounts) {
-          expectedBufferSize += observationCount.addAndGet(bufferActiveBit);
-        }
-      }
-      expectedBufferSize -= expectedCount;
-
-      appendLock.lock();
       try {
-        while (bufferPos < expectedBufferSize) {
-          // Wait until all in-flight threads have added their observations to the buffer.
-          bufferFilled.await();
+        while (!complete.apply(expectedCount)) {
+          // Wait until all in-flight threads have added their observations to the histogram /
+          // summary.
+          // we can't use a condition here, because the other thread doesn't have a lock as it's on
+          // the fast path.
+          Thread.yield();
         }
+        result = createResult.get();
       } finally {
-        appendLock.unlock();
-      }
+        // Signal that the buffer is inactive. This has to happen even when the block above throws:
+        // the bit is what makes append() buffer instead of record, so a bit left set sends every
+        // later run() into the wait loop above forever, holding runLock and with it the scrape.
+        long expectedBufferSize = 0;
+        if (reset) {
+          for (AtomicLong observationCount : stripedObservationCounts) {
+            expectedBufferSize += observationCount.getAndSet(0) & ~bufferActiveBit;
+          }
+          reset = false;
+        } else {
+          for (AtomicLong observationCount : stripedObservationCounts) {
+            expectedBufferSize += observationCount.addAndGet(bufferActiveBit);
+          }
+        }
+        expectedBufferSize -= expectedCount;
+
+        appendLock.lock();
+        try {
+          while (bufferPos < expectedBufferSize) {
+            // Wait until all in-flight threads have added their observations to the buffer.
+            bufferFilled.await();
+          }
+        } finally {
+          appendLock.unlock();
+        }
 
-      buffer = observationBuffer;
-      bufferSize = bufferPos;
-      observationBuffer = new double[0];
-      bufferPos = 0;
+        buffer = observationBuffer;
+        bufferSize = bufferPos;
+        observationBuffer = new double[0];
+        bufferPos = 0;
+      }
     } catch (InterruptedException e) {
       throw new RuntimeException(e);
     } finally {

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

@@ -1,11 +1,33 @@
 package io.prometheus.metrics.core.metrics;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import org.junit.jupiter.api.Test;
 
 class BufferTest {
 
+  @Test
+  void bufferIsDeactivatedWhenCreateResultThrows() {
+    Buffer buffer = new Buffer();
+    assertThat(buffer.append(1.0)).isFalse();
+
+    assertThatThrownBy(
+            () ->
+                buffer.run(
+                    count -> true,
+                    () -> {
+                      throw new IllegalStateException("failed to create the snapshot");
+                    },
+                    value -> {}))
+        .isInstanceOf(IllegalStateException.class)
+        .hasMessage("failed to create the snapshot");
+
+    // The buffer has to be inactive again. While it stayed active, append() kept buffering
+    // observations instead of recording them, and the next run() never left its wait loop.
+    assertThat(buffer.append(2.0)).isFalse();
+  }
+
   @Test
   void stripeIndexDoesNotOverflowWhenThreadIdNarrowsToIntegerMinValue() {
     assertThat(Buffer.stripeIndex(2_147_483_648L, 3)).isEqualTo(2);