account manager: avoid taking locks for long period of time (#3717)

## Summary

The `DeleteOldKeys` method was taking the lock for the duration of the keys deletion. This is not required, as the mutex really just need to be held to synchronize the list of participation keys. The underlying `OneTimeSignatureSecrets` already have a synchronization lock, which is taken as need.

## Test Plan

Unit test added.

The output of the test help to detect the timing issues addressed in this PR. Before this PR, calling Key() 10 times took 4.1 seconds. With this PR, it takes 255us.

Author: tsachiherman

components/mocks/mockParticipationRegistry.go

@@ -0,0 +1,90 @@
+// Copyright (C) 2019-2022 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand 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 Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package mocks
+
+import (
+	"time"
+
+	"github.com/algorand/go-algorand/data/account"
+	"github.com/algorand/go-algorand/data/basics"
+)
+
+// MockParticipationRegistry is a dummy ParticipationRegistry that doesn't do anything
+type MockParticipationRegistry struct {
+}
+
+// Insert adds a record to storage and computes the ParticipationID
+func (m *MockParticipationRegistry) Insert(record account.Participation) (account.ParticipationID, error) {
+	return account.ParticipationID{}, nil
+}
+
+// AppendKeys appends state proof keys to an existing Participation record. Keys can only be appended
+// once, an error will occur when the data is flushed when inserting a duplicate key.
+func (m *MockParticipationRegistry) AppendKeys(id account.ParticipationID, keys account.StateProofKeys) error {
+	return nil
+}
+
+// Delete removes a record from storage.
+func (m *MockParticipationRegistry) Delete(id account.ParticipationID) error {
+	return nil
+}
+
+// DeleteExpired removes all records from storage which are expired on the given round.
+func (m *MockParticipationRegistry) DeleteExpired(round basics.Round) error {
+	return nil
+}
+
+// Get a participation record.
+func (m *MockParticipationRegistry) Get(id account.ParticipationID) account.ParticipationRecord {
+	return account.ParticipationRecord{}
+}
+
+// GetAll of the participation records.
+func (m *MockParticipationRegistry) GetAll() []account.ParticipationRecord {
+	return []account.ParticipationRecord{}
+}
+
+// GetForRound fetches a record with voting secrets for a particular round.
+func (m *MockParticipationRegistry) GetForRound(id account.ParticipationID, round basics.Round) (account.ParticipationRecordForRound, error) {
+	return account.ParticipationRecordForRound{}, nil
+}
+
+// GetStateProofForRound fetches a record with stateproof secrets for a particular round.
+func (m *MockParticipationRegistry) GetStateProofForRound(id account.ParticipationID, round basics.Round) (account.StateProofRecordForRound, error) {
+	return account.StateProofRecordForRound{}, nil
+}
+
+// Register updates the EffectiveFirst and EffectiveLast fields. If there are multiple records for the account
+// then it is possible for multiple records to be updated.
+func (m *MockParticipationRegistry) Register(id account.ParticipationID, on basics.Round) error {
+	return nil
+}
+
+// Record sets the Last* field for the active ParticipationID for the given account.
+func (m *MockParticipationRegistry) Record(account basics.Address, round basics.Round, participationType account.ParticipationAction) error {
+	return nil
+}
+
+// Flush ensures that all changes have been written to the underlying data store.
+func (m *MockParticipationRegistry) Flush(timeout time.Duration) error {
+	return nil
+}
+
+// Close any resources used to implement the interface.
+func (m *MockParticipationRegistry) Close() {
+
+}

crypto/falconWrapper.go

@@ -109,7 +109,7 @@ func (d *FalconVerifier) GetSignatureFixedLengthHashableRepresentation(signature
 	return ctSignature[:], err
 }
 
-// NewFalconSigner creates a falconSinger that is used to sign and verify falcon signatures
+// NewFalconSigner creates a falconSigner that is used to sign and verify falcon signatures
 func NewFalconSigner() (*FalconSigner, error) {
 	var seed FalconSeed
 	RandBytes(seed[:])

data/accountManager.go

@@ -35,10 +35,12 @@ import (
 type AccountManager struct {
 	mu deadlock.Mutex
 
+	// syncronized by mu
 	partKeys map[account.ParticipationKeyIdentity]account.PersistedParticipation
 
 	// Map to keep track of accounts for which we've sent
 	// AccountRegistered telemetry events
+	// syncronized by mu
 	registeredAccounts map[string]bool
 
 	registry account.ParticipationRegistry
@@ -58,9 +60,6 @@ func MakeAccountManager(log logging.Logger, registry account.ParticipationRegist
 
 // Keys returns a list of Participation accounts, and their keys/secrets for requested round.
 func (manager *AccountManager) Keys(rnd basics.Round) (out []account.ParticipationRecordForRound) {
-	manager.mu.Lock()
-	defer manager.mu.Unlock()
-
 	for _, part := range manager.registry.GetAll() {
 		if part.OverlapsInterval(rnd, rnd) {
 			partRndSecrets, err := manager.registry.GetForRound(part.ParticipationID, rnd)
@@ -76,9 +75,6 @@ func (manager *AccountManager) Keys(rnd basics.Round) (out []account.Participati
 
 // StateProofKeys returns a list of Participation accounts, and their stateproof secrets
 func (manager *AccountManager) StateProofKeys(rnd basics.Round) (out []account.StateProofRecordForRound) {
-	manager.mu.Lock()
-	defer manager.mu.Unlock()
-
 	for _, part := range manager.registry.GetAll() {
 		if part.OverlapsInterval(rnd, rnd) {
 			partRndSecrets, err := manager.registry.GetStateProofForRound(part.ParticipationID, rnd)
@@ -168,37 +164,40 @@ func (manager *AccountManager) DeleteOldKeys(latestHdr bookkeeping.BlockHeader,
 
 	manager.mu.Lock()
 	pendingItems := make(map[string]<-chan error, len(manager.partKeys))
-	func() {
-		defer manager.mu.Unlock()
-		for _, part := range manager.partKeys {
-			// We need a key for round r+1 for agreement.
-			nextRound := latestHdr.Round + 1
-
-			if latestHdr.CompactCert[protocol.CompactCertBasic].CompactCertNextRound > 0 {
-				// We need a key for the next compact cert round.
-				// This would be CompactCertNextRound+1 (+1 because compact
-				// cert code uses the next round's ephemeral key), except
-				// if we already used that key to produce a signature (as
-				// reported in ccSigs).
-				nextCC := latestHdr.CompactCert[protocol.CompactCertBasic].CompactCertNextRound + 1
-				if ccSigs[part.Parent] >= nextCC {
-					nextCC = ccSigs[part.Parent] + basics.Round(latestProto.CompactCertRounds) + 1
-				}
-
-				if nextCC < nextRound {
-					nextRound = nextCC
-				}
-			}
 
-			// we pre-create the reported error string here, so that we won't need to have the participation key object if error is detected.
-			first, last := part.ValidInterval()
-			errString := fmt.Sprintf("AccountManager.DeleteOldKeys(): key for %s (%d-%d), nextRound %d",
-				part.Address().String(), first, last, nextRound)
-			errCh := part.DeleteOldKeys(nextRound, agreementProto)
+	partKeys := make([]account.PersistedParticipation, 0, len(manager.partKeys))
+	for _, part := range manager.partKeys {
+		partKeys = append(partKeys, part)
+	}
+	manager.mu.Unlock()
+	for _, part := range partKeys {
+		// We need a key for round r+1 for agreement.
+		nextRound := latestHdr.Round + 1
+
+		if latestHdr.CompactCert[protocol.CompactCertBasic].CompactCertNextRound > 0 {
+			// We need a key for the next compact cert round.
+			// This would be CompactCertNextRound+1 (+1 because compact
+			// cert code uses the next round's ephemeral key), except
+			// if we already used that key to produce a signature (as
+			// reported in ccSigs).
+			nextCC := latestHdr.CompactCert[protocol.CompactCertBasic].CompactCertNextRound + 1
+			if ccSigs[part.Parent] >= nextCC {
+				nextCC = ccSigs[part.Parent] + basics.Round(latestProto.CompactCertRounds) + 1
+			}
 
-			pendingItems[errString] = errCh
+			if nextCC < nextRound {
+				nextRound = nextCC
+			}
 		}
-	}()
+
+		// we pre-create the reported error string here, so that we won't need to have the participation key object if error is detected.
+		first, last := part.ValidInterval()
+		errString := fmt.Sprintf("AccountManager.DeleteOldKeys(): key for %s (%d-%d), nextRound %d",
+			part.Address().String(), first, last, nextRound)
+		errCh := part.DeleteOldKeys(nextRound, agreementProto)
+
+		pendingItems[errString] = errCh
+	}
 
 	// wait for all disk flushes, and report errors as they appear.
 	for errString, errCh := range pendingItems {

data/accountManager_test.go

@@ -0,0 +1,110 @@
+// Copyright (C) 2019-2022 Algorand, Inc.
+// This file is part of go-algorand
+//
+// go-algorand is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as
+// published by the Free Software Foundation, either version 3 of the
+// License, or (at your option) any later version.
+//
+// go-algorand 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 Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with go-algorand.  If not, see <https://www.gnu.org/licenses/>.
+
+package data
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/algorand/go-algorand/components/mocks"
+	"github.com/algorand/go-algorand/config"
+	"github.com/algorand/go-algorand/data/account"
+	"github.com/algorand/go-algorand/data/basics"
+	"github.com/algorand/go-algorand/data/bookkeeping"
+	"github.com/algorand/go-algorand/logging"
+	"github.com/algorand/go-algorand/protocol"
+	"github.com/algorand/go-algorand/test/partitiontest"
+	"github.com/algorand/go-algorand/util/db"
+)
+
+func TestAccountManagerKeys(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	log := logging.TestingLog(t)
+	log.SetLevel(logging.Error)
+	registry := &mocks.MockParticipationRegistry{}
+
+	acctManager := MakeAccountManager(log, registry)
+
+	databaseFiles := make([]string, 0)
+	defer func() {
+		for _, fileName := range databaseFiles {
+			os.Remove(fileName)
+			os.Remove(fileName + "-shm")
+			os.Remove(fileName + "-wal")
+		}
+	}()
+
+	// create participation keys
+	const numPartKeys = 10
+	for partKeyIdx := 0; partKeyIdx < numPartKeys; partKeyIdx++ {
+		rootFilename := t.Name() + "_root_" + strconv.Itoa(partKeyIdx) + ".sqlite"
+		partFilename := t.Name() + "_part_" + strconv.Itoa(partKeyIdx) + ".sqlite"
+		os.Remove(rootFilename)
+		os.Remove(partFilename)
+		rootAccessor, err := db.MakeAccessor(rootFilename, false, true)
+		require.NoError(t, err)
+
+		root, err := account.GenerateRoot(rootAccessor)
+		require.NoError(t, err)
+
+		accessor, err := db.MakeErasableAccessor(partFilename)
+		require.NoError(t, err)
+		accessor.SetLogger(log)
+
+		part, err := account.FillDBWithParticipationKeys(accessor, root.Address(), 0, 100, 10000)
+		require.NoError(t, err)
+
+		rootAccessor.Close()
+		databaseFiles = append(databaseFiles, rootFilename)
+		databaseFiles = append(databaseFiles, partFilename)
+
+		acctManager.AddParticipation(part)
+	}
+
+	keyDeletionDone := make(chan struct{}, 1)
+	nextRoundCh := make(chan basics.Round, 2)
+	// kick off key deletion thread.
+	go func() {
+		defer close(keyDeletionDone)
+		ccSigs := make(map[basics.Address]basics.Round)
+		agreementProto := config.Consensus[protocol.ConsensusCurrentVersion]
+		header := bookkeeping.BlockHeader{}
+		for rnd := range nextRoundCh {
+			header.Round = rnd
+			acctManager.DeleteOldKeys(header, ccSigs, agreementProto)
+		}
+	}()
+
+	testStartTime := time.Now()
+	keysTotalDuration := time.Duration(0)
+	for i := 1; i < 10; i++ {
+		nextRoundCh <- basics.Round(i)
+		startTime := time.Now()
+		acctManager.Keys(basics.Round(i))
+		keysTotalDuration += time.Since(startTime)
+	}
+	close(nextRoundCh)
+	<-keyDeletionDone
+	testDuration := time.Since(testStartTime)
+	require.Lessf(t, keysTotalDuration, testDuration/100, fmt.Sprintf("the time to aquire the keys via Keys() was %v whereas blocking on keys deletion took %v", keysTotalDuration, testDuration))
+	t.Logf("Calling AccountManager.Keys() while AccountManager.DeleteOldKeys() was busy, 10 times in a row, resulted in accumulated delay of %v\n", keysTotalDuration)
+}

node/node.go

@@ -1010,11 +1010,11 @@ func insertStateProofToRegistry(part account.PersistedParticipation, node *Algor
 		return nil
 	}
 	keys := part.StateProofSecrets.GetAllKeys()
-	keysSinger := make(account.StateProofKeys, len(keys))
+	keysSigner := make(account.StateProofKeys, len(keys))
 	for i := uint64(0); i < uint64(len(keys)); i++ {
-		keysSinger[i] = keys[i]
+		keysSigner[i] = keys[i]
 	}
-	return node.accountManager.Registry().AppendKeys(partID, keysSinger)
+	return node.accountManager.Registry().AppendKeys(partID, keysSigner)
 
 }