Bound retry loops and detect reorgs to prevent sync stalls

bchain/coins/avalanche/evm.go

@@ -93,9 +93,12 @@ func (c *AvalancheRPCClient) EthSubscribe(ctx context.Context, channel interface
 func (c *AvalancheRPCClient) CallContext(ctx context.Context, result interface{}, method string, args ...interface{}) error {
 	err := c.Client.CallContext(ctx, result, method, args...)
 	// unfinalized data cannot be queried error returned when trying to query a block height greater than last finalized block
-	// do not throw rpc error and instead treat as ErrBlockNotFound
+	// treat as ErrBlockNotFound so sync retries instead of processing an empty result
 	// https://docs.avax.network/quickstart/exchanges/integrate-exchange-with-avalanche#determining-finality
-	if err != nil && !strings.Contains(err.Error(), "cannot query unfinalized data") {
+	if err != nil {
+		if strings.Contains(err.Error(), "cannot query unfinalized data") {
+			return bchain.ErrBlockNotFound
+		}
 		return err
 	}
 	return nil

bchain/coins/avalanche/evm_test.go

@@ -0,0 +1,73 @@
+package avalanche
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/rpc"
+	"github.com/trezor/blockbook/bchain"
+)
+
+type testAvalancheRPCService struct{}
+
+func (s *testAvalancheRPCService) Unfinalized() (string, error) {
+	return "", errors.New("cannot query unfinalized data")
+}
+
+func (s *testAvalancheRPCService) OtherError() (string, error) {
+	return "", errors.New("other failure")
+}
+
+func (s *testAvalancheRPCService) Success() (string, error) {
+	return "ok", nil
+}
+
+func newTestAvalancheRPCClient(t *testing.T) *AvalancheRPCClient {
+	t.Helper()
+
+	server := rpc.NewServer()
+	if err := server.RegisterName("test", &testAvalancheRPCService{}); err != nil {
+		t.Fatalf("RegisterName() error = %v", err)
+	}
+	client := rpc.DialInProc(server)
+	t.Cleanup(func() {
+		client.Close()
+		server.Stop()
+	})
+
+	return &AvalancheRPCClient{Client: client}
+}
+
+func TestAvalancheRPCClientCallContextMapsUnfinalizedDataToBlockNotFound(t *testing.T) {
+	client := newTestAvalancheRPCClient(t)
+
+	var result string
+	err := client.CallContext(context.Background(), &result, "test_unfinalized")
+	if !errors.Is(err, bchain.ErrBlockNotFound) {
+		t.Fatalf("CallContext() error = %v, want ErrBlockNotFound", err)
+	}
+}
+
+func TestAvalancheRPCClientCallContextReturnsOtherErrors(t *testing.T) {
+	client := newTestAvalancheRPCClient(t)
+
+	var result string
+	err := client.CallContext(context.Background(), &result, "test_otherError")
+	if err == nil || !strings.Contains(err.Error(), "other failure") {
+		t.Fatalf("CallContext() error = %v, want other failure", err)
+	}
+}
+
+func TestAvalancheRPCClientCallContextReturnsResult(t *testing.T) {
+	client := newTestAvalancheRPCClient(t)
+
+	var result string
+	if err := client.CallContext(context.Background(), &result, "test_success"); err != nil {
+		t.Fatalf("CallContext() error = %v", err)
+	}
+	if result != "ok" {
+		t.Fatalf("result = %q, want %q", result, "ok")
+	}
+}

bchain/coins/eth/ethrpc.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/hex"
 	"encoding/json"
+	stdErrors "errors"
 	"fmt"
 	"io"
 	"math/big"
@@ -901,7 +902,7 @@ func (b *EthereumRPC) GetBlockHash(height uint32) (string, error) {
 	defer cancel()
 	h, err := b.Client.HeaderByNumber(ctx, &n)
 	if err != nil {
-		if err == ethereum.NotFound {
+		if err == ethereum.NotFound || stdErrors.Is(err, bchain.ErrBlockNotFound) {
 			return "", bchain.ErrBlockNotFound
 		}
 		return "", errors.Annotatef(err, "height %v", height)

bchain/coins/eth/ethrpc_blockhash_test.go

@@ -0,0 +1,92 @@
+package eth
+
+import (
+	"context"
+	stdErrors "errors"
+	"math/big"
+	"testing"
+	"time"
+
+	ethereum "github.com/ethereum/go-ethereum"
+	"github.com/trezor/blockbook/bchain"
+)
+
+type stubEVMHeader struct {
+	hash string
+	num  *big.Int
+}
+
+func (h *stubEVMHeader) Hash() string         { return h.hash }
+func (h *stubEVMHeader) Number() *big.Int     { return h.num }
+func (h *stubEVMHeader) Difficulty() *big.Int { return big.NewInt(0) }
+
+type stubEVMClient struct {
+	header bchain.EVMHeader
+	err    error
+}
+
+func (s *stubEVMClient) NetworkID(ctx context.Context) (*big.Int, error) { return nil, nil }
+func (s *stubEVMClient) HeaderByNumber(ctx context.Context, number *big.Int) (bchain.EVMHeader, error) {
+	if s.err != nil {
+		return nil, s.err
+	}
+	return s.header, nil
+}
+func (s *stubEVMClient) SuggestGasPrice(ctx context.Context) (*big.Int, error) { return nil, nil }
+func (s *stubEVMClient) EstimateGas(ctx context.Context, msg interface{}) (uint64, error) {
+	return 0, nil
+}
+func (s *stubEVMClient) BalanceAt(ctx context.Context, addrDesc bchain.AddressDescriptor, blockNumber *big.Int) (*big.Int, error) {
+	return nil, nil
+}
+func (s *stubEVMClient) NonceAt(ctx context.Context, addrDesc bchain.AddressDescriptor, blockNumber *big.Int) (uint64, error) {
+	return 0, nil
+}
+
+func newTestEthereumRPC(client bchain.EVMClient) *EthereumRPC {
+	return &EthereumRPC{Client: client, Timeout: time.Second}
+}
+
+func TestGetBlockHashMapsErrBlockNotFound(t *testing.T) {
+	// AvalancheClient.HeaderByNumber returns bchain.ErrBlockNotFound directly
+	// after the unfinalized-data fix. GetBlockHash must recognize that and
+	// surface ErrBlockNotFound to sync.go's stdErrors.Is check; otherwise
+	// fork detection breaks because juju/errors v0.0.0-2017 has no Unwrap.
+	b := newTestEthereumRPC(&stubEVMClient{err: bchain.ErrBlockNotFound})
+	_, err := b.GetBlockHash(123)
+	if !stdErrors.Is(err, bchain.ErrBlockNotFound) {
+		t.Fatalf("GetBlockHash() error = %v, want ErrBlockNotFound", err)
+	}
+}
+
+func TestGetBlockHashMapsEthereumNotFound(t *testing.T) {
+	b := newTestEthereumRPC(&stubEVMClient{err: ethereum.NotFound})
+	_, err := b.GetBlockHash(123)
+	if !stdErrors.Is(err, bchain.ErrBlockNotFound) {
+		t.Fatalf("GetBlockHash() error = %v, want ErrBlockNotFound", err)
+	}
+}
+
+func TestGetBlockHashPropagatesOtherErrors(t *testing.T) {
+	other := stdErrors.New("rpc connection refused")
+	b := newTestEthereumRPC(&stubEVMClient{err: other})
+	_, err := b.GetBlockHash(123)
+	if err == nil {
+		t.Fatal("GetBlockHash() error = nil, want propagated error")
+	}
+	if stdErrors.Is(err, bchain.ErrBlockNotFound) {
+		t.Fatalf("GetBlockHash() error = %v, must not match ErrBlockNotFound", err)
+	}
+}
+
+func TestGetBlockHashReturnsHash(t *testing.T) {
+	header := &stubEVMHeader{hash: "0xdeadbeef", num: big.NewInt(123)}
+	b := newTestEthereumRPC(&stubEVMClient{header: header})
+	got, err := b.GetBlockHash(123)
+	if err != nil {
+		t.Fatalf("GetBlockHash() error = %v", err)
+	}
+	if got != "0xdeadbeef" {
+		t.Fatalf("GetBlockHash() = %q, want %q", got, "0xdeadbeef")
+	}
+}

blockbook.go

@@ -547,19 +547,30 @@ func syncIndexLoop() {
 	defer close(chanSyncIndexDone)
 	glog.Info("syncIndexLoop starting")
 	// resync index about every 15 minutes if there are no chanSyncIndex requests, with debounce 1 second
+	logSyncErr := func(err error, suffix string) {
+		// Transient backend conditions (timeouts, missing-block-at-tip, connection blips)
+		// are expected during sync and the loop already retries from them,
+		// so demote the log level to keep operator alerting (set on level=ERROR) signal-rich.
+		msg := errors.ErrorStack(err) + suffix
+		if db.IsTransientSyncError(err) {
+			glog.Warning("syncIndexLoop ", msg)
+		} else {
+			glog.Error("syncIndexLoop ", msg)
+		}
+	}
 	common.TickAndDebounce(time.Duration(*resyncIndexPeriodMs)*time.Millisecond, time.Duration(*resyncIndexDebounceMs)*time.Millisecond, chanSyncIndex, func() {
 		if err := syncWorker.ResyncIndex(onNewBlock, false); err != nil {
 			if err == db.ErrOperationInterrupted || common.IsInShutdown() {
 				return
 			}
-			glog.Error("syncIndexLoop ", errors.ErrorStack(err), ", will retry...")
+			logSyncErr(err, ", will retry...")
 			// retry once in case of random network error, after a slight delay
 			time.Sleep(time.Millisecond * 2500)
 			if err := syncWorker.ResyncIndex(onNewBlock, false); err != nil {
 				if err == db.ErrOperationInterrupted || common.IsInShutdown() {
 					return
 				}
-				glog.Error("syncIndexLoop ", errors.ErrorStack(err))
+				logSyncErr(err, "")
 			}
 		}
 	})