From 120e7788c992773fe9511e85dd1394e8f4fa4319 Mon Sep 17 00:00:00 2001 From: Sneha Agnihotri Date: Wed, 15 Nov 2023 15:40:45 -0800 Subject: [PATCH 1/5] Bump version and update CHANGELOG for core v2.7.1 Signed-off-by: Sneha Agnihotri --- VERSION | 2 +- docs/CHANGELOG.md | 2 ++ 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 24ba9a38de6..860487ca19c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.7.0 +2.7.1 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index f122b365d28..b9ab38eb3bd 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ... +## 2.7.1 - UNRELEASED + ## 2.7.0 - 2023-11-14 From 184cdfbf2274a8a01be084f127b82bc169eec67c Mon Sep 17 00:00:00 2001 From: amit-momin <108959691+amit-momin@users.noreply.github.com> Date: Thu, 16 Nov 2023 14:44:39 -0600 Subject: [PATCH 2/5] [Hotfix] Update loading next sequence map to avoid startup failure (#11319) --- common/txmgr/broadcaster.go | 77 +++++---- core/chains/evm/txmgr/broadcaster_test.go | 180 +++++++++++++++------- docs/CHANGELOG.md | 6 + 3 files changed, 181 insertions(+), 82 deletions(-) diff --git a/common/txmgr/broadcaster.go b/common/txmgr/broadcaster.go index 6512f67fe0b..1e1c0e0cff3 100644 --- a/common/txmgr/broadcaster.go +++ b/common/txmgr/broadcaster.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "fmt" + "slices" "sync" "time" @@ -243,10 +244,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) star eb.sequenceLock.Lock() defer eb.sequenceLock.Unlock() - eb.nextSequenceMap, err = eb.loadNextSequenceMap(eb.enabledAddresses) - if err != nil { - return errors.Wrap(err, "Broadcaster: failed to load next sequence map") - } + eb.nextSequenceMap = eb.loadNextSequenceMap(eb.enabledAddresses) eb.isStarted = true return nil @@ -326,30 +324,38 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) txIn } // Load the next sequence map using the tx table or on-chain (if not found in tx table) -func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) loadNextSequenceMap(addresses []ADDR) (map[ADDR]SEQ, error) { +func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) loadNextSequenceMap(addresses []ADDR) map[ADDR]SEQ { ctx, cancel := eb.chStop.NewCtx() defer cancel() nextSequenceMap := make(map[ADDR]SEQ) for _, address := range addresses { - // Get the highest sequence from the tx table - // Will need to be incremented since this sequence is already used - seq, err := eb.txStore.FindLatestSequence(ctx, address, eb.chainID) - if err != nil { - // Look for nonce on-chain if no tx found for address in TxStore or if error occurred - // Returns the nonce that should be used for the next transaction so no need to increment - seq, err = eb.client.PendingSequenceAt(ctx, address) - if err != nil { - return nil, errors.New("failed to retrieve next sequence from on-chain causing failure to load next sequence map on broadcaster startup") - } - + seq, err := eb.getSequenceForAddr(ctx, address) + if err == nil { nextSequenceMap[address] = seq - } else { - nextSequenceMap[address] = eb.generateNextSequence(seq) } } - return nextSequenceMap, nil + return nextSequenceMap +} + +func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) getSequenceForAddr(ctx context.Context, address ADDR) (seq SEQ, err error) { + // Get the highest sequence from the tx table + // Will need to be incremented since this sequence is already used + seq, err = eb.txStore.FindLatestSequence(ctx, address, eb.chainID) + if err == nil { + seq = eb.generateNextSequence(seq) + return seq, nil + } + // Look for nonce on-chain if no tx found for address in TxStore or if error occurred + // Returns the nonce that should be used for the next transaction so no need to increment + seq, err = eb.client.PendingSequenceAt(ctx, address) + if err == nil { + return seq, nil + } + eb.logger.Criticalw("failed to retrieve next sequence from on-chain for address: ", "address", address.String()) + return seq, err + } func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) newSequenceSyncBackoff() backoff.Backoff { @@ -432,7 +438,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) moni // syncSequence tries to sync the key sequence, retrying indefinitely until success or stop signal is sent func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) SyncSequence(ctx context.Context, addr ADDR) { sequenceSyncRetryBackoff := eb.newSequenceSyncBackoff() - localSequence, err := eb.GetNextSequence(addr) + localSequence, err := eb.GetNextSequence(ctx, addr) // Address not found in map so skip sync if err != nil { eb.logger.Criticalw("Failed to retrieve local next sequence for address", "address", addr.String(), "err", err) @@ -646,7 +652,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) hand observeTimeUntilBroadcast(eb.chainID, etx.CreatedAt, time.Now()) // Check if from_address exists in map to ensure it is valid before broadcasting var sequence SEQ - sequence, err = eb.GetNextSequence(etx.FromAddress) + sequence, err = eb.GetNextSequence(ctx, etx.FromAddress) if err != nil { return err, true } @@ -704,7 +710,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) hand // Check if from_address exists in map to ensure it is valid before broadcasting var sequence SEQ - sequence, err = eb.GetNextSequence(etx.FromAddress) + sequence, err = eb.GetNextSequence(ctx, etx.FromAddress) if err != nil { return err, true } @@ -741,7 +747,7 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) next return nil, errors.Wrap(err, "findNextUnstartedTransactionFromAddress failed") } - sequence, err := eb.GetNextSequence(etx.FromAddress) + sequence, err := eb.GetNextSequence(ctx, etx.FromAddress) if err != nil { return nil, err } @@ -826,15 +832,32 @@ func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) save } // Used to get the next usable sequence for a transaction -func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) GetNextSequence(address ADDR) (seq SEQ, err error) { +func (eb *Broadcaster[CHAIN_ID, HEAD, ADDR, TX_HASH, BLOCK_HASH, SEQ, FEE]) GetNextSequence(ctx context.Context, address ADDR) (seq SEQ, err error) { eb.sequenceLock.Lock() defer eb.sequenceLock.Unlock() // Get next sequence from map seq, exists := eb.nextSequenceMap[address] - if !exists { - return seq, errors.New(fmt.Sprint("address not found in next sequence map: ", address)) + if exists { + return seq, nil + } + + eb.logger.Infow("address not found in local next sequence map. Attempting to search and populate sequence.", "address", address.String()) + // Check if address is in the enabled address list + if !slices.Contains(eb.enabledAddresses, address) { + return seq, fmt.Errorf("address disabled: %s", address) } - return seq, nil + + // Try to retrieve next sequence from tx table or on-chain to load the map + // A scenario could exist where loading the map during startup failed (e.g. All configured RPC's are unreachable at start) + // The expectation is that the node does not fail startup so sequences need to be loaded during runtime + foundSeq, err := eb.getSequenceForAddr(ctx, address) + if err != nil { + return seq, fmt.Errorf("failed to find next sequence for address: %s", address) + } + + // Set sequence in map + eb.nextSequenceMap[address] = foundSeq + return foundSeq, nil } // Used to increment the sequence in the mapping to have the next usable one available for the next transaction diff --git a/core/chains/evm/txmgr/broadcaster_test.go b/core/chains/evm/txmgr/broadcaster_test.go index 6f9308548b3..dd2a124be49 100644 --- a/core/chains/evm/txmgr/broadcaster_test.go +++ b/core/chains/evm/txmgr/broadcaster_test.go @@ -141,6 +141,43 @@ func TestEthBroadcaster_Lifecycle(t *testing.T) { require.NoError(t, eb.XXXTestCloseInternal()) } +// Failure to load next sequnce map should not fail Broadcaster startup +func TestEthBroadcaster_LoadNextSequenceMapFailure_StartupSuccess(t *testing.T) { + db := pgtest.NewSqlxDB(t) + cfg := configtest.NewTestGeneralConfig(t) + eventBroadcaster := cltest.NewEventBroadcaster(t, cfg.Database().URL()) + err := eventBroadcaster.Start(testutils.Context(t)) + require.NoError(t, err) + t.Cleanup(func() { assert.NoError(t, eventBroadcaster.Close()) }) + txStore := cltest.NewTestTxStore(t, db, cfg.Database()) + evmcfg := evmtest.NewChainScopedConfig(t, cfg) + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + ethKeyStore := cltest.NewKeyStore(t, db, cfg.Database()).Eth() + cltest.MustInsertRandomKeyReturningState(t, ethKeyStore) + estimator := gasmocks.NewEvmFeeEstimator(t) + txBuilder := txmgr.NewEvmTxAttemptBuilder(*ethClient.ConfiguredChainID(), evmcfg.EVM().GasEstimator(), ethKeyStore, estimator) + ethClient.On("PendingNonceAt", mock.Anything, mock.Anything).Return(uint64(0), errors.New("Getting on-chain nonce failed")) + eb := txmgr.NewEvmBroadcaster( + txStore, + txmgr.NewEvmTxmClient(ethClient), + txmgr.NewEvmTxmConfig(evmcfg.EVM()), + txmgr.NewEvmTxmFeeConfig(evmcfg.EVM().GasEstimator()), + evmcfg.EVM().Transactions(), + evmcfg.Database().Listener(), + ethKeyStore, + eventBroadcaster, + txBuilder, + nil, + logger.TestLogger(t), + &testCheckerFactory{}, + false, + ) + + // Instance starts without error even if loading next sequence map fails + err = eb.Start(testutils.Context(t)) + require.NoError(t, err) +} + func TestEthBroadcaster_ProcessUnstartedEthTxs_Success(t *testing.T) { db := pgtest.NewSqlxDB(t) cfg := configtest.NewTestGeneralConfig(t) @@ -961,7 +998,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_ResumingFromCrash(t *testing.T) { } func getLocalNextNonce(t *testing.T, eb *txmgr.Broadcaster, fromAddress gethCommon.Address) uint64 { - n, err := eb.GetNextSequence(fromAddress) + n, err := eb.GetNextSequence(testutils.Context(t), fromAddress) require.NoError(t, err) require.NotNil(t, n) return uint64(n) @@ -987,6 +1024,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient := evmtest.NewEthClientMockWithDefaultChain(t) ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, ethKeyStore, evmcfg, &testCheckerFactory{}, false) + ctx := testutils.Context(t) require.NoError(t, utils.JustError(db.Exec(`SET CONSTRAINTS pipeline_runs_pipeline_spec_id_fkey DEFERRED`))) @@ -998,7 +1036,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Successful, errors.New("replacement transaction underpriced")).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1034,7 +1072,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.Fatal, errors.New(fatalErrorExample)).Once() - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1051,7 +1089,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { // Check that the key had its nonce reset var nonce evmtypes.Nonce - nonce, err = eb.GetNextSequence(fromAddress) + nonce, err = eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) // Saved NextNonce must be the same as before because this transaction // was not accepted by the eth node and never can be @@ -1084,7 +1122,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.Fatal, errors.New(fatalErrorExample)).Once() - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), "something exploded in the callback") assert.True(t, retryable) @@ -1106,7 +1144,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Fatal, errors.New(fatalErrorExample)).Once() { - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) } @@ -1124,7 +1162,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(localNextNonce), nil).Once() eb2 := txmgr.NewEvmBroadcaster(txStore, txmgr.NewEvmTxmClient(ethClient), txmgr.NewEvmTxmConfig(evmcfg.EVM()), txmgr.NewEvmTxmFeeConfig(evmcfg.EVM().GasEstimator()), evmcfg.EVM().Transactions(), evmcfg.Database().Listener(), ethKeyStore, eventBroadcaster, txBuilder, nil, lggr, &testCheckerFactory{}, false) require.NoError(t, err) - retryable, err := eb2.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb2.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) }) @@ -1146,7 +1184,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { // another node even if the primary one returns "exceeds the configured // cap" - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) assert.Contains(t, err.Error(), "tx fee (1.10 ether) exceeds the configured cap (1.00 ether)") assert.Contains(t, err.Error(), "error while sending transaction") @@ -1166,7 +1204,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { // Check that the key had its nonce reset var nonce evmtypes.Nonce - nonce, err = eb.GetNextSequence(fromAddress) + nonce, err = eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) // Saved NextNonce must be the same as before because this transaction // was not accepted by the eth node and never can be @@ -1175,7 +1213,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { // On the second try, the tx has been accepted into the mempool ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(localNextNonce+1), nil).Once() - retryable, err = eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err = eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1203,7 +1241,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(localNextNonce), nil).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), retryableErrorExample) assert.True(t, retryable) @@ -1226,7 +1264,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.Successful, nil).Once() - retryable, err = eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err = eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1254,7 +1292,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), errors.New("pending nonce fetch failed")).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), retryableErrorExample) require.Contains(t, err.Error(), "pending nonce fetch failed") @@ -1278,7 +1316,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.Successful, nil).Once() - retryable, err = eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err = eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1307,7 +1345,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(localNextNonce+1), nil).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.NoError(t, err) assert.False(t, retryable) @@ -1349,7 +1387,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Successful, nil).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.NoError(t, err) assert.False(t, retryable) @@ -1385,7 +1423,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Retryable, failedToReachNodeError).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) assert.Contains(t, err.Error(), "context deadline exceeded") assert.True(t, retryable) @@ -1416,7 +1454,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Successful, errors.New(temporarilyUnderpricedError)).Once() // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) assert.NoError(t, err) assert.False(t, retryable) @@ -1456,7 +1494,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Underpriced, errors.New(underpricedError)).Once() // Do the thing - retryable, err := eb2.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb2.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), "bumped fee price of 20 gwei is equal to original fee price of 20 gwei. ACTION REQUIRED: This is a configuration error, you must increase either FeeEstimator.BumpPercent or FeeEstimator.BumpMin") assert.True(t, retryable) @@ -1473,7 +1511,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.InsufficientFunds, errors.New(insufficientEthError)).Once() - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) assert.Contains(t, err.Error(), "insufficient funds for transfer") assert.True(t, retryable) @@ -1503,7 +1541,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce }), fromAddress).Return(clienttypes.Retryable, errors.New(nonceGapError)).Once() - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) assert.Contains(t, err.Error(), nonceGapError) assert.True(t, retryable) @@ -1548,7 +1586,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { }), fromAddress).Return(clienttypes.Underpriced, errors.New(underpricedError)).Once() // Check gas tip cap verification - retryable, err := eb2.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb2.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), "bumped gas tip cap of 1 wei is less than or equal to original gas tip cap of 1 wei") assert.True(t, retryable) @@ -1572,7 +1610,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(localNextNonce, nil).Once() eb2 := NewTestEthBroadcaster(t, txStore, ethClient, ethKeyStore, evmcfg2, &testCheckerFactory{}, false) - retryable, err := eb2.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb2.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), "specified gas tip cap of 0 is below min configured gas tip of 1 wei for key") assert.True(t, retryable) @@ -1599,7 +1637,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_Errors(t *testing.T) { return tx.Nonce() == localNextNonce && tx.GasTipCap().Cmp(big.NewInt(0).Add(gasTipCapDefault.ToInt(), big.NewInt(0).Mul(evmcfg2.EVM().GasEstimator().BumpMin().ToInt(), big.NewInt(2)))) == 0 }), fromAddress).Return(clienttypes.Successful, nil).Once() - retryable, err = eb2.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err = eb2.ProcessUnstartedTxs(ctx, fromAddress) require.NoError(t, err) assert.False(t, retryable) @@ -1631,7 +1669,8 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_KeystoreErrors(t *testing.T) { kst.On("EnabledAddressesForChain", &cltest.FixtureChainID).Return(addresses, nil).Once() ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, kst, evmcfg, &testCheckerFactory{}, false) - _, err := eb.GetNextSequence(fromAddress) + ctx := testutils.Context(t) + _, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) t.Run("tx signing fails", func(t *testing.T) { @@ -1645,7 +1684,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_KeystoreErrors(t *testing.T) { })).Return(&tx, errors.New("could not sign transaction")) // Do the thing - retryable, err := eb.ProcessUnstartedTxs(testutils.Context(t), fromAddress) + retryable, err := eb.ProcessUnstartedTxs(ctx, fromAddress) require.Error(t, err) require.Contains(t, err.Error(), "could not sign transaction") assert.True(t, retryable) @@ -1659,7 +1698,7 @@ func TestEthBroadcaster_ProcessUnstartedEthTxs_KeystoreErrors(t *testing.T) { // Check that the key did not have its nonce incremented var nonce types.Nonce - nonce, err = eb.GetNextSequence(fromAddress) + nonce, err = eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) require.Equal(t, int64(localNonce), int64(nonce)) }) @@ -1697,12 +1736,13 @@ func TestEthBroadcaster_IncrementNextNonce(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, kst, evmcfg, &testCheckerFactory{}, false) - nonce, err := eb.GetNextSequence(fromAddress) + ctx := testutils.Context(t) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) eb.IncrementNextSequence(fromAddress, nonce) // Nonce bumped to 1 - nonce, err = eb.GetNextSequence(fromAddress) + nonce, err = eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) require.Equal(t, int64(1), int64(nonce)) } @@ -1784,7 +1824,7 @@ func TestEthBroadcaster_SyncNonce(t *testing.T) { kst.On("EnabledAddressesForChain", &cltest.FixtureChainID).Return(addresses, nil).Once() ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), nil).Once() eb := txmgr.NewEvmBroadcaster(txStore, txmgr.NewEvmTxmClient(ethClient), evmTxmCfg, txmgr.NewEvmTxmFeeConfig(ge), evmcfg.EVM().Transactions(), cfg.Database().Listener(), kst, eventBroadcaster, txBuilder, nil, lggr, checkerFactory, false) - err := eb.Start(testutils.Context(t)) + err := eb.Start(ctx) assert.NoError(t, err) defer func() { assert.NoError(t, eb.Close()) }() @@ -1810,12 +1850,12 @@ func TestEthBroadcaster_SyncNonce(t *testing.T) { testutils.WaitForLogMessage(t, observed, "Fast-forward sequence") // Check nextSequenceMap to make sure it has correct nonce assigned - nonce, err := eb.GetNextSequence(fromAddress) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) - assert.Equal(t, strconv.FormatUint(ethNodeNonce, 10), nonce.String()) + require.Equal(t, strconv.FormatUint(ethNodeNonce, 10), nonce.String()) // The disabled key did not get updated - _, err = eb.GetNextSequence(disabledAddress) + _, err = eb.GetNextSequence(ctx, disabledAddress) require.Error(t, err) }) @@ -1844,19 +1884,19 @@ func TestEthBroadcaster_SyncNonce(t *testing.T) { testutils.WaitForLogMessage(t, observed, "Fast-forward sequence") // Check keyState to make sure it has correct nonce assigned - nonce, err := eb.GetNextSequence(fromAddress) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) assert.Equal(t, int64(ethNodeNonce), int64(nonce)) // The disabled key did not get updated - _, err = eb.GetNextSequence(disabledAddress) + _, err = eb.GetNextSequence(ctx, disabledAddress) require.Error(t, err) }) - } func Test_LoadSequenceMap(t *testing.T) { t.Parallel() + ctx := testutils.Context(t) t.Run("set next nonce using entries from tx table", func(t *testing.T) { db := pgtest.NewSqlxDB(t) cfg := configtest.NewTestGeneralConfig(t) @@ -1871,9 +1911,9 @@ func Test_LoadSequenceMap(t *testing.T) { cltest.MustInsertUnconfirmedEthTx(t, txStore, int64(1), fromAddress) eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) - nonce, err := eb.GetNextSequence(fromAddress) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) - assert.Equal(t, int64(2), int64(nonce)) + require.Equal(t, int64(2), int64(nonce)) }) t.Run("set next nonce using client when not found in tx table", func(t *testing.T) { @@ -1889,9 +1929,9 @@ func Test_LoadSequenceMap(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(10), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) - nonce, err := eb.GetNextSequence(fromAddress) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) - assert.Equal(t, int64(10), int64(nonce)) + require.Equal(t, int64(10), int64(nonce)) }) } @@ -1910,25 +1950,53 @@ func Test_NextNonce(t *testing.T) { _, addr1 := cltest.MustInsertRandomKey(t, ks) ethClient.On("PendingNonceAt", mock.Anything, addr1).Return(uint64(randNonce), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) - + ctx := testutils.Context(t) cltest.MustInsertRandomKey(t, ks, *utils.NewBig(testutils.FixtureChainID)) - nonce, err := eb.GetNextSequence(addr1) + nonce, err := eb.GetNextSequence(ctx, addr1) require.NoError(t, err) - assert.Equal(t, randNonce, int64(nonce)) + require.Equal(t, randNonce, int64(nonce)) randAddr1 := utils.RandomAddress() - _, err = eb.GetNextSequence(randAddr1) + _, err = eb.GetNextSequence(ctx, randAddr1) require.Error(t, err) - assert.Contains(t, err.Error(), fmt.Sprintf("address not found in next sequence map: %s", randAddr1.Hex())) + require.Contains(t, err.Error(), fmt.Sprintf("address disabled: %s", randAddr1.Hex())) randAddr2 := utils.RandomAddress() - _, err = eb.GetNextSequence(randAddr2) + _, err = eb.GetNextSequence(ctx, randAddr2) require.Error(t, err) - assert.Contains(t, err.Error(), fmt.Sprintf("address not found in next sequence map: %s", randAddr2.Hex())) + require.Contains(t, err.Error(), fmt.Sprintf("address disabled: %s", randAddr2.Hex())) } +func Test_SetNonceAfterInit(t *testing.T) { + t.Parallel() + + db := pgtest.NewSqlxDB(t) + cfg := configtest.NewTestGeneralConfig(t) + txStore := cltest.NewTestTxStore(t, db, cfg.Database()) + ks := cltest.NewKeyStore(t, db, cfg.Database()).Eth() + + ethClient := evmtest.NewEthClientMockWithDefaultChain(t) + evmcfg := evmtest.NewChainScopedConfig(t, cfg) + checkerFactory := &txmgr.CheckerFactory{Client: ethClient} + randNonce := testutils.NewRandomPositiveInt64() + _, addr1 := cltest.MustInsertRandomKey(t, ks) + ethClient.On("PendingNonceAt", mock.Anything, addr1).Return(uint64(0), errors.New("failed to retrieve nonce at startup")).Once() + ethClient.On("PendingNonceAt", mock.Anything, addr1).Return(uint64(randNonce), nil).Once() + eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) + + ctx := testutils.Context(t) + nonce, err := eb.GetNextSequence(ctx, addr1) + require.NoError(t, err) + require.Equal(t, randNonce, int64(nonce)) + + // Test that the new nonce is set in the map and does not need a client call to retrieve on subsequent calls + nonce, err = eb.GetNextSequence(ctx, addr1) + require.NoError(t, err) + require.Equal(t, randNonce, int64(nonce)) +} + func Test_IncrementNextNonce(t *testing.T) { t.Parallel() @@ -1945,26 +2013,27 @@ func Test_IncrementNextNonce(t *testing.T) { ethClient.On("PendingNonceAt", mock.Anything, addr1).Return(uint64(randNonce), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) - nonce, err := eb.GetNextSequence(addr1) + ctx := testutils.Context(t) + nonce, err := eb.GetNextSequence(ctx, addr1) require.NoError(t, err) eb.IncrementNextSequence(addr1, nonce) - nonce, err = eb.GetNextSequence(addr1) + nonce, err = eb.GetNextSequence(ctx, addr1) require.NoError(t, err) assert.Equal(t, randNonce+1, int64(nonce)) eb.IncrementNextSequence(addr1, nonce) - nonce, err = eb.GetNextSequence(addr1) + nonce, err = eb.GetNextSequence(ctx, addr1) require.NoError(t, err) assert.Equal(t, randNonce+2, int64(nonce)) randAddr1 := utils.RandomAddress() - _, err = eb.GetNextSequence(randAddr1) + _, err = eb.GetNextSequence(ctx, randAddr1) require.Error(t, err) - assert.Contains(t, err.Error(), fmt.Sprintf("address not found in next sequence map: %s", randAddr1.Hex())) + assert.Contains(t, err.Error(), fmt.Sprintf("address disabled: %s", randAddr1.Hex())) // verify it didnt get changed by any erroring calls - nonce, err = eb.GetNextSequence(addr1) + nonce, err = eb.GetNextSequence(ctx, addr1) require.NoError(t, err) assert.Equal(t, randNonce+2, int64(nonce)) } @@ -1983,14 +2052,15 @@ func Test_SetNextNonce(t *testing.T) { _, fromAddress := cltest.MustInsertRandomKey(t, ks) ethClient.On("PendingNonceAt", mock.Anything, fromAddress).Return(uint64(0), nil).Once() eb := NewTestEthBroadcaster(t, txStore, ethClient, ks, evmcfg, checkerFactory, false) + ctx := testutils.Context(t) t.Run("update next nonce", func(t *testing.T) { - nonce, err := eb.GetNextSequence(fromAddress) + nonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) assert.Equal(t, int64(0), int64(nonce)) eb.SetNextSequence(fromAddress, evmtypes.Nonce(24)) - newNextNonce, err := eb.GetNextSequence(fromAddress) + newNextNonce, err := eb.GetNextSequence(ctx, fromAddress) require.NoError(t, err) assert.Equal(t, int64(24), int64(newNextNonce)) }) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index b9ab38eb3bd..a58a2eb4a76 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -13,6 +13,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## 2.7.1 - UNRELEASED +### Fixed + +- Fixed a bug that causes the node to shutdown if all configured RPC's are unreachable during startup. + ## 2.7.0 - 2023-11-14 @@ -39,7 +43,9 @@ These will eventually replace `TelemetryIngress.URL` and `TelemetryIngress.Serve - `P2P.V2` is now enabled (`Enabled = true`) by default. ### Upcoming Required Configuration Changes + Starting in `v2.9.0`: + - `TelemetryIngress.URL` and `TelemetryIngress.ServerPubKey` will no longer be allowed. Any TOML configuration that sets this fields will prevent the node from booting. These fields will be replaced by `[[TelemetryIngress.Endpoints]]` - `P2P.V1` will no longer be supported and must not be set in TOML configuration in order to boot. Use `P2P.V2` instead. If you are using both, `V1` can simply be removed. From a564461f6c2eedb3b324b37d8cdf56f7cec95a69 Mon Sep 17 00:00:00 2001 From: Sneha Agnihotri Date: Tue, 21 Nov 2023 11:01:31 -0800 Subject: [PATCH 3/5] Finalize date on changelog for 2.7.1 --- docs/CHANGELOG.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index a58a2eb4a76..50c8a389932 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -11,14 +11,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ... -## 2.7.1 - UNRELEASED + + +## 2.7.1 - 2023-11-21 ### Fixed - Fixed a bug that causes the node to shutdown if all configured RPC's are unreachable during startup. - - ## 2.7.0 - 2023-11-14 ### Added From 39ba467207bf2f5cf73f5a067616261bcba8499c Mon Sep 17 00:00:00 2001 From: anirudhwarrier <12178754+anirudhwarrier@users.noreply.github.com> Date: Thu, 23 Nov 2023 10:49:53 +0400 Subject: [PATCH 4/5] fix failing CI tests for 2.7.1 --- integration-tests/go.mod | 2 +- integration-tests/go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/integration-tests/go.mod b/integration-tests/go.mod index 7dd2d017785..b90ba784a38 100644 --- a/integration-tests/go.mod +++ b/integration-tests/go.mod @@ -21,7 +21,7 @@ require ( github.com/segmentio/ksuid v1.0.4 github.com/slack-go/slack v0.12.2 github.com/smartcontractkit/chainlink-env v0.38.3 - github.com/smartcontractkit/chainlink-testing-framework v1.17.12-0.20231018101901-23824db88d36 + github.com/smartcontractkit/chainlink-testing-framework v1.17.13 github.com/smartcontractkit/chainlink/v2 v2.0.0-00010101000000-000000000000 github.com/smartcontractkit/libocr v0.0.0-20231020123319-d255366a6545 github.com/smartcontractkit/ocr2keepers v0.7.27 diff --git a/integration-tests/go.sum b/integration-tests/go.sum index 3be74077277..8c3dc2c65c2 100644 --- a/integration-tests/go.sum +++ b/integration-tests/go.sum @@ -2370,8 +2370,8 @@ github.com/smartcontractkit/chainlink-solana v1.0.3-0.20231023133638-72f4e799ab0 github.com/smartcontractkit/chainlink-solana v1.0.3-0.20231023133638-72f4e799ab05/go.mod h1:o0Pn1pbaUluboaK6/yhf8xf7TiFCkyFl6WUOdwqamuU= github.com/smartcontractkit/chainlink-starknet/relayer v0.0.1-beta-test.0.20231024133459-1ef3a11319eb h1:HiluOfEVGOQTM6BTDImOqYdMZZ7qq7fkZ3TJdmItNr8= github.com/smartcontractkit/chainlink-starknet/relayer v0.0.1-beta-test.0.20231024133459-1ef3a11319eb/go.mod h1:/30flFG4L/iCYAFeA3DUzR0xuHSxAMONiWTzyzvsNwo= -github.com/smartcontractkit/chainlink-testing-framework v1.17.12-0.20231018101901-23824db88d36 h1:ow84QG8vEHMvfjGg0RF8HNYh80WcHci6PIenXyY6K8Y= -github.com/smartcontractkit/chainlink-testing-framework v1.17.12-0.20231018101901-23824db88d36/go.mod h1:RWlmjwnjIGbQAnRfKwe02Ife82nNI3rZmdI0zgkfbyk= +github.com/smartcontractkit/chainlink-testing-framework v1.17.13 h1:C8E+P3/ElUxmsIHE0fOd1hDoSc3AbbqcePmN9sNKYdM= +github.com/smartcontractkit/chainlink-testing-framework v1.17.13/go.mod h1:RWlmjwnjIGbQAnRfKwe02Ife82nNI3rZmdI0zgkfbyk= github.com/smartcontractkit/go-plugin v0.0.0-20231003134350-e49dad63b306 h1:ko88+ZznniNJZbZPWAvHQU8SwKAdHngdDZ+pvVgB5ss= github.com/smartcontractkit/go-plugin v0.0.0-20231003134350-e49dad63b306/go.mod h1:w1sAEES3g3PuV/RzUrgow20W2uErMly84hhD3um1WL4= github.com/smartcontractkit/grpc-proxy v0.0.0-20230731113816-f1be6620749f h1:hgJif132UCdjo8u43i7iPN1/MFnu49hv7lFGFftCHKU= From 19e3b4ed835bb664cbb642af4c8355307153f93a Mon Sep 17 00:00:00 2001 From: Tate Date: Mon, 27 Nov 2023 09:15:09 -0700 Subject: [PATCH 5/5] bump solana version in tests built --- .github/workflows/integration-tests.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml index aadb14f1284..b1a1e663569 100644 --- a/.github/workflows/integration-tests.yml +++ b/.github/workflows/integration-tests.yml @@ -747,7 +747,7 @@ jobs: uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3.6.0 with: repository: smartcontractkit/chainlink-solana - ref: 23816fcf7d380a30c87b6d87e4fb0ca94419b259 # swtich back to this after the next solana release${{ needs.get_solana_sha.outputs.sha }} + ref: a28100b7f2954604a8ca2ff9ec7bccc6ec952953 # swtich back to this after the next solana release${{ needs.get_solana_sha.outputs.sha }} - name: Build Test Image if: needs.changes.outputs.src == 'true' && needs.solana-test-image-exists.outputs.exists == 'false' uses: ./.github/actions/build-test-image @@ -805,7 +805,7 @@ jobs: uses: actions/checkout@f43a0e5ff2bd294095638e18286ca9a3d1956744 # v3.6.0 with: repository: smartcontractkit/chainlink-solana - ref: ${{ needs.get_solana_sha.outputs.sha }} + ref: a28100b7f2954604a8ca2ff9ec7bccc6ec952953 # temporarily using specific commit for release branch ${{ needs.get_solana_sha.outputs.sha }} - name: Run Setup if: needs.changes.outputs.src == 'true' uses: smartcontractkit/chainlink-github-actions/chainlink-testing-framework/setup-run-tests-environment@eccde1970eca69f079d3efb3409938a72ade8497 # v2.2.13