From 37852111d96305eba3c73d60b421c9256046e05b Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Sat, 22 Aug 2026 17:12:44 -0400 Subject: [PATCH 1/9] reject sectors that have been uploaded more than the temporary storage duration ago --- api/app/app.go | 10 ++- api/app/app_test.go | 18 +++++- api/app/client.go | 22 ++++++- contracts/maintenance.go | 2 +- contracts/manager.go | 7 ++- openapi/app.yml | 23 +++++++ persist/postgres/sectors.go | 15 +++-- persist/postgres/sectors_test.go | 105 ++++++++++++++++++++++++++----- slabs/slabs.go | 40 ++++++++++-- slabs/slabs_test.go | 38 +++++++++-- 10 files changed, 240 insertions(+), 40 deletions(-) diff --git a/api/app/app.go b/api/app/app.go index 61133c13..7827520e 100644 --- a/api/app/app.go +++ b/api/app/app.go @@ -547,15 +547,19 @@ func (a *app) handlePOSTSlabs(jc jape.Context, pk types.PublicKey) { if !ok { return } + now := time.Now() for _, param := range params { - if err := param.Validate(); err != nil { + if err := param.Validate(now); 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) { + slabIDs, err := a.slabs.PinSlabs(jc.Request.Context(), proto.Account(pk), now.Add(6*time.Hour), params...) + if errors.Is(err, slabs.ErrBadHosts) || + errors.Is(err, slabs.ErrMinShards) || + errors.Is(err, slabs.ErrSlabUploadTooOld) || + errors.Is(err, slabs.ErrSlabUploadInFuture) { jc.Error(err, http.StatusBadRequest) return } else if jc.Check("failed to pin slab", err) != nil { diff --git a/api/app/app_test.go b/api/app/app_test.go index 01a22428..d2e5a031 100644 --- a/api/app/app_test.go +++ b/api/app/app_test.go @@ -115,6 +115,8 @@ func newAccount(t *testing.T, cluster *testutils.Cluster) (types.PrivateKey, acc func uploadRandomSlab(t testing.TB, client *client.Client, sk types.PrivateKey, hosts []hosts.Host) slabs.SlabPinParams { t.Helper() + uploadedAt := time.Now() + // prepare sectors var sectors []slabs.PinnedSector for _, h := range hosts { @@ -137,6 +139,7 @@ func uploadRandomSlab(t testing.TB, client *client.Client, sk types.PrivateKey, EncryptionKey: frand.Entropy256(), MinShards: 4, Sectors: sectors, + UploadedAt: &uploadedAt, } } @@ -209,8 +212,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.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.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.UploadedAt = nil p.Sectors = p.Sectors[:5] _, err = client.PinSlabs(context.Background(), sk, p) if err == nil || !strings.Contains(err.Error(), "too low") { diff --git a/api/app/client.go b/api/app/client.go index 12635f60..088919ca 100644 --- a/api/app/client.go +++ b/api/app/client.go @@ -68,6 +68,21 @@ func (e *HTTPError) Error() string { return fmt.Sprintf("HTTP %d: %s", e.StatusCode, msg) } +// matchBadRequest joins err with the first of sentinels whose message appears +// in the body of a 400 response so callers can match it with errors.Is. +func matchBadRequest(err error, sentinels ...error) error { + httpErr, ok := errors.AsType[*HTTPError](err) + if !ok || httpErr.StatusCode != http.StatusBadRequest { + return err + } + for _, sentinel := range sentinels { + if strings.Contains(httpErr.Body, sentinel.Error()) { + return fmt.Errorf("%w: %w", sentinel, err) + } + } + return err +} + // 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,9 +202,12 @@ 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 slab with an unacceptable upload time +// is rejected with slabs.ErrSlabUploadTooOld, meaning it has to be re-uploaded, +// or slabs.ErrSlabUploadInFuture, meaning the client's clock is ahead. 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) + err = matchBadRequest(c.signedRequestJSON(ctx, appKey, http.MethodPost, "/slabs", params, &slabIDs), + slabs.ErrSlabUploadTooOld, slabs.ErrSlabUploadInFuture) return } diff --git a/contracts/maintenance.go b/contracts/maintenance.go index ab22fa91..5237ccf1 100644 --- a/contracts/maintenance.go +++ b/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") } diff --git a/contracts/manager.go b/contracts/manager.go index 6871d591..a1d933ca 100644 --- a/contracts/manager.go +++ b/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 diff --git a/openapi/app.yml b/openapi/app.yml index 432cfc32..deb7776c 100644 --- a/openapi/app.yml +++ b/openapi/app.yml @@ -629,6 +629,14 @@ paths: 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 upload of the slab's sectors began. 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 sectors before temporary storage expires. responses: "201": description: Slab pinned successfully @@ -641,6 +649,21 @@ paths: allOf: - $ref: "#/components/schemas/SlabID" - description: The ID of the pinned slab + "400": + description: >- + Invalid slab parameters. A client may identify an upload that must + be repeated by matching the stable "slab upload is too old" text + in the response body. "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 upload is too old (max 48h0m0s)" + inFuture: + value: "invalid slab pin params: slab upload time is in the future (max 5m0s ahead)" x-codeSamples: - lang: Go diff --git a/persist/postgres/sectors.go b/persist/postgres/sectors.go index 124040f5..46eef387 100644 --- a/persist/postgres/sectors.go +++ b/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 slab's upload time, capped at the current time, +// becomes the uploaded_at of its sectors. 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, + slab.UploadedAt) } var badHosts int diff --git a/persist/postgres/sectors_test.go b/persist/postgres/sectors_test.go index 9e33f9c5..19bfa879 100644 --- a/persist/postgres/sectors_test.go +++ b/persist/postgres/sectors_test.go @@ -123,20 +123,6 @@ func TestMigrateSector(t *testing.T) { t.Fatal(err) } - sectorUploadedAt := func(root types.Hash256) (uploadedAt time.Time) { - t.Helper() - - err := store.pool.QueryRow(t.Context(), ` - SELECT uploaded_at - FROM sectors - WHERE sector_root = $1 - `, sqlHash256(root)).Scan(&uploadedAt) - if err != nil { - t.Fatal(err) - } - return - } - // helper to assert sector state assertSector := func(root types.Hash256, expectedHostKey types.PublicKey, expectedContractID types.FileContractID, expectedFailures, expectedMigrated int) { t.Helper() @@ -179,7 +165,7 @@ func TestMigrateSector(t *testing.T) { migrate := func(root types.Hash256, hostKey types.PublicKey, expectedMigrated bool) { t.Helper() - beforeUploadedAt := sectorUploadedAt(root) + beforeUploadedAt := store.sectorUploadedAt(t, root) if migrated, err := store.MigrateSector(root, hostKey); err != nil { t.Fatal(err) } else if migrated != expectedMigrated { @@ -191,7 +177,7 @@ func TestMigrateSector(t *testing.T) { } } - afterUploadedAt := sectorUploadedAt(root) + afterUploadedAt := store.sectorUploadedAt(t, root) if expectedMigrated && afterUploadedAt.Compare(beforeUploadedAt) != 1 { t.Fatal("expected after uploaded at timestamp to be greater than before timestamp") @@ -1465,6 +1451,78 @@ func TestPinSlabsRebindLostSector(t *testing.T) { } } +// TestPinSlabsUploadedAt asserts that a slab's upload time becomes the +// uploaded_at of its sectors and that a re-pin can't move it backwards. +func TestPinSlabsUploadedAt(t *testing.T) { + store := initPostgres(t, zaptest.NewLogger(t).Named("postgres")) + account := proto.Account{1} + store.addTestAccount(t, types.PublicKey(account)) + + hk := store.addTestHost(t) + store.addTestContract(t, hk) + + assertUploadedAt := func(sectors []slabs.PinnedSector, expected time.Time) { + t.Helper() + + for _, sector := range sectors { + if ts := store.sectorUploadedAt(t, sector.Root); !ts.Equal(expected) { + t.Fatalf("expected uploaded_at %v, got %v", expected, ts) + } + } + } + + // a slab without an upload time falls back to now + before := time.Now().Add(-time.Second) + fresh := newTestSlab(hk) + store.pinTestSlabs(t, account, fresh) + if ts := store.sectorUploadedAt(t, fresh.Sectors[0].Root); ts.Before(before) { + t.Fatalf("expected uploaded_at at or after %v, got %v", before, ts) + } + + // an upload time is persisted + uploadedAt := time.Now().Add(-40 * time.Hour).Round(time.Microsecond) + stale := newTestSlab(hk) + stale.UploadedAt = &uploadedAt + store.pinTestSlabs(t, account, stale) + assertUploadedAt(stale.Sectors, uploadedAt) + + // an upload time in the future is capped at now + future := time.Now().Add(time.Hour) + ahead := newTestSlab(hk) + ahead.UploadedAt = &future + store.pinTestSlabs(t, account, ahead) + if ts := store.sectorUploadedAt(t, ahead.Sectors[0].Root); !ts.Before(future) { + t.Fatalf("expected uploaded_at before %v, got %v", future, ts) + } + + // re-pinning existing sectors keeps their upload time + repin := newTestSlab(hk, stale.Sectors...) + repin.UploadedAt = &uploadedAt + store.pinTestSlabs(t, account, repin) + assertUploadedAt(stale.Sectors, uploadedAt) + + // and can not move it backwards + older := uploadedAt.Add(-7 * time.Hour) + repin.UploadedAt = &older + store.pinTestSlabs(t, account, repin) + assertUploadedAt(stale.Sectors, uploadedAt) + + // re-uploading them does move it forward + refreshed := time.Now().Add(-time.Minute).Round(time.Microsecond) + repin.UploadedAt = &refreshed + store.pinTestSlabs(t, account, repin) + assertUploadedAt(stale.Sectors, refreshed) + + // a re-pin without an upload time still falls back to now + repin.UploadedAt = nil + store.pinTestSlabs(t, account, repin) + for _, sector := range stale.Sectors { + if ts := store.sectorUploadedAt(t, sector.Root); !ts.After(refreshed) { + t.Fatalf("expected uploaded_at after %v, got %v", refreshed, ts) + } + } +} + func TestUnpinSlab(t *testing.T) { store := initPostgres(t, zaptest.NewLogger(t).Named("postgres")) @@ -3917,3 +3975,18 @@ func setScannedHeight(t *testing.T, store *Store, height uint64) { t.Fatal(err) } } + +// sectorUploadedAt returns the sector's uploaded_at timestamp. +func (s *Store) sectorUploadedAt(t testing.TB, root types.Hash256) (uploadedAt time.Time) { + t.Helper() + + err := s.pool.QueryRow(t.Context(), ` + SELECT uploaded_at + FROM sectors + WHERE sector_root = $1 + `, sqlHash256(root)).Scan(&uploadedAt) + if err != nil { + t.Fatal(err) + } + return +} diff --git a/slabs/slabs.go b/slabs/slabs.go index d9e4c263..ae38eeab 100644 --- a/slabs/slabs.go +++ b/slabs/slabs.go @@ -9,6 +9,7 @@ import ( proto "go.sia.tech/core/rhp/v4" "go.sia.tech/core/types" + "go.sia.tech/indexd/contracts" ) const ( @@ -24,6 +25,17 @@ const ( maxSlabVersion = 1 ) +const ( + // MaxSlabUploadAge is the oldest upload time accepted when pinning a slab, + // leaving a day to pin its sectors before hosts delete them from temporary + // storage. + MaxSlabUploadAge = contracts.UnpinnableSectorThreshold - 24*time.Hour + + // MaxSlabUploadSkew is the furthest a slab's upload time may be ahead of + // the indexer's clock. + MaxSlabUploadSkew = 5 * time.Minute +) + var ( // ErrSlabNotFound is returned when a slab is not found in the database. ErrSlabNotFound = errors.New("slab not found") @@ -47,6 +59,15 @@ var ( // ErrUnsupportedSlabVersion is returned when attempting to pin a slab with // a version that is not yet supported. ErrUnsupportedSlabVersion = errors.New("unsupported slab version") + + // ErrSlabUploadTooOld is returned when attempting to pin a slab whose + // sectors may already have been deleted from temporary storage. Clients + // match these two by message, so keep the thresholds out of them. + ErrSlabUploadTooOld = errors.New("slab upload is too old") + + // ErrSlabUploadInFuture is returned when attempting to pin a slab whose + // upload time is ahead of the indexer's clock. + ErrSlabUploadInFuture = errors.New("slab upload time is in the future") ) type ( @@ -90,6 +111,7 @@ type ( EncryptionKey EncryptionKey `json:"encryptionKey"` MinShards uint `json:"minShards"` Sectors []PinnedSector `json:"sectors"` + UploadedAt *time.Time `json:"uploadedAt,omitempty"` } // A PinnedSlab is a slab that has been pinned to hosts. @@ -166,9 +188,10 @@ func (s SlabPinParams) DataSize() uint64 { } // Validate checks if the SlabPinParams are valid. It ensures that the -// encryption key is set, the minimum number of shards is met, and that there -// are no duplicate host keys or empty roots in the sectors. -func (s SlabPinParams) Validate() error { +// encryption key is set, the minimum number of shards is met, the optional +// upload time is recent relative to now, and that there are no duplicate host +// keys or empty roots in the sectors. +func (s SlabPinParams) Validate(now time.Time) error { if s.Version > maxSlabVersion { return fmt.Errorf("%w: %d", ErrUnsupportedSlabVersion, s.Version) } else if s.EncryptionKey == ([32]byte{}) { @@ -177,6 +200,14 @@ func (s SlabPinParams) Validate() error { return err } + if s.UploadedAt != nil { + if s.UploadedAt.Before(now.Add(-MaxSlabUploadAge)) { + return fmt.Errorf("%w (max %v)", ErrSlabUploadTooOld, MaxSlabUploadAge) + } else if s.UploadedAt.After(now.Add(MaxSlabUploadSkew)) { + return fmt.Errorf("%w (max %v ahead)", ErrSlabUploadInFuture, MaxSlabUploadSkew) + } + } + hks := make(map[types.PublicKey]struct{}, len(s.Sectors)) for i, sector := range s.Sectors { if sector.Root == (types.Hash256{}) { @@ -195,8 +226,9 @@ func (s SlabPinParams) Validate() error { // PinSlabs adds slabs to the database for pinning. The slabs are associated // with the provided account. func (m *SlabManager) PinSlabs(ctx context.Context, account proto.Account, nextIntegrityCheck time.Time, toPin ...SlabPinParams) ([]SlabID, error) { + now := time.Now() for i := range toPin { - if err := toPin[i].Validate(); err != nil { + if err := toPin[i].Validate(now); err != nil { return nil, fmt.Errorf("slab %d invalid: %w", i, err) } } diff --git a/slabs/slabs_test.go b/slabs/slabs_test.go index 02d8a26e..0baa1a84 100644 --- a/slabs/slabs_test.go +++ b/slabs/slabs_test.go @@ -1,9 +1,11 @@ package slabs_test import ( + "errors" "fmt" "strings" "testing" + "time" proto "go.sia.tech/core/rhp/v4" "go.sia.tech/core/types" @@ -25,32 +27,51 @@ func TestSlabPinParamsValidate(t *testing.T) { return s }(), } - if err := params.Validate(); err != nil { + now := time.Now() + if err := params.Validate(now); err != nil { t.Fatal("unexpected", err) } + // assert the accepted upload time range is inclusive at both bounds + for _, tc := range []struct { + name string + uploadedAt time.Time + want error + }{ + {"oldest accepted", now.Add(-slabs.MaxSlabUploadAge), nil}, + {"one ns too old", now.Add(-slabs.MaxSlabUploadAge - 1), slabs.ErrSlabUploadTooOld}, + {"furthest ahead accepted", now.Add(slabs.MaxSlabUploadSkew), nil}, + {"one ns too far ahead", now.Add(slabs.MaxSlabUploadSkew + 1), slabs.ErrSlabUploadInFuture}, + } { + params.UploadedAt = &tc.uploadedAt + if err := params.Validate(now); !errors.Is(err, tc.want) { + t.Fatalf("%s: expected %v, got %v", tc.name, tc.want, err) + } + } + params.UploadedAt = nil + // assert empty encryption key is illegal params.EncryptionKey = [32]byte{} - if err := params.Validate(); err == nil || !strings.Contains(err.Error(), "encryption key is empty") { + if err := params.Validate(now); err == nil || !strings.Contains(err.Error(), "encryption key is empty") { t.Fatal("unexpected", err) } // assert duplicate host keys are illegal params.EncryptionKey = frand.Entropy256() params.Sectors[2] = params.Sectors[1] - if err := params.Validate(); err == nil || !strings.Contains(err.Error(), "duplicate host key") { + if err := params.Validate(now); err == nil || !strings.Contains(err.Error(), "duplicate host key") { t.Fatal("unexpected", err) } // assert insufficient redundancy is illegal params.Sectors = params.Sectors[:10] - if err := params.Validate(); err == nil || !strings.Contains(err.Error(), "is too low") { + if err := params.Validate(now); err == nil || !strings.Contains(err.Error(), "is too low") { t.Fatal("unexpected", err) } // assert exceeding max total shards is illegal params.Sectors = make([]slabs.PinnedSector, slabs.MaxTotalShards+1) - if err := params.Validate(); err == nil || !strings.Contains(err.Error(), "exceeds maximum") { + if err := params.Validate(now); err == nil || !strings.Contains(err.Error(), "exceeds maximum") { t.Fatal("unexpected", err) } } @@ -85,6 +106,13 @@ func TestSlabPinParamsDigest(t *testing.T) { if slabID != expectedID { t.Fatalf("expected %v, got %v", expectedID, slabID) } + + // assert the upload time does not affect the slab ID + uploadedAt := time.Now() + params.UploadedAt = &uploadedAt + if digest := params.Digest(); digest != slabID { + t.Fatalf("expected upload time to preserve slab ID %v, got %v", slabID, digest) + } } func TestSlabVersionDigest(t *testing.T) { From b0583b8e94d8322fd7b827294cc1c5acfee6c802 Mon Sep 17 00:00:00 2001 From: "knope-bot[bot]" <152252888+knope-bot[bot]@users.noreply.github.com> Date: Sat, 22 Aug 2026 21:14:09 +0000 Subject: [PATCH 2/9] Auto generate changeset --- ...oaded_more_than_the_temporary_storage_duration_ago.md | 9 +++++++++ 1 file changed, 9 insertions(+) create mode 100644 .changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md diff --git a/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md b/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md new file mode 100644 index 00000000..69008320 --- /dev/null +++ b/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md @@ -0,0 +1,9 @@ +--- +default: patch +--- + +# Reject sectors that have been uploaded more than the temporary storage duration ago + +#1068 by @chris124567 + +Close #1067 From 783fee9b50c5901d931bcaf68896996d3ae2f53e Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Sat, 22 Aug 2026 17:26:55 -0400 Subject: [PATCH 3/9] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- persist/postgres/sectors.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/persist/postgres/sectors.go b/persist/postgres/sectors.go index 46eef387..89c5af46 100644 --- a/persist/postgres/sectors.go +++ b/persist/postgres/sectors.go @@ -377,9 +377,7 @@ func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, to batch := &pgx.Batch{} for _, sector := range slab.Sectors { batch.Queue(` - INSERT INTO sectors (sector_root, host_id, next_integrity_check, uploaded_at) - SELECT $1, h.id, $3, LEAST(NOW(), $4::timestamptz) - FROM hosts h + SELECT $1, h.id, $3, LEAST(NOW(), COALESCE($4::timestamptz, NOW())) WHERE h.public_key = $2 ON CONFLICT (sector_root) DO UPDATE SET uploaded_at = GREATEST(sectors.uploaded_at, EXCLUDED.uploaded_at), From e9ce74d1e03674578119d14c0e68c9ff65b8a33e Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Sat, 22 Aug 2026 17:14:22 -0400 Subject: [PATCH 4/9] fix changeset --- ...n_uploaded_more_than_the_temporary_storage_duration_ago.md | 4 ---- 1 file changed, 4 deletions(-) diff --git a/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md b/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md index 69008320..d96957d9 100644 --- a/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md +++ b/.changeset/reject_sectors_that_have_been_uploaded_more_than_the_temporary_storage_duration_ago.md @@ -3,7 +3,3 @@ default: patch --- # Reject sectors that have been uploaded more than the temporary storage duration ago - -#1068 by @chris124567 - -Close #1067 From a343efe17a37cc4bfe3bc105ef6d944616df798d Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Sat, 22 Aug 2026 18:10:51 -0400 Subject: [PATCH 5/9] fix test --- persist/postgres/sectors.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/persist/postgres/sectors.go b/persist/postgres/sectors.go index 89c5af46..814dabab 100644 --- a/persist/postgres/sectors.go +++ b/persist/postgres/sectors.go @@ -377,7 +377,9 @@ func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, to batch := &pgx.Batch{} for _, sector := range slab.Sectors { batch.Queue(` + INSERT INTO sectors (sector_root, host_id, next_integrity_check, uploaded_at) SELECT $1, h.id, $3, LEAST(NOW(), COALESCE($4::timestamptz, NOW())) + FROM hosts h WHERE h.public_key = $2 ON CONFLICT (sector_root) DO UPDATE SET uploaded_at = GREATEST(sectors.uploaded_at, EXCLUDED.uploaded_at), From e9feeb9c73ba90f726e8cfb683c5b5940708a6c0 Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Tue, 25 Aug 2026 14:11:14 -0400 Subject: [PATCH 6/9] review fix --- api/app/app.go | 9 ++++--- api/app/app_test.go | 15 +++++------ api/app/client.go | 27 +++++++------------- openapi/app.yml | 29 +++++++++++---------- persist/postgres/sectors.go | 6 ++--- persist/postgres/sectors_test.go | 40 ++++++++++++++++++----------- slabs/slabs.go | 44 ++++++++++++++++---------------- slabs/slabs_test.go | 8 +++--- 8 files changed, 90 insertions(+), 88 deletions(-) diff --git a/api/app/app.go b/api/app/app.go index 7827520e..bc4bb2aa 100644 --- a/api/app/app.go +++ b/api/app/app.go @@ -20,6 +20,7 @@ import ( "go.sia.tech/coreutils/rhp/v4/siamux" "go.sia.tech/indexd/accounts" "go.sia.tech/indexd/api" + "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/hosts" "go.sia.tech/indexd/sharing" "go.sia.tech/indexd/slabs" @@ -556,10 +557,10 @@ func (a *app) handlePOSTSlabs(jc jape.Context, pk types.PublicKey) { } slabIDs, err := a.slabs.PinSlabs(jc.Request.Context(), proto.Account(pk), now.Add(6*time.Hour), params...) - if errors.Is(err, slabs.ErrBadHosts) || - errors.Is(err, slabs.ErrMinShards) || - errors.Is(err, slabs.ErrSlabUploadTooOld) || - errors.Is(err, slabs.ErrSlabUploadInFuture) { + if statusErr, ok := errors.AsType[*apierr.StatusError](err); ok { + jc.Error(err, statusErr.Status) + return + } else if 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 { diff --git a/api/app/app_test.go b/api/app/app_test.go index d2e5a031..41b519f5 100644 --- a/api/app/app_test.go +++ b/api/app/app_test.go @@ -115,8 +115,6 @@ func newAccount(t *testing.T, cluster *testutils.Cluster) (types.PrivateKey, acc func uploadRandomSlab(t testing.TB, client *client.Client, sk types.PrivateKey, hosts []hosts.Host) slabs.SlabPinParams { t.Helper() - uploadedAt := time.Now() - // prepare sectors var sectors []slabs.PinnedSector for _, h := range hosts { @@ -126,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, }) } } @@ -139,7 +139,6 @@ func uploadRandomSlab(t testing.TB, client *client.Client, sk types.PrivateKey, EncryptionKey: frand.Entropy256(), MinShards: 4, Sectors: sectors, - UploadedAt: &uploadedAt, } } @@ -215,18 +214,18 @@ func TestApplicationAPI(t *testing.T) { // assert upload times outside the accepted range are rejected p := uploadRandomSlab(t, hc, sk, hosts) tooOld := time.Now().Add(-slabs.MaxSlabUploadAge - time.Hour) - p.UploadedAt = &tooOld + 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.UploadedAt = &inFuture + 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.UploadedAt = nil + 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") { diff --git a/api/app/client.go b/api/app/client.go index 088919ca..0b8ce2dc 100644 --- a/api/app/client.go +++ b/api/app/client.go @@ -17,6 +17,7 @@ import ( "go.sia.tech/core/types" "go.sia.tech/indexd/api" + "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/hosts" "go.sia.tech/indexd/sharing" "go.sia.tech/indexd/slabs" @@ -68,19 +69,11 @@ func (e *HTTPError) Error() string { return fmt.Sprintf("HTTP %d: %s", e.StatusCode, msg) } -// matchBadRequest joins err with the first of sentinels whose message appears -// in the body of a 400 response so callers can match it with errors.Is. -func matchBadRequest(err error, sentinels ...error) error { - httpErr, ok := errors.AsType[*HTTPError](err) - if !ok || httpErr.StatusCode != http.StatusBadRequest { - return err - } - for _, sentinel := range sentinels { - if strings.Contains(httpErr.Body, sentinel.Error()) { - return fmt.Errorf("%w: %w", sentinel, err) - } - } - return err +// Is matches target if it is an *apierr.StatusError with the same status code +// and a message contained in the response body. +func (e *HTTPError) Is(target error) bool { + se, ok := errors.AsType[*apierr.StatusError](target) + return ok && e.StatusCode == se.Status && strings.Contains(e.Body, se.Message) } // sign signs the request with the appropriate headers and returns the signed URL @@ -202,12 +195,10 @@ func (c *Client) Hosts(ctx context.Context, appKey types.PrivateKey, opts ...api return } -// PinSlabs pins slabs to the indexer. A slab with an unacceptable upload time -// is rejected with slabs.ErrSlabUploadTooOld, meaning it has to be re-uploaded, -// or slabs.ErrSlabUploadInFuture, meaning the client's clock is ahead. +// 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 = matchBadRequest(c.signedRequestJSON(ctx, appKey, http.MethodPost, "/slabs", params, &slabIDs), - slabs.ErrSlabUploadTooOld, slabs.ErrSlabUploadInFuture) + err = c.signedRequestJSON(ctx, appKey, http.MethodPost, "/slabs", params, &slabIDs) return } diff --git a/openapi/app.yml b/openapi/app.yml index deb7776c..7d8698c2 100644 --- a/openapi/app.yml +++ b/openapi/app.yml @@ -629,14 +629,15 @@ paths: 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 upload of the slab's sectors began. 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 sectors before temporary storage expires. + 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 @@ -651,19 +652,19 @@ paths: - description: The ID of the pinned slab "400": description: >- - Invalid slab parameters. A client may identify an upload that must - be repeated by matching the stable "slab upload is too old" text - in the response body. "slab upload time is in the future" means - the client's clock is more than 5 minutes ahead. + 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 upload is too old (max 48h0m0s)" + value: "invalid slab pin params: sector 3 invalid: slab upload is too old (max 48h0m0s)" inFuture: - value: "invalid slab pin params: slab upload time is in the future (max 5m0s ahead)" + value: "invalid slab pin params: sector 3 invalid: slab upload time is in the future (max 5m0s ahead)" x-codeSamples: - lang: Go diff --git a/persist/postgres/sectors.go b/persist/postgres/sectors.go index 814dabab..55057253 100644 --- a/persist/postgres/sectors.go +++ b/persist/postgres/sectors.go @@ -268,8 +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. A slab's upload time, capped at the current time, -// becomes the uploaded_at of its sectors. +// 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 { @@ -388,7 +388,7 @@ func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, to sqlHash256(sector.Root), sqlPublicKey(sector.HostKey), nextIntegrityCheck, - slab.UploadedAt) + sector.UploadedAt) } var badHosts int diff --git a/persist/postgres/sectors_test.go b/persist/postgres/sectors_test.go index 19bfa879..3afe3c5b 100644 --- a/persist/postgres/sectors_test.go +++ b/persist/postgres/sectors_test.go @@ -1451,8 +1451,8 @@ func TestPinSlabsRebindLostSector(t *testing.T) { } } -// TestPinSlabsUploadedAt asserts that a slab's upload time becomes the -// uploaded_at of its sectors and that a re-pin can't move it backwards. +// TestPinSlabsUploadedAt asserts that a sector's reported upload time becomes +// its uploaded_at and that a re-pin can't move it backwards. func TestPinSlabsUploadedAt(t *testing.T) { store := initPostgres(t, zaptest.NewLogger(t).Named("postgres")) account := proto.Account{1} @@ -1461,6 +1461,13 @@ func TestPinSlabsUploadedAt(t *testing.T) { hk := store.addTestHost(t) store.addTestContract(t, hk) + setUploadedAt := func(params slabs.SlabPinParams, ts *time.Time) slabs.SlabPinParams { + for i := range params.Sectors { + params.Sectors[i].UploadedAt = ts + } + return params + } + assertUploadedAt := func(sectors []slabs.PinnedSector, expected time.Time) { t.Helper() @@ -1471,7 +1478,7 @@ func TestPinSlabsUploadedAt(t *testing.T) { } } - // a slab without an upload time falls back to now + // a sector without an upload time falls back to now before := time.Now().Add(-time.Second) fresh := newTestSlab(hk) store.pinTestSlabs(t, account, fresh) @@ -1481,41 +1488,44 @@ func TestPinSlabsUploadedAt(t *testing.T) { // an upload time is persisted uploadedAt := time.Now().Add(-40 * time.Hour).Round(time.Microsecond) - stale := newTestSlab(hk) - stale.UploadedAt = &uploadedAt + stale := setUploadedAt(newTestSlab(hk), &uploadedAt) store.pinTestSlabs(t, account, stale) assertUploadedAt(stale.Sectors, uploadedAt) + // sectors of one slab keep their own upload times + mixed := newTestSlab(hk) + earlier := uploadedAt.Add(-time.Hour) + mixed.Sectors[0].UploadedAt = &earlier + mixed.Sectors[1].UploadedAt = &uploadedAt + store.pinTestSlabs(t, account, mixed) + assertUploadedAt(mixed.Sectors[:1], earlier) + assertUploadedAt(mixed.Sectors[1:], uploadedAt) + // an upload time in the future is capped at now future := time.Now().Add(time.Hour) - ahead := newTestSlab(hk) - ahead.UploadedAt = &future + ahead := setUploadedAt(newTestSlab(hk), &future) store.pinTestSlabs(t, account, ahead) if ts := store.sectorUploadedAt(t, ahead.Sectors[0].Root); !ts.Before(future) { t.Fatalf("expected uploaded_at before %v, got %v", future, ts) } // re-pinning existing sectors keeps their upload time - repin := newTestSlab(hk, stale.Sectors...) - repin.UploadedAt = &uploadedAt + repin := setUploadedAt(newTestSlab(hk, stale.Sectors...), &uploadedAt) store.pinTestSlabs(t, account, repin) assertUploadedAt(stale.Sectors, uploadedAt) // and can not move it backwards older := uploadedAt.Add(-7 * time.Hour) - repin.UploadedAt = &older - store.pinTestSlabs(t, account, repin) + store.pinTestSlabs(t, account, setUploadedAt(repin, &older)) assertUploadedAt(stale.Sectors, uploadedAt) // re-uploading them does move it forward refreshed := time.Now().Add(-time.Minute).Round(time.Microsecond) - repin.UploadedAt = &refreshed - store.pinTestSlabs(t, account, repin) + store.pinTestSlabs(t, account, setUploadedAt(repin, &refreshed)) assertUploadedAt(stale.Sectors, refreshed) // a re-pin without an upload time still falls back to now - repin.UploadedAt = nil - store.pinTestSlabs(t, account, repin) + store.pinTestSlabs(t, account, setUploadedAt(repin, nil)) for _, sector := range stale.Sectors { if ts := store.sectorUploadedAt(t, sector.Root); !ts.After(refreshed) { t.Fatalf("expected uploaded_at after %v, got %v", refreshed, ts) diff --git a/slabs/slabs.go b/slabs/slabs.go index ae38eeab..e1cee280 100644 --- a/slabs/slabs.go +++ b/slabs/slabs.go @@ -5,10 +5,12 @@ import ( "errors" "fmt" "math" + "net/http" "time" proto "go.sia.tech/core/rhp/v4" "go.sia.tech/core/types" + "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/contracts" ) @@ -60,14 +62,14 @@ var ( // a version that is not yet supported. ErrUnsupportedSlabVersion = errors.New("unsupported slab version") - // ErrSlabUploadTooOld is returned when attempting to pin a slab whose - // sectors may already have been deleted from temporary storage. Clients - // match these two by message, so keep the thresholds out of them. - ErrSlabUploadTooOld = errors.New("slab upload is too old") + // ErrSlabUploadTooOld is returned when pinning a sector that may already + // have been deleted from temporary storage. Clients match these two by + // message, so keep the thresholds out of them. + ErrSlabUploadTooOld = apierr.New(http.StatusBadRequest, "slab upload is too old") - // ErrSlabUploadInFuture is returned when attempting to pin a slab whose - // upload time is ahead of the indexer's clock. - ErrSlabUploadInFuture = errors.New("slab upload time is in the future") + // ErrSlabUploadInFuture is returned when a sector's upload time is ahead of + // the indexer's clock. + ErrSlabUploadInFuture = apierr.New(http.StatusBadRequest, "slab upload time is in the future") ) type ( @@ -99,10 +101,12 @@ type ( PinnedAt time.Time `json:"pinnedAt"` } - // A PinnedSector is a sector that has been pinned to a host. + // A PinnedSector is a sector that has been pinned to a host. UploadedAt is + // when it was written, if reported. PinnedSector struct { - Root types.Hash256 `json:"root"` - HostKey types.PublicKey `json:"hostKey"` + Root types.Hash256 `json:"root"` + HostKey types.PublicKey `json:"hostKey"` + UploadedAt *time.Time `json:"uploadedAt,omitempty"` } // SlabPinParams is the input to PinSlabs @@ -111,7 +115,6 @@ type ( EncryptionKey EncryptionKey `json:"encryptionKey"` MinShards uint `json:"minShards"` Sectors []PinnedSector `json:"sectors"` - UploadedAt *time.Time `json:"uploadedAt,omitempty"` } // A PinnedSlab is a slab that has been pinned to hosts. @@ -188,9 +191,8 @@ func (s SlabPinParams) DataSize() uint64 { } // Validate checks if the SlabPinParams are valid. It ensures that the -// encryption key is set, the minimum number of shards is met, the optional -// upload time is recent relative to now, and that there are no duplicate host -// keys or empty roots in the sectors. +// encryption key is set, the minimum number of shards is met, and that the +// sectors have unique host keys, non-empty roots and recent upload times. func (s SlabPinParams) Validate(now time.Time) error { if s.Version > maxSlabVersion { return fmt.Errorf("%w: %d", ErrUnsupportedSlabVersion, s.Version) @@ -200,14 +202,6 @@ func (s SlabPinParams) Validate(now time.Time) error { return err } - if s.UploadedAt != nil { - if s.UploadedAt.Before(now.Add(-MaxSlabUploadAge)) { - return fmt.Errorf("%w (max %v)", ErrSlabUploadTooOld, MaxSlabUploadAge) - } else if s.UploadedAt.After(now.Add(MaxSlabUploadSkew)) { - return fmt.Errorf("%w (max %v ahead)", ErrSlabUploadInFuture, MaxSlabUploadSkew) - } - } - hks := make(map[types.PublicKey]struct{}, len(s.Sectors)) for i, sector := range s.Sectors { if sector.Root == (types.Hash256{}) { @@ -216,6 +210,12 @@ func (s SlabPinParams) Validate(now time.Time) error { return fmt.Errorf("sector %d invalid: host key is empty", i) } else if _, exists := hks[sector.HostKey]; exists { return fmt.Errorf("sector %d is invalid: duplicate host key %q", i, sector.HostKey) + } else if sector.UploadedAt != nil { + if sector.UploadedAt.Before(now.Add(-MaxSlabUploadAge)) { + return fmt.Errorf("sector %d invalid: %w (max %v)", i, ErrSlabUploadTooOld, MaxSlabUploadAge) + } else if sector.UploadedAt.After(now.Add(MaxSlabUploadSkew)) { + return fmt.Errorf("sector %d invalid: %w (max %v ahead)", i, ErrSlabUploadInFuture, MaxSlabUploadSkew) + } } hks[sector.HostKey] = struct{}{} } diff --git a/slabs/slabs_test.go b/slabs/slabs_test.go index 0baa1a84..f41cc78e 100644 --- a/slabs/slabs_test.go +++ b/slabs/slabs_test.go @@ -43,12 +43,12 @@ func TestSlabPinParamsValidate(t *testing.T) { {"furthest ahead accepted", now.Add(slabs.MaxSlabUploadSkew), nil}, {"one ns too far ahead", now.Add(slabs.MaxSlabUploadSkew + 1), slabs.ErrSlabUploadInFuture}, } { - params.UploadedAt = &tc.uploadedAt + params.Sectors[1].UploadedAt = &tc.uploadedAt if err := params.Validate(now); !errors.Is(err, tc.want) { t.Fatalf("%s: expected %v, got %v", tc.name, tc.want, err) } } - params.UploadedAt = nil + params.Sectors[1].UploadedAt = nil // assert empty encryption key is illegal params.EncryptionKey = [32]byte{} @@ -107,9 +107,9 @@ func TestSlabPinParamsDigest(t *testing.T) { t.Fatalf("expected %v, got %v", expectedID, slabID) } - // assert the upload time does not affect the slab ID + // assert a sector's upload time does not affect the slab ID uploadedAt := time.Now() - params.UploadedAt = &uploadedAt + params.Sectors[0].UploadedAt = &uploadedAt if digest := params.Digest(); digest != slabID { t.Fatalf("expected upload time to preserve slab ID %v, got %v", slabID, digest) } From 89a95730bcfb68c07a0b6031008370c7dc3c0991 Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Tue, 25 Aug 2026 14:15:09 -0400 Subject: [PATCH 7/9] add missing package --- api/apierr/apierr.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 api/apierr/apierr.go diff --git a/api/apierr/apierr.go b/api/apierr/apierr.go new file mode 100644 index 00000000..0a7a2e1e --- /dev/null +++ b/api/apierr/apierr.go @@ -0,0 +1,19 @@ +// Package apierr defines errors that carry the status code the API returns +// them with. +package apierr + +// A StatusError is an error the API returns with a specific status code. +type StatusError struct { + Status int + Message string +} + +// New returns a StatusError with the given status and message. +func New(status int, message string) *StatusError { + return &StatusError{Status: status, Message: message} +} + +// Error implements the error interface. +func (e *StatusError) Error() string { + return e.Message +} From b11acabdad988314197eb8a76ef89076f90299bf Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Tue, 25 Aug 2026 15:00:23 -0400 Subject: [PATCH 8/9] review fix --- api/app/app.go | 10 +---- openapi/app.yml | 112 +++++++++++++++++++++++------------------------ slabs/objects.go | 9 +++- slabs/slabs.go | 6 ++- 4 files changed, 69 insertions(+), 68 deletions(-) diff --git a/api/app/app.go b/api/app/app.go index bc4bb2aa..7f6b0805 100644 --- a/api/app/app.go +++ b/api/app/app.go @@ -548,15 +548,7 @@ func (a *app) handlePOSTSlabs(jc jape.Context, pk types.PublicKey) { if !ok { return } - now := time.Now() - for _, param := range params { - if err := param.Validate(now); 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), now.Add(6*time.Hour), params...) + slabIDs, err := a.slabs.PinSlabs(jc.Request.Context(), proto.Account(pk), time.Now().Add(6*time.Hour), params...) if statusErr, ok := errors.AsType[*apierr.StatusError](err); ok { jc.Error(err, statusErr.Status) return diff --git a/openapi/app.yml b/openapi/app.yml index 7d8698c2..043d6200 100644 --- a/openapi/app.yml +++ b/openapi/app.yml @@ -592,64 +592,64 @@ 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 - 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. + 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 @@ -662,9 +662,9 @@ paths: type: string examples: tooOld: - value: "invalid slab pin params: sector 3 invalid: slab upload is too old (max 48h0m0s)" + value: "invalid slab pin params: slab 0: sector 3 invalid: slab upload is too old (max 48h0m0s)" inFuture: - value: "invalid slab pin params: sector 3 invalid: slab upload time is in the future (max 5m0s ahead)" + value: "invalid slab pin params: slab 0: sector 3 invalid: slab upload time is in the future (max 5m0s ahead)" x-codeSamples: - lang: Go @@ -674,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...) // ... } diff --git a/slabs/objects.go b/slabs/objects.go index 3c925fd7..a7b4d1be 100644 --- a/slabs/objects.go +++ b/slabs/objects.go @@ -346,13 +346,18 @@ func (s SlabSlice) Pin() SlabPinParams { } } -// Slice creates a SlabSlice from the SlabPinParams. +// Slice creates a SlabSlice from the SlabPinParams. Upload times are dropped; +// they describe the upload, not the stored slab. func (s SlabPinParams) Slice(offset, length uint32) SlabSlice { + sectors := slices.Clone(s.Sectors) + for i := range sectors { + sectors[i].UploadedAt = nil + } return SlabSlice{ Version: s.Version, EncryptionKey: s.EncryptionKey, MinShards: s.MinShards, - Sectors: slices.Clone(s.Sectors), + Sectors: sectors, Offset: offset, Length: length, } diff --git a/slabs/slabs.go b/slabs/slabs.go index e1cee280..8318ed06 100644 --- a/slabs/slabs.go +++ b/slabs/slabs.go @@ -62,6 +62,10 @@ var ( // a version that is not yet supported. ErrUnsupportedSlabVersion = errors.New("unsupported slab version") + // ErrInvalidPinParams is returned when the params of a slab to pin are + // invalid. It wraps the underlying reason. + ErrInvalidPinParams = apierr.New(http.StatusBadRequest, "invalid slab pin params") + // ErrSlabUploadTooOld is returned when pinning a sector that may already // have been deleted from temporary storage. Clients match these two by // message, so keep the thresholds out of them. @@ -229,7 +233,7 @@ func (m *SlabManager) PinSlabs(ctx context.Context, account proto.Account, nextI now := time.Now() for i := range toPin { if err := toPin[i].Validate(now); err != nil { - return nil, fmt.Errorf("slab %d invalid: %w", i, err) + return nil, fmt.Errorf("%w: slab %d: %w", ErrInvalidPinParams, i, err) } } return m.store.PinSlabs(account, nextIntegrityCheck, toPin...) From eb58007edc01773e3f9cc112cb853ba6bbfe8419 Mon Sep 17 00:00:00 2001 From: Christopher Tarry Date: Fri, 28 Aug 2026 10:39:36 -0400 Subject: [PATCH 9/9] review fix --- api/apierr/apierr.go | 19 ------------------- api/app/app.go | 6 +----- api/app/client.go | 8 +++----- persist/postgres/sectors.go | 2 +- persist/postgres/sectors_test.go | 1 + slabs/slabs.go | 8 +++----- 6 files changed, 9 insertions(+), 35 deletions(-) delete mode 100644 api/apierr/apierr.go diff --git a/api/apierr/apierr.go b/api/apierr/apierr.go deleted file mode 100644 index 0a7a2e1e..00000000 --- a/api/apierr/apierr.go +++ /dev/null @@ -1,19 +0,0 @@ -// Package apierr defines errors that carry the status code the API returns -// them with. -package apierr - -// A StatusError is an error the API returns with a specific status code. -type StatusError struct { - Status int - Message string -} - -// New returns a StatusError with the given status and message. -func New(status int, message string) *StatusError { - return &StatusError{Status: status, Message: message} -} - -// Error implements the error interface. -func (e *StatusError) Error() string { - return e.Message -} diff --git a/api/app/app.go b/api/app/app.go index 7f6b0805..369b25e3 100644 --- a/api/app/app.go +++ b/api/app/app.go @@ -20,7 +20,6 @@ import ( "go.sia.tech/coreutils/rhp/v4/siamux" "go.sia.tech/indexd/accounts" "go.sia.tech/indexd/api" - "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/hosts" "go.sia.tech/indexd/sharing" "go.sia.tech/indexd/slabs" @@ -549,10 +548,7 @@ func (a *app) handlePOSTSlabs(jc jape.Context, pk types.PublicKey) { return } slabIDs, err := a.slabs.PinSlabs(jc.Request.Context(), proto.Account(pk), time.Now().Add(6*time.Hour), params...) - if statusErr, ok := errors.AsType[*apierr.StatusError](err); ok { - jc.Error(err, statusErr.Status) - return - } else 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 { diff --git a/api/app/client.go b/api/app/client.go index 0b8ce2dc..07c4aca0 100644 --- a/api/app/client.go +++ b/api/app/client.go @@ -17,7 +17,6 @@ import ( "go.sia.tech/core/types" "go.sia.tech/indexd/api" - "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/hosts" "go.sia.tech/indexd/sharing" "go.sia.tech/indexd/slabs" @@ -69,11 +68,10 @@ func (e *HTTPError) Error() string { return fmt.Sprintf("HTTP %d: %s", e.StatusCode, msg) } -// Is matches target if it is an *apierr.StatusError with the same status code -// and a message contained in the response body. +// Is matches target if its message is contained in the response body. func (e *HTTPError) Is(target error) bool { - se, ok := errors.AsType[*apierr.StatusError](target) - return ok && e.StatusCode == se.Status && strings.Contains(e.Body, se.Message) + msg := target.Error() + return msg != "" && strings.Contains(e.Body, msg) } // sign signs the request with the appropriate headers and returns the signed URL diff --git a/persist/postgres/sectors.go b/persist/postgres/sectors.go index 55057253..a205636d 100644 --- a/persist/postgres/sectors.go +++ b/persist/postgres/sectors.go @@ -378,7 +378,7 @@ func (s *Store) PinSlabs(account proto.Account, nextIntegrityCheck time.Time, to for _, sector := range slab.Sectors { batch.Queue(` INSERT INTO sectors (sector_root, host_id, next_integrity_check, uploaded_at) - SELECT $1, h.id, $3, LEAST(NOW(), COALESCE($4::timestamptz, NOW())) + SELECT $1, h.id, $3, LEAST(NOW(), $4::timestamptz) FROM hosts h WHERE h.public_key = $2 ON CONFLICT (sector_root) DO UPDATE SET diff --git a/persist/postgres/sectors_test.go b/persist/postgres/sectors_test.go index 3afe3c5b..d1020dc6 100644 --- a/persist/postgres/sectors_test.go +++ b/persist/postgres/sectors_test.go @@ -1462,6 +1462,7 @@ func TestPinSlabsUploadedAt(t *testing.T) { store.addTestContract(t, hk) setUploadedAt := func(params slabs.SlabPinParams, ts *time.Time) slabs.SlabPinParams { + params.Sectors = slices.Clone(params.Sectors) for i := range params.Sectors { params.Sectors[i].UploadedAt = ts } diff --git a/slabs/slabs.go b/slabs/slabs.go index 8318ed06..13e79f86 100644 --- a/slabs/slabs.go +++ b/slabs/slabs.go @@ -5,12 +5,10 @@ import ( "errors" "fmt" "math" - "net/http" "time" proto "go.sia.tech/core/rhp/v4" "go.sia.tech/core/types" - "go.sia.tech/indexd/api/apierr" "go.sia.tech/indexd/contracts" ) @@ -64,16 +62,16 @@ var ( // ErrInvalidPinParams is returned when the params of a slab to pin are // invalid. It wraps the underlying reason. - ErrInvalidPinParams = apierr.New(http.StatusBadRequest, "invalid slab pin params") + ErrInvalidPinParams = errors.New("invalid slab pin params") // ErrSlabUploadTooOld is returned when pinning a sector that may already // have been deleted from temporary storage. Clients match these two by // message, so keep the thresholds out of them. - ErrSlabUploadTooOld = apierr.New(http.StatusBadRequest, "slab upload is too old") + ErrSlabUploadTooOld = errors.New("slab upload is too old") // ErrSlabUploadInFuture is returned when a sector's upload time is ahead of // the indexer's clock. - ErrSlabUploadInFuture = apierr.New(http.StatusBadRequest, "slab upload time is in the future") + ErrSlabUploadInFuture = errors.New("slab upload time is in the future") ) type (