node, eth, les: use ethclient as a contract backend

cmd/utils/flags.go

@@ -19,11 +19,13 @@ package utils
 
 import (
 	"crypto/ecdsa"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"math/big"
 	"os"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"strconv"
 	"strings"
@@ -44,6 +46,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/ethstats"
 	"github.com/ethereum/go-ethereum/les"
@@ -1175,6 +1178,20 @@ func RegisterEthService(stack *node.Node, cfg *eth.Config) {
 			}
 			return fullNode, err
 		})
+		if err == nil {
+			err = stack.RegisterCallback(reflect.TypeOf(&eth.Ethereum{}), func(service node.Service) error {
+				if e, ok := service.(*eth.Ethereum); ok {
+					rpcClient, err := stack.AttachLocked()
+					if err != nil {
+						return err
+					}
+					e.SetClient(ethclient.NewClient(rpcClient))
+					return nil
+				} else {
+					return errors.New("the service given is not the required type")
+				}
+			})
+		}
 	}
 	if err != nil {
 		Fatalf("Failed to register the Ethereum service: %v", err)

contracts/registrar/registrar.go

@@ -32,7 +32,7 @@ var (
 	RegistrarAddr = map[common.Hash]common.Address{
 		// params.MainnetGenesisHash: common.HexToAddress(""),
 		// params.TestnetGenesisHash: common.HexToAddress(""),
-		params.RinkebyGenesisHash: common.HexToAddress("0xaefc742121f79ce9d0953fea7c9cd10f6586179b"),
+		params.RinkebyGenesisHash: common.HexToAddress("0xc72f57e41e2498ad3dab92f665b0f21e2c4f4b79"),
 	}
 )
 

contracts/registrar/registrar_test.go

@@ -37,8 +37,8 @@ var (
 	addr      = crypto.PubkeyToAddress(key.PublicKey)
 	emptyHash = [32]byte{}
 
-	checkpointHash0 = crypto.Keccak256Hash(common.Hex2Bytes("dead0"), common.Hex2Bytes("beef0"), common.Hex2Bytes("deadbeef0"))
+	checkpointHash0 = crypto.Keccak256Hash(common.FromHex("dead0"), common.FromHex("beef0"), common.FromHex("deadbeef0"))
-	checkpointHash1 = crypto.Keccak256Hash(common.Hex2Bytes("dead1"), common.Hex2Bytes("beef1"), common.Hex2Bytes("deadbeef1"))
+	checkpointHash1 = crypto.Keccak256Hash(common.FromHex("dead1"), common.FromHex("beef1"), common.FromHex("deadbeef1"))
 )
 
 // validateOperation executes the operation, watches and delivers all events fired by the backend and ensures the
@@ -243,7 +243,7 @@ func TestCheckpointRegister(t *testing.T) {
 		return nil
 	}, "register stable checkpoint")
 
-	newHash := crypto.Keccak256Hash([]byte("dead00"), []byte("beef00"), []byte("deadbeef00"))
+	newHash := crypto.Keccak256Hash(common.FromHex("dead00"), common.FromHex("beef00"), common.FromHex("deadbeef00"))
 	// Modify the latest checkpoint
 	validateOperation(t, c, contractBackend, func() {
 		c.SetCheckpoint(transactOpts, big.NewInt(0), newHash)
@@ -280,7 +280,7 @@ func TestCheckpointRegister(t *testing.T) {
 	}, "register stable checkpoint 1")
 
 	contractBackend.ShiftBlocks(light.CheckpointConfirmations)
-	newHash1 := crypto.Keccak256Hash([]byte("dead11"), []byte("beef11"), []byte("deadbeef11"))
+	newHash1 := crypto.Keccak256Hash(common.FromHex("dead11"), common.FromHex("beef11"), common.FromHex("deadbeef11"))
 	// Modify the registered checkpoint after a very long time
 	validateOperation(t, c, contractBackend, func() {
 		c.SetCheckpoint(transactOpts, big.NewInt(1), newHash1)

eth/backend.go

@@ -39,6 +39,7 @@ import (
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/eth/filters"
 	"github.com/ethereum/go-ethereum/eth/gasprice"
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/internal/ethapi"
@@ -57,6 +58,7 @@ type LesServer interface {
 	APIs() []rpc.API
 	Protocols() []p2p.Protocol
 	SetBloomBitsIndexer(bbIndexer *core.ChainIndexer)
+	SetClient(*ethclient.Client)
 }
 
 // Ethereum implements the Ethereum full node service.
@@ -100,6 +102,14 @@ func (s *Ethereum) AddLesServer(ls LesServer) {
 	ls.SetBloomBitsIndexer(s.bloomIndexer)
 }
 
+// SetClient sets a rpc client which connecting to our local node.
+func (s *Ethereum) SetClient(client *ethclient.Client) {
+	// Pass the rpc client to les server if it is enabled.
+	if s.lesServer != nil {
+		s.lesServer.SetClient(client)
+	}
+}
+
 // New creates a new Ethereum object (including the
 // initialisation of the common Ethereum object)
 func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {

eth/bind.go

@@ -1,198 +0,0 @@
-// Copyright 2018 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
-
-package eth
-
-import (
-	"context"
-	"math/big"
-
-	"github.com/ethereum/go-ethereum"
-	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/common/hexutil"
-	"github.com/ethereum/go-ethereum/core/types"
-	"github.com/ethereum/go-ethereum/eth/filters"
-	"github.com/ethereum/go-ethereum/event"
-	"github.com/ethereum/go-ethereum/internal/ethapi"
-	"github.com/ethereum/go-ethereum/rlp"
-	"github.com/ethereum/go-ethereum/rpc"
-)
-
-// ContractBackend implements bind.ContractBackend with direct calls to Ethereum
-// internals to support operating on contracts within subprotocols like eth, les and
-// swarm.
-//
-// Internally this backend uses the already exposed API endpoints of the Ethereum
-// object. These should be rewritten to internal Go method calls when the Go API
-// is refactored to support a clean library use.
-type ContractBackend struct {
-	filterBackend filters.Backend                  // Backend used for filter vm logs.
-	events        *filters.EventSystem             // Event system used to watch new fired vm logs.
-	eapi          *ethapi.PublicEthereumAPI        // Wrapper around the Ethereum object to access metadata
-	bcapi         *ethapi.PublicBlockChainAPI      // Wrapper around the blockchain to access chain data
-	txapi         *ethapi.PublicTransactionPoolAPI // Wrapper around the transaction pool to access transaction data
-}
-
-// NewContractBackend creates a new native contract backend using an existing
-// Ethereum object.
-func NewContractBackend(apiBackend ethapi.Backend, filterBackend filters.Backend, lightMode bool) *ContractBackend {
-	return &ContractBackend{
-		filterBackend: filterBackend,
-		events:        filters.NewEventSystem(apiBackend.EventMux(), filterBackend, lightMode),
-		eapi:          ethapi.NewPublicEthereumAPI(apiBackend),
-		bcapi:         ethapi.NewPublicBlockChainAPI(apiBackend),
-		txapi:         ethapi.NewPublicTransactionPoolAPI(apiBackend, new(ethapi.AddrLocker)),
-	}
-}
-
-// CodeAt implements bind.ContractCaller retrieving any code associated
-// with the contract from the local API.
-func (b *ContractBackend) CodeAt(ctx context.Context, contract common.Address, blockNum *big.Int) ([]byte, error) {
-	return b.bcapi.GetCode(ctx, contract, toBlockNumber(blockNum))
-}
-
-// ContractCall implements bind.ContractCaller executing an Ethereum contract
-// call with the specified data as the input. The pending flag requests execution
-// against the pending block, not the stable head of the chain.
-func (b *ContractBackend) CallContract(ctx context.Context, msg ethereum.CallMsg, blockNum *big.Int) ([]byte, error) {
-	out, err := b.bcapi.Call(ctx, toCallArgs(msg), toBlockNumber(blockNum))
-	return out, err
-}
-
-// PendingCodeAt implements bind.ContractTransactor retrieving any code associated
-// with the contract from the local API.
-func (b *ContractBackend) PendingCodeAt(ctx context.Context, contract common.Address) ([]byte, error) {
-	return b.bcapi.GetCode(ctx, contract, rpc.PendingBlockNumber)
-}
-
-// PendingAccountNonce implements bind.ContractTransactor retrieving the current
-// pending nonce associated with an account.
-func (b *ContractBackend) PendingNonceAt(ctx context.Context, account common.Address) (nonce uint64, err error) {
-	out, err := b.txapi.GetTransactionCount(ctx, account, rpc.PendingBlockNumber)
-	if out != nil {
-		nonce = uint64(*out)
-	}
-	return nonce, err
-}
-
-// SuggestGasPrice implements bind.ContractTransactor retrieving the currently
-// suggested gas price to allow a timely execution of a transaction.
-func (b *ContractBackend) SuggestGasPrice(ctx context.Context) (*big.Int, error) {
-	price, err := b.eapi.GasPrice(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return price.ToInt(), nil
-}
-
-// EstimateGasLimit implements bind.ContractTransactor trying to estimate the gas
-// needed to execute a specific transaction based on the current pending state of
-// the backend blockchain. There is no guarantee that this is the true gas limit
-// requirement as other transactions may be added or removed by miners, but it
-// should provide a basis for setting a reasonable default.
-func (b *ContractBackend) EstimateGas(ctx context.Context, msg ethereum.CallMsg) (uint64, error) {
-	gas, err := b.bcapi.EstimateGas(ctx, toCallArgs(msg))
-	return uint64(gas), err
-}
-
-// SendTransaction implements bind.ContractTransactor injecting the transaction
-// into the pending pool for execution.
-func (b *ContractBackend) SendTransaction(ctx context.Context, tx *types.Transaction) error {
-	raw, _ := rlp.EncodeToBytes(tx)
-	_, err := b.txapi.SendRawTransaction(ctx, raw)
-	return err
-}
-
-// FilterLogs implements bind.ContractFilterer returning logs matching the given argument
-// that are stored within the state.
-func (b *ContractBackend) FilterLogs(ctx context.Context, query ethereum.FilterQuery) ([]types.Log, error) {
-	// Initialize unset filter boundaried to run from genesis to chain head
-	from := int64(0)
-	if query.FromBlock != nil {
-		from = query.FromBlock.Int64()
-	}
-	to := int64(-1)
-	if query.ToBlock != nil {
-		to = query.ToBlock.Int64()
-	}
-	// Construct and execute the filter
-	filter := filters.New(b.filterBackend, from, to, query.Addresses, query.Topics)
-
-	logs, err := filter.Logs(ctx)
-	if err != nil {
-		return nil, err
-	}
-	res := make([]types.Log, len(logs))
-	for i, log := range logs {
-		res[i] = *log
-	}
-	return res, nil
-}
-
-// SubscribeFilterLogs implements bind.ContractFilterer watching new fired logs matching the given argument.
-func (b *ContractBackend) SubscribeFilterLogs(ctx context.Context, query ethereum.FilterQuery, ch chan<- types.Log) (ethereum.Subscription, error) {
-	// Subscribe to contract events
-	sink := make(chan []*types.Log)
-	sub, err := b.events.SubscribeLogs(query, sink)
-	if err != nil {
-		return nil, err
-	}
-	// Since we're getting logs in batches, we need to flatten them into a plain stream
-	return event.NewSubscription(func(quit <-chan struct{}) error {
-		defer sub.Unsubscribe()
-		for {
-			select {
-			case logs := <-sink:
-				for _, log := range logs {
-					select {
-					case ch <- *log:
-					case err := <-sub.Err():
-						return err
-					case <-quit:
-						return nil
-					}
-				}
-			case err := <-sub.Err():
-				return err
-			case <-quit:
-				return nil
-			}
-		}
-	}), nil
-}
-
-func toCallArgs(msg ethereum.CallMsg) ethapi.CallArgs {
-	args := ethapi.CallArgs{
-		To:   msg.To,
-		From: msg.From,
-		Data: msg.Data,
-		Gas:  hexutil.Uint64(msg.Gas),
-	}
-	if msg.GasPrice != nil {
-		args.GasPrice = hexutil.Big(*msg.GasPrice)
-	}
-	if msg.Value != nil {
-		args.Value = hexutil.Big(*msg.Value)
-	}
-	return args
-}
-
-func toBlockNumber(num *big.Int) rpc.BlockNumber {
-	if num == nil {
-		return rpc.LatestBlockNumber
-	}
-	return rpc.BlockNumber(num.Int64())
-}

les/server.go

@@ -21,6 +21,8 @@ import (
 	"crypto/ecdsa"
 	"errors"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/contracts/registrar"
@@ -29,6 +31,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/les/flowcontrol"
 	"github.com/ethereum/go-ethereum/light"
@@ -53,9 +56,10 @@ type LesServer struct {
 	privateKey      *ecdsa.PrivateKey
 	quitSync        chan struct{}
 
-	// Checkpoint relative fields
+	// Checkpoint contract relative fields
-	registrar        *registrar.Registrar     // Handler for checkpoint contract
+	genesis   common.Hash          // Genesis block hash for contract address detection
-	stableCheckpoint *light.TrustedCheckpoint // The nearest stable checkpoint
+	registrar *registrar.Registrar // Handler for checkpoint contract, initialized after server is started.
+	watching  int32                // Indicator whether the checkpoint contract is being watched
 
 	// Indexers
 	chtIndexer       *core.ChainIndexer // Indexers for creating cht root for each block section
@@ -83,6 +87,7 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 		lesTopics:        lesTopics,
 		chtIndexer:       light.NewChtIndexer(e.ChainDb(), false),
 		bloomTrieIndexer: light.NewBloomTrieIndexer(e.ChainDb(), false),
+		genesis:          e.BlockChain().Genesis().Hash(),
 	}
 	logger := log.New()
 
@@ -114,13 +119,6 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 	}
 	srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000)
 	srv.fcCostStats = newCostStats(e.ChainDb())
-	if addr, ok := registrar.RegistrarAddr[e.BlockChain().Genesis().Hash()]; ok {
-		registrar, err := registrar.NewRegistrar(addr, eth.NewContractBackend(e.APIBackend, e.APIBackend, false))
-		if err != nil {
-			return nil, err
-		}
-		srv.registrar = registrar
-	}
 	return srv, nil
 }
 
@@ -145,16 +143,28 @@ func (s *LesServer) Start(srvr *p2p.Server) {
 	}
 	s.privateKey = srvr.PrivateKey
 	s.protocolManager.blockLoop()
-	if s.registrar != nil {
-		s.stableCheckpoint = s.recoverCheckpoint()
-		go s.checkpointLoop()
-	}
 }
 
 func (s *LesServer) SetBloomBitsIndexer(bloomIndexer *core.ChainIndexer) {
 	bloomIndexer.AddChildIndexer(s.bloomTrieIndexer)
 }
 
+// SetClient sets the rpc client and starts watching checkpoint contract if it is not yet watched.
+func (s *LesServer) SetClient(client *ethclient.Client) {
+	if addr, ok := registrar.RegistrarAddr[s.genesis]; ok {
+		if !atomic.CompareAndSwapInt32(&s.watching, 0, 1) {
+			return
+		}
+		registrar, err := registrar.NewRegistrar(addr, client)
+		if err != nil {
+			atomic.StoreInt32(&s.watching, 0)
+			return
+		}
+		s.registrar = registrar
+		go s.checkpointLoop(s.recoverCheckpoint())
+	}
+}
+
 // Stop stops the LES service
 func (s *LesServer) Stop() {
 	s.chtIndexer.Close()
@@ -165,6 +175,7 @@ func (s *LesServer) Stop() {
 		<-s.protocolManager.noMorePeers
 	}()
 	s.protocolManager.Stop()
+	atomic.StoreInt32(&s.watching, 0)
 }
 
 // APIs implements LesServer, returns all API service provided by les server.
@@ -215,7 +226,7 @@ func (s *LesServer) getCheckpoint(index uint64) (common.Hash, common.Hash, commo
 }
 
 // checkpointLoop starts a standalone goroutine to watch new checkpoint events and updates local's stable checkpoint.
-func (s *LesServer) checkpointLoop() (err error) {
+func (s *LesServer) checkpointLoop(checkpoint *light.TrustedCheckpoint) (err error) {
 	var (
 		eventCh      = make(chan *contract.ContractNewCheckpointEvent)
 		headCh       = make(chan core.ChainHeadEvent, SubscribeChainHeadEvent)
@@ -230,9 +241,12 @@ func (s *LesServer) checkpointLoop() (err error) {
 		eventSub.Unsubscribe()
 		return errors.New("subscribe head event failed")
 	}
+
+	ticker := time.NewTicker(5 * time.Minute)
 	defer func() {
 		eventSub.Unsubscribe()
 		headSub.Unsubscribe()
+		ticker.Stop()
 	}()
 
 	for {
@@ -243,8 +257,9 @@ func (s *LesServer) checkpointLoop() (err error) {
 				log.Info("Ignore empty checkpoint event")
 				continue
 			}
-			// Note several duplicate events may be received because of chain reorg and the modification of the latest checkpoint.
+			// Note several events have same index may be received because of chain reorg and
-			if s.stableCheckpoint == nil || event.Index.Uint64() >= s.stableCheckpoint.SectionIdx {
+			// the modification of the latest checkpoint.
+			if checkpoint == nil || event.Index.Uint64() > checkpoint.SectionIdx {
 				log.Info("Receive new checkpoint event", "section", event.Index, "hash", common.Hash(event.CheckpointHash).Hex(),
 					"grantor", event.Grantor.Hex())
 				announcement[event.Index.Uint64()] = common.Hash(event.CheckpointHash)
@@ -255,22 +270,29 @@ func (s *LesServer) checkpointLoop() (err error) {
 				continue
 			}
 			idx := (number-light.CheckpointConfirmations)/light.CheckpointFrequency - 1
-			if s.stableCheckpoint == nil || idx > s.stableCheckpoint.SectionIdx {
+			if checkpoint == nil || idx > checkpoint.SectionIdx {
 				hash, ok := announcement[idx]
 				if !ok {
 					continue
 				}
 				sectionHead := s.bloomTrieIndexer.SectionHead(idx)
-				checkpoint := &light.TrustedCheckpoint{
+				c := &light.TrustedCheckpoint{
 					SectionIdx:    idx,
 					SectionHead:   sectionHead,
 					ChtRoot:       light.GetChtV2Root(s.chaindb, idx, sectionHead),
 					BloomTrieRoot: light.GetBloomTrieRoot(s.chaindb, idx, sectionHead),
 				}
-				if checkpoint.HashEqual(common.Hash(hash)) {
+				if c.HashEqual(common.Hash(hash)) {
-					light.WriteTrustedCheckpoint(s.chaindb, checkpoint)
+					light.WriteTrustedCheckpoint(s.chaindb, c)
-					s.stableCheckpoint = checkpoint
+					checkpoint = c
-					log.Info("Update stable checkpoint", "section", checkpoint.SectionIdx)
+					delete(announcement, idx)
+					log.Info("Update stable checkpoint", "section", checkpoint.SectionIdx, "hash", checkpoint.Hash().Hex())
+				}
+			}
+		case <-ticker.C:
+			// Evict useless announcement every 5 minutes.
+			for idx := range announcement {
+				if checkpoint != nil && checkpoint.SectionIdx >= idx {
 					delete(announcement, idx)
 				}
 			}
@@ -286,13 +308,13 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint {
 	var (
 		sectionCnt, _, _ = s.bloomTrieIndexer.Sections()
 		stable           = light.ReadTrustedCheckpoint(s.chaindb)
-		unstableIdx      = sectionCnt - 1
 		headHash         = rawdb.ReadHeadHeaderHash(s.chaindb)
 		headNumber       = rawdb.ReadHeaderNumber(s.chaindb, headHash)
 	)
-	if headNumber == nil {
+	if headNumber == nil || sectionCnt == 0 {
 		return nil
 	}
+	unstableIdx := sectionCnt - 1
 	for stable == nil || stable.SectionIdx < unstableIdx {
 		if (unstableIdx+1)*light.CheckpointFrequency+light.CheckpointConfirmations <= *headNumber {
 			iter, err := s.registrar.FilterNewCheckpointEvent(*headNumber, unstableIdx, light.CheckpointFrequency, light.CheckpointProcessConfirmations)
@@ -308,7 +330,7 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint {
 					if checkpoint.HashEqual(common.Hash(iter.Event.CheckpointHash)) {
 						light.WriteTrustedCheckpoint(s.chaindb, checkpoint)
 						iter.Close()
-						log.Info("Recover checkpoint", "index", checkpoint.SectionIdx)
+						log.Info("Recover checkpoint", "index", checkpoint.SectionIdx, "hash", checkpoint.Hash().Hex())
 						return checkpoint
 					}
 				}
@@ -323,7 +345,7 @@ func (s *LesServer) recoverCheckpoint() *light.TrustedCheckpoint {
 	if stable == nil {
 		log.Info("No stable checkpoint")
 	} else {
-		log.Info("Recover checkpoint", "index", stable.SectionIdx)
+		log.Info("Recover checkpoint", "index", stable.SectionIdx, "hash", stable.Hash().Hex())
 	}
 	return stable
 }

light/checkpoint.go

@@ -84,7 +84,12 @@ func (c *TrustedCheckpoint) HashEqual(hash common.Hash) bool {
 	if c.SectionHead == (common.Hash{}) && c.ChtRoot == (common.Hash{}) && c.BloomTrieRoot == (common.Hash{}) {
 		return hash == common.Hash{}
 	}
-	return crypto.Keccak256Hash(c.SectionHead.Bytes(), c.ChtRoot.Bytes(), c.BloomTrieRoot.Bytes()) == hash
+	return c.Hash() == hash
+}
+
+// Hash returns the hash of checkpoint three key fields(sectionHead, chtRoot and bloomTrieRoot).
+func (c *TrustedCheckpoint) Hash() common.Hash {
+	return crypto.Keccak256Hash(c.SectionHead.Bytes(), c.ChtRoot.Bytes(), c.BloomTrieRoot.Bytes())
 }
 
 var (

node/node.go

@@ -48,8 +48,9 @@ type Node struct {
 	serverConfig p2p.Config
 	server       *p2p.Server // Currently running P2P networking layer
 
-	serviceFuncs []ServiceConstructor     // Service constructors (in dependency order)
+	serviceFuncs []ServiceConstructor               // Service constructors (in dependency order)
-	services     map[reflect.Type]Service // Currently running services
+	callbacks    map[reflect.Type][]ServiceCallback // Service callback functions
+	services     map[reflect.Type]Service           // Currently running services
 
 	rpcAPIs       []rpc.API   // List of APIs currently provided by the node
 	inprocHandler *rpc.Server // In-process RPC request handler to process the API requests
@@ -113,6 +114,7 @@ func New(conf *Config) (*Node, error) {
 		ephemeralKeystore: ephemeralKeystore,
 		config:            conf,
 		serviceFuncs:      []ServiceConstructor{},
+		callbacks:         make(map[reflect.Type][]ServiceCallback),
 		ipcEndpoint:       conf.IPCEndpoint(),
 		httpEndpoint:      conf.HTTPEndpoint(),
 		wsEndpoint:        conf.WSEndpoint(),
@@ -134,6 +136,18 @@ func (n *Node) Register(constructor ServiceConstructor) error {
 	return nil
 }
 
+// RegisterCallback injects a callback function associated with the specified service.
+func (n *Node) RegisterCallback(typ reflect.Type, callback ServiceCallback) error {
+	n.lock.Lock()
+	defer n.lock.Unlock()
+
+	if n.server != nil {
+		return ErrNodeRunning
+	}
+	n.callbacks[typ] = append(n.callbacks[typ], callback)
+	return nil
+}
+
 // Start create a live P2P node and starts running it.
 func (n *Node) Start() error {
 	n.lock.Lock()
@@ -211,7 +225,7 @@ func (n *Node) Start() error {
 		// Mark the service started for potential cleanup
 		started = append(started, kind)
 	}
-	// Lastly start the configured RPC interfaces
+	// Start the configured RPC interfaces
 	if err := n.startRPC(services); err != nil {
 		for _, service := range services {
 			service.Stop()
@@ -223,7 +237,21 @@ func (n *Node) Start() error {
 	n.services = services
 	n.server = running
 	n.stop = make(chan struct{})
-
+	// Lastly invokes all registered services callbacks after server is started.
+	for typ, callbacks := range n.callbacks {
+		if service, ok := services[typ]; ok {
+			for _, callback := range callbacks {
+				if err := callback(service); err != nil {
+					for _, service := range services {
+						service.Stop()
+					}
+					running.Stop()
+					n.services, n.server, n.stop = nil, nil, nil
+					return err
+				}
+			}
+		}
+	}
 	return nil
 }
 
@@ -480,7 +508,12 @@ func (n *Node) Restart() error {
 func (n *Node) Attach() (*rpc.Client, error) {
 	n.lock.RLock()
 	defer n.lock.RUnlock()
+	return n.AttachLocked()
+}
 
+// AttachLocked creates an RPC client attached to an in-process API handler.
+// Note, this function assumes the lock of node is held.
+func (n *Node) AttachLocked() (*rpc.Client, error) {
 	if n.server == nil {
 		return nil, ErrNodeStopped
 	}

node/service.go

@@ -71,6 +71,10 @@ func (ctx *ServiceContext) Service(service interface{}) error {
 // registered for service instantiation.
 type ServiceConstructor func(ctx *ServiceContext) (Service, error)
 
+// ServiceCallback is the function signature of the callbacks needed to be invoked
+// after associated service is started.
+type ServiceCallback func(service Service) error
+
 // Service is an individual protocol that can be registered into a node.
 //
 // Notes: