Version store

app/app.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"sync"
 
 	"github.com/crypto-org-chain/cronos/x/cronos"
 	"github.com/crypto-org-chain/cronos/x/cronos/middleware"
@@ -22,12 +23,14 @@ import (
 	dbm "github.com/tendermint/tm-db"
 
 	"github.com/cosmos/cosmos-sdk/baseapp"
+	"github.com/cosmos/cosmos-sdk/client/flags"
 	"github.com/cosmos/cosmos-sdk/client/grpc/tmservice"
 	"github.com/cosmos/cosmos-sdk/codec"
 	"github.com/cosmos/cosmos-sdk/server/api"
 	"github.com/cosmos/cosmos-sdk/server/config"
 	servertypes "github.com/cosmos/cosmos-sdk/server/types"
 	"github.com/cosmos/cosmos-sdk/store/streaming"
+	"github.com/cosmos/cosmos-sdk/store/streaming/file"
 	storetypes "github.com/cosmos/cosmos-sdk/store/types"
 	"github.com/cosmos/cosmos-sdk/testutil/testdata"
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -122,7 +125,8 @@ import (
 	gravitytypes "github.com/peggyjv/gravity-bridge/module/v2/x/gravity/types"
 
 	// this line is used by starport scaffolding # stargate/app/moduleImport
-
+	cronosappclient "github.com/crypto-org-chain/cronos/client"
+	cronosfile "github.com/crypto-org-chain/cronos/client/file"
 	"github.com/crypto-org-chain/cronos/versiondb"
 	"github.com/crypto-org-chain/cronos/versiondb/tmdb"
 	cronosclient "github.com/crypto-org-chain/cronos/x/cronos/client"
@@ -148,6 +152,8 @@ const (
 	//
 	// NOTE: In the SDK, the default value is 255.
 	AddrLen = 20
+
+	FileStreamerDirectory = "file_streamer"
 )
 
 // this line is used by starport scaffolding # stargate/wasm/app/enabledProposals
@@ -353,12 +359,33 @@ func New(
 		os.Exit(1)
 	}
 
-	// configure state listening capabilities using AppOptions
-	// we are doing nothing with the returned streamingServices and waitGroup in this case
 	streamers := cast.ToStringSlice(appOpts.Get("store.streamers"))
 	for _, streamerName := range streamers {
+		if streamerName == "file" {
+			streamingDir := filepath.Join(cast.ToString(appOpts.Get(flags.FlagHome)), "data", FileStreamerDirectory)
+			if err := os.MkdirAll(streamingDir, os.ModePerm); err != nil {
+				panic(err)
+			}
+
+			// default to exposing all
+			exposeStoreKeys := make([]storetypes.StoreKey, 0, len(keys))
+			for _, storeKey := range keys {
+				exposeStoreKeys = append(exposeStoreKeys, storeKey)
+			}
+			service, err := file.NewStreamingService(streamingDir, "", exposeStoreKeys, appCodec, false, true, true)
+			if err != nil {
+				panic(err)
+			}
+			bApp.SetStreamingService(service)
+
+			wg := new(sync.WaitGroup)
+			if err := service.Stream(wg); err != nil {
+				panic(err)
+			}
+		}
 		if streamerName == "versiondb" {
-			dataDir := filepath.Join(homePath, "data", "versiondb")
+			rootDir := cast.ToString(appOpts.Get(flags.FlagHome))
+			dataDir := filepath.Join(rootDir, "data", "versiondb")
 			if err := os.MkdirAll(dataDir, os.ModePerm); err != nil {
 				panic(err)
 			}
@@ -376,7 +403,15 @@ func New(
 				panic(err)
 			}
 			versionDB := tmdb.NewStore(plainDB, historyDB, changesetDB)
-
+			isGrpcOnly := cast.ToBool(appOpts.Get(cronosappclient.FlagIsGrpcOnly))
+			remoteGrpcUrl := cast.ToString(appOpts.Get(cronosappclient.FlagRemoteGrpcUrl))
+			isLocal := cast.ToBool(appOpts.Get(cronosappclient.FlagIsLocal))
+			remoteUrl := cast.ToString(appOpts.Get(cronosappclient.FlagRemoteUrl))
+			remoteWsUrl := cast.ToString(appOpts.Get(cronosappclient.FlagRemoteWsUrl))
+			concurrency := cast.ToInt(appOpts.Get(cronosappclient.FlagConcurrency))
+			if isGrpcOnly {
+				cronosfile.Sync(versionDB, remoteGrpcUrl, remoteUrl, remoteWsUrl, rootDir, isLocal, concurrency)
+			}
 			// default to exposing all
 			exposeStoreKeys := make([]storetypes.StoreKey, 0, len(keys))
 			for _, storeKey := range keys {
@@ -388,7 +423,6 @@ func New(
 			qms.MountTransientStores(tkeys)
 			qms.MountMemoryStores(memKeys)
 			bApp.SetQueryMultiStore(qms)
-			break
 		}
 	}
 

client/decoder.go

@@ -0,0 +1,30 @@
+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) {
+	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)
+		if err := proto.Unmarshal(data[offset:offset+int(size)], pair); err != nil {
+			return nil, err
+		}
+		pairs = append(pairs, *pair)
+		offset += int(size)
+	}
+	return
+}

client/file/sync.go

@@ -0,0 +1,210 @@
+package file
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/crypto-org-chain/cronos/client"
+	"github.com/crypto-org-chain/cronos/versiondb/tmdb"
+	tmjson "github.com/tendermint/tendermint/libs/json"
+	tmtypes "github.com/tendermint/tendermint/types"
+)
+
+// Simplify block height for header
+type Header struct {
+	Height int64 `json:"height,omitempty"`
+}
+
+type Block struct {
+	Header Header `json:"header"`
+}
+
+type GetLatestBlockResponse struct {
+	Block *Block `json:"block,omitempty"`
+}
+
+func Sync(versionDB *tmdb.Store, remoteGrpcUrl, remoteUrl, remoteWsUrl, rootDir string, isLocal bool, concurrency int) {
+
+	const defaultMaxRetry = 50
+	latestVersion, err := versionDB.GetLatestVersion()
+	fmt.Printf("mm-latestVersion: %+v\n", latestVersion)
+	if err != nil {
+		panic(err)
+	}
+	startBlockNum := latestVersion
+	if startBlockNum < 0 {
+		startBlockNum = 0
+	}
+	nextBlockNum := int(startBlockNum) + 1
+	maxBlockNum := -1
+	for i := 0; i < defaultMaxRetry; i++ {
+		if i > 0 {
+			time.Sleep(time.Second)
+		}
+		resp, err := http.Get(fmt.Sprintf("%s/cosmos/base/tendermint/v1beta1/blocks/latest", remoteGrpcUrl))
+		if err != nil {
+			fmt.Printf("error making http request: %s\n", err)
+			continue
+		}
+
+		var bz []byte
+		if result := func() bool {
+			defer resp.Body.Close()
+			bz, err = ioutil.ReadAll(resp.Body)
+			if err != nil {
+				return false
+			}
+			if resp.StatusCode != http.StatusOK {
+				return false
+			}
+			return true
+		}(); !result {
+			continue
+		}
+
+		res := new(GetLatestBlockResponse)
+		err = tmjson.Unmarshal(bz, res)
+		if err != nil {
+			fmt.Printf("mm-read-res-err: %+v\n", err)
+			continue
+		}
+		maxBlockNum = int(res.Block.Header.Height)
+		fmt.Printf("mm-maxBlockNum: %d\n", maxBlockNum)
+		break
+	}
+	if maxBlockNum < 0 {
+		panic(fmt.Sprintf("max retries %d reached", defaultMaxRetry))
+	}
+
+	interval := time.Second
+	directory := filepath.Join(rootDir, "data", "file_streamer")
+	// streamer write the file blk by blk with concurrency 1
+	streamer := NewBlockFileWatcher(1, maxBlockNum, func(blockNum int) string {
+		return GetLocalDataFileName(directory, blockNum)
+	}, nil, true)
+	streamer.Start(nextBlockNum, interval)
+	go func() {
+		chData, chErr := streamer.SubscribeData(), streamer.SubscribeError()
+		for {
+			select {
+			case data := <-chData:
+				pairs, err := client.DecodeData(data.Data)
+				fmt.Printf("mm-pairs: %+v, %+v\n", len(pairs), 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")
+				panic(err)
+			}
+		}
+	}()
+
+	synchronizer := NewBlockFileWatcher(
+		concurrency,
+		maxBlockNum,
+		func(blockNum int) string {
+			return fmt.Sprintf("%s/%s", remoteUrl, DataFileName(blockNum))
+		},
+		func(blockNum int) bool {
+			path := GetLocalDataFileName(directory, blockNum)
+			f, err := os.Open(path)
+			if err == nil {
+				defer func() {
+					_ = f.Close()
+				}()
+				// valid 1st 8 bytes for downloaded file
+				var bytes [8]byte
+				if _, err = io.ReadFull(f, bytes[:]); err == nil {
+					size := binary.BigEndian.Uint64(bytes[:])
+					if info, err := f.Stat(); err == nil && size+uint64(8) == uint64(info.Size()) {
+						return false
+					}
+				}
+			}
+			return true
+		},
+		isLocal,
+	)
+	synchronizer.Start(nextBlockNum, interval)
+	go func() {
+		// max retry for temporary io error
+		maxRetry := concurrency * 2
+		retry := 0
+		chData, chErr := synchronizer.SubscribeData(), synchronizer.SubscribeError()
+		for {
+			select {
+			case data := <-chData:
+				file := GetLocalDataFileName(directory, data.BlockNum)
+				fmt.Printf("mm-data.BlockNum: %+v\n", data.BlockNum)
+				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)
+					}
+				}
+				data.ChResult <- err
+			case err := <-chErr:
+				retry++
+				fmt.Println("mm-retry", retry)
+				if retry == maxRetry {
+					// data corrupt
+					fmt.Println("mm-data-corrupt-panic")
+					panic(err)
+				}
+			}
+		}
+	}()
+
+	go func() {
+		for i := 0; i < defaultMaxRetry; i++ {
+			if i > 0 {
+				time.Sleep(time.Second)
+			}
+			wsClient := client.NewWebsocketClient(remoteWsUrl)
+			chResult, err := wsClient.Subscribe()
+			if err != nil {
+				fmt.Printf("mm-subscribed[%+v]: %+v\n", i, err)
+				continue
+			}
+			fmt.Println("subscribing")
+			err = wsClient.Send("subscribe", []string{
+				"tm.event='NewBlockHeader'",
+			})
+			if err != nil {
+				fmt.Printf("mm-subscribed: %+v\n", err)
+				continue
+			}
+			i = 0
+			fmt.Println("subscribed ws")
+			for res := range chResult {
+				if res == nil || res.Data == nil {
+					continue
+				}
+				data, ok := res.Data.(tmtypes.EventDataNewBlockHeader)
+				if !ok {
+					continue
+				}
+				blockNum := int(data.Header.Height)
+				fmt.Printf("mm-set-max-blk: %+v\n", blockNum)
+				synchronizer.SetMaxBlockNum(blockNum)
+			}
+		}
+		panic(fmt.Sprintf("max retries %d reached", defaultMaxRetry))
+	}()
+}