-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
jsonrpc: start to implement the jsonrpc client side
- Loading branch information
Showing
10 changed files
with
317 additions
and
31 deletions.
There are no files selected for viewing
This file contains 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 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 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,117 @@ | ||
| package jsonrpc | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
| "log/slog" | ||
| "net/http" | ||
| "sync" | ||
|
|
||
| "github.com/duneanalytics/blockchain-ingester/lib/hexutils" | ||
| "github.com/duneanalytics/blockchain-ingester/models" | ||
| "github.com/hashicorp/go-retryablehttp" | ||
| ) | ||
|
|
||
| type BlockchainClient interface { | ||
| LatestBlockNumber() (int64, error) | ||
| BlockByNumber(ctx context.Context, blockNumber int64) (models.RPCBlock, error) | ||
|
|
||
| // SendBlocks sends blocks from startBlockNumber to endBlockNumber to outChan, inclusive. | ||
| // If endBlockNumber is -1, it sends blocks from startBlockNumber to the tip of the chain | ||
| // it will run continueously until the context is cancelled | ||
| SendBlocks(ctx context.Context, outChan chan models.RPCBlock, startBlockNumber, endBlockNumber int64) error | ||
|
|
||
| Close() error | ||
| } | ||
|
|
||
| const ( | ||
| MaxRetries = 10 | ||
| ) | ||
|
|
||
| type rpcClient struct { | ||
| client *retryablehttp.Client | ||
| cfg Config | ||
| log *slog.Logger | ||
| bufPool *sync.Pool | ||
| } | ||
|
|
||
| func NewRPCClient(cfg Config, log *slog.Logger) *rpcClient { // revive:disable-line:unexported-return | ||
| client := retryablehttp.NewClient() | ||
| client.RetryMax = MaxRetries | ||
| client.Logger = log | ||
| client.CheckRetry = retryablehttp.DefaultRetryPolicy | ||
| client.Backoff = retryablehttp.LinearJitterBackoff | ||
| return &rpcClient{ | ||
| client: client, | ||
| cfg: cfg, | ||
| log: log, | ||
| bufPool: &sync.Pool{ | ||
| New: func() interface{} { | ||
| return new(bytes.Buffer) | ||
| }, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| func (c *rpcClient) LatestBlockNumber() (int64, error) { | ||
| buf := c.bufPool.Get().(*bytes.Buffer) | ||
| defer c.bufPool.Put(buf) | ||
| buf.Reset() | ||
|
|
||
| err := c.getResponseBody(context.Background(), "eth_blockNumber", []any{}, buf) | ||
| if err != nil { | ||
| c.log.Error("Failed to get response for jsonRPC", | ||
| "method", "eth_blockNumber", | ||
| "error", err, | ||
| ) | ||
| return 0, err | ||
| } | ||
| resp := struct { | ||
| Result string `json:"result"` | ||
| }{} | ||
| if err := json.NewDecoder(buf).Decode(&resp); err != nil { | ||
| c.log.Error("Failed to decode response for jsonRPC", "error", err) | ||
| return 0, err | ||
| } | ||
| return hexutils.IntFromHex(resp.Result) | ||
| } | ||
|
|
||
| // getResponseBody sends a request to the OpStack server and returns the response body | ||
| func (c *rpcClient) getResponseBody( | ||
| ctx context.Context, method string, params interface{}, output *bytes.Buffer, | ||
| ) error { | ||
| reqData := map[string]interface{}{ | ||
| "jsonrpc": "2.0", | ||
| "id": 1, | ||
| "method": method, | ||
| "params": params, | ||
| } | ||
| encoder := json.NewEncoder(output) | ||
| if err := encoder.Encode(reqData); err != nil { | ||
| return err | ||
| } | ||
| req, err := retryablehttp.NewRequestWithContext(ctx, http.MethodPost, c.cfg.URL, output) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| resp, err := c.client.Do(req) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to send request for method %s: %w", method, err) | ||
| } | ||
| defer resp.Body.Close() | ||
| if resp.StatusCode != http.StatusOK { | ||
| return fmt.Errorf("response for method %s has status code %d", method, resp.StatusCode) | ||
| } | ||
|
|
||
| output.Reset() | ||
| if _, err := output.ReadFrom(resp.Body); err != nil { | ||
| return fmt.Errorf("failed to read response body for method %s: %w", method, err) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (c *rpcClient) Close() error { | ||
| return nil | ||
| } |
This file contains 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,8 @@ | ||
| package jsonrpc | ||
|
|
||
| import "time" | ||
|
|
||
| type Config struct { | ||
| URL string | ||
| PollInterval time.Duration | ||
| } |
This file contains 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,104 @@ | ||
| package jsonrpc | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "log/slog" | ||
|
|
||
| "github.com/duneanalytics/blockchain-ingester/models" | ||
| "golang.org/x/sync/errgroup" | ||
| ) | ||
|
|
||
| type OpStackClient struct { | ||
| rpcClient | ||
| } | ||
|
|
||
| var _ BlockchainClient = &OpStackClient{} | ||
|
|
||
| func NewOpStackClient(cfg Config, log *slog.Logger) *OpStackClient { | ||
| rpcClient := NewRPCClient(cfg, log) | ||
| return &OpStackClient{*rpcClient} | ||
| } | ||
|
|
||
| // BlockByNumber returns the block with the given blockNumber. | ||
| // it uses 3 different methods to get the block: | ||
| // 1. eth_getBlockByNumber | ||
| // 2. eth_getBlockReceipts | ||
| // 3. debug_traceBlockByNumber with tracer "callTracer" | ||
| // We encode the payload in NDJSON, in this order. | ||
| // TODO: debug_traceBlockByNumber should be optional | ||
| // | ||
| // we should handle the case where it is not available | ||
| func (c *OpStackClient) BlockByNumber(ctx context.Context, blockNumber int64) (models.RPCBlock, error) { | ||
| blockNumberHex := fmt.Sprintf("0x%x", blockNumber) | ||
|
|
||
| // TODO: split this into mandatory and optional methods | ||
| methods := []string{ | ||
| "eth_getBlockByNumber", | ||
| "eth_getBlockReceipts", | ||
| "debug_traceBlockByNumber", | ||
| } | ||
| methodArgs := map[string][]any{ | ||
| "eth_getBlockByNumber": {blockNumberHex, true}, | ||
| "eth_getBlockReceipts": {blockNumberHex}, | ||
| "debug_traceBlockByNumber": {blockNumberHex, `{"tracer":"callTracer"}`}, | ||
| } | ||
| group, ctx := errgroup.WithContext(ctx) | ||
| results := make([]*bytes.Buffer, len(methods)) | ||
| for i, method := range methods { | ||
|
|
||
| results[i] = c.bufPool.Get().(*bytes.Buffer) | ||
| defer c.bufPool.Put(results[i]) | ||
|
|
||
| group.Go(func() error { | ||
| results[i].Reset() | ||
| err := c.getResponseBody(ctx, method, methodArgs[method], results[i]) | ||
| if err != nil { | ||
| c.log.Error("Failed to get response for jsonRPC", | ||
| "method", method, | ||
| "error", err, | ||
| ) | ||
| } | ||
| return err | ||
| }) | ||
| } | ||
|
|
||
| if err := group.Wait(); err != nil { | ||
| return models.RPCBlock{}, err | ||
| } | ||
|
|
||
| // copy the responses in order | ||
| var buffer bytes.Buffer | ||
| for _, res := range results { | ||
| buffer.Grow(res.Len() + 1) | ||
| buffer.ReadFrom(res) | ||
| buffer.WriteString("\n") | ||
| } | ||
| return models.RPCBlock{ | ||
| BlockNumber: blockNumber, | ||
| Payload: buffer.Bytes(), | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *OpStackClient) SendBlocks( | ||
| ctx context.Context, outChan chan models.RPCBlock, startBlockNumber, endBlockNumber int64, | ||
| ) error { | ||
| dontStop := endBlockNumber <= startBlockNumber | ||
| for blockNumber := startBlockNumber; dontStop || startBlockNumber <= endBlockNumber; blockNumber++ { | ||
| block, err := c.BlockByNumber(ctx, blockNumber) | ||
| if err != nil { | ||
| c.log.Error("Failed to get block by number", | ||
| "blockNumber", blockNumber, | ||
| "error", err, | ||
| ) | ||
| return err | ||
| } | ||
| select { | ||
| case <-ctx.Done(): | ||
| return nil | ||
| case outChan <- block: | ||
| } | ||
| } | ||
| return nil | ||
| } |
This file contains 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 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 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
Oops, something went wrong.