tornadocontrib/infra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

prevent direct access to backend state struct

Browse Source
This commit is contained in:
Felipe Andrade 2023-05-09 02:57:36 -07:00
parent 6302eddef8
commit 32064105c8
1 changed files with 19 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
proxyd/proxyd/consensus_poller.go
View File

@ -203,23 +203,23 @@ func NewConsensusPoller(bg *BackendGroup, opts ...ConsensusOpt) *ConsensusPoller
// UpdateBackend refreshes the consensus state of a single backend // UpdateBackend refreshes the consensus state of a single backend
func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) { func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
bs := cp.backendState[be] _, _, _, _, bannedUntil := cp.getBackendState(be)
if time.Now().Before(bs.bannedUntil) { if time.Now().Before(bannedUntil) {
log.Debug("skipping backend banned", "backend", be.Name, "bannedUntil", bs.bannedUntil) log.Debug("skipping backend banned", "backend", be.Name, "bannedUntil", bannedUntil)
return return
} }
// if backend it not online or not in a health state we'll only resume checkin it after ban // if backend it not online or not in a health state we'll only resume checkin it after ban
if !be.Online() || !be.IsHealthy() { if !be.Online() || !be.IsHealthy() {
log.Warn("backend banned - not online or not healthy", "backend", be.Name, "bannedUntil", bs.bannedUntil) log.Warn("backend banned - not online or not healthy", "backend", be.Name, "bannedUntil", bannedUntil)
bs.bannedUntil = time.Now().Add(cp.banPeriod) bannedUntil = time.Now().Add(cp.banPeriod)
} }
// if backend it not in sync we'll check again after ban // if backend it not in sync we'll check again after ban
inSync, err := cp.isInSync(ctx, be) inSync, err := cp.isInSync(ctx, be)
if err != nil || !inSync { if err != nil || !inSync {
log.Warn("backend banned - not in sync", "backend", be.Name, "bannedUntil", bs.bannedUntil) log.Warn("backend banned - not in sync", "backend", be.Name, "bannedUntil", bannedUntil)
bs.bannedUntil = time.Now().Add(cp.banPeriod) bannedUntil = time.Now().Add(cp.banPeriod)
} }
// if backend exhausted rate limit we'll skip it for now // if backend exhausted rate limit we'll skip it for now
@ -246,7 +246,11 @@ func (cp *ConsensusPoller) UpdateBackend(ctx context.Context, be *Backend) {
if changed { if changed {
RecordBackendLatestBlock(be, latestBlockNumber) RecordBackendLatestBlock(be, latestBlockNumber)
log.Debug("backend state updated", "name", be.Name, "state", bs) log.Debug("backend state updated",
"name", be.Name,
"peerCount", peerCount,
"latestBlockNumber", latestBlockNumber,
"latestBlockHash", latestBlockHash)
} }
} }
@ -258,7 +262,7 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
currentConsensusBlockNumber := cp.GetConsensusBlockNumber() currentConsensusBlockNumber := cp.GetConsensusBlockNumber()
for _, be := range cp.backendGroup.Backends { for _, be := range cp.backendGroup.Backends {
peerCount, backendLatestBlockNumber, backendLatestBlockHash, lastUpdate := cp.getBackendState(be) peerCount, backendLatestBlockNumber, backendLatestBlockHash, lastUpdate, _ := cp.getBackendState(be)
if !be.skipPeerCountCheck && peerCount < cp.minPeerCount { if !be.skipPeerCountCheck && peerCount < cp.minPeerCount {
continue continue
@ -306,10 +310,10 @@ func (cp *ConsensusPoller) UpdateBackendGroupConsensus(ctx context.Context) {
- with minimum peer count - with minimum peer count
- updated recently - updated recently
*/ */
bs := cp.backendState[be] peerCount, _, _, lastUpdate, bannedUntil := cp.getBackendState(be)
notUpdated := bs.lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now()) notUpdated := lastUpdate.Add(cp.maxUpdateThreshold).Before(time.Now())
isBanned := time.Now().Before(bs.bannedUntil) isBanned := time.Now().Before(bannedUntil)
notEnoughPeers := !be.skipPeerCountCheck && bs.peerCount < cp.minPeerCount notEnoughPeers := !be.skipPeerCountCheck && peerCount < cp.minPeerCount
if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers { if !be.IsHealthy() || be.IsRateLimited() || !be.Online() || notUpdated || isBanned || notEnoughPeers {
filteredBackendsNames = append(filteredBackendsNames, be.Name) filteredBackendsNames = append(filteredBackendsNames, be.Name)
continue continue
@ -432,13 +436,14 @@ func (cp *ConsensusPoller) isInSync(ctx context.Context, be *Backend) (result bo
return res, nil return res, nil
} }
func (cp *ConsensusPoller) getBackendState(be *Backend) (peerCount uint64, blockNumber hexutil.Uint64, blockHash string, lastUpdate time.Time) { func (cp *ConsensusPoller) getBackendState(be *Backend) (peerCount uint64, blockNumber hexutil.Uint64, blockHash string, lastUpdate time.Time, bannedUntil time.Time) {
bs := cp.backendState[be] bs := cp.backendState[be]
bs.backendStateMux.Lock() bs.backendStateMux.Lock()
peerCount = bs.peerCount peerCount = bs.peerCount
blockNumber = bs.latestBlockNumber blockNumber = bs.latestBlockNumber
blockHash = bs.latestBlockHash blockHash = bs.latestBlockHash
lastUpdate = bs.lastUpdate lastUpdate = bs.lastUpdate
bannedUntil = bs.bannedUntil
bs.backendStateMux.Unlock() bs.backendStateMux.Unlock()
return return
} }