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 preceding a given round (excluded)
 	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 all keys connected to ParticipationID that came before the given round.
+// The round is excluded.
 func (manager *AccountManager) DeleteStateProofKey(id account.ParticipationID, round basics.Round) error {
 	return manager.registry.DeleteStateProofKeys(id, round)
 }

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,59 @@ 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(latestRoundToKeep basics.Round) {
 	err := spw.db.Atomic(func(ctx context.Context, tx *sql.Tx) error {
-		return deletePendingSigsBeforeRound(tx, oldestRoundToRemove)
+		return deletePendingSigsBeforeRound(tx, latestRoundToKeep)
 	})
 	if err != nil {
-		spw.log.Warnf("deletePendingSigsBeforeRound(%d): %v", oldestRoundToRemove, err)
+		spw.log.Warnf("deletePendingSigsBeforeRound(%d): %v", latestRoundToKeep, err)
 	}
 }
 
-func (spw *Worker) deleteOldBuilders(currentHdr *bookkeeping.BlockHeader) {
-	oldestRoundToRemove := GetOldestExpectedStateProof(currentHdr)
+func (spw *Worker) deleteStaleKeys(latestRoundToKeep basics.Round) {
+	keys := spw.accts.StateProofKeys(latestRoundToKeep)
+	for _, key := range keys {
+		roundToRemove, err := key.StateProofSecrets.FirstRoundInKeyLifetime()
+		if err != nil {
+			spw.log.Errorf("deleteOldKeys: could not calculate keylifetime for account %v on round %s:  %v", key.ParticipationID, roundToRemove, err)
+			continue
+		}
+		err = spw.accts.DeleteStateProofKey(key.ParticipationID, basics.Round(roundToRemove))
+		if err != nil {
+			spw.log.Warnf("deleteOldKeys: could not remove key for account %v on round %s: %v", key.ParticipationID, roundToRemove, err)
+		}
+	}
+}
 
+func (spw *Worker) deleteStaleBuilders(latestRoundToKeep basics.Round) {
 	spw.mu.Lock()
 	defer spw.mu.Unlock()
 
 	for rnd := range spw.builders {
-		if rnd < oldestRoundToRemove {
+		if rnd < latestRoundToKeep {
 			delete(spw.builders, rnd)
 		}
 	}
 
 	err := spw.db.Atomic(func(ctx context.Context, tx *sql.Tx) error {
-		return deleteBuilders(tx, oldestRoundToRemove)
+		return deleteBuilders(tx, latestRoundToKeep)
 	})
 	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

@@ -149,18 +149,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.