adding resource manager for workflows - #22845
Conversation
patrickhuie19
changed the title
055bd9a to
8fd5251
Compare
CORA - Pending ReviewersAll codeowners have approved! ✅ 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
| return fmt.Errorf("failed to start resource manager: %w", err) | ||
| } | ||
| // Register returns a func that unregisters. | ||
| h.rmUnregister = h.resourceManager.Register(h) |
There was a problem hiding this comment.
This logic should move into your wrapper type -- it should accept the workflowArtifactStorage as input and unregister should be called on its Close() method.
| spec.Status = status | ||
| if _, innerErr := h.workflowArtifactsStore.UpsertWorkflowSpec(ctx, spec); innerErr != nil { | ||
| return fmt.Errorf("failed to update workflow spec: %w", innerErr) | ||
| if status == job.WorkflowSpecStatusActive && spec.Workflow == "" { |
There was a problem hiding this comment.
Worth bubbling this up to its own case clause?
| return | ||
| } | ||
| var orgID string | ||
| if h.orgResolver != nil && owner != "" { |
There was a problem hiding this comment.
orgResolver + resourceManager should never be nil if the API is used correctly; I wouldn't try to handle this here if that's true, it feels overly defensive
| } | ||
|
|
||
| if !sourceFetchFailed && len(w.workflowSources) > 0 { | ||
| w.reconcileOrphanedSpecs(ctx, allMetadataIDs) |
There was a problem hiding this comment.
Consider extending generateReconciliationEvents so that it does this bit for you -- i.e. it should detect orphaned rows and emit Delete events for those
These then get picked up + processed by the handler, which no longer requires a running engine
There was a problem hiding this comment.
agreed, thanks
| spec = newSpec | ||
| } else { | ||
| spec.Status = status | ||
| if spec.RegisteredAt == 0 && payload.CreatedAt > 0 { |
There was a problem hiding this comment.
This is basically the same as lines 689 -> 693
Do we need to duplicate the logic? It looks like we'll run it twice for this case
| // folded in once it has been set by the registry syncer. meterIdentity itself | ||
| // is immutable after construction (set via WithIdentity); only the DON id is | ||
| // learned later, so it is read from an atomic and overlaid here. | ||
| func (h *eventHandler) baseIdentity() resourcemanager.ResourceIdentity { |
There was a problem hiding this comment.
This plus ResourceIdentity should be moved into the wrapper type ideally
Also looking at the ResourceManager interface, does a Meterable need to implement ResourceIdentity?
It seems to me that the ResourceManager can be called in one of two ways:
- for metering/usage -> in which case we pass the identity in anyway
- for snapshotting -> in which case GetUtilization is expected to return the identity
So unless I'm missing something, we don't actually need the ResourceIdentity on the Meterable; in which case I wonder if it's worth removing?
There was a problem hiding this comment.
Agreed, i used ResourceIdentity at one point for testing that identities were supplied correctly but it can be stripped out to simplify the interface
| syncerV2.WithLocalSecretOverrides(lggr, cfg.CRE().LocalSecretOverrides()), | ||
| syncerV2.WithShardExecutionGuard(shardOrchestratorClient, shardingEnabled, shardIndex), | ||
| syncerV2.WithShardRoutingSteady(shardRoutingSteady), | ||
| syncerV2.WithResourceManager(resourcemanager.NewResourceManager(lggr, resourcemanager.ResourceManagerConfig{ |
There was a problem hiding this comment.
Here I would create the wrapper type we discussed and then you will effectively have bypassed a layer of abstraction -- there'll be no need for the handler to know anything about snapshotting, identities etc
There was a problem hiding this comment.
agreed, thanks
| // the union of every source's metadata after the loop (see | ||
| // reconcileOrphanedSpecs); a listing failure skips only orphan | ||
| // reconciliation this tick — engines still reconcile. | ||
| persistedSpecs, specsErr := w.handler.ListWorkflowSpecs(ctx) |
There was a problem hiding this comment.
@patrickhuie19 I think you can go one step further with this:
Put this logic in generateReconciliationEvents, and have it generate events that just get processed through the call other events go through (line 983)
There was a problem hiding this comment.
@cedric-cordenier generateReconciliationEvents is called per-source.
Lets say that in the spec_v2 table there are currently two legacy workflows, i.e. workflows created before we added source to the specs_v2 table:
workflow_ID_A - registered in grpc source
workflow_ID_B - registered in contract source
Lets assume we've made it past line 866, where for _, source := range w.workflowSources { we start the loop for each source. generateReconciliationEvents is called in this loop, and we pass source in its arguments.
We call generateReconciliationEvents within the source because if the grpc source were to be unavailable, or the contract source were to be unavailable it allows us to avoid deleting workflows that were not deleted by a customer. This also allows one source to be unavailable while we can still process events from a live source, i.e. the centralized registry goes down but we can still process customer deployment events on the workflow registry.
To call ListWorkflowSpecs in generateReconciliationEvents would require passing a source.
lets say that we assign workflow_ID_A and workflow_ID_B a default source of contract. When we list the specs, and filter for that source, the reconciliation events will see that workflow_ID_B was deployed by the customer, but that workflow_ID_A had been deleted. This would be a bug, so we can't do that.
We could look for a null value of source, and thus exclude all legacy workflows from the reconcile orphan functionality.
Or, we can continue to do what is on the current version of this PR, which is to wait to handle all orphan events until the very end until we've built a union of all sources AND we've seen the each source was healthy.
|
5 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