diff --git a/node/pkg/watchers/evm/by_transaction.go b/node/pkg/watchers/evm/by_transaction.go index 3f46723787..1d6cdfc27a 100644 --- a/node/pkg/watchers/evm/by_transaction.go +++ b/node/pkg/watchers/evm/by_transaction.go @@ -52,8 +52,9 @@ func MessageEventsForTransaction( tx eth_common.Hash) (*types.Receipt, uint64, []*common.MessagePublication, error) { // Get transactions logs from transaction + // API only returns transactions that have been included in a block. Nothing in the mempool receipt, err := ethConn.TransactionReceipt(ctx, tx) - if err != nil { + if receipt == nil || err != nil { return nil, 0, nil, fmt.Errorf("failed to get transaction receipt: %w", err) } diff --git a/node/pkg/watchers/evm/reobserve_test.go b/node/pkg/watchers/evm/reobserve_test.go new file mode 100644 index 0000000000..2ad67dd929 --- /dev/null +++ b/node/pkg/watchers/evm/reobserve_test.go @@ -0,0 +1,356 @@ +package evm + +import ( + "context" + "errors" + "testing" + + "github.com/certusone/wormhole/node/pkg/common" + "github.com/certusone/wormhole/node/pkg/txverifier" + "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors" + eth_common "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" + "github.com/stretchr/testify/require" + "github.com/wormhole-foundation/wormhole/sdk/vaa" +) + +func TestReobserveSingleMessage(t *testing.T) { + tests := []struct { + name string + consistencyLevel uint8 + }{ + {"finalized", vaa.ConsistencyLevelFinalized}, + {"safe", vaa.ConsistencyLevelSafe}, + {"instant", vaa.ConsistencyLevelPublishImmediately}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log := newValidWormholeLog(t, testBlockNumber, tc.consistencyLevel) + mock.seedLog(log) + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(1), numObs) + require.Equal(t, 1, len(msgC)) + + ev, err := mock.ParseLogMessagePublished(*log) + require.NoError(t, err) + msg := recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + require.Equal(t, common.NotVerified, msg.VerificationState()) + assertMessageMatchesEvent(t, msg, ev) + }) + } +} + +// TestReobserveSingleMessageEarly covers cases where the log's block number is ahead of the +// current chain head for the message's consistency level, so the reobservation is dropped. +func TestReobserveSingleMessageEarly(t *testing.T) { + tests := []struct { + name string + consistencyLevel uint8 + finalizedBlockNum uint64 + safeBlockNum uint64 + }{ + {"finalized", vaa.ConsistencyLevelFinalized, testBlockNumber - 1, testSafeBlockNum}, + {"safe", vaa.ConsistencyLevelSafe, testFinalizedBlockNum, testBlockNumber - 1}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log := newValidWormholeLog(t, testBlockNumber, tc.consistencyLevel) + mock.seedLog(log) + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log.TxHash.Bytes(), mock, tc.finalizedBlockNum, tc.safeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(0), numObs) + require.Equal(t, 0, len(msgC)) + }) + } +} + +func TestReobserveInvalidChainId(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + numObs, err := w.handleReobservationRequest(context.TODO(), vaa.ChainIDSolana, []byte{}, mock, testFinalizedBlockNum, testSafeBlockNum) + require.Error(t, err) + require.Equal(t, uint32(0), numObs) + require.Equal(t, 0, len(msgC)) +} + +func TestReobserveReceiptError(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + txHash := eth_common.HexToHash("0x1234") + mock.errors[txHash] = errors.New("rpc failure") + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, txHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.Error(t, err) + require.Equal(t, uint32(0), numObs) + require.Equal(t, 0, len(msgC)) +} + +func TestReobserveFailedTransactionStatus(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log := newValidWormholeLog(t, testBlockNumber, vaa.ConsistencyLevelFinalized) + receipt := newTestReceipt(log.BlockNumber, []*types.Log{log}) + receipt.Status = 0 + mock.receipts[log.TxHash] = receipt + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.Error(t, err) + require.Equal(t, uint32(0), numObs) + require.Equal(t, 0, len(msgC)) +} + +// TestReobserveLogSkipped covers receipt logs that should not produce a MessagePublication. +// Each subtest starts with a fully valid log and mutates a single field so the log is rejected +// by by_transaction.go. All cases should return numObs=0 with no error. +func TestReobserveLogSkipped(t *testing.T) { + tests := []struct { + name string + mutate func(*types.Log) + }{ + { + name: "wrong_contract_address", + mutate: func(l *types.Log) { + l.Address = eth_common.HexToAddress("0x396343362be2A4dA1cE0C1C210945346fb82Aa49") + }, + }, + { + name: "wrong_event_topic", + mutate: func(l *types.Log) { + l.Topics[0] = eth_common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + }, + }, + { + name: "removed_flag_set", + mutate: func(l *types.Log) { + l.Removed = true + }, + }, + { + name: "empty_topics", + mutate: func(l *types.Log) { + l.Topics = []eth_common.Hash{} + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log := newValidWormholeLog(t, testBlockNumber, vaa.ConsistencyLevelFinalized) + tc.mutate(log) + mock.seedLog(log) + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(0), numObs) + require.Equal(t, 0, len(msgC)) + }) + } +} + +// TestReobserveDeterministicOrdering verifies processing the same message +// always leads to the same result. This test is meant to catch non-determinism issues +// that get added to this code path. +func TestReobserveDeterministicOrdering(t *testing.T) { + const numEvents = 20 + const numRuns = 5 + + var firstRun []uint64 + for run := 0; run < numRuns; run++ { + w, mock, _ := newTestWatcher(t) + msgC := make(chan *common.MessagePublication, numEvents) + w.msgC = msgC + + logs := make([]*types.Log, numEvents) + for i := 0; i < numEvents; i++ { + log := newTestLog(t, testLogParams{ + sender: testTokenBridge, + contractAddr: testEmitter, + sequence: uint64(i), // #nosec G115 -- test-only + consistencyLevel: vaa.ConsistencyLevelFinalized, + blockNumber: testBlockNumber, + }) + logs[i] = &log + } + + receipt := newTestReceipt(testBlockNumber, logs) + mock.receipts[logs[0].TxHash] = receipt + mock.blockTimes[receipt.BlockHash] = testBlockTime + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, logs[0].TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(numEvents), numObs) + + seqs := make([]uint64, numEvents) + for i := 0; i < numEvents; i++ { + seqs[i] = recvMsg(t, msgC).Sequence + } + + if run == 0 { + firstRun = seqs + continue + } + require.Equal(t, firstRun, seqs, "run %d produced a different ordering than run 0", run) + } +} + +// TestReobserve1KEventsInReceipt verifies that a receipt containing 1000 LogMessagePublished +// events produces 1000 published MessagePublications with sequences preserved in order. +func TestReobserve1KEventsInReceipt(t *testing.T) { + const numEvents = 1000 + + w, mock, _ := newTestWatcher(t) + msgC := make(chan *common.MessagePublication, numEvents) + w.msgC = msgC + + logs := make([]*types.Log, numEvents) + for i := 0; i < numEvents; i++ { + log := newTestLog(t, testLogParams{ + sender: testTokenBridge, + contractAddr: testEmitter, + sequence: uint64(i), // #nosec G115 -- test-only + consistencyLevel: vaa.ConsistencyLevelFinalized, + blockNumber: testBlockNumber, + }) + logs[i] = &log + } + + receipt := newTestReceipt(testBlockNumber, logs) + mock.receipts[logs[0].TxHash] = receipt + mock.blockTimes[receipt.BlockHash] = testBlockTime + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, logs[0].TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(numEvents), numObs) + require.Equal(t, numEvents, len(msgC)) + + for i := 0; i < numEvents; i++ { + msg := recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + require.Equal(t, uint64(i), msg.Sequence) // #nosec G115 -- test-only + } +} + +func TestReobserveTwoValidEvents(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log1 := newValidWormholeLog(t, testBlockNumber, vaa.ConsistencyLevelFinalized) + mock.seedLog(log1) + + log2 := newTestLog(t, testLogParams{ + sender: testTokenBridge, + contractAddr: testEmitter, + sequence: 2, + consistencyLevel: vaa.ConsistencyLevelFinalized, + blockNumber: testBlockNumber, + }) + + receipt := newTestReceipt(log1.BlockNumber, []*types.Log{log1, &log2}) + mock.receipts[log1.TxHash] = receipt + mock.blockTimes[receipt.BlockHash] = testBlockTime + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log1.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(2), numObs) + require.Equal(t, 2, len(msgC)) + + ev, err := mock.ParseLogMessagePublished(*log1) + require.NoError(t, err) + msg := recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + assertMessageMatchesEvent(t, msg, ev) + + ev, err = mock.ParseLogMessagePublished(log2) + require.NoError(t, err) + msg = recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + assertMessageMatchesEvent(t, msg, ev) +} + +func TestReobserveValidAndInvalid(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + log1 := newValidWormholeLog(t, testBlockNumber, vaa.ConsistencyLevelFinalized) + mock.seedLog(log1) + + log2 := newTestLog(t, testLogParams{ + sender: testTokenBridge, + contractAddr: testEmitter, + sequence: 2, + consistencyLevel: vaa.ConsistencyLevelFinalized, + blockNumber: testBlockNumber, + }) + + log2.Topics[0] = eth_common.HexToHash("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef") + + receipt := newTestReceipt(log1.BlockNumber, []*types.Log{log1, &log2}) + mock.receipts[log1.TxHash] = receipt + mock.blockTimes[receipt.BlockHash] = testBlockTime + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log1.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(1), numObs) + require.Equal(t, 1, len(msgC)) + + ev, err := mock.ParseLogMessagePublished(*log1) + require.NoError(t, err) + msg := recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + assertMessageMatchesEvent(t, msg, ev) +} + +func TestReobserveTxVerifierIntegration(t *testing.T) { + otherSender := eth_common.HexToAddress("0x000000000000000000000000000000000000dEaD") + + tests := []struct { + name string + verifier txverifier.TransferVerifierInterface + sender eth_common.Address + expectState common.VerificationState + }{ + {"success", &MockTransferVerifier[ethclient.Client, connectors.Connector]{success: true}, testTokenBridge, common.Valid}, + {"failure", &MockTransferVerifier[ethclient.Client, connectors.Connector]{success: false}, testTokenBridge, common.Rejected}, + {"notapplicable", &MockTransferVerifier[ethclient.Client, connectors.Connector]{success: false}, otherSender, common.NotApplicable}, + {"notverified", nil, testTokenBridge, common.NotVerified}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + w.txVerifier = tc.verifier + + log := newTestLog(t, testLogParams{ + sender: tc.sender, + contractAddr: testEmitter, + sequence: 1, + consistencyLevel: vaa.ConsistencyLevelFinalized, + blockNumber: testBlockNumber, + }) + mock.seedLog(&log) + + numObs, err := w.handleReobservationRequest(context.TODO(), w.chainID, log.TxHash.Bytes(), mock, testFinalizedBlockNum, testSafeBlockNum) + require.NoError(t, err) + require.Equal(t, uint32(1), numObs) + require.Equal(t, 1, len(msgC)) + + ev, err := mock.ParseLogMessagePublished(log) + require.NoError(t, err) + msg := recvMsg(t, msgC) + require.True(t, msg.IsReobservation) + require.Equal(t, tc.expectState, msg.VerificationState()) + assertMessageMatchesEvent(t, msg, ev) + }) + } +} diff --git a/node/pkg/watchers/evm/watcher.go b/node/pkg/watchers/evm/watcher.go index 9783284de9..1c4ab4cdb5 100644 --- a/node/pkg/watchers/evm/watcher.go +++ b/node/pkg/watchers/evm/watcher.go @@ -19,6 +19,7 @@ import ( "github.com/certusone/wormhole/node/pkg/p2p" gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1" + "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/rpc" "github.com/prometheus/client_golang/prometheus/promauto" @@ -530,7 +531,6 @@ func (w *Watcher) Run(parentCtx context.Context) error { continue } - start := time.Now() currentHash := ev.Hash logger.Debug("processing new header", zap.Stringer("current_block", ev.Number), @@ -540,185 +540,11 @@ func (w *Watcher) Run(parentCtx context.Context) error { ) readiness.SetReady(w.readinessSync) - blockNumberU := ev.Number.Uint64() - var thisConsistencyLevel uint8 - switch ev.Finality { - case connectors.Latest: - thisConsistencyLevel = vaa.ConsistencyLevelPublishImmediately - atomic.StoreUint64(&w.latestBlockNumber, blockNumberU) - currentEthHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) - stats.Height = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely - w.ccqAddLatestBlock(ev) - case connectors.Safe: - thisConsistencyLevel = vaa.ConsistencyLevelSafe - atomic.StoreUint64(&w.latestSafeBlockNumber, blockNumberU) - currentEthSafeHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) - stats.SafeHeight = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely - case connectors.Finalized: - thisConsistencyLevel = vaa.ConsistencyLevelFinalized - atomic.StoreUint64(&w.latestFinalizedBlockNumber, blockNumberU) - currentEthFinalizedHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) - stats.FinalizedHeight = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely - default: - logger.Error("unexpected finality in block", zap.Stringer("finality", ev.Finality), zap.Any("event", ev)) - errC <- fmt.Errorf("unexpected finality in block: %v", ev.Finality) + if err := w.processNewBlock(ctx, ev, &stats); err != nil { + errC <- err p2p.DefaultRegistry.AddErrorCount(w.chainID, 1) - return nil - } - w.updateNetworkStats(&stats) - - w.pendingMu.Lock() - for key, pLock := range w.pending { - // Don't process the observation if it is waiting on a different consistency level. - if !consistencyLevelMatches(thisConsistencyLevel, pLock.effectiveCL) { - continue - } - - // Don't process the observation if we haven't reached the desired block height yet. - // CCL 'additionalBlocks' is used after consistency level handling. - // For instance, we need to wait for FINAL consistency level and then five more blocks. - if pLock.height+pLock.additionalBlocks > blockNumberU { - continue - } - - // Transaction is now ready - msm := time.Now() - timeout, cancel := context.WithTimeout(ctx, 5*time.Second) - tx, err := w.ethConn.TransactionReceipt(timeout, eth_common.BytesToHash(pLock.message.TxID)) - queryLatency.WithLabelValues(w.networkName, "transaction_receipt").Observe(time.Since(msm).Seconds()) - cancel() - - // SECURITY: Protection against reorgs that remove or change observations - // If the node returns an error after waiting expectedConfirmation blocks, - // it means the chain reorged and the transaction was orphaned. The - // TransactionReceipt call is using the same websocket connection than the - // head notifications, so it's guaranteed to be atomic. - // - // Check multiple possible error cases - the node seems to return a - // "not found" error most of the time, but it could conceivably also - // return a nil tx or rpc.ErrNoResult. - if tx == nil || errors.Is(err, rpc.ErrNoResult) || (err != nil && err.Error() == "not found") { - logger.Warn("tx was orphaned", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - zap.Error(err)) - delete(w.pending, key) - ethMessagesOrphaned.WithLabelValues(w.networkName, "not_found").Inc() - continue - } - - // SECURITY: Check that the tx succeeded. - // This should never happen - if we got this far, it means that logs were emitted, - // which is only possible if the transaction succeeded. We check it anyway just - // in case the EVM implementation is buggy. - if tx.Status != gethTypes.ReceiptStatusSuccessful { - logger.Error("transaction receipt with non-success status", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - zap.Error(err)) - delete(w.pending, key) - ethMessagesOrphaned.WithLabelValues(w.networkName, "tx_failed").Inc() - continue - } - - // Any error other than "not found" is likely transient - we retry next block. - if err != nil { - if pLock.height+MaxWaitConfirmations <= blockNumberU { - // An error from this "transient" case has persisted for more than MaxWaitConfirmations. - logger.Info("observation timed out", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - ) - ethMessagesOrphaned.WithLabelValues(w.networkName, "timeout").Inc() - delete(w.pending, key) - } else { - logger.Warn("transaction could not be fetched", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - zap.Error(err)) - } - continue - } - - // It's possible for a transaction to be orphaned and then included in a different block - // but with the same tx hash. Drop the observation (it will be re-observed and needs to - // wait for the full confirmation time again). - if tx.BlockHash != key.BlockHash { - logger.Info("tx got dropped and mined in a different block; the message should have been reobserved", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - ) - delete(w.pending, key) - ethMessagesOrphaned.WithLabelValues(w.networkName, "blockhash_mismatch").Inc() - continue - } - - logger.Info("observation confirmed", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", pLock.message.TxIDString()), - zap.Stringer("blockHash", key.BlockHash), - zap.Uint64("observedHeight", pLock.height), - zap.Uint64("additionalBlocks", pLock.additionalBlocks), - zap.Stringer("current_blockNum", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockHash", currentHash), - ) - delete(w.pending, key) - - // Note that `tx` here is actually a receipt - txHash := eth_common.Hash(pLock.message.TxID) - - // SECURITY: The TxHash existing within the particular block hash guarantees - // that the event exists without being modified. We don't - // check for the event mutating into something that's invalid (wrong event type, etc.) - // because of this. - pubErr := w.verifyAndPublish(pLock.message, ctx, txHash, tx) - if pubErr != nil { - logger.Error("could not publish message", - zap.String("msgId", pLock.message.MessageIDString()), - zap.String("txHash", txHash.String()), - zap.Error(pubErr), - ) - } + return nil //nolint:nilerr // error propagated via errC } - - w.pendingMu.Unlock() - logger.Debug("processed new header", - zap.Stringer("current_block", ev.Number), - zap.Stringer("finality", ev.Finality), - zap.Stringer("current_blockhash", currentHash), - zap.Duration("took", time.Since(start)), - ) } } }) @@ -981,7 +807,7 @@ func (w *Watcher) postMessage( receipt, err := w.ethConn.TransactionReceipt(receiptCtx, ev.Raw.TxHash) receiptCancel() queryLatency.WithLabelValues(w.networkName, "transaction_receipt").Observe(time.Since(msm).Seconds()) - if err != nil { + if receipt == nil || err != nil { w.logger.Error("failed to get transaction receipt for instant message", zap.String("msgId", msg.MessageIDString()), zap.String("txHash", msg.TxIDString()), @@ -1051,6 +877,220 @@ func (w *Watcher) postMessage( w.pendingMu.Unlock() } +// processNewBlock handles a new incoming block from the subscriber. It loops through all w.pending messages for processing. +func (w *Watcher) processNewBlock(ctx context.Context, ev *connectors.NewBlock, stats *gossipv1.Heartbeat_Network) error { + start := time.Now() + logger := w.logger + currentHash := ev.Hash + + blockNumberU := ev.Number.Uint64() + var thisConsistencyLevel uint8 + switch ev.Finality { + case connectors.Latest: + thisConsistencyLevel = vaa.ConsistencyLevelPublishImmediately + atomic.StoreUint64(&w.latestBlockNumber, blockNumberU) + currentEthHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) + stats.Height = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely + w.ccqAddLatestBlock(ev) + case connectors.Safe: + thisConsistencyLevel = vaa.ConsistencyLevelSafe + atomic.StoreUint64(&w.latestSafeBlockNumber, blockNumberU) + currentEthSafeHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) + stats.SafeHeight = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely + case connectors.Finalized: + thisConsistencyLevel = vaa.ConsistencyLevelFinalized + atomic.StoreUint64(&w.latestFinalizedBlockNumber, blockNumberU) + currentEthFinalizedHeight.WithLabelValues(w.networkName).Set(float64(blockNumberU)) + stats.FinalizedHeight = int64(blockNumberU) // #nosec G115 -- This conversion is safe indefinitely + default: + logger.Error("unexpected finality in block", zap.Stringer("finality", ev.Finality), zap.Any("event", ev)) + return fmt.Errorf("unexpected finality in block: %v", ev.Finality) + } + w.updateNetworkStats(stats) + + w.pendingMu.Lock() + defer w.pendingMu.Unlock() + for key, pLock := range w.pending { + // Don't process the observation if it is waiting on a different consistency level. + if !consistencyLevelMatches(thisConsistencyLevel, pLock.effectiveCL) { + continue + } + + // Don't process the observation if we haven't reached the desired block height yet. + if pLock.height+pLock.additionalBlocks > blockNumberU { + continue + } + + // Transaction is now ready + msm := time.Now() + timeout, cancel := context.WithTimeout(ctx, 5*time.Second) + txreceipt, err := w.ethConn.TransactionReceipt(timeout, eth_common.BytesToHash(pLock.message.TxID)) + queryLatency.WithLabelValues(w.networkName, "transaction_receipt").Observe(time.Since(msm).Seconds()) + cancel() + + // SECURITY: Protection against reorgs that remove or change observations + // If the node returns an error after waiting expectedConfirmation blocks, + // it means the chain reorged and the transaction was orphaned. The + // TransactionReceipt call is using the same websocket connection than the + // head notifications, so it's guaranteed to be atomic. + // + // Check multiple possible error cases - the node seems to return a + // "not found" error most of the time, but it could conceivably also + // return a rpc.ErrNoResult. + if errors.Is(err, rpc.ErrNoResult) || errors.Is(err, ethereum.NotFound) || (err != nil && err.Error() == "not found") { + logger.Warn("tx was orphaned", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + zap.Error(err)) + delete(w.pending, key) + ethMessagesOrphaned.WithLabelValues(w.networkName, "not_found").Inc() + continue + } + + // Transient errors (rate limit, transport, etc.) retry next block. + if err != nil { + if pLock.height+MaxWaitConfirmations <= blockNumberU { + // An error from this "transient" case has persisted for more than MaxWaitConfirmations. + logger.Info("observation timed out", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + ) + ethMessagesOrphaned.WithLabelValues(w.networkName, "timeout").Inc() + delete(w.pending, key) + } else { + logger.Warn("transaction could not be fetched", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + zap.Error(err)) + } + continue + } + + // txreceipt is nil and err is nil. A check to prevent panics later + if txreceipt == nil { + logger.Error("connector returned nil receipt with no error", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + ) + delete(w.pending, key) + ethMessagesOrphaned.WithLabelValues(w.networkName, "nil_receipt").Inc() + continue + } + + // SECURITY: Defense in depth against a buggy or malicious connector returning + // a receipt for a different transaction than the one we queried. + expectedTxHash := eth_common.BytesToHash(pLock.message.TxID) + if txreceipt.TxHash != expectedTxHash { + logger.Error("transaction receipt hash does not match queried txreceipt", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("expectedTxHash", expectedTxHash.Hex()), + zap.Stringer("receiptTxHash", txreceipt.TxHash), + zap.Stringer("blockHash", key.BlockHash), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + ) + delete(w.pending, key) + ethMessagesOrphaned.WithLabelValues(w.networkName, "tx_hash_mismatch").Inc() + continue + } + + // SECURITY: Check that the tx succeeded. + // This should never happen - if we got this far, it means that logs were emitted, + // which is only possible if the transaction succeeded. We check it anyway just + // in case the EVM implementation is buggy. + if txreceipt.Status != gethTypes.ReceiptStatusSuccessful { + logger.Error("transaction receipt with non-success status", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + zap.Any("status", txreceipt.Status)) + delete(w.pending, key) + ethMessagesOrphaned.WithLabelValues(w.networkName, "tx_failed").Inc() + continue + } + + // It's possible for a transaction to be orphaned and then included in a different block + // but with the same tx hash. Drop the observation (it will be re-observed and needs to + // wait for the full confirmation time again). + if txreceipt.BlockHash != key.BlockHash { + logger.Info("tx got dropped and mined in a different block; the message should have been reobserved", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + ) + delete(w.pending, key) + ethMessagesOrphaned.WithLabelValues(w.networkName, "blockhash_mismatch").Inc() + continue + } + + logger.Info("observation confirmed", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", pLock.message.TxIDString()), + zap.Stringer("blockHash", key.BlockHash), + zap.Uint64("observedHeight", pLock.height), + zap.Uint64("additionalBlocks", pLock.additionalBlocks), + zap.Stringer("current_blockNum", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockHash", currentHash), + ) + delete(w.pending, key) + + // SECURITY: The TxHash existing within the particular block hash guarantees + // that the event exists without being modified. We don't + // check for the event mutating into something that's invalid (wrong event type, etc.) + // because of this. + txHash := eth_common.Hash(pLock.message.TxID) + pubErr := w.verifyAndPublish(pLock.message, ctx, txHash, txreceipt) + if pubErr != nil { + logger.Error("could not publish message", + zap.String("msgId", pLock.message.MessageIDString()), + zap.String("txHash", txHash.String()), + zap.Error(pubErr), + ) + } + } + + logger.Debug("processed new header", + zap.Stringer("current_block", ev.Number), + zap.Stringer("finality", ev.Finality), + zap.Stringer("current_blockhash", currentHash), + zap.Duration("took", time.Since(start)), + ) + + return nil +} + // blockNotFoundErrors is used by `canRetryGetBlockTime`. It is a map of the error returns from `getBlockTime` that can trigger a retry. var blockNotFoundErrors = map[string]struct{}{ "not found": {}, diff --git a/node/pkg/watchers/evm/watcher_test.go b/node/pkg/watchers/evm/watcher_test.go index aea2c7131a..ba2cb5a6cf 100644 --- a/node/pkg/watchers/evm/watcher_test.go +++ b/node/pkg/watchers/evm/watcher_test.go @@ -4,10 +4,11 @@ import ( "context" "encoding/json" "errors" + "math/big" "testing" "github.com/certusone/wormhole/node/pkg/common" - "github.com/certusone/wormhole/node/pkg/txverifier" + gossipv1 "github.com/certusone/wormhole/node/pkg/proto/gossip/v1" "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors" "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/ethabi" ethereum "github.com/ethereum/go-ethereum" @@ -17,7 +18,6 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/wormhole-foundation/wormhole/sdk/vaa" - "go.uber.org/zap" ) func TestMsgIdFromLogEvent(t *testing.T) { @@ -88,13 +88,12 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.NoError(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg := <-msgC + publishedMsg := recvMsg(t, msgC) require.NotNil(t, publishedMsg) require.Equal(t, 0, len(msgC)) require.Equal(t, common.NotVerified.String(), publishedMsg.VerificationState().String()) - tbAddr, byteErr := vaa.BytesToAddress([]byte{0x01}) - require.NoError(t, byteErr) + tbAddr := PadAddress(testTokenBridge) // Check scenario where transfer verifier is enabled on the watcher level but // there is no Transfer Verifier instantiated. In this case, fail open and continue @@ -108,7 +107,7 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.NoError(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg = <-msgC + publishedMsg = recvMsg(t, msgC) require.Equal(t, common.NotVerified.String(), publishedMsg.VerificationState().String()) // Check that message status is not changed if it didn't come from token bridge. @@ -119,7 +118,7 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.Nil(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg = <-msgC + publishedMsg = recvMsg(t, msgC) require.Equal(t, common.NotVerified.String(), publishedMsg.VerificationState().String()) // Check scenario where the message already has a verification status. @@ -147,7 +146,7 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.Nil(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg = <-msgC + publishedMsg = recvMsg(t, msgC) require.NotNil(t, publishedMsg) require.Equal(t, 0, len(msgC)) require.Equal(t, common.Rejected.String(), publishedMsg.VerificationState().String()) @@ -160,7 +159,7 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.Nil(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg = <-msgC + publishedMsg = recvMsg(t, msgC) require.Equal(t, common.NotApplicable.String(), publishedMsg.VerificationState().String()) // Check happy path where txverifier is enabled, initialized, and the message is from the token bridge. @@ -174,48 +173,16 @@ func TestVerifyAndPublish(t *testing.T) { err = w.verifyAndPublish(&msg, ctx, eth_common.Hash{}, &types.Receipt{}) require.NoError(t, err) require.Equal(t, 1, len(msgC)) - publishedMsg = <-msgC + publishedMsg = recvMsg(t, msgC) require.NotNil(t, publishedMsg) require.Equal(t, 0, len(msgC)) require.Equal(t, common.Valid.String(), publishedMsg.VerificationState().String()) } -// Helper function to set up a test Ethereum Watcher -func NewWatcherForTest(t *testing.T, msgC chan<- *common.MessagePublication) *Watcher { - t.Helper() - logger := zap.NewNop() - - w := &Watcher{ - // this is implicit but added here for clarity - txVerifierEnabled: false, - msgC: msgC, - logger: logger, - } - - return w -} - -type MockTransferVerifier[E ethclient.Client, C connectors.Connector] struct { - success bool -} - -// TransferIsValid simulates the evaluation made by the Transfer Verifier. -// Always returns nil. The error should be non-nil only when a parsing or RPC error occurs. -// For now, these are not included in the unit tests. -func (m *MockTransferVerifier[E, C]) TransferIsValid(_ context.Context, _ string, _ eth_common.Hash, _ *types.Receipt) (bool, error) { - return m.success, nil -} -func (m *MockTransferVerifier[E, C]) Addrs() *txverifier.TVAddresses { - return &txverifier.TVAddresses{ - TokenBridgeAddr: eth_common.BytesToAddress([]byte{0x01}), - } -} - // TestVerifyDoesNotMutateOriginalMessage checks that verify() does not modify // the original MessagePublication passed to it. func TestVerifyDoesNotMutateOriginalMessage(t *testing.T) { - tbAddr, err := vaa.BytesToAddress([]byte{0x01}) - require.NoError(t, err) + tbAddr := PadAddress(testTokenBridge) msg := &common.MessagePublication{ EmitterAddress: tbAddr, @@ -235,6 +202,768 @@ func TestVerifyDoesNotMutateOriginalMessage(t *testing.T) { require.Equal(t, common.NotVerified.String(), msg.VerificationState().String()) } +// Several test cases for a pending message getting processed depending on the block number +func TestProcessBlockPendingByFinality(t *testing.T) { + tests := []struct { + name string + cl uint8 + finality connectors.FinalityLevel + blockNumber uint64 + expectPending int + expectPublish bool + }{ + {"finalized", vaa.ConsistencyLevelFinalized, connectors.Finalized, 105, 0, true}, + {"safe", vaa.ConsistencyLevelSafe, connectors.Safe, 105, 0, true}, + {"instant", vaa.ConsistencyLevelPublishImmediately, connectors.Latest, 105, 0, true}, + {"finalized_before_block", vaa.ConsistencyLevelFinalized, connectors.Finalized, 99, 1, false}, + {"finalized_with_safe_block", vaa.ConsistencyLevelFinalized, connectors.Safe, 105, 1, false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + w.addPendingMsg(txHash, blockHash, tc.cl, 0, 1) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + + err := w.processNewBlock(context.TODO(), newBlock(tc.blockNumber, tc.finality), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + assert.Equal(t, tc.expectPending, len(w.pending)) + if tc.expectPublish { + require.Equal(t, 1, len(msgC)) + assert.Equal(t, txHash.Bytes(), recvMsg(t, msgC).TxID) + } else { + assert.Equal(t, 0, len(msgC)) + } + }) + } +} + +// Handling both a finalized and a safe message +func TestProcessBlockPendingFinalizedAndSafe(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash1 := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + txHash2 := eth_common.HexToHash("0xe2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ee") + blockHash1 := eth_common.BigToHash(big.NewInt(111)) + blockHash2 := eth_common.BigToHash(big.NewInt(222)) + + w.addPendingMsg(txHash1, blockHash1, vaa.ConsistencyLevelFinalized, 0, 1) + w.addPendingMsg(txHash2, blockHash2, vaa.ConsistencyLevelSafe, 0, 1) + mock.receipts[txHash1] = &types.Receipt{Status: 1, BlockHash: blockHash1, TxHash: txHash1} + mock.receipts[txHash2] = &types.Receipt{Status: 1, BlockHash: blockHash2, TxHash: txHash2} + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed one from pending + assert.Equal(t, 1, len(w.pending)) + + // Published finalized message + require.Equal(t, 1, len(msgC)) + assert.Equal(t, txHash1.Bytes(), recvMsg(t, msgC).TxID) + + err = w.processNewBlock(context.TODO(), newBlock(105, connectors.Safe), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed both + assert.Equal(t, 0, len(w.pending)) + + // Published safe message + require.Equal(t, 1, len(msgC)) + assert.Equal(t, txHash2.Bytes(), recvMsg(t, msgC).TxID) +} + +// Removal of the message without publication if the blockhash differs +func TestProcessBlockPendingWrongBlockHash(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHashBlock := eth_common.BigToHash(big.NewInt(111)) + blockHashMessage := eth_common.BigToHash(big.NewInt(222)) + + w.addPendingMsg(txHash, blockHashMessage, vaa.ConsistencyLevelFinalized, 0, 1) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHashBlock, TxHash: txHash} + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) +} + +// Failed transaction status gets rejected +func TestProcessBlockPendingFailedTx(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + mock.receipts[txHash] = &types.Receipt{Status: 0, BlockHash: blockHash, TxHash: txHash} + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) +} + +// Failed receipt test case +func TestProcessBlockValidReceiptWithError(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + mock.errors[txHash] = errors.New("not found") + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) +} + +// No receipt is found. This should remove from the pending list. +func TestProcessBlockInValidReceiptNoError(t *testing.T) { + w, _, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) +} + +// Transient errors on receipt RPC requests should be retried +func TestProcessBlockTransientError(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + mock.errors[txHash] = errors.New("rate limit 429 error") + + w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Not removed from pending + assert.Equal(t, 1, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) + + // Try the message again after the transient error has disappeared + mock.errors[txHash] = nil + err = w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Published with correct TxID + require.Equal(t, 1, len(msgC)) + assert.Equal(t, txHash.Bytes(), recvMsg(t, msgC).TxID) +} + +// A transient error that persists past MaxWaitConfirmations should remove the pending entry. +func TestProcessBlockTransientErrorTimeout(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + mock.errors[txHash] = errors.New("transient error") + + w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + + // blockNumberU = pLock.height + MaxWaitConfirmations exactly hits the timeout branch. + err := w.processNewBlock(context.TODO(), newBlock(100+MaxWaitConfirmations, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending after timeout + assert.Equal(t, 0, len(w.pending)) + + // Not published + assert.Equal(t, 0, len(msgC)) +} + +// AdditionalBlocks test cases for waiting the proper amount of time before publication +func TestProcessBlockAdditionalBlocks(t *testing.T) { + tests := []struct { + name string + cl uint8 + finality connectors.FinalityLevel + additionalBlocks uint64 + blockNumber uint64 + expectPending int + expectPublish bool + }{ + {"before", vaa.ConsistencyLevelFinalized, connectors.Finalized, 20, 100, 1, false}, + {"before_by_one", vaa.ConsistencyLevelFinalized, connectors.Finalized, 20, 119, 1, false}, + {"exact", vaa.ConsistencyLevelFinalized, connectors.Finalized, 20, 120, 0, true}, + {"none", vaa.ConsistencyLevelPublishImmediately, connectors.Latest, 0, 100, 0, true}, + {"maximum", vaa.ConsistencyLevelPublishImmediately, connectors.Latest, 65535, 100 + 65535, 0, true}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + w.addPendingMsg(txHash, blockHash, tc.cl, tc.additionalBlocks, 1) + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + + err := w.processNewBlock(context.TODO(), newBlock(tc.blockNumber, tc.finality), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + assert.Equal(t, tc.expectPending, len(w.pending)) + if tc.expectPublish { + require.Equal(t, 1, len(msgC)) + assert.Equal(t, txHash.Bytes(), recvMsg(t, msgC).TxID) + } else { + assert.Equal(t, 0, len(msgC)) + } + }) + } +} + +// Effective consistency level (CL) and the VAA CL should differ. +func TestProcessBlockCCLEffectiveCLDiffersFromMessageCL(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + // Simulate CCL: message.ConsistencyLevel = Custom, but effectiveCL = Finalized + key := w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 5, 1) + w.pending[key].message.ConsistencyLevel = vaa.ConsistencyLevelCustom + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + + // Finalized block at height+additionalBlocks: should confirm based on effectiveCL + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Removed from pending and published + assert.Equal(t, 0, len(w.pending)) + require.Equal(t, 1, len(msgC)) + + // The published message must still have ConsistencyLevelCustom for VAA hash consistency + published := recvMsg(t, msgC) + assert.Equal(t, vaa.ConsistencyLevelCustom, published.ConsistencyLevel) +} + +// Pending messages with different finalizations and block times +func TestProcessBlockCCLMultiplePendingDifferentAdditionalBlocks(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + txHashA := eth_common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + txHashB := eth_common.HexToHash("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + // Message A: effectiveCL=Finalized, additionalBlocks=0, height=100 -> ready at block 100 + w.addPendingMsg(txHashA, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + mock.receipts[txHashA] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHashA} + + // Message B: effectiveCL=Finalized, additionalBlocks=10, height=100 -> ready at block 110 + w.addPendingMsg(txHashB, blockHash, vaa.ConsistencyLevelFinalized, 10, 2) + mock.receipts[txHashB] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHashB} + + // Send finalized block at 105: A should confirm, B should stay pending + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + assert.Equal(t, 1, len(w.pending), "only message B should remain pending") + require.Equal(t, 1, len(msgC), "only message A should be published") + assert.Equal(t, txHashA.Bytes(), recvMsg(t, msgC).TxID) + + // Send finalized block at 110: B should now confirm + err = w.processNewBlock(context.TODO(), newBlock(110, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + assert.Equal(t, 0, len(w.pending), "no messages should remain pending") + require.Equal(t, 1, len(msgC), "message B should now be published") + assert.Equal(t, txHashB.Bytes(), recvMsg(t, msgC).TxID) +} + +// Orphaned tx handling +func TestProcessBlockOneConfirmedOneOrphaned(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + blockHash := eth_common.BigToHash(big.NewInt(100)) + + txHashGood := eth_common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa") + txHashOrphaned := eth_common.HexToHash("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb") + + w.addPendingMsg(txHashGood, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + w.addPendingMsg(txHashOrphaned, blockHash, vaa.ConsistencyLevelFinalized, 0, 2) + + // Good tx has a valid receipt + mock.receipts[txHashGood] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHashGood} + + // Orphaned tx returns nil receipt (not in mock.receipts, so defaults to nil) + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + // Both removed from pending + assert.Equal(t, 0, len(w.pending)) + + // Only the valid message was published + assert.Equal(t, 1, len(msgC)) + + published := recvMsg(t, msgC) + assert.Equal(t, txHashGood.Bytes(), published.TxID) +} + +// Invalid finality level process. Should return an error. +func TestProcessBlockUnexpectedFinality(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + err := w.processNewBlock(context.TODO(), newBlock(100, connectors.FinalityLevel(99)), &gossipv1.Heartbeat_Network{}) + require.Error(t, err) + assert.Contains(t, err.Error(), "unexpected finality in block") + + // Nothing published + assert.Equal(t, 0, len(msgC)) +} + +// TxVerifier support in processBlock +func TestProcessBlockTxVerifier(t *testing.T) { + tests := []struct { + name string + success bool + useTokenBridge bool + expectState common.VerificationState + }{ + {"success", true, true, common.Valid}, + {"failure", false, true, common.Rejected}, + {"not_applicable", false, false, common.NotApplicable}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + w.txVerifier = &MockTransferVerifier[ethclient.Client, connectors.Connector]{success: tc.success} + + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.BigToHash(big.NewInt(100)) + + key := w.addPendingMsg(txHash, blockHash, vaa.ConsistencyLevelFinalized, 0, 1) + if tc.useTokenBridge { + w.pending[key].message.EmitterAddress = PadAddress(testTokenBridge) + } + mock.receipts[txHash] = &types.Receipt{Status: 1, BlockHash: blockHash, TxHash: txHash} + + err := w.processNewBlock(context.TODO(), newBlock(105, connectors.Finalized), &gossipv1.Heartbeat_Network{}) + require.NoError(t, err) + + assert.Equal(t, 0, len(w.pending)) + require.Equal(t, 1, len(msgC)) + + msg := recvMsg(t, msgC) + assert.Equal(t, txHash.Bytes(), msg.TxID) + assert.Equal(t, tc.expectState, msg.VerificationState()) + }) + } +} + +// Safe and finalized add to pending in postMessage +func TestPostMessageAddsToPending(t *testing.T) { + tests := []struct { + name string + cl uint8 + }{ + {"finalized", vaa.ConsistencyLevelFinalized}, + {"safe", vaa.ConsistencyLevelSafe}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + ev := newTestLogEvent(tc.cl) + w.postMessage(context.TODO(), ev, 1234) + + require.Equal(t, 1, len(w.pending)) + assert.Equal(t, 0, len(msgC)) + + key := pendingKey{ + TxHash: ev.Raw.TxHash, + BlockHash: ev.Raw.BlockHash, + EmitterAddress: PadAddress(ev.Sender), + Sequence: ev.Sequence, + } + pe := w.pending[key] + require.NotNil(t, pe) + + assertMessageMatchesEvent(t, pe.message, ev) + assertPendingMetadata(t, pe, tc.cl, 0) + }) + } +} + +// Custom consistency level (CCL) edge cases lead to finalized by default +func TestPostMessageCustomDefaultToFinalized(t *testing.T) { + tests := []struct { + name string + setupCCL func(w *Watcher) + }{ + {"ccl_disabled", nil}, + {"ccl_enabled_nothing_special", func(w *Watcher) { + w.enableCCL() + w.seedCCLNothingSpecial(testEmitter) + }}, + {"invalid_additional_blocks_config", func(w *Watcher) { + w.enableCCL() + w.seedCCLAdditionalBlocks(testEmitter, 1, 1) + }}, + {"invalid_type", func(w *Watcher) { + w.enableCCL() + w.seedCCLInvalidType(testEmitter) + }}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + if tc.setupCCL != nil { + tc.setupCCL(w) + } + + ev := newTestLogEvent(vaa.ConsistencyLevelCustom) + w.postMessage(context.TODO(), ev, 1234) + + require.Equal(t, 1, len(w.pending)) + assert.Equal(t, 0, len(msgC)) + + key := pendingKey{ + TxHash: ev.Raw.TxHash, + BlockHash: ev.Raw.BlockHash, + EmitterAddress: PadAddress(ev.Sender), + Sequence: ev.Sequence, + } + pe := w.pending[key] + require.NotNil(t, pe) + assertMessageMatchesEvent(t, pe.message, ev) + assertPendingMetadata(t, pe, vaa.ConsistencyLevelFinalized, 0) + }) + } +} + +// AdditionalBlocks basic testing +func TestPostMessageCustomAdditionalBlocks(t *testing.T) { + tests := []struct { + name string + effectiveCL uint8 + additionalBlocks uint16 + }{ + {"finalized", vaa.ConsistencyLevelFinalized, 101}, + {"safe", vaa.ConsistencyLevelSafe, 50}, + {"instant", vaa.ConsistencyLevelPublishImmediately, 10}, + {"zero_blocks", vaa.ConsistencyLevelFinalized, 0}, + {"one_block", vaa.ConsistencyLevelFinalized, 1}, + {"small_blocks", vaa.ConsistencyLevelFinalized, 5}, + {"medium_blocks", vaa.ConsistencyLevelFinalized, 500}, + {"large_blocks", vaa.ConsistencyLevelFinalized, 10000}, + {"max_uint16", vaa.ConsistencyLevelFinalized, 0xFFFF}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + w.enableCCL() + w.seedCCLAdditionalBlocks(testEmitter, tc.effectiveCL, tc.additionalBlocks) + ev := newTestLogEvent(vaa.ConsistencyLevelCustom) + w.postMessage(context.TODO(), ev, 1234) + + require.Equal(t, 1, len(w.pending)) + assert.Equal(t, 0, len(msgC)) + + key := pendingKey{ + TxHash: ev.Raw.TxHash, + BlockHash: ev.Raw.BlockHash, + EmitterAddress: PadAddress(ev.Sender), + Sequence: ev.Sequence, + } + pe := w.pending[key] + require.NotNil(t, pe) + assertMessageMatchesEvent(t, pe.message, ev) + assertPendingMetadata(t, pe, tc.effectiveCL, uint64(tc.additionalBlocks)) + }) + } +} + +// Instant message is published instead of being added to the pending queue +func TestPostMessageInstantPublishes(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + mock.receipts[ev.Raw.TxHash] = newTestReceipt(ev.Raw.BlockNumber, nil) + w.postMessage(context.TODO(), ev, 1234) + + // Should be published instead of being added to the pending queue + require.Equal(t, 0, len(w.pending)) + assert.Equal(t, 1, len(msgC)) + + msg := recvMsg(t, msgC) + + assertMessageMatchesEvent(t, msg, ev) + assert.Equal(t, common.NotVerified, msg.VerificationState()) +} + +// Multiple instant publishes are handled properly +func TestPostMessageTwoInstantPublishes(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + ev1 := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + + ev2 := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + ev2.Sender = eth_common.HexToAddress("0x388C818CA8B9251b393131C08a736A67ccB19297") + ev2.Nonce = 20 + ev2.Sequence = 2 + + mock.receipts[ev1.Raw.TxHash] = newTestReceipt(ev1.Raw.BlockNumber, nil) + mock.receipts[ev2.Raw.TxHash] = newTestReceipt(ev2.Raw.BlockNumber, nil) + w.postMessage(context.TODO(), ev1, 1234) + w.postMessage(context.TODO(), ev2, 1234) + + // Should be added to pending, not published immediately + require.Equal(t, 0, len(w.pending)) + assert.Equal(t, 2, len(msgC)) + + msg1 := recvMsg(t, msgC) + msg2 := recvMsg(t, msgC) + + assertMessageMatchesEvent(t, msg1, ev1) + assert.Equal(t, common.NotVerified, msg1.VerificationState()) + + assertMessageMatchesEvent(t, msg2, ev2) + assert.Equal(t, common.NotVerified, msg2.VerificationState()) +} + +// An instant and a final message go to the proper spots (msgC and pending) +func TestPostMessageInstantAndFinalized(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + // ev1: instant publish + ev1 := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + + // ev2: finalized (goes to pending) — use a different emitter and sequence + ev2 := newTestLogEventFromParams(testLogEventParams{ + sender: eth_common.HexToAddress("0x388C818CA8B9251b393131C08a736A67ccB19297"), + sequence: 2, + blockNumber: 100, + consistencyLevel: vaa.ConsistencyLevelFinalized, + }) + + mock.receipts[ev1.Raw.TxHash] = newTestReceipt(ev1.Raw.BlockNumber, nil) + w.postMessage(context.TODO(), ev1, 1234) + w.postMessage(context.TODO(), ev2, 1234) + + // One message published immediately, one added to pending. + require.Equal(t, 1, len(w.pending)) + assert.Equal(t, 1, len(msgC)) + + // Verify the instant-published message + msg := recvMsg(t, msgC) + assertMessageMatchesEvent(t, msg, ev1) + assert.Equal(t, common.NotVerified, msg.VerificationState()) + + // Verify the finalized pending entry + key := pendingKey{ + TxHash: ev2.Raw.TxHash, + BlockHash: ev2.Raw.BlockHash, + EmitterAddress: PadAddress(ev2.Sender), + Sequence: ev2.Sequence, + } + pe := w.pending[key] + require.NotNil(t, pe) + + assertMessageMatchesEvent(t, pe.message, ev2) + assertPendingMetadata(t, pe, vaa.ConsistencyLevelFinalized, 0) +} + +// Transaction contains multiple events +func TestPostMessageMultipleEventsFromSameTransaction(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + txHash := eth_common.HexToHash("0xd2d35ab0d18dd19e81a58dfe8d97ad8c68659bd81d7017bcdf4d9719b32119ef") + blockHash := eth_common.HexToHash("0xa1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1") + + ev1 := newTestLogEventFromParams(testLogEventParams{ + sender: testEmitter, + sequence: 1, + blockNumber: 100, + consistencyLevel: vaa.ConsistencyLevelFinalized, + txHash: txHash, + blockHash: blockHash, + }) + ev2 := newTestLogEventFromParams(testLogEventParams{ + sender: testEmitter, + sequence: 2, + blockNumber: 100, + consistencyLevel: vaa.ConsistencyLevelFinalized, + txHash: txHash, + blockHash: blockHash, + }) + + w.postMessage(context.TODO(), ev1, 1234) + w.postMessage(context.TODO(), ev2, 1234) + + assert.Equal(t, 0, len(msgC)) + require.Equal(t, 2, len(w.pending)) + + key1 := pendingKey{ + TxHash: txHash, + BlockHash: blockHash, + EmitterAddress: PadAddress(testEmitter), + Sequence: 1, + } + key2 := pendingKey{ + TxHash: txHash, + BlockHash: blockHash, + EmitterAddress: PadAddress(testEmitter), + Sequence: 2, + } + + pe1 := w.pending[key1] + pe2 := w.pending[key2] + require.NotNil(t, pe1) + require.NotNil(t, pe2) + + assert.Equal(t, txHash.Bytes(), pe1.message.TxID) + assert.Equal(t, uint64(1), pe1.message.Sequence) + assert.Equal(t, txHash.Bytes(), pe2.message.TxID) + assert.Equal(t, uint64(2), pe2.message.Sequence) +} + +func TestPostMessageRemovedLogIsIgnored(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + ev.Raw.Removed = true + + w.postMessage(context.TODO(), ev, 1234) + + assert.Equal(t, 0, len(msgC), "removed log should not be published to msgC") + assert.Equal(t, 0, len(w.pending), "removed log should not be added to pending") +} + +func TestPostMessageWrongContractAddressIsIgnored(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + ev.Raw.Address = eth_common.HexToAddress("0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef") + + w.postMessage(context.TODO(), ev, 1234) + + assert.Equal(t, 0, len(msgC), "log from wrong contract should not be published to msgC") + assert.Equal(t, 0, len(w.pending), "log from wrong contract should not be added to pending") +} + +func TestPostMessageWrongEventSignatureIsIgnored(t *testing.T) { + w, _, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + ev.Raw.Topics = []eth_common.Hash{ + eth_common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"), + eth_common.BytesToHash(testEmitter.Bytes()), + } + + w.postMessage(context.TODO(), ev, 1234) + + assert.Equal(t, 0, len(msgC), "log with wrong event signature should not be published to msgC") + assert.Equal(t, 0, len(w.pending), "log with wrong event signature should not be added to pending") +} + +// TxVerifier is used on postMessage +func TestPostMessageTxVerifier(t *testing.T) { + tests := []struct { + name string + success bool + useTokenBridge bool + expectState common.VerificationState + }{ + {"success", true, true, common.Valid}, + {"failure", false, true, common.Rejected}, + {"not_applicable", false, false, common.NotApplicable}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + w.txVerifier = &MockTransferVerifier[ethclient.Client, connectors.Connector]{success: tc.success} + + sender := testEmitter + if tc.useTokenBridge { + sender = testTokenBridge + } + ev := newTestLogEventFromParams(testLogEventParams{ + sender: sender, + sequence: 1, + blockNumber: 100, + consistencyLevel: vaa.ConsistencyLevelPublishImmediately, + }) + mock.receipts[ev.Raw.TxHash] = newTestReceipt(ev.Raw.BlockNumber, nil) + w.postMessage(context.TODO(), ev, 1234) + + require.Equal(t, 0, len(w.pending)) + require.Equal(t, 1, len(msgC)) + + msg := recvMsg(t, msgC) + assertMessageMatchesEvent(t, msg, ev) + assert.Equal(t, tc.expectState, msg.VerificationState()) + }) + } +} + +// Instant publish must drop the message when the receipt cannot be fetched. +func TestPostMessageInstantReceiptError(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + mock.errors[ev.Raw.TxHash] = errors.New("rpc failure") + + w.postMessage(context.TODO(), ev, 1234) + + assert.Equal(t, 0, len(w.pending)) + assert.Equal(t, 0, len(msgC)) +} + +// Instant publish must drop the message when the receipt reports a non-success status. +func TestPostMessageInstantTxFailed(t *testing.T) { + w, mock, msgC := newTestWatcher(t) + + ev := newTestLogEvent(vaa.ConsistencyLevelPublishImmediately) + receipt := newTestReceipt(ev.Raw.BlockNumber, nil) + receipt.Status = 0 + mock.receipts[ev.Raw.TxHash] = receipt + + w.postMessage(context.TODO(), ev, 1234) + + assert.Equal(t, 0, len(w.pending)) + assert.Equal(t, 0, len(msgC)) +} + func TestConsistencyLevelMatches(t *testing.T) { // Success cases. assert.True(t, consistencyLevelMatches(vaa.ConsistencyLevelPublishImmediately, vaa.ConsistencyLevelPublishImmediately)) diff --git a/node/pkg/watchers/evm/watcher_test_helpers_test.go b/node/pkg/watchers/evm/watcher_test_helpers_test.go new file mode 100644 index 0000000000..ee0deb2a0c --- /dev/null +++ b/node/pkg/watchers/evm/watcher_test_helpers_test.go @@ -0,0 +1,455 @@ +package evm + +import ( + "context" + "encoding/binary" + "math/big" + "testing" + "time" + + "github.com/certusone/wormhole/node/pkg/common" + "github.com/certusone/wormhole/node/pkg/txverifier" + "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors" + dgAbi "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/delegated_guardians" + "github.com/certusone/wormhole/node/pkg/watchers/evm/connectors/ethabi" + ethereum "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/accounts/abi" + eth_common "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/rpc" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wormhole-foundation/wormhole/sdk/vaa" + "go.uber.org/zap" +) + +// testEmitter is a default non-zero emitter address for use across tests. +var testEmitter = eth_common.HexToAddress("0x0290FB167208Af455bB137780163b7B7a9a10C16") + +// testTokenBridge is the token bridge address used by MockTransferVerifier. +var testTokenBridge = eth_common.BytesToAddress([]byte{0x01}) + +// Default block heights used across reobserve tests: the log is observed at testBlockNumber, +// with testFinalizedBlockNum / testSafeBlockNum representing current chain heads. +var ( + testBlockNumber = uint64(100) + testFinalizedBlockNum = uint64(200) + testSafeBlockNum = uint64(150) + testBlockTime = uint64(1234) +) + +// NewWatcherForTest creates a minimal Watcher for verifyAndPublish tests. +func NewWatcherForTest(t *testing.T, msgC chan<- *common.MessagePublication) *Watcher { + t.Helper() + logger := zap.NewNop() + + w := &Watcher{ + // this is implicit but added here for clarity + txVerifierEnabled: false, + msgC: msgC, + logger: logger, + } + + return w +} + +// newTestWatcher creates a Watcher wired up with a mockConnector for processNewBlock and postMessage tests. +// Returns the watcher, the mock (for configuring receipts/errors), and the msgC channel (for reading published messages). +func newTestWatcher(t *testing.T) (*Watcher, *mockConnector, chan *common.MessagePublication) { + t.Helper() + mock := newMockConnector(t) + msgC := make(chan *common.MessagePublication, 10) + + w := &Watcher{ + logger: zap.NewNop(), + cclLogger: zap.NewNop(), + ethConn: mock, + msgC: msgC, + pending: make(map[pendingKey]*pendingMessage), + networkName: "test", + chainID: vaa.ChainIDEthereum, + contract: testEmitter, + } + + return w, mock, msgC +} + +// recvMsg reads from msgC, failing the test (rather than blocking forever) if no +// message arrives within a short timeout. Prefer this over a bare `<-msgC` so a +// missing publish surfaces as a clear failure instead of a `go test` hang. +func recvMsg(t *testing.T, msgC <-chan *common.MessagePublication) *common.MessagePublication { + t.Helper() + select { + case msg := <-msgC: + return msg + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for message on msgC") + return nil + } +} + +type MockTransferVerifier[E ethclient.Client, C connectors.Connector] struct { + success bool +} + +// TransferIsValid simulates the evaluation made by the Transfer Verifier. +// Always returns nil. The error should be non-nil only when a parsing or RPC error occurs. +// For now, these are not included in the unit tests. +func (m *MockTransferVerifier[E, C]) TransferIsValid(_ context.Context, _ string, _ eth_common.Hash, _ *types.Receipt) (bool, error) { + return m.success, nil +} +func (m *MockTransferVerifier[E, C]) Addrs() *txverifier.TVAddresses { + return &txverifier.TVAddresses{ + TokenBridgeAddr: testTokenBridge, + } +} + +// mockConnector is a minimal mock of the connectors.Connector interface for testing +// processNewBlock, postMessage, and reobservation flows. TransactionReceipt, TimeOfBlockByHash, +// and ParseLogMessagePublished have real behavior; all other methods panic because they should +// not be called in these tests. +type mockConnector struct { + // receipts maps txHash -> receipt to return + receipts map[eth_common.Hash]*types.Receipt + // errors maps txHash -> error to return from TransactionReceipt + errors map[eth_common.Hash]error + // blockTimes maps blockHash -> time to return from TimeOfBlockByHash + blockTimes map[eth_common.Hash]uint64 + // filterer delegates ParseLogMessagePublished to the real ABI parser. + filterer *ethabi.AbiFilterer +} + +func newMockConnector(t *testing.T) *mockConnector { + t.Helper() + // The zero address and nil filterer are fine here: UnpackLog only needs the parsed ABI + // metadata baked into the generated package, not an RPC-capable filterer. + filterer, err := ethabi.NewAbiFilterer(eth_common.Address{}, nil) + require.NoError(t, err) + return &mockConnector{ + receipts: make(map[eth_common.Hash]*types.Receipt), + errors: make(map[eth_common.Hash]error), + blockTimes: make(map[eth_common.Hash]uint64), + filterer: filterer, + } +} + +// seedLog wires the mock to return a successful receipt wrapping `log` (keyed by its TxHash) +// and the given block time (keyed by the receipt's BlockHash). For receipts with Status != 1 +// or custom shapes, write directly to mock.receipts / mock.blockTimes. +func (m *mockConnector) seedLog(log *types.Log) { + receipt := newTestReceipt(log.BlockNumber, []*types.Log{log}) + m.receipts[log.TxHash] = receipt + m.blockTimes[receipt.BlockHash] = testBlockTime +} + +// TransactionReceipt returns the configured receipt/error. When neither is set for the txHash, +// it returns a "not found" error to match the real ethclient behavior (and avoid nil-deref in +// callers that don't check err before dereferencing the receipt). +func (m *mockConnector) TransactionReceipt(_ context.Context, txHash eth_common.Hash) (*types.Receipt, error) { + err := m.errors[txHash] + r, hasReceipt := m.receipts[txHash] + if err == nil && !hasReceipt { + return nil, ethereum.NotFound + } + return r, err +} + +// Stub implementations that are required by the interface +func (m *mockConnector) NetworkName() string { + panic("not implemented") +} +func (m *mockConnector) ContractAddress() eth_common.Address { + panic("not implemented") +} +func (m *mockConnector) GetCurrentGuardianSetIndex(ctx context.Context) (uint32, error) { + panic("not implemented") +} +func (m *mockConnector) GetGuardianSet(ctx context.Context, index uint32) (ethabi.StructsGuardianSet, error) { + panic("not implemented") +} +func (m *mockConnector) GetDelegatedGuardianConfig(ctx context.Context) ([]dgAbi.WormholeDelegatedGuardiansDelegatedGuardianSet, error) { + panic("not implemented") +} +func (m *mockConnector) WatchLogMessagePublished(ctx context.Context, errC chan error, sink chan<- *ethabi.AbiLogMessagePublished) (event.Subscription, error) { + panic("not implemented") +} + +// TimeOfBlockByHash returns the configured block time for the given hash, or 0 if unset. +// Tests that care about the exact timestamp should seed blockTimes explicitly. +func (m *mockConnector) TimeOfBlockByHash(_ context.Context, hash eth_common.Hash) (uint64, error) { + return m.blockTimes[hash], nil +} + +// ParseLogMessagePublished delegates to the real generated ABI parser so tests exercise the +// actual unpack path against logs produced by newTestLog. +func (m *mockConnector) ParseLogMessagePublished(log types.Log) (*ethabi.AbiLogMessagePublished, error) { + return m.filterer.ParseLogMessagePublished(log) +} +func (m *mockConnector) SubscribeForBlocks(ctx context.Context, errC chan error, sink chan<- *connectors.NewBlock) (ethereum.Subscription, error) { + panic("not implemented") +} +func (m *mockConnector) GetLatest(ctx context.Context) (latest, finalized, safe uint64, err error) { + panic("not implemented") +} +func (m *mockConnector) RawCallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error { + panic("not implemented") +} +func (m *mockConnector) RawBatchCallContext(ctx context.Context, b []rpc.BatchElem) error { + panic("not implemented") +} +func (m *mockConnector) Client() *ethclient.Client { + panic("not implemented") +} +func (m *mockConnector) SubscribeNewHead(ctx context.Context, ch chan<- *types.Header) (ethereum.Subscription, error) { + panic("not implemented") +} +func (m *mockConnector) Close() { + panic("not implemented") +} + +// newBlock creates a *connectors.NewBlock with sensible defaults for testing. +// Hash is deterministically derived from the block number. +func newBlock(number uint64, finality connectors.FinalityLevel) *connectors.NewBlock { + return &connectors.NewBlock{ + Number: big.NewInt(int64(number)), // #nosec G115 -- test-only + Hash: eth_common.BigToHash(big.NewInt(int64(number))), // #nosec G115 -- test-only + Time: number, + Finality: finality, + } +} + +// addPendingMsg adds an entry to w.pending and returns the key. +// The MessagePublication is constructed with TxID, EmitterChain, and ConsistencyLevel set from the arguments. +func (w *Watcher) addPendingMsg( + txHash eth_common.Hash, + blockHash eth_common.Hash, + effectiveCL uint8, + additionalBlocks uint64, + sequence uint64, +) pendingKey { + key := pendingKey{ + TxHash: txHash, + BlockHash: blockHash, + Sequence: sequence, + } + + msg := &common.MessagePublication{ + TxID: txHash.Bytes(), + EmitterChain: w.chainID, + } + + w.pending[key] = &pendingMessage{ + message: msg, + height: testBlockNumber, + effectiveCL: effectiveCL, + additionalBlocks: additionalBlocks, + } + + return key +} + +// enableCCL enables the CCL feature on the watcher and initializes the cache. +func (w *Watcher) enableCCL() { + w.cclEnabled = true + w.cclLogger = zap.NewNop() + w.cclCache = CCLCache{} +} + +// seedCCLAdditionalBlocks pre-populates the CCL cache for emitter with an AdditionalBlocks config. +// Allows for mocking, since no RPC calls will be made. +func (w *Watcher) seedCCLAdditionalBlocks(emitter eth_common.Address, consistencyLevel uint8, additionalBlocks uint16) { + var data [32]byte + data[0] = byte(AdditionalBlocksType) + data[1] = consistencyLevel + binary.BigEndian.PutUint16(data[2:4], additionalBlocks) + + w.cclCache[emitter] = CCLCacheEntry{ + data: data, + readTime: time.Now(), + } +} + +// seedCCLNothingSpecial pre-populates the CCL cache for emitter with a NothingSpecial config +// (all zeros), meaning no custom handling — treated as finalized. +func (w *Watcher) seedCCLNothingSpecial(emitter eth_common.Address) { + w.cclCache[emitter] = CCLCacheEntry{ + data: cclEmptyData, + readTime: time.Now(), + } +} + +// seedCCLInvalidType pre-populates the CCL cache for emitter with an invalid type byte, +// triggering a parse error in cclReadAndParseConfig. +func (w *Watcher) seedCCLInvalidType(emitter eth_common.Address) { + var data [32]byte + data[0] = 0xFF // invalid CCL request type + + w.cclCache[emitter] = CCLCacheEntry{ + data: data, + readTime: time.Now(), + } +} + +// testLogParams holds parameters for constructing a types.Log with ABI-encoded data +// matching the LogMessagePublished event. Only fields that affect control flow are exposed; +// nonce and payload are hardcoded to zero/empty. +type testLogParams struct { + sender eth_common.Address + sequence uint64 + consistencyLevel uint8 + txHash eth_common.Hash + blockHash eth_common.Hash + blockNumber uint64 + contractAddr eth_common.Address +} + +// logMessagePublishedArgs returns the ABI argument types for the non-indexed fields of +// LogMessagePublished(address indexed sender, uint64 sequence, uint32 nonce, bytes payload, uint8 consistencyLevel). +func logMessagePublishedArgs() abi.Arguments { + uint64Ty, _ := abi.NewType("uint64", "", nil) + uint32Ty, _ := abi.NewType("uint32", "", nil) + bytesTy, _ := abi.NewType("bytes", "", nil) + uint8Ty, _ := abi.NewType("uint8", "", nil) + + return abi.Arguments{ + {Name: "sequence", Type: uint64Ty}, + {Name: "nonce", Type: uint32Ty}, + {Name: "payload", Type: bytesTy}, + {Name: "consistencyLevel", Type: uint8Ty}, + } +} + +// newTestLog builds a types.Log with properly ABI-encoded data for the LogMessagePublished event. +// Use this for testing ParseLogMessagePublished and MessageEventsForTransaction. +// To test failure modes, mutate the returned log directly (e.g. truncate Data, swap Topics). +func newTestLog(t *testing.T, p testLogParams) types.Log { + t.Helper() + + txHash := p.txHash + if txHash == (eth_common.Hash{}) { + txHash = eth_common.BigToHash(big.NewInt(int64(p.blockNumber))) // #nosec G115 -- test-only + } + blockHash := p.blockHash + if blockHash == (eth_common.Hash{}) { + blockHash = eth_common.BigToHash(big.NewInt(int64(p.blockNumber + 0xff))) // #nosec G115 -- test-only + } + + data, err := logMessagePublishedArgs().Pack(p.sequence, uint32(0), []byte{}, p.consistencyLevel) + require.NoError(t, err) + + senderTopic := eth_common.BytesToHash(p.sender.Bytes()) + + return types.Log{ + Address: p.contractAddr, + Topics: []eth_common.Hash{LogMessagePublishedTopic, senderTopic}, + Data: data, + BlockNumber: p.blockNumber, + TxHash: txHash, + TxIndex: 0, + BlockHash: blockHash, + Index: 0, + Removed: false, + } +} + +// newValidWormholeLog returns a pointer to a LogMessagePublished log entry with valid ABI-encoded +// data that MessageEventsForTransaction will parse and surface as a MessagePublication. +// Defaults: sender = testEmitter, contract address = testEmitter, sequence = 1. +// newTestWatcher sets w.contract = testEmitter by default, so the by_transaction.go address filter passes. +// For full control over fields, use newTestLog with a testLogParams struct. +func newValidWormholeLog(t *testing.T, blockNumber uint64, consistencyLevel uint8) *types.Log { + t.Helper() + log := newTestLog(t, testLogParams{ + sender: testTokenBridge, + contractAddr: testEmitter, + sequence: 1, + consistencyLevel: consistencyLevel, + blockNumber: blockNumber, + }) + return &log +} + +// newTestReceipt builds a successful (Status=1) receipt at the given block number, wiring logs verbatim. +// BlockHash matches the default derivation used by newTestLog (blockNumber + 0xff) so a log built +// from the same blockNumber will have a matching receipt.BlockHash by default. +// For Status != 1 or a custom BlockHash, construct the receipt inline in the test. +func newTestReceipt(blockNumber uint64, logs []*types.Log) *types.Receipt { + return &types.Receipt{ + Status: 1, + BlockHash: eth_common.BigToHash(big.NewInt(int64(blockNumber + 0xff))), // #nosec G115 -- test-only + BlockNumber: big.NewInt(int64(blockNumber)), // #nosec G115 -- test-only + Logs: logs, + } +} + +// testLogEventParams holds parameters for constructing an AbiLogMessagePublished for postMessage tests. +// Only fields that affect control flow are exposed; nonce and payload are left at zero values. +type testLogEventParams struct { + sender eth_common.Address + sequence uint64 + consistencyLevel uint8 + txHash eth_common.Hash + blockHash eth_common.Hash + blockNumber uint64 +} + +// newTestLogEvent creates an AbiLogMessagePublished with sensible non-zero defaults for postMessage tests. +// Only blockNumber and consistencyLevel are required; sender and sequence get deterministic non-zero values. +func newTestLogEvent(consistencyLevel uint8) *ethabi.AbiLogMessagePublished { + return newTestLogEventFromParams(testLogEventParams{ + sender: testEmitter, + sequence: 1, + blockNumber: testBlockNumber, + consistencyLevel: consistencyLevel, + }) +} + +// newTestLogEventFromParams creates an AbiLogMessagePublished with full control over fields +// that affect control flow (sender, sequence, consistencyLevel, hashes). +func newTestLogEventFromParams(p testLogEventParams) *ethabi.AbiLogMessagePublished { + txHash := p.txHash + if txHash == (eth_common.Hash{}) { + txHash = eth_common.BigToHash(big.NewInt(int64(p.blockNumber))) // #nosec G115 -- test-only + } + blockHash := p.blockHash + if blockHash == (eth_common.Hash{}) { + blockHash = eth_common.BigToHash(big.NewInt(int64(p.blockNumber + 0xff))) // #nosec G115 -- test-only + } + + return ðabi.AbiLogMessagePublished{ + Sender: p.sender, + Sequence: p.sequence, + Payload: []byte{}, + ConsistencyLevel: p.consistencyLevel, + Raw: types.Log{ + Address: testEmitter, + Topics: []eth_common.Hash{LogMessagePublishedTopic}, + TxHash: txHash, + BlockHash: blockHash, + BlockNumber: p.blockNumber, + }, + } +} + +// assertMessageMatchesEvent verifies that all fields on a MessagePublication match the source event. +func assertMessageMatchesEvent(t *testing.T, msg *common.MessagePublication, ev *ethabi.AbiLogMessagePublished) { + t.Helper() + assert.Equal(t, ev.Raw.TxHash.Bytes(), msg.TxID) + assert.Equal(t, time.Unix(int64(testBlockTime), 0), msg.Timestamp) // #nosec G115 -- test-only + assert.Equal(t, ev.Nonce, msg.Nonce) + assert.Equal(t, ev.Sequence, msg.Sequence) + assert.Equal(t, vaa.ChainIDEthereum, msg.EmitterChain) + assert.Equal(t, PadAddress(ev.Sender), msg.EmitterAddress) + assert.Equal(t, ev.Payload, msg.Payload) + assert.Equal(t, ev.ConsistencyLevel, msg.ConsistencyLevel) +} + +// assertPendingMetadata verifies the metadata fields on a pendingMessage. +func assertPendingMetadata(t *testing.T, pe *pendingMessage, effectiveCL uint8, additionalBlocks uint64) { + t.Helper() + assert.Equal(t, effectiveCL, pe.effectiveCL) + assert.Equal(t, testBlockNumber, pe.height) + assert.Equal(t, additionalBlocks, pe.additionalBlocks) +}