-
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.
add Silkworm poller for Antelope EVM (#108)
* add Silkworm poller for Antelope EVM * remove config.yaml
- Loading branch information
Showing
6 changed files
with
250 additions
and
3 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,123 @@ | ||
| package blockfetcher | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/abourget/llerrgroup" | ||
| pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" | ||
| "github.com/streamingfast/eth-go/rpc" | ||
| pbeth "github.com/streamingfast/firehose-ethereum/types/pb/sf/ethereum/type/v2" | ||
| "go.uber.org/zap" | ||
| "google.golang.org/protobuf/types/known/anypb" | ||
| "google.golang.org/protobuf/types/known/timestamppb" | ||
| ) | ||
|
|
||
| type ToEthBlock func(in *rpc.Block, receipts map[string]*rpc.TransactionReceipt) (*pbeth.Block, map[string]bool) | ||
|
|
||
| type BlockFetcher struct { | ||
| rpcClient *rpc.Client | ||
| latest uint64 | ||
| latestBlockRetryInterval time.Duration | ||
| fetchInterval time.Duration | ||
| toEthBlock ToEthBlock | ||
| lastFetchAt time.Time | ||
| logger *zap.Logger | ||
| } | ||
|
|
||
| func NewBlockFetcher(rpcClient *rpc.Client, intervalBetweenFetch, latestBlockRetryInterval time.Duration, toEthBlock ToEthBlock, logger *zap.Logger) *BlockFetcher { | ||
| return &BlockFetcher{ | ||
| rpcClient: rpcClient, | ||
| latestBlockRetryInterval: latestBlockRetryInterval, | ||
| toEthBlock: toEthBlock, | ||
| fetchInterval: intervalBetweenFetch, | ||
| logger: logger, | ||
| } | ||
| } | ||
|
|
||
| func (f *BlockFetcher) Fetch(ctx context.Context, blockNum uint64) (block *pbbstream.Block, err error) { | ||
| f.logger.Debug("fetching block", zap.Uint64("block_num", blockNum)) | ||
| for f.latest < blockNum { | ||
| f.latest, err = f.rpcClient.LatestBlockNum(ctx) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fetching latest block num: %w", err) | ||
| } | ||
|
|
||
| f.logger.Info("got latest block", zap.Uint64("latest", f.latest), zap.Uint64("block_num", blockNum)) | ||
|
|
||
| if f.latest < blockNum { | ||
| time.Sleep(f.latestBlockRetryInterval) | ||
| continue | ||
| } | ||
| break | ||
| } | ||
|
|
||
| sinceLastFetch := time.Since(f.lastFetchAt) | ||
| if sinceLastFetch < f.fetchInterval { | ||
| time.Sleep(f.fetchInterval - sinceLastFetch) | ||
| } | ||
|
|
||
| rpcBlock, err := f.rpcClient.GetBlockByNumber(ctx, rpc.BlockNumber(blockNum), rpc.WithGetBlockFullTransaction()) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fetching block %d: %w", blockNum, err) | ||
| } | ||
|
|
||
| receipts, err := FetchReceipts(ctx, rpcBlock, f.rpcClient) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fetching receipts for block %d %q: %w", rpcBlock.Number, rpcBlock.Hash.Pretty(), err) | ||
| } | ||
|
|
||
| f.logger.Debug("fetched receipts", zap.Int("count", len(receipts))) | ||
|
|
||
| f.lastFetchAt = time.Now() | ||
|
|
||
| if err != nil { | ||
| return nil, fmt.Errorf("fetching logs for block %d %q: %w", rpcBlock.Number, rpcBlock.Hash.Pretty(), err) | ||
| } | ||
|
|
||
| ethBlock, _ := f.toEthBlock(rpcBlock, receipts) | ||
| anyBlock, err := anypb.New(ethBlock) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("create any block: %w", err) | ||
| } | ||
|
|
||
| return &pbbstream.Block{ | ||
| Number: ethBlock.Number, | ||
| Id: ethBlock.GetFirehoseBlockID(), | ||
| ParentId: ethBlock.GetFirehoseBlockParentID(), | ||
| Timestamp: timestamppb.New(ethBlock.GetFirehoseBlockTime()), | ||
| LibNum: ethBlock.LIBNum(), | ||
| ParentNum: ethBlock.GetFirehoseBlockParentNumber(), | ||
| Payload: anyBlock, | ||
| }, nil | ||
| } | ||
|
|
||
| func FetchReceipts(ctx context.Context, block *rpc.Block, client *rpc.Client) (out map[string]*rpc.TransactionReceipt, err error) { | ||
| out = make(map[string]*rpc.TransactionReceipt) | ||
| lock := sync.Mutex{} | ||
|
|
||
| eg := llerrgroup.New(10) | ||
| for _, tx := range block.Transactions.Transactions { | ||
| if eg.Stop() { | ||
| continue // short-circuit the loop if we got an error | ||
| } | ||
| eg.Go(func() error { | ||
| receipt, err := client.TransactionReceipt(ctx, tx.Hash) | ||
| if err != nil { | ||
| return fmt.Errorf("fetching receipt for tx %q: %w", tx.Hash.Pretty(), err) | ||
| } | ||
| lock.Lock() | ||
| out[tx.Hash.Pretty()] = receipt | ||
| lock.Unlock() | ||
| return err | ||
| }) | ||
| } | ||
|
|
||
| if err := eg.Wait(); err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return | ||
| } |
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,29 @@ | ||
| package blockfetcher | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "go.uber.org/zap" | ||
|
|
||
| pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" | ||
| "github.com/streamingfast/eth-go/rpc" | ||
| "github.com/streamingfast/firehose-ethereum/block" | ||
| ) | ||
|
|
||
| type SilkwormBlockFetcher struct { | ||
| fetcher *BlockFetcher | ||
| } | ||
|
|
||
| func NewSilkwormBlockFetcher(rpcClient *rpc.Client, intervalBetweenFetch time.Duration, latestBlockRetryInterval time.Duration, logger *zap.Logger) *SilkwormBlockFetcher { | ||
| fetcher := NewBlockFetcher(rpcClient, intervalBetweenFetch, latestBlockRetryInterval, block.RpcToEthBlock, logger) | ||
| return &SilkwormBlockFetcher{ | ||
| fetcher: fetcher, | ||
| } | ||
| } | ||
|
|
||
| func (f *SilkwormBlockFetcher) PollingInterval() time.Duration { return 1 * time.Second } | ||
|
|
||
| func (f *SilkwormBlockFetcher) Fetch(ctx context.Context, blockNum uint64) (*pbbstream.Block, error) { | ||
| return f.fetcher.Fetch(ctx, blockNum) | ||
| } |
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,79 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "github.com/pinax-network/firehose-antelope/blockfetcher" | ||
| "path" | ||
| "strconv" | ||
| "time" | ||
|
|
||
| "github.com/spf13/cobra" | ||
| "github.com/streamingfast/bstream" | ||
| "github.com/streamingfast/cli/sflags" | ||
| "github.com/streamingfast/eth-go/rpc" | ||
| firecore "github.com/streamingfast/firehose-core" | ||
| "github.com/streamingfast/firehose-core/blockpoller" | ||
| "github.com/streamingfast/logging" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| func newPollerCmd(logger *zap.Logger, tracer logging.Tracer) *cobra.Command { | ||
| cmd := &cobra.Command{ | ||
| Use: "poller", | ||
| Short: "poll blocks from different sources", | ||
| } | ||
|
|
||
| cmd.AddCommand(newSilkwormPollerCmd(logger, tracer)) | ||
| return cmd | ||
| } | ||
|
|
||
| func newSilkwormPollerCmd(logger *zap.Logger, tracer logging.Tracer) *cobra.Command { | ||
| cmd := &cobra.Command{ | ||
| Use: "silkworm <rpc-endpoint> <first-streamable-block>", | ||
| Short: "poll blocks from silkworm rpc", | ||
| Args: cobra.ExactArgs(2), | ||
| RunE: pollerRunE(logger, tracer), | ||
| } | ||
| cmd.Flags().Duration("interval-between-fetch", 0, "interval between fetch") | ||
|
|
||
| return cmd | ||
| } | ||
|
|
||
| func pollerRunE(logger *zap.Logger, tracer logging.Tracer) firecore.CommandExecutor { | ||
| return func(cmd *cobra.Command, args []string) (err error) { | ||
| ctx := cmd.Context() | ||
|
|
||
| rpcEndpoint := args[0] | ||
|
|
||
| dataDir := sflags.MustGetString(cmd, "data-dir") | ||
| stateDir := path.Join(dataDir, "poller-state") | ||
|
|
||
| logger.Info("launching firehose-antelope poller", zap.String("rpc_endpoint", rpcEndpoint), zap.String("data_dir", dataDir), zap.String("state_dir", stateDir)) | ||
|
|
||
| rpcClient := rpc.NewClient(rpcEndpoint) | ||
|
|
||
| firstStreamableBlock, err := strconv.ParseUint(args[1], 10, 64) | ||
| if err != nil { | ||
| return fmt.Errorf("unable to parse first streamable block %d: %w", firstStreamableBlock, err) | ||
| } | ||
|
|
||
| fetchInterval := sflags.MustGetDuration(cmd, "interval-between-fetch") | ||
|
|
||
| fetcher := blockfetcher.NewSilkwormBlockFetcher(rpcClient, fetchInterval, 1*time.Second, logger) | ||
| handler := blockpoller.NewFireBlockHandler("type.googleapis.com/sf.ethereum.type.v2.Block") | ||
| poller := blockpoller.New(fetcher, handler, blockpoller.WithStoringState(stateDir), blockpoller.WithLogger(logger)) | ||
|
|
||
| // there is currently no support for rpc.FinalizedBlock on eos evm, so we use the latest one | ||
| latestBlock, err := rpcClient.GetBlockByNumber(ctx, rpc.LatestBlock) | ||
| if err != nil { | ||
| return fmt.Errorf("getting latest block: %w", err) | ||
| } | ||
|
|
||
| err = poller.Run(ctx, firstStreamableBlock, bstream.NewBlockRef(latestBlock.Hash.String(), uint64(latestBlock.Number))) | ||
| if err != nil { | ||
| return fmt.Errorf("running poller: %w", err) | ||
| } | ||
|
|
||
| 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