Fix/audit2 (#79)

* fix batch info selection logic

* wait until all batch txs are processed before changing batch info

* handle msg update oracle config

* db actions out of iteration

* add some test cases

* add different bridge id test case

* handle multiple output events

Author: sh-cha

challenger/eventhandler/pending_events.go

@@ -7,6 +7,13 @@ import (
 	challengertypes "github.com/initia-labs/opinit-bots/challenger/types"
 )
 
+func (ch *ChallengeEventHandler) SetPendingEvent(event challengertypes.ChallengeEvent) {
+	ch.pendingEventsMu.Lock()
+	defer ch.pendingEventsMu.Unlock()
+
+	ch.pendingEvents[event.Id()] = event
+}
+
 func (ch *ChallengeEventHandler) SetPendingEvents(events []challengertypes.ChallengeEvent) {
 	if len(events) == 0 {
 		return

challenger/host/handler.go

@@ -29,24 +29,21 @@ func (h *Host) endBlockHandler(_ types.Context, args nodetypes.EndBlockArgs) err
 		return errors.Wrap(err, "failed to save pending events on child db")
 	}
 
-	// save all pending events to host db
-	// currently, only output event is considered as pending event
-	if len(h.outputPendingEventQueue) > 1 || (len(h.outputPendingEventQueue) == 1 && h.outputPendingEventQueue[0].Type() != challengertypes.EventTypeOutput) {
-		panic("must not happen, outputPendingEventQueue should have only one output event")
-	}
+	prevEvents := make([]challengertypes.ChallengeEvent, 0)
 
-	err = eventhandler.SavePendingEvents(h.stage, h.outputPendingEventQueue)
-	if err != nil {
-		return err
-	}
+	if len(h.outputPendingEventQueue) > 0 {
+		// save the last output event to host db
+		err = eventhandler.SavePendingEvent(h.stage, h.outputPendingEventQueue[len(h.outputPendingEventQueue)-1])
+		if err != nil {
+			return errors.Wrap(err, "failed to save pending event on host db")
+		}
 
-	prevEvents := make([]challengertypes.ChallengeEvent, 0)
-	for _, pendingEvent := range h.outputPendingEventQueue {
-		prevEvent, ok := h.eventHandler.GetPrevPendingEvent(pendingEvent)
+		prevEvent, ok := h.eventHandler.GetPrevPendingEvent(h.outputPendingEventQueue[0])
 		if ok {
 			prevEvents = append(prevEvents, prevEvent)
 		}
 	}
+
 	unprocessedEvents := h.eventHandler.GetUnprocessedPendingEvents(prevEvents)
 	pendingChallenges, processedEvents := h.eventHandler.CheckTimeout(args.Block.Header.Time, unprocessedEvents)
 	processedEvents = append(processedEvents, prevEvents...)
@@ -69,7 +66,10 @@ func (h *Host) endBlockHandler(_ types.Context, args nodetypes.EndBlockArgs) err
 
 	h.child.SetPendingEvents(h.eventQueue)
 	h.eventHandler.DeletePendingEvents(processedEvents)
-	h.eventHandler.SetPendingEvents(h.outputPendingEventQueue)
+	// save the last output event
+	if len(h.outputPendingEventQueue) > 0 {
+		h.eventHandler.SetPendingEvent(h.outputPendingEventQueue[len(h.outputPendingEventQueue)-1])
+	}
 	h.challenger.SendPendingChallenges(pendingChallenges)
 	return nil
 }

challenger/host/handler_test.go

@@ -214,6 +214,79 @@ func TestEndBlockHandler(t *testing.T) {
 				},
 			},
 			err: false,
+		}, {
+			name:       "update multiple output events",
+			child:      NewMockChild(nil, 1, 1),
+			challenger: NewMockChallenger(nil),
+			pendingEvents: []challengertypes.ChallengeEvent{
+				&challengertypes.Output{
+					EventType:     "Output",
+					L2BlockNumber: 2,
+					OutputIndex:   1,
+					OutputRoot:    []byte(""),
+					Time:          time.Unix(0, 10000).UTC(),
+					Timeout:       false,
+				},
+			},
+			eventQueue: []challengertypes.ChallengeEvent{},
+			outputPendingEventQueue: []challengertypes.ChallengeEvent{
+				&challengertypes.Output{
+					EventType:     "Output",
+					L2BlockNumber: 2,
+					OutputIndex:   2,
+					OutputRoot:    []byte(""),
+					Time:          time.Unix(0, 11000).UTC(),
+					Timeout:       false,
+				},
+				&challengertypes.Output{
+					EventType:     "Output",
+					L2BlockNumber: 2,
+					OutputIndex:   3,
+					OutputRoot:    []byte(""),
+					Time:          time.Unix(0, 11000).UTC(),
+					Timeout:       false,
+				},
+				&challengertypes.Output{
+					EventType:     "Output",
+					L2BlockNumber: 2,
+					OutputIndex:   4,
+					OutputRoot:    []byte(""),
+					Time:          time.Unix(0, 11000).UTC(),
+					Timeout:       false,
+				},
+			},
+			dbChanges: []types.KV{},
+			endBlockArgs: nodetypes.EndBlockArgs{
+				Block: cmtproto.Block{
+					Header: cmtproto.Header{
+						Height: 10,
+						Time:   time.Unix(0, 11000).UTC(),
+					},
+				},
+			},
+			expectedPendingEvents: []challengertypes.ChallengeEvent{
+				&challengertypes.Output{
+					EventType:     "Output",
+					L2BlockNumber: 2,
+					OutputIndex:   4,
+					OutputRoot:    []byte(""),
+					Time:          time.Unix(0, 11000).UTC(),
+					Timeout:       false,
+				},
+			},
+			expectedEventQueue: []challengertypes.ChallengeEvent{},
+			expectedChallenges: []challengertypes.Challenge{},
+			expectedDB: []types.KV{
+				{
+					Key:   []byte("test_host/synced_height"),
+					Value: []byte("10"),
+				},
+				{
+					Key:   append([]byte("test_host/pending_event/"), []byte{0x1, '/', 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x4}...),
+					Value: []byte(`{"event_type":"Output","l2_block_number":2,"output_index":4,"output_root":"","time":"1970-01-01T00:00:00.000011Z","timeout":false}`),
+				},
+			},
+			err: false,
 		},
 		{
 			name:       "output event timeout",

challenger/host/oracle.go

@@ -4,8 +4,13 @@ import (
 	"time"
 
 	comettypes "github.com/cometbft/cometbft/types"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
 
 	challengertypes "github.com/initia-labs/opinit-bots/challenger/types"
+	nodetypes "github.com/initia-labs/opinit-bots/node/types"
+	hostprovider "github.com/initia-labs/opinit-bots/provider/host"
+	"github.com/initia-labs/opinit-bots/types"
 )
 
 func (h *Host) oracleTxHandler(blockHeight int64, blockTime time.Time, oracleDataBytes comettypes.Tx) {
@@ -17,3 +22,21 @@ func (h *Host) oracleTxHandler(blockHeight int64, blockTime time.Time, oracleDat
 
 	h.eventQueue = append(h.eventQueue, oracle)
 }
+
+func (h *Host) updateOracleConfigHandler(ctx types.Context, args nodetypes.EventHandlerArgs) error {
+	bridgeId, oracleEnabled, err := hostprovider.ParseMsgUpdateOracleConfig(args.EventAttributes)
+	if err != nil {
+		return errors.Wrap(err, "failed to parse update oracle config event")
+	}
+	if bridgeId != h.BridgeId() {
+		return nil
+	}
+
+	ctx.Logger().Info("update oracle config",
+		zap.Uint64("bridge_id", bridgeId),
+		zap.Bool("oracle_enabled", oracleEnabled),
+	)
+
+	h.UpdateOracleEnabled(oracleEnabled)
+	return nil
+}

challenger/host/oracle_test.go

@@ -1,6 +1,7 @@
 package host
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -10,7 +11,9 @@ import (
 	"github.com/initia-labs/opinit-bots/node"
 	nodetypes "github.com/initia-labs/opinit-bots/node/types"
 	hostprovider "github.com/initia-labs/opinit-bots/provider/host"
+	"github.com/initia-labs/opinit-bots/types"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
 	"golang.org/x/crypto/sha3"
 )
 
@@ -74,3 +77,59 @@ func TestOracleTxHandler(t *testing.T) {
 	}
 	require.NoError(t, err)
 }
+
+func TestUpdateOracleConfigHandler(t *testing.T) {
+	db, err := db.NewMemDB()
+	require.NoError(t, err)
+	hostNode := node.NewTestNode(nodetypes.NodeConfig{}, db.WithPrefix([]byte("test_host")), nil, nil, nil, nil)
+
+	h := Host{
+		BaseHost: hostprovider.NewTestBaseHost(0, hostNode, ophosttypes.QueryBridgeResponse{
+			BridgeId: 1,
+			BridgeConfig: ophosttypes.BridgeConfig{
+				OracleEnabled: false,
+			},
+		}, nodetypes.NodeConfig{}, nil),
+	}
+
+	cases := []struct {
+		name                  string
+		bridgeId              uint64
+		oracleEnabled         bool
+		expectedOracleEnabled bool
+		err                   bool
+	}{
+		{
+			name:                  "oracle enabled",
+			bridgeId:              1,
+			oracleEnabled:         true,
+			expectedOracleEnabled: true,
+		},
+		{
+			name:                  "oracle disabled",
+			bridgeId:              1,
+			oracleEnabled:         false,
+			expectedOracleEnabled: false,
+		},
+		{
+			name:                  "another bridge id",
+			bridgeId:              2,
+			oracleEnabled:         true,
+			expectedOracleEnabled: false,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := h.updateOracleConfigHandler(types.NewContext(context.Background(), zap.NewNop(), ""), nodetypes.EventHandlerArgs{
+				EventAttributes: hostprovider.UpdateOracleConfigEvents(tc.bridgeId, tc.oracleEnabled),
+			})
+			if tc.err {
+				require.Error(t, err)
+			} else {
+				require.NoError(t, err)
+				require.Equal(t, tc.expectedOracleEnabled, h.OracleEnabled())
+			}
+		})
+	}
+}

executor/batchsubmitter/batch.go

@@ -21,7 +21,7 @@ import (
 // prepareBatch prepares the batch info for the given block height.
 // if there is more than one batch info in the queue and the start block height of the local batch info is greater than the l2 block number of the next batch info,
 // it finalizes the current batch and occurs a panic to restart the block process from the next batch info's l2 block number + 1.
-func (bs *BatchSubmitter) prepareBatch(blockHeight int64) error {
+func (bs *BatchSubmitter) prepareBatch(ctx types.Context, blockHeight int64) error {
 	localBatchInfo, err := GetLocalBatchInfo(bs.DB())
 	if err != nil {
 		return errors.Wrap(err, "failed to get local batch info")
@@ -32,6 +32,35 @@ func (bs *BatchSubmitter) prepareBatch(blockHeight int64) error {
 	if nextBatchInfo := bs.NextBatchInfo(); nextBatchInfo != nil &&
 		types.MustUint64ToInt64(nextBatchInfo.Output.L2BlockNumber) < bs.localBatchInfo.Start {
 		// if the next batch info is reached, finalize the current batch and update the batch info.
+
+		// wait until all processed msgs and pending txs are processed
+		timer := time.NewTicker(1 * time.Second)
+		defer timer.Stop()
+		for {
+			lenProcessedBatchMsgs, err := bs.da.LenProcessedBatchMsgs()
+			if err != nil {
+				return errors.Wrap(err, "failed to get processed msgs length")
+			}
+			lenPendingBatchTxs, err := bs.da.LenPendingBatchTxs()
+			if err != nil {
+				return errors.Wrap(err, "failed to get pending txs length")
+			}
+			if lenProcessedBatchMsgs == 0 && lenPendingBatchTxs == 0 {
+				break
+			}
+
+			ctx.Logger().Info("waiting for processed batch msgs and pending batch txs to be processed before changing batch info",
+				zap.Int("processed_batch_msgs", lenProcessedBatchMsgs),
+				zap.Int("pending_batch_txs", lenPendingBatchTxs),
+			)
+
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-timer.C:
+			}
+		}
+
 		if bs.batchWriter != nil {
 			err := bs.batchWriter.Close()
 			if err != nil {
@@ -56,15 +85,6 @@ func (bs *BatchSubmitter) prepareBatch(blockHeight int64) error {
 			return errors.Wrap(err, "failed to set synced height")
 		}
 
-		err = node.DeleteProcessedMsgs(bs.da.DB())
-		if err != nil {
-			return errors.Wrap(err, "failed to delete processed msgs")
-		}
-		err = node.DeletePendingTxs(bs.da.DB())
-		if err != nil {
-			return errors.Wrap(err, "failed to delete pending txs")
-		}
-
 		// error will restart block process from nextBatchInfo.Output.L2BlockNumber + 1
 		panic(fmt.Errorf("batch info updated: reset from %d", nextBatchInfo.Output.L2BlockNumber))
 	}

executor/batchsubmitter/batch_submitter.go

@@ -107,11 +107,14 @@ func (bs *BatchSubmitter) Initialize(ctx types.Context, syncedHeight int64, host
 	if len(bs.batchInfos) == 0 {
 		return errors.New("no batch info")
 	}
-	for _, batchInfo := range bs.batchInfos {
-		if len(bs.batchInfos) == 1 || types.MustUint64ToInt64(batchInfo.Output.L2BlockNumber+1) >= bs.node.GetHeight() {
+
+	for i, batchInfo := range bs.batchInfos {
+		if batchInfo.Output.L2BlockNumber != 0 && types.MustUint64ToInt64(batchInfo.Output.L2BlockNumber+1) > bs.node.GetHeight() {
 			break
+		} else if i > 0 {
+			// dequeue the previous batch info
+			bs.DequeueBatchInfo()
 		}
-		bs.DequeueBatchInfo()
 	}
 
 	fileFlag := os.O_CREATE | os.O_RDWR | os.O_APPEND

executor/batchsubmitter/batch_test.go

@@ -191,6 +191,7 @@ func TestPrepareBatch(t *testing.T) {
 				node:        batchNode,
 				batchInfoMu: &sync.Mutex{},
 				batchInfos:  tc.batchInfoQueue,
+				da:          NewMockDA(nil, nil, 1, ""),
 			}
 
 			err = SaveLocalBatchInfo(batchDB, tc.existingLocalBatchInfo)
@@ -204,9 +205,10 @@ func TestPrepareBatch(t *testing.T) {
 			require.NoError(t, err)
 			defer batchSubmitter.batchWriter.Close()
 
+			ctx := types.NewContext(context.Background(), zap.NewNop(), "")
 			if tc.panic {
 				require.Panics(t, func() {
-					batchSubmitter.prepareBatch(tc.blockHeight) //nolint:errcheck
+					batchSubmitter.prepareBatch(ctx, tc.blockHeight) //nolint:errcheck
 				})
 				for _, expectedKV := range tc.expectedChanges {
 					value, err := baseDB.Get(expectedKV.Key)
@@ -218,7 +220,7 @@ func TestPrepareBatch(t *testing.T) {
 				require.NoError(t, err)
 				require.Equal(t, int64(0), fileSize)
 			} else {
-				err := batchSubmitter.prepareBatch(tc.blockHeight)
+				err := batchSubmitter.prepareBatch(ctx, tc.blockHeight)
 				if tc.err {
 					require.Error(t, err)
 				} else {

executor/batchsubmitter/common_test.go

@@ -104,6 +104,14 @@ func (m *mockDA) BroadcastProcessedMsgs(msgs ...btypes.ProcessedMsgs) {
 	m.processedMsgs = append(m.processedMsgs, msgs...)
 }
 
+func (m mockDA) LenProcessedBatchMsgs() (int, error) {
+	return len(m.processedMsgs), nil
+}
+
+func (m mockDA) LenPendingBatchTxs() (int, error) {
+	return 0, nil
+}
+
 var _ executortypes.DANode = (*mockDA)(nil)
 
 func logCapturer() (*zap.Logger, *observer.ObservedLogs) {

executor/batchsubmitter/handler.go

@@ -18,7 +18,7 @@ func (bs *BatchSubmitter) rawBlockHandler(ctx types.Context, args nodetypes.RawB
 	bs.processedMsgs = bs.processedMsgs[:0]
 	bs.stage.Reset()
 
-	err := bs.prepareBatch(args.BlockHeight)
+	err := bs.prepareBatch(ctx, args.BlockHeight)
 	if err != nil {
 		return errors.Wrap(err, "failed to prepare batch")
 	}