achuprov/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#798] morph/subscriber: Remove RPC Init timeout

Browse source 
More convenient way is to fail straight away and
expect external restart.

Signed-off-by: Alex Vanin <alexey@nspcc.ru>
This commit is contained in:
Alex Vanin 2021-09-07 13:15:56 +03:00 committed by Alex Vanin
parent 4874b4ae92
commit 1edf40f4d6
3 changed files with 11 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
cmd/neofs-node/morph.go
View file

@ -190,7 +190,6 @@ func listenMorphNotifications(c *cfg) {
Log: c.log, Log: c.log,
Endpoint: endpoints[i], Endpoint: endpoints[i],
DialTimeout: timeout, DialTimeout: timeout,
RPCInitTimeout: 10 * time.Second,
StartFromBlock: fromSideChainBlock, StartFromBlock: fromSideChainBlock,
}) })
if err == nil { if err == nil {

2
pkg/innerring/innerring.go
View file

@ -7,7 +7,6 @@ import (
"fmt" "fmt"
"io" "io"
"net" "net"
"time"
"github.com/nspcc-dev/neo-go/pkg/core/block" "github.com/nspcc-dev/neo-go/pkg/core/block"
"github.com/nspcc-dev/neo-go/pkg/core/transaction" "github.com/nspcc-dev/neo-go/pkg/core/transaction"
@ -895,7 +894,6 @@ func createListener(ctx context.Context, p *chainParams) (event.Listener, error)
Log: p.log, Log: p.log,
Endpoint: p.cfg.GetString(p.name + ".endpoint.notification"), Endpoint: p.cfg.GetString(p.name + ".endpoint.notification"),
DialTimeout: p.cfg.GetDuration(p.name + ".dial_timeout"), DialTimeout: p.cfg.GetDuration(p.name + ".dial_timeout"),
RPCInitTimeout: 10 * time.Second,
StartFromBlock: p.from, StartFromBlock: p.from,
}) })
if err != nil { if err != nil {

27
pkg/morph/subscriber/subscriber.go
View file

@ -43,7 +43,6 @@ type (
Log *zap.Logger Log *zap.Logger
Endpoint string Endpoint string
DialTimeout time.Duration DialTimeout time.Duration
RPCInitTimeout time.Duration
StartFromBlock uint32 StartFromBlock uint32
} }
) )
@ -198,7 +197,7 @@ func New(ctx context.Context, p *Params) (Subscriber, error) {
zap.String("endpoint", p.Endpoint), zap.String("endpoint", p.Endpoint),
zap.Uint32("min_block_height", p.StartFromBlock)) zap.Uint32("min_block_height", p.StartFromBlock))
err = awaitHeight(wsClient, p.StartFromBlock, p.RPCInitTimeout) err = awaitHeight(wsClient, p.StartFromBlock)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@ -226,23 +225,19 @@ func New(ctx context.Context, p *Params) (Subscriber, error) {
// This function is required to avoid connections to unsynced RPC nodes, because // This function is required to avoid connections to unsynced RPC nodes, because
// they can produce events from the past that should not be processed by // they can produce events from the past that should not be processed by
// NeoFS nodes. // NeoFS nodes.
func awaitHeight(wsClient *client.WSClient, startFrom uint32, timeout time.Duration) error { func awaitHeight(wsClient *client.WSClient, startFrom uint32) error {
if startFrom == 0 { if startFrom == 0 {
return nil return nil
} }
for ch := time.After(timeout); ; { height, err := wsClient.GetBlockCount()
select { if err != nil {
case <-ch: return fmt.Errorf("could not get block height: %w", err)
return fmt.Errorf("could not init ws client: didn't reach expected height %d", startFrom)
default:
}
height, err := wsClient.GetBlockCount()
if err != nil {
return fmt.Errorf("could not get block height: %w", err)
} else if height >= startFrom {
return nil
}
time.Sleep(100 * time.Millisecond)
} }
if height < startFrom {
return fmt.Errorf("RPC block counter %d didn't reach expected height %d", height, startFrom)
}
return nil
} }