feat(zetaclient)!: orchestrator V2

changelog.md

@@ -20,6 +20,7 @@
 * [3170](https://github.com/zeta-chain/node/pull/3170) - revamp TSS package in zetaclient
 * [3291](https://github.com/zeta-chain/node/pull/3291) - revamp zetaclient initialization (+ graceful shutdown)
 * [3319](https://github.com/zeta-chain/node/pull/3319) - implement scheduler for zetaclient
+* [3332](https://github.com/zeta-chain/node/pull/3332) - implement orchestrator V2. Move BTC observer-signer to V2
 
 ### Fixes
 

cmd/zetaclientd/inbound.go

@@ -6,16 +6,17 @@ import (
 	"strconv"
 	"strings"
 
-	"cosmossdk.io/errors"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	ethcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/spf13/cobra"
 
 	"github.com/zeta-chain/node/pkg/coin"
 	"github.com/zeta-chain/node/testutil/sample"
 	"github.com/zeta-chain/node/zetaclient/chains/base"
 	btcobserver "github.com/zeta-chain/node/zetaclient/chains/bitcoin/observer"
+	btcrpc "github.com/zeta-chain/node/zetaclient/chains/bitcoin/rpc"
 	evmobserver "github.com/zeta-chain/node/zetaclient/chains/evm/observer"
 	"github.com/zeta-chain/node/zetaclient/config"
 	zctx "github.com/zeta-chain/node/zetaclient/context"
@@ -156,17 +157,36 @@ func InboundGetBallot(_ *cobra.Command, args []string) error {
 		}
 		fmt.Println("CoinType : ", coinType)
 	} else if chain.IsBitcoin() {
-		observer, ok := observers[chainID]
-		if !ok {
-			return fmt.Errorf("observer not found for btc chain %d", chainID)
+		bitcoinConfig, found := appContext.Config().GetBTCConfig(chain.ID())
+		if !found {
+			return fmt.Errorf("unable to find btc config")
 		}
 
-		btcObserver, ok := observer.(*btcobserver.Observer)
-		if !ok {
-			return fmt.Errorf("observer is not btc observer for chain %d", chainID)
+		rpcClient, err := btcrpc.NewRPCClient(bitcoinConfig)
+		if err != nil {
+			return errors.Wrap(err, "unable to create rpc client")
+		}
+
+		database, err := db.NewFromSqliteInMemory(true)
+		if err != nil {
+			return errors.Wrap(err, "unable to open database")
+		}
+
+		observer, err := btcobserver.NewObserver(
+			*chain.RawChain(),
+			rpcClient,
+			*chain.Params(),
+			client,
+			nil,
+			database,
+			baseLogger,
+			nil,
+		)
+		if err != nil {
+			return errors.Wrap(err, "unable to create btc observer")
 		}
 
-		ballotIdentifier, err = btcObserver.CheckReceiptForBtcTxHash(ctx, inboundHash, false)
+		ballotIdentifier, err = observer.CheckReceiptForBtcTxHash(ctx, inboundHash, false)
 		if err != nil {
 			return err
 		}

cmd/zetaclientd/start.go

@@ -14,6 +14,7 @@ import (
 	"github.com/zeta-chain/node/pkg/constant"
 	"github.com/zeta-chain/node/pkg/graceful"
 	zetaos "github.com/zeta-chain/node/pkg/os"
+	"github.com/zeta-chain/node/pkg/scheduler"
 	"github.com/zeta-chain/node/zetaclient/chains/base"
 	"github.com/zeta-chain/node/zetaclient/config"
 	zctx "github.com/zeta-chain/node/zetaclient/context"
@@ -115,6 +116,8 @@ func Start(_ *cobra.Command, _ []string) error {
 		return errors.Wrap(err, "unable to setup TSS service")
 	}
 
+	graceful.AddStopper(tss.Stop)
+
 	// Starts various background TSS listeners.
 	// Shuts down zetaclientd if any is triggered.
 	maintenance.NewTSSListener(zetacoreClient, logger.Std).Listen(ctx, func() {
@@ -159,9 +162,27 @@ func Start(_ *cobra.Command, _ []string) error {
 		return errors.Wrap(err, "unable to create orchestrator")
 	}
 
+	taskScheduler := scheduler.New(logger.Std)
+	maestroV2Deps := &orchestrator.Dependencies{
+		Zetacore:  zetacoreClient,
+		TSS:       tss,
+		DBPath:    dbPath,
+		Telemetry: telemetry,
+	}
+
+	maestroV2, err := orchestrator.NewV2(taskScheduler, maestroV2Deps, logger)
+	if err != nil {
+		return errors.Wrap(err, "unable to create orchestrator V2")
+	}
+
 	// Start orchestrator with all observers and signers
 	graceful.AddService(ctx, maestro)
 
+	// Start orchestrator V2
+	// V2 will co-exist with V1 until all types of chains will be refactored (BTC, EVM, SOL, TON).
+	// (currently it's only BTC)
+	graceful.AddService(ctx, maestroV2)
+
 	// Block current routine until a shutdown signal is received
 	graceful.WaitForShutdown()
 

e2e/e2etests/test_bitcoin_deposit_and_call_revert.go

@@ -6,7 +6,7 @@ import (
 	"github.com/zeta-chain/node/e2e/runner"
 	"github.com/zeta-chain/node/e2e/utils"
 	"github.com/zeta-chain/node/testutil/sample"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinDepositAndCallRevert(r *runner.E2ERunner, args []string) {

e2e/e2etests/test_bitcoin_deposit_and_call_revert_with_dust.go

@@ -10,7 +10,7 @@ import (
 	"github.com/zeta-chain/node/pkg/constant"
 	"github.com/zeta-chain/node/testutil/sample"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 // TestBitcoinDepositAndCallRevertWithDust sends a Bitcoin deposit that reverts with a dust amount in the revert outbound.

e2e/e2etests/test_bitcoin_deposit_call.go

@@ -9,7 +9,7 @@ import (
 	"github.com/zeta-chain/node/e2e/utils"
 	testcontract "github.com/zeta-chain/node/testutil/contracts"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	"github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinDepositAndCall(r *runner.E2ERunner, args []string) {
@@ -20,7 +20,7 @@ func TestBitcoinDepositAndCall(r *runner.E2ERunner, args []string) {
 	// Given amount to send
 	require.Len(r, args, 1)
 	amount := utils.ParseFloat(r, args[0])
-	amountTotal := amount + zetabitcoin.DefaultDepositorFee
+	amountTotal := amount + common.DefaultDepositorFee
 
 	// Given a list of UTXOs
 	utxos, err := r.ListDeployerUTXOs()
@@ -45,7 +45,7 @@ func TestBitcoinDepositAndCall(r *runner.E2ERunner, args []string) {
 	utils.RequireCCTXStatus(r, cctx, crosschaintypes.CctxStatus_OutboundMined)
 
 	// check if example contract has been called, 'bar' value should be set to amount
-	amountSats, err := zetabitcoin.GetSatoshis(amount)
+	amountSats, err := common.GetSatoshis(amount)
 	require.NoError(r, err)
 	utils.MustHaveCalledExampleContract(r, contract, big.NewInt(amountSats))
 }

e2e/e2etests/test_bitcoin_donation.go

@@ -9,7 +9,7 @@ import (
 	"github.com/zeta-chain/node/e2e/utils"
 	"github.com/zeta-chain/node/pkg/constant"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinDonation(r *runner.E2ERunner, args []string) {

e2e/e2etests/test_bitcoin_std_deposit.go

@@ -10,7 +10,7 @@ import (
 	"github.com/zeta-chain/node/e2e/utils"
 	"github.com/zeta-chain/node/pkg/memo"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	"github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	"github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinStdMemoDeposit(r *runner.E2ERunner, args []string) {
@@ -54,7 +54,7 @@ func TestBitcoinStdMemoDeposit(r *runner.E2ERunner, args []string) {
 
 	// the runner balance should be increased by the deposit amount
 	amountIncreased := new(big.Int).Sub(balanceAfter, balanceBefore)
-	amountSatoshis, err := bitcoin.GetSatoshis(amount)
+	amountSatoshis, err := common.GetSatoshis(amount)
 	require.NoError(r, err)
 	require.Positive(r, amountSatoshis)
 	// #nosec G115 always positive

e2e/e2etests/test_bitcoin_std_deposit_and_call.go

@@ -10,7 +10,7 @@ import (
 	"github.com/zeta-chain/node/pkg/memo"
 	testcontract "github.com/zeta-chain/node/testutil/contracts"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinStdMemoDepositAndCall(r *runner.E2ERunner, args []string) {

e2e/e2etests/test_bitcoin_std_memo_inscribed_deposit_and_call.go

@@ -10,7 +10,7 @@ import (
 	"github.com/zeta-chain/node/pkg/memo"
 	testcontract "github.com/zeta-chain/node/testutil/contracts"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	"github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 )
 
 func TestBitcoinStdMemoInscribedDepositAndCall(r *runner.E2ERunner, args []string) {
@@ -53,7 +53,7 @@ func TestBitcoinStdMemoInscribedDepositAndCall(r *runner.E2ERunner, args []strin
 	utils.RequireCCTXStatus(r, cctx, crosschaintypes.CctxStatus_OutboundMined)
 
 	// check if example contract has been called, 'bar' value should be set to correct amount
-	depositFeeSats, err := zetabitcoin.GetSatoshis(zetabitcoin.DefaultDepositorFee)
+	depositFeeSats, err := common.GetSatoshis(common.DefaultDepositorFee)
 	require.NoError(r, err)
 	receiveAmount := depositAmount - depositFeeSats
 	utils.MustHaveCalledExampleContract(r, contract, big.NewInt(receiveAmount))

e2e/runner/bitcoin.go

@@ -23,7 +23,7 @@ import (
 	"github.com/zeta-chain/node/pkg/constant"
 	"github.com/zeta-chain/node/pkg/memo"
 	crosschaintypes "github.com/zeta-chain/node/x/crosschain/types"
-	zetabitcoin "github.com/zeta-chain/node/zetaclient/chains/bitcoin"
+	zetabtc "github.com/zeta-chain/node/zetaclient/chains/bitcoin/common"
 	btcobserver "github.com/zeta-chain/node/zetaclient/chains/bitcoin/observer"
 	"github.com/zeta-chain/node/zetaclient/chains/bitcoin/signer"
 )
@@ -100,7 +100,7 @@ func (r *E2ERunner) DepositBTCWithAmount(amount float64, memo *memo.InboundMemo)
 	r.Logger.Info("Now sending two txs to TSS address...")
 
 	// add depositor fee so that receiver gets the exact given 'amount' in ZetaChain
-	amount += zetabitcoin.DefaultDepositorFee
+	amount += zetabtc.DefaultDepositorFee
 
 	// deposit to TSS address
 	var txHash *chainhash.Hash
@@ -148,7 +148,7 @@ func (r *E2ERunner) DepositBTC(receiver common.Address) {
 	r.Logger.Info("Now sending two txs to TSS address and tester ZEVM address...")
 
 	// send initial BTC to the tester ZEVM address
-	amount := 1.15 + zetabitcoin.DefaultDepositorFee
+	amount := 1.15 + zetabtc.DefaultDepositorFee
 	txHash, err := r.DepositBTCWithLegacyMemo(amount, utxos[:2], receiver)
 	require.NoError(r, err)
 
@@ -241,7 +241,7 @@ func (r *E2ERunner) sendToAddrFromDeployerWithMemo(
 
 	// use static fee 0.0005 BTC to calculate change
 	feeSats := btcutil.Amount(0.0005 * btcutil.SatoshiPerBitcoin)
-	amountInt, err := zetabitcoin.GetSatoshis(amount)
+	amountInt, err := zetabtc.GetSatoshis(amount)
 	require.NoError(r, err)
 	amountSats := btcutil.Amount(amountInt)
 	change := inputSats - feeSats - amountSats
@@ -351,7 +351,7 @@ func (r *E2ERunner) InscribeToTSSFromDeployerWithMemo(
 
 	// parameters to build the reveal transaction
 	commitOutputIdx := uint32(0)
-	commitAmount, err := zetabitcoin.GetSatoshis(amount)
+	commitAmount, err := zetabtc.GetSatoshis(amount)
 	require.NoError(r, err)
 
 	// build the reveal transaction to spend above funds
@@ -412,7 +412,7 @@ func (r *E2ERunner) QueryOutboundReceiverAndAmount(txid string) (string, int64)
 	// parse receiver address from pkScript
 	txOutput := revertTx.MsgTx().TxOut[1]
 	pkScript := txOutput.PkScript
-	receiver, err := zetabitcoin.DecodeScriptP2WPKH(hex.EncodeToString(pkScript), r.BitcoinParams)
+	receiver, err := zetabtc.DecodeScriptP2WPKH(hex.EncodeToString(pkScript), r.BitcoinParams)
 	require.NoError(r, err)
 
 	return receiver, txOutput.Value

go.mod

@@ -163,7 +163,7 @@ require (
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/orderedcode v0.0.1 // indirect
 	github.com/google/s2a-go v0.1.7 // indirect
-	github.com/google/uuid v1.6.0 // indirect
+	github.com/google/uuid v1.6.0
 	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
 	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
 	github.com/gorilla/handlers v1.5.1 // indirect
@@ -249,8 +249,8 @@ require (
 	github.com/pelletier/go-toml/v2 v2.1.0 // indirect
 	github.com/petermattis/goid v0.0.0-20230317030725-371a4b8eda08 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_model v0.4.0 // indirect
-	github.com/prometheus/common v0.42.0 // indirect
+	github.com/prometheus/client_model v0.4.0
+	github.com/prometheus/common v0.42.0
 	github.com/prometheus/procfs v0.9.0 // indirect
 	github.com/raulk/go-watchdog v1.3.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect

pkg/chains/chain.go

@@ -8,6 +8,8 @@
 	"github.com/btcsuite/btcd/chaincfg"
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/tonkeeper/tongo/ton"
+
+	"github.com/zeta-chain/node/zetaclient/logs"
 )
 
 // Validate checks whether the chain is valid
@@ -108,6 +110,13 @@
 	return chain.Consensus == Consensus_catchain_consensus
 }
 
+func (chain Chain) LogFields() map[string]any {
+	return map[string]any{
+		logs.FieldChain:        chain.ChainId,
+		logs.FieldChainNetwork: chain.Network.String(),
+	}
+}
+
 // DecodeAddressFromChainID decode the address string to bytes
 // additionalChains is a list of additional chains to search from
 // in practice, it is used in the protocol to dynamically support new chains without doing an upgrade

pkg/fanout/fanout.go

@@ -0,0 +1,66 @@
+// Package fanout provides a fan-out pattern implementation.
+// It allows one channel to stream data to multiple independent channels.
+// Note that context handling is out of the scope of this package.
+package fanout
+
+import "sync"
+
+const DefaultBuffer = 8
+
+// FanOut is a fan-out pattern implementation.
+// It is NOT a worker pool, so use it wisely.
+type FanOut[T any] struct {
+	input   <-chan T
+	outputs []chan T
+
+	// outputBuffer chan buffer size for outputs channels.
+	// This helps with writing to chan in case of slow consumers.
+	outputBuffer int
+
+	mu sync.RWMutex
+}
+
+// New constructs FanOut
+func New[T any](source <-chan T, buf int) *FanOut[T] {
+	return &FanOut[T]{
+		input:        source,
+		outputs:      make([]chan T, 0),
+		outputBuffer: buf,
+	}
+}
+
+func (f *FanOut[T]) Add() <-chan T {
+	out := make(chan T, f.outputBuffer)
+
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	f.outputs = append(f.outputs, out)
+
+	return out
+}
+
+// Start starts the fan-out process
+func (f *FanOut[T]) Start() {
+	go func() {
+		// loop for new data
+		for data := range f.input {
+			f.mu.RLock()
+			for _, output := range f.outputs {
+				// note that this might spawn lots of goroutines.
+				// it is a naive approach, but should be more than enough for our use cases.
+				go func(output chan<- T) { output <- data }(output)
+			}
+			f.mu.RUnlock()
+		}
+
+		// at this point, the input was closed
+		f.mu.Lock()
+		defer f.mu.Unlock()
+		for _, out := range f.outputs {
+			close(out)
+		}
+
+		f.outputs = nil
+	}()
+}