Feat/batch (#15)

* change batch spec adding header bytes to batch data

* bump opinit version

* delete oracle option

* bug fix marshal unmarshal batch data

Author: sh-cha

executor/batch/batch.go

@@ -1,19 +1,17 @@
 package batch
 
 import (
+	"compress/gzip"
 	"context"
 	"errors"
-	"io"
 	"os"
 	"sync"
-	"time"
 
 	"go.uber.org/zap"
 
 	opchildtypes "github.com/initia-labs/OPinit/x/opchild/types"
 	ophosttypes "github.com/initia-labs/OPinit/x/ophost/types"
 
-	dbtypes "github.com/initia-labs/opinit-bots/db/types"
 	executortypes "github.com/initia-labs/opinit-bots/executor/types"
 	"github.com/initia-labs/opinit-bots/node"
 	btypes "github.com/initia-labs/opinit-bots/node/broadcaster/types"
@@ -26,14 +24,6 @@ type hostNode interface {
 	QueryBatchInfos(context.Context, uint64) (*ophosttypes.QueryBatchInfosResponse, error)
 }
 
-type compressionFunc interface {
-	Write([]byte) (int, error)
-	Reset(io.Writer)
-	Close() error
-}
-
-var SubmissionKey = []byte("submission_time")
-
 type BatchSubmitter struct {
 	version uint8
 
@@ -50,19 +40,17 @@ type BatchSubmitter struct {
 
 	opchildQueryClient opchildtypes.QueryClient
 
-	batchInfoMu *sync.Mutex
-	batchInfos  []ophosttypes.BatchInfoWithOutput
-	batchWriter compressionFunc
-	batchFile   *os.File
-	batchHeader *executortypes.BatchHeader
+	batchInfoMu    *sync.Mutex
+	batchInfos     []ophosttypes.BatchInfoWithOutput
+	batchWriter    *gzip.Writer
+	batchFile      *os.File
+	localBatchInfo *executortypes.LocalBatchInfo
 
 	processedMsgs []btypes.ProcessedMsgs
 
 	chainID  string
 	homePath string
 
-	lastSubmissionTime time.Time
-
 	// status info
 	LastBatchEndBlockNumber uint64
 }
@@ -98,7 +86,8 @@ func NewBatchSubmitterV0(
 
 		opchildQueryClient: opchildtypes.NewQueryClient(node.GetRPCClient()),
 
-		batchInfoMu: &sync.Mutex{},
+		batchInfoMu:    &sync.Mutex{},
+		localBatchInfo: &executortypes.LocalBatchInfo{},
 
 		processedMsgs: make([]btypes.ProcessedMsgs, 0),
 		homePath:      homePath,
@@ -131,17 +120,26 @@ func (bs *BatchSubmitter) Initialize(ctx context.Context, startHeight uint64, ho
 	}
 
 	fileFlag := os.O_CREATE | os.O_RDWR
-	// if the node has already processed blocks, append to the file
-	if !bs.node.HeightInitialized() {
+	if bs.node.HeightInitialized() {
+		bs.localBatchInfo.Start = bs.node.GetHeight()
+		bs.localBatchInfo.End = 0
+		bs.localBatchInfo.BatchFileSize = 0
+
+		err = bs.saveLocalBatchInfo()
+		if err != nil {
+			return err
+		}
+	} else {
+		// if the node has already processed blocks, append to the file
 		fileFlag |= os.O_APPEND
 	}
 
 	bs.batchFile, err = os.OpenFile(bs.homePath+"/batch", fileFlag, 0666)
 	if err != nil {
 		return err
 	}
-
-	err = bs.LoadSubmissionInfo()
+	// linux command gzip use level 6 as default
+	bs.batchWriter, err = gzip.NewWriterLevel(bs.batchFile, 6)
 	if err != nil {
 		return err
 	}
@@ -168,25 +166,6 @@ func (bs *BatchSubmitter) SetBridgeInfo(bridgeInfo opchildtypes.BridgeInfo) {
 	bs.bridgeInfo = bridgeInfo
 }
 
-func (bs *BatchSubmitter) LoadSubmissionInfo() error {
-	val, err := bs.db.Get(SubmissionKey)
-	if err != nil {
-		if err == dbtypes.ErrNotFound {
-			return nil
-		}
-		return err
-	}
-	bs.lastSubmissionTime = time.Unix(0, dbtypes.ToInt64(val))
-	return nil
-}
-
-func (bs *BatchSubmitter) SubmissionInfoToRawKV(timestamp int64) types.RawKV {
-	return types.RawKV{
-		Key:   bs.db.PrefixedKey(SubmissionKey),
-		Value: dbtypes.FromInt64(timestamp),
-	}
-}
-
 func (bs *BatchSubmitter) ChainID() string {
 	return bs.chainID
 }

executor/batch/db.go

@@ -0,0 +1,40 @@
+package batch
+
+import (
+	"encoding/json"
+
+	dbtypes "github.com/initia-labs/opinit-bots/db/types"
+	"github.com/initia-labs/opinit-bots/types"
+)
+
+var LocalBatchInfoKey = []byte("local_batch_info")
+
+func (bs *BatchSubmitter) loadLocalBatchInfo() error {
+	val, err := bs.db.Get(LocalBatchInfoKey)
+	if err != nil {
+		if err == dbtypes.ErrNotFound {
+			return nil
+		}
+		return err
+	}
+	return json.Unmarshal(val, &bs.localBatchInfo)
+}
+
+func (bs *BatchSubmitter) localBatchInfoToRawKV() (types.RawKV, error) {
+	value, err := json.Marshal(bs.localBatchInfo)
+	if err != nil {
+		return types.RawKV{}, err
+	}
+	return types.RawKV{
+		Key:   bs.db.PrefixedKey(LocalBatchInfoKey),
+		Value: value,
+	}, nil
+}
+
+func (bs *BatchSubmitter) saveLocalBatchInfo() error {
+	value, err := json.Marshal(bs.localBatchInfo)
+	if err != nil {
+		return err
+	}
+	return bs.db.Set(LocalBatchInfoKey, value)
+}