Contract Pruning

api/worker.go

@@ -49,6 +49,12 @@ type RHPScanRequest struct {
 	Timeout time.Duration   `json:"timeout"`
 }
 
+// RHPPruneContractRequest is the request type for the /rhp/contract/:id/prune
+// endpoint.
+type RHPPruneContractRequest struct {
+	Timeout time.Duration `json:"timeout"`
+}
+
 // RHPPriceTableRequest is the request type for the /rhp/pricetable endpoint.
 type RHPPriceTableRequest struct {
 	HostKey    types.PublicKey `json:"hostKey"`

bus/bus.go

@@ -100,7 +100,7 @@ type (
 		RenewedContract(ctx context.Context, renewedFrom types.FileContractID) (api.ContractMetadata, error)
 		SetContractSet(ctx context.Context, set string, contracts []types.FileContractID) error
 
-		ContractRoots(ctx context.Context, id types.FileContractID) ([]types.Hash256, error)
+		ContractRoots(ctx context.Context, id types.FileContractID) ([]types.Hash256, []types.Hash256, error)
 		ContractSizes(ctx context.Context) (map[types.FileContractID]api.ContractSize, error)
 		ContractSize(ctx context.Context, id types.FileContractID) (api.ContractSize, error)
 
@@ -126,6 +126,10 @@ type (
 		RefreshHealth(ctx context.Context) error
 		UnhealthySlabs(ctx context.Context, healthCutoff float64, set string, limit int) ([]api.UnhealthySlab, error)
 		UpdateSlab(ctx context.Context, s object.Slab, contractSet string, usedContracts map[types.PublicKey]types.FileContractID) error
+
+		TrackUpload(uID api.UploadID) error
+		AddUploadingSector(uID api.UploadID, id types.FileContractID, root types.Hash256) error
+		FinishUpload(uID api.UploadID)
 	}
 
 	// An AutopilotStore stores autopilots.
@@ -167,10 +171,9 @@ type bus struct {
 
 	eas EphemeralAccountStore
 
-	logger           *zap.SugaredLogger
-	accounts         *accounts
-	contractLocks    *contractLocks
-	uploadingSectors *uploadingSectorsCache
+	logger        *zap.SugaredLogger
+	accounts      *accounts
+	contractLocks *contractLocks
 
 	startTime time.Time
 }
@@ -816,11 +819,11 @@ func (b *bus) contractIDRootsHandlerGET(jc jape.Context) {
 		return
 	}
 
-	roots, err := b.ms.ContractRoots(jc.Request.Context(), id)
+	roots, uploading, err := b.ms.ContractRoots(jc.Request.Context(), id)
 	if jc.Check("couldn't fetch contract sectors", err) == nil {
 		jc.Encode(api.ContractRootsResponse{
 			Roots:     roots,
-			Uploading: b.uploadingSectors.sectors(id),
+			Uploading: uploading,
 		})
 	}
 }
@@ -1506,7 +1509,7 @@ func (b *bus) stateHandlerGET(jc jape.Context) {
 func (b *bus) uploadTrackHandlerPOST(jc jape.Context) {
 	var id api.UploadID
 	if jc.DecodeParam("id", &id) == nil {
-		jc.Check("failed to track upload", b.uploadingSectors.trackUpload(id))
+		jc.Check("failed to track upload", b.ms.TrackUpload(id))
 	}
 }
 
@@ -1519,13 +1522,13 @@ func (b *bus) uploadAddSectorHandlerPOST(jc jape.Context) {
 	if jc.Decode(&req) != nil {
 		return
 	}
-	jc.Check("failed to add sector", b.uploadingSectors.addUploadingSector(id, req.ContractID, req.Root))
+	jc.Check("failed to add sector", b.ms.AddUploadingSector(id, req.ContractID, req.Root))
 }
 
 func (b *bus) uploadFinishedHandlerDELETE(jc jape.Context) {
 	var id api.UploadID
 	if jc.DecodeParam("id", &id) == nil {
-		b.uploadingSectors.finishUpload(id)
+		b.ms.FinishUpload(id)
 	}
 }
 
@@ -1578,21 +1581,20 @@ func (b *bus) webhookHandlerPost(jc jape.Context) {
 // New returns a new Bus.
 func New(s Syncer, am *alerts.Manager, hm *webhooks.Manager, cm ChainManager, tp TransactionPool, w Wallet, hdb HostDB, as AutopilotStore, ms MetadataStore, ss SettingStore, eas EphemeralAccountStore, l *zap.Logger) (*bus, error) {
 	b := &bus{
-		alerts:           alerts.WithOrigin(am, "bus"),
-		alertMgr:         am,
-		hooks:            hm,
-		s:                s,
-		cm:               cm,
-		tp:               tp,
-		w:                w,
-		hdb:              hdb,
-		as:               as,
-		ms:               ms,
-		ss:               ss,
-		eas:              eas,
-		contractLocks:    newContractLocks(),
-		uploadingSectors: newUploadingSectorsCache(),
-		logger:           l.Sugar().Named("bus"),
+		alerts:        alerts.WithOrigin(am, "bus"),
+		alertMgr:      am,
+		hooks:         hm,
+		s:             s,
+		cm:            cm,
+		tp:            tp,
+		w:             w,
+		hdb:           hdb,
+		as:            as,
+		ms:            ms,
+		ss:            ss,
+		eas:           eas,
+		contractLocks: newContractLocks(),
+		logger:        l.Sugar().Named("bus"),
 
 		startTime: time.Now(),
 	}
@@ -1773,14 +1775,13 @@ func (b *bus) Handler() http.Handler {
 		"POST /search/hosts":   b.searchHostsHandlerPOST,
 		"GET  /search/objects": b.searchObjectsHandlerGET,
 
-		"GET    /stats/objects": b.objectsStatshandlerGET,
-
 		"GET    /settings":     b.settingsHandlerGET,
 		"GET    /setting/:key": b.settingKeyHandlerGET,
 		"PUT    /setting/:key": b.settingKeyHandlerPUT,
 		"DELETE /setting/:key": b.settingKeyHandlerDELETE,
 
-		"GET    /state": b.stateHandlerGET,
+		"GET    /state":         b.stateHandlerGET,
+		"GET    /stats/objects": b.objectsStatshandlerGET,
 
 		"POST   /upload/:id":        b.uploadTrackHandlerPOST,
 		"POST   /upload/:id/sector": b.uploadAddSectorHandlerPOST,

internal/testing/cluster_test.go

@@ -2025,7 +2025,11 @@ func TestWallet(t *testing.T) {
 		// to the amount of money sent as well as the miner fees used.
 		spendableDiff := wallet.Spendable.Sub(updated.Spendable)
 		if updated.Unconfirmed.Cmp(spendableDiff) > 0 {
-			t.Fatalf("unconfirmed balance can't be greater than the difference in spendable balance here, confirmed %v->%v unconfirmed %v->%v spendable %v->%v fee %v", wallet.Confirmed, updated.Confirmed, wallet.Unconfirmed, updated.Unconfirmed, wallet.Spendable, updated.Spendable, minerFee)
+			t.Fatalf("unconfirmed balance can't be greater than the difference in spendable balance here: \nconfirmed %v (%v) - >%v (%v) \nunconfirmed %v (%v) -> %v (%v) \nspendable %v (%v) -> %v (%v) \nfee %v (%v)",
+				wallet.Confirmed, wallet.Confirmed.ExactString(), updated.Confirmed, updated.Confirmed.ExactString(),
+				wallet.Unconfirmed, wallet.Unconfirmed.ExactString(), updated.Unconfirmed, updated.Unconfirmed.ExactString(),
+				wallet.Spendable, wallet.Spendable.ExactString(), updated.Spendable, updated.Spendable.ExactString(),
+				minerFee, minerFee.ExactString())
 		}
 		withdrawnAmt := spendableDiff.Sub(updated.Unconfirmed)
 		expectedWithdrawnAmt := sendAmt.Add(minerFee)

internal/testing/pruning_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"math"
 	"strings"
 	"testing"
 	"time"
@@ -192,6 +193,8 @@ func TestSectorPruning(t *testing.T) {
 	w := cluster.Worker
 	b := cluster.Bus
 
+	numObjects := 10
+
 	// add hosts
 	_, err = cluster.AddHostsBlocking(int(cfg.Contracts.Amount))
 	if err != nil {
@@ -210,7 +213,7 @@ func TestSectorPruning(t *testing.T) {
 	}
 
 	// add several objects
-	for i := 0; i < 10; i++ {
+	for i := 0; i < numObjects; i++ {
 		filename := fmt.Sprintf("obj_%d", i)
 		if err := w.UploadObject(context.Background(), bytes.NewReader([]byte(filename)), filename); err != nil {
 			t.Fatal(err)
@@ -238,7 +241,7 @@ func TestSectorPruning(t *testing.T) {
 		}
 		n += len(cRoots)
 	}
-	if n != rs.TotalShards*10 {
+	if n != rs.TotalShards*numObjects {
 		t.Fatal("unexpected number of roots", n)
 	}
 
@@ -254,6 +257,7 @@ func TestSectorPruning(t *testing.T) {
 		}
 	}()
 	b = cluster2.Bus
+	w = cluster2.Worker
 
 	// assert prunable data is 0
 	if res, err := b.PrunableData(context.Background()); err != nil {
@@ -262,8 +266,8 @@ func TestSectorPruning(t *testing.T) {
 		t.Fatal("expected 0 prunable data", n)
 	}
 
-	// delete a random number of objects
-	for i := 0; i < 10; i += 2 {
+	// delete every other object
+	for i := 0; i < numObjects; i += 2 {
 		filename := fmt.Sprintf("obj_%d", i)
 		if err := b.DeleteObject(context.Background(), filename, false); err != nil {
 			t.Fatal(err)
@@ -273,7 +277,40 @@ func TestSectorPruning(t *testing.T) {
 	// assert amount of prunable data
 	if res, err := b.PrunableData(context.Background()); err != nil {
 		t.Fatal(err)
-	} else if res.TotalPrunable != 5*uint64(rs.TotalShards)*rhpv2.SectorSize {
+	} else if res.TotalPrunable != uint64(math.Ceil(float64(numObjects)/2))*uint64(rs.TotalShards)*rhpv2.SectorSize {
 		t.Fatal("unexpected prunable data", n)
 	}
+
+	// prune all contracts
+	for _, c := range contracts {
+		if _, err := w.RHPPruneContract(context.Background(), c.ID, 0); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// assert spending records were updated and prunable data is 0
+	if err = Retry(10, testBusFlushInterval, func() error {
+		if res, err := b.PrunableData(context.Background()); err != nil {
+			t.Fatal(err)
+		} else if res.TotalPrunable != 0 {
+			return fmt.Errorf("unexpected prunable data: %d", n)
+		}
+		return nil
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// assert spending was updated
+	for _, c := range contracts {
+		c, err := b.Contract(context.Background(), c.ID)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if c.Spending.SectorRoots.IsZero() {
+			t.Fatal("spending record not updated")
+		}
+		if c.Spending.Deletions.IsZero() {
+			t.Fatal("spending record not updated")
+		}
+	}
 }

stores/metadata.go

@@ -606,9 +606,9 @@ func (s *SQLStore) Contract(ctx context.Context, id types.FileContractID) (api.C
 	return contract.convert(), nil
 }
 
-func (s *SQLStore) ContractRoots(ctx context.Context, id types.FileContractID) (roots []types.Hash256, err error) {
+func (s *SQLStore) ContractRoots(ctx context.Context, id types.FileContractID) (roots, uploading []types.Hash256, err error) {
 	if !s.isKnownContract(id) {
-		return nil, api.ErrContractNotFound
+		return nil, nil, api.ErrContractNotFound
 	}
 
 	var dbRoots []hash256
@@ -627,6 +627,7 @@ WHERE c.fcid = ?
 		}
 	}
 
+	uploading = s.uploadingSectors.sectors(id)
 	return
 }
 
@@ -671,9 +672,21 @@ GROUP BY c.fcid
 
 	sizes := make(map[types.FileContractID]api.ContractSize)
 	for _, row := range rows {
-		if types.FileContractID(row.Fcid) == (types.FileContractID{}) {
+		fcid := types.FileContractID(row.Fcid)
+		if fcid == (types.FileContractID{}) {
 			return nil, errors.New("invalid file contract id")
 		}
+
+		// adjust the amount of prunable data with the pending uploads, due to how
+		// we record contract spending a contract's size might already include
+		// pending sectors
+		pending := s.uploadingSectors.pending(fcid)
+		if pending > row.Prunable {
+			row.Prunable = 0
+		} else {
+			row.Prunable -= pending
+		}
+
 		sizes[types.FileContractID(row.Fcid)] = api.ContractSize{
 			Size:     row.Size,
 			Prunable: row.Prunable,
@@ -694,7 +707,7 @@ func (s *SQLStore) ContractSize(ctx context.Context, id types.FileContractID) (a
 
 	if err := s.db.
 		Raw(`
-SELECT c.size, CASE WHEN c.size>(COUNT(cs.db_sector_id) * ?) THEN c.size-(COUNT(cs.db_sector_id) * ?) ELSE 0 END as prunable
+SELECT MAX(c.size), CASE WHEN MAX(c.size)>(COUNT(cs.db_sector_id) * ?) THEN MAX(c.size)-(COUNT(cs.db_sector_id) * ?) ELSE 0 END as prunable
 FROM contracts c
 LEFT JOIN contract_sectors cs ON cs.db_contract_id = c.id
 WHERE c.fcid = ?
@@ -704,6 +717,16 @@ WHERE c.fcid = ?
 		return api.ContractSize{}, err
 	}
 
+	// adjust the amount of prunable data with the pending uploads, due to how
+	// we record contract spending a contract's size might already include
+	// pending sectors
+	pending := s.uploadingSectors.pending(id)
+	if pending > size.Prunable {
+		size.Prunable = 0
+	} else {
+		size.Prunable -= pending
+	}
+
 	return api.ContractSize{
 		Size:     size.Size,
 		Prunable: size.Prunable,
@@ -1548,6 +1571,18 @@ func (s *SQLStore) MarkPackedSlabsUploaded(ctx context.Context, slabs []api.Uplo
 	return nil
 }
 
+func (s *SQLStore) TrackUpload(uID api.UploadID) error {
+	return s.uploadingSectors.trackUpload(uID)
+}
+
+func (s *SQLStore) AddUploadingSector(uID api.UploadID, id types.FileContractID, root types.Hash256) error {
+	return s.uploadingSectors.addUploadingSector(uID, id, root)
+}
+
+func (s *SQLStore) FinishUpload(uID api.UploadID) {
+	s.uploadingSectors.finishUpload(uID)
+}
+
 func markPackedSlabUploaded(tx *gorm.DB, slab api.UploadedPackedSlab, contracts map[types.PublicKey]dbContract) (string, error) {
 	// find the slab
 	var sla dbSlab

stores/metadata_test.go

@@ -437,7 +437,7 @@ func TestContractRoots(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	roots, err := cs.ContractRoots(context.Background(), fcids[0])
+	roots, _, err := cs.ContractRoots(context.Background(), fcids[0])
 	if err != nil {
 		t.Fatal(err)
 	}

bus/uploadingsectors.go → stores/sectorscache.go

@@ -1,10 +1,11 @@
-package bus
+package stores
 
 import (
 	"fmt"
 	"sync"
 	"time"
 
+	"go.sia.tech/core/rhp/v2"
 	"go.sia.tech/core/types"
 	"go.sia.tech/renterd/api"
 )
@@ -65,6 +66,20 @@ func (usc *uploadingSectorsCache) addUploadingSector(uID api.UploadID, fcid type
 	return fmt.Errorf("%w; id '%v'", api.ErrUnknownUpload, uID)
 }
 
+func (usc *uploadingSectorsCache) pending(fcid types.FileContractID) (size uint64) {
+	usc.mu.Lock()
+	var uploads []*ongoingUpload
+	for _, ongoing := range usc.uploads {
+		uploads = append(uploads, ongoing)
+	}
+	usc.mu.Unlock()
+
+	for _, ongoing := range uploads {
+		size += uint64(len(ongoing.sectors(fcid))) * rhp.SectorSize
+	}
+	return
+}
+
 func (usc *uploadingSectorsCache) sectors(fcid types.FileContractID) (roots []types.Hash256) {
 	usc.mu.Lock()
 	var uploads []*ongoingUpload

bus/uploadingsectors_test.go → stores/sectorscache_test.go

@@ -1,4 +1,4 @@
-package bus
+package stores
 
 import (
 	"errors"