WIP: adding resource managers for triggers

chain_capabilities/evm/go.mod

@@ -2,17 +2,26 @@ module github.com/smartcontractkit/capabilities/chain_capabilities/evm
 
 go 1.26.2
 
+// Unpublished local stack for SHARED-2709; drop once chainlink-common and
+// chainlink-protos/metering/go are tagged.
+replace (
+	github.com/smartcontractkit/chainlink-common => ../../../chainlink-common
+	github.com/smartcontractkit/chainlink-protos/metering/go => ../../../chainlink-protos/metering/go
+)
+
 require (
 	github.com/ethereum/go-ethereum v1.17.0
 	github.com/google/go-cmp v0.7.0
+	github.com/jonboulle/clockwork v0.5.0
 	github.com/smartcontractkit/capabilities/chain_capabilities/common v0.0.0-20260615195421-fb87220e503f
 	github.com/smartcontractkit/capabilities/libs v0.0.0-20260609124022-2749e4a32bfb
 	github.com/smartcontractkit/chain-selectors v1.0.104
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260619153749-934b00c44d37
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260706203129-712bad3efbc6
 	github.com/smartcontractkit/chainlink-evm v0.3.4-0.20260410162948-2dca02f24e98
 	github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20251022073203-7d8ae8cf67c1
 	github.com/smartcontractkit/chainlink-framework/multinode v0.0.0-20260410144512-ca02ad6ed16a
-	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260618082634-432eb85805e7
+	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260622152157-c8e129347b8b
+	github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260706185759-873029fd9019
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.43.0
 	go.uber.org/zap v1.27.1
@@ -69,7 +78,6 @@ require (
 	github.com/jackc/pgx/v5 v5.9.2 // indirect
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jackpal/go-nat-pmp v1.0.2 // indirect
-	github.com/jonboulle/clockwork v0.5.0 // indirect
 	github.com/klauspost/compress v1.18.2 // indirect
 	github.com/klauspost/cpuid/v2 v2.2.10 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
@@ -90,7 +98,7 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/smartcontractkit/chainlink-common/keystore v1.1.1-0.20260529092756-a94bc8ce96d6 // indirect
-	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260601211238-9f526774fef0 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260626151909-052e55e62e62 // indirect
 	github.com/smartcontractkit/chainlink-framework/chains v0.0.0-20260326122810-b657beadfb57 // indirect
 	github.com/smartcontractkit/chainlink-framework/metrics v0.0.0-20260401162955-be2bc6b5264b // indirect
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b // indirect

chain_capabilities/evm/main.go

@@ -28,11 +28,13 @@ import (
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/trigger"
 	"github.com/smartcontractkit/capabilities/libs/loopserver"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	evmcappb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/evm"
 	evmcapserver "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/evm/server"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop"
+	"github.com/smartcontractkit/chainlink-common/pkg/resourcemanager"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	"github.com/smartcontractkit/chainlink-common/pkg/types"
 	"github.com/smartcontractkit/chainlink-common/pkg/types/core"
@@ -46,6 +48,40 @@ type capabilityGRPCService struct {
 	capability
 	lggr          logger.Logger
 	limitsFactory limits.Factory
+	// metering carries emission toggles and deployment/node identity dimensions
+	// delivered to the plugin process via loop.EnvConfig (set at startup). The
+	// zero value is valid and leaves those dimensions empty/disabled.
+	metering meteringConfig
+}
+
+type meteringConfig struct {
+	meterRecordsEnabled   bool
+	meterSnapshotsEnabled bool
+	deployment            resourcemanager.DeploymentIdentity
+}
+
+func newMeteringConfig(env loop.EnvConfig) meteringConfig {
+	return meteringConfig{
+		meterRecordsEnabled:   env.MeterRecordsEnabled,
+		meterSnapshotsEnabled: env.MeterSnapshotsEnabled,
+		deployment: resourcemanager.DeploymentIdentity{
+			Product:         env.MeterProduct,
+			Tenant:          env.MeterTenant,
+			NumericTenantID: env.MeterNumericTenantID,
+			Environment:     env.MeterEnvironment,
+			Zone:            env.MeterZone,
+			NodeID:          env.MeterNodeID,
+		},
+	}
+}
+
+func (m meteringConfig) resourceManagerConfig() resourcemanager.ResourceManagerConfig {
+	return resourcemanager.ResourceManagerConfig{
+		MeterRecordsEnabled:   m.meterRecordsEnabled,
+		MeterSnapshotsEnabled: m.meterSnapshotsEnabled,
+		Emitter:               beholder.GetEmitter(),
+		SnapshotInterval:      resourcemanager.DefaultSnapshotInterval,
+	}
 }
 
 type capability struct {
@@ -62,7 +98,12 @@ var _ evmcapserver.ClientCapability = &capabilityGRPCService{}
 
 func main() {
 	loopserver.ServeNew(CapabilityName, func(s *loop.Server) loop.StandardCapabilities {
-		return evmcapserver.NewClientServer(&capabilityGRPCService{lggr: s.Logger, limitsFactory: s.LimitsFactory})
+		meteringCfg := newMeteringConfig(s.EnvConfig)
+		return evmcapserver.NewClientServer(&capabilityGRPCService{
+			lggr:          s.Logger,
+			limitsFactory: s.LimitsFactory,
+			metering:      meteringCfg,
+		})
 	}, loop.WithOtelViews(append(consMetrics.MetricViews(), monitoring.MetricViews()...)))
 }
 
@@ -160,9 +201,15 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 
 	// TODO: add org resolver
 	capabilityID := fmt.Sprintf("%s (%d)", c.id, cfg.ChainID)
-	c.triggerService, err = trigger.NewLogTriggerService(evmRelayer, trigger.NewLogTriggerStore(), c.lggr, capabilityID, capabilityDonID, processor, messageBuilder,
+	// The ResourceManager owns the snapshot tick; the LogTriggerService starts it
+	// as a sub-service and registers itself, so it must be configured with a
+	// snapshot interval here. Identity/snapshots are gated by the same metering
+	// env flag as MeterRecords.
+	resourceManager := resourcemanager.NewResourceManager(c.lggr, c.metering.resourceManagerConfig())
+	baseIdentity := newBaseMeteringIdentity(dependencies, c.metering.deployment)
+	c.triggerService, err = trigger.NewLogTriggerService(evmRelayer, trigger.NewLogTriggerStore(), c.lggr, capabilityID, processor, messageBuilder,
 		cfg.LogTriggerPollInterval, cfg.LogTriggerSendChannelBufferSize, cfg.LogTriggerLimitQueryLogSize, c.limitsFactory,
-		dependencies.OrgResolver, dependencies.TriggerEventStore)
+		dependencies.OrgResolver, dependencies.TriggerEventStore, resourceManager, baseIdentity, c.chainSelector)
 	if err != nil {
 		return fmt.Errorf("error when creating trigger: %w", err)
 	}
@@ -197,6 +244,47 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 	return nil
 }
 
+// defaultMeteringProduct is the fallback metering product dimension used when
+// the host did not provide one via loop.EnvConfig (a legacy node or a boot path
+// not yet updated). The other deployment dimensions (environment, zone,
+// node_id) have no meaningful constant and are left empty in that case.
+const defaultMeteringProduct = "cre"
+
+// newBaseMeteringIdentity builds the EVM log trigger's base metering identity.
+// The deployment/node dimensions come from deployment (delivered via
+// loop.EnvConfig); the DON dimension comes from the host-injected
+// CapabilityDonID. It carries the six coarse dimensions plus the service-level
+// resource/resource_type; the per-resource ResourceID is set per emit/snapshot.
+// When CapabilityDonID is 0, the DON identifier is left empty here and resolved per emit from
+// the consumer's WorkflowDonID (see LogTriggerService.resolveDONID).
+func newBaseMeteringIdentity(deps core.StandardCapabilitiesDependencies, deployment resourcemanager.DeploymentIdentity) resourcemanager.ResourceIdentity {
+	product := deployment.Product
+	if product == "" {
+		product = defaultMeteringProduct
+	}
+	var donID string
+	if deps.CapabilityDonID != 0 {
+		donID = strconv.FormatUint(uint64(deps.CapabilityDonID), 10)
+	}
+	var donIdentity *resourcemanager.DonIdentity
+	if donID != "" || deployment.NodeID != "" {
+		donIdentity = &resourcemanager.DonIdentity{
+			DonID:  donID,
+			NodeID: deployment.NodeID,
+		}
+	}
+	return resourcemanager.ResourceIdentity{
+		Product:         product,
+		Tenant:          deployment.Tenant,
+		NumericTenantID: deployment.NumericTenantID,
+		Environment:     deployment.Environment,
+		Zone:            deployment.Zone,
+		Don:             donIdentity,
+		Service:         trigger.MeteringService,
+		ResourcePool:    trigger.MeteringResource,
+	}
+}
+
 func (c *capabilityGRPCService) unmarshalConfig(configStr string) (*config.Config, error) {
 	var cfg config.Config
 	if err := json.Unmarshal([]byte(configStr), &cfg); err != nil {

chain_capabilities/evm/trigger/physical_filter_id.go

@@ -0,0 +1,63 @@
+package trigger
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sort"
+	"strings"
+
+	evmtypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/evm"
+)
+
+// physicalFilterID returns the workflow-independent content identity of an EVM
+// log filter: the lowercase hex SHA-256 over a canonical encoding of the
+// filter's physical matching criteria (chain selector, addresses, event
+// signatures, and positional topic slots). Two filters that match exactly the
+// same on-chain logs hash to the same ID regardless of which workflow or
+// trigger registered them, or of the order their addresses/sigs/topics were
+// supplied. It is used as ResourceIdentity.ResourceID and as the
+// RESERVE/RELEASE event identity so identical filters share one billable
+// physical resource (R4).
+//
+// Canonicalization rules (each rule defeats a source of non-determinism):
+//   - addresses and event sigs are lowercased 0x-prefixed hex and sorted
+//     ascending: the matching set is order-independent;
+//   - topic2/topic3/topic4 are POSITIONAL — a value in topic2 is a different
+//     filter than the same value in topic3 — so each slot is encoded under its
+//     own positional tag, and within a slot the values are sorted ascending;
+//   - the chain selector scopes the hash so identical filters on different
+//     chains stay distinct.
+//
+// The preimage uses "|" as a top-level separator and "," within a set; the
+// per-element hex encodings are fixed-width and contain neither, so the
+// encoding is unambiguous.
+func physicalFilterID(chainSelector string, addresses []evmtypes.Address, eventSigs, topic2, topic3, topic4 []evmtypes.Hash) string {
+	sortedAddrs := make([]string, len(addresses))
+	for i, a := range addresses {
+		sortedAddrs[i] = "0x" + hex.EncodeToString(a[:])
+	}
+	sort.Strings(sortedAddrs)
+
+	canonHashes := func(hs []evmtypes.Hash) string {
+		out := make([]string, len(hs))
+		for i, h := range hs {
+			out[i] = "0x" + hex.EncodeToString(h[:])
+		}
+		sort.Strings(out)
+		return strings.Join(out, ",")
+	}
+
+	// Topic slots are encoded positionally so the same value in different slots
+	// produces a different identity.
+	preimage := strings.Join([]string{
+		"cs=" + chainSelector,
+		"addrs=" + strings.Join(sortedAddrs, ","),
+		"sigs=" + canonHashes(eventSigs),
+		"t2=" + canonHashes(topic2),
+		"t3=" + canonHashes(topic3),
+		"t4=" + canonHashes(topic4),
+	}, "|")
+
+	sum := sha256.Sum256([]byte(preimage))
+	return hex.EncodeToString(sum[:])
+}

chain_capabilities/evm/trigger/store.go

@@ -12,7 +12,26 @@ import (
 )
 
 type filter struct {
-	filterID    string
+	filterID string
+	// physicalFilterID is the workflow-independent content hash of the filter's
+	// physical matching criteria (chain selector + canonicalized addresses,
+	// event sigs, and positional topics). It is the metering ResourceID and the
+	// RESERVE/RELEASE event identity, so the unregister, cleanup, snapshot, and
+	// graceful-close paths all reuse it from here without the request input.
+	physicalFilterID string
+	// reservedAddressCount is the number of filter addresses metered in the
+	// RESERVE record when the filter was registered. The matching RELEASE
+	// must carry the same value, and UnregisterLogTrigger ignores its request
+	// input, so the count is stashed here at registration.
+	reservedAddressCount int64
+	// donID is stashed from the registration RequestMetadata so the
+	// unregister/cleanup/snapshot/close paths can emit a metering record with
+	// the same identity as the RESERVE, without the original request. It is the
+	// resolved metering DON ID string (capability DON, or the consumer
+	// WorkflowDonID fallback when the host did not inject a capability DON);
+	// empty when neither is known.
+	donID       string
+	orgID       string
 	expressions []query.Expression
 	confidence  primitives.ConfidenceLevel
 }