Algod Importer: Longer Timeout

conduit/plugins/importers/algod/algod_importer.go

@@ -42,7 +42,7 @@
 )
 
 var (
-	waitForRoundTimeout = 5 * time.Second
+	waitForRoundTimeout = 15 * time.Second
 )
 
 const catchpointsURL = "https://algorand-catchpoints.s3.us-east-2.amazonaws.com/consolidated/%s_catchpoints.txt"
@@ -68,14 +68,14 @@
 }
 
 func (algodImp *algodImporter) OnComplete(input data.BlockData) error {
 	if algodImp.mode == followerMode {
 		syncRound := input.Round() + 1
 		_, err := algodImp.aclient.SetSyncRound(syncRound).Do(algodImp.ctx)
 		algodImp.logger.Tracef("importer algod.OnComplete(BlockData) called SetSyncRound(syncRound=%d) err: %v", syncRound, err)
 		return err
 	}
 
 	return nil
 }
 
 func (algodImp *algodImporter) Metadata() plugins.Metadata {
@@ -165,11 +165,11 @@
 func getMissingCatchpointLabel(URL string, nextRound uint64) (string, error) {
 	resp, err := http.Get(URL)
 	if err != nil {
 		return "", err
 	}
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return "", fmt.Errorf("failed to read catchpoint label response: %w", err)
 	}
 
 	if resp.StatusCode != 200 {
@@ -184,7 +184,7 @@
 		line := scanner.Text()
 		round, err := parseCatchpointRound(line)
 		if err != nil {
 			return "", err
 		}
 		// TODO: Change >= to > after go-algorand#5352 is fixed.
 		if uint64(round) >= nextRound {
@@ -370,7 +370,7 @@
 	algodImp.aclient = client
 	genesisResponse, err := algodImp.aclient.GetGenesis().Do(algodImp.ctx)
 	if err != nil {
 		return err
 	}
 
 	genesis := sdk.Genesis{}
@@ -378,7 +378,7 @@
 	// Don't fail on unknown properties here since the go-algorand and SDK genesis types differ slightly
 	err = json.LenientDecode([]byte(genesisResponse), &genesis)
 	if err != nil {
 		return err
 	}
 	if reflect.DeepEqual(genesis, sdk.Genesis{}) {
 		return fmt.Errorf("unable to fetch genesis file from API at %s", algodImp.cfg.NetAddr)
@@ -418,13 +418,29 @@
 }
 
 // SyncError is used to indicate algod and conduit are not synchronized.
+// The retrievedRound is the round returned from an algod status call.
+// The expectedRound is the round conduit expected to have gotten back.
 type SyncError struct {
-	rnd      uint64
-	expected uint64
+	retrievedRound uint64
+	expectedRound  uint64
+	err            error
+}
+
+// NewSyncError creates a new SyncError.
+func NewSyncError(retrievedRound, expectedRound uint64, err error) *SyncError {
+	return &SyncError{
+		retrievedRound: retrievedRound,
+		expectedRound:  expectedRound,
+		err:            err,
+	}
 }
 
 func (e *SyncError) Error() string {
-	return fmt.Sprintf("wrong round returned from status for round: %d != %d", e.rnd, e.expected)
+	return fmt.Sprintf("wrong round returned from status for round: retrieved(%d) != expected(%d): %v", e.retrievedRound, e.expectedRound, e.err)
+}
+
+func (e *SyncError) Unwrap() error {
+	return e.err
 }
 
 func waitForRoundWithTimeout(ctx context.Context, l *logrus.Logger, c *algod.Client, rnd uint64, to time.Duration) (uint64, error) {
@@ -440,10 +456,8 @@
 		if rnd <= status.LastRound {
 			return status.LastRound, nil
 		}
-		return 0, &SyncError{
-			rnd:      status.LastRound,
-			expected: rnd,
-		}
+		// algod's timeout should not be reached because context.WithTimeout is used
+		return 0, NewSyncError(status.LastRound, rnd, fmt.Errorf("sync error, likely due to status after block timeout: %w", err))
 	}
 
 	// If there was a different error and the node is responsive, call status before returning a SyncError.
@@ -453,15 +467,12 @@
 		// If there was an error getting status, return the original error.
 		return 0, fmt.Errorf("unable to get status after block and status: %w", errors.Join(err, err2))
 	}
 	if status2.LastRound < rnd {
-		return 0, &SyncError{
-			rnd:      status.LastRound,
-			expected: rnd,
-		}
+		return 0, NewSyncError(status2.LastRound, rnd, fmt.Errorf("status2.LastRound mismatch: %w", err))
 	}
 
 	// This is probably a connection error, not a SyncError.
 	return 0, fmt.Errorf("unknown errors: StatusAfterBlock(%w), Status(%w)", err, err2)
 }
 
 func (algodImp *algodImporter) getBlockInner(rnd uint64) (data.BlockData, error) {
@@ -470,11 +481,10 @@
 
 	nodeRound, err := waitForRoundWithTimeout(algodImp.ctx, algodImp.logger, algodImp.aclient, rnd, waitForRoundTimeout)
 	if err != nil {
-		// If context has expired.
 		if algodImp.ctx.Err() != nil {
-			return blk, fmt.Errorf("GetBlock ctx error: %w", err)
+			return blk, fmt.Errorf("importer algod.GetBlock() ctx cancelled: %w", err)
 		}
-		algodImp.logger.Errorf(err.Error())
+		algodImp.logger.Errorf("importer algod.GetBlock() called waitForRoundWithTimeout: %v", err)
 		return data.BlockData{}, err
 	}
 	start := time.Now()
@@ -483,14 +493,14 @@
 	dt := time.Since(start)
 	getAlgodRawBlockTimeSeconds.Observe(dt.Seconds())
 	if err != nil {
-		algodImp.logger.Errorf("error getting block for round %d: %s", rnd, err.Error())
+		algodImp.logger.Errorf("importer algod.GetBlock() error getting block for round %d: %s", rnd, err.Error())
 		return data.BlockData{}, err
 	}
 	tmpBlk := new(models.BlockResponse)
 	err = msgpack.Decode(blockbytes, tmpBlk)
 	if err != nil {
 		return blk, err
 	}
 
 	blk.BlockHeader = tmpBlk.Block.BlockHeader
 	blk.Payset = tmpBlk.Block.Payset
@@ -503,9 +513,9 @@
 			delta, err = algodImp.getDelta(rnd)
 			if err != nil {
 				if nodeRound < rnd {
-					err = fmt.Errorf("ledger state delta not found: node round (%d) is behind required round (%d), ensure follower node has its sync round set to the required round: %w", nodeRound, rnd, err)
+					err = fmt.Errorf("importer algod.GetBlock() ledger state delta not found: node round (%d) is behind required round (%d), ensure follower node has its sync round set to the required round: %w", nodeRound, rnd, err)
 				} else {
-					err = fmt.Errorf("ledger state delta not found: node round (%d), required round (%d): verify follower node configuration and ensure follower node has its sync round set to the required round, re-deploying the follower node may be necessary: %w", nodeRound, rnd, err)
+					err = fmt.Errorf("importer algod.GetBlock() ledger state delta not found: node round (%d), required round (%d): verify follower node configuration and ensure follower node has its sync round set to the required round, re-deploying the follower node may be necessary: %w", nodeRound, rnd, err)
 				}
 				algodImp.logger.Error(err.Error())
 				return data.BlockData{}, err
@@ -523,10 +533,10 @@
 	if err != nil {
 		target := &SyncError{}
 		if errors.As(err, &target) {
-			algodImp.logger.Warnf("Sync error detected, attempting to set the sync round to recover the node: %s", err.Error())
+			algodImp.logger.Warnf("importer algod.GetBlock() sync error detected, attempting to set the sync round to recover the node: %s", err.Error())
 			_, _ = algodImp.aclient.SetSyncRound(rnd).Do(algodImp.ctx)
 		} else {
-			err = fmt.Errorf("error getting block for round %d, check node configuration: %s", rnd, err)
+			err = fmt.Errorf("importer algod.GetBlock() error getting block for round %d, check node configuration: %s", rnd, err)
 			algodImp.logger.Errorf(err.Error())
 		}
 		return data.BlockData{}, err

conduit/plugins/importers/algod/algod_importer_test.go

@@ -3,17 +3,18 @@ package algodimporter
 import (
 	"context"
 	"fmt"
-	"github.com/sirupsen/logrus"
-	"github.com/sirupsen/logrus/hooks/test"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"gopkg.in/yaml.v3"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/sirupsen/logrus"
+	"github.com/sirupsen/logrus/hooks/test"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
+
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/algod"
 	"github.com/algorand/go-algorand-sdk/v2/client/v2/common/models"
 	sdk "github.com/algorand/go-algorand-sdk/v2/types"
@@ -699,8 +700,8 @@ func TestGetBlockErrors(t *testing.T) {
 			blockAfterResponder: MakeBlockAfterResponder(models.NodeStatus{LastRound: 50}),
 			blockResponder:      BlockResponder,
 			deltaResponder:      MakeMsgpStatusResponder("get", "/v2/deltas/", http.StatusNotFound, ""),
-			err:                 fmt.Sprintf("wrong round returned from status for round: 50 != 200"),
-			logs:                []string{"wrong round returned from status for round: 50 != 200", "Sync error detected, attempting to set the sync round to recover the node"},
+			err:                 fmt.Sprintf("wrong round returned from status for round: retrieved(50) != expected(200)"),
+			logs:                []string{"wrong round returned from status for round: retrieved(50) != expected(200)", "sync error detected, attempting to set the sync round to recover the node"},
 		},
 		{
 			name:                "Cannot get delta (caught up)",
@@ -755,7 +756,8 @@ func TestGetBlockErrors(t *testing.T) {
 			// Make sure each of the expected log messages are present
 			for _, log := range tc.logs {
 				found := false
-				for _, entry := range hook.AllEntries() {
+				hookEntries := hook.AllEntries()
+				for _, entry := range hookEntries {
 					fmt.Println(strings.Contains(entry.Message, log))
 					found = found || strings.Contains(entry.Message, log)
 				}