From c8276578254a5cc4a863235b177ba80f37f1a826 Mon Sep 17 00:00:00 2001 From: akrem-chabchoub Date: Thu, 15 Oct 2026 17:42:35 +0200 Subject: [PATCH 1/2] feat(check): add corrupt chunk check --- cmd/beekeeper/cmd/check.go | 5 +- config/local.yaml | 16 +- pkg/bee/api/node.go | 4 +- pkg/bee/client.go | 5 + pkg/check/corruptchunk/corruptchunk.go | 419 +++++++++++++++++++++++++ pkg/config/check.go | 33 +- 6 files changed, 475 insertions(+), 7 deletions(-) create mode 100644 pkg/check/corruptchunk/corruptchunk.go diff --git a/cmd/beekeeper/cmd/check.go b/cmd/beekeeper/cmd/check.go index 8c087b12d..7b5dbb83b 100644 --- a/cmd/beekeeper/cmd/check.go +++ b/cmd/beekeeper/cmd/check.go @@ -104,8 +104,9 @@ Use --metrics-enabled to collect and push metrics to Prometheus.`, // set global config checkGlobalConfig := config.CheckGlobalConfig{ - Seed: c.globalConfig.GetInt64(optionNameSeed), - GethURL: c.globalConfig.GetString(optionNameGethURL), + Seed: c.globalConfig.GetInt64(optionNameSeed), + GethURL: c.globalConfig.GetString(optionNameGethURL), + K8sClient: c.k8sClient, } checkRunner := check.NewCheckRunner(checkGlobalConfig, c.config.Checks, cluster, metricsPusher, tracer, c.log) diff --git a/config/local.yaml b/config/local.yaml index d389ee2b3..98de0b80f 100644 --- a/config/local.yaml +++ b/config/local.yaml @@ -23,7 +23,7 @@ clusters: bee: bee-config: bee-local config: local - count: 2 + count: 8 mode: node local-dns: _inherit: "local" @@ -101,7 +101,7 @@ clusters: node-groups: local: _inherit: "" - image: k3d-registry.localhost:5000/ethersphere/bee:latest + image: akram951/bee:before-5634 image-pull-policy: Always ingress-annotations: nginx.ingress.kubernetes.io/affinity: "cookie" @@ -427,6 +427,18 @@ checks: seed: data-size: type: redundancy + ci-corrupt-chunk: + options: + postage-ttl: 24h + postage-depth: 22 + postage-label: test-label + seed: + import-image: ethersphere/bee:2.8.2 + restart-image: + expect-removed: true + ready-timeout: 5m + timeout: 30m + type: corrupt-chunk ci-gsoc: options: postage-ttl: 24h diff --git a/pkg/bee/api/node.go b/pkg/bee/api/node.go index 789eeccc3..d2ef3ef07 100644 --- a/pkg/bee/api/node.go +++ b/pkg/bee/api/node.go @@ -97,7 +97,9 @@ func (n *NodeService) HasChunk(ctx context.Context, a swarm.Address) (bool, erro // Health represents node's health type Health struct { - Status string `json:"status"` + Status string `json:"status"` + Version string `json:"version"` + APIVersion string `json:"apiVersion"` } // Health returns node's health diff --git a/pkg/bee/client.go b/pkg/bee/client.go index 711ffeb65..a0c2c2f41 100644 --- a/pkg/bee/client.go +++ b/pkg/bee/client.go @@ -211,6 +211,11 @@ func (c *Client) DownloadBytes(ctx context.Context, a swarm.Address, opts *api.D return io.ReadAll(r) } +// Health returns node's health +func (c *Client) Health(ctx context.Context) (api.Health, error) { + return c.api.Node.Health(ctx) +} + // DownloadChunk downloads chunk from the node func (c *Client) DownloadChunk(ctx context.Context, a swarm.Address, targets string, opts *api.DownloadOptions) (data []byte, err error) { r, err := c.api.Chunks.Download(ctx, a, targets, opts) diff --git a/pkg/check/corruptchunk/corruptchunk.go b/pkg/check/corruptchunk/corruptchunk.go new file mode 100644 index 000000000..6651f2568 --- /dev/null +++ b/pkg/check/corruptchunk/corruptchunk.go @@ -0,0 +1,419 @@ +package corruptchunk + +import ( + "archive/tar" + "bytes" + "context" + "encoding/base64" + "encoding/binary" + "encoding/hex" + "errors" + "fmt" + "math/rand" + "net/http" + "strings" + "time" + + "github.com/ethersphere/bee/v2/pkg/cac" + "github.com/ethersphere/bee/v2/pkg/postage" + "github.com/ethersphere/bee/v2/pkg/soc" + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/ethersphere/beekeeper/pkg/bee" + "github.com/ethersphere/beekeeper/pkg/bee/api" + "github.com/ethersphere/beekeeper/pkg/beekeeper" + "github.com/ethersphere/beekeeper/pkg/k8s" + "github.com/ethersphere/beekeeper/pkg/logging" + "github.com/ethersphere/beekeeper/pkg/orchestration" + "github.com/ethersphere/beekeeper/pkg/random" + appsv1 "k8s.io/api/apps/v1" + "k8s.io/client-go/util/retry" +) + +const ( + beeContainerName = "bee" + dataVolumeName = "data" + dataDir = "/home/bee/.bee" +) + +type Options struct { + K8sClient *k8s.Client + ImportImage string + RestartImage string + ExpectRemoved bool + ReadyTimeout time.Duration + PostageDepth uint64 + PostageLabel string + PostageTTL time.Duration + Seed int64 +} + +func NewDefaultOptions() Options { + return Options{ + ImportImage: "ethersphere/bee:2.8.2", + ExpectRemoved: true, + ReadyTimeout: 5 * time.Minute, + PostageDepth: 22, + PostageLabel: "test-label", + PostageTTL: 24 * time.Hour, + Seed: random.Int64(), + } +} + +var _ beekeeper.Action = (*Check)(nil) + +type Check struct { + logger logging.Logger +} + +func NewCheck(logger logging.Logger) beekeeper.Action { + return &Check{ + logger: logger, + } +} + +func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, o any) error { + opts, ok := o.(Options) + if !ok { + return fmt.Errorf("invalid options type") + } + if opts.K8sClient == nil { + return errors.New("kubernetes client is required, run with --enable-k8s") + } + if opts.ImportImage == "" { + return errors.New("import image is required") + } + + c.logger.Infof("seed: %d", opts.Seed) + rnd := random.PseudoGenerator(opts.Seed) + + clients, err := cluster.ShuffledFullNodeClients(ctx, rnd) + if err != nil { + return fmt.Errorf("get shuffled full node clients: %w", err) + } + if len(clients) == 0 { + return errors.New("no full nodes in the cluster") + } + node := clients[0] + namespace := cluster.Namespace() + c.logger.Infof("target node: %s, namespace: %s", node.Name(), namespace) + + original, err := opts.K8sClient.StatefulSet.Get(ctx, node.Name(), namespace) + if err != nil { + return fmt.Errorf("get stateful set %s: %w", node.Name(), err) + } + // if err := checkPersistence(original); err != nil { + // return err + // } + + health, err := c.waitReady(ctx, node, opts.ReadyTimeout) + if err != nil { + return err + } + c.logger.Infof("node %s runs bee %s, image %s", node.Name(), health.Version, beeImage(original)) + c.logStatus(ctx, node, "before") + + batchID, err := node.GetOrCreateMutableBatch(ctx, opts.PostageTTL, opts.PostageDepth, opts.PostageLabel) + if err != nil { + return fmt.Errorf("get or create batch: %w", err) + } + c.logger.Infof("postage batch: %s", batchID) + + ch, archive, err := invalidChunk(rnd, batchID) + if err != nil { + return err + } + c.logger.Infof("invalid chunk: %s (%d bytes, neither cac nor soc)", ch.Address(), len(ch.Data())) + + if _, err := node.DownloadChunk(ctx, ch.Address(), "", nil); err == nil { + return fmt.Errorf("chunk %s is retrievable before import", ch.Address()) + } + + sts := &statefulSet{client: opts.K8sClient, namespace: namespace, original: original, logger: c.logger} + defer sts.restore(context.WithoutCancel(ctx)) + + c.logger.Infof("importing invalid chunk with %s", opts.ImportImage) + if err := sts.runImport(ctx, opts.ImportImage, importCommand(archive)); err != nil { + return fmt.Errorf("import invalid chunk: %w", err) + } + c.logger.Infof("invalid chunk imported into %s reserve", node.Name()) + + restartImage := opts.RestartImage + if restartImage == "" { + restartImage = beeImage(original) + } + c.logger.Infof("restarting %s with %s", node.Name(), restartImage) + if err := sts.start(ctx, restartImage); err != nil { + return fmt.Errorf("restart node: %w", err) + } + + health, err = c.waitReady(ctx, node, opts.ReadyTimeout) + if err != nil { + return err + } + c.logger.Infof("node %s is back, runs bee %s", node.Name(), health.Version) + c.logStatus(ctx, node, "after") + + cache := false + got, err := node.DownloadChunk(ctx, ch.Address(), "", &api.DownloadOptions{Cache: &cache}) + c.logger.Infof("invalid chunk %s lookup on %s: %d bytes, error: %v", ch.Address(), node.Name(), len(got), err) + + var kept bool + switch { + case err == nil && bytes.Equal(got, ch.Data()): + kept = true + case err == nil: + return fmt.Errorf("node %s returned %d unexpected bytes for invalid chunk %s", node.Name(), len(got), ch.Address()) + case !isChunkAbsent(err): + return fmt.Errorf("inconclusive lookup of invalid chunk %s on %s: %w", ch.Address(), node.Name(), err) + } + + switch { + case opts.ExpectRemoved && kept: + return fmt.Errorf("invalid chunk %s is still served by %s (bee %s): reserve kept it", ch.Address(), node.Name(), health.Version) + case !opts.ExpectRemoved && !kept: + return fmt.Errorf("invalid chunk %s is not served by %s (bee %s), expected it to be kept: %w", ch.Address(), node.Name(), health.Version, err) + case kept: + c.logger.Infof("invalid chunk kept and served, as expected") + default: + c.logger.Infof("invalid chunk removed, as expected") + } + + return nil +} + +func (c *Check) logStatus(ctx context.Context, node *bee.Client, when string) { + status, err := node.Status(ctx) + if err != nil { + c.logger.Warningf("node %s status %s: %v", node.Name(), when, err) + return + } + c.logger.Infof("node %s %s: reserve %d (%d within radius), storage radius %d, peers %d", + node.Name(), when, status.ReserveSize, status.ReserveSizeWithinRadius, status.StorageRadius, status.ConnectedPeers) +} + +func (c *Check) waitReady(ctx context.Context, node *bee.Client, timeout time.Duration) (api.Health, error) { + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + ticker := time.NewTicker(2 * time.Second) + defer ticker.Stop() + + for { + health, err := node.Health(ctx) + if err == nil && health.Status == "ok" { + if _, err = node.Status(ctx); err == nil { + return health, nil + } + } + c.logger.Debugf("waiting for node %s api: %v", node.Name(), err) + + select { + case <-ctx.Done(): + return api.Health{}, fmt.Errorf("node %s api not ready after %s: %w", node.Name(), timeout, err) + case <-ticker.C: + } + } +} + +func isChunkAbsent(err error) bool { + var httpErr *api.HTTPStatusError + if !errors.As(err, &httpErr) { + return false + } + switch httpErr.Code { + case http.StatusNotFound: + return true + case http.StatusInternalServerError: + return strings.Contains(err.Error(), "read chunk failed") + } + return false +} + +func checkPersistence(ss *appsv1.StatefulSet) error { + for _, v := range ss.Spec.Template.Spec.Volumes { + if v.Name == dataVolumeName && v.EmptyDir != nil { + return fmt.Errorf("node %s stores data on emptyDir, the imported chunk would be lost on restart: set persistence-enabled: true", ss.Name) + } + } + for _, pvc := range ss.Spec.VolumeClaimTemplates { + if pvc.Name == dataVolumeName { + return nil + } + } + return fmt.Errorf("node %s has no %q volume claim", ss.Name, dataVolumeName) +} + +func beeImage(ss *appsv1.StatefulSet) string { + if i := beeContainerIndex(ss); i >= 0 { + return ss.Spec.Template.Spec.Containers[i].Image + } + return "" +} + +func beeContainerIndex(ss *appsv1.StatefulSet) int { + for i, c := range ss.Spec.Template.Spec.Containers { + if c.Name == beeContainerName { + return i + } + } + return -1 +} + +func invalidChunk(rnd *rand.Rand, batchID string) (swarm.Chunk, []byte, error) { + id, err := hex.DecodeString(batchID) + if err != nil { + return nil, nil, fmt.Errorf("decode batch id %q: %w", batchID, err) + } + + addr := make([]byte, swarm.HashSize) + if _, err := rnd.Read(addr); err != nil { + return nil, nil, err + } + data := make([]byte, swarm.SpanSize+swarm.ChunkSize) + if _, err := rnd.Read(data[swarm.SpanSize:]); err != nil { + return nil, nil, err + } + binary.LittleEndian.PutUint64(data, swarm.ChunkSize) + + ch := swarm.NewChunk(swarm.NewAddress(addr), data) + if cac.Valid(ch) || soc.Valid(ch) { + return nil, nil, fmt.Errorf("generated chunk %s is valid", ch.Address()) + } + + index := make([]byte, 8) + binary.BigEndian.PutUint32(index, binary.BigEndian.Uint32(addr)>>(32-postage.BucketDepth)) + timestamp := make([]byte, 8) + binary.BigEndian.PutUint64(timestamp, uint64(time.Now().UnixNano())) + stamp, err := postage.NewStamp(id, index, timestamp, make([]byte, 65)).MarshalBinary() + if err != nil { + return nil, nil, fmt.Errorf("marshal stamp: %w", err) + } + + body := make([]byte, 4, 4+len(data)+len(stamp)) + binary.BigEndian.PutUint32(body, uint32(len(data))) + body = append(body, data...) + body = append(body, stamp...) + + var buf bytes.Buffer + tw := tar.NewWriter(&buf) + if err := tw.WriteHeader(&tar.Header{Name: ch.Address().String(), Mode: 0o644, Size: int64(len(body))}); err != nil { + return nil, nil, fmt.Errorf("write tar header: %w", err) + } + if _, err := tw.Write(body); err != nil { + return nil, nil, fmt.Errorf("write tar body: %w", err) + } + if err := tw.Close(); err != nil { + return nil, nil, fmt.Errorf("close tar: %w", err) + } + + return ch, buf.Bytes(), nil +} + +func importCommand(archive []byte) []string { + return []string{"sh", "-c", fmt.Sprintf("echo %s | base64 -d | bee db import reserve - --data-dir=%s", + base64.StdEncoding.EncodeToString(archive), dataDir)} +} + +type statefulSet struct { + client *k8s.Client + namespace string + original *appsv1.StatefulSet + logger logging.Logger + changed bool +} + +func (s *statefulSet) podName() string { + return s.original.Name + "-0" +} + +func (s *statefulSet) update(ctx context.Context, mutate func(*appsv1.StatefulSet) error) error { + return retry.RetryOnConflict(retry.DefaultRetry, func() error { + latest, err := s.client.StatefulSet.Get(ctx, s.original.Name, s.namespace) + if err != nil { + return fmt.Errorf("get stateful set %s: %w", s.original.Name, err) + } + if err := mutate(latest); err != nil { + return err + } + return s.client.StatefulSet.Update(ctx, s.namespace, latest) + }) +} + +func (s *statefulSet) runImport(ctx context.Context, image string, command []string) error { + s.changed = true + err := s.update(ctx, func(ss *appsv1.StatefulSet) error { + i := beeContainerIndex(ss) + if i < 0 { + return fmt.Errorf("stateful set %s has no %q container", ss.Name, beeContainerName) + } + ss.Spec.UpdateStrategy.Type = appsv1.OnDeleteStatefulSetStrategyType + ss.Spec.UpdateStrategy.RollingUpdate = nil + ss.Spec.Template.Spec.Containers[i].Image = image + ss.Spec.Template.Spec.Containers[i].Command = command + ss.Spec.Template.Spec.Containers[i].ReadinessProbe = nil + ss.Spec.Template.Spec.Containers[i].LivenessProbe = nil + return nil + }) + if err != nil { + return fmt.Errorf("set import spec: %w", err) + } + + errc := make(chan error, 1) + go func() { + errc <- s.client.Pods.WaitForPodRecreationAndCompletion(ctx, s.namespace, s.podName(), beeContainerName) + }() + s.logger.Debugf("deleting pod %s to run the import", s.podName()) + if _, err := s.client.Pods.Delete(ctx, s.podName(), s.namespace); err != nil { + return fmt.Errorf("delete pod %s: %w", s.podName(), err) + } + if err := <-errc; err != nil { + return fmt.Errorf("import in pod %s: %w", s.podName(), err) + } + return nil +} + +func (s *statefulSet) start(ctx context.Context, image string) error { + err := s.update(ctx, func(ss *appsv1.StatefulSet) error { + ss.Spec.Template = *s.original.Spec.Template.DeepCopy() + if i := beeContainerIndex(ss); i >= 0 { + ss.Spec.Template.Spec.Containers[i].Image = image + } + return nil + }) + if err != nil { + return fmt.Errorf("set start spec: %w", err) + } + s.changed = image != beeImage(s.original) + return s.recreate(ctx) +} + +func (s *statefulSet) recreate(ctx context.Context) error { + s.logger.Debugf("deleting pod %s", s.podName()) + if _, err := s.client.Pods.Delete(ctx, s.podName(), s.namespace); err != nil { + return fmt.Errorf("delete pod %s: %w", s.podName(), err) + } + if err := s.client.Pods.WaitForRunning(ctx, s.namespace, s.podName()); err != nil { + return fmt.Errorf("wait for pod %s: %w", s.podName(), err) + } + return nil +} + +func (s *statefulSet) restore(ctx context.Context) { + s.logger.Infof("restoring stateful set %s", s.original.Name) + err := s.update(ctx, func(ss *appsv1.StatefulSet) error { + ss.Spec.Template = *s.original.Spec.Template.DeepCopy() + ss.Spec.UpdateStrategy = s.original.Spec.UpdateStrategy + ss.Spec.Replicas = s.original.Spec.Replicas + return nil + }) + if err != nil { + s.logger.Errorf("restore stateful set %s: %v", s.original.Name, err) + return + } + if !s.changed { + return + } + if err := s.recreate(ctx); err != nil { + s.logger.Errorf("restore pod %s: %v", s.podName(), err) + } +} diff --git a/pkg/config/check.go b/pkg/config/check.go index 541afb7e5..991e7a8bf 100644 --- a/pkg/config/check.go +++ b/pkg/config/check.go @@ -12,6 +12,7 @@ import ( "github.com/ethersphere/beekeeper/pkg/check/autotls" "github.com/ethersphere/beekeeper/pkg/check/balances" "github.com/ethersphere/beekeeper/pkg/check/cashout" + "github.com/ethersphere/beekeeper/pkg/check/corruptchunk" "github.com/ethersphere/beekeeper/pkg/check/datadurability" "github.com/ethersphere/beekeeper/pkg/check/feed" "github.com/ethersphere/beekeeper/pkg/check/fileretrieval" @@ -36,6 +37,7 @@ import ( "github.com/ethersphere/beekeeper/pkg/check/soc" "github.com/ethersphere/beekeeper/pkg/check/stake" "github.com/ethersphere/beekeeper/pkg/check/withdraw" + "github.com/ethersphere/beekeeper/pkg/k8s" "github.com/ethersphere/beekeeper/pkg/logging" "github.com/ethersphere/beekeeper/pkg/random" "gopkg.in/yaml.v3" @@ -56,8 +58,9 @@ type CheckType struct { // CheckGlobalConfig represents global configs for all checks type CheckGlobalConfig struct { - Seed int64 - GethURL string + Seed int64 + GethURL string + K8sClient *k8s.Client } // Checks represents all available check types @@ -655,6 +658,32 @@ var Checks = map[string]CheckType{ return opts, nil }, }, + "corrupt-chunk": { + NewAction: corruptchunk.NewCheck, + NewOptions: func(checkGlobalConfig CheckGlobalConfig, check Check) (any, error) { + checkOpts := new(struct { + ImportImage *string `yaml:"import-image"` + RestartImage *string `yaml:"restart-image"` + ExpectRemoved *bool `yaml:"expect-removed"` + ReadyTimeout *time.Duration `yaml:"ready-timeout"` + PostageDepth *uint64 `yaml:"postage-depth"` + PostageLabel *string `yaml:"postage-label"` + PostageTTL *time.Duration `yaml:"postage-ttl"` + Seed *int64 `yaml:"seed"` + }) + if err := check.Options.Decode(checkOpts); err != nil { + return nil, fmt.Errorf("decoding check %s options: %w", check.Type, err) + } + opts := corruptchunk.NewDefaultOptions() + + if err := applyCheckConfig(checkGlobalConfig, checkOpts, &opts); err != nil { + return nil, fmt.Errorf("applying options: %w", err) + } + opts.K8sClient = checkGlobalConfig.K8sClient + + return opts, nil + }, + }, "withdraw": { NewAction: withdraw.NewCheck, NewOptions: func(checkGlobalConfig CheckGlobalConfig, check Check) (any, error) { From 4d9cfc39f477d1d213506a273f6ea8ad0d72bfb9 Mon Sep 17 00:00:00 2001 From: akrem-chabchoub Date: Thu, 15 Oct 2026 18:47:33 +0200 Subject: [PATCH 2/2] chore(config): update image and enable persistence for local node group --- config/local.yaml | 8 +++++--- pkg/check/corruptchunk/corruptchunk.go | 17 ++++++++++++++++- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/config/local.yaml b/config/local.yaml index 98de0b80f..10dd763c4 100644 --- a/config/local.yaml +++ b/config/local.yaml @@ -101,7 +101,7 @@ clusters: node-groups: local: _inherit: "" - image: akram951/bee:before-5634 + image: akram951/bee:before-5640 image-pull-policy: Always ingress-annotations: nginx.ingress.kubernetes.io/affinity: "cookie" @@ -122,7 +122,7 @@ node-groups: beekeeper.ethswarm.org/node-funder: "true" node-selector: node-group: "local" - persistence-enabled: false + persistence-enabled: true pod-management-policy: "OrderedReady" resources-limit-cpu: "250m" resources-limit-memory: 0.5Gi @@ -130,6 +130,8 @@ node-groups: resources-request-memory: 0.25Gi restart-policy: "Always" update-strategy: "RollingUpdate" + persistence-storage-class: "local-path" + persistence-storage-request: "1Mi" local-dns: _inherit: "local" local-autotls: @@ -433,7 +435,7 @@ checks: postage-depth: 22 postage-label: test-label seed: - import-image: ethersphere/bee:2.8.2 + import-image: akram951/bee:before-5640 restart-image: expect-removed: true ready-timeout: 5m diff --git a/pkg/check/corruptchunk/corruptchunk.go b/pkg/check/corruptchunk/corruptchunk.go index 6651f2568..838926f2c 100644 --- a/pkg/check/corruptchunk/corruptchunk.go +++ b/pkg/check/corruptchunk/corruptchunk.go @@ -49,7 +49,7 @@ type Options struct { func NewDefaultOptions() Options { return Options{ - ImportImage: "ethersphere/bee:2.8.2", + ImportImage: "akram951/bee:before-5640", ExpectRemoved: true, ReadyTimeout: 5 * time.Minute, PostageDepth: 22, @@ -61,6 +61,21 @@ func NewDefaultOptions() Options { var _ beekeeper.Action = (*Check)(nil) +// Check verifies how a node handles an invalid chunk (neither CAC nor SOC) +// already sitting in its reserve, the case bee PR #5640 cleans up with +// migration step 8 on startup. +// +// Steps: +// 1. Pick a full node; its data must be on a persistent volume. +// 2. Build an invalid chunk: random bytes under a random address, stamped +// with a real batch. +// 3. Swap the node's StatefulSet to ImportImage (bee 7515c4c5, the commit +// before PR #5640, whose reserve does not validate chunk types) running +// `bee db import reserve` to write the chunk. +// 4. Restart the node with RestartImage (the original image if empty). +// 5. GET /chunks/
: served means the chunk was kept, not found means +// it was removed; compare with ExpectRemoved. +// 6. Always restore the original StatefulSet. type Check struct { logger logging.Logger }