-
Notifications
You must be signed in to change notification settings - Fork 4.2k
feat(client/v2): broadcast logic #22282
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.
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
Merged
Merged
Changes from 3 commits
Commits
Show all changes
22 commits
Select commit
Hold shift + click to select a range
cb00e03
add: broadcast logic
JulianToledano 2025987
lint :)
JulianToledano 015eab4
Merge branch 'main' into julian/client-v2-broadcast
JulianToledano 39b5126
fix: make sure interfaces are implemented
JulianToledano 49598dd
del: server/v2 dependency
JulianToledano 8656f1f
fix: tests
JulianToledano b33e5a2
CHANGELOG
JulianToledano c13fe79
update: factory register
JulianToledano 4bd1fbb
update: factory and types public
JulianToledano bee57fa
update: mock
JulianToledano b0094f3
Merge branch 'main' into julian/client-v2-broadcast
JulianToledano 71e1c4a
./scripts/go-mod-tidy-all.sh
JulianToledano dc9e1f4
make mocks
JulianToledano ea3ffb8
update: rabbit enablements
JulianToledano cb472db
fix: create makes use of context
JulianToledano 4ecd77f
update: use broadcaster in v2/tx
JulianToledano 00fe5a3
update: remove factory + public
JulianToledano 86b8d58
lint + typo
JulianToledano c840773
lint :/
JulianToledano 5b3db1e
update: rabbit enhancements
JulianToledano 0c3855d
Merge branch 'main' into julian/client-v2-broadcast
JulianToledano f86767d
fix: checkCometError check
JulianToledano File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,51 @@ | ||
| package broadcast | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| ) | ||
|
|
||
| const ( | ||
| // BroadcastSync defines a tx broadcasting mode where the client waits for | ||
| // a CheckTx execution response only. | ||
| BroadcastSync = "sync" | ||
| // BroadcastAsync defines a tx broadcasting mode where the client returns | ||
| // immediately. | ||
| BroadcastAsync = "async" | ||
|
|
||
| // cometBftConsensus is the identifier for the CometBFT consensus engine. | ||
| cometBftConsensus = "comet" | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| ) | ||
|
|
||
| type ( | ||
| // Broadcaster defines an interface for broadcasting transactions to the consensus engine. | ||
| Broadcaster interface { | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| // Broadcast sends a transaction to the network and returns the result. | ||
| // | ||
| // It returns a byte slice containing the formatted result that will be | ||
| // passed to the output writer, and an error if the broadcast failed. | ||
| Broadcast(ctx context.Context, txBytes []byte) ([]byte, error) | ||
| } | ||
|
|
||
| // factory defines a generic interface for creating a Broadcaster. | ||
| factory interface { | ||
| // Create creates a new Broadcaster instance of type T. | ||
| create(ctx context.Context, consensus, url string, opts ...Option) (Broadcaster, error) | ||
| } | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
|
|
||
| // Option is a function that configures a Broadcaster. | ||
| Option func(Broadcaster) | ||
| ) | ||
|
|
||
| // broadcasterFactory is a factory for creating Broadcaster instances. | ||
| type broadcasterFactory struct{} | ||
|
|
||
| // create creates a new Broadcaster based on the given consensus type. | ||
| func (f broadcasterFactory) create(_ context.Context, consensus, url string, opts ...Option) (Broadcaster, error) { | ||
| switch consensus { | ||
| case cometBftConsensus: | ||
| return NewCometBftBroadcaster(url, opts...) | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| default: | ||
| return nil, fmt.Errorf("invalid consensus type: %s", consensus) | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package broadcast | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func Test_newBroadcaster(t *testing.T) { | ||
| var f factory | ||
| f = broadcasterFactory{} | ||
| tests := []struct { | ||
| name string | ||
| consensus string | ||
| opts []Option | ||
| want Broadcaster | ||
| wantErr bool | ||
| }{ | ||
| { | ||
| name: "comet", | ||
| consensus: "comet", | ||
| opts: []Option{ | ||
| withMode(BroadcastSync), | ||
| }, | ||
| want: &CometBftBroadcaster{}, | ||
| }, | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| } | ||
| for _, tt := range tests { | ||
| t.Run(tt.name, func(t *testing.T) { | ||
| got, err := f.create(context.Background(), tt.consensus, "localhost:26657", tt.opts...) | ||
| if tt.wantErr { | ||
| require.Error(t, err) | ||
| } else { | ||
| require.NoError(t, err) | ||
| require.NotNil(t, got) | ||
| require.IsType(t, tt.want, got) | ||
| } | ||
| }) | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,157 @@ | ||
| package broadcast | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "strings" | ||
|
|
||
| "github.com/cometbft/cometbft/mempool" | ||
| rpchttp "github.com/cometbft/cometbft/rpc/client/http" | ||
| coretypes "github.com/cometbft/cometbft/rpc/core/types" | ||
| cmttypes "github.com/cometbft/cometbft/types" | ||
|
|
||
| apiacbci "cosmossdk.io/api/cosmos/base/abci/v1beta1" | ||
| serverrpc "cosmossdk.io/server/v2/cometbft/client/rpc" | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
|
|
||
| "github.com/cosmos/cosmos-sdk/codec" | ||
| sdkerrors "github.com/cosmos/cosmos-sdk/types/errors" | ||
| ) | ||
|
JulianToledano marked this conversation as resolved.
|
||
|
|
||
| // CometBftBroadcaster implements the Broadcaster interface for CometBFT consensus engine. | ||
| type CometBftBroadcaster struct { | ||
| rpcClient serverrpc.CometRPC | ||
| mode string | ||
| cdc codec.JSONCodec | ||
| } | ||
|
|
||
| func withMode(mode string) func(broadcaster Broadcaster) { | ||
| return func(b Broadcaster) { | ||
| cbc, ok := b.(*CometBftBroadcaster) | ||
| if !ok { | ||
| return | ||
| } | ||
| cbc.mode = mode | ||
| } | ||
| } | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
|
|
||
| func withJsonCodec(codec codec.JSONCodec) func(broadcaster Broadcaster) { | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| return func(b Broadcaster) { | ||
| cbc, ok := b.(*CometBftBroadcaster) | ||
| if !ok { | ||
| return | ||
| } | ||
| cbc.cdc = codec | ||
| } | ||
| } | ||
|
|
||
| // NewCometBftBroadcaster creates a new CometBftBroadcaster. | ||
| func NewCometBftBroadcaster(rpcURL string, opts ...Option) (*CometBftBroadcaster, error) { | ||
| rpcClient, err := rpchttp.New(rpcURL) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create CometBft RPC client: %w", err) | ||
| } | ||
|
|
||
| bc := &CometBftBroadcaster{} | ||
| for _, opt := range opts { | ||
| opt(bc) | ||
| } | ||
|
|
||
| bc.rpcClient = *rpcClient | ||
| return bc, nil | ||
| } | ||
|
|
||
| // Broadcast sends a transaction to the network and returns the result. | ||
| // returns a byte slice containing the JSON-encoded result and an error if the broadcast failed. | ||
| func (c CometBftBroadcaster) Broadcast(ctx context.Context, txBytes []byte) ([]byte, error) { | ||
| var fn func(ctx context.Context, tx cmttypes.Tx) (*coretypes.ResultBroadcastTx, error) | ||
| switch c.mode { | ||
| case BroadcastSync: | ||
| fn = c.rpcClient.BroadcastTxSync | ||
| case BroadcastAsync: | ||
| fn = c.rpcClient.BroadcastTxAsync | ||
| default: | ||
| return []byte{}, fmt.Errorf("unknown broadcast mode: %s", c.mode) | ||
| } | ||
|
|
||
| res, err := c.broadcast(ctx, txBytes, fn) | ||
| if err != nil { | ||
| return []byte{}, err | ||
| } | ||
|
|
||
| return c.cdc.MarshalJSON(res) | ||
| } | ||
|
|
||
| // broadcast sends a transaction to the CometBFT network using the provided function. | ||
| func (c CometBftBroadcaster) broadcast(ctx context.Context, txbytes []byte, | ||
| fn func(ctx context.Context, tx cmttypes.Tx) (*coretypes.ResultBroadcastTx, error), | ||
| ) (*apiacbci.TxResponse, error) { | ||
| bResult, err := fn(ctx, txbytes) | ||
| if errRes := checkCometError(err, txbytes); err != nil { | ||
| return errRes, nil | ||
| } | ||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
|
|
||
| return newResponseFormatBroadcastTx(bResult), err | ||
| } | ||
|
|
||
|
JulianToledano marked this conversation as resolved.
Outdated
|
||
| // checkCometError checks for errors returned by the CometBFT network and returns an appropriate TxResponse. | ||
| // It extracts error information and constructs a TxResponse with the error details. | ||
| func checkCometError(err error, tx cmttypes.Tx) *apiacbci.TxResponse { | ||
| if err == nil { | ||
| return nil | ||
| } | ||
|
|
||
| errStr := strings.ToLower(err.Error()) | ||
| txHash := fmt.Sprintf("%X", tx.Hash()) | ||
|
|
||
| switch { | ||
| case strings.Contains(errStr, strings.ToLower(mempool.ErrTxInCache.Error())): | ||
| return &apiacbci.TxResponse{ | ||
| Code: sdkerrors.ErrTxInMempoolCache.ABCICode(), | ||
| Codespace: sdkerrors.ErrTxInMempoolCache.Codespace(), | ||
| Txhash: txHash, | ||
| } | ||
|
|
||
| case strings.Contains(errStr, "mempool is full"): | ||
| return &apiacbci.TxResponse{ | ||
| Code: sdkerrors.ErrMempoolIsFull.ABCICode(), | ||
| Codespace: sdkerrors.ErrMempoolIsFull.Codespace(), | ||
| Txhash: txHash, | ||
| } | ||
|
|
||
| case strings.Contains(errStr, "tx too large"): | ||
| return &apiacbci.TxResponse{ | ||
| Code: sdkerrors.ErrTxTooLarge.ABCICode(), | ||
| Codespace: sdkerrors.ErrTxTooLarge.Codespace(), | ||
| Txhash: txHash, | ||
| } | ||
|
|
||
| default: | ||
| return nil | ||
| } | ||
| } | ||
|
|
||
| // newResponseFormatBroadcastTx returns a TxResponse given a ResultBroadcastTx from cometbft | ||
| func newResponseFormatBroadcastTx(res *coretypes.ResultBroadcastTx) *apiacbci.TxResponse { | ||
| if res == nil { | ||
| return nil | ||
| } | ||
|
|
||
| parsedLogs, _ := parseABCILogs(res.Log) | ||
|
|
||
|
JulianToledano marked this conversation as resolved.
|
||
| return &apiacbci.TxResponse{ | ||
| Code: res.Code, | ||
| Codespace: res.Codespace, | ||
| Data: res.Data.String(), | ||
| RawLog: res.Log, | ||
| Logs: parsedLogs, | ||
| Txhash: res.Hash.String(), | ||
| } | ||
| } | ||
|
|
||
| // parseABCILogs attempts to parse a stringified ABCI tx log into a slice of | ||
| // ABCIMessageLog types. It returns an error upon JSON decoding failure. | ||
| func parseABCILogs(logs string) (res []*apiacbci.ABCIMessageLog, err error) { | ||
| err = json.Unmarshal([]byte(logs), &res) | ||
| return res, err | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.