Add counter for unaccessed failed promise warnings

CHANGELOG.md

@@ -20,6 +20,8 @@ to docs, or any other relevant information.
 ## [Unreleased]
 
 ### Added
+- Added `temporal_workflow_unaccessed_failed_promise`, counting failed promises reported by the
+  existing unaccessed-promise warning during runner cleanup.
 - Added experimental `ChildWorkflowOptions.Builder.setVersioningOverride` and
   `VersioningOverride.OneTimeVersioningOverride` for explicit pinned, auto-upgrade, and one-time
   child workflow routing. Invalid child overrides are reported as `InvalidVersioningOverrideFailure`

temporal-sdk/src/main/java/io/temporal/internal/sync/DeterministicRunnerImpl.java

@@ -8,6 +8,7 @@
 import io.temporal.internal.context.ContextThreadLocal;
 import io.temporal.internal.worker.WorkflowExecutorCache;
 import io.temporal.serviceclient.CheckedExceptionWrapper;
+import io.temporal.worker.MetricsType;
 import io.temporal.workflow.Promise;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -339,6 +340,10 @@ private void close(boolean fromWorkflowThread) {
               f.get();
               throw new Error("unreachable");
             } catch (RuntimeException e) {
+              workflowContext
+                  .getMetricsScope()
+                  .counter(MetricsType.WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER)
+                  .inc(1);
               log.warn(
                   "Promise completed with exception and was never accessed. The ignored exception:",
                   CheckedExceptionWrapper.unwrap(e));

temporal-sdk/src/main/java/io/temporal/worker/MetricsType.java

@@ -17,6 +17,15 @@ private MetricsType() {}
   public static final String WORKFLOW_FAILED_COUNTER = TEMPORAL_METRICS_PREFIX + "workflow_failed";
   public static final String WORKFLOW_CONTINUE_AS_NEW_COUNTER =
       TEMPORAL_METRICS_PREFIX + "workflow_continue_as_new";
+
+  /**
+   * Counts failed promises reported by the unaccessed-promise warning during runner cleanup,
+   * including cache eviction and Worker shutdown. Replay-time increments are suppressed; separate
+   * runners may report the same failure.
+   */
+  public static final String WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER =
+      TEMPORAL_METRICS_PREFIX + "workflow_unaccessed_failed_promise";
+
   public static final String CORRUPTED_SIGNALS_COUNTER =
       TEMPORAL_METRICS_PREFIX + "corrupted_signals";
 

temporal-sdk/src/test/java/io/temporal/internal/sync/DeterministicRunnerTest.java

@@ -4,6 +4,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.mockito.Mockito.*;
 
+import com.uber.m3.tally.Counter;
 import com.uber.m3.tally.NoopScope;
 import com.uber.m3.tally.RootScopeBuilder;
 import com.uber.m3.tally.Scope;
@@ -1000,4 +1001,42 @@ public void testSupplierCalledMultipleWithoutCaching() {
             });
     d.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS);
   }
+
+  @Test
+  public void testUnaccessedFailedPromiseMetric() {
+    Counter counter = closeRunnerWithFailedPromises(false);
+    verify(counter, times(2)).inc(1);
+  }
+
+  @Test
+  public void testAccessedFailedPromisesDoNotIncrementMetric() {
+    Counter counter = closeRunnerWithFailedPromises(true);
+    verifyNoInteractions(counter);
+  }
+
+  private Counter closeRunnerWithFailedPromises(boolean accessFailures) {
+    Counter counter = mock(Counter.class);
+    Scope scope = mock(Scope.class);
+    when(scope.counter(MetricsType.WORKFLOW_UNACCESSED_FAILED_PROMISE_COUNTER)).thenReturn(counter);
+    SyncWorkflowContext context = spy(DummySyncWorkflowContext.newDummySyncWorkflowContext());
+    doReturn(scope).when(context).getMetricsScope();
+    DeterministicRunner runner =
+        DeterministicRunner.newRunner(
+            threadPool::submit,
+            context,
+            () -> {
+              Promise<Object> read = Workflow.newFailedPromise(new RuntimeException("read"));
+              Promise<Object> handled = Workflow.newFailedPromise(new RuntimeException("handled"));
+              if (accessFailures) {
+                read.getFailure();
+                handled.exceptionally(failure -> null);
+              }
+              WorkflowThread.await("blocked", () -> false);
+            });
+    runner.runUntilAllBlocked(DeterministicRunner.DEFAULT_DEADLOCK_DETECTION_TIMEOUT_MS);
+    assertFalse(runner.isDone());
+    runner.close();
+    runner.close();
+    return counter;
+  }
 }