fail early if incomplete

allow retry if incomplete

client/decoder.go

@@ -1,13 +1,22 @@
 package client
 
 import (
+	"fmt"
+
 	"github.com/cosmos/cosmos-sdk/store/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/gogo/protobuf/proto"
 )
 
 func DecodeData(data []byte) (pairs []types.StoreKVPair, err error) {
-	offset := 8
-	for offset < len(data) {
+	const prefixLen = 8
+	offset := prefixLen
+	dataSize := sdk.BigEndianToUint64(data[:offset])
+	size := len(data)
+	if int(dataSize)+prefixLen != size {
+		return nil, fmt.Errorf("incomplete file: %v vs %v", dataSize, size)
+	}
+	for offset < size {
 		size, n := proto.DecodeVarint(data[offset:])
 		offset += n
 		pair := new(types.StoreKVPair)

client/file/sync.go

@@ -96,12 +96,13 @@ func Sync(versionDB *tmdb.Store, remoteGrpcUrl, remoteUrl, remoteWsUrl, rootDir
 			case data := <-chData:
 				pairs, err := client.DecodeData(data.Data)
 				fmt.Printf("mm-pairs: %+v, %+v\n", len(pairs), err)
-				if err == nil {
-					if err = versionDB.PutAtVersion(int64(data.BlockNum), pairs); err != nil {
-						fmt.Println("mm-put-at-version-panic")
-						panic(err)
-					}
+				if err != nil {
+					fmt.Println("invalid decode")
+				} else if err = versionDB.PutAtVersion(int64(data.BlockNum), pairs); err != nil {
+					fmt.Println("mm-put-at-version-panic")
+					panic(err)
 				}
+				data.ChResult <- err
 			case err := <-chErr:
 				// fail read
 				fmt.Println("mm-fail-read-panic")
@@ -150,12 +151,14 @@ func Sync(versionDB *tmdb.Store, remoteGrpcUrl, remoteUrl, remoteWsUrl, rootDir
 				if err := os.WriteFile(file, data.Data, 0600); err != nil {
 					fmt.Println("mm-WriteFile-panic")
 					panic(err)
+				} else {
+					retry = 0
+					fmt.Println("mm-reset-retry")
+					if data.BlockNum > maxBlockNum {
+						streamer.SetMaxBlockNum(data.BlockNum)
+					}
 				}
-				retry = 0
-				fmt.Println("mm-reset-retry")
-				if data.BlockNum > maxBlockNum {
-					streamer.SetMaxBlockNum(data.BlockNum)
-				}
+				data.ChResult <- err
 			case err := <-chErr:
 				retry++
 				fmt.Println("mm-retry", retry)
@@ -200,7 +203,6 @@ func Sync(versionDB *tmdb.Store, remoteGrpcUrl, remoteUrl, remoteWsUrl, rootDir
 				blockNum := int(data.Header.Height)
 				fmt.Printf("mm-set-max-blk: %+v\n", blockNum)
 				synchronizer.SetMaxBlockNum(blockNum)
-				streamer.SetMaxBlockNum(blockNum)
 			}
 		}
 		panic(fmt.Sprintf("max retries %d reached", defaultMaxRetry))

client/file/watcher.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path/filepath"
 	"sync"
-	"sync/atomic"
 	"time"
 )
 
@@ -56,18 +55,20 @@ func (d *httpFileDownloader) GetData(path string) ([]byte, error) {
 type BlockData struct {
 	BlockNum int
 	Data     []byte
+	ChResult chan<- error
 }
 
 type BlockFileWatcher struct {
 	concurrency   int
-	maxBlockNum   int64
+	maxBlockNum   int
 	getFilePath   func(blockNum int) string
 	onBeforeFetch func(blockNum int) bool
 	downloader    fileDownloader
 	chData        chan *BlockData
 	chError       chan error
 	chDone        chan bool
 	startLock     *sync.Mutex
+	maxBlockLock  *sync.RWMutex
 }
 
 func NewBlockFileWatcher(
@@ -79,12 +80,13 @@ func NewBlockFileWatcher(
 ) *BlockFileWatcher {
 	w := &BlockFileWatcher{
 		concurrency:   concurrency,
-		maxBlockNum:   int64(maxBlockNum),
+		maxBlockNum:   maxBlockNum,
 		getFilePath:   getFilePath,
 		onBeforeFetch: onBeforeFetch,
 		chData:        make(chan *BlockData),
 		chError:       make(chan error),
 		startLock:     new(sync.Mutex),
+		maxBlockLock:  new(sync.RWMutex),
 	}
 	if isLocal {
 		w.downloader = new(localFileDownloader)
@@ -111,7 +113,12 @@ func (w *BlockFileWatcher) SubscribeError() <-chan error {
 }
 
 func (w *BlockFileWatcher) SetMaxBlockNum(num int) {
-	atomic.StoreInt64(&w.maxBlockNum, int64(num))
+	w.maxBlockLock.Lock()
+	defer w.maxBlockLock.Unlock()
+	// avoid dup job when set max to smaller one while locking
+	if num > w.maxBlockNum {
+		w.maxBlockNum = num
+	}
 }
 
 func (w *BlockFileWatcher) fetch(blockNum int) error {
@@ -131,11 +138,14 @@ func (w *BlockFileWatcher) fetch(blockNum int) error {
 		}
 		return err
 	}
+
+	chResult := make(chan error)
 	w.chData <- &BlockData{
 		BlockNum: blockNum,
 		Data:     data,
+		ChResult: chResult,
 	}
-	return nil
+	return <-chResult
 }
 
 func (w *BlockFileWatcher) Start(
@@ -158,7 +168,9 @@ func (w *BlockFileWatcher) Start(
 			default:
 				wg := new(sync.WaitGroup)
 				currentBlockNum := blockNum
-				maxBlockNum := int(atomic.LoadInt64(&w.maxBlockNum))
+				w.maxBlockLock.RLock()
+				maxBlockNum := w.maxBlockNum
+				w.maxBlockLock.RUnlock()
 				concurrency := w.concurrency
 				if diff := maxBlockNum - currentBlockNum; diff < concurrency {
 					if diff <= 0 {