Add arb_checkPublisherHealth method

arbnode/api.go

@@ -22,11 +22,25 @@ import (
 )
 
 type BlockValidatorAPI struct {
+	val *validator.BlockValidator
+}
+
+func (a *BlockValidatorAPI) LatestValidatedBlock(ctx context.Context) (hexutil.Uint64, error) {
+	block := a.val.LastBlockValidated()
+	return hexutil.Uint64(block), nil
+}
+
+func (a *BlockValidatorAPI) LatestValidatedBlockHash(ctx context.Context) (common.Hash, error) {
+	_, hash, _ := a.val.LastBlockValidatedAndHash()
+	return hash, nil
+}
+
+type BlockValidatorDebugAPI struct {
 	val        *validator.BlockValidator
 	blockchain *core.BlockChain
 }
 
-func (a *BlockValidatorAPI) RevalidateBlock(ctx context.Context, blockNum rpc.BlockNumberOrHash, moduleRootOptional *common.Hash) (bool, error) {
+func (a *BlockValidatorDebugAPI) RevalidateBlock(ctx context.Context, blockNum rpc.BlockNumberOrHash, moduleRootOptional *common.Hash) (bool, error) {
 	header, err := arbitrum.HeaderByNumberOrHash(a.blockchain, blockNum)
 	if err != nil {
 		return false, err
@@ -47,14 +61,12 @@ func (a *BlockValidatorAPI) RevalidateBlock(ctx context.Context, blockNum rpc.Bl
 	return a.val.ValidateBlock(ctx, header, moduleRoot)
 }
 
-func (a *BlockValidatorAPI) LatestValidatedBlock(ctx context.Context) (hexutil.Uint64, error) {
-	block := a.val.LastBlockValidated()
-	return hexutil.Uint64(block), nil
+type ArbAPI struct {
+	txPublisher TransactionPublisher
 }
 
-func (a *BlockValidatorAPI) LatestValidatedBlockHash(ctx context.Context) (common.Hash, error) {
-	_, hash, _ := a.val.LastBlockValidatedAndHash()
-	return hash, nil
+func (a *ArbAPI) CheckPublisherHealth(ctx context.Context) error {
+	return a.txPublisher.CheckHealth(ctx)
 }
 
 type ArbDebugAPI struct {

arbnode/arb_interface.go

@@ -12,6 +12,7 @@ import (
 
 type TransactionPublisher interface {
 	PublishTransaction(ctx context.Context, tx *types.Transaction) error
+	CheckHealth(ctx context.Context) error
 	Initialize(context.Context) error
 	Start(context.Context) error
 	StopAndWait()

arbnode/forwarder.go

@@ -5,19 +5,26 @@ package arbnode
 
 import (
 	"context"
+	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/pkg/errors"
 )
 
 type TxForwarder struct {
-	enabled int32
-	target  string
-	timeout time.Duration
-	client  *ethclient.Client
+	enabled   int32
+	target    string
+	timeout   time.Duration
+	rpcClient *rpc.Client
+	ethClient *ethclient.Client
+
+	healthMutex   sync.Mutex
+	healthErr     error
+	healthChecked time.Time
 }
 
 func NewForwarder(target string, timeout time.Duration) *TxForwarder {
@@ -40,22 +47,46 @@ func (f *TxForwarder) PublishTransaction(inctx context.Context, tx *types.Transa
 	}
 	ctx, cancelFunc := f.ctxWithTimeout(inctx)
 	defer cancelFunc()
-	return f.client.SendTransaction(ctx, tx)
+	return f.ethClient.SendTransaction(ctx, tx)
+}
+
+const cacheUpstreamHealth = 2 * time.Second
+const maxHealthTimeout = 10 * time.Second
+
+func (f *TxForwarder) CheckHealth(inctx context.Context) error {
+	if atomic.LoadInt32(&f.enabled) == 0 {
+		return ErrNoSequencer
+	}
+	f.healthMutex.Lock()
+	defer f.healthMutex.Unlock()
+	if time.Since(f.healthChecked) > cacheUpstreamHealth {
+		timeout := f.timeout
+		if timeout == time.Duration(0) || timeout >= maxHealthTimeout {
+			timeout = maxHealthTimeout
+		}
+		ctx, cancelFunc := context.WithTimeout(context.Background(), timeout)
+		defer cancelFunc()
+		f.healthErr = f.rpcClient.CallContext(ctx, nil, "arb_checkPublisherHealth")
+		f.healthChecked = time.Now()
+	}
+	return f.healthErr
 }
 
 func (f *TxForwarder) Initialize(inctx context.Context) error {
 	if f.target == "" {
-		f.client = nil
+		f.rpcClient = nil
+		f.ethClient = nil
 		f.enabled = 0
 		return nil
 	}
 	ctx, cancelFunc := f.ctxWithTimeout(inctx)
 	defer cancelFunc()
-	client, err := ethclient.DialContext(ctx, f.target)
+	rpcClient, err := rpc.DialContext(ctx, f.target)
 	if err != nil {
 		return err
 	}
-	f.client = client
+	f.rpcClient = rpcClient
+	f.ethClient = ethclient.NewClient(rpcClient)
 	f.enabled = 1
 	return nil
 }
@@ -77,8 +108,14 @@ func NewTxDropper() *TxDropper {
 	return &TxDropper{}
 }
 
+var txDropperErr = errors.New("publishing transactions not supported by this endpoint")
+
 func (f *TxDropper) PublishTransaction(ctx context.Context, tx *types.Transaction) error {
-	return errors.New("transactions not supported by this endpoint")
+	return txDropperErr
+}
+
+func (f *TxDropper) CheckHealth(ctx context.Context) error {
+	return txDropperErr
 }
 
 func (f *TxDropper) Initialize(ctx context.Context) error { return nil }

arbnode/node.go

@@ -1056,11 +1056,23 @@ func CreateNode(
 		apis = append(apis, rpc.API{
 			Namespace: "arb",
 			Version:   "1.0",
-			Service:   &BlockValidatorAPI{val: currentNode.BlockValidator, blockchain: l2BlockChain},
+			Service:   &BlockValidatorAPI{val: currentNode.BlockValidator},
+			Public:    false,
+		})
+		apis = append(apis, rpc.API{
+			Namespace: "arbdebug",
+			Version:   "1.0",
+			Service:   &BlockValidatorDebugAPI{val: currentNode.BlockValidator, blockchain: l2BlockChain},
 			Public:    false,
 		})
 	}
 
+	apis = append(apis, rpc.API{
+		Namespace: "arb",
+		Version:   "1.0",
+		Service:   &ArbAPI{currentNode.TxPublisher},
+		Public:    false,
+	})
 	apis = append(apis, rpc.API{
 		Namespace: "arbdebug",
 		Version:   "1.0",

arbnode/sequencer.go

@@ -124,6 +124,20 @@ func (s *Sequencer) postTxFilter(state *arbosState.ArbosState, tx *types.Transac
 	return nil
 }
 
+func (s *Sequencer) CheckHealth(ctx context.Context) error {
+	s.forwarderMutex.Lock()
+	forwarder := s.forwarder
+	s.forwarderMutex.Unlock()
+	if forwarder != nil {
+		return forwarder.CheckHealth(ctx)
+	}
+
+	if s.txStreamer.coordinator != nil && !s.txStreamer.coordinator.CurrentlyChosen() {
+		return ErrNoSequencer
+	}
+	return nil
+}
+
 func (s *Sequencer) ForwardTarget() string {
 	s.forwarderMutex.Lock()
 	defer s.forwarderMutex.Unlock()

arbnode/tx_pre_checker.go

@@ -142,3 +142,7 @@ func (c *TxPreChecker) PublishTransaction(ctx context.Context, tx *types.Transac
 	}
 	return c.publisher.PublishTransaction(ctx, tx)
 }
+
+func (c *TxPreChecker) CheckHealth(ctx context.Context) error {
+	return c.publisher.CheckHealth(ctx)
+}

cmd/genericconf/server.go

@@ -24,7 +24,7 @@ type HTTPConfig struct {
 var HTTPConfigDefault = HTTPConfig{
 	Addr:           node.DefaultConfig.HTTPHost,
 	Port:           8547,
-	API:            append(node.DefaultConfig.HTTPModules, "eth"),
+	API:            append(node.DefaultConfig.HTTPModules, "eth", "arb"),
 	RPCPrefix:      node.DefaultConfig.HTTPPathPrefix,
 	CORSDomain:     node.DefaultConfig.HTTPCors,
 	VHosts:         node.DefaultConfig.HTTPVirtualHosts,
@@ -89,7 +89,7 @@ type WSConfig struct {
 var WSConfigDefault = WSConfig{
 	Addr:      node.DefaultConfig.WSHost,
 	Port:      8548,
-	API:       append(node.DefaultConfig.WSModules, "eth"),
+	API:       append(node.DefaultConfig.WSModules, "eth", "arb"),
 	RPCPrefix: node.DefaultConfig.WSPathPrefix,
 	Origins:   node.DefaultConfig.WSOrigins,
 	ExposeAll: node.DefaultConfig.WSExposeAll,