Merge pull request #222 from SiaFoundation/nate/add-sector-verify-end…

…point

Add sector verify endpoint

api/api.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"go.sia.tech/core/consensus"
+	rhp2 "go.sia.tech/core/rhp/v2"
 	rhp3 "go.sia.tech/core/rhp/v3"
 	"go.sia.tech/core/types"
 	"go.sia.tech/hostd/alerts"
@@ -63,6 +64,10 @@ type (
 		SetReadOnly(id int64, readOnly bool) error
 		RemoveSector(root types.Hash256) error
 		ResizeCache(size uint32)
+		Read(types.Hash256) (*[rhp2.SectorSize]byte, error)
+
+		// SectorReferences returns the references to a sector
+		SectorReferences(root types.Hash256) (storage.SectorReference, error)
 	}
 
 	// A ContractManager manages the host's contracts
@@ -198,7 +203,8 @@ func NewServer(name string, hostKey types.PublicKey, a Alerts, g Syncer, chain C
 		"GET /accounts":                  api.handleGETAccounts,
 		"GET /accounts/:account/funding": api.handleGETAccountFunding,
 		// sector endpoints
-		"DELETE /sectors/:root": api.handleDeleteSector,
+		"DELETE /sectors/:root":     api.handleDeleteSector,
+		"GET /sectors/:root/verify": api.handleGETVerifySector,
 		// volume endpoints
 		"GET /volumes":               api.handleGETVolumes,
 		"POST /volumes":              api.handlePOSTVolume,

api/types.go

@@ -142,6 +142,12 @@ type (
 	CreateDirRequest struct {
 		Path string `json:"path"`
 	}
+
+	// VerifySectorResponse is the response body for the [GET] /sectors/:root/verify endpoint.
+	VerifySectorResponse struct {
+		storage.SectorReference
+		Error string `json:"error,omitempty"`
+	}
 )
 
 // MarshalJSON implements json.Marshaler

api/volumes.go

@@ -7,6 +7,8 @@ import (
 	"net/http"
 	"sync"
 
+	rhp2 "go.sia.tech/core/rhp/v2"
+	"go.sia.tech/core/types"
 	"go.sia.tech/hostd/host/storage"
 	"go.sia.tech/jape"
 )
@@ -197,3 +199,36 @@ func (a *api) handleDELETEVolumeCancelOp(c jape.Context) {
 	err := a.volumeJobs.Cancel(id)
 	a.checkServerError(c, "failed to cancel operation", err)
 }
+
+func (a *api) handleGETVerifySector(jc jape.Context) {
+	var root types.Hash256
+	if err := jc.DecodeParam("root", &root); err != nil {
+		return
+	}
+
+	refs, err := a.volumes.SectorReferences(root)
+	if err != nil {
+		jc.Error(err, http.StatusInternalServerError)
+		return
+	}
+
+	resp := VerifySectorResponse{
+		SectorReference: refs,
+	}
+
+	// if the sector is not referenced return the empty response without
+	// attempting to read the sector data
+	if len(refs.Contracts) == 0 && refs.TempStorage == 0 && refs.Locks == 0 {
+		jc.Encode(resp)
+		return
+	}
+
+	// try to read the sector data and verify the root
+	data, err := a.volumes.Read(root)
+	if err != nil {
+		resp.Error = err.Error()
+	} else if calc := rhp2.SectorRoot(data); calc != root {
+		resp.Error = fmt.Sprintf("sector is corrupt: expected root %q, got %q", root, calc)
+	}
+	jc.Encode(resp)
+}

host/storage/persist.go

@@ -68,6 +68,8 @@ type (
 		ExpireTempSectors(height uint64) error
 		// IncrementSectorStats increments sector stats
 		IncrementSectorStats(reads, writes, cacheHit, cacheMiss uint64) error
+		// SectorReferences returns the references to a sector
+		SectorReferences(types.Hash256) (SectorReference, error)
 	}
 )
 

host/storage/storage.go

@@ -64,6 +64,13 @@ type (
 		Expiration uint64
 	}
 
+	// A SectorReference contains the references to a sector.
+	SectorReference struct {
+		Contracts   []types.FileContractID `json:"contracts"`
+		TempStorage int                    `json:"tempStorage"`
+		Locks       int                    `json:"locks"`
+	}
+
 	// A VolumeManager manages storage using local volumes.
 	VolumeManager struct {
 		cacheHits   uint64 // ensure 64-bit alignment on 32-bit systems
@@ -418,6 +425,11 @@ func (vm *VolumeManager) Close() error {
 	return nil
 }
 
+// SectorReferences returns the references to a sector.
+func (vm *VolumeManager) SectorReferences(root types.Hash256) (SectorReference, error) {
+	return vm.vs.SectorReferences(root)
+}
+
 // Usage returns the total and used storage space, in sectors, from the storage manager.
 func (vm *VolumeManager) Usage() (usedSectors uint64, totalSectors uint64, err error) {
 	done, err := vm.tg.Add()

persist/sqlite/sectors.go

@@ -178,6 +178,63 @@ func (s *Store) HasSector(root types.Hash256) (bool, error) {
 	return true, nil
 }
 
+// SectorReferences returns the references, if any of a sector root
+func (s *Store) SectorReferences(root types.Hash256) (refs storage.SectorReference, err error) {
+	err = s.transaction(func(tx txn) error {
+		dbID, err := sectorDBID(tx, root)
+		if err != nil {
+			return fmt.Errorf("failed to get sector id: %w", err)
+		}
+
+		// check if the sector is referenced by a contract
+		refs.Contracts, err = contractSectorRefs(tx, dbID)
+		if err != nil {
+			return fmt.Errorf("failed to get contracts: %w", err)
+		}
+
+		// check if the sector is referenced by temp storage
+		refs.TempStorage, err = getTempStorageCount(tx, dbID)
+		if err != nil {
+			return fmt.Errorf("failed to get temp storage: %w", err)
+		}
+
+		// check if the sector is locked
+		refs.Locks, err = getSectorLockCount(tx, dbID)
+		if err != nil {
+			return fmt.Errorf("failed to get locks: %w", err)
+		}
+		return nil
+	})
+	return
+}
+
+func contractSectorRefs(tx txn, sectorID int64) (contractIDs []types.FileContractID, err error) {
+	rows, err := tx.Query(`SELECT DISTINCT contract_id FROM contract_sector_roots WHERE sector_id=$1;`, sectorID)
+	if err != nil {
+		return nil, fmt.Errorf("failed to select contracts: %w", err)
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		var contractID types.FileContractID
+		if err := rows.Scan((*sqlHash256)(&contractID)); err != nil {
+			return nil, fmt.Errorf("failed to scan contract id: %w", err)
+		}
+		contractIDs = append(contractIDs, contractID)
+	}
+	return
+}
+
+func getTempStorageCount(tx txn, sectorID int64) (n int, err error) {
+	err = tx.QueryRow(`SELECT COUNT(*) FROM temp_storage_sector_roots WHERE sector_id=$1;`, sectorID).Scan(&n)
+	return
+}
+
+func getSectorLockCount(tx txn, sectorID int64) (n int, err error) {
+	err = tx.QueryRow(`SELECT COUNT(*) FROM locked_sectors WHERE sector_id=$1;`, sectorID).Scan(&n)
+	return
+}
+
 func incrementVolumeUsage(tx txn, volumeID int64, delta int) error {
 	var used int64
 	err := tx.QueryRow(`UPDATE storage_volumes SET used_sectors=used_sectors+$1 WHERE id=$2 RETURNING used_sectors;`, delta, volumeID).Scan(&used)