[codex] Add workflow up-down-counter lifecycle cleanup

packages/test/src/mock-native-worker.ts

@@ -12,6 +12,7 @@ import { MetricMeterWithComposedTags } from '@temporalio/common/lib/metrics';
 import type { CompiledWorkerOptions, WorkerOptions } from '@temporalio/worker/lib/worker-options';
 import { compileWorkerOptions } from '@temporalio/worker/lib/worker-options';
 import type { WorkflowCreator } from '@temporalio/worker/lib/workflow/interface';
+import { WorkflowMetricsTracker } from '@temporalio/worker/lib/workflow/metrics-tracker';
 import * as activities from './activities';
 
 const sleep = (ms: number): Promise<void> => new Promise((resolve) => setTimeout(resolve, ms));
@@ -182,7 +183,17 @@ export class Worker extends RealWorker {
       taskQueue: opts.taskQueue,
     });
     const nativeWorker = new MockNativeWorker();
-    super(runtime, nativeWorker, workflowCreator, opts, logger, runtime.metricMeter, opts.plugins ?? []);
+    const metricsTracker = new WorkflowMetricsTracker(runtime.metricMeter);
+    super(
+      runtime,
+      nativeWorker,
+      workflowCreator,
+      opts,
+      logger,
+      runtime.metricMeter,
+      metricsTracker,
+      opts.plugins ?? []
+    );
   }
 
   public runWorkflows(...args: Parameters<Worker['workflow$']>): Promise<void> {
@@ -207,6 +218,7 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work
     namespace: options.namespace ?? 'default',
     taskQueue: options.taskQueue ?? 'default',
   });
+  const metricsTracker = new WorkflowMetricsTracker(metricMeter);
   return new Worker(
     {
       async createWorkflow() {
@@ -216,6 +228,6 @@ export function isolateFreeWorker(options: WorkerOptions = defaultOptions): Work
         /* Nothing to destroy */
       },
     },
-    compileWorkerOptions(options, logger, metricMeter)
+    compileWorkerOptions(options, logger, metricsTracker)
   );
 }

packages/test/src/test-metrics-custom.ts

@@ -89,6 +89,13 @@ test('Custom Metrics - Bridge supports works properly (no tags)', async (t) => {
   await assertMetricReported(t, /my_float_histogram_bucket{le="0.05"} 1/);
   await assertMetricReported(t, /my_float_histogram_bucket{le="0.1"} 2/);
   await assertMetricReported(t, /my_float_histogram_bucket{le="1"} 3/);
+
+  // UpDownCounter
+  const upDown = meter.createUpDownCounter!('my-up-down-counter', 'my-up-down-counter-unit', 'my-up-down-counter-description');
+  upDown.add(1);
+  upDown.add(5);
+  upDown.add(-3); // 1 + 5 - 3 = 3
+  await assertMetricReported(t, /my_up_down_counter 3/);
 });
 
 /**
@@ -127,6 +134,11 @@ test('Custom Metrics - Tags composition works properly', async (t) => {
     t,
     /my_float_histogram_bucket{labelA="value-a",labelB="true",labelC="123",labelD="123.456",le="0.5"} 1/
   );
+
+  // UpDownCounter
+  const upDown = meter.createUpDownCounter!('my-up-down-counter', 'my-up-down-counter-unit', 'my-up-down-counter-description');
+  upDown.add(2, { labelA: 'value-a', labelB: true, labelC: 123, labelD: 123.456 });
+  await assertMetricReported(t, /my_up_down_counter{labelA="value-a",labelB="true",labelC="123",labelD="123.456"} 2/);
 });
 
 export async function metricWorksWorkflow(): Promise<void> {
@@ -161,12 +173,18 @@ export async function metricWorksWorkflow(): Promise<void> {
     'workflow-float-gauge-unit',
     'workflow-float-gauge-description'
   );
+  const myUpDownCounterMetric = metricMeter.createUpDownCounter!(
+    'workflow-up-down-counter',
+    'workflow-up-down-counter-unit',
+    'workflow-up-down-counter-description'
+  );
 
   myCounterMetric.add(1);
   myHistogramMetric.record(1);
   myFloatHistogramMetric.record(0.01);
   myGaugeMetric.set(1);
   myFloatGaugeMetric.set(0.1);
+  myUpDownCounterMetric.add(2);
 
   // Pause here, so that we can force replay to a distinct worker
   let signalReceived = false;
@@ -181,6 +199,7 @@ export async function metricWorksWorkflow(): Promise<void> {
   myFloatHistogramMetric.record(0.03);
   myGaugeMetric.set(3);
   myFloatGaugeMetric.set(0.3);
+  myUpDownCounterMetric.add(-1);
 }
 
 test('Metric in Workflow works and are not replayed', async (t) => {
@@ -212,6 +231,7 @@ test('Metric in Workflow works and are not replayed', async (t) => {
   await assertMetricReported(t, /workflow_float_histogram_bucket{[^}]+?,le="0.05"} 2/);
   await assertMetricReported(t, /workflow_gauge{[^}]+} 1/);
   await assertMetricReported(t, /workflow_float_gauge{[^}]+} 0.1/);
+  await assertMetricReported(t, /workflow_up_down_counter{[^}]+} 4/);
 
   const worker2 = await createWorker();
   await worker2.runUntil(async () => {
@@ -224,6 +244,7 @@ test('Metric in Workflow works and are not replayed', async (t) => {
   await assertMetricReported(t, /workflow_float_histogram_bucket{[^}]+?,le="0.05"} 4/);
   await assertMetricReported(t, /workflow_gauge{[^}]+} 3/);
   await assertMetricReported(t, /workflow_float_gauge{[^}]+} 0.3/);
+  await assertMetricReported(t, /workflow_up_down_counter{[^}]+} 2/);
 });
 
 export async function MetricTagsWorkflow(): Promise<void> {

packages/test/src/test-up-down-counter-replay.ts

@@ -0,0 +1,220 @@
+import test from 'ava';
+import type {
+  MetricMeter,
+  MetricTags,
+  MetricUpDownCounter,
+  MetricCounter,
+  MetricHistogram,
+  MetricGauge,
+  NumericMetricValueType,
+} from '@temporalio/common';
+import { WorkflowMetricsTracker } from '@temporalio/worker/lib/workflow/metrics-tracker';
+
+interface RecordedAdd {
+  name: string;
+  value: number;
+  tags: MetricTags;
+}
+
+interface RecordedInstrument {
+  name: string;
+  unit: string | undefined;
+  description: string | undefined;
+}
+
+class FakeUpDownCounter implements MetricUpDownCounter {
+  public readonly kind = 'up-down-counter' as const;
+  public readonly valueType = 'int' as const;
+  constructor(
+    public readonly name: string,
+    public readonly unit: string | undefined,
+    public readonly description: string | undefined,
+    private readonly recorded: RecordedAdd[]
+  ) {}
+  add(value: number, tags: MetricTags = {}): void {
+    this.recorded.push({ name: this.name, value, tags });
+  }
+  withTags(_tags: MetricTags): MetricUpDownCounter {
+    throw new Error('not used in this test');
+  }
+}
+
+class FakeMetricMeter implements MetricMeter {
+  public readonly recorded: RecordedAdd[] = [];
+  public readonly instruments: RecordedInstrument[] = [];
+  createCounter(_n: string, _u?: string, _d?: string): MetricCounter {
+    throw new Error('unused');
+  }
+  createHistogram(_n: string, _v?: NumericMetricValueType, _u?: string, _d?: string): MetricHistogram {
+    throw new Error('unused');
+  }
+  createGauge(_n: string, _v?: NumericMetricValueType, _u?: string, _d?: string): MetricGauge {
+    throw new Error('unused');
+  }
+  createUpDownCounter(name: string, unit?: string, description?: string): MetricUpDownCounter {
+    this.instruments.push({ name, unit, description });
+    return new FakeUpDownCounter(name, unit, description, this.recorded);
+  }
+  withTags(_t: MetricTags): MetricMeter {
+    throw new Error('unused');
+  }
+}
+
+function callUpDownCounterSink(
+  tracker: WorkflowMetricsTracker,
+  runId: string,
+  name: string,
+  netValue: number,
+  attrs: MetricTags = {}
+) {
+  const sinks = tracker.getInjectedSinks();
+  const fn = sinks.__temporal_metrics.addMetricUpDownCounterValue.fn;
+  void fn({ runId } as any, name, undefined, undefined, netValue, attrs);
+}
+
+test('first emission applies full delta', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 5);
+  t.deepEqual(meter.recorded, [{ name: 'inflight', value: 5, tags: {} }]);
+});
+
+test('idempotent re-emission with same net value applies no delta', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 5);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 5);
+  t.deepEqual(meter.recorded, [{ name: 'inflight', value: 5, tags: {} }]);
+});
+
+test('different tag combinations are tracked independently', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { region: 'us' });
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { region: 'eu' });
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 3, { region: 'us' });
+  t.deepEqual(meter.recorded, [
+    { name: 'inflight', value: 1, tags: { region: 'us' } },
+    { name: 'inflight', value: 1, tags: { region: 'eu' } },
+    { name: 'inflight', value: 2, tags: { region: 'us' } },
+  ]);
+});
+
+test('tag keys do not collide when tag values contain separators', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { a: 'b,c=d' });
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 1, { a: 'b', c: 'd' });
+  t.deepEqual(meter.recorded, [
+    { name: 'inflight', value: 1, tags: { a: 'b,c=d' } },
+    { name: 'inflight', value: 1, tags: { a: 'b', c: 'd' } },
+  ]);
+});
+
+test('same metric name with different descriptors creates independent instruments', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  const sinks = tracker.getInjectedSinks();
+  const fn = sinks.__temporal_metrics.addMetricUpDownCounterValue.fn;
+
+  void fn({ runId: 'run-1' } as any, 'inflight', 'items', 'Item count', 1, {});
+  void fn({ runId: 'run-1' } as any, 'inflight', 'bytes', 'Byte count', 2, {});
+
+  t.deepEqual(meter.instruments, [
+    { name: 'inflight', unit: 'items', description: 'Item count' },
+    { name: 'inflight', unit: 'bytes', description: 'Byte count' },
+  ]);
+  t.deepEqual(meter.recorded, [
+    { name: 'inflight', value: 1, tags: {} },
+    { name: 'inflight', value: 2, tags: {} },
+  ]);
+});
+
+test('negative net values produce negative deltas', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', 1);
+  callUpDownCounterSink(tracker, 'run-1', 'inflight', -2);
+  t.deepEqual(meter.recorded, [
+    { name: 'inflight', value: 1, tags: {} },
+    { name: 'inflight', value: -3, tags: {} },
+  ]);
+});
+
+test('notifyWorkflowEvicted undoes contributions for that runId only', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-A', 'inflight', 4);
+  callUpDownCounterSink(tracker, 'run-B', 'inflight', 7);
+  tracker.notifyWorkflowEvicted('run-A');
+  t.deepEqual(meter.recorded, [
+    { name: 'inflight', value: 4, tags: {} },
+    { name: 'inflight', value: 7, tags: {} },
+    { name: 'inflight', value: -4, tags: {} },
+  ]);
+});
+
+test('notifyWorkflowEvicted with no contributions is a no-op', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-A', 'inflight', 0);
+  tracker.notifyWorkflowEvicted('run-A');
+  // The first emission (netValue=0) results in delta=0 → no native add.
+  // Eviction has nothing to undo.
+  t.deepEqual(meter.recorded, []);
+});
+
+test('full lifecycle: emit, evict, re-emit reaches steady state', (t) => {
+  const meter = new FakeMetricMeter();
+  const tracker = new WorkflowMetricsTracker(meter);
+  callUpDownCounterSink(tracker, 'run-A', 'inflight', 1);
+  tracker.notifyWorkflowEvicted('run-A');
+  callUpDownCounterSink(tracker, 'run-A', 'inflight', 1);
+  // Sum of recorded values: 1 + (-1) + 1 = 1
+  const sum = meter.recorded.reduce((acc, r) => acc + r.value, 0);
+  t.is(sum, 1);
+});
+
+test('Counter sink applies values to the underlying meter', (t) => {
+  const counterAdds: RecordedAdd[] = [];
+
+  class FakeCounter {
+    public readonly kind = 'counter' as const;
+    public readonly valueType = 'int' as const;
+    constructor(
+      public readonly name: string,
+      public readonly unit?: string,
+      public readonly description?: string
+    ) {}
+    add(value: number, tags: MetricTags = {}) {
+      counterAdds.push({ name: this.name, value, tags });
+    }
+    withTags(): any {
+      throw new Error('unused');
+    }
+  }
+
+  const meter: MetricMeter = {
+    createCounter: (name, unit, description) => new FakeCounter(name, unit, description),
+    createHistogram: () => {
+      throw new Error('unused');
+    },
+    createGauge: () => {
+      throw new Error('unused');
+    },
+    createUpDownCounter: () => {
+      throw new Error('unused');
+    },
+    withTags: () => {
+      throw new Error('unused');
+    },
+  };
+
+  const tracker = new WorkflowMetricsTracker(meter);
+  const sinks = tracker.getInjectedSinks();
+  sinks.__temporal_metrics.addMetricCounterValue.fn({ runId: 'r' } as any, 'requests', undefined, undefined, 3, {
+    region: 'us',
+  });
+  t.deepEqual(counterAdds, [{ name: 'requests', value: 3, tags: { region: 'us' } }]);
+  t.is(sinks.__temporal_metrics.addMetricCounterValue.callDuringReplay, false);
+});

packages/worker/src/worker-options.ts

@@ -6,7 +6,6 @@ import type {
   ActivityFunction,
   DataConverter,
   LoadedDataConverter,
-  MetricMeter,
   VersioningBehavior,
   WorkerDeploymentVersion,
 } from '@temporalio/common';
@@ -22,7 +21,7 @@ import type { NativeConnection } from './connection';
 import type { CompiledWorkerInterceptors, WorkerInterceptors } from './interceptors';
 import type { Logger } from './logger';
 import { initLoggerSink } from './workflow/logger';
-import { initMetricSink } from './workflow/metrics';
+import type { WorkflowMetricsTracker } from './workflow/metrics-tracker';
 import { Runtime } from './runtime';
 import type { InjectedSinks } from './sinks';
 import { MiB } from './utils';
@@ -920,7 +919,7 @@ export type CompiledWorkerOptionsWithBuildId = CompiledWorkerOptions & {
 function addDefaultWorkerOptions(
   options: WorkerOptions,
   logger: Logger,
-  metricMeter: MetricMeter
+  metricsTracker: WorkflowMetricsTracker
 ): WorkerOptionsWithDefaults {
   const {
     buildId,
@@ -1045,7 +1044,7 @@ function addDefaultWorkerOptions(
     nonStickyToStickyPollRatio: nonStickyToStickyPollRatio ?? 0.2,
     sinks: {
       ...initLoggerSink(logger),
-      ...initMetricSink(metricMeter),
+      ...metricsTracker.getInjectedSinks(),
       // Fix deprecated registration of the 'defaultWorkerLogger' sink
       ...(sinks?.defaultWorkerLogger ? { __temporal_logger: sinks.defaultWorkerLogger } : {}),
       ...sinks,
@@ -1059,7 +1058,7 @@ function addDefaultWorkerOptions(
 export function compileWorkerOptions(
   rawOpts: WorkerOptions,
   logger: Logger,
-  metricMeter: MetricMeter
+  metricsTracker: WorkflowMetricsTracker
 ): CompiledWorkerOptions {
   // Validate sink names to ensure they don't use reserved prefixes/names
   if (rawOpts.sinks) {
@@ -1068,7 +1067,7 @@ export function compileWorkerOptions(
     }
   }
 
-  const opts = addDefaultWorkerOptions(rawOpts, logger, metricMeter);
+  const opts = addDefaultWorkerOptions(rawOpts, logger, metricsTracker);
   if (opts.maxCachedWorkflows !== 0 && opts.maxCachedWorkflows < 2) {
     logger.warn('maxCachedWorkflows must be either 0 (ie. cache is disabled) or greater than 1. Defaulting to 2.');
     opts.maxCachedWorkflows = 2;