From 660e5fd479bf81405af9625933bc015e009dd16b Mon Sep 17 00:00:00 2001 From: Nate Date: Tue, 15 Sep 2026 09:51:56 -0700 Subject: [PATCH 1/5] cleanup sector root cache --- .changeset/release_cached_sector_roots.md | 7 +++ host/contracts/cache.go | 49 +++++++++++++++++++ host/contracts/integrity.go | 2 +- host/contracts/lock.go | 2 +- host/contracts/manager.go | 58 ++++++++++------------- host/contracts/manager_test.go | 24 +++++++++- host/contracts/persist.go | 5 ++ host/contracts/update.go | 8 ++-- persist/sqlite/consensus.go | 48 +++++++++---------- persist/sqlite/contracts.go | 49 +++++++++++++++++-- persist/sqlite/init.sql | 3 ++ persist/sqlite/migrations.go | 21 ++++++++ persist/sqlite/migrations_test.go | 57 ++++++++++++++++++++++ 13 files changed, 265 insertions(+), 68 deletions(-) create mode 100644 .changeset/release_cached_sector_roots.md create mode 100644 host/contracts/cache.go diff --git a/.changeset/release_cached_sector_roots.md b/.changeset/release_cached_sector_roots.md new file mode 100644 index 00000000..28fa7169 --- /dev/null +++ b/.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. diff --git a/host/contracts/cache.go b/host/contracts/cache.go new file mode 100644 index 00000000..41889f7f --- /dev/null +++ b/host/contracts/cache.go @@ -0,0 +1,49 @@ +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.Lock() + defer rc.mu.Unlock() + if height > rc.lastExpiredHeight { + expired, err := rc.store.ExpiredV2Contracts(rc.lastExpiredHeight, height) + if err != nil { + return fmt.Errorf("failed to get expired contracts: %w", err) + } + for _, id := range expired { + delete(rc.contractSectors, id) + } + } + rc.lastExpiredHeight = height + return nil +} diff --git a/host/contracts/integrity.go b/host/contracts/integrity.go index af683f32..72b9ea24 100644 --- a/host/contracts/integrity.go +++ b/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 { diff --git a/host/contracts/lock.go b/host/contracts/lock.go index a2db5b9d..e0fb1711 100644 --- a/host/contracts/lock.go +++ b/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 } diff --git a/host/contracts/manager.go b/host/contracts/manager.go index a11b7263..82351252 100644 --- a/host/contracts/manager.go +++ b/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), @@ -315,7 +292,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 +310,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 +362,27 @@ func NewManager(store ContractStore, storage StorageManager, chain ChainManager, opt(cm) } + cm.log.Debug("building sector roots cache") start := time.Now() - roots, err := store.V2SectorRoots() 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))) + + 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 + } + + cm.roots = &rootsCache{ + store: store, + contractSectors: roots, + lastExpiredHeight: expireHeight, + } return cm, nil } diff --git a/host/contracts/manager_test.go b/host/contracts/manager_test.go index 06b297b3..a1dcee47 100644 --- a/host/contracts/manager_test.go +++ b/host/contracts/manager_test.go @@ -280,6 +280,13 @@ func TestV2ContractLifecycle(t *testing.T) { } } + assertCachedRoots := func(t *testing.T, contractID types.FileContractID, n int) { + t.Helper() + if roots := node.Contracts.SectorRoots(contractID); len(roots) != n { + t.Fatalf("expected %v cached sector roots, got %v", n, len(roots)) + } + } + assertStorageMetrics := func(t *testing.T, contractSectors, physicalSectors uint64) { t.Helper() time.Sleep(2 * time.Second) // wait for the volume manager to prune sectors @@ -371,6 +378,7 @@ func TestV2ContractLifecycle(t *testing.T) { // metrics should not have been updated, contract is still pending assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) assertStorageMetrics(t, 0, 1) + assertCachedRoots(t, contractID, 1) // mine to confirm the contract testutil.MineAndSync(t, node, types.VoidAddress, 1) @@ -386,11 +394,13 @@ func TestV2ContractLifecycle(t *testing.T) { assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) // sector metrics should not change due to the reorg buffer assertStorageMetrics(t, 0, 1) + assertCachedRoots(t, contractID, 1) // mine through the reorg buffer so the sectors will be garbage // collected testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertStorageMetrics(t, 0, 0) + assertCachedRoots(t, contractID, 0) }) t.Run("failed storage proof", func(t *testing.T) { @@ -404,6 +414,7 @@ func TestV2ContractLifecycle(t *testing.T) { // metrics should not have been updated, contract is still pending assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) assertStorageMetrics(t, 0, 1) + assertCachedRoots(t, contractID, 1) // mine to confirm the contract testutil.MineAndSync(t, node, types.VoidAddress, 1) @@ -419,11 +430,13 @@ func TestV2ContractLifecycle(t *testing.T) { assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) // storage metrics will not change due to the reorg buffer assertStorageMetrics(t, 0, 1) + assertCachedRoots(t, contractID, 1) // mine through the reorg buffer so the sectors will be // garbage collected testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertStorageMetrics(t, 0, 0) + assertCachedRoots(t, contractID, 0) }) t.Run("renewal", func(t *testing.T) { @@ -481,6 +494,8 @@ func TestV2ContractLifecycle(t *testing.T) { // not change due to the reorg buffer assertContractMetrics(t, types.Siacoins(22), renewal.RiskedCollateral()) assertStorageMetrics(t, 1, 1) + assertCachedRoots(t, contractID, 1) + assertCachedRoots(t, renewalID, 1) // try to revise the original contract after the renewal is confirmed err = node.Contracts.ReviseV2Contract(contractID, fc, []types.Hash256{}, proto4.Usage{}) @@ -492,6 +507,8 @@ func TestV2ContractLifecycle(t *testing.T) { // garbage collected testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertStorageMetrics(t, 1, 1) + assertCachedRoots(t, contractID, 0) + assertCachedRoots(t, renewalID, 1) // mine until the renewed contract is successful testutil.MineAndSync(t, node, types.VoidAddress, int(renewal.ProofHeight-node.Chain.Tip().Height+1)) @@ -506,6 +523,7 @@ func TestV2ContractLifecycle(t *testing.T) { // collected testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertStorageMetrics(t, 0, 0) + assertCachedRoots(t, renewalID, 0) // try to revise the original contract after the renewal is successful err = node.Contracts.ReviseV2Contract(contractID, fc, []types.Hash256{}, proto4.Usage{}) @@ -731,6 +749,7 @@ func TestV2ContractLifecycle(t *testing.T) { assertContractStatus(t, contractID, contracts.V2ContractStatusPending) assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) assertStorageMetrics(t, 0, 1) + assertCachedRoots(t, contractID, 1) // mine until the contract is rejected testutil.MineAndSync(t, node, types.VoidAddress, 20) @@ -738,6 +757,7 @@ func TestV2ContractLifecycle(t *testing.T) { assertContractStatus(t, contractID, contracts.V2ContractStatusRejected) assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) assertStorageMetrics(t, 0, 0) + assertCachedRoots(t, contractID, 0) }) t.Run("rejected renewal with storage", func(t *testing.T) { @@ -1795,10 +1815,12 @@ func TestV2SectorRootConsistency(t *testing.T) { assertDBRoots(t, contractID, roots) } - testutil.MineAndSync(t, node, types.VoidAddress, int(fc.ExpirationHeight-node.Chain.Tip().Height)+1) + // roots are cached until the resolution is outside the reorg buffer + testutil.MineAndSync(t, node, types.VoidAddress, int(fc.ProofHeight-node.Chain.Tip().Height)+1) assertRoots(t, contractID, roots) testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) + assertRoots(t, contractID, nil) }) t.Run("renewal inherits roots", func(t *testing.T) { diff --git a/host/contracts/persist.go b/host/contracts/persist.go index 6dcb4c4d..4f8aadbf 100644 --- a/host/contracts/persist.go +++ b/host/contracts/persist.go @@ -13,6 +13,8 @@ type ( // ContractActions returns the lifecycle actions for the contract at the // given index. ContractActions(index types.ChainIndex, revisionBroadcastHeight uint64) (LifecycleActions, error) + // Tip returns the last scanned chain index. + Tip() (types.ChainIndex, error) // V2SectorRoots returns the sector roots for all v2 contracts. V2SectorRoots() (map[types.FileContractID][]types.Hash256, error) @@ -52,6 +54,9 @@ type ( // ExpireV2ContractSectors removes sector roots for any v2 contracts that are // rejected or past their proof window. ExpireV2ContractSectors(height uint64) error + // ExpiredV2Contracts returns the IDs of v2 contracts that were + // rejected or resolved at a height in [minHeight, maxHeight). + ExpiredV2Contracts(minHeight, maxHeight uint64) ([]types.FileContractID, error) // RHP4AccountBalance returns the balance of an account. RHP4AccountBalance(proto4.Account) (types.Currency, error) diff --git a/host/contracts/update.go b/host/contracts/update.go index 2290c82a..c8993a4f 100644 --- a/host/contracts/update.go +++ b/host/contracts/update.go @@ -98,7 +98,7 @@ type ( // RejectContracts sets the status of any v1 and v2 contracts with a // negotiation height before the provided height and that have not // been confirmed to rejected - RejectContracts(height uint64) (v1, v2 []types.FileContractID, err error) + RejectContracts(index types.ChainIndex, height uint64) (v1, v2 []types.FileContractID, err error) // AddContractChainIndexElement adds or updates the merkle proof of // chain index state elements @@ -130,7 +130,7 @@ func (cm *Manager) buildV2StorageProof(cs consensus.State, ele V2ProofElement, l sectorIndex := leafIndex / proto4.LeavesPerSector segmentIndex := leafIndex % proto4.LeavesPerSector - roots := cm.getSectorRoots(contractID) + roots := cm.roots.SectorRoots(contractID) contractRoot := proto4.MetaRoot(roots) if contractRoot != revision.FileMerkleRoot { log.Error("unexpected contract root", zap.Stringer("expectedRoot", revision.FileMerkleRoot), zap.Stringer("actualRoot", contractRoot)) @@ -337,6 +337,8 @@ func (cm *Manager) ProcessActions(index types.ChainIndex) error { // 6 block buffer for reorg protection if err := cm.store.ExpireV2ContractSectors(expireHeight); err != nil { return fmt.Errorf("failed to expire v2 contract sectors: %w", err) + } else if err := cm.roots.ExpireContracts(expireHeight); err != nil { + return fmt.Errorf("failed to expire sector roots: %w", err) } return nil } @@ -488,7 +490,7 @@ func (cm *Manager) UpdateChainState(tx UpdateStateTx, reverted []chain.RevertUpd index := cau.State.Index if index.Height >= cm.rejectBuffer { minNegotiationHeight := index.Height - cm.rejectBuffer - rejectedV1, rejectedV2, err := tx.RejectContracts(minNegotiationHeight) + rejectedV1, rejectedV2, err := tx.RejectContracts(index, minNegotiationHeight) if err != nil { return fmt.Errorf("failed to reject contracts: %w", err) } diff --git a/persist/sqlite/consensus.go b/persist/sqlite/consensus.go index aa06a4b6..d2abb3a1 100644 --- a/persist/sqlite/consensus.go +++ b/persist/sqlite/consensus.go @@ -347,15 +347,15 @@ func (ux *updateTx) RevertContracts(index types.ChainIndex, state contracts.Stat } // v2 - if err := revertV2ContractFormation(ux.tx, state.ConfirmedV2); err != nil { + if err := revertV2ContractFormation(ux.tx, index, state.ConfirmedV2); err != nil { return fmt.Errorf("failed to revert v2 contract formation: %w", err) } else if err := applyV2ContractRevision(ux.tx, state.RevisedV2); err != nil { // note: this is correct. The previous revision is being applied return fmt.Errorf("failed to revert v2 contract revisions: %w", err) - } else if err := revertSuccessfulV2Contracts(ux.tx, contracts.V2ContractStatusSuccessful, state.SuccessfulV2); err != nil { + } else if err := revertSuccessfulV2Contracts(ux.tx, index, contracts.V2ContractStatusSuccessful, state.SuccessfulV2); err != nil { return fmt.Errorf("failed to revert v2 successful resolution: %w", err) - } else if err := revertSuccessfulV2Contracts(ux.tx, contracts.V2ContractStatusRenewed, state.RenewedV2); err != nil { + } else if err := revertSuccessfulV2Contracts(ux.tx, index, contracts.V2ContractStatusRenewed, state.RenewedV2); err != nil { return fmt.Errorf("failed to revert v2 renewed resolution: %w", err) - } else if err := revertFailedV2Contracts(ux.tx, state.FailedV2); err != nil { + } else if err := revertFailedV2Contracts(ux.tx, index, state.FailedV2); err != nil { return fmt.Errorf("failed to revert v2 failure resolution: %w", err) } return nil @@ -363,7 +363,7 @@ func (ux *updateTx) RevertContracts(index types.ChainIndex, state contracts.Stat // RejectContracts returns any contracts with a negotiation height // before the provided height that have not been confirmed. -func (ux *updateTx) RejectContracts(height uint64) (rejected []types.FileContractID, rejectedV2 []types.FileContractID, err error) { +func (ux *updateTx) RejectContracts(index types.ChainIndex, height uint64) (rejected []types.FileContractID, rejectedV2 []types.FileContractID, err error) { log := ux.tx.log.Named("RejectContracts").With(zap.Uint64("height", height)) rejected, err = rejectContracts(ux.tx, height, log.Named("v1")) @@ -371,7 +371,7 @@ func (ux *updateTx) RejectContracts(height uint64) (rejected []types.FileContrac return nil, nil, fmt.Errorf("failed to reject v1 contracts: %w", err) } - rejectedV2, err = rejectV2Contracts(ux.tx, height, log.Named("v2")) + rejectedV2, err = rejectV2Contracts(ux.tx, index, height, log.Named("v2")) if err != nil { return nil, nil, fmt.Errorf("failed to get rejected v2 contracts: %w", err) } @@ -1298,7 +1298,7 @@ func applyV2ContractFormation(tx *txn, index types.ChainIndex, confirmed []types } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=$1, contract_status=$2 WHERE id=$3`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=$1, contract_status=$2, last_updated_height=$3, last_updated_block_id=$4 WHERE id=$5`) if err != nil { return fmt.Errorf("failed to prepare update status statement: %w", err) } @@ -1326,7 +1326,7 @@ func applyV2ContractFormation(tx *txn, index types.ChainIndex, confirmed []types } // update the contract table with the confirmation index and new status. - res, err := updateStmt.Exec(encode(index), contracts.V2ContractStatusActive, state.ID) + res, err := updateStmt.Exec(encode(index), contracts.V2ContractStatusActive, index.Height, encode(index.ID), state.ID) if err != nil { return fmt.Errorf("failed to update state %q: %w", fce.ID, err) } else if n, err := res.RowsAffected(); err != nil { @@ -1350,7 +1350,7 @@ func applyV2ContractFormation(tx *txn, index types.ChainIndex, confirmed []types // revertV2ContractFormation reverts the contract formation by setting the // confirmation index to null and the status to pending. -func revertV2ContractFormation(tx *txn, reverted []types.V2FileContractElement) error { +func revertV2ContractFormation(tx *txn, index types.ChainIndex, reverted []types.V2FileContractElement) error { if len(reverted) == 0 { return nil } @@ -1373,7 +1373,7 @@ func revertV2ContractFormation(tx *txn, reverted []types.V2FileContractElement) } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=NULL, contract_status=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1410,7 +1410,7 @@ func revertV2ContractFormation(tx *txn, reverted []types.V2FileContractElement) } // set the contract status to pending - if res, err := updateStmt.Exec(contracts.V2ContractStatusPending, state.ID); err != nil { + if res, err := updateStmt.Exec(contracts.V2ContractStatusPending, index.Height, encode(index.ID), state.ID); err != nil { return fmt.Errorf("failed to revert contract formation %q: %w", fce.ID, err) } else if n, err := res.RowsAffected(); err != nil { return fmt.Errorf("failed to get rows affected: %w", err) @@ -1470,7 +1470,7 @@ func applySuccessfulV2Contracts(tx *txn, index types.ChainIndex, status contract } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1508,7 +1508,7 @@ func applySuccessfulV2Contracts(tx *txn, index types.ChainIndex, status contract } // update the contract's resolution index and status - if res, err := updateStmt.Exec(encode(index.ID), index.Height, status, state.ID); err != nil { + if res, err := updateStmt.Exec(encode(index.ID), index.Height, status, index.Height, encode(index.ID), state.ID); err != nil { return fmt.Errorf("failed to update contract %q: %w", contractID, err) } else if n, err := res.RowsAffected(); err != nil { return fmt.Errorf("failed to get rows affected: %w", err) @@ -1555,7 +1555,7 @@ func applyFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.File } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1594,7 +1594,7 @@ func applyFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.File } // update the contract's resolution index and status - if res, err := updateStmt.Exec(encode(index.ID), index.Height, contracts.V2ContractStatusFailed, state.ID); err != nil { + if res, err := updateStmt.Exec(encode(index.ID), index.Height, contracts.V2ContractStatusFailed, index.Height, encode(index.ID), state.ID); err != nil { return fmt.Errorf("failed to update contract %q: %w", contractID, err) } else if n, err := res.RowsAffected(); err != nil { return fmt.Errorf("failed to get rows affected: %w", err) @@ -1632,7 +1632,7 @@ func applyFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.File // revertSuccessfulV2Contracts clears the resolution index, sets the status to // active and updates the revenue metrics. -func revertSuccessfulV2Contracts(tx *txn, status contracts.V2ContractStatus, successful []types.FileContractID) error { +func revertSuccessfulV2Contracts(tx *txn, index types.ChainIndex, status contracts.V2ContractStatus, successful []types.FileContractID) error { if len(successful) == 0 { return nil } @@ -1643,7 +1643,7 @@ func revertSuccessfulV2Contracts(tx *txn, status contracts.V2ContractStatus, suc } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1674,7 +1674,7 @@ func revertSuccessfulV2Contracts(tx *txn, status contracts.V2ContractStatus, suc } // update the contract's resolution index and status - if res, err := updateStmt.Exec(contracts.V2ContractStatusActive, state.ID); err != nil { + if res, err := updateStmt.Exec(contracts.V2ContractStatusActive, index.Height, encode(index.ID), state.ID); err != nil { return fmt.Errorf("failed to update contract %q: %w", contractID, err) } else if n, err := res.RowsAffected(); err != nil { return fmt.Errorf("failed to get rows affected: %w", err) @@ -1704,7 +1704,7 @@ func revertSuccessfulV2Contracts(tx *txn, status contracts.V2ContractStatus, suc // revertFailedV2Contracts sets the contract status to active and adds the // potential revenue and collateral metrics. -func revertFailedV2Contracts(tx *txn, failed []types.FileContractID) error { +func revertFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.FileContractID) error { if len(failed) == 0 { return nil } @@ -1715,7 +1715,7 @@ func revertFailedV2Contracts(tx *txn, failed []types.FileContractID) error { } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1749,7 +1749,7 @@ func revertFailedV2Contracts(tx *txn, failed []types.FileContractID) error { } // update the contract's resolution index and status - if res, err := updateStmt.Exec(contracts.V2ContractStatusActive, state.ID); err != nil { + if res, err := updateStmt.Exec(contracts.V2ContractStatusActive, index.Height, encode(index.ID), state.ID); err != nil { return fmt.Errorf("failed to update contract %q: %w", contractID, err) } else if n, err := res.RowsAffected(); err != nil { return fmt.Errorf("failed to get rows affected: %w", err) @@ -1967,7 +1967,7 @@ func resetRejectedPoolFunding(tx *txn, contractDBID int64, log *zap.Logger) erro return nil } -func rejectV2Contracts(tx *txn, height uint64, log *zap.Logger) (rejected []types.FileContractID, err error) { +func rejectV2Contracts(tx *txn, index types.ChainIndex, height uint64, log *zap.Logger) (rejected []types.FileContractID, err error) { rejected, err = v2ContractsToReject(tx, height) if err != nil { return nil, fmt.Errorf("failed to get rejected v2 contracts: %w", err) @@ -1981,7 +1981,7 @@ func rejectV2Contracts(tx *txn, height uint64, log *zap.Logger) (rejected []type } defer stateDone() - updateStatus, err := tx.Prepare(`UPDATE contracts_v2 SET contract_status=? WHERE id=?`) + updateStatus, err := tx.Prepare(`UPDATE contracts_v2 SET contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) if err != nil { return nil, fmt.Errorf("failed to prepare v2 update statement: %w", err) } @@ -2012,7 +2012,7 @@ func rejectV2Contracts(tx *txn, height uint64, log *zap.Logger) (rejected []type } // update metrics - if _, err := updateStatus.Exec(contracts.V2ContractStatusRejected, state.ID); err != nil { + if _, err := updateStatus.Exec(contracts.V2ContractStatusRejected, index.Height, encode(index.ID), state.ID); err != nil { return nil, fmt.Errorf("failed to update contract status: %w", err) } else if err := updateV2StatusMetrics(state.Status, contracts.V2ContractStatusRejected, incrementNumericStat); err != nil { return nil, fmt.Errorf("failed to update contract metrics: %w", err) diff --git a/persist/sqlite/contracts.go b/persist/sqlite/contracts.go index 974072c4..b7a23d1f 100644 --- a/persist/sqlite/contracts.go +++ b/persist/sqlite/contracts.go @@ -448,6 +448,25 @@ func (s *Store) ExpireV2ContractSectors(height uint64) error { } } +// ExpiredV2Contracts returns the IDs of v2 contracts that were rejected or +// resolved at a height in [minHeight, maxHeight). +func (s *Store) ExpiredV2Contracts(minHeight, maxHeight uint64) (ids []types.FileContractID, err error) { + err = s.transaction(func(tx *txn) error { + const query = `SELECT contract_id FROM contracts_v2 +WHERE contract_status IN ($1, $2, $3, $4) AND last_updated_height >= $5 AND last_updated_height < $6` + rows, err := tx.Query(query, contracts.V2ContractStatusRejected, contracts.V2ContractStatusSuccessful, contracts.V2ContractStatusFailed, contracts.V2ContractStatusRenewed, minHeight, maxHeight) + if err != nil { + return fmt.Errorf("failed to query contracts: %w", err) + } + ids, err = collectRows(rows, func(s scanner) (id types.FileContractID, err error) { + err = s.Scan(decode(&id)) + return id, err + }) + return err + }) + return +} + func getContract(tx *txn, contractID int64) (contracts.Contract, error) { const query = `SELECT c.contract_id, rt.contract_id AS renewed_to, rf.contract_id AS renewed_from, c.contract_status, c.negotiation_height, c.formation_confirmed, COALESCE(c.revision_number=c.confirmed_revision_number, false) AS revision_confirmed, c.resolution_height, c.locked_collateral, c.rpc_revenue, @@ -522,9 +541,15 @@ LIMIT $3)` // updateResolvedV2Contract clears a contract and returns its ID func updateResolvedV2Contract(tx *txn, contractID types.FileContractID, renewedDBID int64) (dbID int64, err error) { - const clearQuery = `UPDATE contracts_v2 SET renewed_to=$1 WHERE contract_id=$2 RETURNING id;` + index, err := lastScannedIndex(tx) + if err != nil { + return 0, fmt.Errorf("failed to get last scanned index: %w", err) + } + const clearQuery = `UPDATE contracts_v2 SET renewed_to=$1, last_updated_height=$2, last_updated_block_id=$3 WHERE contract_id=$4 RETURNING id;` err = tx.QueryRow(clearQuery, renewedDBID, + index.Height, + encode(index.ID), encode(contractID), ).Scan(&dbID) return @@ -864,17 +889,28 @@ func v2ContractRoots(tx *txn, contractMapID, contractMapRevision int64, maxSecto }) } +func lastScannedIndex(tx *txn) (index types.ChainIndex, err error) { + err = tx.QueryRow(`SELECT last_scanned_index FROM global_settings`).Scan(decodeNullable(&index)) + return +} + func insertV2Contract(tx *txn, contract contracts.V2Contract, mapID, mapRevisionNumber int64, formationSet rhp4.TransactionSet) (dbID int64, err error) { const query = `INSERT INTO contracts_v2 (contract_id, renter_id, locked_collateral, rpc_revenue, storage_revenue, ingress_revenue, egress_revenue, account_funding, risked_collateral, revision_number, negotiation_height, proof_height, expiration_height, formation_txn_set, -formation_txn_set_basis, raw_revision, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number) VALUES - ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) RETURNING id;` +formation_txn_set_basis, raw_revision, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number, +last_updated_height, last_updated_block_id) VALUES + ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) RETURNING id;` renterID, err := renterDBID(tx, contract.RenterPublicKey) if err != nil { return 0, fmt.Errorf("failed to get renter id: %w", err) } + index, err := lastScannedIndex(tx) + if err != nil { + return 0, fmt.Errorf("failed to get last scanned index: %w", err) + } + err = tx.QueryRow(query, encode(contract.ID), renterID, @@ -896,6 +932,8 @@ formation_txn_set_basis, raw_revision, contract_status, sector_count, contract_v contract.V2FileContract.Filesize/proto4.SectorSize, mapID, mapRevisionNumber, + index.Height, + encode(index.ID), ).Scan(&dbID) return dbID, err } @@ -993,7 +1031,10 @@ func reviseV2Contract(tx *txn, id types.FileContractID, revision types.V2FileCon return 0, fmt.Errorf("revision number went backwards: existing=%d revised=%d", existingRevision, revision.RevisionNumber) } - if _, err := tx.Exec(`UPDATE contracts_v2 SET raw_revision=?, revision_number=?, sector_count=? WHERE id=?`, encode(revision), encode(revision.RevisionNumber), revision.Filesize/proto4.SectorSize, contractDBID); err != nil { + index, err := lastScannedIndex(tx) + if err != nil { + return 0, fmt.Errorf("failed to get last scanned index: %w", err) + } else if _, err := tx.Exec(`UPDATE contracts_v2 SET raw_revision=?, revision_number=?, sector_count=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`, encode(revision), encode(revision.RevisionNumber), revision.Filesize/proto4.SectorSize, index.Height, encode(index.ID), contractDBID); err != nil { return 0, fmt.Errorf("failed to update contract: %w", err) } else if err := updateV2ContractUsage(tx, contractDBID, usage); err != nil { return 0, fmt.Errorf("failed to update contract usage: %w", err) diff --git a/persist/sqlite/init.sql b/persist/sqlite/init.sql index 05252498..5ad9e8bb 100644 --- a/persist/sqlite/init.sql +++ b/persist/sqlite/init.sql @@ -183,6 +183,8 @@ CREATE TABLE contracts_v2 ( resolution_height INTEGER CHECK((resolution_height IS NULL) = (resolution_block_id IS NULL)), -- null if the resolution has not been confirmed on the blockchain contract_status TEXT NOT NULL, sector_count INTEGER NOT NULL, -- used for cleanup + last_updated_height INTEGER NOT NULL DEFAULT 0, + last_updated_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000', contract_v2_roots_map_id INTEGER NOT NULL, contract_v2_roots_map_revision_number INTEGER NOT NULL, @@ -199,6 +201,7 @@ CREATE INDEX contracts_v2_contract_status ON contracts_v2(contract_status); CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_proof_height ON contracts_v2(confirmation_index, resolution_block_id, proof_height); CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_expiration_height ON contracts_v2(confirmation_index, resolution_block_id, expiration_height); CREATE INDEX contracts_v2_resolution_height ON contracts_v2(resolution_height); +CREATE INDEX contracts_v2_contract_status_last_updated_height ON contracts_v2(contract_status, last_updated_height); CREATE INDEX contracts_v2_confirmation_index_proof_height ON contracts_v2(confirmation_index, proof_height); CREATE INDEX contracts_v2_confirmation_index_negotiation_height ON contracts_v2(confirmation_index, negotiation_height); CREATE INDEX contracts_v2_roots_map_id_contract_v2_roots_map_revision_number ON contracts_v2(contract_v2_roots_map_id, contract_v2_roots_map_revision_number); diff --git a/persist/sqlite/migrations.go b/persist/sqlite/migrations.go index 0c76a530..afa061e9 100644 --- a/persist/sqlite/migrations.go +++ b/persist/sqlite/migrations.go @@ -13,6 +13,26 @@ import ( "go.uber.org/zap" ) +// migrateVersion56 adds the last updated index to contracts_v2. Resolved +// contracts are set to their resolution index, all other contracts to the last +// scanned index. +func migrateVersion56(tx *txn, _ *zap.Logger) error { + var index types.ChainIndex + if err := tx.QueryRow(`SELECT last_scanned_index FROM global_settings`).Scan(decodeNullable(&index)); err != nil { + return fmt.Errorf("failed to get last scanned index: %w", err) + } + _, err := tx.Exec(` +ALTER TABLE contracts_v2 ADD COLUMN last_updated_height INTEGER NOT NULL DEFAULT 0; +ALTER TABLE contracts_v2 ADD COLUMN last_updated_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000'; +UPDATE contracts_v2 SET last_updated_height=resolution_height, last_updated_block_id=resolution_block_id WHERE resolution_height IS NOT NULL; +CREATE INDEX contracts_v2_contract_status_last_updated_height ON contracts_v2(contract_status, last_updated_height);`) + if err != nil { + return fmt.Errorf("failed to add last updated columns: %w", err) + } + _, err = tx.Exec(`UPDATE contracts_v2 SET last_updated_height=$1, last_updated_block_id=$2 WHERE resolution_height IS NULL`, index.Height, encode(index.ID)) + return err +} + // migrateVersion55 adds a reference count to stored_sectors maintained by // triggers on the contract and temp storage root tables, replaces the last // access timestamp with volume_sector_locks and indexes unreferenced sectors. @@ -1612,4 +1632,5 @@ var migrations = []func(tx *txn, log *zap.Logger) error{ migrateVersion53, migrateVersion54, migrateVersion55, + migrateVersion56, } diff --git a/persist/sqlite/migrations_test.go b/persist/sqlite/migrations_test.go index a1d27f79..c8013674 100644 --- a/persist/sqlite/migrations_test.go +++ b/persist/sqlite/migrations_test.go @@ -767,6 +767,63 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $ // TestMigrateV55 ensures the migration from version 54 to 55 computes the // reference count of existing sectors and that pruning works afterwards. +func TestMigrateV56(t *testing.T) { + log := zaptest.NewLogger(t) + fp := filepath.Join(t.TempDir(), "hostd.sqlite3") + store := initDBVersion(t, fp, 55, log) + + scanned := types.ChainIndex{Height: 200, ID: frand.Entropy256()} + resolution := types.ChainIndex{Height: 150, ID: frand.Entropy256()} + resolvedID, activeID := types.FileContractID(frand.Entropy256()), types.FileContractID(frand.Entropy256()) + + // populate the pre-migration schema directly + err := store.transaction(func(tx *txn) error { + if _, err := tx.Exec(`UPDATE global_settings SET last_scanned_index=$1`, encode(scanned)); err != nil { + return err + } else if _, err := tx.Exec(`INSERT INTO contract_renters (id, public_key) VALUES (1, $1)`, encode(types.GeneratePrivateKey().PublicKey())); err != nil { + return err + } else if _, err := tx.Exec(`INSERT INTO contract_v2_roots_map (id, revision_number) VALUES (1, 0), (2, 0)`); err != nil { + return err + } + + const query = `INSERT INTO contracts_v2 (contract_id, renter_id, revision_number, formation_txn_set, formation_txn_set_basis, +locked_collateral, rpc_revenue, storage_revenue, ingress_revenue, egress_revenue, account_funding, risked_collateral, raw_revision, +negotiation_height, proof_height, expiration_height, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number, +resolution_block_id, resolution_height) VALUES ($1, 1, $2, $3, $4, $5, $5, $5, $5, $5, $5, $5, $6, 100, 140, 160, $7, 0, $8, 0, $9, $10)` + insert := func(id types.FileContractID, mapID int64, status contracts.V2ContractStatus, resolutionID, resolutionHeight any) error { + _, err := tx.Exec(query, encode(id), encode(uint64(0)), []byte{}, encode(types.ChainIndex{}), encode(types.ZeroCurrency), encode(types.V2FileContract{}), status, mapID, resolutionID, resolutionHeight) + return err + } + if err := insert(resolvedID, 1, contracts.V2ContractStatusSuccessful, encode(resolution.ID), resolution.Height); err != nil { + return err + } + return insert(activeID, 2, contracts.V2ContractStatusActive, nil, nil) + }) + if err != nil { + t.Fatal(err) + } else if err := store.Close(); err != nil { + t.Fatal(err) + } + + store, err = OpenDatabase(fp, log) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + assertLastUpdated := func(t *testing.T, id types.FileContractID, expected types.ChainIndex) { + t.Helper() + var index types.ChainIndex + if err := store.readerDB.QueryRow(`SELECT last_updated_height, last_updated_block_id FROM contracts_v2 WHERE contract_id=$1`, encode(id)).Scan(&index.Height, decode(&index.ID)); err != nil { + t.Fatal(err) + } else if index != expected { + t.Fatalf("expected last updated index %v for %v, got %v", expected, id, index) + } + } + assertLastUpdated(t, resolvedID, resolution) + assertLastUpdated(t, activeID, scanned) +} + func TestMigrateV55(t *testing.T) { log := zaptest.NewLogger(t) fp := filepath.Join(t.TempDir(), "hostd.sqlite3") From d51f2330ba0ccffbcfcaaef5cedf8fb3ea4fc5aa Mon Sep 17 00:00:00 2001 From: Nate Date: Tue, 15 Sep 2026 10:20:40 -0700 Subject: [PATCH 2/5] address comments --- host/contracts/manager.go | 1 + host/contracts/manager_test.go | 116 ++++++++++++++++++++++++++++++ persist/sqlite/migrations_test.go | 114 ++++++++++++++--------------- 3 files changed, 174 insertions(+), 57 deletions(-) diff --git a/host/contracts/manager.go b/host/contracts/manager.go index 82351252..07ea9ae0 100644 --- a/host/contracts/manager.go +++ b/host/contracts/manager.go @@ -258,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 } diff --git a/host/contracts/manager_test.go b/host/contracts/manager_test.go index a1dcee47..472988cc 100644 --- a/host/contracts/manager_test.go +++ b/host/contracts/manager_test.go @@ -760,6 +760,122 @@ func TestV2ContractLifecycle(t *testing.T) { assertCachedRoots(t, contractID, 0) }) + t.Run("rejected with storage resubmitted", func(t *testing.T) { + cm := node.Chain + c := node.Contracts + w := node.Wallet + + renterFunds, hostFunds := types.Siacoins(10), types.Siacoins(20) + duration := uint64(10) + cs := cm.TipState() + fc := types.V2FileContract{ + RevisionNumber: 0, + Filesize: 0, + Capacity: 0, + FileMerkleRoot: types.Hash256{}, + ProofHeight: cs.Index.Height + duration, + ExpirationHeight: cs.Index.Height + duration + 10, + RenterOutput: types.SiacoinOutput{ + Value: renterFunds, + Address: w.Address(), + }, + HostOutput: types.SiacoinOutput{ + Value: hostFunds, + Address: w.Address(), + }, + MissedHostValue: hostFunds, + TotalCollateral: hostFunds, + RenterPublicKey: renterKey.PublicKey(), + HostPublicKey: hostKey.PublicKey(), + } + fundAmount := cs.V2FileContractTax(fc).Add(hostFunds).Add(renterFunds) + sigHash := cs.ContractSigHash(fc) + fc.HostSignature = hostKey.SignHash(sigHash) + fc.RenterSignature = renterKey.SignHash(sigHash) + + txn := types.V2Transaction{ + FileContracts: []types.V2FileContract{fc}, + } + + basis, toSign, err := w.FundV2Transaction(&txn, fundAmount, false) + if err != nil { + t.Fatal("failed to fund transaction:", err) + } + w.SignV2Inputs(&txn, toSign) + formationSet := rhp4.TransactionSet{ + Transactions: []types.V2Transaction{txn}, + Basis: basis, + } + contractID := txn.V2FileContractID(txn.ID(), 0) + // corrupt the formation set to trigger a rejection + formationSet.Transactions[len(formationSet.Transactions)-1].SiacoinInputs[0].SatisfiedPolicy.Signatures[0] = types.Signature{} + if err := c.AddV2Contract(formationSet, proto4.Usage{}); err != nil { + t.Fatal("failed to add contract:", err) + } + expectedStatuses[contracts.V2ContractStatusPending]++ + assertContractStatus(t, contractID, contracts.V2ContractStatusPending) + + // add a root to the contract + var sector [proto4.SectorSize]byte + frand.Read(sector[:]) + root := proto4.SectorRoot(§or) + roots := []types.Hash256{root} + + if err := node.Volumes.StoreSector(root, §or, proto4.CachedSectorSubtrees(§or), 1); err != nil { + t.Fatal(err) + } + + fc.Filesize = proto4.SectorSize + fc.Capacity = proto4.SectorSize + fc.FileMerkleRoot = proto4.MetaRoot(roots) + fc.RevisionNumber++ + revisionSigHash := cm.TipState().ContractSigHash(fc) + fc.HostSignature = hostKey.SignHash(revisionSigHash) + fc.RenterSignature = renterKey.SignHash(revisionSigHash) + if err := c.ReviseV2Contract(contractID, fc, roots, proto4.Usage{}); err != nil { + t.Fatal(err) + } + assertCachedRoots(t, contractID, 1) + + // mine one block at a time until the contract is rejected so the + // rejection is still inside the reorg buffer and the roots are still + // cached + rejectContract := func(t *testing.T) { + t.Helper() + for i := 0; ; i++ { + testutil.MineAndSync(t, node, types.VoidAddress, 1) + if contract, err := c.V2Contract(contractID); err != nil { + t.Fatal(err) + } else if contract.Status == contracts.V2ContractStatusRejected { + break + } else if i > 20 { + t.Fatal("contract was not rejected") + } + } + expectedStatuses[contracts.V2ContractStatusPending]-- + expectedStatuses[contracts.V2ContractStatusRejected]++ + } + rejectContract(t) + assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) + assertStorageMetrics(t, 0, 0) + assertCachedRoots(t, contractID, 1) + + // resubmitting the same contract replaces the rejected contract with a + // pending contract that has no roots + if err := c.AddV2Contract(formationSet, proto4.Usage{}); err != nil { + t.Fatal("failed to resubmit contract:", err) + } + expectedStatuses[contracts.V2ContractStatusRejected]-- + expectedStatuses[contracts.V2ContractStatusPending]++ + assertContractStatus(t, contractID, contracts.V2ContractStatusPending) + assertCachedRoots(t, contractID, 0) + + // the formation set is still corrupt, leave the contract rejected + rejectContract(t) + assertContractMetrics(t, types.ZeroCurrency, types.ZeroCurrency) + assertCachedRoots(t, contractID, 0) + }) + t.Run("rejected renewal with storage", func(t *testing.T) { cm := node.Chain c := node.Contracts diff --git a/persist/sqlite/migrations_test.go b/persist/sqlite/migrations_test.go index c8013674..0c8f78b9 100644 --- a/persist/sqlite/migrations_test.go +++ b/persist/sqlite/migrations_test.go @@ -767,63 +767,6 @@ VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $ // TestMigrateV55 ensures the migration from version 54 to 55 computes the // reference count of existing sectors and that pruning works afterwards. -func TestMigrateV56(t *testing.T) { - log := zaptest.NewLogger(t) - fp := filepath.Join(t.TempDir(), "hostd.sqlite3") - store := initDBVersion(t, fp, 55, log) - - scanned := types.ChainIndex{Height: 200, ID: frand.Entropy256()} - resolution := types.ChainIndex{Height: 150, ID: frand.Entropy256()} - resolvedID, activeID := types.FileContractID(frand.Entropy256()), types.FileContractID(frand.Entropy256()) - - // populate the pre-migration schema directly - err := store.transaction(func(tx *txn) error { - if _, err := tx.Exec(`UPDATE global_settings SET last_scanned_index=$1`, encode(scanned)); err != nil { - return err - } else if _, err := tx.Exec(`INSERT INTO contract_renters (id, public_key) VALUES (1, $1)`, encode(types.GeneratePrivateKey().PublicKey())); err != nil { - return err - } else if _, err := tx.Exec(`INSERT INTO contract_v2_roots_map (id, revision_number) VALUES (1, 0), (2, 0)`); err != nil { - return err - } - - const query = `INSERT INTO contracts_v2 (contract_id, renter_id, revision_number, formation_txn_set, formation_txn_set_basis, -locked_collateral, rpc_revenue, storage_revenue, ingress_revenue, egress_revenue, account_funding, risked_collateral, raw_revision, -negotiation_height, proof_height, expiration_height, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number, -resolution_block_id, resolution_height) VALUES ($1, 1, $2, $3, $4, $5, $5, $5, $5, $5, $5, $5, $6, 100, 140, 160, $7, 0, $8, 0, $9, $10)` - insert := func(id types.FileContractID, mapID int64, status contracts.V2ContractStatus, resolutionID, resolutionHeight any) error { - _, err := tx.Exec(query, encode(id), encode(uint64(0)), []byte{}, encode(types.ChainIndex{}), encode(types.ZeroCurrency), encode(types.V2FileContract{}), status, mapID, resolutionID, resolutionHeight) - return err - } - if err := insert(resolvedID, 1, contracts.V2ContractStatusSuccessful, encode(resolution.ID), resolution.Height); err != nil { - return err - } - return insert(activeID, 2, contracts.V2ContractStatusActive, nil, nil) - }) - if err != nil { - t.Fatal(err) - } else if err := store.Close(); err != nil { - t.Fatal(err) - } - - store, err = OpenDatabase(fp, log) - if err != nil { - t.Fatal(err) - } - defer store.Close() - - assertLastUpdated := func(t *testing.T, id types.FileContractID, expected types.ChainIndex) { - t.Helper() - var index types.ChainIndex - if err := store.readerDB.QueryRow(`SELECT last_updated_height, last_updated_block_id FROM contracts_v2 WHERE contract_id=$1`, encode(id)).Scan(&index.Height, decode(&index.ID)); err != nil { - t.Fatal(err) - } else if index != expected { - t.Fatalf("expected last updated index %v for %v, got %v", expected, id, index) - } - } - assertLastUpdated(t, resolvedID, resolution) - assertLastUpdated(t, activeID, scanned) -} - func TestMigrateV55(t *testing.T) { log := zaptest.NewLogger(t) fp := filepath.Join(t.TempDir(), "hostd.sqlite3") @@ -897,3 +840,60 @@ func TestMigrateV55(t *testing.T) { t.Fatalf("expected ErrSectorNotFound, got %v", err) } } + +func TestMigrateV56(t *testing.T) { + log := zaptest.NewLogger(t) + fp := filepath.Join(t.TempDir(), "hostd.sqlite3") + store := initDBVersion(t, fp, 55, log) + + scanned := types.ChainIndex{Height: 200, ID: frand.Entropy256()} + resolution := types.ChainIndex{Height: 150, ID: frand.Entropy256()} + resolvedID, activeID := types.FileContractID(frand.Entropy256()), types.FileContractID(frand.Entropy256()) + + // populate the pre-migration schema directly + err := store.transaction(func(tx *txn) error { + if _, err := tx.Exec(`UPDATE global_settings SET last_scanned_index=$1`, encode(scanned)); err != nil { + return err + } else if _, err := tx.Exec(`INSERT INTO contract_renters (id, public_key) VALUES (1, $1)`, encode(types.GeneratePrivateKey().PublicKey())); err != nil { + return err + } else if _, err := tx.Exec(`INSERT INTO contract_v2_roots_map (id, revision_number) VALUES (1, 0), (2, 0)`); err != nil { + return err + } + + const query = `INSERT INTO contracts_v2 (contract_id, renter_id, revision_number, formation_txn_set, formation_txn_set_basis, +locked_collateral, rpc_revenue, storage_revenue, ingress_revenue, egress_revenue, account_funding, risked_collateral, raw_revision, +negotiation_height, proof_height, expiration_height, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number, +resolution_block_id, resolution_height) VALUES ($1, 1, $2, $3, $4, $5, $5, $5, $5, $5, $5, $5, $6, 100, 140, 160, $7, 0, $8, 0, $9, $10)` + insert := func(id types.FileContractID, mapID int64, status contracts.V2ContractStatus, resolutionID, resolutionHeight any) error { + _, err := tx.Exec(query, encode(id), encode(uint64(0)), []byte{}, encode(types.ChainIndex{}), encode(types.ZeroCurrency), encode(types.V2FileContract{}), status, mapID, resolutionID, resolutionHeight) + return err + } + if err := insert(resolvedID, 1, contracts.V2ContractStatusSuccessful, encode(resolution.ID), resolution.Height); err != nil { + return err + } + return insert(activeID, 2, contracts.V2ContractStatusActive, nil, nil) + }) + if err != nil { + t.Fatal(err) + } else if err := store.Close(); err != nil { + t.Fatal(err) + } + + store, err = OpenDatabase(fp, log) + if err != nil { + t.Fatal(err) + } + defer store.Close() + + assertLastUpdated := func(t *testing.T, id types.FileContractID, expected types.ChainIndex) { + t.Helper() + var index types.ChainIndex + if err := store.readerDB.QueryRow(`SELECT last_updated_height, last_updated_block_id FROM contracts_v2 WHERE contract_id=$1`, encode(id)).Scan(&index.Height, decode(&index.ID)); err != nil { + t.Fatal(err) + } else if index != expected { + t.Fatalf("expected last updated index %v for %v, got %v", expected, id, index) + } + } + assertLastUpdated(t, resolvedID, resolution) + assertLastUpdated(t, activeID, scanned) +} From a0b85185ccd79634e4782a10b021620f0a60f559 Mon Sep 17 00:00:00 2001 From: Nate Date: Tue, 15 Sep 2026 13:45:04 -0700 Subject: [PATCH 3/5] address comments --- host/contracts/manager.go | 16 ++++++++-------- host/contracts/manager_test.go | 26 +++++++++++++++++++++----- host/contracts/persist.go | 6 ++++-- persist/sqlite/consensus.go | 3 --- persist/sqlite/contracts.go | 10 ++++++---- persist/sqlite/contracts_test.go | 2 +- 6 files changed, 40 insertions(+), 23 deletions(-) diff --git a/host/contracts/manager.go b/host/contracts/manager.go index 07ea9ae0..240f60d0 100644 --- a/host/contracts/manager.go +++ b/host/contracts/manager.go @@ -363,14 +363,6 @@ func NewManager(store ContractStore, storage StorageManager, chain ChainManager, opt(cm) } - cm.log.Debug("building sector roots cache") - start := time.Now() - roots, err := store.V2SectorRoots() - if err != nil { - return nil, fmt.Errorf("failed to get v2 sector roots: %w", err) - } - cm.log.Debug("loaded sector roots", zap.Duration("elapsed", time.Since(start))) - tip, err := store.Tip() if err != nil { return nil, fmt.Errorf("failed to get tip: %w", err) @@ -380,6 +372,14 @@ func NewManager(store ContractStore, storage StorageManager, chain ChainManager, expireHeight -= ReorgBuffer } + 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.log.Debug("loaded sector roots", zap.Duration("elapsed", time.Since(start))) + cm.roots = &rootsCache{ store: store, contractSectors: roots, diff --git a/host/contracts/manager_test.go b/host/contracts/manager_test.go index 472988cc..c3053072 100644 --- a/host/contracts/manager_test.go +++ b/host/contracts/manager_test.go @@ -1645,7 +1645,7 @@ func TestV2SectorRoots(t *testing.T) { } } - dbRoots, err := node.Store.V2SectorRoots() + dbRoots, err := node.Store.V2SectorRoots(node.Chain.Tip().Height - contracts.ReorgBuffer) if err != nil { t.Fatal(err) } @@ -1902,7 +1902,7 @@ func TestV2SectorRootConsistency(t *testing.T) { assertDBRoots := func(t *testing.T, contractID types.FileContractID, expected []types.Hash256) { t.Helper() - dbRoots, err := node.Store.V2SectorRoots() + dbRoots, err := node.Store.V2SectorRoots(node.Chain.Tip().Height - contracts.ReorgBuffer) if err != nil { t.Fatal("failed to load sector roots:", err) } @@ -2016,9 +2016,19 @@ func TestV2SectorRootConsistency(t *testing.T) { } assertRoots(t, renewalID2, roots) assertDBRoots(t, renewalID2, roots) + // the renewed contract's roots are retained until the renewal is + // outside the reorg buffer + assertDBRoots(t, renewalID1, roots) + + testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertDBRoots(t, renewalID1, nil) + assertDBRoots(t, renewalID2, roots) + + // roots are retained until the resolution is outside the reorg buffer + testutil.MineAndSync(t, node, types.VoidAddress, int(renewal2.ProofHeight-node.Chain.Tip().Height)+1) + assertDBRoots(t, renewalID2, roots) - testutil.MineAndSync(t, node, types.VoidAddress, int(renewal2.ExpirationHeight-node.Chain.Tip().Height)+1) + testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertDBRoots(t, renewalID1, nil) assertDBRoots(t, renewalID2, nil) }) @@ -2075,7 +2085,7 @@ func TestV2SectorRootConsistency(t *testing.T) { t.Fatalf("expected rejected, got %v", contract.Status) } - dbRoots, err := node.Store.V2SectorRoots() + dbRoots, err := node.Store.V2SectorRoots(node.Chain.Tip().Height - contracts.ReorgBuffer) if err != nil { t.Fatal(err) } @@ -2280,8 +2290,14 @@ func TestV2SectorRootConsistency(t *testing.T) { assertRoots(t, renewalID, renewalRoots) assertDBRoots(t, renewalID, renewalRoots) - // original contract's roots should be cleaned up after renewal is confirmed + // original contract's roots are retained until the renewal is outside + // the reorg buffer testutil.MineAndSync(t, node, types.VoidAddress, 1) + assertDBRoots(t, contractID, roots) + assertRoots(t, renewalID, renewalRoots) + assertDBRoots(t, renewalID, renewalRoots) + + testutil.MineAndSync(t, node, types.VoidAddress, contracts.ReorgBuffer+1) assertDBRoots(t, contractID, nil) // note: roots is not checked because it doesn't get cleared on expiration. assertRoots(t, renewalID, renewalRoots) assertDBRoots(t, renewalID, renewalRoots) diff --git a/host/contracts/persist.go b/host/contracts/persist.go index 4f8aadbf..6523f2ad 100644 --- a/host/contracts/persist.go +++ b/host/contracts/persist.go @@ -16,8 +16,10 @@ type ( // Tip returns the last scanned chain index. Tip() (types.ChainIndex, error) - // V2SectorRoots returns the sector roots for all v2 contracts. - V2SectorRoots() (map[types.FileContractID][]types.Hash256, error) + // V2SectorRoots returns the sector roots of all v2 contracts that + // have not been rejected and are either unresolved or were resolved + // at or after minHeight. + V2SectorRoots(minHeight uint64) (map[types.FileContractID][]types.Hash256, error) // Contracts returns a paginated list of contracts sorted by expiration // asc. diff --git a/persist/sqlite/consensus.go b/persist/sqlite/consensus.go index d2abb3a1..026940b6 100644 --- a/persist/sqlite/consensus.go +++ b/persist/sqlite/consensus.go @@ -1743,9 +1743,6 @@ func revertFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.Fil // panic if the contract is not failed. Proper reverts should have // ensured that this never happens. panic(fmt.Errorf("unexpected contract state transition %q -> %q", state.Status, contracts.V2ContractStatusFailed)) - } else if state.Status == contracts.V2ContractStatusFailed { - // skip update, most likely rescanning - continue } // update the contract's resolution index and status diff --git a/persist/sqlite/contracts.go b/persist/sqlite/contracts.go index b7a23d1f..cfa48bf2 100644 --- a/persist/sqlite/contracts.go +++ b/persist/sqlite/contracts.go @@ -360,12 +360,14 @@ func (s *Store) ReviseContract(revision contracts.SignedRevision, oldRoots, newR }) } -// V2SectorRoots returns the sector roots for all active v2 contracts. -func (s *Store) V2SectorRoots() (roots map[types.FileContractID][]types.Hash256, err error) { +// V2SectorRoots returns the sector roots of all v2 contracts that have not +// been rejected and are either unresolved or were resolved at or after +// minHeight. +func (s *Store) V2SectorRoots(minHeight uint64) (roots map[types.FileContractID][]types.Hash256, err error) { err = s.transaction(func(tx *txn) error { const contractsQuery = `SELECT contract_id, raw_revision, contract_v2_roots_map_id, contract_v2_roots_map_revision_number FROM contracts_v2 -WHERE contract_status <> $1 AND resolution_height IS NULL;` - rows, err := tx.Query(contractsQuery, contracts.V2ContractStatusRejected) +WHERE contract_status <> $1 AND (resolution_height IS NULL OR last_updated_height >= $2);` + rows, err := tx.Query(contractsQuery, contracts.V2ContractStatusRejected, minHeight) if err != nil { return err } diff --git a/persist/sqlite/contracts_test.go b/persist/sqlite/contracts_test.go index 14706e2e..8157c50b 100644 --- a/persist/sqlite/contracts_test.go +++ b/persist/sqlite/contracts_test.go @@ -961,7 +961,7 @@ WHERE c.contract_id = $1`, encode(contractID)).Scan(&count) t.Run("renewal rejected then renewed again", func(t *testing.T) { checkSectorRoots := func(t *testing.T, contractID types.FileContractID, expected []types.Hash256) { t.Helper() - roots, err := db.V2SectorRoots() + roots, err := db.V2SectorRoots(0) if err != nil { t.Fatal(err) } From a6b7a69fe4ba4c04f3d96a370fabe82d0baf8ed4 Mon Sep 17 00:00:00 2001 From: Nate Date: Wed, 16 Sep 2026 14:24:22 -0700 Subject: [PATCH 4/5] address comments --- host/contracts/cache.go | 24 ++++++++++++++++-------- persist/sqlite/consensus.go | 14 +++++++------- persist/sqlite/contracts.go | 19 +++++-------------- persist/sqlite/init.sql | 6 +++--- persist/sqlite/migrations.go | 15 ++++++--------- persist/sqlite/migrations_test.go | 10 +++++----- 6 files changed, 42 insertions(+), 46 deletions(-) diff --git a/host/contracts/cache.go b/host/contracts/cache.go index 41889f7f..7736bb99 100644 --- a/host/contracts/cache.go +++ b/host/contracts/cache.go @@ -33,16 +33,24 @@ func (rc *rootsCache) UpdateSectorRoots(id types.FileContractID, roots []types.H // 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() - if height > rc.lastExpiredHeight { - expired, err := rc.store.ExpiredV2Contracts(rc.lastExpiredHeight, height) - if err != nil { - return fmt.Errorf("failed to get expired contracts: %w", err) - } - for _, id := range expired { - delete(rc.contractSectors, id) - } + for _, id := range expired { + delete(rc.contractSectors, id) } rc.lastExpiredHeight = height return nil diff --git a/persist/sqlite/consensus.go b/persist/sqlite/consensus.go index 026940b6..8484470e 100644 --- a/persist/sqlite/consensus.go +++ b/persist/sqlite/consensus.go @@ -1298,7 +1298,7 @@ func applyV2ContractFormation(tx *txn, index types.ChainIndex, confirmed []types } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=$1, contract_status=$2, last_updated_height=$3, last_updated_block_id=$4 WHERE id=$5`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=$1, contract_status=$2, last_status_update_height=$3, last_status_update_block_id=$4 WHERE id=$5`) if err != nil { return fmt.Errorf("failed to prepare update status statement: %w", err) } @@ -1373,7 +1373,7 @@ func revertV2ContractFormation(tx *txn, index types.ChainIndex, reverted []types } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET confirmation_index=NULL, contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1470,7 +1470,7 @@ func applySuccessfulV2Contracts(tx *txn, index types.ChainIndex, status contract } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1555,7 +1555,7 @@ func applyFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.File } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=?, resolution_height=?, contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1643,7 +1643,7 @@ func revertSuccessfulV2Contracts(tx *txn, index types.ChainIndex, status contrac } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1715,7 +1715,7 @@ func revertFailedV2Contracts(tx *txn, index types.ChainIndex, failed []types.Fil } defer done() - updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStmt, err := tx.Prepare(`UPDATE contracts_v2 SET resolution_block_id=NULL, resolution_height=NULL, contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return fmt.Errorf("failed to prepare update statement: %w", err) } @@ -1978,7 +1978,7 @@ func rejectV2Contracts(tx *txn, index types.ChainIndex, height uint64, log *zap. } defer stateDone() - updateStatus, err := tx.Prepare(`UPDATE contracts_v2 SET contract_status=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`) + updateStatus, err := tx.Prepare(`UPDATE contracts_v2 SET contract_status=?, last_status_update_height=?, last_status_update_block_id=? WHERE id=?`) if err != nil { return nil, fmt.Errorf("failed to prepare v2 update statement: %w", err) } diff --git a/persist/sqlite/contracts.go b/persist/sqlite/contracts.go index cfa48bf2..c3bb6292 100644 --- a/persist/sqlite/contracts.go +++ b/persist/sqlite/contracts.go @@ -366,7 +366,7 @@ func (s *Store) ReviseContract(revision contracts.SignedRevision, oldRoots, newR func (s *Store) V2SectorRoots(minHeight uint64) (roots map[types.FileContractID][]types.Hash256, err error) { err = s.transaction(func(tx *txn) error { const contractsQuery = `SELECT contract_id, raw_revision, contract_v2_roots_map_id, contract_v2_roots_map_revision_number FROM contracts_v2 -WHERE contract_status <> $1 AND (resolution_height IS NULL OR last_updated_height >= $2);` +WHERE contract_status <> $1 AND (resolution_height IS NULL OR last_status_update_height >= $2);` rows, err := tx.Query(contractsQuery, contracts.V2ContractStatusRejected, minHeight) if err != nil { return err @@ -455,7 +455,7 @@ func (s *Store) ExpireV2ContractSectors(height uint64) error { func (s *Store) ExpiredV2Contracts(minHeight, maxHeight uint64) (ids []types.FileContractID, err error) { err = s.transaction(func(tx *txn) error { const query = `SELECT contract_id FROM contracts_v2 -WHERE contract_status IN ($1, $2, $3, $4) AND last_updated_height >= $5 AND last_updated_height < $6` +WHERE contract_status IN ($1, $2, $3, $4) AND last_status_update_height >= $5 AND last_status_update_height < $6` rows, err := tx.Query(query, contracts.V2ContractStatusRejected, contracts.V2ContractStatusSuccessful, contracts.V2ContractStatusFailed, contracts.V2ContractStatusRenewed, minHeight, maxHeight) if err != nil { return fmt.Errorf("failed to query contracts: %w", err) @@ -543,15 +543,9 @@ LIMIT $3)` // updateResolvedV2Contract clears a contract and returns its ID func updateResolvedV2Contract(tx *txn, contractID types.FileContractID, renewedDBID int64) (dbID int64, err error) { - index, err := lastScannedIndex(tx) - if err != nil { - return 0, fmt.Errorf("failed to get last scanned index: %w", err) - } - const clearQuery = `UPDATE contracts_v2 SET renewed_to=$1, last_updated_height=$2, last_updated_block_id=$3 WHERE contract_id=$4 RETURNING id;` + const clearQuery = `UPDATE contracts_v2 SET renewed_to=$1 WHERE contract_id=$2 RETURNING id;` err = tx.QueryRow(clearQuery, renewedDBID, - index.Height, - encode(index.ID), encode(contractID), ).Scan(&dbID) return @@ -900,7 +894,7 @@ func insertV2Contract(tx *txn, contract contracts.V2Contract, mapID, mapRevision const query = `INSERT INTO contracts_v2 (contract_id, renter_id, locked_collateral, rpc_revenue, storage_revenue, ingress_revenue, egress_revenue, account_funding, risked_collateral, revision_number, negotiation_height, proof_height, expiration_height, formation_txn_set, formation_txn_set_basis, raw_revision, contract_status, sector_count, contract_v2_roots_map_id, contract_v2_roots_map_revision_number, -last_updated_height, last_updated_block_id) VALUES +last_status_update_height, last_status_update_block_id) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22) RETURNING id;` renterID, err := renterDBID(tx, contract.RenterPublicKey) @@ -1033,10 +1027,7 @@ func reviseV2Contract(tx *txn, id types.FileContractID, revision types.V2FileCon return 0, fmt.Errorf("revision number went backwards: existing=%d revised=%d", existingRevision, revision.RevisionNumber) } - index, err := lastScannedIndex(tx) - if err != nil { - return 0, fmt.Errorf("failed to get last scanned index: %w", err) - } else if _, err := tx.Exec(`UPDATE contracts_v2 SET raw_revision=?, revision_number=?, sector_count=?, last_updated_height=?, last_updated_block_id=? WHERE id=?`, encode(revision), encode(revision.RevisionNumber), revision.Filesize/proto4.SectorSize, index.Height, encode(index.ID), contractDBID); err != nil { + if _, err := tx.Exec(`UPDATE contracts_v2 SET raw_revision=?, revision_number=?, sector_count=? WHERE id=?`, encode(revision), encode(revision.RevisionNumber), revision.Filesize/proto4.SectorSize, contractDBID); err != nil { return 0, fmt.Errorf("failed to update contract: %w", err) } else if err := updateV2ContractUsage(tx, contractDBID, usage); err != nil { return 0, fmt.Errorf("failed to update contract usage: %w", err) diff --git a/persist/sqlite/init.sql b/persist/sqlite/init.sql index 5ad9e8bb..aea0062d 100644 --- a/persist/sqlite/init.sql +++ b/persist/sqlite/init.sql @@ -183,8 +183,8 @@ CREATE TABLE contracts_v2 ( resolution_height INTEGER CHECK((resolution_height IS NULL) = (resolution_block_id IS NULL)), -- null if the resolution has not been confirmed on the blockchain contract_status TEXT NOT NULL, sector_count INTEGER NOT NULL, -- used for cleanup - last_updated_height INTEGER NOT NULL DEFAULT 0, - last_updated_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000', + last_status_update_height INTEGER NOT NULL DEFAULT 0, + last_status_update_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000', contract_v2_roots_map_id INTEGER NOT NULL, contract_v2_roots_map_revision_number INTEGER NOT NULL, @@ -201,7 +201,7 @@ CREATE INDEX contracts_v2_contract_status ON contracts_v2(contract_status); CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_proof_height ON contracts_v2(confirmation_index, resolution_block_id, proof_height); CREATE INDEX contracts_v2_confirmation_index_resolution_block_id_expiration_height ON contracts_v2(confirmation_index, resolution_block_id, expiration_height); CREATE INDEX contracts_v2_resolution_height ON contracts_v2(resolution_height); -CREATE INDEX contracts_v2_contract_status_last_updated_height ON contracts_v2(contract_status, last_updated_height); +CREATE INDEX contracts_v2_contract_status_last_status_update_height ON contracts_v2(contract_status, last_status_update_height); CREATE INDEX contracts_v2_confirmation_index_proof_height ON contracts_v2(confirmation_index, proof_height); CREATE INDEX contracts_v2_confirmation_index_negotiation_height ON contracts_v2(confirmation_index, negotiation_height); CREATE INDEX contracts_v2_roots_map_id_contract_v2_roots_map_revision_number ON contracts_v2(contract_v2_roots_map_id, contract_v2_roots_map_revision_number); diff --git a/persist/sqlite/migrations.go b/persist/sqlite/migrations.go index afa061e9..078a4888 100644 --- a/persist/sqlite/migrations.go +++ b/persist/sqlite/migrations.go @@ -13,7 +13,7 @@ import ( "go.uber.org/zap" ) -// migrateVersion56 adds the last updated index to contracts_v2. Resolved +// migrateVersion56 adds the last status update index to contracts_v2. Resolved // contracts are set to their resolution index, all other contracts to the last // scanned index. func migrateVersion56(tx *txn, _ *zap.Logger) error { @@ -22,14 +22,11 @@ func migrateVersion56(tx *txn, _ *zap.Logger) error { return fmt.Errorf("failed to get last scanned index: %w", err) } _, err := tx.Exec(` -ALTER TABLE contracts_v2 ADD COLUMN last_updated_height INTEGER NOT NULL DEFAULT 0; -ALTER TABLE contracts_v2 ADD COLUMN last_updated_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000'; -UPDATE contracts_v2 SET last_updated_height=resolution_height, last_updated_block_id=resolution_block_id WHERE resolution_height IS NOT NULL; -CREATE INDEX contracts_v2_contract_status_last_updated_height ON contracts_v2(contract_status, last_updated_height);`) - if err != nil { - return fmt.Errorf("failed to add last updated columns: %w", err) - } - _, err = tx.Exec(`UPDATE contracts_v2 SET last_updated_height=$1, last_updated_block_id=$2 WHERE resolution_height IS NULL`, index.Height, encode(index.ID)) +ALTER TABLE contracts_v2 ADD COLUMN last_status_update_height INTEGER NOT NULL DEFAULT 0; +ALTER TABLE contracts_v2 ADD COLUMN last_status_update_block_id BLOB NOT NULL DEFAULT x'0000000000000000000000000000000000000000000000000000000000000000'; +UPDATE contracts_v2 SET last_status_update_height=resolution_height, last_status_update_block_id=resolution_block_id WHERE resolution_height IS NOT NULL; +UPDATE contracts_v2 SET last_status_update_height=$1, last_status_update_block_id=$2 WHERE resolution_height IS NULL; +CREATE INDEX contracts_v2_contract_status_last_status_update_height ON contracts_v2(contract_status, last_status_update_height);`, index.Height, encode(index.ID)) return err } diff --git a/persist/sqlite/migrations_test.go b/persist/sqlite/migrations_test.go index 0c8f78b9..dda242bc 100644 --- a/persist/sqlite/migrations_test.go +++ b/persist/sqlite/migrations_test.go @@ -885,15 +885,15 @@ resolution_block_id, resolution_height) VALUES ($1, 1, $2, $3, $4, $5, $5, $5, $ } defer store.Close() - assertLastUpdated := func(t *testing.T, id types.FileContractID, expected types.ChainIndex) { + assertLastStatusUpdate := func(t *testing.T, id types.FileContractID, expected types.ChainIndex) { t.Helper() var index types.ChainIndex - if err := store.readerDB.QueryRow(`SELECT last_updated_height, last_updated_block_id FROM contracts_v2 WHERE contract_id=$1`, encode(id)).Scan(&index.Height, decode(&index.ID)); err != nil { + if err := store.readerDB.QueryRow(`SELECT last_status_update_height, last_status_update_block_id FROM contracts_v2 WHERE contract_id=$1`, encode(id)).Scan(&index.Height, decode(&index.ID)); err != nil { t.Fatal(err) } else if index != expected { - t.Fatalf("expected last updated index %v for %v, got %v", expected, id, index) + t.Fatalf("expected last status update index %v for %v, got %v", expected, id, index) } } - assertLastUpdated(t, resolvedID, resolution) - assertLastUpdated(t, activeID, scanned) + assertLastStatusUpdate(t, resolvedID, resolution) + assertLastStatusUpdate(t, activeID, scanned) } From 492f9ee181b0c3948659e5bbdb1b7bfad7caadb5 Mon Sep 17 00:00:00 2001 From: Nate Date: Wed, 16 Sep 2026 14:26:25 -0700 Subject: [PATCH 5/5] update changeset with warning --- .changeset/fix_stored_sectors_table_bloat.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.changeset/fix_stored_sectors_table_bloat.md b/.changeset/fix_stored_sectors_table_bloat.md index edc6aca1..f12ebe73 100644 --- a/.changeset/fix_stored_sectors_table_bloat.md +++ b/.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. \ No newline at end of file