From 1d03154a18499fe1511b67ca2964c6cce2177eb0 Mon Sep 17 00:00:00 2001 From: Erik Burton Date: Thu, 14 Nov 2024 14:11:32 -0800 Subject: [PATCH 01/10] feat: delete merge queue caches on pr merge (#15234) * feat: delete merge queue caches on pr merge * fix: use env, rename variables --- .github/workflows/delete-caches.yml | 51 ++++++++++++++++++++--------- 1 file changed, 36 insertions(+), 15 deletions(-) diff --git a/.github/workflows/delete-caches.yml b/.github/workflows/delete-caches.yml index 870b6b6d086..aaba0dda623 100644 --- a/.github/workflows/delete-caches.yml +++ b/.github/workflows/delete-caches.yml @@ -16,27 +16,48 @@ jobs: # See also: https://docs.github.com/en/rest/actions/cache?apiVersion=2022-11-28#delete-a-github-actions-cache-for-a-repository-using-a-cache-id actions: write contents: read + env: + REPO: ${{ github.repository }} + PR_NUMBER: ${{ github.event.pull_request.number }} steps: - name: Check out code uses: actions/checkout@v4.1.2 - - name: Cleanup Branch Caches - run: | - gh extension install actions/gh-actions-cache - - REPO=${{ github.repository }} - BRANCH=refs/pull/${{ github.event.pull_request.number }}/merge + - name: Setup gh-actions-cache extension + run: gh extension install actions/gh-actions-cache - echo "Fetching list of cache key" - cacheKeysForPR=$(gh actions-cache list -R $REPO -B $BRANCH | cut -f 1 ) + - name: Retrieve Trunk SHA + id: get-sha + run: | + SHA=$(gh pr view -R $REPO $PR_NUMBER --json mergeCommit --jq .mergeCommit.oid) + echo "sha=$SHA" >> $GITHUB_OUTPUT - ## Setting this to not fail the workflow while deleting cache keys. + - name: Cleanup Caches + env: + TRUNK_SHA: ${{ steps.get-sha.outputs.sha }} + GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | set +e - echo "Deleting caches..." - for cacheKey in $cacheKeysForPR - do - gh actions-cache delete $cacheKey -R $REPO -B $BRANCH --confirm + + PR_BRANCH=refs/pull/$PR_NUMBER/merge + echo "Fetching list of cache keys for the PR branch ($PR_BRANCH)" + PR_CACHE_KEYS=$(gh actions-cache list -R $REPO -B $PR_BRANCH | cut -f 1) + + echo "Deleting caches for PR branch ($PR_BRANCH)..." + for CACHE_KEY in $PR_CACHE_KEYS; do + gh actions-cache delete $CACHE_KEY -R $REPO -B $PR_BRANCH --confirm done + + if [[ -n "$TRUNK_SHA" ]]; then + echo "Found corresponding merge commit $TRUNK_SHA" + QUEUE_BRANCH="gh-readonly-queue/develop/pr-${PR_NUMBER}-${TRUNK_SHA}" + echo "Fetching list of cache keys for the merge queue branch ($QUEUE_BRANCH)" + QUEUE_CACHE_KEYS=$(gh actions-cache list -R $REPO -B $QUEUE_BRANCH | cut -f 1) + + echo "Deleting caches for merge queue branch ($QUEUE_BRANCH)..." + for CACHE_KEY in $QUEUE_CACHE_KEYS; do + gh actions-cache delete $CACHE_KEY -R $REPO -B $QUEUE_BRANCH --confirm + done + fi + echo "Done" - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} \ No newline at end of file From 0240fb8750640619a0862903c6ad5e94c110b8e3 Mon Sep 17 00:00:00 2001 From: Erik Burton Date: Thu, 14 Nov 2024 14:29:10 -0800 Subject: [PATCH 02/10] chore: update goreleaser github action dependencies (#15251) --- .github/actions/goreleaser-build-sign-publish/action.yml | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/.github/actions/goreleaser-build-sign-publish/action.yml b/.github/actions/goreleaser-build-sign-publish/action.yml index 0a70cc479c9..c57bff91488 100644 --- a/.github/actions/goreleaser-build-sign-publish/action.yml +++ b/.github/actions/goreleaser-build-sign-publish/action.yml @@ -33,14 +33,14 @@ runs: name: Set up QEMU uses: docker/setup-qemu-action@49b3bc8e6bdd4a60e6116a5414239cba5943d3cf # v3.2.0 - name: Setup docker buildx - uses: docker/setup-buildx-action@2b51285047da1547ffb1b2203d8be4c0af6b1f20 # v3.2.0 + uses: docker/setup-buildx-action@c47758b77c9736f4b2ef4073d4d51994fabfe349 # v3.7.0 - name: Set up Go uses: ./.github/actions/setup-go with: go-version-file: 'go.mod' only-modules: 'true' - name: Setup goreleaser - uses: goreleaser/goreleaser-action@286f3b13b1b49da4ac219696163fb8c1c93e1200 # v6.0.0 + uses: goreleaser/goreleaser-action@9ed2f89a662bf1735a48bc8557fd212fa902bebf # v6.1.0 with: distribution: goreleaser-pro install-only: true @@ -49,12 +49,12 @@ runs: GORELEASER_KEY: ${{ inputs.goreleaser-key }} - name: Login to docker registry - uses: docker/login-action@e92390c5fb421da1463c202d546fed0ec5c39f20 # v3.1.0 + uses: docker/login-action@9780b0c442fbb1117ed29e0efdff1e18412f7567 # v3.3.0 with: registry: ${{ inputs.docker-registry }} - name: Install syft - uses: anchore/sbom-action/download-syft@61119d458adab75f756bc0b9e4bde25725f86a7a # v0.17.2 + uses: anchore/sbom-action/download-syft@fc46e51fd3cb168ffb36c6d1915723c47db58abb # v0.17.7 - name: Run goreleaser release shell: bash From 1a80fec3917b7b96d95d85dee2ccdc17f88f27ca Mon Sep 17 00:00:00 2001 From: Bolek <1416262+bolekk@users.noreply.github.com> Date: Thu, 14 Nov 2024 15:21:15 -0800 Subject: [PATCH 03/10] Prune workfllw DB entries (#15226) --- core/services/chainlink/application.go | 1 + core/services/workflows/delegate.go | 1 + core/services/workflows/store/store_db.go | 72 +++++++++++++++++++++-- core/web/testdata/body/health.html | 4 ++ core/web/testdata/body/health.json | 9 +++ core/web/testdata/body/health.txt | 1 + testdata/scripts/health/default.txtar | 10 ++++ testdata/scripts/health/multi-chain.txtar | 10 ++++ 8 files changed, 104 insertions(+), 4 deletions(-) diff --git a/core/services/chainlink/application.go b/core/services/chainlink/application.go index 0b2352f67d4..abbe9dad9ab 100644 --- a/core/services/chainlink/application.go +++ b/core/services/chainlink/application.go @@ -400,6 +400,7 @@ func NewApplication(opts ApplicationOpts) (Application, error) { streamRegistry = streams.NewRegistry(globalLogger, pipelineRunner) workflowORM = workflowstore.NewDBStore(opts.DS, globalLogger, clockwork.NewRealClock()) ) + srvcs = append(srvcs, workflowORM) promReporter := headreporter.NewPrometheusReporter(opts.DS, legacyEVMChains) chainIDs := make([]*big.Int, legacyEVMChains.Len()) diff --git a/core/services/workflows/delegate.go b/core/services/workflows/delegate.go index 72aff3033d0..1db26729ca6 100644 --- a/core/services/workflows/delegate.go +++ b/core/services/workflows/delegate.go @@ -75,6 +75,7 @@ func (d *Delegate) ServicesForSpec(ctx context.Context, spec job.Job) ([]job.Ser if err != nil { return nil, err } + d.logger.Infow("Creating Workflow Engine for workflow spec", "workflowID", spec.WorkflowSpec.WorkflowID, "workflowOwner", spec.WorkflowSpec.WorkflowOwner, "workflowName", spec.WorkflowSpec.WorkflowName, "jobName", spec.Name) return []job.ServiceCtx{engine}, nil } diff --git a/core/services/workflows/store/store_db.go b/core/services/workflows/store/store_db.go index 926dd06d09b..f15a6928e7e 100644 --- a/core/services/workflows/store/store_db.go +++ b/core/services/workflows/store/store_db.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "sync" "time" "google.golang.org/protobuf/proto" @@ -15,17 +16,31 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/values" valuespb "github.com/smartcontractkit/chainlink-common/pkg/values/pb" + commonservices "github.com/smartcontractkit/chainlink-common/pkg/services" "github.com/smartcontractkit/chainlink/v2/core/logger" + "github.com/smartcontractkit/chainlink/v2/core/services" +) + +const ( + defaultPruneFrequencySec = 20 + defaultPruneTimeoutSec = 60 + defaultPruneRecordAgeHours = 3 + defaultPruneBatchSize = 3000 ) // `DBStore` is a postgres-backed // data store that persists workflow progress. type DBStore struct { - lggr logger.Logger - db sqlutil.DataSource - clock clockwork.Clock + commonservices.StateMachine + lggr logger.Logger + db sqlutil.DataSource + shutdownWaitGroup sync.WaitGroup + chStop commonservices.StopChan + clock clockwork.Clock } +var _ services.ServiceCtx = (*DBStore)(nil) + // `workflowExecutionRow` describes a row // of the `workflow_executions` table type workflowExecutionRow struct { @@ -70,6 +85,47 @@ type workflowExecutionWithStep struct { WEFinishedAt *time.Time `db:"we_finished_at"` } +func (d *DBStore) Start(context.Context) error { + return d.StartOnce("DBStore", func() error { + d.shutdownWaitGroup.Add(1) + go d.pruneDBEntries() + return nil + }) +} + +func (d *DBStore) Close() error { + return d.StopOnce("DBStore", func() error { + close(d.chStop) + d.shutdownWaitGroup.Wait() + return nil + }) +} + +func (d *DBStore) pruneDBEntries() { + defer d.shutdownWaitGroup.Done() + ticker := time.NewTicker(defaultPruneFrequencySec * time.Second) + defer ticker.Stop() + for { + select { + case <-d.chStop: + return + case <-ticker.C: + ctx, cancel := d.chStop.CtxWithTimeout(defaultPruneTimeoutSec * time.Second) + err := sqlutil.TransactDataSource(ctx, d.db, nil, func(tx sqlutil.DataSource) error { + stmt := fmt.Sprintf("DELETE FROM workflow_executions WHERE (id) IN (SELECT id FROM workflow_executions WHERE (created_at < now() - interval '%d hours') LIMIT %d);", defaultPruneRecordAgeHours, defaultPruneBatchSize) + _, err := tx.ExecContext(ctx, stmt) + return err + }) + if err != nil { + d.lggr.Errorw("Failed to prune workflow_executions", "err", err) + } else { + d.lggr.Infow("Pruned oldest workflow_executions", "batchSize", defaultPruneBatchSize, "ageLimitHours", defaultPruneRecordAgeHours) + } + cancel() + } + } +} + // `UpdateStatus` updates the status of the given workflow execution func (d *DBStore) UpdateStatus(ctx context.Context, executionID string, status string) error { sql := `UPDATE workflow_executions SET status = $1, updated_at = $2 WHERE id = $3` @@ -407,5 +463,13 @@ func (d *DBStore) GetUnfinished(ctx context.Context, workflowID string, offset, } func NewDBStore(ds sqlutil.DataSource, lggr logger.Logger, clock clockwork.Clock) *DBStore { - return &DBStore{db: ds, lggr: lggr.Named("WorkflowDBStore"), clock: clock} + return &DBStore{db: ds, lggr: lggr.Named("WorkflowDBStore"), clock: clock, chStop: make(chan struct{})} +} + +func (d *DBStore) HealthReport() map[string]error { + return map[string]error{d.Name(): d.Healthy()} +} + +func (d *DBStore) Name() string { + return d.lggr.Name() } diff --git a/core/web/testdata/body/health.html b/core/web/testdata/body/health.html index 2bf427f5e00..4692d452a5b 100644 --- a/core/web/testdata/body/health.html +++ b/core/web/testdata/body/health.html @@ -156,3 +156,7 @@
TelemetryManager
+
+ WorkflowDBStore +
+ diff --git a/core/web/testdata/body/health.json b/core/web/testdata/body/health.json index d573e0bd5fc..8c4c3b312de 100644 --- a/core/web/testdata/body/health.json +++ b/core/web/testdata/body/health.json @@ -287,6 +287,15 @@ "status": "passing", "output": "" } + }, + { + "type": "checks", + "id": "WorkflowDBStore", + "attributes": { + "name": "WorkflowDBStore", + "status": "passing", + "output": "" + } } ] } diff --git a/core/web/testdata/body/health.txt b/core/web/testdata/body/health.txt index fde038dfc63..a098f906146 100644 --- a/core/web/testdata/body/health.txt +++ b/core/web/testdata/body/health.txt @@ -31,3 +31,4 @@ ok StarkNet.Baz.Chain ok StarkNet.Baz.Relayer ok StarkNet.Baz.Txm ok TelemetryManager +ok WorkflowDBStore \ No newline at end of file diff --git a/testdata/scripts/health/default.txtar b/testdata/scripts/health/default.txtar index a7db2308e35..73b82bc7e39 100644 --- a/testdata/scripts/health/default.txtar +++ b/testdata/scripts/health/default.txtar @@ -41,6 +41,7 @@ ok PipelineRunner ok PipelineRunner.BridgeCache ok RetirementReportCache ok TelemetryManager +ok WorkflowDBStore -- out.json -- { @@ -134,6 +135,15 @@ ok TelemetryManager "status": "passing", "output": "" } + }, + { + "type": "checks", + "id": "WorkflowDBStore", + "attributes": { + "name": "WorkflowDBStore", + "status": "passing", + "output": "" + } } ] } diff --git a/testdata/scripts/health/multi-chain.txtar b/testdata/scripts/health/multi-chain.txtar index f53bbfebf8c..d3a0caf67b5 100644 --- a/testdata/scripts/health/multi-chain.txtar +++ b/testdata/scripts/health/multi-chain.txtar @@ -101,6 +101,7 @@ ok StarkNet.Baz.Chain ok StarkNet.Baz.Relayer ok StarkNet.Baz.Txm ok TelemetryManager +ok WorkflowDBStore -- out-unhealthy.txt -- ! EVM.1.HeadTracker.HeadListener @@ -396,6 +397,15 @@ ok TelemetryManager "status": "passing", "output": "" } + }, + { + "type": "checks", + "id": "WorkflowDBStore", + "attributes": { + "name": "WorkflowDBStore", + "status": "passing", + "output": "" + } } ] } From 93ee17772daeab9d03c356b4b3df1019f1d35c5c Mon Sep 17 00:00:00 2001 From: Cedric Date: Fri, 15 Nov 2024 11:28:12 +0000 Subject: [PATCH 04/10] [chore] Small fixes (#15230) * [chore] Small fixes - Ensure we start the handler - Ensure fetch message ID is < 128 * Fix tests --- core/capabilities/compute/compute.go | 1 - core/capabilities/compute/compute_test.go | 1 - core/services/standardcapabilities/delegate.go | 2 +- 3 files changed, 1 insertion(+), 3 deletions(-) diff --git a/core/capabilities/compute/compute.go b/core/capabilities/compute/compute.go index 78a4cc1e033..cdb8ee19a6a 100644 --- a/core/capabilities/compute/compute.go +++ b/core/capabilities/compute/compute.go @@ -294,7 +294,6 @@ func (c *Compute) createFetcher() func(ctx context.Context, req *wasmpb.FetchReq ) messageID := strings.Join([]string{ - req.Metadata.WorkflowId, req.Metadata.WorkflowExecutionId, ghcapabilities.MethodComputeAction, c.idGenerator(), diff --git a/core/capabilities/compute/compute_test.go b/core/capabilities/compute/compute_test.go index 719bff82edf..e0f5a2f9750 100644 --- a/core/capabilities/compute/compute_test.go +++ b/core/capabilities/compute/compute_test.go @@ -189,7 +189,6 @@ func TestComputeFetch(t *testing.T) { th.connector.EXPECT().GatewayIDs().Return([]string{"gateway1", "gateway2"}) msgID := strings.Join([]string{ - workflowID, workflowExecutionID, ghcapabilities.MethodComputeAction, validRequestUUID, diff --git a/core/services/standardcapabilities/delegate.go b/core/services/standardcapabilities/delegate.go index a92e082dead..ceb69883e90 100644 --- a/core/services/standardcapabilities/delegate.go +++ b/core/services/standardcapabilities/delegate.go @@ -254,7 +254,7 @@ func (d *Delegate) ServicesForSpec(ctx context.Context, spec job.Job) ([]job.Ser } computeSrvc := compute.NewAction(cfg, log, d.registry, handler, idGeneratorFn) - return []job.ServiceCtx{computeSrvc}, nil + return []job.ServiceCtx{handler, computeSrvc}, nil } standardCapability := newStandardCapabilities(log, spec.StandardCapabilitiesSpec, d.cfg, telemetryService, kvStore, d.registry, errorLog, From 827e003a0c8c0212c6de0cb330e1f27a9c583dab Mon Sep 17 00:00:00 2001 From: chudilka1 Date: Fri, 15 Nov 2024 14:00:09 +0200 Subject: [PATCH 05/10] Update Log.Level config docs (#15244) --- .changeset/rude-geckos-switch.md | 5 +++++ core/config/docs/core.toml | 4 ++-- docs/CONFIG.md | 4 ++-- 3 files changed, 9 insertions(+), 4 deletions(-) create mode 100644 .changeset/rude-geckos-switch.md diff --git a/.changeset/rude-geckos-switch.md b/.changeset/rude-geckos-switch.md new file mode 100644 index 00000000000..866b1c40c63 --- /dev/null +++ b/.changeset/rude-geckos-switch.md @@ -0,0 +1,5 @@ +--- +"chainlink": minor +--- + +#bugfix Update Log.Level and MaxSize configs description in the docs diff --git a/core/config/docs/core.toml b/core/config/docs/core.toml index 083633ab57f..e0fc76f449c 100644 --- a/core/config/docs/core.toml +++ b/core/config/docs/core.toml @@ -124,7 +124,7 @@ JsonWrapperKey = 'event' # Example Headers = ['Authorization: token', 'X-SomeOther-Header: value with spaces | and a bar+*'] # Example [Log] -# Level determines both what is printed on the screen and what is written to the log file. +# Level determines only what is printed on the screen/console. This configuration does not apply to the logs that are recorded in a file (see [`Log.File`](#logfile) for more details). # # The available levels are: # - "debug": Useful for forensic debugging of issues. @@ -145,7 +145,7 @@ UnixTS = false # Default [Log.File] # Dir sets the log directory. By default, Chainlink nodes write log data to `$ROOT/log.jsonl`. Dir = '/my/log/directory' # Example -# MaxSize determines the log file's max size in megabytes before file rotation. Having this not set will disable logging to disk. If your disk doesn't have enough disk space, the logging will pause and the application will log errors until space is available again. +# MaxSize determines the log file's max size before file rotation. Having this not set or set to a value smaller than 1Mb will disable logging to disk. If your disk doesn't have enough disk space, the logging will pause and the application will log errors until space is available again. # # Values must have suffixes with a unit like: `5120mb` (5,120 megabytes). If no unit suffix is provided, the value defaults to `b` (bytes). The list of valid unit suffixes are: # diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 47ba5b574cd..ff918468c07 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -389,7 +389,7 @@ UnixTS = false # Default ```toml Level = 'info' # Default ``` -Level determines both what is printed on the screen and what is written to the log file. +Level determines only what is printed on the screen/console. This configuration does not apply to the logs that are recorded in a file (see [`Log.File`](#logfile) for more details). The available levels are: - "debug": Useful for forensic debugging of issues. @@ -434,7 +434,7 @@ Dir sets the log directory. By default, Chainlink nodes write log data to `$ROOT ```toml MaxSize = '5120mb' # Default ``` -MaxSize determines the log file's max size in megabytes before file rotation. Having this not set will disable logging to disk. If your disk doesn't have enough disk space, the logging will pause and the application will log errors until space is available again. +MaxSize determines the log file's max size before file rotation. Having this not set or set to a value smaller than 1Mb will disable logging to disk. If your disk doesn't have enough disk space, the logging will pause and the application will log errors until space is available again. Values must have suffixes with a unit like: `5120mb` (5,120 megabytes). If no unit suffix is provided, the value defaults to `b` (bytes). The list of valid unit suffixes are: From 5f6d46d194c83610f075d12099f53073a2606051 Mon Sep 17 00:00:00 2001 From: Lukasz <120112546+lukaszcl@users.noreply.github.com> Date: Fri, 15 Nov 2024 13:43:30 +0100 Subject: [PATCH 06/10] Fix workflow for nightly flaky test detector (#15260) --- .github/workflows/run-nightly-flaky-test-detector.yml | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/.github/workflows/run-nightly-flaky-test-detector.yml b/.github/workflows/run-nightly-flaky-test-detector.yml index 615233a6106..1c5dc72d4a3 100644 --- a/.github/workflows/run-nightly-flaky-test-detector.yml +++ b/.github/workflows/run-nightly-flaky-test-detector.yml @@ -4,6 +4,7 @@ on: schedule: # Run every night at 3:00 AM UTC - cron: '0 3 * * *' + workflow_dispatch: # Allows manual trigger for debugging jobs: trigger-flaky-test-detection: @@ -14,8 +15,8 @@ jobs: baseRef: 'origin/develop' projectPath: '.' runThreshold: '1' - runAllTests: 'true' - extraArgs: '{ "skipped_tests": "TestChainComponents", "test_repeat_count": "5", "all_tests_runner": "ubuntu22.04-32cores-128GB", "all_tests_runner_count": "3", "min_pass_ratio": "0" }' + runAllTests: true + extraArgs: '{ "skipped_tests": "TestChainComponents", "test_repeat_count": "5", "all_tests_runner": "ubuntu22.04-32cores-128GB", "all_tests_runner_count": "3", "min_pass_ratio": "0", "run_with_race": "false" }' secrets: SLACK_BOT_TOKEN: ${{ secrets.QA_SLACK_API_KEY }} \ No newline at end of file From a4d3c22e4495b0d6731a398e46d4844b4e65b226 Mon Sep 17 00:00:00 2001 From: george-dorin <120329946+george-dorin@users.noreply.github.com> Date: Fri, 15 Nov 2024 15:18:24 +0200 Subject: [PATCH 07/10] Add dual contract transmitter (#15202) * -Remove adaptive send -Add dual transmitter relay config -Add dual transmitter as contract transmitter * -Fix lint * Add tx.Meta * Revert ocr2 helpers * Fix lint * Fix lint * Change tx.meta to pointers Send secondary tx even if primary fails * Add meta fields to test * Remove unused code Change DualTransmissionConfig to pointer * Implement feedback * Fix failing test --- common/txmgr/types/tx.go | 4 ++ core/services/job/job_orm_test.go | 63 +++++++++++++---- core/services/job/models.go | 24 ------- core/services/job/models_test.go | 61 ----------------- core/services/job/orm.go | 28 ++++++-- core/services/ocr2/validate/validate.go | 10 --- core/services/ocrcommon/transmitter.go | 73 +++++++++++++++++++- core/services/ocrcommon/transmitter_test.go | 76 +++++++++++++++++++++ core/services/relay/evm/evm.go | 1 + core/services/relay/evm/types/types.go | 10 +++ core/testdata/testspecs/v2_specs.go | 30 +++----- 11 files changed, 244 insertions(+), 136 deletions(-) diff --git a/common/txmgr/types/tx.go b/common/txmgr/types/tx.go index b65f7edf6e5..f04047a36c1 100644 --- a/common/txmgr/types/tx.go +++ b/common/txmgr/types/tx.go @@ -159,6 +159,10 @@ type TxMeta[ADDR types.Hashable, TX_HASH types.Hashable] struct { MessageIDs []string `json:"MessageIDs,omitempty"` // SeqNumbers is used by CCIP for tx to committed sequence numbers correlation in logs SeqNumbers []uint64 `json:"SeqNumbers,omitempty"` + + // Dual Broadcast + DualBroadcast *bool `json:"DualBroadcast,omitempty"` + DualBroadcastParams *string `json:"DualBroadcastParams,omitempty"` } type TxAttempt[ diff --git a/core/services/job/job_orm_test.go b/core/services/job/job_orm_test.go index 0aa79f2f0b2..9db99fcd48d 100644 --- a/core/services/job/job_orm_test.go +++ b/core/services/job/job_orm_test.go @@ -2014,7 +2014,7 @@ func mustInsertPipelineRun(t *testing.T, orm pipeline.ORM, j job.Job) pipeline.R return run } -func TestORM_CreateJob_OCR2_With_AdaptiveSend(t *testing.T) { +func TestORM_CreateJob_OCR2_With_DualTransmission(t *testing.T) { ctx := testutils.Context(t) customChainID := big.New(testutils.NewRandomEVMChainID()) @@ -2030,27 +2030,66 @@ func TestORM_CreateJob_OCR2_With_AdaptiveSend(t *testing.T) { db := pgtest.NewSqlxDB(t) keyStore := cltest.NewKeyStore(t, db) require.NoError(t, keyStore.OCR2().Add(ctx, cltest.DefaultOCR2Key)) - _, transmitterID := cltest.MustInsertRandomKey(t, keyStore.Eth()) + baseJobSpec := fmt.Sprintf(testspecs.OCR2EVMDualTransmissionSpecMinimalTemplate, transmitterID.String()) + lggr := logger.TestLogger(t) pipelineORM := pipeline.NewORM(db, lggr, config.JobPipeline().MaxSuccessfulRuns()) bridgesORM := bridges.NewORM(db) jobORM := NewTestORM(t, db, pipelineORM, bridgesORM, keyStore) - adaptiveSendKey := cltest.MustGenerateRandomKey(t) + // Enabled but no config set + enabledDualTransmissionSpec := ` + enableDualTransmission=true` - jb, err := ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), testspecs.GetOCR2EVMWithAdaptiveSendSpecMinimal(cltest.DefaultOCR2Key.ID(), transmitterID.String(), adaptiveSendKey.EIP55Address.String()), nil) + jb, err := ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+enabledDualTransmissionSpec, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "dual transmission is enabled but no dual transmission config present") + + // ContractAddress not set + emptyContractAddress := ` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress="" + ` + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+emptyContractAddress, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid contract address in dual transmission config") + + // Transmitter address not set + emptyTransmitterAddress := ` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '' + ` + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+emptyTransmitterAddress, nil) + require.NoError(t, err) + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "invalid transmitter address in dual transmission config") + + dtTransmitterAddress := cltest.MustGenerateRandomKey(t) + completeDualTransmissionSpec := fmt.Sprintf(` + enableDualTransmission=true + [relayConfig.dualTransmission] + contractAddress = '0x613a38AC1659769640aaE063C651F48E0250454C' + transmitterAddress = '%s' + [relayConfig.dualTransmission.meta] + key1 = 'val1' + key2 = ['val2','val3'] + `, + dtTransmitterAddress.Address.String()) + + jb, err = ocr2validate.ValidatedOracleSpecToml(testutils.Context(t), config.OCR2(), config.Insecure(), baseJobSpec+completeDualTransmissionSpec, nil) require.NoError(t, err) - require.Equal(t, "arbitrary-value", jb.AdaptiveSendSpec.Metadata["arbitraryParam"]) - t.Run("unknown transmitter address", func(t *testing.T) { - require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "failed to validate AdaptiveSendSpec.TransmitterAddress: no EVM key matching") - }) + jb.OCR2OracleSpec.TransmitterID = null.StringFrom(transmitterID.String()) - t.Run("multiple jobs", func(t *testing.T) { - keyStore.Eth().XXXTestingOnlyAdd(ctx, adaptiveSendKey) - require.NoError(t, jobORM.CreateJob(ctx, &jb), "failed to validate AdaptiveSendSpec.TransmitterAddress: no EVM key matching") - }) + // Unknown transmitter address + require.ErrorContains(t, jobORM.CreateJob(ctx, &jb), "unknown dual transmission transmitterAddress: no EVM key matching:") + + // Should not error + keyStore.Eth().XXXTestingOnlyAdd(ctx, dtTransmitterAddress) + require.NoError(t, jobORM.CreateJob(ctx, &jb)) } diff --git a/core/services/job/models.go b/core/services/job/models.go index 5054abd08b5..231bf10fda0 100644 --- a/core/services/job/models.go +++ b/core/services/job/models.go @@ -185,7 +185,6 @@ type Job struct { CCIPSpecID *int32 CCIPSpec *CCIPSpec CCIPBootstrapSpecID *int32 - AdaptiveSendSpec *AdaptiveSendSpec `toml:"adaptiveSend"` JobSpecErrors []SpecError Type Type `toml:"type"` SchemaVersion uint32 `toml:"schemaVersion"` @@ -1061,26 +1060,3 @@ type CCIPSpec struct { // and RMN network info for offchain blessing. PluginConfig JSONConfig `toml:"pluginConfig"` } - -type AdaptiveSendSpec struct { - TransmitterAddress *evmtypes.EIP55Address `toml:"transmitterAddress"` - ContractAddress *evmtypes.EIP55Address `toml:"contractAddress"` - Delay time.Duration `toml:"delay"` - Metadata JSONConfig `toml:"metadata"` -} - -func (o *AdaptiveSendSpec) Validate() error { - if o.TransmitterAddress == nil { - return errors.New("no AdaptiveSendSpec.TransmitterAddress found") - } - - if o.ContractAddress == nil { - return errors.New("no AdaptiveSendSpec.ContractAddress found") - } - - if o.Delay.Seconds() <= 1 { - return errors.New("AdaptiveSendSpec.Delay not set or smaller than 1s") - } - - return nil -} diff --git a/core/services/job/models_test.go b/core/services/job/models_test.go index a21a4553219..5ef36ea9f48 100644 --- a/core/services/job/models_test.go +++ b/core/services/job/models_test.go @@ -11,8 +11,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/codec" "github.com/smartcontractkit/chainlink-common/pkg/types" pkgworkflows "github.com/smartcontractkit/chainlink-common/pkg/workflows" - "github.com/smartcontractkit/chainlink/v2/core/internal/cltest" - "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" "github.com/smartcontractkit/chainlink/v2/core/services/job" "github.com/smartcontractkit/chainlink/v2/core/services/relay" @@ -352,62 +350,3 @@ func TestWorkflowSpec_Validate(t *testing.T) { require.NotEmpty(t, w.WorkflowID) }) } - -func TestAdaptiveSendConfig(t *testing.T) { - tests := []struct { - name string - shouldError bool - expectedErrorMessage string - config job.AdaptiveSendSpec - }{ - { - name: "AdaptiveSendSpec.TransmitterAddress not set", - shouldError: true, - expectedErrorMessage: "no AdaptiveSendSpec.TransmitterAddress found", - config: job.AdaptiveSendSpec{ - TransmitterAddress: nil, - ContractAddress: ptr(cltest.NewEIP55Address()), - Delay: time.Second * 30, - }, - }, - { - name: "AdaptiveSendSpec.ContractAddress not set", - shouldError: true, - expectedErrorMessage: "no AdaptiveSendSpec.ContractAddress found", - config: job.AdaptiveSendSpec{ - TransmitterAddress: ptr(cltest.NewEIP55Address()), - ContractAddress: nil, - Delay: time.Second * 30, - }, - }, - { - name: "AdaptiveSendSpec.Delay not set", - shouldError: true, - expectedErrorMessage: "AdaptiveSendSpec.Delay not set or smaller than 1s", - config: job.AdaptiveSendSpec{ - TransmitterAddress: ptr(cltest.NewEIP55Address()), - ContractAddress: ptr(cltest.NewEIP55Address()), - }, - }, - { - name: "AdaptiveSendSpec.Delay set to 50ms", - shouldError: true, - expectedErrorMessage: "AdaptiveSendSpec.Delay not set or smaller than 1s", - config: job.AdaptiveSendSpec{ - TransmitterAddress: ptr(cltest.NewEIP55Address()), - ContractAddress: ptr(cltest.NewEIP55Address()), - Delay: time.Millisecond * 50, - }, - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - if test.shouldError { - require.ErrorContains(t, test.config.Validate(), test.expectedErrorMessage) - } else { - require.NoError(t, test.config.Validate()) - } - }) - } -} diff --git a/core/services/job/orm.go b/core/services/job/orm.go index 1e1d8a98700..5e8b5ce127f 100644 --- a/core/services/job/orm.go +++ b/core/services/job/orm.go @@ -20,7 +20,6 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" "github.com/smartcontractkit/chainlink-common/pkg/types" - "github.com/smartcontractkit/chainlink/v2/core/bridges" evmconfig "github.com/smartcontractkit/chainlink/v2/core/chains/evm/config" evmtypes "github.com/smartcontractkit/chainlink/v2/core/chains/evm/types" @@ -304,10 +303,29 @@ func (o *orm) CreateJob(ctx context.Context, jb *Job) error { } } - if jb.AdaptiveSendSpec != nil { - err = validateKeyStoreMatchForRelay(ctx, jb.OCR2OracleSpec.Relay, tx.keyStore, jb.AdaptiveSendSpec.TransmitterAddress.String()) - if err != nil { - return fmt.Errorf("failed to validate AdaptiveSendSpec.TransmitterAddress: %w", err) + if enableDualTransmission, ok := jb.OCR2OracleSpec.RelayConfig["enableDualTransmission"]; ok && enableDualTransmission != nil { + rawDualTransmissionConfig, ok := jb.OCR2OracleSpec.RelayConfig["dualTransmission"] + if !ok { + return errors.New("dual transmission is enabled but no dual transmission config present") + } + + dualTransmissionConfig, ok := rawDualTransmissionConfig.(map[string]interface{}) + if !ok { + return errors.New("invalid dual transmission config") + } + + dtContractAddress, ok := dualTransmissionConfig["contractAddress"].(string) + if !ok || !common.IsHexAddress(dtContractAddress) { + return errors.New("invalid contract address in dual transmission config") + } + + dtTransmitterAddress, ok := dualTransmissionConfig["transmitterAddress"].(string) + if !ok || !common.IsHexAddress(dtTransmitterAddress) { + return errors.New("invalid transmitter address in dual transmission config") + } + + if err = validateKeyStoreMatchForRelay(ctx, jb.OCR2OracleSpec.Relay, tx.keyStore, dtTransmitterAddress); err != nil { + return errors.Wrap(err, "unknown dual transmission transmitterAddress") } } diff --git a/core/services/ocr2/validate/validate.go b/core/services/ocr2/validate/validate.go index 40e2c11c3e7..c2f3e455232 100644 --- a/core/services/ocr2/validate/validate.go +++ b/core/services/ocr2/validate/validate.go @@ -73,9 +73,6 @@ func ValidatedOracleSpecToml(ctx context.Context, config OCR2Config, insConf Ins if err = validateTimingParameters(config, insConf, spec); err != nil { return jb, err } - if err = validateAdaptiveSendSpec(ctx, jb); err != nil { - return jb, err - } return jb, nil } @@ -380,10 +377,3 @@ func validateOCR2LLOSpec(jsonConfig job.JSONConfig) error { } return pkgerrors.Wrap(pluginConfig.Validate(), "LLO PluginConfig is invalid") } - -func validateAdaptiveSendSpec(ctx context.Context, spec job.Job) error { - if spec.AdaptiveSendSpec != nil { - return spec.AdaptiveSendSpec.Validate() - } - return nil -} diff --git a/core/services/ocrcommon/transmitter.go b/core/services/ocrcommon/transmitter.go index 5d2de45295f..8121f3778d2 100644 --- a/core/services/ocrcommon/transmitter.go +++ b/core/services/ocrcommon/transmitter.go @@ -2,7 +2,10 @@ package ocrcommon import ( "context" + errors2 "errors" + "fmt" "math/big" + "net/url" "slices" "github.com/ethereum/go-ethereum/common" @@ -11,6 +14,7 @@ import ( "github.com/smartcontractkit/chainlink/v2/common/txmgr/types" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/forwarders" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" + types2 "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/types" ) type roundRobinKeystore interface { @@ -88,13 +92,14 @@ func NewOCR2FeedsTransmitter( checker txmgr.TransmitCheckerSpec, chainID *big.Int, keystore roundRobinKeystore, + dualTransmissionConfig *types2.DualTransmissionConfig, ) (Transmitter, error) { // Ensure that a keystore is provided. if keystore == nil { return nil, errors.New("nil keystore provided to transmitter") } - return &ocr2FeedsTransmitter{ + baseTransmitter := &ocr2FeedsTransmitter{ ocr2Aggregator: ocr2Aggregator, txManagerOCR2: txm, transmitter: transmitter{ @@ -107,7 +112,17 @@ func NewOCR2FeedsTransmitter( chainID: chainID, keystore: keystore, }, - }, nil + } + + if dualTransmissionConfig != nil { + return &ocr2FeedsDualTransmission{ + transmitter: *baseTransmitter, + secondaryContractAddress: dualTransmissionConfig.ContractAddress, + secondaryFromAddress: dualTransmissionConfig.TransmitterAddress, + secondaryMeta: dualTransmissionConfig.Meta, + }, nil + } + return baseTransmitter, nil } func (t *transmitter) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error { @@ -203,3 +218,57 @@ func (t *ocr2FeedsTransmitter) forwarderAddress(ctx context.Context, eoa, ocr2Ag return forwarderAddress, nil } + +type ocr2FeedsDualTransmission struct { + transmitter ocr2FeedsTransmitter + + secondaryContractAddress common.Address + secondaryFromAddress common.Address + secondaryMeta map[string][]string +} + +func (t *ocr2FeedsDualTransmission) CreateEthTransaction(ctx context.Context, toAddress common.Address, payload []byte, txMeta *txmgr.TxMeta) error { + // Primary transmission + errPrimary := t.transmitter.CreateEthTransaction(ctx, toAddress, payload, txMeta) + if errPrimary != nil { + errPrimary = fmt.Errorf("skipped primary transmission: %w", errPrimary) + } + + if txMeta == nil { + txMeta = &txmgr.TxMeta{} + } + + dualBroadcast := true + dualBroadcastParams := t.urlParams() + + txMeta.DualBroadcast = &dualBroadcast + txMeta.DualBroadcastParams = &dualBroadcastParams + + // Secondary transmission + _, errSecondary := t.transmitter.txm.CreateTransaction(ctx, txmgr.TxRequest{ + FromAddress: t.secondaryFromAddress, + ToAddress: t.secondaryContractAddress, + EncodedPayload: payload, + FeeLimit: t.transmitter.gasLimit, + Strategy: t.transmitter.strategy, + Checker: t.transmitter.checker, + Meta: txMeta, + }) + + errSecondary = errors.Wrap(errSecondary, "skipped secondary transmission") + return errors2.Join(errPrimary, errSecondary) +} + +func (t *ocr2FeedsDualTransmission) FromAddress(ctx context.Context) common.Address { + return t.transmitter.FromAddress(ctx) +} + +func (t *ocr2FeedsDualTransmission) urlParams() string { + values := url.Values{} + for k, v := range t.secondaryMeta { + for _, p := range v { + values.Add(k, p) + } + } + return values.Encode() +} diff --git a/core/services/ocrcommon/transmitter_test.go b/core/services/ocrcommon/transmitter_test.go index d6a07190800..5f434e59c62 100644 --- a/core/services/ocrcommon/transmitter_test.go +++ b/core/services/ocrcommon/transmitter_test.go @@ -5,16 +5,19 @@ import ( "testing" "github.com/ethereum/go-ethereum/common" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" commontxmmocks "github.com/smartcontractkit/chainlink/v2/common/txmgr/types/mocks" "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr" txmmocks "github.com/smartcontractkit/chainlink/v2/core/chains/evm/txmgr/mocks" + "github.com/smartcontractkit/chainlink/v2/core/chains/evm/utils" "github.com/smartcontractkit/chainlink/v2/core/internal/cltest" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils" "github.com/smartcontractkit/chainlink/v2/core/internal/testutils/pgtest" "github.com/smartcontractkit/chainlink/v2/core/services/ocrcommon" + "github.com/smartcontractkit/chainlink/v2/core/services/relay/evm/types" ) func newMockTxStrategy(t *testing.T) *commontxmmocks.TxStrategy { @@ -169,3 +172,76 @@ func Test_DefaultTransmitter_Forwarding_Enabled_CreateEthTransaction_No_Keystore ) require.Error(t, err) } + +func Test_DualTransmitter(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + ethKeyStore := cltest.NewKeyStore(t, db).Eth() + + _, fromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + _, secondaryFromAddress := cltest.MustInsertRandomKey(t, ethKeyStore) + + contractAddress := utils.RandomAddress() + secondaryContractAddress := utils.RandomAddress() + + gasLimit := uint64(1000) + chainID := big.NewInt(0) + effectiveTransmitterAddress := fromAddress + toAddress := testutils.NewAddress() + payload := []byte{1, 2, 3} + txm := txmmocks.NewMockEvmTxManager(t) + strategy := newMockTxStrategy(t) + dualTransmissionConfig := &types.DualTransmissionConfig{ + ContractAddress: secondaryContractAddress, + TransmitterAddress: secondaryFromAddress, + Meta: map[string][]string{ + "key1": {"value1"}, + "key2": {"value2", "value3"}, + "key3": {"value4", "value5", "value6"}, + }, + } + + transmitter, err := ocrcommon.NewOCR2FeedsTransmitter( + txm, + []common.Address{fromAddress}, + contractAddress, + gasLimit, + effectiveTransmitterAddress, + strategy, + txmgr.TransmitCheckerSpec{}, + chainID, + ethKeyStore, + dualTransmissionConfig, + ) + require.NoError(t, err) + + primaryTxConfirmed := false + secondaryTxConfirmed := false + + txm.On("CreateTransaction", mock.Anything, mock.MatchedBy(func(tx txmgr.TxRequest) bool { + switch tx.FromAddress { + case fromAddress: + // Primary transmission + assert.Equal(t, tx.ToAddress, toAddress, "unexpected primary toAddress") + assert.Nil(t, tx.Meta, "Meta should be empty") + primaryTxConfirmed = true + case secondaryFromAddress: + // Secondary transmission + assert.Equal(t, tx.ToAddress, secondaryContractAddress, "unexpected secondary toAddress") + assert.True(t, *tx.Meta.DualBroadcast, "DualBroadcast should be true") + assert.Equal(t, "key1=value1&key2=value2&key2=value3&key3=value4&key3=value5&key3=value6", *tx.Meta.DualBroadcastParams, "DualBroadcastParams not equal") + secondaryTxConfirmed = true + default: + // Should never be reached + return false + } + + return true + })).Twice().Return(txmgr.Tx{}, nil) + + require.NoError(t, transmitter.CreateEthTransaction(testutils.Context(t), toAddress, payload, nil)) + + require.True(t, primaryTxConfirmed) + require.True(t, secondaryTxConfirmed) +} diff --git a/core/services/relay/evm/evm.go b/core/services/relay/evm/evm.go index 7c070cc282d..db0fe90796b 100644 --- a/core/services/relay/evm/evm.go +++ b/core/services/relay/evm/evm.go @@ -806,6 +806,7 @@ func generateTransmitterFrom(ctx context.Context, rargs commontypes.RelayArgs, e checker, configWatcher.chain.ID(), ethKeystore, + relayConfig.DualTransmissionConfig, ) case commontypes.CCIPExecution: transmitter, err = cciptransmitter.NewTransmitterWithStatusChecker( diff --git a/core/services/relay/evm/types/types.go b/core/services/relay/evm/types/types.go index e1a61098be5..2b56aee6379 100644 --- a/core/services/relay/evm/types/types.go +++ b/core/services/relay/evm/types/types.go @@ -192,6 +192,12 @@ func (c LLOConfigMode) String() string { return string(c) } +type DualTransmissionConfig struct { + ContractAddress common.Address `json:"contractAddress" toml:"contractAddress"` + TransmitterAddress common.Address `json:"transmitterAddress" toml:"transmitterAddress"` + Meta map[string][]string `json:"meta" toml:"meta"` +} + type RelayConfig struct { ChainID *big.Big `json:"chainID"` FromBlock uint64 `json:"fromBlock"` @@ -215,6 +221,10 @@ type RelayConfig struct { // LLO-specific LLODONID uint32 `json:"lloDonID" toml:"lloDonID"` LLOConfigMode LLOConfigMode `json:"lloConfigMode" toml:"lloConfigMode"` + + // DualTransmission specific + EnableDualTransmission bool `json:"enableDualTransmission" toml:"enableDualTransmission"` + DualTransmissionConfig *DualTransmissionConfig `json:"dualTransmission" toml:"dualTransmission"` } var ErrBadRelayConfig = errors.New("bad relay config") diff --git a/core/testdata/testspecs/v2_specs.go b/core/testdata/testspecs/v2_specs.go index d3d72467a83..d519ace6479 100644 --- a/core/testdata/testspecs/v2_specs.go +++ b/core/testdata/testspecs/v2_specs.go @@ -951,42 +951,28 @@ targets: inputs: consensus_output: $(a-consensus.outputs) ` -var OCR2EVMSpecMinimalWithAdaptiveSendTemplate = ` +var OCR2EVMDualTransmissionSpecMinimalTemplate = ` type = "offchainreporting2" schemaVersion = 1 -name = "%s" +name = "test-job" +relay = "evm" contractID = "0x613a38AC1659769640aaE063C651F48E0250454C" p2pv2Bootstrappers = [] -ocrKeyBundleID = "%s" -relay = "evm" -pluginType = "median" transmitterID = "%s" - +pluginType = "median" observationSource = """ ds [type=http method=GET url="https://chain.link/ETH-USD"]; ds_parse [type=jsonparse path="data.price" separator="."]; ds_multiply [type=multiply times=100]; ds -> ds_parse -> ds_multiply; """ - [pluginConfig] juelsPerFeeCoinSource = """ - ds1 [type=http method=GET url="https://chain.link/jules" allowunrestrictednetworkaccess="true"]; - ds1_parse [type=jsonparse path="answer"]; - ds1_multiply [type=multiply times=1]; - ds1 -> ds1_parse -> ds1_multiply; + ds [type=http method=GET url="https://chain.link/ETH-USD"]; + ds_parse [type=jsonparse path="data.price" separator="."]; + ds_multiply [type=multiply times=100]; + ds -> ds_parse -> ds_multiply; """ [relayConfig] chainID = 0 - -[adaptiveSend] -transmitterAddress = '%s' -contractAddress = '0xF67D0290337bca0847005C7ffD1BC75BA9AAE6e4' -delay = '30s' -[adaptiveSend.metadata] -arbitraryParam = 'arbitrary-value' ` - -func GetOCR2EVMWithAdaptiveSendSpecMinimal(keyBundle, transmitterID, secondaryTransmitterAddress string) string { - return fmt.Sprintf(OCR2EVMSpecMinimalWithAdaptiveSendTemplate, uuid.New(), keyBundle, transmitterID, secondaryTransmitterAddress) -} From cb9c185910f054c9e162f8181cb2bebaa81988b9 Mon Sep 17 00:00:00 2001 From: Vyzaldy Sanchez Date: Fri, 15 Nov 2024 11:14:16 -0400 Subject: [PATCH 08/10] Add compute metrics (#15246) * Adds compute metrics * Fixes lint * Removes execution ID from metrics labels * Fixes CI * Fixes CI * Fixes CI --- core/capabilities/compute/compute.go | 23 +++++++++--- core/capabilities/compute/compute_test.go | 4 ++- core/capabilities/compute/monitoring.go | 35 +++++++++++++++++++ .../services/standardcapabilities/delegate.go | 6 +++- core/services/workflows/engine_test.go | 6 ++-- 5 files changed, 66 insertions(+), 8 deletions(-) diff --git a/core/capabilities/compute/compute.go b/core/capabilities/compute/compute.go index cdb8ee19a6a..7b11072b161 100644 --- a/core/capabilities/compute/compute.go +++ b/core/capabilities/compute/compute.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "net/http" + "strconv" "strings" "sync" "time" @@ -20,6 +21,7 @@ import ( capabilitiespb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/pb" "github.com/smartcontractkit/chainlink-common/pkg/custmsg" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink-common/pkg/metrics" "github.com/smartcontractkit/chainlink-common/pkg/services" coretypes "github.com/smartcontractkit/chainlink-common/pkg/types/core" "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host" @@ -75,8 +77,9 @@ var ( var _ capabilities.ActionCapability = (*Compute)(nil) type Compute struct { - stopCh services.StopChan - log logger.Logger + stopCh services.StopChan + log logger.Logger + metrics *computeMetricsLabeler // emitter is used to emit messages from the WASM module to a configured collector. emitter custmsg.MessageEmitter @@ -328,6 +331,13 @@ func (c *Compute) createFetcher() func(ctx context.Context, req *wasmpb.FetchReq return nil, fmt.Errorf("failed to unmarshal fetch response: %w", err) } + c.metrics.with( + "status", strconv.FormatUint(uint64(response.StatusCode), 10), + platform.KeyWorkflowID, req.Metadata.WorkflowId, + platform.KeyWorkflowName, req.Metadata.WorkflowName, + platform.KeyWorkflowOwner, req.Metadata.WorkflowOwner, + ).incrementHTTPRequestCounter(ctx) + // Only log if the response is not in the 200 range if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices { msg := fmt.Sprintf("compute fetch request failed with status code %d", response.StatusCode) @@ -357,10 +367,14 @@ func NewAction( handler *webapi.OutgoingConnectorHandler, idGenerator func() string, opts ...func(*Compute), -) *Compute { +) (*Compute, error) { if config.NumWorkers == 0 { config.NumWorkers = defaultNumWorkers } + metricsLabeler, err := newComputeMetricsLabeler(metrics.NewLabeler().With("capability", CapabilityIDCompute)) + if err != nil { + return nil, fmt.Errorf("failed to create compute metrics labeler: %w", err) + } var ( lggr = logger.Named(log, "CustomCompute") labeler = custmsg.NewLabeler() @@ -368,6 +382,7 @@ func NewAction( stopCh: make(services.StopChan), log: lggr, emitter: labeler, + metrics: metricsLabeler, registry: registry, modules: newModuleCache(clockwork.NewRealClock(), 1*time.Minute, 10*time.Minute, 3), transformer: NewTransformer(lggr, labeler), @@ -382,5 +397,5 @@ func NewAction( opt(compute) } - return compute + return compute, nil } diff --git a/core/capabilities/compute/compute_test.go b/core/capabilities/compute/compute_test.go index e0f5a2f9750..c4146b7408e 100644 --- a/core/capabilities/compute/compute_test.go +++ b/core/capabilities/compute/compute_test.go @@ -18,6 +18,7 @@ import ( cappkg "github.com/smartcontractkit/chainlink-common/pkg/capabilities" "github.com/smartcontractkit/chainlink-common/pkg/utils/tests" "github.com/smartcontractkit/chainlink-common/pkg/values" + corecapabilities "github.com/smartcontractkit/chainlink/v2/core/capabilities" "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi" "github.com/smartcontractkit/chainlink/v2/core/services/gateway/api" @@ -60,7 +61,8 @@ func setup(t *testing.T, config Config) testHarness { connectorHandler, err := webapi.NewOutgoingConnectorHandler(connector, config.ServiceConfig, ghcapabilities.MethodComputeAction, log) require.NoError(t, err) - compute := NewAction(config, log, registry, connectorHandler, idGeneratorFn) + compute, err := NewAction(config, log, registry, connectorHandler, idGeneratorFn) + require.NoError(t, err) compute.modules.clock = clockwork.NewFakeClock() return testHarness{ diff --git a/core/capabilities/compute/monitoring.go b/core/capabilities/compute/monitoring.go index 4b676c25f7d..71354c014a7 100644 --- a/core/capabilities/compute/monitoring.go +++ b/core/capabilities/compute/monitoring.go @@ -1,3 +1,38 @@ package compute +import ( + "context" + "fmt" + + "go.opentelemetry.io/otel/metric" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" + "github.com/smartcontractkit/chainlink-common/pkg/metrics" + + localMonitoring "github.com/smartcontractkit/chainlink/v2/core/monitoring" +) + const timestampKey = "computeTimestamp" + +type computeMetricsLabeler struct { + metrics.Labeler + computeHTTPRequestCounter metric.Int64Counter +} + +func newComputeMetricsLabeler(l metrics.Labeler) (*computeMetricsLabeler, error) { + computeHTTPRequestCounter, err := beholder.GetMeter().Int64Counter("capabilities_compute_http_request_count") + if err != nil { + return nil, fmt.Errorf("failed to register compute http request counter: %w", err) + } + + return &computeMetricsLabeler{Labeler: l, computeHTTPRequestCounter: computeHTTPRequestCounter}, nil +} + +func (c *computeMetricsLabeler) with(keyValues ...string) *computeMetricsLabeler { + return &computeMetricsLabeler{c.With(keyValues...), c.computeHTTPRequestCounter} +} + +func (c *computeMetricsLabeler) incrementHTTPRequestCounter(ctx context.Context) { + otelLabels := localMonitoring.KvMapToOtelAttributes(c.Labels) + c.computeHTTPRequestCounter.Add(ctx, 1, metric.WithAttributes(otelLabels...)) +} diff --git a/core/services/standardcapabilities/delegate.go b/core/services/standardcapabilities/delegate.go index ceb69883e90..f4c0f360bf3 100644 --- a/core/services/standardcapabilities/delegate.go +++ b/core/services/standardcapabilities/delegate.go @@ -12,6 +12,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/sqlutil" "github.com/smartcontractkit/chainlink-common/pkg/types" "github.com/smartcontractkit/chainlink-common/pkg/types/core" + "github.com/smartcontractkit/chainlink/v2/core/capabilities/compute" gatewayconnector "github.com/smartcontractkit/chainlink/v2/core/capabilities/gateway_connector" "github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi" @@ -253,7 +254,10 @@ func (d *Delegate) ServicesForSpec(ctx context.Context, spec job.Job) ([]job.Ser return uuid.New().String() } - computeSrvc := compute.NewAction(cfg, log, d.registry, handler, idGeneratorFn) + computeSrvc, err := compute.NewAction(cfg, log, d.registry, handler, idGeneratorFn) + if err != nil { + return nil, err + } return []job.ServiceCtx{handler, computeSrvc}, nil } diff --git a/core/services/workflows/engine_test.go b/core/services/workflows/engine_test.go index e6667fe0bc6..3a2bc17bc36 100644 --- a/core/services/workflows/engine_test.go +++ b/core/services/workflows/engine_test.go @@ -1448,7 +1448,8 @@ func TestEngine_WithCustomComputeStep(t *testing.T) { require.NoError(t, err) idGeneratorFn := func() string { return "validRequestID" } - compute := compute.NewAction(cfg, log, reg, handler, idGeneratorFn) + compute, err := compute.NewAction(cfg, log, reg, handler, idGeneratorFn) + require.NoError(t, err) require.NoError(t, compute.Start(ctx)) defer compute.Close() @@ -1513,7 +1514,8 @@ func TestEngine_CustomComputePropagatesBreaks(t *testing.T) { require.NoError(t, err) idGeneratorFn := func() string { return "validRequestID" } - compute := compute.NewAction(cfg, log, reg, handler, idGeneratorFn) + compute, err := compute.NewAction(cfg, log, reg, handler, idGeneratorFn) + require.NoError(t, err) require.NoError(t, compute.Start(ctx)) defer compute.Close() From d691d59cda52f24cd4eba06a71e58eef2b71a7d5 Mon Sep 17 00:00:00 2001 From: Lukasz <120112546+lukaszcl@users.noreply.github.com> Date: Fri, 15 Nov 2024 16:56:59 +0100 Subject: [PATCH 09/10] Bump flakeguard (#15267) --- .github/workflows/find-new-flaky-tests.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/find-new-flaky-tests.yml b/.github/workflows/find-new-flaky-tests.yml index ee27ac37562..c9a015206fd 100644 --- a/.github/workflows/find-new-flaky-tests.yml +++ b/.github/workflows/find-new-flaky-tests.yml @@ -100,7 +100,7 @@ jobs: - name: Install flakeguard shell: bash - run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@897bca304fc9f0e68b87579558750c4a3e83adec + run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@9da200418b01268201a0e6477832734a14720d07 - name: Find new or updated test packages if: ${{ inputs.runAllTests == false }} @@ -259,7 +259,7 @@ jobs: - name: Install flakeguard shell: bash - run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@897bca304fc9f0e68b87579558750c4a3e83adec + run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@9da200418b01268201a0e6477832734a14720d07 - name: Run tests with flakeguard shell: bash @@ -301,7 +301,7 @@ jobs: - name: Install flakeguard shell: bash - run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@897bca304fc9f0e68b87579558750c4a3e83adec + run: go install github.com/smartcontractkit/chainlink-testing-framework/tools/flakeguard@9da200418b01268201a0e6477832734a14720d07 - name: Set combined test results id: set_test_results From 452a99e6a282a5c75d978631754e69f69be92e94 Mon Sep 17 00:00:00 2001 From: Connor Stein Date: Fri, 15 Nov 2024 13:25:59 -0500 Subject: [PATCH 10/10] Common deploy helper (#15268) * Common deploy * Can remove interface * Helpers --- deployment/ccip/deploy.go | 169 +++++++++------------------ deployment/ccip/deploy_home_chain.go | 26 ++--- deployment/ccip/test_helpers.go | 25 ++-- deployment/go.mod | 1 + deployment/helpers.go | 44 +++++++ 5 files changed, 123 insertions(+), 142 deletions(-) diff --git a/deployment/ccip/deploy.go b/deployment/ccip/deploy.go index 77df3aab60f..8a33cf0d26d 100644 --- a/deployment/ccip/deploy.go +++ b/deployment/ccip/deploy.go @@ -7,21 +7,16 @@ import ( "github.com/ethereum/go-ethereum/accounts/abi/bind" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" - "github.com/ethereum/go-ethereum/core/types" "github.com/smartcontractkit/ccip-owner-contracts/pkg/config" owner_helpers "github.com/smartcontractkit/ccip-owner-contracts/pkg/gethwrappers" "github.com/smartcontractkit/chainlink-common/pkg/logger" + "github.com/smartcontractkit/chainlink/deployment" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/fee_quoter" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/registry_module_owner_custom" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/rmn_home" - "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/generated/aggregator_v3_interface" - "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/shared/generated/erc20" - - "github.com/smartcontractkit/chainlink/deployment" - "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/burn_mint_token_pool" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/ccip_home" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/maybe_revert_message_receiver" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/nonce_manager" @@ -32,7 +27,6 @@ import ( "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/router" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/token_admin_registry" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/ccip/generated/weth9" - "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/keystone/generated/capabilities_registry" "github.com/smartcontractkit/chainlink/v2/core/gethwrappers/shared/generated/burn_mint_erc677" ) @@ -66,65 +60,6 @@ var ( BurnMintTokenPool deployment.ContractType = "BurnMintTokenPool" ) -type Contracts interface { - *capabilities_registry.CapabilitiesRegistry | - *rmn_proxy_contract.RMNProxyContract | - *ccip_home.CCIPHome | - *rmn_home.RMNHome | - *nonce_manager.NonceManager | - *fee_quoter.FeeQuoter | - *router.Router | - *token_admin_registry.TokenAdminRegistry | - *registry_module_owner_custom.RegistryModuleOwnerCustom | - *weth9.WETH9 | - *rmn_remote.RMNRemote | - *owner_helpers.ManyChainMultiSig | - *owner_helpers.RBACTimelock | - *offramp.OffRamp | - *onramp.OnRamp | - *burn_mint_erc677.BurnMintERC677 | - *burn_mint_token_pool.BurnMintTokenPool | - *maybe_revert_message_receiver.MaybeRevertMessageReceiver | - *aggregator_v3_interface.AggregatorV3Interface | - *erc20.ERC20 -} - -type ContractDeploy[C Contracts] struct { - // We just keep all the deploy return values - // since some will be empty if there's an error. - Address common.Address - Contract C - Tx *types.Transaction - Tv deployment.TypeAndVersion - Err error -} - -// TODO: pull up to general deployment pkg somehow -// without exposing all product specific contracts? -func deployContract[C Contracts]( - lggr logger.Logger, - chain deployment.Chain, - addressBook deployment.AddressBook, - deploy func(chain deployment.Chain) ContractDeploy[C], -) (*ContractDeploy[C], error) { - contractDeploy := deploy(chain) - if contractDeploy.Err != nil { - lggr.Errorw("Failed to deploy contract", "err", contractDeploy.Err) - return nil, contractDeploy.Err - } - _, err := chain.Confirm(contractDeploy.Tx) - if err != nil { - lggr.Errorw("Failed to confirm deployment", "err", err) - return nil, err - } - err = addressBook.Save(chain.Selector, contractDeploy.Address.String(), contractDeploy.Tv) - if err != nil { - lggr.Errorw("Failed to save contract address", "err", err) - return nil, err - } - return &contractDeploy, nil -} - type DeployCCIPContractConfig struct { HomeChainSel uint64 FeedChainSel uint64 @@ -268,15 +203,15 @@ func DeployMCMSWithConfig( chain deployment.Chain, ab deployment.AddressBook, mcmConfig config.Config, -) (*ContractDeploy[*owner_helpers.ManyChainMultiSig], error) { +) (*deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig], error) { groupQuorums, groupParents, signerAddresses, signerGroups := mcmConfig.ExtractSetConfigInputs() - mcm, err := deployContract(lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*owner_helpers.ManyChainMultiSig] { + mcm, err := deployment.DeployContract(lggr, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig] { mcmAddr, tx, mcm, err2 := owner_helpers.DeployManyChainMultiSig( chain.DeployerKey, chain.Client, ) - return ContractDeploy[*owner_helpers.ManyChainMultiSig]{ + return deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig]{ mcmAddr, mcm, tx, deployment.NewTypeAndVersion(contractType, deployment.Version1_0_0), err2, } }) @@ -299,11 +234,11 @@ func DeployMCMSWithConfig( } type MCMSContracts struct { - Admin *ContractDeploy[*owner_helpers.ManyChainMultiSig] - Canceller *ContractDeploy[*owner_helpers.ManyChainMultiSig] - Bypasser *ContractDeploy[*owner_helpers.ManyChainMultiSig] - Proposer *ContractDeploy[*owner_helpers.ManyChainMultiSig] - Timelock *ContractDeploy[*owner_helpers.RBACTimelock] + Admin *deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig] + Canceller *deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig] + Bypasser *deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig] + Proposer *deployment.ContractDeploy[*owner_helpers.ManyChainMultiSig] + Timelock *deployment.ContractDeploy[*owner_helpers.RBACTimelock] } // DeployMCMSContracts deploys the MCMS contracts for the given configuration @@ -331,8 +266,8 @@ func DeployMCMSContracts( return nil, err } - timelock, err := deployContract(lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*owner_helpers.RBACTimelock] { + timelock, err := deployment.DeployContract(lggr, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*owner_helpers.RBACTimelock] { timelock, tx2, cc, err2 := owner_helpers.DeployRBACTimelock( chain.DeployerKey, chain.Client, @@ -343,7 +278,7 @@ func DeployMCMSContracts( []common.Address{canceller.Address}, // cancellers []common.Address{bypasser.Address}, // bypassers ) - return ContractDeploy[*owner_helpers.RBACTimelock]{ + return deployment.ContractDeploy[*owner_helpers.RBACTimelock]{ timelock, cc, tx2, deployment.NewTypeAndVersion(RBACTimelock, deployment.Version1_0_0), err2, } }) @@ -374,13 +309,13 @@ func DeployFeeTokensToChains(lggr logger.Logger, ab deployment.AddressBook, chai // DeployFeeTokens deploys link and weth9. This is _usually_ for test environments only, // real environments they tend to already exist, but sometimes we still have to deploy them to real chains. func DeployFeeTokens(lggr logger.Logger, chain deployment.Chain, ab deployment.AddressBook) (FeeTokenContracts, error) { - weth9, err := deployContract(lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*weth9.WETH9] { + weth9, err := deployment.DeployContract(lggr, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*weth9.WETH9] { weth9Addr, tx2, weth9c, err2 := weth9.DeployWETH9( chain.DeployerKey, chain.Client, ) - return ContractDeploy[*weth9.WETH9]{ + return deployment.ContractDeploy[*weth9.WETH9]{ weth9Addr, weth9c, tx2, deployment.NewTypeAndVersion(WETH9, deployment.Version1_0_0), err2, } }) @@ -390,8 +325,8 @@ func DeployFeeTokens(lggr logger.Logger, chain deployment.Chain, ab deployment.A } lggr.Infow("deployed weth9", "addr", weth9.Address) - linkToken, err := deployContract(lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*burn_mint_erc677.BurnMintERC677] { + linkToken, err := deployment.DeployContract(lggr, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*burn_mint_erc677.BurnMintERC677] { linkTokenAddr, tx2, linkToken, err2 := burn_mint_erc677.DeployBurnMintERC677( chain.DeployerKey, chain.Client, @@ -400,7 +335,7 @@ func DeployFeeTokens(lggr logger.Logger, chain deployment.Chain, ab deployment.A uint8(18), big.NewInt(0).Mul(big.NewInt(1e9), big.NewInt(1e18)), ) - return ContractDeploy[*burn_mint_erc677.BurnMintERC677]{ + return deployment.ContractDeploy[*burn_mint_erc677.BurnMintERC677]{ linkTokenAddr, linkToken, tx2, deployment.NewTypeAndVersion(LinkToken, deployment.Version1_0_0), err2, } }) @@ -432,14 +367,14 @@ func DeployChainContracts( if err != nil { return err } - ccipReceiver, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*maybe_revert_message_receiver.MaybeRevertMessageReceiver] { + ccipReceiver, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*maybe_revert_message_receiver.MaybeRevertMessageReceiver] { receiverAddr, tx, receiver, err2 := maybe_revert_message_receiver.DeployMaybeRevertMessageReceiver( chain.DeployerKey, chain.Client, false, ) - return ContractDeploy[*maybe_revert_message_receiver.MaybeRevertMessageReceiver]{ + return deployment.ContractDeploy[*maybe_revert_message_receiver.MaybeRevertMessageReceiver]{ receiverAddr, receiver, tx, deployment.NewTypeAndVersion(CCIPReceiver, deployment.Version1_0_0), err2, } }) @@ -450,14 +385,14 @@ func DeployChainContracts( e.Logger.Infow("deployed receiver", "addr", ccipReceiver.Address) // TODO: Correctly configure RMN remote. - rmnRemote, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*rmn_remote.RMNRemote] { + rmnRemote, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*rmn_remote.RMNRemote] { rmnRemoteAddr, tx, rmnRemote, err2 := rmn_remote.DeployRMNRemote( chain.DeployerKey, chain.Client, chain.Selector, ) - return ContractDeploy[*rmn_remote.RMNRemote]{ + return deployment.ContractDeploy[*rmn_remote.RMNRemote]{ rmnRemoteAddr, rmnRemote, tx, deployment.NewTypeAndVersion(RMNRemote, deployment.Version1_6_0_dev), err2, } }) @@ -486,14 +421,14 @@ func DeployChainContracts( return err } - rmnProxy, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*rmn_proxy_contract.RMNProxyContract] { + rmnProxy, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*rmn_proxy_contract.RMNProxyContract] { rmnProxyAddr, tx2, rmnProxy, err2 := rmn_proxy_contract.DeployRMNProxyContract( chain.DeployerKey, chain.Client, rmnRemote.Address, ) - return ContractDeploy[*rmn_proxy_contract.RMNProxyContract]{ + return deployment.ContractDeploy[*rmn_proxy_contract.RMNProxyContract]{ rmnProxyAddr, rmnProxy, tx2, deployment.NewTypeAndVersion(ARMProxy, deployment.Version1_0_0), err2, } }) @@ -503,15 +438,15 @@ func DeployChainContracts( } e.Logger.Infow("deployed rmnProxy", "addr", rmnProxy.Address) - routerContract, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*router.Router] { + routerContract, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*router.Router] { routerAddr, tx2, routerC, err2 := router.DeployRouter( chain.DeployerKey, chain.Client, contractConfig.Weth9.Address(), rmnProxy.Address, ) - return ContractDeploy[*router.Router]{ + return deployment.ContractDeploy[*router.Router]{ routerAddr, routerC, tx2, deployment.NewTypeAndVersion(Router, deployment.Version1_2_0), err2, } }) @@ -521,15 +456,15 @@ func DeployChainContracts( } e.Logger.Infow("deployed router", "addr", routerContract.Address) - testRouterContract, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*router.Router] { + testRouterContract, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*router.Router] { routerAddr, tx2, routerC, err2 := router.DeployRouter( chain.DeployerKey, chain.Client, contractConfig.Weth9.Address(), rmnProxy.Address, ) - return ContractDeploy[*router.Router]{ + return deployment.ContractDeploy[*router.Router]{ routerAddr, routerC, tx2, deployment.NewTypeAndVersion(TestRouter, deployment.Version1_2_0), err2, } }) @@ -539,12 +474,12 @@ func DeployChainContracts( } e.Logger.Infow("deployed test router", "addr", testRouterContract.Address) - tokenAdminRegistry, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*token_admin_registry.TokenAdminRegistry] { + tokenAdminRegistry, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*token_admin_registry.TokenAdminRegistry] { tokenAdminRegistryAddr, tx2, tokenAdminRegistry, err2 := token_admin_registry.DeployTokenAdminRegistry( chain.DeployerKey, chain.Client) - return ContractDeploy[*token_admin_registry.TokenAdminRegistry]{ + return deployment.ContractDeploy[*token_admin_registry.TokenAdminRegistry]{ tokenAdminRegistryAddr, tokenAdminRegistry, tx2, deployment.NewTypeAndVersion(TokenAdminRegistry, deployment.Version1_5_0), err2, } }) @@ -554,13 +489,13 @@ func DeployChainContracts( } e.Logger.Infow("deployed tokenAdminRegistry", "addr", tokenAdminRegistry) - customRegistryModule, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*registry_module_owner_custom.RegistryModuleOwnerCustom] { + customRegistryModule, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*registry_module_owner_custom.RegistryModuleOwnerCustom] { regModAddr, tx2, regMod, err2 := registry_module_owner_custom.DeployRegistryModuleOwnerCustom( chain.DeployerKey, chain.Client, tokenAdminRegistry.Address) - return ContractDeploy[*registry_module_owner_custom.RegistryModuleOwnerCustom]{ + return deployment.ContractDeploy[*registry_module_owner_custom.RegistryModuleOwnerCustom]{ regModAddr, regMod, tx2, deployment.NewTypeAndVersion(RegistryModule, deployment.Version1_5_0), err2, } }) @@ -584,14 +519,14 @@ func DeployChainContracts( e.Logger.Infow("assigned registry module on token admin registry") - nonceManager, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*nonce_manager.NonceManager] { + nonceManager, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*nonce_manager.NonceManager] { nonceManagerAddr, tx2, nonceManager, err2 := nonce_manager.DeployNonceManager( chain.DeployerKey, chain.Client, []common.Address{}, // Need to add onRamp after ) - return ContractDeploy[*nonce_manager.NonceManager]{ + return deployment.ContractDeploy[*nonce_manager.NonceManager]{ nonceManagerAddr, nonceManager, tx2, deployment.NewTypeAndVersion(NonceManager, deployment.Version1_6_0_dev), err2, } }) @@ -601,8 +536,8 @@ func DeployChainContracts( } e.Logger.Infow("Deployed nonce manager", "addr", nonceManager.Address) - feeQuoter, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*fee_quoter.FeeQuoter] { + feeQuoter, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*fee_quoter.FeeQuoter] { prAddr, tx2, pr, err2 := fee_quoter.DeployFeeQuoter( chain.DeployerKey, chain.Client, @@ -627,7 +562,7 @@ func DeployChainContracts( }, []fee_quoter.FeeQuoterDestChainConfigArgs{}, ) - return ContractDeploy[*fee_quoter.FeeQuoter]{ + return deployment.ContractDeploy[*fee_quoter.FeeQuoter]{ prAddr, pr, tx2, deployment.NewTypeAndVersion(FeeQuoter, deployment.Version1_6_0_dev), err2, } }) @@ -637,8 +572,8 @@ func DeployChainContracts( } e.Logger.Infow("Deployed fee quoter", "addr", feeQuoter.Address) - onRamp, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*onramp.OnRamp] { + onRamp, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*onramp.OnRamp] { onRampAddr, tx2, onRamp, err2 := onramp.DeployOnRamp( chain.DeployerKey, chain.Client, @@ -654,7 +589,7 @@ func DeployChainContracts( }, []onramp.OnRampDestChainConfigArgs{}, ) - return ContractDeploy[*onramp.OnRamp]{ + return deployment.ContractDeploy[*onramp.OnRamp]{ onRampAddr, onRamp, tx2, deployment.NewTypeAndVersion(OnRamp, deployment.Version1_6_0_dev), err2, } }) @@ -664,8 +599,8 @@ func DeployChainContracts( } e.Logger.Infow("Deployed onramp", "addr", onRamp.Address) - offRamp, err := deployContract(e.Logger, chain, ab, - func(chain deployment.Chain) ContractDeploy[*offramp.OffRamp] { + offRamp, err := deployment.DeployContract(e.Logger, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*offramp.OffRamp] { offRampAddr, tx2, offRamp, err2 := offramp.DeployOffRamp( chain.DeployerKey, chain.Client, @@ -682,7 +617,7 @@ func DeployChainContracts( }, []offramp.OffRampSourceChainConfigArgs{}, ) - return ContractDeploy[*offramp.OffRamp]{ + return deployment.ContractDeploy[*offramp.OffRamp]{ offRampAddr, offRamp, tx2, deployment.NewTypeAndVersion(OffRamp, deployment.Version1_6_0_dev), err2, } }) diff --git a/deployment/ccip/deploy_home_chain.go b/deployment/ccip/deploy_home_chain.go index c9c88d35328..d17a501783d 100644 --- a/deployment/ccip/deploy_home_chain.go +++ b/deployment/ccip/deploy_home_chain.go @@ -87,30 +87,30 @@ func MustABIEncode(abiString string, args ...interface{}) []byte { } // DeployCapReg deploys the CapabilitiesRegistry contract if it is not already deployed -// and returns a ContractDeploy struct with the address and contract instance. +// and returns a deployment.ContractDeploy struct with the address and contract instance. func DeployCapReg( lggr logger.Logger, state CCIPOnChainState, ab deployment.AddressBook, chain deployment.Chain, -) (*ContractDeploy[*capabilities_registry.CapabilitiesRegistry], error) { +) (*deployment.ContractDeploy[*capabilities_registry.CapabilitiesRegistry], error) { homeChainState, exists := state.Chains[chain.Selector] if exists { cr := homeChainState.CapabilityRegistry if cr != nil { lggr.Infow("Found CapabilitiesRegistry in chain state", "address", cr.Address().String()) - return &ContractDeploy[*capabilities_registry.CapabilitiesRegistry]{ + return &deployment.ContractDeploy[*capabilities_registry.CapabilitiesRegistry]{ Address: cr.Address(), Contract: cr, Tv: deployment.NewTypeAndVersion(CapabilitiesRegistry, deployment.Version1_0_0), }, nil } } - capReg, err := deployContract(lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*capabilities_registry.CapabilitiesRegistry] { + capReg, err := deployment.DeployContract(lggr, chain, ab, + func(chain deployment.Chain) deployment.ContractDeploy[*capabilities_registry.CapabilitiesRegistry] { crAddr, tx, cr, err2 := capabilities_registry.DeployCapabilitiesRegistry( chain.DeployerKey, chain.Client, ) - return ContractDeploy[*capabilities_registry.CapabilitiesRegistry]{ + return deployment.ContractDeploy[*capabilities_registry.CapabilitiesRegistry]{ Address: crAddr, Contract: cr, Tv: deployment.NewTypeAndVersion(CapabilitiesRegistry, deployment.Version1_0_0), Tx: tx, Err: err2, } }) @@ -130,7 +130,7 @@ func DeployHomeChain( rmnHomeDynamic rmn_home.RMNHomeDynamicConfig, nodeOps []capabilities_registry.CapabilitiesRegistryNodeOperator, nodeP2PIDsPerNodeOpAdmin map[string][][32]byte, -) (*ContractDeploy[*capabilities_registry.CapabilitiesRegistry], error) { +) (*deployment.ContractDeploy[*capabilities_registry.CapabilitiesRegistry], error) { // load existing state state, err := LoadOnchainState(e) if err != nil { @@ -143,15 +143,15 @@ func DeployHomeChain( } lggr.Infow("deployed/connected to capreg", "addr", capReg.Address) - ccipHome, err := deployContract( + ccipHome, err := deployment.DeployContract( lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*ccip_home.CCIPHome] { + func(chain deployment.Chain) deployment.ContractDeploy[*ccip_home.CCIPHome] { ccAddr, tx, cc, err2 := ccip_home.DeployCCIPHome( chain.DeployerKey, chain.Client, capReg.Address, ) - return ContractDeploy[*ccip_home.CCIPHome]{ + return deployment.ContractDeploy[*ccip_home.CCIPHome]{ Address: ccAddr, Tv: deployment.NewTypeAndVersion(CCIPHome, deployment.Version1_6_0_dev), Tx: tx, Err: err2, Contract: cc, } }) @@ -161,14 +161,14 @@ func DeployHomeChain( } lggr.Infow("deployed CCIPHome", "addr", ccipHome.Address) - rmnHome, err := deployContract( + rmnHome, err := deployment.DeployContract( lggr, chain, ab, - func(chain deployment.Chain) ContractDeploy[*rmn_home.RMNHome] { + func(chain deployment.Chain) deployment.ContractDeploy[*rmn_home.RMNHome] { rmnAddr, tx, rmn, err2 := rmn_home.DeployRMNHome( chain.DeployerKey, chain.Client, ) - return ContractDeploy[*rmn_home.RMNHome]{ + return deployment.ContractDeploy[*rmn_home.RMNHome]{ Address: rmnAddr, Tv: deployment.NewTypeAndVersion(RMNHome, deployment.Version1_6_0_dev), Tx: tx, Err: err2, Contract: rmn, } }, diff --git a/deployment/ccip/test_helpers.go b/deployment/ccip/test_helpers.go index f12a475bc2f..70be57b2f38 100644 --- a/deployment/ccip/test_helpers.go +++ b/deployment/ccip/test_helpers.go @@ -15,6 +15,7 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/pkg/errors" + cciptypes "github.com/smartcontractkit/chainlink-ccip/pkg/types/ccipocr3" commonutils "github.com/smartcontractkit/chainlink-common/pkg/utils" "github.com/smartcontractkit/chainlink-testing-framework/lib/utils/testcontext" @@ -360,7 +361,7 @@ var ( func DeployFeeds(lggr logger.Logger, ab deployment.AddressBook, chain deployment.Chain) (map[string]common.Address, error) { linkTV := deployment.NewTypeAndVersion(PriceFeed, deployment.Version1_0_0) - mockLinkFeed := func(chain deployment.Chain) ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface] { + mockLinkFeed := func(chain deployment.Chain) deployment.ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface] { linkFeed, tx, _, err1 := mock_v3_aggregator_contract.DeployMockV3Aggregator( chain.DeployerKey, chain.Client, @@ -369,12 +370,12 @@ func DeployFeeds(lggr logger.Logger, ab deployment.AddressBook, chain deployment ) aggregatorCr, err2 := aggregator_v3_interface.NewAggregatorV3Interface(linkFeed, chain.Client) - return ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface]{ + return deployment.ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface]{ Address: linkFeed, Contract: aggregatorCr, Tv: linkTV, Tx: tx, Err: multierr.Append(err1, err2), } } - mockWethFeed := func(chain deployment.Chain) ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface] { + mockWethFeed := func(chain deployment.Chain) deployment.ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface] { wethFeed, tx, _, err1 := mock_ethusd_aggregator_wrapper.DeployMockETHUSDAggregator( chain.DeployerKey, chain.Client, @@ -382,7 +383,7 @@ func DeployFeeds(lggr logger.Logger, ab deployment.AddressBook, chain deployment ) aggregatorCr, err2 := aggregator_v3_interface.NewAggregatorV3Interface(wethFeed, chain.Client) - return ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface]{ + return deployment.ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface]{ Address: wethFeed, Contract: aggregatorCr, Tv: linkTV, Tx: tx, Err: multierr.Append(err1, err2), } } @@ -409,11 +410,11 @@ func deploySingleFeed( lggr logger.Logger, ab deployment.AddressBook, chain deployment.Chain, - deployFunc func(deployment.Chain) ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface], + deployFunc func(deployment.Chain) deployment.ContractDeploy[*aggregator_v3_interface.AggregatorV3Interface], symbol TokenSymbol, ) (common.Address, string, error) { //tokenTV := deployment.NewTypeAndVersion(PriceFeed, deployment.Version1_0_0) - mockTokenFeed, err := deployContract(lggr, chain, ab, deployFunc) + mockTokenFeed, err := deployment.DeployContract(lggr, chain, ab, deployFunc) if err != nil { lggr.Errorw("Failed to deploy token feed", "err", err, "symbol", symbol) return common.Address{}, "", err @@ -677,8 +678,8 @@ func deployTransferTokenOneEnd( } } - tokenContract, err := deployContract(lggr, chain, addressBook, - func(chain deployment.Chain) ContractDeploy[*burn_mint_erc677.BurnMintERC677] { + tokenContract, err := deployment.DeployContract(lggr, chain, addressBook, + func(chain deployment.Chain) deployment.ContractDeploy[*burn_mint_erc677.BurnMintERC677] { USDCTokenAddr, tx, token, err2 := burn_mint_erc677.DeployBurnMintERC677( chain.DeployerKey, chain.Client, @@ -687,7 +688,7 @@ func deployTransferTokenOneEnd( uint8(18), big.NewInt(0).Mul(big.NewInt(1e9), big.NewInt(1e18)), ) - return ContractDeploy[*burn_mint_erc677.BurnMintERC677]{ + return deployment.ContractDeploy[*burn_mint_erc677.BurnMintERC677]{ USDCTokenAddr, token, tx, deployment.NewTypeAndVersion(BurnMintToken, deployment.Version1_0_0), err2, } }) @@ -705,8 +706,8 @@ func deployTransferTokenOneEnd( return nil, nil, err } - tokenPool, err := deployContract(lggr, chain, addressBook, - func(chain deployment.Chain) ContractDeploy[*burn_mint_token_pool.BurnMintTokenPool] { + tokenPool, err := deployment.DeployContract(lggr, chain, addressBook, + func(chain deployment.Chain) deployment.ContractDeploy[*burn_mint_token_pool.BurnMintTokenPool] { tokenPoolAddress, tx, tokenPoolContract, err2 := burn_mint_token_pool.DeployBurnMintTokenPool( chain.DeployerKey, chain.Client, @@ -715,7 +716,7 @@ func deployTransferTokenOneEnd( common.HexToAddress(rmnAddress), common.HexToAddress(routerAddress), ) - return ContractDeploy[*burn_mint_token_pool.BurnMintTokenPool]{ + return deployment.ContractDeploy[*burn_mint_token_pool.BurnMintTokenPool]{ tokenPoolAddress, tokenPoolContract, tx, deployment.NewTypeAndVersion(BurnMintTokenPool, deployment.Version1_0_0), err2, } }) diff --git a/deployment/go.mod b/deployment/go.mod index 73e23aa83f3..ed05d136993 100644 --- a/deployment/go.mod +++ b/deployment/go.mod @@ -533,4 +533,5 @@ replace ( github.com/btcsuite/btcd/btcec/v2 => github.com/btcsuite/btcd/btcec/v2 v2.3.2 // replicating the replace directive on cosmos SDK github.com/gogo/protobuf => github.com/regen-network/protobuf v1.3.3-alpha.regen.1 + github.com/sourcegraph/sourcegraph/lib => github.com/sourcegraph/sourcegraph-public-snapshot/lib v0.0.0-20240822153003-c864f15af264 ) diff --git a/deployment/helpers.go b/deployment/helpers.go index 420e9e03f06..1f0dc3064d6 100644 --- a/deployment/helpers.go +++ b/deployment/helpers.go @@ -15,6 +15,8 @@ import ( "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/crypto" "github.com/pkg/errors" + + "github.com/smartcontractkit/chainlink-common/pkg/logger" ) // OCRSecrets are used to disseminate a shared secret to OCR nodes @@ -108,3 +110,45 @@ func ParseErrorFromABI(errorString string, contractABI string) (string, error) { } return "", errors.New("error not found in ABI") } + +// ContractDeploy represents the result of an EVM contract deployment +// via an abigen Go binding. It contains all the return values +// as they are useful in different ways. +type ContractDeploy[C any] struct { + Address common.Address // We leave this incase a Go binding doesn't have Address() + Contract C // Expected to be a Go binding + Tx *types.Transaction // Incase the caller needs for example tx hash info for + Tv TypeAndVersion + Err error +} + +// DeployContract deploys an EVM contract and +// records the address in the provided address book +// if the deployment was confirmed onchain. +// Deploying and saving the address is a very common pattern +// so this helps to reduce boilerplate. +// It returns an error if the deployment failed, the tx was not +// confirmed or the address could not be saved. +func DeployContract[C any]( + lggr logger.Logger, + chain Chain, + addressBook AddressBook, + deploy func(chain Chain) ContractDeploy[C], +) (*ContractDeploy[C], error) { + contractDeploy := deploy(chain) + if contractDeploy.Err != nil { + lggr.Errorw("Failed to deploy contract", "err", contractDeploy.Err) + return nil, contractDeploy.Err + } + _, err := chain.Confirm(contractDeploy.Tx) + if err != nil { + lggr.Errorw("Failed to confirm deployment", "err", err) + return nil, err + } + err = addressBook.Save(chain.Selector, contractDeploy.Address.String(), contractDeploy.Tv) + if err != nil { + lggr.Errorw("Failed to save contract address", "err", err) + return nil, err + } + return &contractDeploy, nil +}