Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[ship-3971] remove ws dep ccip #16143

Draft
wants to merge 5 commits into
base: develop
Choose a base branch
from
+467 −69
Draft

[ship-3971] remove ws dep ccip #16143

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
680151e
Fully functional HTTP
davidcauchi Jan 8, 2025
8f74051
Cleanup
davidcauchi Jan 30, 2025
eade523
Clean duplicate import
davidcauchi Jan 30, 2025
bb129d1
Fix linting issues
davidcauchi Jan 30, 2025
dccfbad
Fix linting issues
davidcauchi Jan 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
330 changes: 265 additions & 65 deletions integration-tests/ccip-tests/actions/ccip_helpers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@ import (
"golang.org/x/exp/rand"
"golang.org/x/sync/errgroup"

"github.com/smartcontractkit/chainlink-testing-framework/sentinel"

chainselectors "github.com/smartcontractkit/chain-selectors"

commonconfig "github.com/smartcontractkit/chainlink-common/pkg/config"
Expand Down Expand Up @@ -3477,38 +3479,7 @@ func (lane *CCIPLane) StartEventWatchers() error {
for {
select {
case e := <-sendReqEventLatest:
lane.Logger.Info().Msgf("CCIPSendRequested event received for seq number %d", e.Message.SequenceNumber)
eventsForTx, ok := lane.Source.CCIPSendRequestedWatcher.Load(e.Raw.TxHash.Hex())
if ok {
lane.Source.CCIPSendRequestedWatcher.Store(e.Raw.TxHash.Hex(), append(eventsForTx.([]*contracts.SendReqEventData),
&contracts.SendReqEventData{
MessageId: e.Message.MessageId,
SequenceNumber: e.Message.SequenceNumber,
DataLength: len(e.Message.Data),
NoOfTokens: len(e.Message.TokenAmounts),
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
Fee: e.Message.FeeTokenAmount,
}))
} else {
lane.Source.CCIPSendRequestedWatcher.Store(e.Raw.TxHash.Hex(), []*contracts.SendReqEventData{
{
MessageId: e.Message.MessageId,
SequenceNumber: e.Message.SequenceNumber,
DataLength: len(e.Message.Data),
NoOfTokens: len(e.Message.TokenAmounts),
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
Fee: e.Message.FeeTokenAmount,
},
})
}

lane.Source.CCIPSendRequestedWatcher = testutils.DeleteNilEntriesFromMap(lane.Source.CCIPSendRequestedWatcher)
processSendRequestedEvent(lane, e)
case <-lane.Context.Done():
return
}
Expand All @@ -3531,19 +3502,7 @@ func (lane *CCIPLane) StartEventWatchers() error {
for {
select {
case e := <-reportAcceptedEvent:
lane.Logger.Info().Interface("Interval", e.Report.Interval).Msgf("ReportAccepted event received")
for i := e.Report.Interval.Min; i <= e.Report.Interval.Max; i++ {
lane.Dest.ReportAcceptedWatcher.Store(i, &contracts.CommitStoreReportAccepted{
Min: e.Report.Interval.Min,
Max: e.Report.Interval.Max,
MerkleRoot: e.Report.MerkleRoot,
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
})
}
lane.Dest.ReportAcceptedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ReportAcceptedWatcher)
processReportAcceptedEvent(lane, e)
case <-lane.Context.Done():
return
}
Expand All @@ -3567,14 +3526,7 @@ func (lane *CCIPLane) StartEventWatchers() error {
for {
select {
case e := <-reportBlessedEvent:
lane.Logger.Info().Msgf("TaggedRootBlessed event received for root %x", e.TaggedRoot.Root)
if e.TaggedRoot.CommitStore == lane.Dest.CommitStore.EthAddress {
lane.Dest.ReportBlessedWatcher.Store(e.TaggedRoot.Root, &contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
})
}
lane.Dest.ReportBlessedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ReportBlessedWatcher)
processTaggedRootBlessedEvent(lane, e)
case <-lane.Context.Done():
return
}
Expand All @@ -3598,18 +3550,7 @@ func (lane *CCIPLane) StartEventWatchers() error {
for {
select {
case e := <-execStateChangedEventLatest:
lane.Logger.Info().Msgf("Execution state changed event received for seq number %d", e.SequenceNumber)
lane.Dest.ExecStateChangedWatcher.Store(e.SequenceNumber, &contracts.EVM2EVMOffRampExecutionStateChanged{
SequenceNumber: e.SequenceNumber,
MessageId: e.MessageId,
State: e.State,
ReturnData: e.ReturnData,
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
})
lane.Dest.ExecStateChangedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ExecStateChangedWatcher)
processExecutionStateChangedEvent(lane, e)
case <-lane.Context.Done():
return
}
Expand All @@ -3618,6 +3559,265 @@ func (lane *CCIPLane) StartEventWatchers() error {
return nil
}

func processSendRequestedEvent(lane *CCIPLane, e *evm_2_evm_onramp.EVM2EVMOnRampCCIPSendRequested) {
lane.Logger.Info().Msgf("CCIPSendRequested event received for seq number %d", e.Message.SequenceNumber)
eventsForTx, ok := lane.Source.CCIPSendRequestedWatcher.Load(e.Raw.TxHash.Hex())
if ok {
lane.Source.CCIPSendRequestedWatcher.Store(e.Raw.TxHash.Hex(), append(eventsForTx.([]*contracts.SendReqEventData),
&contracts.SendReqEventData{
MessageId: e.Message.MessageId,
SequenceNumber: e.Message.SequenceNumber,
DataLength: len(e.Message.Data),
NoOfTokens: len(e.Message.TokenAmounts),
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
Fee: e.Message.FeeTokenAmount,
}))
} else {
lane.Source.CCIPSendRequestedWatcher.Store(e.Raw.TxHash.Hex(), []*contracts.SendReqEventData{
{
MessageId: e.Message.MessageId,
SequenceNumber: e.Message.SequenceNumber,
DataLength: len(e.Message.Data),
NoOfTokens: len(e.Message.TokenAmounts),
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
Fee: e.Message.FeeTokenAmount,
},
})
}

lane.Source.CCIPSendRequestedWatcher = testutils.DeleteNilEntriesFromMap(lane.Source.CCIPSendRequestedWatcher)
}

func processReportAcceptedEvent(lane *CCIPLane, e *commit_store.CommitStoreReportAccepted) {
lane.Logger.Info().Interface("Interval", e.Report.Interval).Msgf("ReportAccepted event received")
for i := e.Report.Interval.Min; i <= e.Report.Interval.Max; i++ {
lane.Dest.ReportAcceptedWatcher.Store(i, &contracts.CommitStoreReportAccepted{
Min: e.Report.Interval.Min,
Max: e.Report.Interval.Max,
MerkleRoot: e.Report.MerkleRoot,
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
})
}

// Clean up nil entries
lane.Dest.ReportAcceptedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ReportAcceptedWatcher)
}

func processTaggedRootBlessedEvent(lane *CCIPLane, e *rmn_contract.RMNContractTaggedRootBlessed) {
// Process the event
lane.Logger.Info().Msgf("TaggedRootBlessed event received for root %x", e.TaggedRoot.Root)
if e.TaggedRoot.CommitStore == lane.Dest.CommitStore.EthAddress {
lane.Dest.ReportBlessedWatcher.Store(e.TaggedRoot.Root, &contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
})
}

// Clean up nil entries
lane.Dest.ReportBlessedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ReportBlessedWatcher)
}

func processExecutionStateChangedEvent(lane *CCIPLane, e *evm_2_evm_offramp.EVM2EVMOffRampExecutionStateChanged) {
lane.Logger.Info().Msgf("Execution state changed event received for seq number %d", e.SequenceNumber)
lane.Dest.ExecStateChangedWatcher.Store(e.SequenceNumber, &contracts.EVM2EVMOffRampExecutionStateChanged{
SequenceNumber: e.SequenceNumber,
MessageId: e.MessageId,
State: e.State,
ReturnData: e.ReturnData,
LogInfo: contracts.LogInfo{
BlockNumber: e.Raw.BlockNumber,
TxHash: e.Raw.TxHash,
},
})
lane.Dest.ExecStateChangedWatcher = testutils.DeleteNilEntriesFromMap(lane.Dest.ExecStateChangedWatcher)
}

func (lane *CCIPLane) StartEventWatchersPolling(sc *sentinel.SentinelCoordinator) error {
lane.Logger.Info().Msg("Starting event watchers, Polling")
if lane.Source.Common.ChainClient.GetNetworkConfig().FinalityDepth == 0 {
err := lane.Source.Common.ChainClient.PollFinality()
if err != nil {
return err
}
}

sendReqEventSub, err := sc.Sentinel.Subscribe(
lane.SourceChain.GetChainID().Int64(),
lane.Source.OnRamp.EthAddress,
evm_2_evm_onramp.EVM2EVMOnRampCCIPSendRequested{}.Topic(),
)
if err != nil {
log.Error().Err(err).Msg("error in setting up polling to CCIPSendRequested event")
return err
}
go func() {
defer func() {
if err := sc.Sentinel.Unsubscribe(lane.SourceChain.GetChainID().Int64(), lane.Source.OnRamp.EthAddress, evm_2_evm_onramp.EVM2EVMOnRampCCIPSendRequested{}.Topic(), sendReqEventSub); err != nil {
lane.Logger.Error().Err(err).Msg("failed to unsubscribe from CCIPSendRequested event")
}
}()
for {
select {
case <-sc.Ctx.Done():
return
case event, ok := <-sendReqEventSub:
if !ok {
lane.Logger.Info().Msg("Sentinel log channel closed. Exiting goroutine.")
return
}
typesLog, _ := sentinel.ConvertAPILogToTypesLog(event)
e, _ := lane.Source.OnRamp.Instance.Latest.EVM2EVMOnRampFilterer.ParseCCIPSendRequested(*typesLog)
processSendRequestedEvent(lane, e)
case <-lane.Context.Done():
return
}
}
}()

reportAcceptedSub, err := sc.Sentinel.Subscribe(
lane.DestChain.GetChainID().Int64(),
lane.Dest.CommitStore.EthAddress,
commit_store.CommitStoreReportAccepted{}.Topic(),
)
if err != nil {
log.Error().Err(err).Msg("error in setting up polling to ReportAccepted event")
return errors.New("failed to subscribe to ReportAccepted event")
}
go func() {
defer func() {
if err := sc.Sentinel.Unsubscribe(lane.DestChain.GetChainID().Int64(), lane.Dest.CommitStore.EthAddress, commit_store.CommitStoreReportAccepted{}.Topic(), reportAcceptedSub); err != nil {
lane.Logger.Error().Err(err).Msg("failed to unsubscribe from ReportAccepted event")
}
}()
for {
select {
case <-sc.Ctx.Done():
return
case event, ok := <-reportAcceptedSub:
if !ok {
lane.Logger.Info().Msg("Sentinel log channel closed. Exiting goroutine.")
return
}

// Convert the log and parse the ReportAccepted event
typesLog, err := sentinel.ConvertAPILogToTypesLog(event)
if err != nil {
lane.Logger.Error().Err(err).Msg("error converting Sentinel log to types.Log")
continue
}
e, err := lane.Dest.CommitStore.Instance.Latest.CommitStoreFilterer.ParseReportAccepted(*typesLog)
if err != nil {
lane.Logger.Error().Err(err).Msg("error parsing ReportAccepted event")
continue
}
processReportAcceptedEvent(lane, e)
case <-lane.Context.Done():
return
}
}
}()

if lane.Dest.Common.ARM != nil {
reportBlessedEventSub, err := sc.Sentinel.Subscribe(
lane.DestChain.GetChainID().Int64(),
lane.Dest.Common.ARM.EthAddress,
rmn_contract.RMNContractTaggedRootBlessed{}.Topic(),
)
if err != nil {
log.Error().Err(err).Msg("error in setting up polling to TaggedRootBlessed event")
return errors.New("failed to subscribe to TaggedRootBlessed event")
}

go func() {
defer func() {
if err := sc.Sentinel.Unsubscribe(lane.DestChain.GetChainID().Int64(), lane.Dest.Common.ARM.EthAddress, rmn_contract.RMNContractTaggedRootBlessed{}.Topic(), reportBlessedEventSub); err != nil {
lane.Logger.Error().Err(err).Msg("failed to unsubscribe from TaggedRootBlessed event")
}
}()
for {
select {
case <-sc.Ctx.Done():
return
case event, ok := <-reportBlessedEventSub:
if !ok {
lane.Logger.Info().Msg("Sentinel log channel closed. Exiting goroutine.")
return
}

// Convert the log and parse the TaggedRootBlessed event
typesLog, err := sentinel.ConvertAPILogToTypesLog(event)
if err != nil {
lane.Logger.Error().Err(err).Msg("error converting Sentinel log to types.Log")
continue
}
e, err := lane.Dest.Common.ARM.Instance.RMNContractFilterer.ParseTaggedRootBlessed(*typesLog)
if err != nil {
lane.Logger.Error().Err(err).Msg("error parsing TaggedRootBlessed event")
continue
}
processTaggedRootBlessedEvent(lane, e)
case <-lane.Context.Done():
return
}
}
}()
}

execStateChangedEventSub, err := sc.Sentinel.Subscribe(
lane.DestChain.GetChainID().Int64(),
lane.Dest.OffRamp.EthAddress,
evm_2_evm_offramp.EVM2EVMOffRampExecutionStateChanged{}.Topic(),
)
if err != nil {
log.Error().Err(err).Msg("error in setting up polling to ExecutionStateChanged event")
return errors.New("failed to subscribe to ExecutionStateChanged event")
}

go func() {
defer func() {
if err := sc.Sentinel.Unsubscribe(lane.DestChain.GetChainID().Int64(), lane.Dest.OffRamp.EthAddress, evm_2_evm_offramp.EVM2EVMOffRampExecutionStateChanged{}.Topic(), execStateChangedEventSub); err != nil {
lane.Logger.Error().Err(err).Msg("failed to unsubscribe from ExecutionStateChanged event")
}
}()
for {
select {
case <-sc.Ctx.Done():
return
case event, ok := <-execStateChangedEventSub:
if !ok {
lane.Logger.Info().Msg("Sentinel log channel closed. Exiting goroutine.")
return
}

// Convert the log and parse the ExecutionStateChanged event
typesLog, err := sentinel.ConvertAPILogToTypesLog(event)
if err != nil {
lane.Logger.Error().Err(err).Msg("error converting Sentinel log to types.Log")
continue
}
e, err := lane.Dest.OffRamp.Instance.Latest.EVM2EVMOffRampFilterer.ParseExecutionStateChanged(*typesLog)
if err != nil {
lane.Logger.Error().Err(err).Msg("error parsing ExecutionStateChanged event")
continue
}
processExecutionStateChangedEvent(lane, e)
case <-lane.Context.Done():
return
}
}
}()
return nil
}

func (lane *CCIPLane) CleanUp(clearFees bool) error {
lane.Logger.Info().Msg("Cleaning up lane")
if lane.Source.Common.ChainClient.GetNetworkConfig().FinalityDepth == 0 {
Expand Down
14 changes: 14 additions & 0 deletions integration-tests/ccip-tests/contracts/contract_models.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1442,6 +1442,20 @@ func (r *Router) SetOnRamp(chainSelector uint64, onRamp common.Address) error {

func (r *Router) CCIPSend(destChainSelector uint64, msg router.ClientEVM2AnyMessage, valueForNative *big.Int) (*types.Transaction, error) {
opts, err := r.client.TransactionOpts(r.client.GetDefaultWallet())
// print out opts
r.logger.Info().
Str("from", opts.From.Hex()).
Str("nonce", fmt.Sprintf("%v", opts.Nonce)).
Str("value", fmt.Sprintf("%v", opts.Value)).
Str("gasPrice", fmt.Sprintf("%v", opts.GasPrice)).
Str("gasFeeCap", fmt.Sprintf("%v", opts.GasFeeCap)).
Str("gasTipCap", fmt.Sprintf("%v", opts.GasTipCap)).
Uint64("gasLimit", opts.GasLimit).
Str("accessList", fmt.Sprintf("%v", opts.AccessList)).
Str("context", fmt.Sprintf("%v", opts.Context)).
Bool("noSend", opts.NoSend).
Msg("TransactOpts")

if err != nil {
return nil, fmt.Errorf("error getting transaction opts: %w", err)
}
Expand Down
Loading
Loading