feat: relax Arbitrum block finality check to speed up deposits

cmd/vega/commands/node/node.go

@@ -476,9 +476,13 @@ func (n *Command) startBlockchainClients() error {
 		return fmt.Errorf("could not instantiate secondary ethereum client: %w", err)
 	}
 
-	n.primaryEthConfirmations = ethclient.NewEthereumConfirmations(n.conf.Ethereum, n.primaryEthClient, nil)
+	n.primaryEthConfirmations = ethclient.NewEthereumConfirmations(n.conf.Ethereum, n.primaryEthClient, nil, ethclient.FinalityStateFinalized)
 
-	n.secondaryEthConfirmations = ethclient.NewEthereumConfirmations(n.conf.Ethereum, n.secondaryEthClient, nil)
+	// for arbitrum we can use the weaker check for finality and only require that the block is marked as "safe".
+	// This is because "safe" means that the batch has been send to L1 Ethereum and from then on its "final" on
+	// Arbitrum. If the batched-transaction is part of a re-org on Ethereum, it doesn't matter to Vega because core
+	// is only looking at the Arbitrum blocks we don't track the batch, so we don't need to wait for full finality.
+	n.secondaryEthConfirmations = ethclient.NewEthereumConfirmations(n.conf.Ethereum, n.secondaryEthClient, nil, ethclient.FinalityStateSafe)
 
 	return nil
 }

core/banking/asset_action.go

@@ -76,6 +76,10 @@ func (t *assetAction) GetType() types.NodeVoteType {
 	}
 }
 
+func (t *assetAction) GetChainID() string {
+	return t.chainID
+}
+
 func (t *assetAction) IsBuiltinAssetDeposit() bool {
 	return t.builtinD != nil
 }

core/client/eth/ethereum_confirmations.go

@@ -28,7 +28,14 @@ import (
 var (
 	ErrMissingConfirmations = errors.New("not enough confirmations")
 	ErrBlockNotFinalized    = errors.New("block not finalized")
-	finalised               = big.NewInt(-3)
+)
+
+type FinalityState int
+
+const (
+	FinalityStateSafe FinalityState = iota
+	FinalityStateFinalized
+	FinalityStateLatest
 )
 
 //go:generate go run github.com/golang/mock/mockgen -destination mocks/ethereum_client_confirmations_mock.go -package mocks code.vegaprotocol.io/vega/core/staking EthereumClientConfirmations
@@ -58,17 +65,31 @@ type EthereumConfirmations struct {
 	curHeightLastUpdate time.Time
 	finHeight           uint64
 	finHeightLastUpdate time.Time
+	finState            *big.Int
 }
 
-func NewEthereumConfirmations(cfg Config, ethClient EthereumClientConfirmations, time Time) *EthereumConfirmations {
+func NewEthereumConfirmations(cfg Config, ethClient EthereumClientConfirmations, time Time, cs FinalityState) *EthereumConfirmations {
 	if time == nil {
 		time = StdTime{}
 	}
-	return &EthereumConfirmations{
+
+	conf := &EthereumConfirmations{
 		retryDelay: cfg.RetryDelay.Get(),
 		ethClient:  ethClient,
 		time:       time,
 	}
+
+	switch cs {
+	case FinalityStateSafe:
+		conf.finState = big.NewInt(-4)
+	case FinalityStateFinalized:
+		conf.finState = big.NewInt(-3)
+	case FinalityStateLatest:
+		conf.finState = nil
+	default:
+		panic("unexpected confirmation state")
+	}
+	return conf
 }
 
 func (e *EthereumConfirmations) GetConfirmations() uint64 {
@@ -117,7 +138,7 @@ func (e *EthereumConfirmations) finalizedHeight(ctx context.Context) (uint64, er
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
-	h, lastUpdate, err := e.getHeight(ctx, e.finHeight, e.finHeightLastUpdate, finalised)
+	h, lastUpdate, err := e.getHeight(ctx, e.finHeight, e.finHeightLastUpdate, e.finState)
 	if err != nil {
 		return e.finHeight, err
 	}

core/client/eth/ethereum_confirmations_test.go

@@ -35,7 +35,7 @@ func TestEthereumConfirmations(t *testing.T) {
 	tim := localMocks.NewMockTime(ctrl)
 	cfg := eth.NewDefaultConfig()
 	cfg.RetryDelay.Duration = 15 * time.Second
-	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim)
+	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim, eth.FinalityStateFinalized)
 	defer ctrl.Finish()
 
 	ethCfns.UpdateConfirmations(30)
@@ -100,7 +100,7 @@ func TestBlockFinalisation(t *testing.T) {
 	tim := localMocks.NewMockTime(ctrl)
 	cfg := eth.NewDefaultConfig()
 	cfg.RetryDelay.Duration = 15 * time.Second
-	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim)
+	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim, eth.FinalityStateFinalized)
 	defer ctrl.Finish()
 
 	ethCfns.UpdateConfirmations(10)
@@ -131,7 +131,7 @@ func TestCheckRequiredConfirmations(t *testing.T) {
 	tim := localMocks.NewMockTime(ctrl)
 	cfg := eth.NewDefaultConfig()
 	cfg.RetryDelay.Duration = 15 * time.Second
-	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim)
+	ethCfns := eth.NewEthereumConfirmations(cfg, ethClient, tim, eth.FinalityStateFinalized)
 	defer ctrl.Finish()
 
 	tim.EXPECT().Now().Times(1).Return(time.Unix(10, 0))

core/client/eth/l2_clients.go

@@ -73,7 +73,7 @@ func NewL2Clients(
 		}
 
 		clients[v.ChainID] = clt
-		confirmations[v.ChainID] = NewEthereumConfirmations(cfg, clt, nil)
+		confirmations[v.ChainID] = NewEthereumConfirmations(cfg, clt, nil, FinalityStateLatest)
 	}
 
 	return &L2Clients{
@@ -153,7 +153,7 @@ func (e *L2Clients) ReloadConf(cfg Config) {
 		}
 
 		e.clients[v.ChainID] = clt
-		e.confirmations[v.ChainID] = NewEthereumConfirmations(cfg, clt, nil)
+		e.confirmations[v.ChainID] = NewEthereumConfirmations(cfg, clt, nil, FinalityStateLatest)
 	}
 }
 

core/datasource/external/ethverifier/verifier.go

@@ -124,6 +124,12 @@ type pendingCallEvent struct {
 }
 
 func (p pendingCallEvent) GetID() string { return p.callEvent.Hash() }
+func (p pendingCallEvent) GetChainID() string {
+	if p.callEvent.SourceChainID == nil {
+		return ""
+	}
+	return strconv.Itoa(int(*p.callEvent.SourceChainID))
+}
 
 func (p pendingCallEvent) GetType() types.NodeVoteType {
 	return types.NodeVoteTypeEthereumContractCallResult

core/governance/node_validation.go

@@ -64,6 +64,16 @@ func (n *nodeProposal) GetID() string {
 	return n.ID
 }
 
+func (n *nodeProposal) GetChainID() string {
+	switch na := n.Terms.Change.(type) {
+	case *types.ProposalTermsNewAsset:
+		if erc20 := na.NewAsset.Changes.GetERC20(); erc20 != nil {
+			return erc20.ChainID
+		}
+	}
+	return ""
+}
+
 func (n *nodeProposal) GetType() types.NodeVoteType {
 	return types.NodeVoteTypeGovernanceValidateAsset
 }

core/netparams/bridge_mapping.go

@@ -25,7 +25,8 @@ var stagnet1 = `{
 	"multisig_control_contract": {
 	  "address": "0x764c51de728f09407f7f073f63fc0a8a6adf110e",
 	  "deployment_block_height": 27160717
-	}
+	},
+	"block_time": "250ms"
   }`
 
 var testnet = `{
@@ -38,7 +39,8 @@ var testnet = `{
 	"multisig_control_contract": {
 	  "address": "0x0A3f3E72FCe9862c750B0682aA75bb7261b3eb15",
 	  "deployment_block_height": 31628794
-	}
+	},
+	"block_time": "250ms"
   }`
 
 var bridgeMapping = map[string]string{

core/protocol/all_services.go

@@ -962,7 +962,7 @@ func (svcs *allServices) setupNetParameters(powWatchers []netparams.WatchParam)
 					return nil
 				}
 
-				svcs.witness.SetDefaultConfirmations(ethCfg.Confirmations())
+				svcs.witness.SetPrimaryDefaultConfirmations(ethCfg.ChainID(), ethCfg.Confirmations())
 				return nil
 			},
 		},
@@ -987,6 +987,7 @@ func (svcs *allServices) setupNetParameters(powWatchers []netparams.WatchParam)
 				}
 
 				svcs.banking.OnSecondaryEthChainIDUpdated(ethCfg.ChainID())
+				svcs.witness.SetSecondaryDefaultConfirmations(ethCfg.ChainID(), ethCfg.Confirmations(), ethCfg.BlockTime())
 				return nil
 			},
 		},

core/staking/accounting.go

@@ -68,6 +68,7 @@ type Accounting struct {
 
 	stakingAssetTotalSupply *num.Uint
 	stakingBridgeAddress    ethcmn.Address
+	chainID                 string
 
 	// snapshot bits
 	accState accountingSnapshotState
@@ -80,14 +81,17 @@ type Accounting struct {
 }
 
 type pendingStakeTotalSupply struct {
-	sts   *types.StakeTotalSupply
-	check func() error
+	sts     *types.StakeTotalSupply
+	chainID string
+	check   func() error
 }
 
 func (p pendingStakeTotalSupply) GetID() string {
 	return hex.EncodeToString(vgcrypto.Hash([]byte(p.sts.String())))
 }
 
+func (p pendingStakeTotalSupply) GetChainID() string { return p.chainID }
+
 func (p pendingStakeTotalSupply) GetType() types.NodeVoteType {
 	return types.NodeVoteTypeStakeTotalSupply
 }
@@ -171,8 +175,9 @@ func (a *Accounting) GetAllAvailableBalances() map[string]*num.Uint {
 	return balances
 }
 
-func (a *Accounting) UpdateStakingBridgeAddress(stakingBridgeAddress ethcmn.Address) error {
-	a.stakingBridgeAddress = stakingBridgeAddress
+func (a *Accounting) UpdateStakingBridgeAddress(ethCfg *types.EthereumConfig) error {
+	a.stakingBridgeAddress = ethCfg.StakingBridgeAddresses()[0]
+	a.chainID = ethCfg.ChainID()
 
 	if !a.accState.isRestoring {
 		if err := a.updateStakingAssetTotalSupply(); err != nil {
@@ -195,7 +200,8 @@ func (a *Accounting) ProcessStakeTotalSupply(_ context.Context, evt *types.Stake
 	expectedSupply := evt.TotalSupply.Clone()
 
 	a.pendingStakeTotalSupply = &pendingStakeTotalSupply{
-		sts: evt,
+		sts:     evt,
+		chainID: a.chainID,
 		check: func() error {
 			totalSupply, err := a.getStakeAssetTotalSupply(a.stakingBridgeAddress)
 			if err != nil {

core/staking/accounting_snapshot.go

@@ -172,7 +172,8 @@ func (a *Accounting) restoreStakingAccounts(ctx context.Context, accounts *types
 	if pendingSupply != nil {
 		expectedSupply := pendingSupply.TotalSupply.Clone()
 		a.pendingStakeTotalSupply = &pendingStakeTotalSupply{
-			sts: pendingSupply,
+			sts:     pendingSupply,
+			chainID: a.chainID,
 			check: func() error {
 				totalSupply, err := a.getStakeAssetTotalSupply(a.stakingBridgeAddress)
 				if err != nil {

core/staking/stake_verifier.go

@@ -89,19 +89,23 @@ type StakeVerifier struct {
 
 type pendingSD struct {
 	*types.StakeDeposited
-	check func() error
+	chainID string
+	check   func() error
 }
 
 func (p pendingSD) GetID() string                    { return p.ID }
+func (p pendingSD) GetChainID() string               { return p.chainID }
 func (p pendingSD) GetType() types.NodeVoteType      { return types.NodeVoteTypeStakeDeposited }
 func (p *pendingSD) Check(ctx context.Context) error { return p.check() }
 
 type pendingSR struct {
 	*types.StakeRemoved
-	check func() error
+	chainID string
+	check   func() error
 }
 
 func (p pendingSR) GetID() string                    { return p.ID }
+func (p pendingSR) GetChainID() string               { return p.chainID }
 func (p pendingSR) GetType() types.NodeVoteType      { return types.NodeVoteTypeStakeRemoved }
 func (p *pendingSR) Check(ctx context.Context) error { return p.check() }
 
@@ -166,6 +170,7 @@ func (s *StakeVerifier) ProcessStakeRemoved(
 
 	pending := &pendingSR{
 		StakeRemoved: event,
+		chainID:      s.accs.chainID,
 		check:        func() error { return s.ocv.CheckStakeRemoved(event) },
 	}
 	s.pendingSRs = append(s.pendingSRs, pending)
@@ -191,6 +196,7 @@ func (s *StakeVerifier) ProcessStakeDeposited(
 
 	pending := &pendingSD{
 		StakeDeposited: event,
+		chainID:        s.accs.chainID,
 		check:          func() error { return s.ocv.CheckStakeDeposited(event) },
 	}
 

core/staking/stake_verifier_snapshot.go

@@ -137,6 +137,7 @@ func (s *StakeVerifier) restorePendingSD(ctx context.Context, deposited []*types
 
 		pending := &pendingSD{
 			StakeDeposited: d,
+			chainID:        s.accs.chainID,
 			check:          func() error { return s.ocv.CheckStakeDeposited(d) },
 		}
 
@@ -189,6 +190,7 @@ func (s *StakeVerifier) restorePendingSR(ctx context.Context, removed []*types.S
 
 		pending := &pendingSR{
 			StakeRemoved: r,
+			chainID:      s.accs.chainID,
 			check:        func() error { return s.ocv.CheckStakeRemoved(r) },
 		}
 

core/staking/staking.go

@@ -73,7 +73,7 @@ func New(
 			ocv.UpdateStakingBridgeAddresses(ethCfg.StakingBridgeAddresses())
 
 			// We just need one of the staking bridges.
-			if err := accs.UpdateStakingBridgeAddress(ethCfg.StakingBridgeAddresses()[0]); err != nil {
+			if err := accs.UpdateStakingBridgeAddress(ethCfg); err != nil {
 				return fmt.Errorf("couldn't update Ethereum configuration in accounting: %w", err)
 			}
 

core/types/ethereum.go

@@ -41,6 +41,7 @@ var (
 	ErrDuplicateChainID                                   = errors.New("duplicate chain ID name")
 	ErrCannotRemoveL2Config                               = errors.New("L2 config cannot be removed")
 	ErrCanOnlyAmendedConfirmationsAndBlockInterval        = errors.New("can only amended L2 config confirmations and block interval")
+	ErrInvalidBlockLengthDuration                         = errors.New("block-length duration is invalid")
 )
 
 type EthereumConfig struct {

core/types/ethereum_secondary.go

@@ -17,6 +17,7 @@ package types
 
 import (
 	"fmt"
+	"time"
 
 	vgreflect "code.vegaprotocol.io/vega/libs/reflect"
 	proto "code.vegaprotocol.io/vega/protos/vega"
@@ -28,6 +29,7 @@ type EVMChainConfig struct {
 	confirmations    uint64
 	collateralBridge EthereumContract
 	multiSigControl  EthereumContract
+	blockTime        time.Duration
 }
 
 func EVMChainConfigFromUntypedProto(v interface{}) (*EVMChainConfig, error) {
@@ -46,7 +48,7 @@ func EVMChainConfigFromUntypedProto(v interface{}) (*EVMChainConfig, error) {
 
 func SecondaryConfigFromProto(cfgProto *proto.EVMChainConfig) (*EVMChainConfig, error) {
 	if err := CheckEVMChainConfig(cfgProto); err != nil {
-		return nil, fmt.Errorf("invalid second ethereum configuration: %w", err)
+		return nil, fmt.Errorf("invalid EVM chain configuration: %w", err)
 	}
 
 	cfg := &EVMChainConfig{
@@ -62,9 +64,21 @@ func SecondaryConfigFromProto(cfgProto *proto.EVMChainConfig) (*EVMChainConfig,
 		},
 	}
 
+	if len(cfgProto.BlockTime) != 0 {
+		bl, err := time.ParseDuration(cfgProto.BlockTime)
+		if err != nil {
+			return nil, fmt.Errorf("invalid EVM chain configuration, block_length: %w", err)
+		}
+		cfg.blockTime = bl
+	}
+
 	return cfg, nil
 }
 
+func (c *EVMChainConfig) BlockTime() time.Duration {
+	return c.blockTime
+}
+
 func (c *EVMChainConfig) ChainID() string {
 	return c.chainID
 }
@@ -123,6 +137,13 @@ func CheckEVMChainConfig(cfgProto *proto.EVMChainConfig) error {
 		return ErrUnsupportedCollateralBridgeDeploymentBlockHeight
 	}
 
+	if len(cfgProto.BlockTime) != 0 {
+		_, err := time.ParseDuration(cfgProto.BlockTime)
+		if err != nil {
+			return ErrInvalidBlockLengthDuration
+		}
+	}
+
 	return nil
 }