new functionalities for state sync client

api/service/stagedstreamsync/adapter.go

@@ -18,6 +18,8 @@ type syncProtocol interface {
 	GetRawBlocksByNumber(ctx context.Context, bns []uint64, opts ...syncproto.Option) ([][]byte, [][]byte, sttypes.StreamID, error)
 	GetBlockHashes(ctx context.Context, bns []uint64, opts ...syncproto.Option) ([]common.Hash, sttypes.StreamID, error)
 	GetBlocksByHashes(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) ([]*types.Block, sttypes.StreamID, error)
+	GetReceipts(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) (receipts []types.Receipts, stid sttypes.StreamID, err error)
+	GetNodeData(ctx context.Context, hs []common.Hash, opts ...syncproto.Option) (data [][]byte, stid sttypes.StreamID, err error)
 
 	RemoveStream(stID sttypes.StreamID) // If a stream delivers invalid data, remove the stream
 	StreamFailed(stID sttypes.StreamID, reason string)

consensus/engine/consensus_engine.go

@@ -23,6 +23,27 @@ type ChainReader interface {
 	// Config retrieves the blockchain's chain configuration.
 	Config() *params.ChainConfig
 
+	// TrieNode retrieves a blob of data associated with a trie node
+	// either from ephemeral in-memory cache, or from persistent storage.
+	TrieNode(hash common.Hash) ([]byte, error)
+
+	// ContractCode retrieves a blob of data associated with a contract
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ContractCode(hash common.Hash) ([]byte, error)
+
+	// ValidatorCode retrieves a blob of data associated with a validator
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ValidatorCode(hash common.Hash) ([]byte, error)
+
+	// GetReceiptsByHash retrieves the receipts for all transactions in a given block.
+	GetReceiptsByHash(hash common.Hash) types.Receipts
+
 	// CurrentHeader retrieves the current header from the local chain.
 	CurrentHeader() *block.Header
 

core/blockchain.go

@@ -80,6 +80,21 @@ type BlockChain interface {
 	GetBlockByNumber(number uint64) *types.Block
 	// GetReceiptsByHash retrieves the receipts for all transactions in a given block.
 	GetReceiptsByHash(hash common.Hash) types.Receipts
+	// TrieNode retrieves a blob of data associated with a trie node
+	// either from ephemeral in-memory cache, or from persistent storage.
+	TrieNode(hash common.Hash) ([]byte, error)
+	// ContractCode retrieves a blob of data associated with a contract
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ContractCode(hash common.Hash) ([]byte, error)
+	// ValidatorCode retrieves a blob of data associated with a validator
+	// hash either from ephemeral in-memory cache, or from persistent storage.
+	//
+	// If the code doesn't exist in the in-memory cache, check the storage with
+	// new code scheme.
+	ValidatorCode(hash common.Hash) ([]byte, error)
 	// Stop stops the blockchain service. If any imports are currently in progress
 	// it will abort them using the procInterrupt.
 	Stop()

core/blockchain_impl.go

@@ -1098,6 +1098,24 @@ func (bc *BlockChainImpl) GetBlocksFromHash(hash common.Hash, n int) (blocks []*
 	return
 }
 
+// ContractCode retrieves a blob of data associated with a contract
+// hash either from ephemeral in-memory cache, or from persistent storage.
+//
+// If the code doesn't exist in the in-memory cache, check the storage with
+// new code scheme.
+func (bc *BlockChainImpl) ContractCode(hash common.Hash) ([]byte, error) {
+	return bc.stateCache.ContractCode(common.Hash{}, hash)
+}
+
+// ValidatorCode retrieves a blob of data associated with a validator
+// hash either from ephemeral in-memory cache, or from persistent storage.
+//
+// If the code doesn't exist in the in-memory cache, check the storage with
+// new code scheme.
+func (bc *BlockChainImpl) ValidatorCode(hash common.Hash) ([]byte, error) {
+	return bc.stateCache.ValidatorCode(common.Hash{}, hash)
+}
+
 func (bc *BlockChainImpl) GetUnclesInChain(b *types.Block, length int) []*block.Header {
 	uncles := []*block.Header{}
 	for i := 0; b != nil && i < length; i++ {

core/blockchain_stub.go

@@ -62,6 +62,10 @@ func (a Stub) StateAt(common.Hash) (*state.DB, error) {
 	return nil, errors.Errorf("method StateAt not implemented for %s", a.Name)
 }
 
+func (a Stub) TrieNode(hash common.Hash) ([]byte, error) {
+	return []byte{}, errors.Errorf("method TrieNode not implemented for %s", a.Name)
+}
+
 func (a Stub) HasBlock(hash common.Hash, number uint64) bool {
 	return false
 }
@@ -90,6 +94,14 @@ func (a Stub) GetReceiptsByHash(hash common.Hash) types.Receipts {
 	return nil
 }
 
+func (a Stub) ContractCode(hash common.Hash) ([]byte, error) {
+	return []byte{}, nil
+}
+
+func (a Stub) ValidatorCode(hash common.Hash) ([]byte, error) {
+	return []byte{}, nil
+}
+
 func (a Stub) Stop() {
 }
 

hmy/downloader/adapter_test.go

@@ -88,12 +88,16 @@ func (bc *testBlockChain) changeBlockNumber(val uint64) {
 
 func (bc *testBlockChain) ShardID() uint32                                          { return 0 }
 func (bc *testBlockChain) ReadShardState(epoch *big.Int) (*shard.State, error)      { return nil, nil }
+func (bc *testBlockChain) TrieNode(hash common.Hash) ([]byte, error)                { return []byte{}, nil }
 func (bc *testBlockChain) Config() *params.ChainConfig                              { return nil }
 func (bc *testBlockChain) WriteCommitSig(blockNum uint64, lastCommits []byte) error { return nil }
 func (bc *testBlockChain) GetHeader(hash common.Hash, number uint64) *block.Header  { return nil }
 func (bc *testBlockChain) GetHeaderByNumber(number uint64) *block.Header            { return nil }
 func (bc *testBlockChain) GetHeaderByHash(hash common.Hash) *block.Header           { return nil }
 func (bc *testBlockChain) GetBlock(hash common.Hash, number uint64) *types.Block    { return nil }
+func (bc *testBlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts        { return nil }
+func (bc *testBlockChain) ContractCode(hash common.Hash) ([]byte, error)            { return []byte{}, nil }
+func (bc *testBlockChain) ValidatorCode(hash common.Hash) ([]byte, error)           { return []byte{}, nil }
 func (bc *testBlockChain) ReadValidatorList() ([]common.Address, error)             { return nil, nil }
 func (bc *testBlockChain) ReadCommitSig(blockNum uint64) ([]byte, error)            { return nil, nil }
 func (bc *testBlockChain) ReadBlockRewardAccumulator(uint64) (*big.Int, error)      { return nil, nil }

internal/chain/engine_test.go

@@ -323,8 +323,12 @@ func (bc *fakeBlockChain) CurrentHeader() *block.Header {
 func (bc *fakeBlockChain) GetBlock(hash common.Hash, number uint64) *types.Block    { return nil }
 func (bc *fakeBlockChain) GetHeader(hash common.Hash, number uint64) *block.Header  { return nil }
 func (bc *fakeBlockChain) GetHeaderByHash(hash common.Hash) *block.Header           { return nil }
+func (bc *fakeBlockChain) GetReceiptsByHash(hash common.Hash) types.Receipts        { return nil }
+func (bc *fakeBlockChain) ContractCode(hash common.Hash) ([]byte, error)            { return []byte{}, nil }
+func (bc *fakeBlockChain) ValidatorCode(hash common.Hash) ([]byte, error)           { return []byte{}, nil }
 func (bc *fakeBlockChain) ShardID() uint32                                          { return 0 }
 func (bc *fakeBlockChain) ReadShardState(epoch *big.Int) (*shard.State, error)      { return nil, nil }
+func (bc *fakeBlockChain) TrieNode(hash common.Hash) ([]byte, error)                { return []byte{}, nil }
 func (bc *fakeBlockChain) WriteCommitSig(blockNum uint64, lastCommits []byte) error { return nil }
 func (bc *fakeBlockChain) GetHeaderByNumber(number uint64) *block.Header            { return nil }
 func (bc *fakeBlockChain) ReadValidatorList() ([]common.Address, error)             { return nil, nil }

p2p/stream/protocols/sync/chain.go

@@ -16,6 +16,8 @@ type chainHelper interface {
 	getBlockHashes(bns []uint64) []common.Hash
 	getBlocksByNumber(bns []uint64) ([]*types.Block, error)
 	getBlocksByHashes(hs []common.Hash) ([]*types.Block, error)
+	getNodeData(hs []common.Hash) ([][]byte, error)
+	getReceipts(hs []common.Hash) ([]types.Receipts, error)
 }
 
 type chainHelperImpl struct {
@@ -139,3 +141,44 @@ func (ch *chainHelperImpl) getBlockSigFromNextBlock(header *block.Header) []byte
 func (ch *chainHelperImpl) getBlockSigFromDB(header *block.Header) ([]byte, error) {
 	return ch.chain.ReadCommitSig(header.Number().Uint64())
 }
+
+// getNodeData assembles the response to a node data query.
+func (ch *chainHelperImpl) getNodeData(hs []common.Hash) ([][]byte, error) {
+	var (
+		bytes int
+		nodes [][]byte
+	)
+	for _, hash := range hs {
+		// Retrieve the requested state entry
+		entry, err := ch.chain.TrieNode(hash)
+		if len(entry) == 0 || err != nil {
+			// Read the contract/validator code with prefix only to save unnecessary lookups.
+			entry, err = ch.chain.ContractCode(hash)
+			if len(entry) == 0 || err != nil {
+				entry, err = ch.chain.ValidatorCode(hash)
+			}
+		}
+		if err == nil && len(entry) > 0 {
+			nodes = append(nodes, entry)
+			bytes += len(entry)
+		}
+	}
+	return nodes, nil
+}
+
+// getReceipts assembles the response to a receipt query.
+func (ch *chainHelperImpl) getReceipts(hs []common.Hash) ([]types.Receipts, error) {
+	var receipts []types.Receipts
+
+	for i, hash := range hs {
+		// Retrieve the requested block's receipts
+		results := ch.chain.GetReceiptsByHash(hash)
+		if results == nil {
+			if header := ch.chain.GetHeaderByHash(hash); header == nil || header.ReceiptHash() != types.EmptyRootHash {
+				continue
+			}
+		}
+		receipts[i] = append(receipts[i], results...)
+	}
+	return receipts, nil
+}

p2p/stream/protocols/sync/chain_test.go

@@ -46,6 +46,20 @@ func (tch *testChainHelper) getBlocksByHashes(hs []common.Hash) ([]*types.Block,
 	return bs, nil
 }
 
+func (tch *testChainHelper) getNodeData(hs []common.Hash) ([][]byte, error) {
+	data := makeTestNodeData(len(hs))
+	return data, nil
+}
+
+func (tch *testChainHelper) getReceipts(hs []common.Hash) ([]types.Receipts, error) {
+	testReceipts := makeTestReceipts(len(hs), 3)
+	receipts := make([]types.Receipts, len(hs)*3)
+	for i, _ := range hs {
+		receipts[i] = testReceipts
+	}
+	return receipts, nil
+}
+
 func numberToHash(bn uint64) common.Hash {
 	var h common.Hash
 	binary.LittleEndian.PutUint64(h[:], bn)
@@ -90,6 +104,28 @@ func makeTestBlock(bn uint64) *types.Block {
 	return types.NewBlockWithHeader(&block.Header{Header: header})
 }
 
+// makeTestReceipts creates fake node data
+func makeTestNodeData(n int) [][]byte {
+	testData := make([][]byte, n)
+	for i := 0; i < n; i++ {
+		testData[i] = types.EmptyRootHash.Bytes()
+	}
+	return testData
+}
+
+// makeTestReceipts creates fake receipts
+func makeTestReceipts(n int, nPerBlock int) []*types.Receipt {
+	receipts := make([]*types.Receipt, nPerBlock)
+	for i := 0; i < nPerBlock; i++ {
+		receipts[i] = &types.Receipt{
+			Status:            types.ReceiptStatusSuccessful,
+			CumulativeGasUsed: 0x888888888,
+			Logs:              make([]*types.Log, 5),
+		}
+	}
+	return receipts
+}
+
 func decodeBlocksBytes(bbs [][]byte) ([]*types.Block, error) {
 	blocks := make([]*types.Block, 0, len(bbs))