Billing: emit compute usage MeterRecord per execution

.changeset/compute-usage-meter-records.md

@@ -0,0 +1,5 @@
+---
+"chainlink": patch
+---
+
+Emit a `cre:workflow:compute` usage MeterRecord per workflow execution when `[Metering].MeterRecordsEnabled` is on. The compute duration is reported by the module that ran the workflow (the guest run for WASM, the enclave-measured run for confidential workflows). #internal

core/scripts/go.mod

@@ -29,7 +29,7 @@ require (
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/chain-selectors v1.0.112
 	github.com/smartcontractkit/chainlink-ccip/chains/evm v0.0.0-20261001213322-a2d686f610ca
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20261014202458-d375ab88face
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20261015152047-f1f2b6d3bd57
 	github.com/smartcontractkit/chainlink-common/keystore v1.3.1-0.20260903141829-ef07b52a737d
 	github.com/smartcontractkit/chainlink-deployments-framework v0.123.3
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20261005112317-b723176adfe8

core/services/cre/cre.go

@@ -982,6 +982,14 @@ func newWorkflowRegistrySyncerV2(
 		handlerOpts = append(handlerOpts, syncerV2.WithSpecMeter(specMeter))
 	}
 
+	// Compute usage records share the [Metering].MeterRecordsEnabled gate.
+	if meterRecordsEnabled {
+		usageIdentity := meterIdentity
+		usageIdentity.Service = resourcemanager.EmittingServiceWorkflowEngine
+		usageIdentity = resourcemanager.WithWorkflowUsagePool(usageIdentity, resourcemanager.ResourceTypeWorkflowCompute)
+		handlerOpts = append(handlerOpts, syncerV2.WithUsageMetering(usageIdentity))
+	}
+
 	mc := capCfg.WorkflowRegistry().ModuleCache()
 	cacheEnabled := mc.Enabled()
 	diskMonitorEnabled := mc.DiskMonitorEnabled() || cacheEnabled

core/services/workflows/syncer/v2/handler.go

@@ -108,6 +108,10 @@ type eventHandler struct {
 	// disabled: its handler-facing methods are nil-receiver safe no-ops.
 	specMeter *SpecMeter
 
+	// meterRecordsEnabled gates the engines' compute usage records.
+	meterRecordsEnabled bool
+	usageIdentity       resourcemanager.ResourceIdentity
+
 	orgResolver    orgresolver.OrgResolver
 	secretsFetcher v2.SecretsFetcher
 	// localSecretOverrides is keyed by owner address; values are secret id -> secret value
@@ -193,6 +197,14 @@ func WithBillingClient(client metering.BillingClient) func(*eventHandler) {
 	}
 }
 
+// WithUsageMetering enables compute usage records on the engines with identity.
+func WithUsageMetering(identity resourcemanager.ResourceIdentity) func(*eventHandler) {
+	return func(e *eventHandler) {
+		e.meterRecordsEnabled = true
+		e.usageIdentity = identity
+	}
+}
+
 // WithSpecMeter supplies the SpecMeter that emits metering.v1.MeterRecord
 // events for the workflow_specs_v2 storage lifecycle. The handler runs it as a
 // sub-service and reports storage transitions through EmitSpecDelta; all other
@@ -1505,6 +1517,8 @@ func (h *eventHandler) newV2EngineConfig(
 		WorkflowRegistryAddress:           h.workflowRegistryAddress,
 		WorkflowRegistryChainSelector:     h.workflowRegistryChainSelector,
 		OrgResolver:                       h.orgResolver,
+		MeterRecordsEnabled:               h.meterRecordsEnabled,
+		UsageIdentity:                     h.usageIdentity,
 		SecretsFetcher:                    h.secretsFetcher,
 		OverrideFetcher:                   h.overrideFetcherForOwner(owner),
 		DebugMode:                         h.debugMode,

core/services/workflows/v2/base_engine.go

@@ -25,8 +25,10 @@ import (
 	commonconfig "github.com/smartcontractkit/chainlink-common/pkg/config"
 	"github.com/smartcontractkit/chainlink-common/pkg/contexts"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink-common/pkg/durableemitter"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/metrics"
+	"github.com/smartcontractkit/chainlink-common/pkg/resourcemanager"
 	"github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/services/orgresolver"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/cresettings"
@@ -111,7 +113,8 @@ type baseEngine struct {
 	// tracer is the OTel tracer for this engine. It's a noop tracer when DebugMode is false.
 	tracer trace.Tracer
 
-	orgID string
+	usageMeter *resourcemanager.ResourceManager
+	orgID      string
 	// orgIDMissingReason records why orgID is empty ("resolver_nil",
 	// "resolver_error", or "empty_response"). It is set once in startWith() and
 	// read by startExecution() to label the org_id_missing counter. It is only
@@ -186,6 +189,17 @@ func newBaseEngine(cfg *EngineConfig, engineMode string) (*baseEngine, logger.Su
 	engine.setLogger(beholderLogger)
 	engine.meterReports = metering.NewReports(cfg.BillingClient, cfg.WorkflowOwner, cfg.WorkflowID, beholderLogger, labelsMap, metricsLabeler, cfg.WorkflowRegistryAddress, cfg.WorkflowRegistryChainSelector, metering.EngineVersionV2)
 	engine.metrics = metricsLabeler
+	if cfg.MeterRecordsEnabled {
+		emitter := cfg.UsageEmitter
+		if emitter == nil {
+			if d := durableemitter.GetGlobalEmitter(); d != nil {
+				emitter = d
+			} else {
+				beholderLogger.Errorw("Compute usage metering enabled but no durable emitter is configured; compute usage records will not be delivered")
+			}
+		}
+		engine.usageMeter = resourcemanager.NewResourceManager(cfg.Lggr, resourcemanager.ResourceManagerConfig{MeterRecordsEnabled: true, Emitter: emitter})
+	}
 	engine.loggerLabels.Store(&labelsMap)
 	engine.localNode.Store(&localNode)
 
@@ -728,13 +742,6 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina
 		}
 	}
 
-	// Track time the guest spends blocked in DON-time host calls so it can be
-	// excluded from metered compute (see compute_metering.go). The wrapper is
-	// always installed; whether the recorded wait is subtracted is gated by
-	// e.cfg.MeterComputeExcludingHostWait below.
-	suspension := &suspensionTracker{}
-	timeProvider = newMeasuredTimeProvider(timeProvider, e.cfg.Clock, suspension)
-
 	// Limit is always usable even on a read failure; err is advisory.
 	moduleExecuteMaxResponseSizeBytes, err := e.cfg.LocalLimiters.ExecutionResponse.Limit(ctx)
 	if err != nil {
@@ -761,7 +768,6 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina
 		base: e, WorkflowExecutionID: executionID, ExecutionTimestamp: executionTimestamp,
 		UserLogChan: userLogChan, TimeProvider: timeProvider, SecretsFetcher: e.secretsFetcher(executionID),
 		executionProfile: newExecutionProfileCollector(),
-		suspension:       suspension,
 	}
 
 	execHelper.initLimiters(e.cfg.LocalLimiters)
@@ -783,25 +789,14 @@ func (e *baseEngine) startExecution(ctx context.Context, event triggers.Coordina
 	endTime := e.cfg.Clock.Now()
 	executionDuration := endTime.Sub(startTime)
 
-	// computeDuration is what we meter as RESOURCE_TYPE_COMPUTE. We start from the
-	// end-to-end wall-clock and subtract time the guest spent blocked in DON-time
-	// host calls: that wait is not compute and is non-deterministic across nodes
-	// (a node whose DON-time request lands in the current consensus round returns
-	// in microseconds; one whose request slips to the next round blocks ~a full
-	// round). Operational latency metrics below intentionally keep using
-	// executionDuration.
+	// Metered compute: the module's reported duration, else wall-clock.
 	computeDuration := executionDuration
-	if hostWait := suspension.total(); hostWait > 0 {
-		computeDuration -= hostWait
-		if computeDuration < 0 {
-			computeDuration = 0
-		}
-		executionLogger.Debugw("Excluding DON-time wait from metered compute",
-			"wallClockMs", executionDuration.Milliseconds(),
-			"hostWaitMs", hostWait.Milliseconds(),
-			"computeMs", computeDuration.Milliseconds())
+	if result != nil && result.ComputeDurationMs != nil {
+		computeDuration = time.Duration(result.GetComputeDurationMs()) * time.Millisecond
 	}
 
+	e.emitComputeUsage(ctx, executionLogger, executionID, computeDuration)
+
 	if isMetering {
 		computeUnit := billing.ResourceType_name[int32(billing.ResourceType_RESOURCE_TYPE_COMPUTE)]
 		mrErr := meteringReport.Settle(computeUnit,
@@ -1146,3 +1141,36 @@ func resolveOrgID(ctx context.Context, resolver orgresolver.OrgResolver, workflo
 	}
 	return resolvedOrg{ID: orgID}
 }
+
+// emitComputeUsage emits the cre:workflow:compute usage MeterRecord for one
+// execution and logs the emission. The log line is a contract consumed by the
+// billing reconciler (fields: executionID, eventID, resourceType, value, orgID)
+// and must stay stable. Fail-open: never returns an error to the execution.
+func (e *baseEngine) emitComputeUsage(ctx context.Context, lggr logger.Logger, executionID string, computeDuration time.Duration) {
+	if e.usageMeter == nil {
+		return
+	}
+	resourceID, err := resourcemanager.WorkflowUsageResourceID(e.cfg.WorkflowID, executionID)
+	if err != nil {
+		lggr.Errorw("Compute usage meter record not emitted", "err", err)
+		return
+	}
+	identity := e.cfg.UsageIdentity
+	if node := e.localNode.Load(); node != nil {
+		identity = identity.WithDonID(strconv.FormatUint(uint64(node.WorkflowDON.ID), 10))
+	}
+	valueMs := computeDuration.Milliseconds()
+	// The capability event id for compute is the execution id itself: one
+	// compute record per execution, identical on every node of the DON.
+	e.usageMeter.EmitUsage(ctx, identity, executionID, valueMs, resourcemanager.UtilizationFields{
+		ResourceType: resourcemanager.ResourceTypeWorkflowCompute,
+		ResourceID:   resourceID,
+		OrgID:        e.orgID,
+	})
+	lggr.Infow("Emitted capability usage meter record",
+		"eventID", executionID,
+		"resourceType", resourcemanager.ResourceTypeWorkflowCompute,
+		"value", valueMs,
+		"orgID", e.orgID,
+	)
+}

core/services/workflows/v2/capability_executor.go

@@ -46,10 +46,6 @@ type ExecutionHelper struct {
 	callCounts         map[limits.Limiter[int]]int
 
 	executionProfile *executionProfileCollector
-
-	// suspension accumulates wall-clock time the guest spent blocked in DON-time
-	// host calls, so it can be excluded from metered compute. See compute_metering.go.
-	suspension *suspensionTracker
 }
 
 func (c *ExecutionHelper) initLimiters(limiters *EngineLimiters) {