WIP: adding resource managers for triggers

chain_capabilities/evm/go.mod

@@ -5,14 +5,16 @@ go 1.26.2
 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-20260604174211-7f26071a47e0
+	github.com/smartcontractkit/capabilities/libs v0.0.0-20260609124022-2749e4a32bfb
 	github.com/smartcontractkit/chain-selectors v1.0.106
-	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260714130758-475cb096a3e1
+	github.com/smartcontractkit/chainlink-common v0.11.2-0.20260810181408-53d56f533c59
 	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-20260707195416-ca350beacd4b
+	github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260804191526-b7a850ae7648
+	github.com/smartcontractkit/chainlink-protos/metering/go v0.0.0-20260729184203-90b4cdd48536
 	github.com/stretchr/testify v1.11.1
 	go.opentelemetry.io/otel v1.43.0
 	go.uber.org/zap v1.27.1
@@ -69,7 +71,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 +91,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.20260626151909-052e55e62e62 // indirect
+	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260724142814-45996a1bcb72 // 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
@@ -217,3 +218,5 @@ require (
 )
 
 replace github.com/fbsobreira/gotron-sdk => github.com/smartcontractkit/chainlink-tron/relayer/gotron-sdk v0.0.5-0.20250528121202-292529af39df
+
+replace github.com/smartcontractkit/capabilities/libs => ../../libs

chain_capabilities/evm/main.go

@@ -33,6 +33,7 @@ import (
 	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 +47,10 @@ type capabilityGRPCService struct {
 	capability
 	lggr          logger.Logger
 	limitsFactory limits.Factory
+	// metering is the resolved metering Config (ResourceManagerConfig +
+	// DeploymentIdentity) produced by loop.Server.MeteringConfig at startup.
+	// The zero value is valid and leaves those dimensions empty/disabled.
+	metering resourcemanager.Config
 }
 
 type capability struct {
@@ -62,7 +67,15 @@ 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})
+		// Server.MeteringConfig is the single, canonical loop-env -> metering
+		// mapping (enable flags, snapshot interval, deployment identity); no
+		// per-main copy of that mapping, and no reaching for a process-global
+		// emitter (it injects the server's own durable emitter).
+		return evmcapserver.NewClientServer(&capabilityGRPCService{
+			lggr:          s.Logger,
+			limitsFactory: s.LimitsFactory,
+			metering:      s.MeteringConfig(),
+		})
 	}, loop.WithOtelViews(append(consMetrics.MetricViews(), monitoring.MetricViews()...)))
 }
 
@@ -127,10 +140,12 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 	//   - job-spec boot path: populated when unambiguous, otherwise 0 (e.g. a node
 	//     that belongs to multiple DONs running this capability, or a core node
 	//     that pre-dates CRE-4409).
-	// When it is 0 the trigger service falls back to the consumer workflow's DON
-	// ID (see trigger.NewLogTriggerService). We deliberately do NOT re-resolve it
-	// from the registry here: that lookup cannot disambiguate multi-DON nodes and
-	// would emit a guess instead of the safe workflow-DON fallback. See CRE-4409.
+	// When it is 0, event labels fall back to the consumer workflow's DON ID at
+	// their own call sites, while metering snapshots carry no DON dimension at
+	// all (the workflow DON is never substituted — see triggermeter.DonID). We
+	// deliberately do NOT re-resolve it from the registry here: that lookup
+	// cannot disambiguate multi-DON nodes and would emit a guess instead of the
+	// safe per-call-site degradation. See CRE-4409.
 	capabilityDonID := dependencies.CapabilityDonID
 
 	var scheduler ts.TransmissionScheduler
@@ -158,11 +173,15 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 		return fmt.Errorf("failed to init evm relayer for chainID %d from relayer: %w", cfg.ChainID, err)
 	}
 
-	// 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 wraps it
+	// in a triggermeter (which owns its lifecycle, identity, and snapshot
+	// registration), so it must be configured with a snapshot interval here.
+	resourceManager := resourcemanager.NewResourceManager(c.lggr, c.metering.ResourceManagerConfig)
+	orgResolver := dependencies.OrgResolver
+	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)
+		orgResolver, dependencies.TriggerEventStore, resourceManager, c.metering.DeploymentIdentity, dependencies.CapabilityDonID, c.chainSelector)
 	if err != nil {
 		return fmt.Errorf("error when creating trigger: %w", err)
 	}

chain_capabilities/evm/trigger/physical_filter_id.go

@@ -0,0 +1,62 @@
+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 the snapshot resource_id and dedup key 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,30 @@ 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, so the
+	// snapshot path reuses it from here without the request input. Identical
+	// filters registered by different triggers share one physicalFilterID and
+	// are billed once: the snapshot path dedups on it (see snapshotRows).
+	physicalFilterID string
+	// reservedAddressCount is the number of filter addresses this filter bills:
+	// the physical filter's snapshot level carries this value.
+	// UnregisterLogTrigger ignores its request input, so the count is stashed
+	// here at registration.
+	reservedAddressCount int64
+	// donID is the capability DON ID resolved at registration, stashed so the
+	// snapshot path reproduces the same identity without the original request.
+	// The consumer workflow's DON ID is never substituted for it; it is empty
+	// when the host did not inject a capability DON.
+	donID string
+	// workflowOwner is stored for attribution.
+	workflowOwner string
+	// orgID is the organization ID resolved from workflowOwner at registration
+	// time and stored alongside so that emit and snapshot paths can use it
+	// without a network call.
+	orgID       string
 	expressions []query.Expression
 	confidence  primitives.ConfidenceLevel
 }