Skip to content

Commit 10dc967

Browse files
strahestrahe
committed
fix(storage): validate replacements and restore coverage gates
1 parent ef458fc commit 10dc967

7 files changed

Lines changed: 277 additions & 21 deletions

File tree

‎Makefile‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,10 @@ build:
1616
test:
1717
go test ./...
1818

19-
# Run benchmarks — auto-discovers packages containing *_bench_test.go files
19+
# Run benchmarks — auto-discovers packages containing *_bench_test.go files.
20+
# Skip hidden dirs (worktrees, .git) and local reference checkouts.
2021
bench:
21-
go test -bench=. -benchmem $(shell find . -name '*_bench_test.go' | sed 's|/[^/]*$$||' | sort -u)
22+
go test -bench=. -benchmem $(shell find . \( -path '*/.*' -o -path './lotus' -o -path './curio' -o -path './synapse-sdk' -o -path './go-synapse' \) -prune -o -name '*_bench_test.go' -print | sed 's|/[^/]*$$||' | sort -u)
2223

2324
# Run tests with race detector
2425
test-race:

‎storage/context_test.go‎

Lines changed: 83 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"errors"
77
"fmt"
88
"io"
9+
"log/slog"
910
"math/big"
1011
"strings"
1112
"sync"
@@ -294,6 +295,28 @@ func TestDataSetContextConstructionAndDefensiveCopies(t *testing.T) {
294295
if _, ok := providerCtx.DataSetRef(); ok {
295296
t.Fatal("ForDataSet mutated its ProviderContext receiver")
296297
}
298+
299+
logged := mustProviderContext(t, &fakePDPProviderClient{}, WithLogger(slog.New(slog.DiscardHandler)), WithCDN(true))
300+
dataSetLogged, err := logged.ForDataSet(testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)))
301+
if err != nil {
302+
t.Fatalf("ForDataSet: %v", err)
303+
}
304+
if logged.ServiceURL() != testProvider().ServiceURL || dataSetLogged.ServiceURL() != testProvider().ServiceURL {
305+
t.Fatalf("ServiceURL provider=%q data-set=%q", logged.ServiceURL(), dataSetLogged.ServiceURL())
306+
}
307+
if !logged.CDNEnabled() || !dataSetLogged.CDNEnabled() || !logged.WithCDN() || !dataSetLogged.WithCDN() {
308+
t.Fatal("CDNEnabled/WithCDN mismatch")
309+
}
310+
if dataSetLogged.GetProviderInfo().ServiceURL != testProvider().ServiceURL {
311+
t.Fatalf("GetProviderInfo=%+v", dataSetLogged.GetProviderInfo())
312+
}
313+
if !dataSetLogged.ClientDataSetID().Equal(types.NewBigInt(7)) || !dataSetLogged.DataSetID().Equal(types.NewBigInt(42)) {
314+
t.Fatalf("DataSetContext IDs dataSet=%s client=%s", dataSetLogged.DataSetID(), dataSetLogged.ClientDataSetID())
315+
}
316+
pieceCID := mustPieceInfo(t).CIDv2
317+
if logged.PieceURL(pieceCID) == "" || dataSetLogged.PieceURL(pieceCID) == "" {
318+
t.Fatal("PieceURL empty")
319+
}
297320
}
298321

299322
func TestProviderContextForDataSetConcurrent(t *testing.T) {
@@ -881,13 +904,25 @@ func TestContextStoreUploadsAndWaits(t *testing.T) {
881904
return nil
882905
},
883906
}
884-
c := mustProviderContext(t, client)
885-
result, err := c.Store(context.Background(), bytes.NewReader(data), nil)
886-
if err != nil {
887-
t.Fatalf("Store: %v", err)
888-
}
889-
if result.PieceCID != info.CIDv2 || result.Size != int64(len(data)) {
890-
t.Fatalf("Store result=%+v", result)
907+
for _, name := range []string{"provider", "data-set"} {
908+
t.Run(name, func(t *testing.T) {
909+
var (
910+
result *StoreResult
911+
err error
912+
)
913+
switch name {
914+
case "provider":
915+
result, err = mustProviderContext(t, client).Store(context.Background(), bytes.NewReader(data), nil)
916+
default:
917+
result, err = mustDataSetContext(t, client, testDataSetRef(types.NewBigInt(42), types.NewBigInt(7))).Store(context.Background(), bytes.NewReader(data), nil)
918+
}
919+
if err != nil {
920+
t.Fatalf("Store: %v", err)
921+
}
922+
if result.PieceCID != info.CIDv2 || result.Size != int64(len(data)) {
923+
t.Fatalf("Store result=%+v", result)
924+
}
925+
})
891926
}
892927
}
893928

@@ -915,6 +950,47 @@ func TestContextValidationErrors(t *testing.T) {
915950
}
916951
}
917952

953+
func TestContextPresignAndPullRejectInvalidInputs(t *testing.T) {
954+
info := mustPieceInfo(t)
955+
c := mustWritableProviderContext(t, &fakePDPProviderClient{})
956+
957+
tooMany := make([]PieceInput, pdp.MaxAddPiecesBatchSize+1)
958+
for i := range tooMany {
959+
tooMany[i] = PieceInput{PieceCID: info.CIDv2}
960+
}
961+
_, err := c.PresignForCommit(context.Background(), tooMany)
962+
if !errors.Is(err, ErrInvalidArgument) || !errors.Is(err, pdp.ErrTooManyPieces) {
963+
t.Fatalf("too many pieces error=%v", err)
964+
}
965+
966+
_, err = c.PresignForCommit(context.Background(), []PieceInput{{
967+
PieceCID: info.CIDv2,
968+
PieceMetadata: map[string]string{strings.Repeat("k", maxMetadataKeyLength+1): "v"},
969+
}})
970+
if err == nil || !strings.Contains(err.Error(), "exceeds max length") {
971+
t.Fatalf("long metadata key error=%v", err)
972+
}
973+
974+
_, err = c.PresignForCommit(context.Background(), []PieceInput{{}})
975+
if !errors.Is(err, ErrInvalidArgument) {
976+
t.Fatalf("undefined pieceCID error=%v", err)
977+
}
978+
979+
unsigned := mustProviderContext(t, &fakePDPProviderClient{})
980+
_, err = unsigned.PresignForCommit(context.Background(), []PieceInput{{PieceCID: info.CIDv2}})
981+
if !errors.Is(err, ErrInvalidArgument) {
982+
t.Fatalf("missing payer/chain/recordKeeper error=%v", err)
983+
}
984+
985+
_, err = c.Pull(context.Background(), PullRequest{
986+
Pieces: []cid.Cid{info.CIDv2},
987+
From: func(cid.Cid) string { return "" },
988+
})
989+
if !errors.Is(err, ErrInvalidArgument) {
990+
t.Fatalf("empty source URL error=%v", err)
991+
}
992+
}
993+
918994
func TestDataSetContextCommitRejectsMismatchedConfirmation(t *testing.T) {
919995
info := mustPieceInfo(t)
920996
client := &fakePDPProviderClient{

‎storage/create.go‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -400,3 +400,17 @@ func validateExcludedStorageContexts(op string, contexts []StorageContext, exclu
400400
}
401401
return nil
402402
}
403+
404+
func (s *Service) validateUploadReplacement(op string, replacement StorageContext, usedProviders map[string]types.BigInt, excluded []types.BigInt) error {
405+
if err := s.validateStorageContexts(op, []StorageContext{replacement}); err != nil {
406+
return err
407+
}
408+
if err := validateExcludedStorageContexts(op, []StorageContext{replacement}, excluded); err != nil {
409+
return err
410+
}
411+
id := replacement.ProviderID()
412+
if _, exists := usedProviders[idconv.Key(id)]; exists {
413+
return fmt.Errorf("%s: %w: duplicate providerID %s", op, ErrInvalidArgument, id.String())
414+
}
415+
return nil
416+
}

‎storage/download_test.go‎

Lines changed: 21 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"github.com/ipfs/go-cid"
1616

1717
"github.com/strahe/synapse-go/piece"
18+
"github.com/strahe/synapse-go/types"
1819
)
1920

2021
func TestContextDownload_UsesPDPProviderClientAndValidatesPiece(t *testing.T) {
@@ -32,23 +33,31 @@ func TestContextDownload_UsesPDPProviderClientAndValidatesPiece(t *testing.T) {
3233
return io.NopCloser(bytes.NewReader(data)), int64(len(data)), nil
3334
},
3435
}
35-
ctx, err := NewProviderContext(testProvider(), fake, mustTestSigner(t))
36+
providerCtx, err := NewProviderContext(testProvider(), fake, mustTestSigner(t))
3637
if err != nil {
37-
t.Fatalf("NewContext: %v", err)
38+
t.Fatalf("NewProviderContext: %v", err)
3839
}
39-
40-
reader, err := ctx.Download(context.Background(), info.CIDv2)
40+
dataSetCtx, err := NewDataSetContext(testProvider(), fake, mustTestSigner(t), testDataSetRef(types.NewBigInt(42), types.NewBigInt(7)))
4141
if err != nil {
42-
t.Fatalf("Download: %v", err)
42+
t.Fatalf("NewDataSetContext: %v", err)
4343
}
44-
defer func() { _ = reader.Close() }()
4544

46-
got, err := io.ReadAll(reader)
47-
if err != nil {
48-
t.Fatalf("ReadAll: %v", err)
49-
}
50-
if !bytes.Equal(got, data) {
51-
t.Fatal("downloaded bytes mismatch")
45+
for _, storageCtx := range []StorageContext{providerCtx, dataSetCtx} {
46+
reader, err := storageCtx.Download(context.Background(), info.CIDv2)
47+
if err != nil {
48+
t.Fatalf("%T.Download: %v", storageCtx, err)
49+
}
50+
got, err := io.ReadAll(reader)
51+
closeErr := reader.Close()
52+
if err != nil {
53+
t.Fatalf("ReadAll: %v", err)
54+
}
55+
if closeErr != nil {
56+
t.Fatalf("Close: %v", closeErr)
57+
}
58+
if !bytes.Equal(got, data) {
59+
t.Fatalf("%T downloaded bytes mismatch", storageCtx)
60+
}
5261
}
5362
}
5463

‎storage/selector_test.go‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,39 @@ func TestServiceResolverResolveUploadContexts_AutoSelectsApprovedProvidersAndReu
110110
}
111111
}
112112

113+
func TestServiceResolverSelectProviderContext_ReturnsUnboundHealthyProvider(t *testing.T) {
114+
resolver := newTestServiceResolver(t, serviceResolverFixture{
115+
approvedProviderIDs: []types.BigInt{testID(1), testID(2)},
116+
activeProviders: []spregistry.PDPProvider{
117+
testPDPProvider(testID(1), "https://sp-1.example.com"),
118+
testPDPProvider(testID(2), "https://sp-2.example.com"),
119+
},
120+
clientDataSets: []*warmstorage.DataSetInfo{
121+
{DataSetID: testID(11), ProviderID: testID(1), ClientDataSetID: testID(101)},
122+
},
123+
providerPing: func(_ context.Context, serviceURL string) error {
124+
if serviceURL == "https://sp-1.example.com" {
125+
return errors.New("provider unavailable")
126+
}
127+
return nil
128+
},
129+
})
130+
131+
got, err := resolver.SelectProviderContext(context.Background(), SelectProviderContextOptions{
132+
ExcludeProviderIDs: []types.BigInt{testID(1)},
133+
DataSetMetadata: map[string]string{"app": "photos"},
134+
})
135+
if err != nil {
136+
t.Fatalf("SelectProviderContext: %v", err)
137+
}
138+
if got == nil || !got.ProviderID().Equal(testID(2)) {
139+
t.Fatalf("provider=%v want 2", got)
140+
}
141+
if _, bound := got.DataSetRef(); bound {
142+
t.Fatal("SelectProviderContext returned a bound context")
143+
}
144+
}
145+
113146
func TestServiceResolverSelectReplacement_ExcludesUsedProviders(t *testing.T) {
114147
resolver := newTestServiceResolver(t, serviceResolverFixture{
115148
approvedProviderIDs: []types.BigInt{testID(1), testID(2), testID(3), testID(4)},

‎storage/service.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -425,6 +425,10 @@ func (s *Service) uploadWithContexts(ctx context.Context, r io.Reader, contexts
425425
id := c.ProviderID()
426426
usedProviders[idconv.Key(id)] = id
427427
}
428+
var excluded []types.BigInt
429+
if opts != nil {
430+
excluded = opts.ExcludeProviderIDs
431+
}
428432

429433
type successfulSecondary struct {
430434
ctx StorageContext
@@ -506,6 +510,21 @@ func (s *Service) uploadWithContexts(ctx context.Context, r io.Reader, contexts
506510
if replErr != nil {
507511
break
508512
}
513+
if err := s.validateUploadReplacement("storage.Service.Upload", replacement, usedProviders, excluded); err != nil {
514+
var providerID types.BigInt
515+
if !isNilStorageContext(replacement) {
516+
providerID = replacement.ProviderID()
517+
}
518+
failedAttempts = append(failedAttempts, FailedAttempt{
519+
ProviderID: providerID,
520+
Role: CopyRoleSecondary,
521+
Stage: CopyStagePresign,
522+
Err: err,
523+
Explicit: explicitProviders,
524+
})
525+
attemptsUsed++
526+
continue
527+
}
509528
id := replacement.ProviderID()
510529
usedProviders[idconv.Key(id)] = id
511530
attemptsUsed++

‎storage/service_test.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -629,6 +629,110 @@ func TestManagerUpload_ImplicitSecondaryReplacement(t *testing.T) {
629629
}
630630
}
631631

632+
func TestServiceUploadReplacementRejectsInvalidTargets(t *testing.T) {
633+
data := bytes.Repeat([]byte("rq"), 128)
634+
info, err := piece.CalculateFromBytes(data)
635+
if err != nil {
636+
t.Fatalf("CalculateFromBytes: %v", err)
637+
}
638+
639+
newPrimary := func() *fakeUploadContext {
640+
return &fakeUploadContext{
641+
id: types.NewBigInt(101),
642+
endpoint: "https://primary.example.com",
643+
pieceURL: "https://primary.example.com/piece/" + info.CIDv2.String(),
644+
storeFn: func(_ context.Context, _ io.Reader, _ *StoreOptions) (*StoreResult, error) {
645+
return &StoreResult{PieceCID: info.CIDv2, Size: int64(len(data))}, nil
646+
},
647+
commitFn: func(_ context.Context, _ CommitRequest) (*CommitResult, error) {
648+
return &CommitResult{DataSetID: types.NewBigInt(1001), PieceIDs: []types.BigInt{types.NewBigInt(2001)}, IsNewDataSet: true}, nil
649+
},
650+
}
651+
}
652+
newFailedSecondary := func() *fakeUploadContext {
653+
return &fakeUploadContext{
654+
id: types.NewBigInt(202),
655+
endpoint: "https://secondary.example.com",
656+
presignFn: func(_ context.Context, _ []PieceInput) ([]byte, error) {
657+
return []byte{0x01}, nil
658+
},
659+
pullFn: func(_ context.Context, _ PullRequest) (*PullResult, error) {
660+
return nil, errors.New("pull failed")
661+
},
662+
}
663+
}
664+
665+
wrongIdentity := serviceTestIdentity()
666+
wrongIdentity.Payer = common.HexToAddress("0x9999")
667+
tests := map[string]struct {
668+
replacement StorageContext
669+
exclude []types.BigInt
670+
}{
671+
"nil-core": {replacement: &ProviderContext{}},
672+
"typed nil": {replacement: (*ProviderContext)(nil)},
673+
"identity mismatch": {replacement: &fakeUploadContext{
674+
id: types.NewBigInt(303),
675+
endpoint: "https://replacement.example.com",
676+
identity: &wrongIdentity,
677+
presignFn: func(_ context.Context, _ []PieceInput) ([]byte, error) {
678+
t.Fatal("PresignForCommit must not run for an identity-mismatched replacement")
679+
return nil, nil
680+
},
681+
}},
682+
"excluded": {
683+
replacement: &fakeUploadContext{
684+
id: types.NewBigInt(303),
685+
endpoint: "https://replacement.example.com",
686+
presignFn: func(_ context.Context, _ []PieceInput) ([]byte, error) {
687+
t.Fatal("PresignForCommit must not run for an excluded replacement")
688+
return nil, nil
689+
},
690+
},
691+
exclude: []types.BigInt{types.NewBigInt(303)},
692+
},
693+
"duplicate primary": {replacement: &fakeUploadContext{
694+
id: types.NewBigInt(101),
695+
endpoint: "https://replacement.example.com",
696+
presignFn: func(_ context.Context, _ []PieceInput) ([]byte, error) {
697+
t.Fatal("PresignForCommit must not run for a duplicate replacement")
698+
return nil, nil
699+
},
700+
}},
701+
}
702+
703+
for name, tt := range tests {
704+
t.Run(name, func(t *testing.T) {
705+
defer func() {
706+
if recovered := recover(); recovered != nil {
707+
t.Fatalf("Upload panicked: %v", recovered)
708+
}
709+
}()
710+
svc := mustNewService(t, Options{
711+
Resolver: &fakeResolver{
712+
contexts: []StorageContext{newPrimary(), newFailedSecondary()},
713+
replacements: []StorageContext{tt.replacement},
714+
},
715+
})
716+
got, err := svc.Upload(context.Background(), bytes.NewReader(data), &UploadOptions{
717+
Copies: 2,
718+
ExcludeProviderIDs: tt.exclude,
719+
})
720+
if err != nil {
721+
t.Fatalf("Upload: %v", err)
722+
}
723+
if got.Complete || got.SuccessCount() != 1 {
724+
t.Fatalf("complete=%v success=%d want partial primary", got.Complete, got.SuccessCount())
725+
}
726+
if len(got.FailedAttempts) < 2 {
727+
t.Fatalf("failedAttempts=%+v want secondary failure and rejected replacement", got.FailedAttempts)
728+
}
729+
if !errors.Is(got.FailedAttempts[len(got.FailedAttempts)-1].Err, ErrInvalidArgument) {
730+
t.Fatalf("replacement error=%v want ErrInvalidArgument", got.FailedAttempts[len(got.FailedAttempts)-1].Err)
731+
}
732+
})
733+
}
734+
}
735+
632736
func TestManagerUpload_ReplacementKeepsImmutableClientDataSetID(t *testing.T) {
633737
data := bytes.Repeat([]byte("ij"), 128)
634738
info, err := piece.CalculateFromBytes(data)

0 commit comments

Comments
 (0)