Emit confirmed info event per confirmed block (#143)

config/config.go

@@ -31,12 +31,13 @@ var (
 
 // Config is the main config for the fpd cli command
 type Config struct {
-	LogLevel       string         `long:"loglevel" description:"Logging level for all subsystems" choice:"trace" choice:"debug" choice:"info" choice:"warn" choice:"error" choice:"fatal"`
-	BitcoinNetwork string         `long:"bitcoinnetwork" description:"Bitcoin network to run on" choice:"mainnet" choice:"regtest" choice:"testnet" choice:"simnet" choice:"signet"`
-	BTCConfig      *BTCConfig     `group:"btcconfig" namespace:"btcconfig"`
-	DatabaseConfig *DBConfig      `group:"dbconfig" namespace:"dbconfig"`
-	QueueConfig    *QueueConfig   `group:"queueconfig" namespace:"queueconfig"`
-	MetricsConfig  *MetricsConfig `group:"metricsconfig" namespace:"metricsconfig"`
+	LogLevel          string         `long:"loglevel" description:"Logging level for all subsystems" choice:"trace" choice:"debug" choice:"info" choice:"warn" choice:"error" choice:"fatal"`
+	BitcoinNetwork    string         `long:"bitcoinnetwork" description:"Bitcoin network to run on" choice:"mainnet" choice:"regtest" choice:"testnet" choice:"simnet" choice:"signet"`
+	ExtraEventEnabled bool           `long:"extraeventenabled" description:"Whether emitting non-default events is allowed"`
+	BTCConfig         *BTCConfig     `group:"btcconfig" namespace:"btcconfig"`
+	DatabaseConfig    *DBConfig      `group:"dbconfig" namespace:"dbconfig"`
+	QueueConfig       *QueueConfig   `group:"queueconfig" namespace:"queueconfig"`
+	MetricsConfig     *MetricsConfig `group:"metricsconfig" namespace:"metricsconfig"`
 
 	BTCNetParams chaincfg.Params
 }

consumer/event_consumer.go

@@ -10,5 +10,6 @@ type EventConsumer interface {
 	PushUnbondingEvent(ev *client.UnbondingStakingEvent) error
 	PushWithdrawEvent(ev *client.WithdrawStakingEvent) error
 	PushBtcInfoEvent(ev *client.BtcInfoEvent) error
+	PushConfirmedInfoEvent(ev *client.ConfirmedInfoEvent) error
 	Stop() error
 }

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/avast/retry-go/v4 v4.5.1
 	github.com/babylonchain/babylon v0.9.0-rc.1
 	github.com/babylonchain/networks/parameters v0.2.1
-	github.com/babylonchain/staking-queue-client v0.2.1
+	github.com/babylonchain/staking-queue-client v0.4.0
 	github.com/btcsuite/btcd v0.24.0
 	github.com/btcsuite/btcd/btcec/v2 v2.3.2
 	github.com/btcsuite/btcd/btcutil v1.1.5

go.sum

@@ -285,8 +285,8 @@ github.com/babylonchain/babylon v0.9.0-rc.1 h1:mZYKQVHVKFUA2xaEAzJloB1kyePHvZECJ
 github.com/babylonchain/babylon v0.9.0-rc.1/go.mod h1:YFALTW+Kp/b5jSDoA7Z70RggJjAedlmQTrpdeU8c3hY=
 github.com/babylonchain/networks/parameters v0.2.1 h1:OKHiCnwL/UdVN17cMwCrHz/bAjO/USauLiPyNlnVl6E=
 github.com/babylonchain/networks/parameters v0.2.1/go.mod h1:nejhvrL7Iwh5Vunvkg7pnomQZlHnyNzOY9lQaDp6tOA=
-github.com/babylonchain/staking-queue-client v0.2.1 h1:FKbxUUOoCydAsUoj9XQMIQT1S79ThX9p7vVc5yjWm8c=
-github.com/babylonchain/staking-queue-client v0.2.1/go.mod h1:mEgA6N3SnwFwGEOsUYr/HdjpKa13Wck08MS7VY/Icvo=
+github.com/babylonchain/staking-queue-client v0.4.0 h1:uKClZqFgksUo9tfVwIVCMtDjJk8YzOBsx7IwlJEP3fw=
+github.com/babylonchain/staking-queue-client v0.4.0/go.mod h1:mEgA6N3SnwFwGEOsUYr/HdjpKa13Wck08MS7VY/Icvo=
 github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=

indexer/indexer.go

@@ -383,6 +383,18 @@ func (si *StakingIndexer) HandleConfirmedBlock(b *types.IndexedBlock) error {
 		return fmt.Errorf("failed to save the last processed height: %w", err)
 	}
 
+	if si.cfg.ExtraEventEnabled {
+		// emit ConfirmedInfoEvent to send the confirmed height and tvl
+		confirmedTvl, err := si.is.GetConfirmedTvl()
+		if err != nil {
+			return fmt.Errorf("failed to get the confirmed tvl: %w", err)
+		}
+		confirmedInfoEvent := queuecli.NewConfirmedInfoEvent(uint64(b.Height), confirmedTvl)
+		if err := si.consumer.PushConfirmedInfoEvent(&confirmedInfoEvent); err != nil {
+			return fmt.Errorf("failed to push the confirmed info event: %w", err)
+		}
+	}
+
 	// record metrics
 	lastProcessedBtcHeight.Set(float64(b.Height))
 

itest/e2e_test.go

@@ -99,11 +99,14 @@ func TestStakingLifeCycle(t *testing.T) {
 	stakingTxHash := stakingTx.TxHash()
 	tm.SendTxWithNConfirmations(t, stakingTx, int(k))
 
+	tm.CheckConfirmedInfoEvent(t, 100, 0)
+
 	// check that the staking tx is already stored
 	_ = tm.WaitForStakingTxStored(t, stakingTxHash)
 
 	// check the staking event is received by the queue
 	tm.CheckNextStakingEvent(t, stakingTxHash)
+	tm.CheckConfirmedInfoEvent(t, 102, uint64(testStakingData.StakingAmount))
 
 	// wait for the staking tx expires
 	if uint64(testStakingData.StakingTime) > k {

itest/test_manager.go

@@ -39,23 +39,24 @@ import (
 )
 
 type TestManager struct {
-	Config             *config.Config
-	Db                 kvdb.Backend
-	Si                 *indexer.StakingIndexer
-	BS                 *btcscanner.BtcPoller
-	WalletPrivKey      *btcec.PrivateKey
-	serverStopper      *signal.Interceptor
-	wg                 *sync.WaitGroup
-	BitcoindHandler    *BitcoindTestHandler
-	WalletClient       *rpcclient.Client
-	MinerAddr          btcutil.Address
-	DirPath            string
-	QueueConsumer      *queuemngr.QueueManager
-	StakingEventChan   <-chan queuecli.QueueMessage
-	UnbondingEventChan <-chan queuecli.QueueMessage
-	WithdrawEventChan  <-chan queuecli.QueueMessage
-	BtcInfoEventChan   <-chan queuecli.QueueMessage
-	VersionedParams    *parser.ParsedGlobalParams
+	Config                 *config.Config
+	Db                     kvdb.Backend
+	Si                     *indexer.StakingIndexer
+	BS                     *btcscanner.BtcPoller
+	WalletPrivKey          *btcec.PrivateKey
+	serverStopper          *signal.Interceptor
+	wg                     *sync.WaitGroup
+	BitcoindHandler        *BitcoindTestHandler
+	WalletClient           *rpcclient.Client
+	MinerAddr              btcutil.Address
+	DirPath                string
+	QueueConsumer          *queuemngr.QueueManager
+	StakingEventChan       <-chan queuecli.QueueMessage
+	UnbondingEventChan     <-chan queuecli.QueueMessage
+	WithdrawEventChan      <-chan queuecli.QueueMessage
+	BtcInfoEventChan       <-chan queuecli.QueueMessage
+	ConfirmedInfoEventChan <-chan queuecli.QueueMessage
+	VersionedParams        *parser.ParsedGlobalParams
 }
 
 // bitcoin params used for testing
@@ -146,6 +147,8 @@ func StartWithBitcoinHandler(t *testing.T, h *BitcoindTestHandler, minerAddress
 	require.NoError(t, err)
 	unconfirmedEventChan, err := queueConsumer.BtcInfoQueue.ReceiveMessages()
 	require.NoError(t, err)
+	confirmedInfoEventChan, err := queueConsumer.ConfirmedInfoQueue.ReceiveMessages()
+	require.NoError(t, err)
 
 	db, err := cfg.DatabaseConfig.GetDbBackend()
 	require.NoError(t, err)
@@ -176,22 +179,23 @@ func StartWithBitcoinHandler(t *testing.T, h *BitcoindTestHandler, minerAddress
 	time.Sleep(3 * time.Second)
 
 	return &TestManager{
-		Config:             cfg,
-		Si:                 si,
-		BS:                 scanner,
-		serverStopper:      &interceptor,
-		wg:                 &wg,
-		BitcoindHandler:    h,
-		WalletClient:       rpcclient,
-		WalletPrivKey:      walletPrivKey.PrivKey,
-		MinerAddr:          minerAddress,
-		DirPath:            dirPath,
-		QueueConsumer:      queueConsumer,
-		StakingEventChan:   stakingEventChan,
-		UnbondingEventChan: unbondingEventChan,
-		WithdrawEventChan:  withdrawEventChan,
-		BtcInfoEventChan:   unconfirmedEventChan,
-		VersionedParams:    versionedParams,
+		Config:                 cfg,
+		Si:                     si,
+		BS:                     scanner,
+		serverStopper:          &interceptor,
+		wg:                     &wg,
+		BitcoindHandler:        h,
+		WalletClient:           rpcclient,
+		WalletPrivKey:          walletPrivKey.PrivKey,
+		MinerAddr:              minerAddress,
+		DirPath:                dirPath,
+		QueueConsumer:          queueConsumer,
+		StakingEventChan:       stakingEventChan,
+		UnbondingEventChan:     unbondingEventChan,
+		WithdrawEventChan:      withdrawEventChan,
+		BtcInfoEventChan:       unconfirmedEventChan,
+		ConfirmedInfoEventChan: confirmedInfoEventChan,
+		VersionedParams:        versionedParams,
 	}
 }
 
@@ -214,7 +218,10 @@ func ReStartFromHeight(t *testing.T, tm *TestManager, height uint64) *TestManage
 func DefaultStakingIndexerConfig(homePath string) *config.Config {
 	defaultConfig := config.DefaultConfigWithHome(homePath)
 
-	// both wallet and node are bicoind
+	// enable emitting extra events for testing
+	defaultConfig.ExtraEventEnabled = true
+
+	// both wallet and node are bitcoind
 	defaultConfig.BTCNetParams = *regtestParams
 
 	bitcoindHost := "127.0.0.1:18443"
@@ -361,6 +368,23 @@ func (tm *TestManager) CheckNextWithdrawEvent(t *testing.T, stakingTxHash chainh
 	require.NoError(t, err)
 }
 
+func (tm *TestManager) CheckConfirmedInfoEvent(t *testing.T, height, tvl uint64) {
+	var confirmedInfoEv queuecli.ConfirmedInfoEvent
+
+	for {
+		confirmedInfoEventBytes := <-tm.ConfirmedInfoEventChan
+		err := tm.QueueConsumer.ConfirmedInfoQueue.DeleteMessage(confirmedInfoEventBytes.Receipt)
+		require.NoError(t, err)
+		err = json.Unmarshal([]byte(confirmedInfoEventBytes.Body), &confirmedInfoEv)
+		require.NoError(t, err)
+		if height != confirmedInfoEv.Height {
+			continue
+		}
+		require.Equal(t, confirmedInfoEv.Tvl, tvl)
+		return
+	}
+}
+
 func (tm *TestManager) CheckNextUnconfirmedEvent(t *testing.T, confirmedTvl, totalTvl uint64) {
 	var btcInfoEvent queuecli.BtcInfoEvent
 
@@ -376,6 +400,7 @@ func (tm *TestManager) CheckNextUnconfirmedEvent(t *testing.T, confirmedTvl, tot
 		if totalTvl != btcInfoEvent.UnconfirmedTvl {
 			continue
 		}
+
 		return
 	}
 }