diff --git a/config/config.yaml b/config/config.yaml index 4f8c6fdf..ea35588b 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -319,6 +319,10 @@ checks: postage-depth: 21 postage-label: test-label download-timeout: 60s + chunk-retrieval-timeout: # empty keeps Bee's default (30s) + upload-concurrency: 16 + every-leaf: false # withhold the threshold from every leaf group of one large blob + every-leaf-data-size: 4194304 # 4 MiB, about 1,000 chunks: too slow for bee-local timeout: 1h type: erasure-coding retrieval: diff --git a/config/local.yaml b/config/local.yaml index 43e75bd7..cbfbe70a 100644 --- a/config/local.yaml +++ b/config/local.yaml @@ -435,6 +435,10 @@ checks: postage-depth: 21 postage-label: test-label download-timeout: 60s + chunk-retrieval-timeout: # empty keeps Bee's default (30s) + upload-concurrency: 16 + every-leaf: false # withhold the threshold from every leaf group of one large blob + every-leaf-data-size: 4194304 # 4 MiB, about 1,000 chunks: too slow for bee-local timeout: 1h type: erasure-coding ci-gsoc: diff --git a/config/testnet-bee-playground.yaml b/config/testnet-bee-playground.yaml index 8004d7c3..4e741dd2 100644 --- a/config/testnet-bee-playground.yaml +++ b/config/testnet-bee-playground.yaml @@ -180,6 +180,10 @@ checks: postage-depth: 22 postage-label: test-label download-timeout: 60s + chunk-retrieval-timeout: # empty keeps Bee's default (30s) + upload-concurrency: 16 + every-leaf: false # withhold the threshold from every leaf group of one large blob + every-leaf-data-size: 4194304 # 4 MiB, about 1,000 chunks: too slow for bee-local timeout: 1h type: erasure-coding pg-testnet-load: diff --git a/pkg/bee/api/api.go b/pkg/bee/api/api.go index f943fc33..ffe64e0f 100644 --- a/pkg/bee/api/api.go +++ b/pkg/bee/api/api.go @@ -33,6 +33,7 @@ const ( swarmIndexDocumentHeader = "Swarm-Index-Document" swarmErrorDocumentHeader = "Swarm-Error-Document" swarmRedundancyLevelHeader = "Swarm-Redundancy-Level" + swarmChunkRetrievalTimeout = "Swarm-Chunk-Retrieval-Timeout" ) // Client manages communication with the Bee API. @@ -230,6 +231,9 @@ func (c *Client) requestDataGetHeader(ctx context.Context, method, path string, if opts != nil && opts.RedundancyFallbackMode != nil { req.Header.Set(swarmRedundancyFallbackMode, strconv.FormatBool(*opts.RedundancyFallbackMode)) } + if opts != nil && opts.ChunkRetrievalTimeout != nil { + req.Header.Set(swarmChunkRetrievalTimeout, opts.ChunkRetrievalTimeout.String()) + } r, err := c.httpClient.Do(req) if err != nil { return nil, nil, err diff --git a/pkg/bee/api/options.go b/pkg/bee/api/options.go index a88e7bcd..1f5d14ef 100644 --- a/pkg/bee/api/options.go +++ b/pkg/bee/api/options.go @@ -1,6 +1,8 @@ package api import ( + "time" + "github.com/ethersphere/bee/v2/pkg/file/redundancy" "github.com/ethersphere/bee/v2/pkg/swarm" ) @@ -27,5 +29,6 @@ type DownloadOptions struct { Cache *bool RLevel *redundancy.Level RedundancyFallbackMode *bool + ChunkRetrievalTimeout *time.Duration OnlyRootChunk *bool } diff --git a/pkg/check/erasurecoding/erasurecoding.go b/pkg/check/erasurecoding/erasurecoding.go index 7b843aa2..233f7ecb 100644 --- a/pkg/check/erasurecoding/erasurecoding.go +++ b/pkg/check/erasurecoding/erasurecoding.go @@ -5,7 +5,9 @@ import ( "context" "errors" "fmt" + "io" "math/rand" + "net/http" "strings" "time" @@ -20,30 +22,34 @@ import ( "golang.org/x/sync/errgroup" ) -const uploadConcurrency = 16 - // Options for the erasure-coding check. type Options struct { - Levels []uint8 // redundancy levels to test: 1 medium, 2 strong, 3 insane, 4 paranoid - DataSizes []int64 // sizes of the random blobs to split - AllDepths bool // also test the intermediate chunks above the leaf level - PostageDepth uint64 // depth of the postage batch - PostageLabel string // label of the postage batch - PostageTTL time.Duration // lifetime of the postage batch - DownloadTimeout time.Duration // time allowed for each download - Seed int64 + Levels []uint8 // redundancy levels to test: 1 medium, 2 strong, 3 insane, 4 paranoid + DataSizes []int64 // sizes of the random blobs to split + AllDepths bool // also test the intermediate chunks above the leaf level + EveryLeaf bool // also withhold the threshold from every leaf group at once + EveryLeafDataSize int64 // size of the random blob for the every-leaf scenario + PostageDepth uint64 // depth of the postage batch + PostageLabel string // label of the postage batch + PostageTTL time.Duration // lifetime of the postage batch + UploadConcurrency int // chunks uploaded in parallel + DownloadTimeout time.Duration // time allowed for each download + ChunkRetrievalTimeout time.Duration // per-chunk retrieval timeout sent to Bee; zero keeps Bee's default + Seed int64 } // NewDefaultOptions returns the default options. func NewDefaultOptions() Options { return Options{ - Levels: []uint8{uint8(redundancy.MEDIUM), uint8(redundancy.STRONG), uint8(redundancy.INSANE), uint8(redundancy.PARANOID)}, - DataSizes: []int64{32 * 1024}, // 32 KiB: one group at every redundancy level - PostageDepth: 22, - PostageLabel: "erasure-coding-check", - PostageTTL: 24 * time.Hour, - DownloadTimeout: 60 * time.Second, - Seed: time.Now().UnixNano(), + Levels: []uint8{uint8(redundancy.MEDIUM), uint8(redundancy.STRONG), uint8(redundancy.INSANE), uint8(redundancy.PARANOID)}, + DataSizes: []int64{32 * 1024}, // 32 KiB: one group at every redundancy level + EveryLeafDataSize: 4 << 20, // 4 MiB: many leaf groups at every redundancy level + PostageDepth: 22, + PostageLabel: "erasure-coding-check", + PostageTTL: 24 * time.Hour, + UploadConcurrency: 16, + DownloadTimeout: 60 * time.Second, + Seed: time.Now().UnixNano(), } } @@ -62,15 +68,21 @@ func NewCheck(logger logging.Logger) beekeeper.Action { // scenario is one upload and download. type scenario struct { - level redundancy.Level - size int64 - group int - c Case - fallback bool + level redundancy.Level + size int64 + group int + c Case + fallback bool + everyLeaf bool // withhold from every leaf group at once instead of from group } func (s scenario) String() string { - return fmt.Sprintf("level %d, %d bytes, group %d, %s, fallback %v", s.level, s.size, s.group, s.c, s.fallback) + target := fmt.Sprintf("group %d", s.group) + if s.everyLeaf { + target = "every leaf group" + } + + return fmt.Sprintf("level %d, %d bytes, %s, %s, fallback %v", s.level, s.size, target, s.c, s.fallback) } // Run implements beekeeper.Action. @@ -101,6 +113,25 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, o any) e var failures []string total := 0 + // runScenario records a failed scenario and carries on; it stops only when the check is cancelled. + runScenario := func(s scenario) error { + total++ + + if err := c.run(ctx, s, uploader, downloader, batchID, rnd, opts); err != nil { + if ctx.Err() != nil { + return fmt.Errorf("%s: %w", s, ctx.Err()) + } + c.logger.Errorf("FAIL %s: %v", s, err) + failures = append(failures, fmt.Sprintf("%s: %v", s, err)) + + return nil + } + + c.logger.Infof("ok %s", s) + + return nil + } + for _, lvl := range opts.Levels { level := redundancy.Level(lvl) @@ -120,24 +151,21 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, o any) e // Fresh data for every upload: chunks left in the network or in the // download node's cache by one scenario must not satisfy the next. for _, fallback := range []bool{false, true} { - s := scenario{level: level, size: size, group: group, c: cs, fallback: fallback} - total++ - - if err := c.run(ctx, s, uploader, downloader, batchID, rnd, opts); err != nil { - if ctx.Err() != nil { - return fmt.Errorf("%s: %w", s, ctx.Err()) - } - c.logger.Errorf("FAIL %s: %v", s, err) - failures = append(failures, fmt.Sprintf("%s: %v", s, err)) - - continue + if err := runScenario(scenario{level: level, size: size, group: group, c: cs, fallback: fallback}); err != nil { + return err } - - c.logger.Infof("ok %s", s) } } } } + + if opts.EveryLeaf { + for _, fallback := range []bool{false, true} { + if err := runScenario(scenario{level: level, size: opts.EveryLeafDataSize, c: AtThreshold, fallback: fallback, everyLeaf: true}); err != nil { + return err + } + } + } } if len(failures) > 0 { @@ -158,21 +186,30 @@ func (c *Check) run(ctx context.Context, s scenario, uploader, downloader *bee.C return err } - groups := tree.SelectGroups(opts.AllDepths) - if s.group >= len(groups) { - return fmt.Errorf("tree has %d groups under test, want group %d", len(groups), s.group) + var withheld []swarm.Address + if s.everyLeaf { + withheld = tree.LeafDrops(s.c, rnd) + } else { + groups := tree.SelectGroups(opts.AllDepths) + if s.group >= len(groups) { + return fmt.Errorf("tree has %d groups under test, want group %d", len(groups), s.group) + } + withheld = groups[s.group].Drops(s.c, rnd) } - withheld := groups[s.group].Drops(s.c, rnd) - - if err := upload(ctx, uploader, batchID, tree.Chunks, withheld); err != nil { + if err := upload(ctx, uploader, batchID, tree.Chunks, withheld, opts.UploadConcurrency); err != nil { return err } dctx, cancel := context.WithTimeout(ctx, opts.DownloadTimeout) defer cancel() - got, dlErr := downloader.DownloadBytes(dctx, tree.Root, &api.DownloadOptions{RedundancyFallbackMode: &s.fallback}) + dlOpts := &api.DownloadOptions{RedundancyFallbackMode: &s.fallback} + if opts.ChunkRetrievalTimeout > 0 { + dlOpts.ChunkRetrievalTimeout = &opts.ChunkRetrievalTimeout + } + + got, dlErr := downloader.DownloadBytes(dctx, tree.Root, dlOpts) downloaded := dlErr == nil && bytes.Equal(got, data) c.logger.Debugf("%s: withheld %d of %d chunks, download ok %v, err %v", s, len(withheld), len(tree.Chunks), downloaded, dlErr) @@ -184,20 +221,35 @@ func (c *Check) run(ctx context.Context, s scenario, uploader, downloader *bee.C return fmt.Errorf("withheld %d of %d chunks, download returned different data", len(withheld), len(tree.Chunks)) case !want && downloaded: return fmt.Errorf("withheld %d of %d chunks, expected the download to fail but it returned the data", len(withheld), len(tree.Chunks)) + case !want && dlErr == nil: + return fmt.Errorf("withheld %d of %d chunks, download returned %d of %d bytes without an error", len(withheld), len(tree.Chunks), len(got), len(data)) + case !want && !isRecoveryFailure(dlErr): + return fmt.Errorf("withheld %d of %d chunks, download failed for an unexpected reason: %w", len(withheld), len(tree.Chunks), dlErr) } return nil } +// isRecoveryFailure reports whether a download error is Bee failing to rebuild the data, +// as opposed to a timeout or a broken connection that would hide a real problem. +func isRecoveryFailure(err error) bool { + var httpErr *api.HTTPStatusError + if errors.As(err, &httpErr) { + return httpErr.Code == http.StatusNotFound || httpErr.Code == http.StatusInternalServerError + } + + return errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF) +} + // upload sends every chunk except the withheld ones. -func upload(ctx context.Context, client *bee.Client, batchID string, chunks []swarm.Chunk, withheld []swarm.Address) error { +func upload(ctx context.Context, client *bee.Client, batchID string, chunks []swarm.Chunk, withheld []swarm.Address, concurrency int) error { skip := make(map[string]bool, len(withheld)) for _, a := range withheld { skip[a.ByteString()] = true } group, gctx := errgroup.WithContext(ctx) - group.SetLimit(uploadConcurrency) + group.SetLimit(concurrency) for _, ch := range chunks { if skip[ch.Address().ByteString()] { @@ -240,9 +292,18 @@ func (o Options) validate() error { return fmt.Errorf("data size %d must be larger than one chunk (%d bytes) to have an intermediate chunk", s, swarm.ChunkSize) } } + if o.EveryLeaf && o.EveryLeafDataSize <= swarm.ChunkSize { + return fmt.Errorf("every-leaf data size %d must be larger than one chunk (%d bytes)", o.EveryLeafDataSize, swarm.ChunkSize) + } + if o.UploadConcurrency < 1 { + return fmt.Errorf("upload concurrency must be positive, got %d", o.UploadConcurrency) + } if o.DownloadTimeout <= 0 { return errors.New("download timeout must be positive") } + if o.ChunkRetrievalTimeout < 0 { + return errors.New("chunk retrieval timeout must not be negative") + } return nil } diff --git a/pkg/check/erasurecoding/erasurecoding_test.go b/pkg/check/erasurecoding/erasurecoding_test.go index 96963a43..f0c34eae 100644 --- a/pkg/check/erasurecoding/erasurecoding_test.go +++ b/pkg/check/erasurecoding/erasurecoding_test.go @@ -3,8 +3,11 @@ package erasurecoding_test import ( "bytes" "context" + "errors" + "fmt" "io" "math/rand" + "net/http" "testing" "time" @@ -13,6 +16,7 @@ import ( "github.com/ethersphere/bee/v2/pkg/file/redundancy/getter" "github.com/ethersphere/bee/v2/pkg/storage" "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/ethersphere/beekeeper/pkg/bee/api" "github.com/ethersphere/beekeeper/pkg/check/erasurecoding" ) @@ -215,3 +219,115 @@ func TestGroupDrops_matchBeeRecovery(t *testing.T) { } } } + +// TestSplit_treeShape pins the group sizes of a 300 KiB blob, so a change in Bee's +// splitter that moves the thresholds shows up here before it shows up on a cluster. +func TestSplit_treeShape(t *testing.T) { + t.Parallel() + + type shape struct{ data, parity int } + + tests := []struct { + level redundancy.Level + root shape + leaves []shape + }{ + {level: redundancy.MEDIUM, root: shape{75, 8}, leaves: []shape{{75, 8}}}, + {level: redundancy.STRONG, root: shape{75, 17}, leaves: []shape{{75, 17}}}, + {level: redundancy.INSANE, root: shape{75, 27}, leaves: []shape{{75, 27}}}, + {level: redundancy.PARANOID, root: shape{2, 23}, leaves: []shape{{39, 89}, {36, 87}}}, + } + + for _, tc := range tests { + t.Run(fmt.Sprintf("level_%d", tc.level), func(t *testing.T) { + t.Parallel() + + tree, err := erasurecoding.Split(context.Background(), randomData(t, 300*1024), tc.level) + if err != nil { + t.Fatal(err) + } + + root := tree.Groups[0] + if got := (shape{len(root.Data), len(root.Parity)}); got != tc.root { + t.Errorf("root: got %d data + %d parities, want %d + %d", got.data, got.parity, tc.root.data, tc.root.parity) + } + + var leaves []shape + for _, g := range tree.Groups { + if g.LeafParent { + leaves = append(leaves, shape{len(g.Data), len(g.Parity)}) + } + } + if fmt.Sprint(leaves) != fmt.Sprint(tc.leaves) { + t.Errorf("leaf groups (data, parities): got %v, want %v", leaves, tc.leaves) + } + }) + } +} + +// TestLeafDrops_matchBeeRecovery runs the every-leaf scenario through Bee's joiner: +// the threshold withheld from every leaf group at once is rebuilt only with the fallback on. +func TestLeafDrops_matchBeeRecovery(t *testing.T) { + t.Parallel() + + for _, level := range levels { + t.Run(fmt.Sprintf("level_%d", level), func(t *testing.T) { + t.Parallel() + + data := randomData(t, 1024*1024) + + tree, err := erasurecoding.Split(context.Background(), data, level) + if err != nil { + t.Fatal(err) + } + + drops := tree.LeafDrops(erasurecoding.AtThreshold, rand.New(rand.NewSource(7))) + + want := 0 + for _, g := range tree.Groups { + if g.LeafParent { + want += len(g.Parity) + } + } + if len(drops) != want { + t.Fatalf("%d drops, want %d (the parity count of every leaf group)", len(drops), want) + } + + for _, fallback := range []bool{true, false} { + got, err := join(t, tree, drops, fallback) + gotOK := err == nil && bytes.Equal(got, data) + + if want := erasurecoding.AtThreshold.ExpectDownload(fallback); gotOK != want { + t.Errorf("fallback=%v: download ok = %v (err %v), want %v", fallback, gotOK, err, want) + } + } + }) + } +} + +func TestIsRecoveryFailure(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + err error + want bool + }{ + {name: "not found", err: api.NewHTTPStatusError(http.StatusNotFound), want: true}, + {name: "internal server error", err: api.NewHTTPStatusError(http.StatusInternalServerError), want: true}, + {name: "truncated body", err: fmt.Errorf("read: %w", io.ErrUnexpectedEOF), want: true}, + {name: "bad request", err: api.NewHTTPStatusError(http.StatusBadRequest), want: false}, + {name: "timeout", err: fmt.Errorf("download: %w", context.DeadlineExceeded), want: false}, + {name: "other", err: errors.New("connection refused"), want: false}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + if got := erasurecoding.IsRecoveryFailure(tc.err); got != tc.want { + t.Errorf("got %v, want %v", got, tc.want) + } + }) + } +} diff --git a/pkg/check/erasurecoding/export_test.go b/pkg/check/erasurecoding/export_test.go new file mode 100644 index 00000000..c4207e26 --- /dev/null +++ b/pkg/check/erasurecoding/export_test.go @@ -0,0 +1,3 @@ +package erasurecoding + +var IsRecoveryFailure = isRecoveryFailure diff --git a/pkg/check/erasurecoding/tree.go b/pkg/check/erasurecoding/tree.go index bbc2a6b6..7fb244ae 100644 --- a/pkg/check/erasurecoding/tree.go +++ b/pkg/check/erasurecoding/tree.go @@ -98,6 +98,19 @@ func (g Group) withhold(n int, rnd *rand.Rand) []swarm.Address { return picked } +// LeafDrops picks the chunk addresses to withhold for the case from every group whose +// children are leaf data chunks, so one download has to rebuild many groups at once. +func (t *Tree) LeafDrops(c Case, rnd *rand.Rand) []swarm.Address { + var drops []swarm.Address + for _, g := range t.Groups { + if g.LeafParent { + drops = append(drops, g.Drops(c, rnd)...) + } + } + + return drops +} + // Tree is the chunk tree of one split blob. type Tree struct { Root swarm.Address diff --git a/pkg/config/check.go b/pkg/config/check.go index ef45e6e3..2dd1da29 100644 --- a/pkg/config/check.go +++ b/pkg/config/check.go @@ -660,14 +660,18 @@ var Checks = map[string]CheckType{ NewAction: erasurecoding.NewCheck, NewOptions: func(checkGlobalConfig CheckGlobalConfig, check Check) (any, error) { checkOpts := new(struct { - Levels *[]uint8 `yaml:"levels"` - DataSizes *[]int64 `yaml:"data-sizes"` - AllDepths *bool `yaml:"all-depths"` - PostageDepth *uint64 `yaml:"postage-depth"` - PostageLabel *string `yaml:"postage-label"` - PostageTTL *time.Duration `yaml:"postage-ttl"` - DownloadTimeout *time.Duration `yaml:"download-timeout"` - Seed *int64 `yaml:"seed"` + Levels *[]uint8 `yaml:"levels"` + DataSizes *[]int64 `yaml:"data-sizes"` + AllDepths *bool `yaml:"all-depths"` + EveryLeaf *bool `yaml:"every-leaf"` + EveryLeafDataSize *int64 `yaml:"every-leaf-data-size"` + PostageDepth *uint64 `yaml:"postage-depth"` + PostageLabel *string `yaml:"postage-label"` + PostageTTL *time.Duration `yaml:"postage-ttl"` + UploadConcurrency *int `yaml:"upload-concurrency"` + DownloadTimeout *time.Duration `yaml:"download-timeout"` + ChunkRetrievalTimeout *time.Duration `yaml:"chunk-retrieval-timeout"` + Seed *int64 `yaml:"seed"` }) if err := check.Options.Decode(checkOpts); err != nil { return nil, fmt.Errorf("decoding check %s options: %w", check.Type, err)