Rosetta Implementation - pt2 FIX2 (Stage 3.2 of Node API Overhaul)

core/blockchain.go

@@ -161,13 +161,14 @@ type BlockChain struct {
 	procInterrupt int32          // interrupt signaler for block processing
 	wg            sync.WaitGroup // chain processing wait group for shutting down
 
-	engine         consensus_engine.Engine
-	processor      Processor // block processor interface
-	validator      Validator // block and state validator interface
-	vmConfig       vm.Config
-	badBlocks      *lru.Cache              // Bad block cache
-	shouldPreserve func(*types.Block) bool // Function used to determine whether should preserve the given block.
-	pendingSlashes slash.Records
+	engine                 consensus_engine.Engine
+	processor              Processor // block processor interface
+	validator              Validator // block and state validator interface
+	vmConfig               vm.Config
+	badBlocks              *lru.Cache              // Bad block cache
+	shouldPreserve         func(*types.Block) bool // Function used to determine whether should preserve the given block.
+	pendingSlashes         slash.Records
+	maxGarbCollectedBlkNum int64
 }
 
 // NewBlockChain returns a fully initialised block chain using information
@@ -228,6 +229,7 @@ func NewBlockChain(
 		vmConfig:                      vmConfig,
 		badBlocks:                     badBlocks,
 		pendingSlashes:                slash.Records{},
+		maxGarbCollectedBlkNum:        -1,
 	}
 	bc.SetValidator(NewBlockValidator(chainConfig, bc, engine))
 	bc.SetProcessor(NewStateProcessor(chainConfig, bc, engine))
@@ -1168,6 +1170,9 @@ func (bc *BlockChain) WriteBlockWithState(
 						bc.triegc.Push(root, number)
 						break
 					}
+					if -number > bc.maxGarbCollectedBlkNum {
+						bc.maxGarbCollectedBlkNum = -number
+					}
 					triedb.Dereference(root.(common.Hash))
 				}
 			}
@@ -1202,6 +1207,11 @@ func (bc *BlockChain) WriteBlockWithState(
 	return CanonStatTy, nil
 }
 
+// GetMaxGarbageCollectedBlockNumber ..
+func (bc *BlockChain) GetMaxGarbageCollectedBlockNumber() int64 {
+	return bc.maxGarbCollectedBlkNum
+}
+
 // InsertChain attempts to insert the given batch of blocks in to the canonical
 // chain or, otherwise, create a fork. If an error is returned it will return
 // the index number of the failing block as well an error describing what went
@@ -2021,10 +2031,10 @@ func (bc *BlockChain) ReadPendingCrossLinks() ([]types.CrossLink, error) {
 // WritePendingCrossLinks saves the pending crosslinks
 func (bc *BlockChain) WritePendingCrossLinks(crossLinks []types.CrossLink) error {
 	// deduplicate crosslinks if any
-	m := map[uint32]map[uint64](types.CrossLink){}
+	m := map[uint32]map[uint64]types.CrossLink{}
 	for _, cl := range crossLinks {
 		if _, ok := m[cl.ShardID()]; !ok {
-			m[cl.ShardID()] = map[uint64](types.CrossLink){}
+			m[cl.ShardID()] = map[uint64]types.CrossLink{}
 		}
 		m[cl.ShardID()][cl.BlockNum()] = cl
 	}
@@ -2111,10 +2121,10 @@ func (bc *BlockChain) DeleteFromPendingCrossLinks(crossLinks []types.CrossLink)
 		return 0, err
 	}
 
-	m := map[uint32]map[uint64](struct{}){}
+	m := map[uint32]map[uint64]struct{}{}
 	for _, cl := range crossLinks {
 		if _, ok := m[cl.ShardID()]; !ok {
-			m[cl.ShardID()] = map[uint64](struct{}){}
+			m[cl.ShardID()] = map[uint64]struct{}{}
 		}
 		m[cl.ShardID()][cl.BlockNum()] = struct{}{}
 	}

hmy/hmy.go

@@ -29,9 +29,10 @@ import (
 const (
 	// BloomBitsBlocks is the number of blocks a single bloom bit section vector
 	// contains on the server side.
-	BloomBitsBlocks         uint64 = 4096
-	leaderCacheSize                = 250 // Approx number of BLS keys in committee
-	totalStakeCacheDuration        = 20  // number of blocks where the returned total stake will remain the same
+	BloomBitsBlocks              uint64 = 4096
+	leaderCacheSize                     = 250 // Approx number of BLS keys in committee
+	undelegationPayoutsCacheSize        = 500 // max number of epochs to store in cache
+	totalStakeCacheDuration             = 20  // number of blocks where the returned total stake will remain the same
 )
 
 var (
@@ -64,6 +65,8 @@ type Harmony struct {
 	group singleflight.Group
 	// leaderCache to save on recomputation every epoch.
 	leaderCache *lru.Cache
+	// undelegationPayoutsCache to save on recomputation every epoch
+	undelegationPayoutsCache *lru.Cache
 	// totalStakeCache to save on recomputation for `totalStakeCacheDuration` blocks.
 	totalStakeCache *totalStakeCache
 }
@@ -98,24 +101,26 @@ func New(
 ) *Harmony {
 	chainDb := nodeAPI.Blockchain().ChainDB()
 	leaderCache, _ := lru.New(leaderCacheSize)
+	undelegationPayoutsCache, _ := lru.New(undelegationPayoutsCacheSize)
 	totalStakeCache := newTotalStakeCache(totalStakeCacheDuration)
 	bloomIndexer := NewBloomIndexer(chainDb, params.BloomBitsBlocks, params.BloomConfirms)
 	bloomIndexer.Start(nodeAPI.Blockchain())
 	return &Harmony{
-		ShutdownChan:    make(chan bool),
-		BloomRequests:   make(chan chan *bloombits.Retrieval),
-		BloomIndexer:    bloomIndexer,
-		BlockChain:      nodeAPI.Blockchain(),
-		BeaconChain:     nodeAPI.Beaconchain(),
-		TxPool:          txPool,
-		CxPool:          cxPool,
-		eventMux:        new(event.TypeMux),
-		chainDb:         chainDb,
-		NodeAPI:         nodeAPI,
-		ChainID:         nodeAPI.Blockchain().Config().ChainID.Uint64(),
-		ShardID:         shardID,
-		leaderCache:     leaderCache,
-		totalStakeCache: totalStakeCache,
+		ShutdownChan:             make(chan bool),
+		BloomRequests:            make(chan chan *bloombits.Retrieval),
+		BloomIndexer:             bloomIndexer,
+		BlockChain:               nodeAPI.Blockchain(),
+		BeaconChain:              nodeAPI.Beaconchain(),
+		TxPool:                   txPool,
+		CxPool:                   cxPool,
+		eventMux:                 new(event.TypeMux),
+		chainDb:                  chainDb,
+		NodeAPI:                  nodeAPI,
+		ChainID:                  nodeAPI.Blockchain().Config().ChainID.Uint64(),
+		ShardID:                  shardID,
+		leaderCache:              leaderCache,
+		totalStakeCache:          totalStakeCache,
+		undelegationPayoutsCache: undelegationPayoutsCache,
 	}
 }
 

hmy/staking.go

@@ -7,10 +7,12 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rpc"
 	"github.com/harmony-one/harmony/consensus/quorum"
 	"github.com/harmony-one/harmony/core/rawdb"
 	"github.com/harmony-one/harmony/core/types"
-	internal_common "github.com/harmony-one/harmony/internal/common"
+	"github.com/harmony-one/harmony/internal/chain"
+	internalCommon "github.com/harmony-one/harmony/internal/common"
 	"github.com/harmony-one/harmony/numeric"
 	commonRPC "github.com/harmony-one/harmony/rpc/common"
 	"github.com/harmony-one/harmony/shard"
@@ -22,7 +24,8 @@ import (
 )
 
 var (
-	zero = numeric.ZeroDec()
+	zero    = numeric.ZeroDec()
+	bigZero = big.NewInt(0)
 )
 
 func (hmy *Harmony) readAndUpdateRawStakes(
@@ -125,6 +128,16 @@ func (hmy *Harmony) IsStakingEpoch(epoch *big.Int) bool {
 	return hmy.BlockChain.Config().IsStaking(epoch)
 }
 
+// IsPreStakingEpoch ...
+func (hmy *Harmony) IsPreStakingEpoch(epoch *big.Int) bool {
+	return hmy.BlockChain.Config().IsPreStaking(epoch)
+}
+
+// GetDelegationLockingPeriodInEpoch ...
+func (hmy *Harmony) GetDelegationLockingPeriodInEpoch(epoch *big.Int) int {
+	return chain.GetLockPeriodInEpoch(hmy.BlockChain, epoch)
+}
+
 // SendStakingTx adds a staking transaction
 func (hmy *Harmony) SendStakingTx(ctx context.Context, signedStakingTx *staking.StakingTransaction) error {
 	stx, _, _, _ := rawdb.ReadStakingTransaction(hmy.chainDb, signedStakingTx.Hash())
@@ -252,7 +265,7 @@ func (hmy *Harmony) GetValidatorInformation(
 	bc := hmy.BlockChain
 	wrapper, err := bc.ReadValidatorInformationAt(addr, block.Root())
 	if err != nil {
-		s, _ := internal_common.AddressToBech32(addr)
+		s, _ := internalCommon.AddressToBech32(addr)
 		return nil, errors.Wrapf(err, "not found address in current state %s", s)
 	}
 
@@ -433,6 +446,21 @@ func (hmy *Harmony) GetDelegationsByValidator(validator common.Address) []*staki
 	return delegations
 }
 
+// GetDelegationsByValidatorAtBlock returns all delegation information of a validator at the given block
+func (hmy *Harmony) GetDelegationsByValidatorAtBlock(
+	validator common.Address, block *types.Block,
+) []*staking.Delegation {
+	wrapper, err := hmy.BlockChain.ReadValidatorInformationAt(validator, block.Root())
+	if err != nil || wrapper == nil {
+		return nil
+	}
+	delegations := []*staking.Delegation{}
+	for i := range wrapper.Delegations {
+		delegations = append(delegations, &wrapper.Delegations[i])
+	}
+	return delegations
+}
+
 // GetDelegationsByDelegator returns all delegation information of a delegator
 func (hmy *Harmony) GetDelegationsByDelegator(
 	delegator common.Address,
@@ -471,6 +499,56 @@ func (hmy *Harmony) GetDelegationsByDelegatorByBlock(
 	return addresses, delegations
 }
 
+// UndelegationPayouts ..
+type UndelegationPayouts map[common.Address]*big.Int
+
+// GetUndelegationPayouts returns the undelegation payouts for each delegator
+//
+// Due to in-memory caching, it is possible to get undelegation payouts for a state / epoch
+// that has been pruned but have it be lost (and unable to recompute) after the node restarts.
+// This not a problem if a full (archival) DB is used.
+func (hmy *Harmony) GetUndelegationPayouts(
+	ctx context.Context, epoch *big.Int,
+) (UndelegationPayouts, error) {
+	if !hmy.IsPreStakingEpoch(epoch) {
+		return nil, fmt.Errorf("not pre-staking epoch or later")
+	}
+
+	payouts, ok := hmy.undelegationPayoutsCache.Get(epoch.Uint64())
+	if ok {
+		return payouts.(UndelegationPayouts), nil
+	}
+	undelegationPayouts := UndelegationPayouts{}
+	// require second to last block as saved undelegations are AFTER undelegations are payed out
+	blockNumber := shard.Schedule.EpochLastBlock(epoch.Uint64()) - 1
+	undelegationPayoutBlock, err := hmy.BlockByNumber(ctx, rpc.BlockNumber(blockNumber))
+	if err != nil || undelegationPayoutBlock == nil {
+		// Block not found, so no undelegationPayouts (not an error)
+		return undelegationPayouts, nil
+	}
+
+	lockingPeriod := hmy.GetDelegationLockingPeriodInEpoch(undelegationPayoutBlock.Epoch())
+	for _, validator := range hmy.GetAllValidatorAddresses() {
+		wrapper, err := hmy.BlockChain.ReadValidatorInformationAt(validator, undelegationPayoutBlock.Root())
+		if err != nil || wrapper == nil {
+			continue // Not a validator at this epoch or unable to fetch validator info because of pruned state.
+		}
+		for _, delegation := range wrapper.Delegations {
+			withdraw := delegation.RemoveUnlockedUndelegations(epoch, wrapper.LastEpochInCommittee, lockingPeriod)
+			if withdraw.Cmp(bigZero) == 1 {
+				if totalPayout, ok := undelegationPayouts[delegation.DelegatorAddress]; ok {
+					undelegationPayouts[delegation.DelegatorAddress] = new(big.Int).Add(totalPayout, withdraw)
+				} else {
+					undelegationPayouts[delegation.DelegatorAddress] = withdraw
+				}
+			}
+		}
+	}
+
+	hmy.undelegationPayoutsCache.Add(epoch.Uint64(), undelegationPayouts)
+	return undelegationPayouts, nil
+}
+
 // GetTotalStakingSnapshot ..
 func (hmy *Harmony) GetTotalStakingSnapshot() *big.Int {
 	if stake := hmy.totalStakeCache.pop(hmy.CurrentBlock().NumberU64()); stake != nil {

internal/chain/engine.go

@@ -292,12 +292,7 @@ func payoutUndelegations(
 				"[Finalize] failed to get validator from state to finalize",
 			)
 		}
-		lockPeriod := staking.LockPeriodInEpoch
-		if chain.Config().IsRedelegation(header.Epoch()) {
-			lockPeriod = staking.LockPeriodInEpoch
-		} else if chain.Config().IsQuickUnlock(header.Epoch()) {
-			lockPeriod = staking.LockPeriodInEpochV2
-		}
+		lockPeriod := GetLockPeriodInEpoch(chain, header.Epoch())
 		for i := range wrapper.Delegations {
 			delegation := &wrapper.Delegations[i]
 			totalWithdraw := delegation.RemoveUnlockedUndelegations(
@@ -547,3 +542,14 @@ func GetPublicKeys(
 	}
 	return subCommittee.BLSPublicKeys()
 }
+
+// GetLockPeriodInEpoch returns the delegation lock period for the given chain
+func GetLockPeriodInEpoch(chain engine.ChainReader, epoch *big.Int) int {
+	lockPeriod := staking.LockPeriodInEpoch
+	if chain.Config().IsRedelegation(epoch) {
+		lockPeriod = staking.LockPeriodInEpoch
+	} else if chain.Config().IsQuickUnlock(epoch) {
+		lockPeriod = staking.LockPeriodInEpochV2
+	}
+	return lockPeriod
+}

rosetta/common/errors.go

@@ -1,6 +1,8 @@
 package common
 
 import (
+	"fmt"
+
 	"github.com/coinbase/rosetta-sdk-go/types"
 	"github.com/harmony-one/harmony/rpc"
 )
@@ -65,15 +67,14 @@ var (
 
 // NewError create a new error with a given detail structure
 func NewError(rosettaError types.Error, detailStructure interface{}) *types.Error {
+	newError := rosettaError
 	details, err := rpc.NewStructuredResponse(detailStructure)
 	if err != nil {
-		newError := CatchAllError
-		CatchAllError.Details = map[string]interface{}{
-			"message": err.Error(),
+		newError.Details = map[string]interface{}{
+			"message": fmt.Sprintf("unable to get error details: %v", err.Error()),
 		}
-		return &newError
+	} else {
+		newError.Details = details
 	}
-	newError := rosettaError
-	newError.Details = details
 	return &newError
 }

rosetta/common/operations.go

@@ -21,8 +21,11 @@ const (
 	// GenesisFundsOperation ..
 	GenesisFundsOperation = "Genesis"
 
-	// PreStakingEraBlockRewardOperation ..
-	PreStakingEraBlockRewardOperation = "PreStakingBlockReward"
+	// PreStakingBlockRewardOperation ..
+	PreStakingBlockRewardOperation = "PreOpenStakingBlockReward"
+
+	// UndelegationPayoutOperation ..
+	UndelegationPayoutOperation = "UndelegationPayout"
 )
 
 var (
@@ -33,7 +36,8 @@ var (
 		CrossShardTransferOperation,
 		ContractCreationOperation,
 		GenesisFundsOperation,
-		PreStakingEraBlockRewardOperation,
+		PreStakingBlockRewardOperation,
+		UndelegationPayoutOperation,
 	}
 
 	// StakingOperationTypes ..

rosetta/common/operations_test.go

@@ -54,7 +54,8 @@ func TestPlainOperationTypes(t *testing.T) {
 		CrossShardTransferOperation,
 		ContractCreationOperation,
 		GenesisFundsOperation,
-		PreStakingEraBlockRewardOperation,
+		PreStakingBlockRewardOperation,
+		UndelegationPayoutOperation,
 	}
 	sort.Strings(referenceOperationTypes)
 	sort.Strings(plainOperationTypes)