adding resource manager for workflows - #22845
Conversation
patrickhuie19
changed the title
055bd9a to
8fd5251
Compare
CORA - Pending Reviewers
Legend: ✅ Approved | ❌ Changes Requested | 💬 Commented | 🚫 Dismissed | ⏳ Pending | ❓ Unknown For more details, see the full review summary. |
|
I see you updated files related to
|
|
✅ No conflicts with other open PRs targeting |
|
|
…re-anchor Phase 4 of the delta-based metering redesign, built against the new chainlink-common resourcemanager API (EmitDelta/EmitUsage, RM-generated event_id, no UtilizationFields.EventID). TOML-only gating (kills split-brain): - Delete env.MeterRecordsEnabled and the CL_METER_RECORDS_ENABLED LOOP pass-through (provably shadowed by AsCmdEnv) + its subtests. - Delete cre.meterRecordsEnabled(); gate the syncer ResourceManager on cfg.Metering().MeterRecordsEnabled()/MeterSnapshotsEnabled(). - Default [Metering].Product = "cre" (accessor + core.toml + CONFIG.md). Dead-plumbing removal: - Remove standardcapabilities.NodeIdentity struct, delegate field/ctor param, and its population in application.go. Identity delivery is loop.EnvConfig. node_id logical name: - loop_registry already sources NodeID from [Metering].NodeID; fix fixtures (csa-pubkey-1 -> clp-cre-wf-zone-a-1) and drop the CSA-pubkey comment. Host-side CapDONID (rule 8): - Delegate already resolves and populates CapabilityDonID on the dependencies handed to capability LOOPs at Initialise; add a test asserting a nonzero CapDONID round-trips (and zero is preserved for the workflow-DON fallback). Syncer v2 metering re-anchor (handler.go): - Emit anchors on artifact persistence: EmitDelta(+1) when a new spec's artifacts are first persisted, EmitDelta(-1) when a real delete removes them. Pause/activate and status-only updates emit nothing. - Delete emitGracefulCloseReleases and its call in close (no lifecycle emits). - GetUtilization enumerates persisted specs (WorkflowSpecsDS.GetWorkflowSpecList) at value 1 per workflow; org resolved from the stored owner via the shared CachingOrgResolver. Records resolve org from payload owner via ResolveOrEmpty. - Default ResourceManager to nil (nil-guarded), reuse one emitSpecDelta helper. Regenerate txtar validate/merge goldens and effective TOMLs for the new Product default and the Tenant/NumericTenantID dimensions. Note: local replace directives point chainlink, core/scripts and deployment at the ../chainlink-common and ../chainlink-protos/metering/go siblings so this branch builds against the finished upstream API; the maintainer drops these and bumps the module pins at merge time.
…w syncer
The common EmitDelta now requires a producer-supplied event_id. The syncer is a
workflow-DON service driven by reconciliation against the shared on-chain
WorkflowRegistry, so every workflow-DON node sees the same event and derives the
identical id via resourcemanager.EventID:
- create (+1): EventID("workflow-spec-register", workflowID, CreatedAt) — the
on-chain CreatedAt is DON-consistent and distinguishes a re-registration from
the original (re-activate-after-delete does not collide).
- delete (-1): EventID("workflow-spec-delete", workflowID). WorkflowDeletedEvent
carries only the workflow ID (no on-chain CreatedAt/block), so repeated
delete cycles of the same workflowID share an event_id; this residual drift is
bounded by snapshot reconciliation. See the returned blocker note.
Scrubs the "UUIDv4" event_id wording and the UUID-format validation from the
metering test (event_id is an opaque string; the only check is empty-vs-nonempty).
Aligns the metering pin to the merged chainlink-common pin.
…w_specs_v2 to eliminate delta drift on register->pause->activate flow within billing dedup time bucket
|
| ) | ||
|
|
||
| require ( |
There was a problem hiding this comment.
Stray indirect block
| ) | |
| require ( |
| // Product is the deployment product identity dimension, e.g. "cre". | ||
| Product *string | ||
| // Tenant is the human-readable tenant name, e.g. "mainline". | ||
| Tenant *string |
There was a problem hiding this comment.
Is this different from the cresettings TenantID?
| Telemetry Telemetry `toml:",omitempty"` | ||
| Metering Metering `toml:",omitempty"` |
There was a problem hiding this comment.
I wonder if just Metering is clear, especially alongside Telemetry. What about something like ResourceMetering? Or would it make sense as CRE.Metering?
| } | ||
| h.emitSpecDelta(ctx, -1, workflowID, owner, resourcemanager.EventID("workflow-spec-delete", parts...)) | ||
| } | ||
| h.cleanupModuleCache(workflowID) |
There was a problem hiding this comment.
// does this belong here or with the caller?
| } | ||
| var orgID string | ||
| if h.orgResolver != nil && spec.WorkflowOwner != "" { | ||
| if resolved, err := h.orgResolver.Get(ctx, spec.WorkflowOwner); err == nil { |
There was a problem hiding this comment.
Pointing out a discrepancy between what this says and how this is being implemented.
// GetUtilization returns the current level of the producer's currently
// active resources, one SnapshotEntry per resource. The manager emits one
// MeterSnapshot per entry.
//
// It is called on the snapshot tick and MUST be a cheap, non-blocking
// read-snapshot of in-memory state: no network, no disk, no lock held
// across I/O. It must tolerate ctx cancellation (returning promptly, and
// nil/empty is acceptable) and tolerate concurrent registration of new
// resources. An empty or nil return is valid and means nothing is currently
// active: no snapshots are emitted, and billing zeroes the resource out by
// its absence from subsequent snapshots.
GetUtilization(ctx context.Context) []SnapshotEntry
This is a gRPC round-trip plus JWT signing that is blocking.
The ResourceManager gives each registrant max(SnapshotInterval/4, 5s) = 15s at the 60s default. At ~100 workflows and ~150ms per lookup, the deadline already falls into fast fail mid-list.
The comment from chainlink-common would imply that there should be a cache here.
Caveat a cache will still miss the first run on a cold start
| var orgID string | ||
| if h.orgResolver != nil && spec.WorkflowOwner != "" { | ||
| if resolved, err := h.orgResolver.Get(ctx, spec.WorkflowOwner); err == nil { | ||
| orgID = resolved |
There was a problem hiding this comment.
Should we log a fail here? Otherwise an empty Org will be silently failing
| } | ||
| var orgID string | ||
| if h.orgResolver != nil && owner != "" { | ||
| if resolved, err := h.orgResolver.Get(ctx, owner); err != nil { |
There was a problem hiding this comment.
Can we get this from context instead of refetching?
Might be a few places we could do this too
| // through setDefaults returns "unset" (the nil-pointer fallback below). | ||
| func (b *meteringConfig) Product() string { | ||
| if b.s.Product == nil { | ||
| return "unset" |
There was a problem hiding this comment.
chainlink-common/pkg/resourcemanager.UnsetProduct ?
| ) | ||
| } | ||
|
|
||
| if !sourceFetchFailed && len(w.workflowSources) > 0 { |
There was a problem hiding this comment.
Might be worth adding some defense here against a source that returns an empty list (which I'm imagining could happen from an RPC having problems). Because if it hits that scenario it will clear all specs and -1 them, even though they might not be orphaned.
Guard could be if allMetadataIDs is empty while specs exist. We expect it won't be empty unless malfunctioning.
| if rowsAffected == 0 { | ||
| return sql.ErrNoRows // No spec deleted | ||
| func (orm *orm) DeleteWorkflowSpec(ctx context.Context, id string) (*job.WorkflowSpec, error) { | ||
| query := `DELETE FROM workflow_specs_v2 WHERE workflow_id = $1 RETURNING *` |
There was a problem hiding this comment.
RETURNING * would be better as explicit columns like the other ORM methods
| } | ||
|
|
||
| // SetWorkflowDon supplies the launcher-resolved workflow DON identity for | ||
| // metering. Called by the registry after WaitForDon, before any event is |
There was a problem hiding this comment.
Can the snapshot run before this is set? I see RM is .start'ed on event handler start
| } | ||
|
|
||
| // syncUsingReconciliationStrategy syncs workflow registry contract state by polling the workflow metadata state and comparing to local state. | ||
| // NOTE: In this mode paused states will be treated as a deleted workflow. Workflows will not be registered as paused. |
There was a problem hiding this comment.
Stale comment now with tombstones
| @@ -875,6 +890,7 @@ | |||
| if w.shardingEnabled { | |||
| filteredWorkflowsMetadata, err = w.filterWorkflowsByShard(ctx, workflows) | |||
There was a problem hiding this comment.
Haven't had a chance to think through this yet, but note to self to double check everything works as expected with sharding moving a workflow off
| // Test_specStorage_StateMachine exercises the storage state machine across | ||
| // (row state × event) transitions, verifying row existence, Status, artifact | ||
| // presence, RegisteredAt, engine-registry contents, and fetch call counts. | ||
| func Test_specStorage_StateMachine(t *testing.T) { |
There was a problem hiding this comment.
Can we also test the new ORM methods against real pgtest like in orm_test.go?
| copy(cp, h.released) | ||
| return cp | ||
| } | ||
|
|
There was a problem hiding this comment.
The orphan sweeping could use more testing
- A source succeeds with an empty list
- One source fails, another succeeds. Nothing verifies it short-circuits the sweep.
- Shard filter error sets the same flag at workflow_registry.go line 893
- Sharding enabled, workflow filtered to another shard. Currently swept with a -1
| @@ -0,0 +1,5 @@ | |||
| -- +goose Up | |||
| ALTER TABLE workflow_specs_v2 ADD COLUMN registered_at bigint NOT NULL DEFAULT 0; | |||
There was a problem hiding this comment.
Is this going to cause a problem if not all nodes update at the same time, and then with the mismatch a workflow gets deleted?
| // event and derives the identical id from it (see the call sites). It is never | ||
| // generated here; an empty eventID is rejected by the ResourceManager. | ||
| func (h *eventHandler) emitSpecDelta(ctx context.Context, delta int64, workflowID, owner, eventID string) { | ||
| if h.resourceManager == nil { |
There was a problem hiding this comment.
h.resourceManager is always non-nil because of https://github.com/smartcontractkit/chainlink/pull/22845/changes#diff-45d214eb19532a6195e33d2e90ee1b2b5534cea9c6ae823135c033ca5e2359b7R1060 even with metering flags off. So this guard doesn't do much
3 participants
Durable resource metering for workflow specs (
workflow-syncer-v2× ResourceManager)SHARED-2712
Requires
TL;DR
This PR makes the v2 workflow registry syncer emit billing-grade
MeterRecorddeltas for workflow-spec storage:+1when a workflow's artifacts are first persisted,-1when they are deleted. Pause and Activate do not produceMeterRecords. To ensure no drift from the deltas (i.e. think of register -> delete -> re-register) we change the ArtifactStorageORM to add aregistered_atfield to the table, sourced from theCreatedAtuint64 on the smartcontract and centralized registry sources. To preserve the previous functionality we still clear out theworkflowandconfigfields. This additional state prevents us from emitting the sameeventIDwithin a dedup window on the billing service.1. What is being metered
One metered unit = one persisted registration generation in
workflow_specs_v2. That row is the node's durable copy of the workflow: theworkflowcolumn holds the complete hex-encoded wasm binary andconfigholds its config (the on-disk module cache and in-memory module LRU are rebuildable caches on top of it).Identity on every record:
service=workflow-syncer-v2,resource_pool=workflow_specs_v2,resource_type=operations,resource_id=workflow_id, plus deployment dimensions (product/tenant/environment/zone/node) from the new[Metering]config anddon_idresolved from the DON notifier. Org ID is resolved fail-open from the workflow owner at emit time.2. Constraints
event_id, so all nodes fielding the same logical transition must derive byte-identical event_ids from DON-consistent inputs only (resourcemanager.EventID). Any locally-derived discriminator (wall clock, block head at tick time, per-node counters) silently breaks dedup and double-counts.WorkflowRegistry.sol,_applyPause/_applyActivatemutate onlystatus;createdAtis set once at registration and never changes. The metadata view the syncer reconciles against has noupdatedAt, no nonce, no status-change block.WorkflowActivatedevent (see the comment ingenerateReconciliationEvents: "we can't tell the difference between an activation and registration without holding state in the db").Given the above constraints, if pause deleted the row and emitted
-1, the subsequent activation would re-persist the row through the create path and be forced to reuse the register event_id.(wfID, on-chain createdAt)is the only DON-consistent material available, and it's identical to the original registration's. The consumer would then drop that+1as a duplicate whenever register → pause → activate happen within one dedup window:There is no id scheme that escapes this without new state. So we hold state: the row survives pause, and deltas are emitted only at generation boundaries, where each fires at most once per
(workflow_id, registered_at)and ids never collide across generations. Zero drift by construction, not by reconciliation.3. Flows
stateDiagram-v2 direction LR [*] --> Absent Absent --> Active: register / activate — fetch + INSERT — emit +1 Active --> Tombstone: pause — one UPDATE sets status=paused and clears binary+config — no delta Tombstone --> Active: activate — refetch + UPSERT via existing create path — no delta Active --> Absent: delete or orphan sweep — DELETE — emit -1 Tombstone --> Absent: delete or orphan sweep — DELETE — emit -1 Active --> Active: status-only flip or redelivered event — no delta Tombstone --> Tombstone: redelivered pause — no-op Absent --> Absent: redelivered pause / delete — no-op, no deltaEvent_idsregistered_atEventID("workflow-spec-register", wfID, onChainCreatedAt)status='paused',workflow='',config=''(one statement)status='active'DELETE … RETURNING registered_atEventID("workflow-spec-delete", wfID, registered_at)— timestamp part omitted when0(pre-migration rows)registered_atis a new column (migration0302) persisting the on-chain registration timestamp. the generation discriminator. Without it, a delete → re-register (same content ⇒ same workflowID) → delete sequence inside one dedup window would collide the two-1s.4. ArtifactStorage Lifecycle Change
Prior to this change
workflowPausedEventdelegates to the delete path. The entire row inworkflow_specs_v2is removed, and activation later refetches the binary/config from the storage service and re-inserts it.Now: pause = drain engine (unchanged
ErrDrainInProgresssemantics) → stop engine → singleUPDATE workflow_specs_v2 SET status='paused', workflow='', config='', updated_at=now()→ module-cache cleanup → engine-registry pop. Same step order as the delete path; only the storage operation differs.I considered keeping the binary/config, but decided against it:
spec.Workflow == ""— an invariant, sincecreateWorkflowSpecalways persists a non-empty hex binary; commented in code) and runs the existingcreateWorkflowSpecfetch/upsert path — the exact same storage-service fetch it performs today, minus the delta. Same failure mode on fetch errors (event retried, engine not started).5. DB changes
0302_workflow_specs_v2_registered_at.sql:ADD COLUMN registered_at bigint NOT NULL DEFAULT 0.0→ their delete-id falls back to the timestamp-less form. Backfill is opportunistic: any event that touches an existing row withregistered_at = 0writes the payload'sCreatedAt. The first post-upgrade boot deliversWorkflowActivatedfor every engine-less active workflow, so fleets converge in one reconciliation pass — before metering can emit anything, because[Metering].MeterRecordsEnableddefaults tofalseand enabling it requires a restart.6. Testing
Storage state machine —
Test_specStorage_StateMachine. Table-driven over (row state: absent | active | paused-tombstone | paused-with-artifacts) × (event: Registered(active), Registered(paused), Activated, Paused, Deleted — each delivered twice). After every step it asserts: row existence, status, payload emptiness,registered_at, engine-registry contents, artifact-fetch call count, returned error.Metering lifecycle —
handler_metering_test.go(reworked).Test_meterRecords_FullLifecycleEmitsExactlyTwoRecords: register(+1) → pause(0) → activate(0) → delete(−1); asserts exactly two records with the exact event_ids,registered_atflowing insert → RETURNING.Test_meterRecords_PauseActivateCycleIsLevelNeutral; redelivered delete emits exactly one −1; drain-deferred delete emits nothing until the successful retry;Test_meterRecords_TransientDeleteErrorThenRetryEmitsOnce(the lost-−1 bug);Test_meterRecords_LegacyRowDeleteUsesFallbackID; reprocessed register yields the identical event_id; org-ID and don_id stamping; disabled ResourceManager emits nothing; emitter failures never fail event handling.Test_meterRecords_FailOpenEquivalence: identical event sequences with the ResourceManager nil / disabled / erroring produce byte-identical row and engine outcomes and zero handler errors — metering cannot alter deployment behavior by construction.Test_meterRecords_SweepReleasesPausedTombstone: deleted-while-paused through the sweep → exactly one −1 with the generation id.Source parity —
Test_handler_SourceParity_PauseActivateCycle. The full register → pause → activate → delete cycle run twice, once with a contract-source identifier and once with a centralized-registry (gRPC) identifier, asserting identical storage effects, engine-registry source scoping, and identical metering records — pinning that mainline and centralized flows cannot diverge.Sweep —
Test_reconcileOrphanedSpecs_ReleasesEnginelessOrphansOnly(+ engine-present orphans skipped; zeroHandledispatches).Deliberately flipped assertions (the complete list — everything else in the existing syncer suite passes unchanged):
Test_workflowDeletedHandlerand drain tests move to the 2-argworkflowDeletedEventsignature; tests asserting "pause deletes the row" now assert "pause tombstones the row"; ORM/store delete tests assert(registeredAt, deleted, err).End-to-end: local CRE environment (this repo's tooling, metering-enabled TOML in the DON configs) — register → pause → activate → delete on both sources; chip-ingress shows exactly one
+1and one-1per generation with event_ids identical across all four workflow nodes; engines healthy throughout; during the paused interval the row is present with an emptyworkflowcolumnReviewer guide
core/services/workflows/syncer/v2/handler.goworkflowPausedEvent(the behavior change),workflowRegisteredEventrestore branch,releaseSpecStorage,stopEngineextraction,emitSpecDeltacore/services/workflows/syncer/v2/workflow_registry.goreconcileOrphanedSpecs,ReleaseOrphanedSpeconevtHandler)core/services/workflows/artifacts/v2/{orm,store}.go, migration0302registered_at,DELETE … RETURNING,PauseWorkflowSpeccore/config/**,core/services/chainlink/**,plugins/loop_registry.go[Metering]config plumbing + LOOP env propagationcore/services/cre/cre.goTest_specStorage_StateMachineandTest_meterRecords_FullLifecycleEmitsExactlyTwoRecords