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-20260710151514-27b5a126dabe
 	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,11 +98,11 @@ 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
-	github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260205130626-db2a2aab956b // indirect
+	github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260709145319-7782fb89eb16 // indirect
 	github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260528173149-f5b8336b19d9 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/supranational/blst v0.3.16-0.20250831170142-f48500c1fdbe // indirect

chain_capabilities/evm/main.go

@@ -31,8 +31,10 @@ import (
 	"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/durableemitter"
 	"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,11 @@ type capabilityGRPCService struct {
 	capability
 	lggr          logger.Logger
 	limitsFactory limits.Factory
+	// metering is the resolved metering Config (ResourceManagerConfig +
+	// DeploymentIdentity) produced from loop.EnvConfig by
+	// EnvConfig.MeteringConfig at startup. The zero value is valid and
+	// leaves those dimensions empty/disabled.
+	metering resourcemanager.Config
 }
 
 type capability struct {
@@ -62,7 +69,21 @@ 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})
+		// EnvConfig.MeteringConfig is the single, canonical loop-env -> metering
+		// mapping (enable flags, snapshot interval, deployment identity); no
+		// per-main copy of that mapping. The durable emitter is resolved here
+		// and injected, since resourcemanager itself must not reach for the
+		// process-global emitter.
+		var emitter resourcemanager.Emitter
+		if de := durableemitter.GetGlobalEmitter(); de != nil {
+			emitter = de
+		}
+		meteringCfg := s.EnvConfig.MeteringConfig(emitter)
+		return evmcapserver.NewClientServer(&capabilityGRPCService{
+			lggr:          s.Logger,
+			limitsFactory: s.LimitsFactory,
+			metering:      meteringCfg,
+		})
 	}, loop.WithOtelViews(append(consMetrics.MetricViews(), monitoring.MetricViews()...)))
 }
 
@@ -158,11 +179,20 @@ 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 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 := resourcemanager.NewBaseIdentity(c.metering.DeploymentIdentity, trigger.MeteringService, trigger.MeteringResource)
+	if dependencies.CapabilityDonID != 0 {
+		baseIdentity = baseIdentity.WithDonID(strconv.FormatUint(uint64(dependencies.CapabilityDonID), 10))
+	}
+	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, baseIdentity, 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,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,9 +12,34 @@ import (
 )
 
 type filter struct {
-	filterID    string
-	expressions []query.Expression
-	confidence  primitives.ConfidenceLevel
+	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
+	// shared-resource refcount key, so the unregister, cleanup, and snapshot
+	// paths all reuse it from here without the request input. Identical filters
+	// registered by different triggers share one physicalFilterID and are billed
+	// once (a +delta on the 0->1 activation, a -delta on the 1->0 release).
+	physicalFilterID string
+	// reservedAddressCount is the number of filter addresses this filter bills:
+	// the +delta emitted on the physical filter's 0->1 activation, and the
+	// -delta on its 1->0 release, both carry this value. 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 paths reproduce the same identity as the
+	// activation delta 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
+	// 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
 }
 
 type logTriggerState struct {
@@ -38,6 +63,8 @@ type LogTriggerStore interface {
 	Read(triggerID string) (value logTriggerState, ok bool)
 	ReadAll() (values map[string]logTriggerState)
 	Write(triggerID string, value logTriggerState)
+	WriteAndIsFirstForPhysical(triggerID string, value logTriggerState) (firstForPhysical bool)
+	DeleteAndIsLastForPhysical(triggerID, physicalFilterID string) (found, lastForPhysical bool)
 	Update(triggerID string, lastBlock *big.Int, unfinalizedSentEventIDs map[string]*big.Int) error
 	Delete(triggerID string)
 }
@@ -71,6 +98,50 @@ func (cs *logTriggerStore) Write(triggerID string, value logTriggerState) {
 	cs.triggers[triggerID] = value
 }
 
+// WriteAndIsFirstForPhysical writes value under triggerID and reports whether,
+// at the instant of the write, no OTHER trigger already held
+// value.physicalFilterID. A true result is the 0->1 activation of that shared
+// physical filter — the only transition that bills a +delta. Deriving the
+// transition from owned state at operation time keeps the emitter stateless
+// (no ledger). The scan and write are atomic under the store lock so two
+// concurrent registrations of the same physical filter can never both observe
+// zero and double-bill.
+func (cs *logTriggerStore) WriteAndIsFirstForPhysical(triggerID string, value logTriggerState) (firstForPhysical bool) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	firstForPhysical = true
+	for id, existing := range cs.triggers {
+		if id == triggerID {
+			continue
+		}
+		if existing.physicalFilterID == value.physicalFilterID {
+			firstForPhysical = false
+			break
+		}
+	}
+	cs.triggers[triggerID] = value
+	return firstForPhysical
+}
+
+// DeleteAndIsLastForPhysical deletes triggerID and reports whether any remaining
+// trigger still holds physicalFilterID. lastForPhysical is true when none
+// remain — the 1->0 deactivation that bills a -delta. found reports whether the
+// trigger existed. Scan and delete are atomic under the store lock.
+func (cs *logTriggerStore) DeleteAndIsLastForPhysical(triggerID, physicalFilterID string) (found, lastForPhysical bool) {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	_, found = cs.triggers[triggerID]
+	delete(cs.triggers, triggerID)
+	lastForPhysical = true
+	for _, existing := range cs.triggers {
+		if existing.physicalFilterID == physicalFilterID {
+			lastForPhysical = false
+			break
+		}
+	}
+	return found, lastForPhysical
+}
+
 func (cs *logTriggerStore) Update(triggerID string, lastBlock *big.Int, unfinalizedSentEventIDs map[string]*big.Int) error {
 	cs.mu.Lock()
 	defer cs.mu.Unlock()