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"
@@ -108,6 +109,8 @@ func Start(_ *cobra.Command, _ []string) error {
 		return nil
 	}
 
+	graceful.AddStopper(tss.Stop)
+
 	// Starts various background TSS listeners.
 	// Shuts down zetaclientd if any is triggered.
 	maintenance.NewTSSListener(zetacoreClient, logger.Std).Listen(ctx, func() {
@@ -152,9 +155,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 @@ import (
 	"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 @@ func (chain Chain) IsTONChain() bool {
 	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/scheduler/opts.go

@@ -29,14 +29,17 @@ func Interval(interval time.Duration) Opt {
 	return func(_ *Task, opts *taskOpts) { opts.interval = interval }
 }
 
-// Skipper sets task skipper function
+// Skipper sets task skipper function. If it returns true, the task is skipped.
 func Skipper(skipper func() bool) Opt {
 	return func(t *Task, _ *taskOpts) { t.skipper = skipper }
 }
 
 // IntervalUpdater sets interval updater function.
 func IntervalUpdater(intervalUpdater func() time.Duration) Opt {
-	return func(_ *Task, opts *taskOpts) { opts.intervalUpdater = intervalUpdater }
+	return func(_ *Task, opts *taskOpts) {
+		opts.interval = intervalUpdater()
+		opts.intervalUpdater = intervalUpdater
+	}
 }
 
 // BlockTicker makes Task to listen for new zeta blocks

pkg/scheduler/scheduler.go

@@ -107,6 +107,18 @@ func (s *Scheduler) Register(ctx context.Context, exec Executable, opts ...Opt)
 	return task
 }
 
+func (s *Scheduler) Tasks() map[uuid.UUID]*Task {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	copied := make(map[uuid.UUID]*Task, len(s.tasks))
+	for k, v := range s.tasks {
+		copied[k] = v
+	}
+
+	return copied
+}
+
 // Stop stops all tasks.
 func (s *Scheduler) Stop() {
 	s.StopGroup("")
@@ -132,6 +144,11 @@ func (s *Scheduler) StopGroup(group Group) {
 		return
 	}
 
+	s.logger.Info().
+		Int("tasks", len(selectedTasks)).
+		Str("group", string(group)).
+		Msg("Stopping scheduler group")
+
 	// Stop all selected tasks concurrently
 	var wg sync.WaitGroup
 	wg.Add(len(selectedTasks))
@@ -161,6 +178,14 @@ func (t *Task) Stop() {
 	t.logger.Info().Int64("time_taken_ms", timeTakenMS).Msg("Stopped scheduler task")
 }
 
+func (t *Task) Group() Group {
+	return t.group
+}
+
+func (t *Task) Name() string {
+	return t.name
+}
+
 // execute executes Task with additional logging and metrics.
 func (t *Task) execute(ctx context.Context) error {
 	startedAt := time.Now().UTC()