chore: sync upstream SiaFoundation/indexd

.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md

@@ -0,0 +1,5 @@
+---
+default: patch
+---
+
+# Reject sectors that have been uploaded more than the temporary storage duration ago

api/app/app.go

@@ -547,15 +547,8 @@ func (a *app) handlePOSTSlabs(jc jape.Context, pk types.PublicKey) {
 	if !ok {
 		return
 	}
-	for _, param := range params {
-		if err := param.Validate(); err != nil {
-			jc.Error(fmt.Errorf("invalid slab pin params: %w", err), http.StatusBadRequest)
-			return
-		}
-	}
-
 	slabIDs, err := a.slabs.PinSlabs(jc.Request.Context(), proto.Account(pk), time.Now().Add(6*time.Hour), params...)
-	if errors.Is(err, slabs.ErrBadHosts) || errors.Is(err, slabs.ErrMinShards) {
+	if errors.Is(err, slabs.ErrInvalidPinParams) || errors.Is(err, slabs.ErrBadHosts) || errors.Is(err, slabs.ErrMinShards) {
 		jc.Error(err, http.StatusBadRequest)
 		return
 	} else if jc.Check("failed to pin slab", err) != nil {

api/app/app_test.go

@@ -124,12 +124,14 @@ func uploadRandomSlab(t testing.TB, client *client.Client, sk types.PrivateKey,
 
 		// upload sector
 		hk := h.PublicKey
+		uploadedAt := time.Now()
 		if result, err := client.WriteSector(context.Background(), sk, hk, sector[:]); err != nil {
 			t.Fatal(err)
 		} else {
 			sectors = append(sectors, slabs.PinnedSector{
-				Root:    result.Root,
-				HostKey: hk,
+				Root:       result.Root,
+				HostKey:    hk,
+				UploadedAt: &uploadedAt,
 			})
 		}
 	}
@@ -209,8 +211,21 @@ func TestApplicationAPI(t *testing.T) {
 		t.Fatal("failed to unpin slab:", err)
 	}
 
-	// assert minimum redundancy is enforced
+	// assert upload times outside the accepted range are rejected
 	p := uploadRandomSlab(t, hc, sk, hosts)
+	tooOld := time.Now().Add(-slabs.MaxSlabUploadAge - time.Hour)
+	p.Sectors[3].UploadedAt = &tooOld
+	if _, err := client.PinSlabs(context.Background(), sk, p); !errors.Is(err, slabs.ErrSlabUploadTooOld) {
+		t.Fatal("expected stale upload error, got:", err)
+	}
+	inFuture := time.Now().Add(slabs.MaxSlabUploadSkew + time.Minute)
+	p.Sectors[3].UploadedAt = &inFuture
+	if _, err := client.PinSlabs(context.Background(), sk, p); !errors.Is(err, slabs.ErrSlabUploadInFuture) {
+		t.Fatal("expected future upload error, got:", err)
+	}
+
+	// assert minimum redundancy is enforced
+	p.Sectors[3].UploadedAt = nil
 	p.Sectors = p.Sectors[:5]
 	_, err = client.PinSlabs(context.Background(), sk, p)
 	if err == nil || !strings.Contains(err.Error(), "too low") {

api/app/client.go

@@ -68,6 +68,12 @@ func (e *HTTPError) Error() string {
 	return fmt.Sprintf("HTTP %d: %s", e.StatusCode, msg)
 }
 
+// Is matches target if its message is contained in the response body.
+func (e *HTTPError) Is(target error) bool {
+	msg := target.Error()
+	return msg != "" && strings.Contains(e.Body, msg)
+}
+
 // sign signs the request with the appropriate headers and returns the signed URL
 // and request body.
 func sign(appKey types.PrivateKey, validUntil time.Time, method, endpointURL string, requestBuf []byte) (*url.URL, io.Reader, error) {
@@ -187,7 +193,8 @@ func (c *Client) Hosts(ctx context.Context, appKey types.PrivateKey, opts ...api
 	return
 }
 
-// PinSlabs pins slabs to the indexer.
+// PinSlabs pins slabs to the indexer. A sector with an unacceptable upload time
+// is rejected with slabs.ErrSlabUploadTooOld or slabs.ErrSlabUploadInFuture.
 func (c *Client) PinSlabs(ctx context.Context, appKey types.PrivateKey, params ...slabs.SlabPinParams) (slabIDs []slabs.SlabID, err error) {
 	err = c.signedRequestJSON(ctx, appKey, http.MethodPost, "/slabs", params, &slabIDs)
 	return

contracts/maintenance.go

@@ -200,7 +200,7 @@ func (cm *ContractManager) maintenanceLoop(ctx context.Context) {
 			logError(cm.performSectorPinning(ctx, pinningLog), pinningLog)
 
 			unpinnableLog := log.Named("unpinnable")
-			threshold := time.Now().Add(-unpinnableSectorThreshold)
+			threshold := time.Now().Add(-UnpinnableSectorThreshold)
 			logError(cm.store.MarkSectorsUnpinnable(threshold), unpinnableLog)
 			log.Debug("maintenance complete")
 		}

contracts/manager.go

@@ -26,12 +26,15 @@ const (
 	pinTimeout   = 2 * time.Minute
 	pruneTimeout = 2 * time.Minute
 
-	unpinnableSectorThreshold = 3 * 24 * time.Hour
-
 	pruneIntervalSuccess = 24 * time.Hour
 	pruneIntervalFailure = 3 * time.Hour
 )
 
+// UnpinnableSectorThreshold is how long a sector may stay unpinned before it is
+// marked unpinnable, matching the duration hosts keep sectors in temporary
+// storage.
+const UnpinnableSectorThreshold = 3 * 24 * time.Hour
+
 var (
 	// DefaultMaintenanceSettings are the default settings for contract
 	// maintenance. These settings are configured in the database as defaults

openapi/app.yml

@@ -592,55 +592,79 @@ paths:
     post:
       tags:
         - slabs
-      summary: Pin a slab to the indexer
-      operationId: pinSlab
+      summary: Pin slabs to the indexer
+      operationId: pinSlabs
       requestBody:
         required: true
         content:
           application/json:
             schema:
-              type: object
-              properties:
-                version:
-                  type: integer
-                  format: uint8
-                  maximum: 1
-                  default: 0
-                  description: The slab encoding version.
-                encryptionKey:
-                  allOf:
-                    - $ref: "#/components/schemas/EncryptionKey"
-                    - description: The encryption key used to encrypt the shards
-                minShards:
-                  type: integer
-                  minimum: 1
-                  format: uint
-                  description: The number of data shards a piece gets erasure-coded into
-                sectors:
-                  type: array
-                  items:
-                    type: object
-                    properties:
-                      root:
-                        allOf:
-                          - $ref: "#/components/schemas/Hash256"
-                          - description: The root of the sector
-                      hostKey:
-                        allOf:
-                          - $ref: "#/components/schemas/PublicKey"
-                          - description: The public key of the host that stores the sector
+              type: array
+              items:
+                type: object
+                properties:
+                  version:
+                    type: integer
+                    format: uint8
+                    maximum: 1
+                    default: 0
+                    description: The slab encoding version.
+                  encryptionKey:
+                    allOf:
+                      - $ref: "#/components/schemas/EncryptionKey"
+                      - description: The encryption key used to encrypt the shards
+                  minShards:
+                    type: integer
+                    minimum: 1
+                    format: uint
+                    description: The number of data shards a piece gets erasure-coded into
+                  sectors:
+                    type: array
+                    items:
+                      type: object
+                      properties:
+                        root:
+                          allOf:
+                            - $ref: "#/components/schemas/Hash256"
+                            - description: The root of the sector
+                        hostKey:
+                          allOf:
+                            - $ref: "#/components/schemas/PublicKey"
+                            - description: The public key of the host that stores the sector
+                        uploadedAt:
+                          type: string
+                          format: date-time
+                          description: >-
+                            When the sector was written to the host. If
+                            provided, it must be no more than 48 hours old and
+                            no more than 5 minutes ahead of the indexer's clock,
+                            leaving enough time to pin the sector before
+                            temporary storage expires.
       responses:
-        "201":
-          description: Slab pinned successfully
+        "200":
+          description: Slabs pinned successfully
           content:
             application/json:
               schema:
-                type: object
-                properties:
-                  slabID:
-                    allOf:
-                      - $ref: "#/components/schemas/SlabID"
-                      - description: The ID of the pinned slab
+                type: array
+                description: The IDs of the pinned slabs, in request order
+                items:
+                  $ref: "#/components/schemas/SlabID"
+        "400":
+          description: >-
+            Invalid slab parameters. The stable "slab upload is too old" text
+            in the response body identifies a sector that must be re-uploaded
+            and names its index. "slab upload time is in the future" means the
+            client's clock is more than 5 minutes ahead.
+          content:
+            text/plain:
+              schema:
+                type: string
+              examples:
+                tooOld:
+                  value: "invalid slab pin params: slab 0: sector 3 invalid: slab upload is too old (max 48h0m0s)"
+                inFuture:
+                  value: "invalid slab pin params: slab 0: sector 3 invalid: slab upload time is in the future (max 5m0s ahead)"
 
       x-codeSamples:
         - lang: Go
@@ -650,16 +674,16 @@ paths:
 
             import (
               "context"
-              "go.sia.tech/indexd/app"
-              "go.sia.tech/indexd/slabs"
               "go.sia.tech/core/types"
+              "go.sia.tech/indexd/api/app"
+              "go.sia.tech/indexd/slabs"
             )
 
             func main() {
               var appKey types.PrivateKey
-              client, err := app.NewClient("http://localhost:9982", appKey)
-              var params slabs.SlabPinParams
-              id, err := client.PinSlab(context.Background(), params)
+              client := app.NewClient("http://localhost:9982")
+              var params []slabs.SlabPinParams
+              ids, err := client.PinSlabs(context.Background(), appKey, params...)
               // ...
             }
 

persist/postgres/sectors.go

@@ -268,7 +268,8 @@ func (s *Store) markFailingSectorsLostBatch(hostKey types.PublicKey, maxChecks,
 }
 
 // PinSlabs adds slabs to the database for pinning. The slabs are associated
-// with the provided account.
+// with the provided account. A sector's reported upload time, capped at now,
+// becomes its uploaded_at.
 func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, toPin ...slabs.SlabPinParams) ([]slabs.SlabID, error) {
 	var digests []slabs.SlabID
 	err := s.transaction(func(ctx context.Context, tx *txn) error {
@@ -371,21 +372,23 @@ func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, to
 
 			// insert the slab's sectors. For a slab that already
 			// exists this may rebind any sectors that were marked
-			// lost since it was pinned.
+			// lost since it was pinned. An existing sector keeps the
+			// later upload time.
 			batch := &pgx.Batch{}
 			for _, sector := range slab.Sectors {
 				batch.Queue(`
-				INSERT INTO sectors (sector_root, host_id, next_integrity_check)
-				SELECT $1, h.id, $3
+				INSERT INTO sectors (sector_root, host_id, next_integrity_check, uploaded_at)
+				SELECT $1, h.id, $3, LEAST(NOW(), $4::timestamptz)
 				FROM hosts h
 				WHERE h.public_key = $2
 				ON CONFLICT (sector_root) DO UPDATE SET
-					uploaded_at = NOW(),
+					uploaded_at = GREATEST(sectors.uploaded_at, EXCLUDED.uploaded_at),
 					host_id = COALESCE(sectors.host_id, EXCLUDED.host_id)
 				RETURNING id, host_id, (OLD.id IS NULL) AS inserted, (OLD.id IS NOT NULL AND OLD.host_id IS NULL) AS rebound`,
 					sqlHash256(sector.Root),
 					sqlPublicKey(sector.HostKey),
-					nextIntegrityCheck)
+					nextIntegrityCheck,
+					sector.UploadedAt)
 			}
 
 			var badHosts int