Cleanup sector root cache

.changeset/fix_stored_sectors_table_bloat.md

@@ -7,3 +7,5 @@ default: patch
 The cache kept 32 KiB of subtree roots inline on every sector row, which slowed
 sector reads, pruning and contract root lookups. Cached roots are discarded on
 upgrade and rebuilt on the next read.
+
+This database migration will take a long time on large hosts. Plan accordingly.

.changeset/release_cached_sector_roots.md

@@ -0,0 +1,7 @@
+---
+default: patch
+---
+
+# Release cached sector roots of resolved and rejected contracts
+
+Fixes a memory leak where the in-memory sector root cache kept the roots of every resolved, renewed, and rejected contract for the lifetime of the process.

host/contracts/cache.go

@@ -0,0 +1,57 @@
+package contracts
+
+import (
+	"fmt"
+	"slices"
+	"sync"
+
+	"go.sia.tech/core/types"
+)
+
+type rootsCache struct {
+	store ContractStore
+
+	mu                sync.RWMutex // protects the fields below
+	lastExpiredHeight uint64
+	contractSectors   map[types.FileContractID][]types.Hash256
+}
+
+// SectorRoots gets the cached sector roots for the contract
+func (rc *rootsCache) SectorRoots(id types.FileContractID) []types.Hash256 {
+	rc.mu.RLock()
+	defer rc.mu.RUnlock()
+	return slices.Clone(rc.contractSectors[id])
+}
+
+// UpdateSectorRoots replaces the cached sector roots for the given file contract
+func (rc *rootsCache) UpdateSectorRoots(id types.FileContractID, roots []types.Hash256) {
+	rc.mu.Lock()
+	defer rc.mu.Unlock()
+	rc.contractSectors[id] = slices.Clone(roots)
+}
+
+// ExpireContracts removes the cached sector roots of contracts that were
+// rejected or resolved before expireHeight.
+func (rc *rootsCache) ExpireContracts(height uint64) error {
+	rc.mu.RLock()
+	lastExpiredHeight := rc.lastExpiredHeight
+	rc.mu.RUnlock()
+	if height <= lastExpiredHeight {
+		rc.mu.Lock()
+		rc.lastExpiredHeight = height
+		rc.mu.Unlock()
+		return nil
+	}
+
+	expired, err := rc.store.ExpiredV2Contracts(lastExpiredHeight, height)
+	if err != nil {
+		return fmt.Errorf("failed to get expired contracts: %w", err)
+	}
+	rc.mu.Lock()
+	defer rc.mu.Unlock()
+	for _, id := range expired {
+		delete(rc.contractSectors, id)
+	}
+	rc.lastExpiredHeight = height
+	return nil
+}

host/contracts/integrity.go

@@ -80,7 +80,7 @@ func (cm *Manager) CheckIntegrity(ctx context.Context, contractID types.FileCont
 
 	expectedRoots := contract.Revision.Filesize / proto4.SectorSize
 
-	roots := cm.getSectorRoots(contractID)
+	roots := cm.roots.SectorRoots(contractID)
 	if uint64(len(roots)) != expectedRoots {
 		return nil, 0, fmt.Errorf("expected %v sector roots, got %v", expectedRoots, len(roots))
 	} else if calculated := proto4.MetaRoot(roots); contract.Revision.FileMerkleRoot != calculated {

host/contracts/lock.go

@@ -139,7 +139,7 @@ func (cm *Manager) LockV2Contract(id types.FileContractID) (rev rhp4.RevisionSta
 		Revision:  contract.V2FileContract,
 		Renewed:   renewed,
 		Revisable: revisable,
-		Roots:     cm.getSectorRoots(id),
+		Roots:     cm.roots.SectorRoots(id),
 	}
 	return state, func() { cm.locks.Unlock(id) }, nil
 }

host/contracts/manager.go

@@ -4,7 +4,6 @@ import (
 	"errors"
 	"fmt"
 	"math"
-	"sync"
 	"time"
 
 	"go.sia.tech/core/consensus"
@@ -77,10 +76,7 @@ type (
 
 		locks *locker // contracts must be locked while they are being modified
 
-		mu sync.Mutex // guards the following fields
-		// caches the sector roots of all contracts to avoid long reads from
-		// the store
-		sectorRoots map[types.FileContractID][]types.Hash256
+		roots *rootsCache
 	}
 )
 
@@ -89,25 +85,6 @@ var (
 	ErrAlreadyRenewed = errors.New("renewed contracts cannot be revised")
 )
 
-func (cm *Manager) getSectorRoots(id types.FileContractID) []types.Hash256 {
-	cm.mu.Lock()
-	defer cm.mu.Unlock()
-
-	roots, ok := cm.sectorRoots[id]
-	if !ok {
-		return nil
-	}
-	// return a deep copy of the roots
-	return append([]types.Hash256(nil), roots...)
-}
-
-func (cm *Manager) setSectorRoots(id types.FileContractID, roots []types.Hash256) {
-	cm.mu.Lock()
-	defer cm.mu.Unlock()
-	// deep copy the roots
-	cm.sectorRoots[id] = append([]types.Hash256(nil), roots...)
-}
-
 // Contracts returns a paginated list of contracts matching the filter and the
 // total number of contracts matching the filter.
 func (cm *Manager) Contracts(filter ContractFilter) ([]Contract, int, error) {
@@ -161,7 +138,7 @@ func (cm *Manager) RenewContract(renewal SignedRevision, existing SignedRevision
 	defer done()
 
 	// sanity checks
-	existingRoots := cm.getSectorRoots(existing.Revision.ParentID)
+	existingRoots := cm.roots.SectorRoots(existing.Revision.ParentID)
 	if existing.Revision.FileMerkleRoot != (types.Hash256{}) {
 		return errors.New("existing contract must be cleared")
 	} else if existing.Revision.Filesize != 0 {
@@ -177,7 +154,7 @@ func (cm *Manager) RenewContract(renewal SignedRevision, existing SignedRevision
 	if err := cm.store.RenewContract(renewal, existing, formationSet, lockedCollateral, clearingUsage, initialUsage, cm.chain.TipState().Index.Height); err != nil {
 		return err
 	}
-	cm.setSectorRoots(renewal.Revision.ParentID, existingRoots)
+	cm.roots.UpdateSectorRoots(renewal.Revision.ParentID, existingRoots)
 	cm.log.Debug("contract renewed", zap.Stringer("renewalID", renewal.Revision.ParentID), zap.Stringer("existingID", existing.Revision.ParentID))
 	return nil
 }
@@ -203,7 +180,7 @@ func (cm *Manager) ReviseV2Contract(contractID types.FileContractID, revision ty
 		return fmt.Errorf("revision number went backwards: existing=%d revised=%d", existing.RevisionNumber, revision.RevisionNumber)
 	}
 
-	oldRoots := cm.getSectorRoots(contractID)
+	oldRoots := cm.roots.SectorRoots(contractID)
 
 	// validate the contract revision fields
 	switch {
@@ -241,7 +218,7 @@ func (cm *Manager) ReviseV2Contract(contractID types.FileContractID, revision ty
 		return err
 	}
 	// update the sector roots cache
-	cm.setSectorRoots(contractID, newRoots)
+	cm.roots.UpdateSectorRoots(contractID, newRoots)
 	cm.log.Debug("contract revised",
 		zap.Stringer("contractID", contractID),
 		zap.Uint64("previousRevisionNumber", existing.RevisionNumber),
@@ -281,6 +258,7 @@ func (cm *Manager) AddV2Contract(formation rhp4.TransactionSet, usage proto4.Usa
 	if err := cm.store.AddV2Contract(contract, formation); err != nil {
 		return err
 	}
+	cm.roots.UpdateSectorRoots(contractID, nil)
 	cm.log.Debug("contract formed", zap.Stringer("contractID", contractID))
 	return nil
 }
@@ -315,7 +293,7 @@ func (cm *Manager) RenewV2Contract(renewal rhp4.TransactionSet, usage proto4.Usa
 	fc := resolution.NewContract
 
 	existingID := types.FileContractID(existing.ID)
-	existingRoots := cm.getSectorRoots(existingID)
+	existingRoots := cm.roots.SectorRoots(existingID)
 	if fc.FileMerkleRoot != proto4.MetaRoot(existingRoots) {
 		return errors.New("renewal root does not match existing roots")
 	}
@@ -333,14 +311,14 @@ func (cm *Manager) RenewV2Contract(renewal rhp4.TransactionSet, usage proto4.Usa
 	if err := cm.store.RenewV2Contract(contract, renewal, existingID); err != nil {
 		return err
 	}
-	cm.setSectorRoots(contract.ID, existingRoots)
+	cm.roots.UpdateSectorRoots(contract.ID, existingRoots)
 	cm.log.Debug("contract renewed", zap.Stringer("formedID", contract.ID), zap.Stringer("existingID", existingID))
 	return nil
 }
 
 // SectorRoots returns the roots of all sectors stored by the contract.
 func (cm *Manager) SectorRoots(id types.FileContractID) []types.Hash256 {
-	return cm.getSectorRoots(id)
+	return cm.roots.SectorRoots(id)
 }
 
 // Close closes the contract manager.
@@ -385,14 +363,27 @@ func NewManager(store ContractStore, storage StorageManager, chain ChainManager,
 		opt(cm)
 	}
 
-	start := time.Now()
+	tip, err := store.Tip()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get tip: %w", err)
+	}
+	expireHeight := tip.Height
+	if expireHeight > ReorgBuffer {
+		expireHeight -= ReorgBuffer
+	}
 
-	roots, err := store.V2SectorRoots()
+	cm.log.Debug("building sector roots cache")
+	start := time.Now()
+	roots, err := store.V2SectorRoots(expireHeight)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get v2 sector roots: %w", err)
 	}
-
-	cm.sectorRoots = roots
 	cm.log.Debug("loaded sector roots", zap.Duration("elapsed", time.Since(start)))
+
+	cm.roots = &rootsCache{
+		store:             store,
+		contractSectors:   roots,
+		lastExpiredHeight: expireHeight,
+	}
 	return cm, nil
 }