incentives: cache top online accounts and use when building AbsentParticipationAccounts

Makefile

@@ -290,6 +290,9 @@ $(GOPATH1)/bin/%:
 test: build
 	$(GOTESTCOMMAND) $(GOTAGS) -race $(UNIT_TEST_SOURCES) -timeout 1h -coverprofile=coverage.txt -covermode=atomic
 
+testc:
+	echo $(UNIT_TEST_SOURCES) | xargs -P8 -n1 go test -c
+
 benchcheck: build
 	$(GOTESTCOMMAND) $(GOTAGS) -race $(UNIT_TEST_SOURCES) -run ^NOTHING -bench Benchmark -benchtime 1x -timeout 1h
 

cmd/tealdbg/localLedger.go

@@ -359,6 +359,10 @@ func (l *localLedger) LookupAgreement(rnd basics.Round, addr basics.Address) (ba
 	}, nil
 }
 
+func (l *localLedger) GetKnockOfflineCandidates(basics.Round, config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, nil
+}
+
 func (l *localLedger) OnlineCirculation(rnd basics.Round, voteRound basics.Round) (basics.MicroAlgos, error) {
 	// A constant is fine for tealdbg
 	return basics.Algos(1_000_000_000), nil // 1B

daemon/algod/api/server/v2/dryrun.go

@@ -329,6 +329,10 @@ func (dl *dryrunLedger) LookupAgreement(rnd basics.Round, addr basics.Address) (
 	}, nil
 }
 
+func (dl *dryrunLedger) GetKnockOfflineCandidates(basics.Round, config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, nil
+}
+
 func (dl *dryrunLedger) OnlineCirculation(rnd basics.Round, voteRnd basics.Round) (basics.MicroAlgos, error) {
 	// dryrun doesn't support setting the global online stake, so we'll just return a constant
 	return basics.Algos(1_000_000_000), nil // 1B

data/basics/userBalance.go

@@ -111,7 +111,10 @@ type VotingData struct {
 type OnlineAccountData struct {
 	MicroAlgosWithRewards MicroAlgos
 	VotingData
+
 	IncentiveEligible bool
+	LastProposed      Round
+	LastHeartbeat     Round
 }
 
 // AccountData contains the data associated with a given address.
@@ -561,6 +564,8 @@ func (u AccountData) OnlineAccountData() OnlineAccountData {
 			VoteKeyDilution: u.VoteKeyDilution,
 		},
 		IncentiveEligible: u.IncentiveEligible,
+		LastProposed:      u.LastProposed,
+		LastHeartbeat:     u.LastHeartbeat,
 	}
 }
 

ledger/acctonline.go

@@ -622,11 +622,6 @@ func (ao *onlineAccounts) onlineTotals(rnd basics.Round) (basics.MicroAlgos, pro
 	return basics.MicroAlgos{Raw: onlineRoundParams.OnlineSupply}, onlineRoundParams.CurrentProtocol, nil
 }
 
-// LookupOnlineAccountData returns the online account data for a given address at a given round.
-func (ao *onlineAccounts) LookupOnlineAccountData(rnd basics.Round, addr basics.Address) (data basics.OnlineAccountData, err error) {
-	return ao.lookupOnlineAccountData(rnd, addr)
-}
-
 // roundOffset calculates the offset of the given round compared to the current dbRound. Requires that the lock would be taken.
 func (ao *onlineAccounts) roundOffset(rnd basics.Round) (offset uint64, err error) {
 	if rnd < ao.cachedDBRoundOnline {

ledger/eval/appcow_test.go

@@ -56,6 +56,10 @@ func (ml *emptyLedger) onlineStake() (basics.MicroAlgos, error) {
 	return basics.MicroAlgos{}, nil
 }
 
+func (ml *emptyLedger) knockOfflineCandidates() (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, nil
+}
+
 func (ml *emptyLedger) lookupAppParams(addr basics.Address, aidx basics.AppIndex, cacheOnly bool) (ledgercore.AppParamsDelta, bool, error) {
 	return ledgercore.AppParamsDelta{}, true, nil
 }

ledger/eval/cow.go

@@ -47,6 +47,7 @@ type roundCowParent interface {
 	// lookup retrieves agreement data about an address, querying the ledger if necessary.
 	lookupAgreement(basics.Address) (basics.OnlineAccountData, error)
 	onlineStake() (basics.MicroAlgos, error)
+	knockOfflineCandidates() (map[basics.Address]basics.OnlineAccountData, error)
 
 	// lookupAppParams, lookupAssetParams, lookupAppLocalState, and lookupAssetHolding retrieve data for a given address and ID.
 	// If cacheOnly is set, the ledger DB will not be queried, and only the cache will be consulted.
@@ -192,6 +193,10 @@ func (cb *roundCowState) lookupAgreement(addr basics.Address) (data basics.Onlin
 	return cb.lookupParent.lookupAgreement(addr)
 }
 
+func (cb *roundCowState) knockOfflineCandidates() (map[basics.Address]basics.OnlineAccountData, error) {
+	return cb.lookupParent.knockOfflineCandidates()
+}
+
 func (cb *roundCowState) lookupAppParams(addr basics.Address, aidx basics.AppIndex, cacheOnly bool) (ledgercore.AppParamsDelta, bool, error) {
 	params, ok := cb.mods.Accts.GetAppParams(addr, aidx)
 	if ok {

ledger/eval/cow_test.go

@@ -73,6 +73,10 @@ func (ml *mockLedger) onlineStake() (basics.MicroAlgos, error) {
 	return basics.Algos(55_555), nil
 }
 
+func (ml *mockLedger) knockOfflineCandidates() (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, nil
+}
+
 func (ml *mockLedger) lookupAppParams(addr basics.Address, aidx basics.AppIndex, cacheOnly bool) (ledgercore.AppParamsDelta, bool, error) {
 	params, ok := ml.balanceMap[addr].AppParams[aidx]
 	return ledgercore.AppParamsDelta{Params: &params}, ok, nil // XXX make a copy?

ledger/eval/eval.go

@@ -48,6 +48,7 @@ type LedgerForCowBase interface {
 	CheckDup(config.ConsensusParams, basics.Round, basics.Round, basics.Round, transactions.Txid, ledgercore.Txlease) error
 	LookupWithoutRewards(basics.Round, basics.Address) (ledgercore.AccountData, basics.Round, error)
 	LookupAgreement(basics.Round, basics.Address) (basics.OnlineAccountData, error)
+	GetKnockOfflineCandidates(basics.Round, config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error)
 	LookupAsset(basics.Round, basics.Address, basics.AssetIndex) (ledgercore.AssetResource, error)
 	LookupApplication(basics.Round, basics.Address, basics.AppIndex) (ledgercore.AppResource, error)
 	LookupKv(basics.Round, string) ([]byte, error)
@@ -237,6 +238,10 @@ func (x *roundCowBase) lookupAgreement(addr basics.Address) (basics.OnlineAccoun
 	return ad, err
 }
 
+func (x *roundCowBase) knockOfflineCandidates() (map[basics.Address]basics.OnlineAccountData, error) {
+	return x.l.GetKnockOfflineCandidates(x.rnd, x.proto)
+}
+
 // onlineStake returns the total online stake as of the start of the round. It
 // caches the result to prevent repeated calls to the ledger.
 func (x *roundCowBase) onlineStake() (basics.MicroAlgos, error) {
@@ -1611,20 +1616,68 @@ func (eval *BlockEvaluator) generateKnockOfflineAccountsList() {
 	if !eval.generate {
 		return
 	}
-	current := eval.Round()
 
+	current := eval.Round()
 	maxExpirations := eval.proto.MaxProposedExpiredOnlineAccounts
 	maxSuspensions := eval.proto.Payouts.MaxMarkAbsent
 
 	updates := &eval.block.ParticipationUpdates
 
-	ch := activeChallenge(&eval.proto, uint64(eval.Round()), eval.state)
+	ch := activeChallenge(&eval.proto, uint64(current), eval.state)
+
+	// Make a set of candidate addresses to check for absentee status.
+	type candidateData struct {
+		VoteLastValid         basics.Round
+		VoteID                crypto.OneTimeSignatureVerifier
+		Status                basics.Status
+		LastProposed          basics.Round
+		LastHeartbeat         basics.Round
+		MicroAlgosWithRewards basics.MicroAlgos
+		IncentiveEligible     bool // currently unused below, but may be needed in the future
+	}
+	absentCandidates := make(map[basics.Address]candidateData)
+
+	// First, ask the ledger for the top N online accounts, with their latest
+	// online account data, current up to the previous round.
+	if maxSuspensions > 0 {
+		knockOfflineCandidates, err := eval.state.knockOfflineCandidates()
+		if err != nil {
+			// Log an error and keep going; generating lists of absent and expired
+			// accounts is not required by block validation rules.
+			logging.Base().Warnf("error fetching knockOfflineCandidates: %v", err)
+			knockOfflineCandidates = nil
+		}
+		for accountAddr, acctData := range knockOfflineCandidates {
+			// acctData is from previous block: doesn't include any updates in mods
+			absentCandidates[accountAddr] = candidateData{
+				VoteLastValid:         acctData.VoteLastValid,
+				VoteID:                acctData.VoteID,
+				Status:                basics.Online, // from lookupOnlineAccountData, which only returns online accounts
+				LastProposed:          acctData.LastProposed,
+				LastHeartbeat:         acctData.LastHeartbeat,
+				MicroAlgosWithRewards: acctData.MicroAlgosWithRewards,
+				IncentiveEligible:     acctData.IncentiveEligible,
+			}
+		}
+	}
 
+	// Then add any accounts modified in this block, with their state at the
+	// end of the round.
 	for _, accountAddr := range eval.state.modifiedAccounts() {
 		acctData, found := eval.state.mods.Accts.GetData(accountAddr)
 		if !found {
 			continue
 		}
+		// This will overwrite data from the knockOfflineCandidates() list, if they were modified in the current block.
+		absentCandidates[accountAddr] = candidateData{
+			VoteLastValid:         acctData.VoteLastValid,
+			VoteID:                acctData.VoteID,
+			Status:                acctData.Status,
+			LastProposed:          acctData.LastProposed,
+			LastHeartbeat:         acctData.LastHeartbeat,
+			MicroAlgosWithRewards: acctData.WithUpdatedRewards(eval.proto, eval.state.rewardsLevel()).MicroAlgos,
+			IncentiveEligible:     acctData.IncentiveEligible,
+		}
 
 		// Regardless of being online or suspended, if voting data exists, the
 		// account can be expired to remove it.  This means an offline account
@@ -1637,17 +1690,24 @@ func (eval *BlockEvaluator) generateKnockOfflineAccountsList() {
 					updates.ExpiredParticipationAccounts,
 					accountAddr,
 				)
-				continue // if marking expired, do not also suspend
+				delete(absentCandidates, accountAddr) // if marking expired, do not also suspend
 			}
 		}
+	}
+
+	// Now, check these candidate accounts to see if they are expired or absent.
+	for accountAddr, acctData := range absentCandidates {
+		if acctData.MicroAlgosWithRewards.IsZero() {
+			continue // should only happen in tests; prevents panic in isAbsent
+		}
 
 		if len(updates.AbsentParticipationAccounts) >= maxSuspensions {
 			continue // no more room (don't break the loop, since we may have more expiries)
 		}
 
 		if acctData.Status == basics.Online {
 			lastSeen := max(acctData.LastProposed, acctData.LastHeartbeat)
-			if isAbsent(eval.state.prevTotals.Online.Money, acctData.MicroAlgos, lastSeen, current) ||
+			if isAbsent(eval.state.prevTotals.Online.Money, acctData.MicroAlgosWithRewards, lastSeen, current) ||
 				failsChallenge(ch, accountAddr, lastSeen) {
 				updates.AbsentParticipationAccounts = append(
 					updates.AbsentParticipationAccounts,
@@ -1658,14 +1718,6 @@ func (eval *BlockEvaluator) generateKnockOfflineAccountsList() {
 	}
 }
 
-// delete me in Go 1.21
-func max(a, b basics.Round) basics.Round {
-	if a > b {
-		return a
-	}
-	return b
-}
-
 // bitsMatch checks if the first n bits of two byte slices match. Written to
 // work on arbitrary slices, but we expect that n is small. Only user today
 // calls with n=5.
@@ -1821,6 +1873,9 @@ func (eval *BlockEvaluator) validateAbsentOnlineAccounts() error {
 		if acctData.Status != basics.Online {
 			return fmt.Errorf("proposed absent account %v was %v, not Online", accountAddr, acctData.Status)
 		}
+		if acctData.MicroAlgos.IsZero() {
+			return fmt.Errorf("proposed absent account %v with zero algos", accountAddr)
+		}
 
 		lastSeen := max(acctData.LastProposed, acctData.LastHeartbeat)
 		if isAbsent(eval.state.prevTotals.Online.Money, acctData.MicroAlgos, lastSeen, eval.Round()) {

ledger/eval/eval_test.go

@@ -793,6 +793,17 @@ func (ledger *evalTestLedger) LookupAgreement(rnd basics.Round, addr basics.Addr
 	return convertToOnline(ad), err
 }
 
+func (ledger *evalTestLedger) GetKnockOfflineCandidates(rnd basics.Round, _ config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	// simulate by returning all online accounts known by the test ledger
+	ret := make(map[basics.Address]basics.OnlineAccountData)
+	for addr, data := range ledger.roundBalances[rnd] {
+		if data.Status == basics.Online && !data.MicroAlgos.IsZero() {
+			ret[addr] = data.OnlineAccountData()
+		}
+	}
+	return ret, nil
+}
+
 // OnlineCirculation just returns a deterministic value for a given round.
 func (ledger *evalTestLedger) OnlineCirculation(rnd, voteRound basics.Round) (basics.MicroAlgos, error) {
 	return basics.MicroAlgos{Raw: uint64(rnd) * 1_000_000}, nil
@@ -1025,6 +1036,10 @@ func (l *testCowBaseLedger) LookupAgreement(rnd basics.Round, addr basics.Addres
 	return basics.OnlineAccountData{}, errors.New("not implemented")
 }
 
+func (l *testCowBaseLedger) GetKnockOfflineCandidates(basics.Round, config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, errors.New("not implemented")
+}
+
 func (l *testCowBaseLedger) OnlineCirculation(rnd, voteRnd basics.Round) (basics.MicroAlgos, error) {
 	return basics.MicroAlgos{}, errors.New("not implemented")
 }

ledger/eval/prefetcher/prefetcher_alignment_test.go

@@ -119,16 +119,23 @@ func (l *prefetcherAlignmentTestLedger) LookupWithoutRewards(_ basics.Round, add
 	}
 	return ledgercore.AccountData{}, 0, nil
 }
+
 func (l *prefetcherAlignmentTestLedger) LookupAgreement(_ basics.Round, addr basics.Address) (basics.OnlineAccountData, error) {
 	// prefetch alignment tests do not check for prefetching of online account data
 	// because it's quite different and can only occur in AVM opcodes, which
 	// aren't handled anyway (just as we don't know if a holding or app local
 	// will be accessed in AVM.)
 	return basics.OnlineAccountData{}, errors.New("not implemented")
 }
+
 func (l *prefetcherAlignmentTestLedger) OnlineCirculation(rnd, voteRnd basics.Round) (basics.MicroAlgos, error) {
 	return basics.MicroAlgos{}, errors.New("not implemented")
 }
+
+func (l *prefetcherAlignmentTestLedger) GetKnockOfflineCandidates(basics.Round, config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	return nil, errors.New("not implemented")
+}
+
 func (l *prefetcherAlignmentTestLedger) LookupApplication(rnd basics.Round, addr basics.Address, aidx basics.AppIndex) (ledgercore.AppResource, error) {
 	l.mu.Lock()
 	if l.requestedApps == nil {
@@ -144,6 +151,7 @@ func (l *prefetcherAlignmentTestLedger) LookupApplication(rnd basics.Round, addr
 
 	return l.apps[addr][aidx], nil
 }
+
 func (l *prefetcherAlignmentTestLedger) LookupAsset(rnd basics.Round, addr basics.Address, aidx basics.AssetIndex) (ledgercore.AssetResource, error) {
 	l.mu.Lock()
 	if l.requestedAssets == nil {
@@ -159,9 +167,11 @@ func (l *prefetcherAlignmentTestLedger) LookupAsset(rnd basics.Round, addr basic
 
 	return l.assets[addr][aidx], nil
 }
+
 func (l *prefetcherAlignmentTestLedger) LookupKv(rnd basics.Round, key string) ([]byte, error) {
 	panic("not implemented")
 }
+
 func (l *prefetcherAlignmentTestLedger) GetCreatorForRound(_ basics.Round, cidx basics.CreatableIndex, ctype basics.CreatableType) (basics.Address, bool, error) {
 	l.mu.Lock()
 	if l.requestedCreators == nil {
@@ -175,6 +185,7 @@ func (l *prefetcherAlignmentTestLedger) GetCreatorForRound(_ basics.Round, cidx
 	}
 	return basics.Address{}, false, nil
 }
+
 func (l *prefetcherAlignmentTestLedger) GenesisHash() crypto.Digest {
 	return crypto.Digest{}
 }

ledger/ledger.go

@@ -638,10 +638,39 @@ func (l *Ledger) LookupAgreement(rnd basics.Round, addr basics.Address) (basics.
 	defer l.trackerMu.RUnlock()
 
 	// Intentionally apply (pending) rewards up to rnd.
-	data, err := l.acctsOnline.LookupOnlineAccountData(rnd, addr)
+	data, err := l.acctsOnline.lookupOnlineAccountData(rnd, addr)
 	return data, err
 }
 
+// GetKnockOfflineCandidates retrieves a list of online accounts who will be
+// checked to a recent proposal or heartbeat. Large accounts are the ones worth checking.
+func (l *Ledger) GetKnockOfflineCandidates(rnd basics.Round, proto config.ConsensusParams) (map[basics.Address]basics.OnlineAccountData, error) {
+	l.trackerMu.RLock()
+	defer l.trackerMu.RUnlock()
+
+	// get state proof worker's most recent list for top N addresses
+	if proto.StateProofInterval == 0 {
+		return nil, nil
+	}
+	// get latest state proof voters information, up to rnd, without calling cond.Wait()
+	_, voters := l.acctsOnline.voters.LatestCompletedVotersUpTo(rnd)
+	if voters == nil { // no cached voters found < rnd
+		return nil, nil
+	}
+
+	// fetch fresh data up to this round from online account cache. These accounts should all
+	// be in cache, as long as proto.StateProofTopVoters < onlineAccountsCacheMaxSize.
+	ret := make(map[basics.Address]basics.OnlineAccountData)
+	for addr := range voters.AddrToPos {
+		data, err := l.acctsOnline.lookupOnlineAccountData(rnd, addr)
+		if err != nil {
+			continue // skip missing / not online accounts
+		}
+		ret[addr] = data
+	}
+	return ret, nil
+}
+
 // LookupWithoutRewards is like Lookup but does not apply pending rewards up
 // to the requested round rnd.
 func (l *Ledger) LookupWithoutRewards(rnd basics.Round, addr basics.Address) (ledgercore.AccountData, basics.Round, error) {

ledger/ledgercore/accountdata.go

@@ -187,6 +187,8 @@ func (u AccountData) OnlineAccountData(proto config.ConsensusParams, rewardsLeve
 		MicroAlgosWithRewards: microAlgos,
 		VotingData:            u.VotingData,
 		IncentiveEligible:     u.IncentiveEligible,
+		LastProposed:          u.LastProposed,
+		LastHeartbeat:         u.LastHeartbeat,
 	}
 }
 

ledger/ledgercore/onlineacct.go

@@ -22,7 +22,7 @@ import (
 )
 
 // An OnlineAccount corresponds to an account whose AccountData.Status
-// is Online.  This is used for a Merkle tree commitment of online
+// is Online. This is used for a Merkle tree commitment of online
 // accounts, which is subsequently used to validate participants for
 // a state proof.
 type OnlineAccount struct {

ledger/ledgercore/votersForRound.go

@@ -183,3 +183,11 @@ func (tr *VotersForRound) Wait() error {
 	}
 	return nil
 }
+
+// Completed returns true if the tree has finished being constructed.
+// If there was an error constructing the tree, the error is also returned.
+func (tr *VotersForRound) Completed() (bool, error) {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	return tr.Tree != nil || tr.loadTreeError != nil, tr.loadTreeError
+}