add reader-node-firehose #63
Conversation
| a.zlogger.Info("launching reader log plugin") | ||
| mindreaderLogPlugin, err := mindreader.NewMindReaderPlugin( | ||
| a.Config.OneBlocksStoreURL, | ||
| a.Config.WorkingDir, | ||
| firehoseReader.NoopConsoleReader, | ||
| a.Config.StartBlockNum, | ||
| a.Config.StopBlockNum, | ||
| a.Config.MindReadBlocksChanCapacity, | ||
| a.modules.MetricsAndReadinessManager.UpdateHeadBlock, | ||
| func(_ error) {}, | ||
| a.Config.OneBlockSuffix, | ||
| blockStreamServer, | ||
| a.zlogger, | ||
| a.tracer, | ||
| ) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| a.zlogger.Debug("configuring shutter") | ||
| mindreaderLogPlugin.OnTerminated(a.Shutdown) | ||
| a.OnTerminating(mindreaderLogPlugin.Shutdown) |
There was a problem hiding this comment.
Ideally, we would avoid completely the mindreader plugin, there is overall little reason to use outside avoiding re-implementing two things:
- Save received one block to one-blocks-store
- Emit block on block stream server
I understand it make the implementation easier. But I have the overall feeling that it will make a much simpler implementation since we will move away from the node-manager.
We should also move to something else than node-manager as the top-level package.
There was a problem hiding this comment.
Sounds good. Should I move it into its own top-level package or is there another package to include it in?
There was a problem hiding this comment.
A new top-level reader-node-firehose seems good, the package <name> could be shorter if needed.
| stream, err := f.firehoseClient.Blocks(context.Background(), &pbfirehose.Request{ | ||
| StartBlockNum: int64(startBlock), | ||
| Cursor: string(cursor), | ||
| StopBlockNum: stopBlock, | ||
| FinalBlocksOnly: false, | ||
| }, f.callOpts...) | ||
| if err != nil { | ||
| return fmt.Errorf("failed to request block stream from Firehose: %w", err) | ||
| } |
There was a problem hiding this comment.
How do we deal with connection errors?
| ctx, s.cancelPeriodicLogger = context.WithCancel(ctx) | ||
|
|
||
| go func() { | ||
| ticker := time.NewTicker(logEach) |
There was a problem hiding this comment.
ticker has a Stop() let's defer stop it.
There was a problem hiding this comment.
blockRate too has a Stop method. Could go in StopPeriodicLogToZap() too.
|
Fred will continue on this in the new year. |
|
@matthewdarwin I reviewed the SQL PRs too. If there is something I missed, let me know. |
This adds a new reader-node app that reads blocks from another Firehose endpoint. This allows indexers to bootstrap a Firehose/Substreams stack without having to set up an instrumented blockchain node initially.