Algod: state-proof key deletion safety

data/account/participationRegistry.go

@@ -232,7 +232,7 @@ type ParticipationRegistry interface {
 	// once, an error will occur when the data is flushed when inserting a duplicate key.
 	AppendKeys(id ParticipationID, keys StateProofKeys) error
 
-	// DeleteStateProofKeys removes all stateproof keys preceding a given round (including)
+	// DeleteStateProofKeys removes all stateproof keys up to, and not including, a given round
 	DeleteStateProofKeys(id ParticipationID, round basics.Round) error
 
 	// Delete removes a record from storage.
@@ -347,7 +347,7 @@ const (
 	insertKeysetQuery         = `INSERT INTO Keysets (participationID, account, firstValidRound, lastValidRound, keyDilution, vrf, stateProof) VALUES (?, ?, ?, ?, ?, ?, ?)`
 	insertRollingQuery        = `INSERT INTO Rolling (pk, voting) VALUES (?, ?)`
 	appendStateProofKeysQuery = `INSERT INTO StateProofKeys (pk, round, key) VALUES(?, ?, ?)`
-	deleteStateProofKeysQuery = `DELETE FROM StateProofKeys WHERE pk=? AND round<=?`
+	deleteStateProofKeysQuery = `DELETE FROM StateProofKeys WHERE pk=? AND round<?`
 
 	// SELECT pk FROM Keysets WHERE participationID = ?
 	selectPK      = `SELECT pk FROM Keysets WHERE participationID = ? LIMIT 1`

data/accountManager.go

@@ -45,7 +45,8 @@ type AccountManager struct {
 	log      logging.Logger
 }
 
-// DeleteStateProofKey deletes all keys connected to ParticipationID that came before (including) the given round.
+// DeleteStateProofKey deletes keys related to a ParticipationID. The function removes
+// all keys up to, and not including, a given round the given round.
 func (manager *AccountManager) DeleteStateProofKey(id account.ParticipationID, round basics.Round) error {
 	return manager.registry.DeleteStateProofKeys(id, round)
 }
@@ -61,6 +62,19 @@ func MakeAccountManager(log logging.Logger, registry account.ParticipationRegist
 	return manager
 }
 
+// DeleteStateProofKeysForExpiredAccounts removes ephemeral keys for every expired account
+func (manager *AccountManager) DeleteStateProofKeysForExpiredAccounts(currentRound basics.Round) {
+	for _, part := range manager.registry.GetAll() {
+		if currentRound <= part.LastValid {
+			continue
+		}
+		// since DeleteStateProofKeys doesn't remove the last round, we add one to make sure all secrets are being removed
+		if err := manager.DeleteStateProofKey(part.ParticipationID, part.LastValid+1); err != nil {
+			manager.log.Warnf("error while removing state proof keys for participant %v on round %v: %v", part.ParticipationID, part.LastValid+1, err)
+		}
+	}
+}
+
 // Keys returns a list of Participation accounts, and their keys/secrets for requested round.
 func (manager *AccountManager) Keys(rnd basics.Round) (out []account.ParticipationRecordForRound) {
 	for _, part := range manager.registry.GetAll() {
@@ -82,7 +96,7 @@ func (manager *AccountManager) StateProofKeys(rnd basics.Round) (out []account.S
 		if part.StateProof != nil && part.OverlapsInterval(rnd, rnd) {
 			partRndSecrets, err := manager.registry.GetStateProofSecretsForRound(part.ParticipationID, rnd)
 			if err != nil {
-				manager.log.Errorf("error while loading round secrets from participation registry: %v", err)
+				manager.log.Warnf("could not load state proof keys from participation registry: %v", err)
 				continue
 			}
 			out = append(out, partRndSecrets)

data/accountManager_test.go

@@ -250,6 +250,63 @@ func TestAccountManagerOverlappingStateProofKeys(t *testing.T) {
 	a.Equal(1, len(res))
 }
 
+func TestAccountManagerRemoveStateProofKeysForExpiredAccounts(t *testing.T) {
+	partitiontest.PartitionTest(t)
+	a := assert.New(t)
+
+	registry, dbName := getRegistryImpl(t, false, true)
+	defer registryCloseTest(t, registry, dbName)
+
+	log := logging.TestingLog(t)
+	log.SetLevel(logging.Error)
+
+	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")
+			os.Remove(fileName + "-journal")
+		}
+	}()
+
+	store, err := db.MakeAccessor("stateprooftest", false, true)
+	a.NoError(err)
+	root, err := account.GenerateRoot(store)
+	a.NoError(err)
+	part1, err := account.FillDBWithParticipationKeys(store, root.Address(), 0, basics.Round(merklesignature.KeyLifetimeDefault*2), 3)
+	a.NoError(err)
+	store.Close()
+
+	keys1 := part1.StateProofSecrets.GetAllKeys()
+
+	// Add participations to the registry and append StateProof keys as well
+	part1ID, err := acctManager.registry.Insert(part1.Participation)
+	a.NoError(err)
+	err = registry.AppendKeys(part1ID, keys1)
+	a.NoError(err)
+
+	err = acctManager.registry.Flush(10 * time.Second)
+	a.NoError(err)
+
+	for i := 1; i <= 2; i++ {
+		res := acctManager.StateProofKeys(basics.Round(i * merklesignature.KeyLifetimeDefault))
+		a.Equal(1, len(res))
+	}
+
+	acctManager.DeleteStateProofKeysForExpiredAccounts(part1.LastValid + 1)
+	err = acctManager.registry.Flush(10 * time.Second)
+	a.NoError(err)
+
+	for i := 1; i <= 2; i++ {
+		res := acctManager.StateProofKeys(basics.Round(i * merklesignature.KeyLifetimeDefault))
+		a.Equal(0, len(res))
+	}
+
+}
+
 func TestGetStateProofKeysDontLogErrorOnNilStateProof(t *testing.T) {
 	partitiontest.PartitionTest(t)
 	a := assert.New(t)

stateproof/abstractions.go

@@ -55,6 +55,7 @@ type Network interface {
 type Accounts interface {
 	StateProofKeys(basics.Round) []account.StateProofSecretsForRound
 	DeleteStateProofKey(id account.ParticipationID, round basics.Round) error
+	DeleteStateProofKeysForExpiredAccounts(currentRound basics.Round)
 }
 
 // BlockHeaderFetcher captures the aspects of the Ledger that is used to fetch block headers

stateproof/builder.go

@@ -366,8 +366,7 @@ func (spw *Worker) builder(latest basics.Round) {
 			continue
 		}
 
-		spw.deleteOldSigs(&hdr)
-		spw.deleteOldBuilders(&hdr)
+		spw.deleteStaleStateProofBuildData(&hdr)
 
 		// Broadcast signatures based on the previous block(s) that
 		// were agreed upon.  This ensures that, if we send a signature
@@ -450,31 +449,60 @@ func (spw *Worker) broadcastSigs(brnd basics.Round, proto config.ConsensusParams
 	}
 }
 
-func (spw *Worker) deleteOldSigs(currentHdr *bookkeeping.BlockHeader) {
-	oldestRoundToRemove := GetOldestExpectedStateProof(currentHdr)
+func (spw *Worker) deleteStaleStateProofBuildData(currentHdr *bookkeeping.BlockHeader) {
+	proto := config.Consensus[currentHdr.CurrentProtocol]
+	stateProofNextRound := currentHdr.StateProofTracking[protocol.StateProofBasic].StateProofNextRound
+	if proto.StateProofInterval == 0 || stateProofNextRound == 0 {
+		return
+	}
+
+	if spw.LastCleanupRound == stateProofNextRound {
+		return
+	}
+
+	spw.deleteStaleSigs(stateProofNextRound)
+	spw.deleteStaleKeys(stateProofNextRound)
+	spw.deleteStaleBuilders(stateProofNextRound)
+	spw.LastCleanupRound = stateProofNextRound
+}
 
+func (spw *Worker) deleteStaleSigs(retainRound basics.Round) {
 	err := spw.db.Atomic(func(ctx context.Context, tx *sql.Tx) error {
-		return deletePendingSigsBeforeRound(tx, oldestRoundToRemove)
+		return deletePendingSigsBeforeRound(tx, retainRound)
 	})
 	if err != nil {
-		spw.log.Warnf("deletePendingSigsBeforeRound(%d): %v", oldestRoundToRemove, err)
+		spw.log.Warnf("deleteStaleSigs(%d): %v", retainRound, err)
 	}
 }
 
-func (spw *Worker) deleteOldBuilders(currentHdr *bookkeeping.BlockHeader) {
-	oldestRoundToRemove := GetOldestExpectedStateProof(currentHdr)
+func (spw *Worker) deleteStaleKeys(retainRound basics.Round) {
+	keys := spw.accts.StateProofKeys(retainRound)
+	for _, key := range keys {
+		firstRoundAtKeyLifeTime, err := key.StateProofSecrets.FirstRoundInKeyLifetime()
+		if err != nil {
+			spw.log.Errorf("deleteStaleKeys: could not calculate keylifetime for account %v on round %s:  %v", key.ParticipationID, firstRoundAtKeyLifeTime, err)
+			continue
+		}
+		err = spw.accts.DeleteStateProofKey(key.ParticipationID, basics.Round(firstRoundAtKeyLifeTime))
+		if err != nil {
+			spw.log.Warnf("deleteStaleKeys: could not remove key for account %v on round %s: %v", key.ParticipationID, firstRoundAtKeyLifeTime, err)
+		}
+	}
+	spw.accts.DeleteStateProofKeysForExpiredAccounts(retainRound)
+}
 
+func (spw *Worker) deleteStaleBuilders(retainRound basics.Round) {
 	spw.mu.Lock()
 	defer spw.mu.Unlock()
 
 	for rnd := range spw.builders {
-		if rnd < oldestRoundToRemove {
+		if rnd < retainRound {
 			delete(spw.builders, rnd)
 		}
 	}
 
 	err := spw.db.Atomic(func(ctx context.Context, tx *sql.Tx) error {
-		return deleteBuilders(tx, oldestRoundToRemove)
+		return deleteBuilders(tx, retainRound)
 	})
 	if err != nil {
 		spw.log.Warnf("deleteOldBuilders: failed to delete builders from database: %v", err)

stateproof/signer.go

@@ -21,7 +21,6 @@ import (
 
 	"github.com/algorand/go-algorand/config"
 	"github.com/algorand/go-algorand/crypto/merklesignature"
-	"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/protocol"
@@ -117,8 +116,6 @@ func (spw *Worker) signStateProof(hdr bookkeeping.BlockHeader) {
 	}
 
 	sigs := make([]sigFromAddr, 0, len(keys))
-	ids := make([]account.ParticipationID, 0, len(keys))
-	usedSigners := make([]*merklesignature.Signer, 0, len(keys))
 
 	stateproofMessage, err := GenerateStateProofMessage(spw.ledger, uint64(votersHdr.Round), hdr)
 	if err != nil {
@@ -155,30 +152,15 @@ func (spw *Worker) signStateProof(hdr bookkeeping.BlockHeader) {
 			Round:         hdr.Round,
 			Sig:           sig,
 		})
-		ids = append(ids, key.ParticipationID)
-		usedSigners = append(usedSigners, key.StateProofSecrets)
 	}
 
 	// any error in handle sig indicates the signature wasn't stored in disk, thus we cannot delete the key.
-	for i, sfa := range sigs {
+	for _, sfa := range sigs {
 		if _, err := spw.handleSig(sfa, nil); err != nil {
 			spw.log.Warnf("spw.signBlock(%d): handleSig: %v", hdr.Round, err)
 			continue
 		}
 
 		spw.log.Infof("spw.signBlock(%d): sp message was signed with address %v", hdr.Round, sfa.SignerAddress)
-		firstRoundInKeyLifetime, err := usedSigners[i].FirstRoundInKeyLifetime() // Calculate first round of the key in order to delete all previous keys (and keep the current one for now)
-		if err != nil {
-			spw.log.Warnf("spw.signBlock(%d): Signer.FirstRoundInKeyLifetime: %v", hdr.Round, err)
-			continue
-		}
-		if firstRoundInKeyLifetime == 0 {
-			continue // No previous keys to delete (also underflows when subtracting 1)
-		}
-
-		// Safe to delete key for sfa.Round because the signature is now stored in the disk.
-		if err := spw.accts.DeleteStateProofKey(ids[i], basics.Round(firstRoundInKeyLifetime-1)); err != nil { // Subtract 1 to delete all keys up to this one
-			spw.log.Warnf("spw.signBlock(%d): DeleteStateProofKey: %v", hdr.Round, err)
-		}
 	}
 }

stateproof/stateproofMessageGenerator_test.go

@@ -53,6 +53,10 @@ func (s *workerForStateProofMessageTests) DeleteStateProofKey(id account.Partici
 	return s.w.DeleteStateProofKey(id, round)
 }
 
+func (s *workerForStateProofMessageTests) DeleteStateProofKeysForExpiredAccounts(currentRound basics.Round) {
+	s.w.DeleteStateProofKeysForExpiredAccounts(currentRound)
+}
+
 func (s *workerForStateProofMessageTests) Latest() basics.Round {
 	return s.w.Latest()
 }
@@ -149,18 +153,18 @@ func (s *workerForStateProofMessageTests) addBlockWithStateProofHeaders(ccNextRo
 
 func newWorkerForStateProofMessageStubs(keys []account.Participation, totalWeight int) *workerForStateProofMessageTests {
 	s := &testWorkerStubs{
-		t:                     nil,
-		mu:                    deadlock.Mutex{},
-		latest:                0,
-		waiters:               make(map[basics.Round]chan struct{}),
-		waitersCount:          make(map[basics.Round]int),
-		blocks:                make(map[basics.Round]bookkeeping.BlockHeader),
-		keys:                  keys,
-		keysForVoters:         keys,
-		sigmsg:                make(chan []byte, 1024),
-		txmsg:                 make(chan transactions.SignedTxn, 1024),
-		totalWeight:           totalWeight,
-		deletedStateProofKeys: map[account.ParticipationID]basics.Round{},
+		t:                         nil,
+		mu:                        deadlock.Mutex{},
+		latest:                    0,
+		waiters:                   make(map[basics.Round]chan struct{}),
+		waitersCount:              make(map[basics.Round]int),
+		blocks:                    make(map[basics.Round]bookkeeping.BlockHeader),
+		keys:                      keys,
+		keysForVoters:             keys,
+		sigmsg:                    make(chan []byte, 1024),
+		txmsg:                     make(chan transactions.SignedTxn, 1024),
+		totalWeight:               totalWeight,
+		deletedKeysBeforeRoundMap: map[account.ParticipationID]basics.Round{},
 	}
 	sm := workerForStateProofMessageTests{w: s}
 	return &sm

stateproof/worker.go

@@ -65,8 +65,9 @@ type Worker struct {
 	shutdown context.CancelFunc
 	wg       sync.WaitGroup
 
-	signed   basics.Round
-	signedCh chan struct{}
+	signed           basics.Round
+	signedCh         chan struct{}
+	LastCleanupRound basics.Round
 }
 
 // NewWorker constructs a new Worker, as used by the node.