Add blocksapi blocks loader #122
Conversation
|
Amazing work! I will have some comments though, already in process of review |
| ToBlock uint64 `mapstructure:"toBlock"` | ||
| ForceReindex bool `mapstructure:"forceReindex"` | ||
|
|
||
| RetryCountOnFailure uint8 `mapstructure:"retryCountOnFailure"` |
There was a problem hiding this comment.
Nitpick: those two parameters are also refiner-specific
indexer/indexer.go
Outdated
| fromBlockUpdated = true | ||
| } | ||
| } | ||
| } | ||
|
|
||
| if config.FromBlock < fromBlock { |
There was a problem hiding this comment.
I think you broke the case when config.FromBlock < config.GenesisBlock
In this case config.FromBlock must become equal to config.GenesisBlock
There was a problem hiding this comment.
It will be checked in indexer.go:108 for both implementations
indexer/indexer.go
Outdated
| if (config.ToBlock > DefaultToBlock) && (config.ToBlock <= config.FromBlock) { | ||
| if fromBlockUpdated && (cfgFromBlockOrigin < config.ToBlock) { | ||
| if fromBlockUpdated && (config.FromBlock < config.ToBlock) { |
There was a problem hiding this comment.
Aha, this condition becomes broken as well actually.
It should essentially track wether it's config that is wrong, or we just already progressed further.
So let's restore that cfgFromBlockOrigin variable, it's actually useful here
main.go
Outdated
| signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) | ||
| <-sig | ||
|
|
||
| // graceful interrupt net requests and io | ||
| cancel() |
There was a problem hiding this comment.
nitpick: better to defer it. That will also handle panics
| @@ -0,0 +1,298 @@ | |||
| package indexer | |||
|
|
|||
There was a problem hiding this comment.
Somehow git doesn't recognize this as a renamed file.
Just a suggestion on how to try achieving this next time (no worries now):
git mv indexer.go refiner.go
git commit ...And only then modify and create another indexer.go
indexer/blockapi.go
Outdated
| l *log.Logger | ||
| b broker.Broker | ||
| grpc *grpc.ClientConn | ||
| height uint64 |
There was a problem hiding this comment.
I would name it nextHeight, a bit more descriptive
| return | ||
| } | ||
|
|
||
| switch r := response.Response.(type) { |
There was a problem hiding this comment.
There's also two other cases that have to be handled:
case *blocksapi.ReceiveBlocksResponse_Error_- basically error on server side. Need to be logged and then you get back to reconnection loopcase *blocksapi.ReceiveBlocksResponse_Done_- you have reached theStopTargetsuccessfully
indexer/blockapi.go
Outdated
|
|
||
| callCtx := metadata.NewOutgoingContext(ctx, md) | ||
|
|
||
| go i.run(callCtx) |
There was a problem hiding this comment.
Now on the reconnection loop.
It can happen that connection was interrupted, or server has returned error because of some temporary problem.
Currently this code only does single attempt and will basically hang in case of any error.
What needs to be done is a loop.
Each time connection error, EOF or receive error happens - run needs to be called again. So just make a loop around that
There was a problem hiding this comment.
Something like:
wg.Add(1)
go func() {
defer wg.Done()
for {
if indexerCtx.Err() != nil {
return
}
if err := run(indexerCtx); err == nil {
// Maybe we finished completely (if !force-reindex && toblock != 0)
// in that case just return
// otherwise handle switch from reindex to normal endless indexing
}
// TODO: Log error
// That function doesn't exist but I guess you understand what I mean
if !sleepCtx(indexerCtx, time.Second/5) {
return
}
}
}()| callCtx := metadata.NewOutgoingContext(ctx, md) | ||
|
|
||
| go i.run(callCtx) | ||
| } |
There was a problem hiding this comment.
And we should also repeat the logic from here:
relayer2-public/indexer/indexer.go
Line 262 in 4be29a0
Basically in case of
ForceReindex - we should first do the reindex, and once reindex is done - we should start reading from whatever is the current state.
indexer/refiner_test.go
Outdated
| @@ -263,7 +263,7 @@ func TestConfiguration(t *testing.T) { | |||
| args[0].(*db.StoreHandler).Close() | |||
| }, | |||
| call: func(args ...interface{}) (interface{}, error) { | |||
| i, err := New(args[0].(*db.StoreHandler)) | |||
| i, err := NewIndexerRefiner(args[0].(*db.StoreHandler), nil) | |||
There was a problem hiding this comment.
not enough arguments in call to NewIndexerRefiner
(here and below)
| // Register protobuf plugin for generate optimized marshall & unmarshal code | ||
| encoding.RegisterCodec(vtgrpc.Codec{}) | ||
| } | ||
|
|
There was a problem hiding this comment.
Also please add some logging so the progress of indexing is visible.
I would simply start second goroutine that would every couple of seconds log the current height, speed, etc
Something like I did here: https://github.com/aurora-is-near/block-storage/blob/main/cmd/testconsumer/testconsumer.go#L213
Indexer and this logging goroutine can exchange state via atomic vars easily
mrdimidium
force-pushed
the
nikolay/blockapi
branch
from
1584dcc to
ebc9bcd
Compare
mrdimidium
force-pushed
the
nikolay/blockapi
branch
from
ebc9bcd to
9cc80e6
Compare
Blocks come in Borealis format, not JSON
Remove dependency of account setting in uncommitted local file
Adds a new way to get blocks from the Aurora API. Set
source: "blocksapi"in the relay config to use this method.