From 74285cd835db5ed7dcb03cfec25754c09a501249 Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Fri, 2 Jan 2026 18:44:33 +0200 Subject: [PATCH 1/6] fix: gsoc and pss checks --- pkg/check/gsoc/gsoc.go | 66 +++++++++++++++++++++++++++++++----------- pkg/check/pss/pss.go | 34 ++++++++++++++++++++-- 2 files changed, 81 insertions(+), 19 deletions(-) diff --git a/pkg/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index c19b95ab5..b92ddf7a0 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/pkg/check/gsoc/gsoc.go @@ -29,14 +29,16 @@ type Options struct { PostageTTL time.Duration PostageDepth uint64 PostageLabel string + Chunks int } // NewDefaultOptions returns new default options func NewDefaultOptions() Options { return Options{ PostageTTL: 24 * time.Hour, - PostageDepth: 17, + PostageDepth: 22, PostageLabel: "test-label", + Chunks: 3, } } @@ -95,14 +97,14 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts any } c.logger.Infof("send messages with different postage batches sequentially...") - err = run(ctx, uploadClient, listenClient, batches, c.logger, false) + err = run(ctx, uploadClient, listenClient, batches, c.logger, false, o.Chunks) if err != nil { return fmt.Errorf("sequential: %w", err) } c.logger.Infof("done") c.logger.Infof("send messages with different postage batches parallel...") - err = run(ctx, uploadClient, listenClient, batches, c.logger, true) + err = run(ctx, uploadClient, listenClient, batches, c.logger, true, o.Chunks) if err != nil { return fmt.Errorf("parallel: %w", err) } @@ -111,8 +113,10 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts any return nil } -func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client, batches []string, logger logging.Logger, parallel bool) error { - const numChunks = 10 +func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client, batches []string, logger logging.Logger, parallel bool, numChunks int) error { + if numChunks <= 0 { + numChunks = 3 + } privKey, err := crypto.GenerateSecp256k1Key() if err != nil { return err @@ -171,21 +175,49 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client return err } - select { - case <-time.After(3 * time.Minute): - return fmt.Errorf("timeout: not all messages received") - case <-done: - } + // Wait for all messages to be received or timeout + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + timeout := time.After(3 * time.Minute) - receivedMtx.Lock() - defer receivedMtx.Unlock() - for i := range numChunks { - want := fmt.Sprintf("data %d", i) - if !received[want] { - return fmt.Errorf("message '%s' not received", want) + for { + select { + case <-done: + return nil + case <-timeout: + return fmt.Errorf("timeout: not all messages received") + case <-ticker.C: + receivedMtx.Lock() + if len(received) == numChunks { + receivedMtx.Unlock() + return nil + } + + var missing []int + for i := range numChunks { + want := fmt.Sprintf("data %d", i) + if !received[want] { + missing = append(missing, i) + } + } + receivedMtx.Unlock() + + if len(missing) == 0 { + continue + } + + logger.Infof("gsoc: still missing %d chunks: %v. retrying...", len(missing), missing) + + // Retry missing chunks sequentially to avoid flooding + for _, i := range missing { + payload := fmt.Sprintf("data %d", i) + logger.Infof("gsoc: retrying soc to node=%s, payload=%s", uploadClient.Name(), payload) + if err := uploadSoc(ctx, uploadClient, payload, resourceId, batches[i%2], privKey); err != nil { + logger.Errorf("gsoc: retry failed for %s: %v", payload, err) + } + } } } - return nil } func uploadSoc(ctx context.Context, client *bee.Client, payload string, resourceId []byte, batchID string, privKey *ecdsa.PrivateKey) error { diff --git a/pkg/check/pss/pss.go b/pkg/check/pss/pss.go index 6856ad67f..dd502ee23 100644 --- a/pkg/check/pss/pss.go +++ b/pkg/check/pss/pss.go @@ -33,7 +33,7 @@ func NewDefaultOptions() Options { AddressPrefix: 1, GasPrice: "", PostageTTL: 24 * time.Hour, - PostageDepth: 16, + PostageDepth: 22, PostageLabel: "test-label", RequestTimeout: 5 * time.Minute, Seed: random.Int64(), @@ -131,8 +131,38 @@ func (c *Check) testPss(nodeAName, nodeBName string, clients map[string]*bee.Cli defer closer() tStart := time.Now() - err = nodeA.SendPSSMessage(ctx, addrB.Overlay, addrB.PSSPublicKey, testTopic, o.AddressPrefix, testData, batchID) + c.metrics.SendAndReceiveGauge.WithLabelValues(nodeAName, nodeBName).Set(0) + for range 5 { + err = nodeA.SendPSSMessage(ctx, addrB.Overlay, addrB.PSSPublicKey, testTopic, o.AddressPrefix, testData, batchID) + if err == nil { + break + } + + // check if message is received + select { + case msg := <-ch: + if msg == string(testData) { + c.logger.Info("pss: message received despite send failure") + return nil + } + default: + // continue + } + + c.logger.Infof("pss: send failed, retrying in 1s: %v", err) + time.Sleep(1 * time.Second) + } if err != nil { + // check if message is received + select { + case msg := <-ch: + if msg == string(testData) { + c.logger.Info("pss: message received despite send failure") + return nil + } + default: + // continue + } return err } c.logger.Infof("pss: test data sent successfully to node %s. Waiting for response from node %s", nodeAName, nodeBName) From 9b2cbeb3227d8e9ae2f9e75e255ed51a89319826 Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Tue, 20 Jan 2026 16:19:30 +0200 Subject: [PATCH 2/6] fix: addressing review comments --- pkg/check/gsoc/gsoc.go | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/pkg/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index b92ddf7a0..0d6e21a45 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/pkg/check/gsoc/gsoc.go @@ -115,7 +115,7 @@ func (c *Check) Run(ctx context.Context, cluster orchestration.Cluster, opts any func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client, batches []string, logger logging.Logger, parallel bool, numChunks int) error { if numChunks <= 0 { - numChunks = 3 + return fmt.Errorf("chunks must be greater than 0") } privKey, err := crypto.GenerateSecp256k1Key() if err != nil { @@ -210,6 +210,12 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client // Retry missing chunks sequentially to avoid flooding for _, i := range missing { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + payload := fmt.Sprintf("data %d", i) logger.Infof("gsoc: retrying soc to node=%s, payload=%s", uploadClient.Name(), payload) if err := uploadSoc(ctx, uploadClient, payload, resourceId, batches[i%2], privKey); err != nil { From 620110faea120344d191f66d0519c29b3433a7f5 Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Tue, 20 Jan 2026 16:53:10 +0200 Subject: [PATCH 3/6] fix: add chuncks in checks and local config --- config/local.yaml | 1 + pkg/config/check.go | 1 + 2 files changed, 2 insertions(+) diff --git a/config/local.yaml b/config/local.yaml index d389ee2b3..9ac70bb61 100644 --- a/config/local.yaml +++ b/config/local.yaml @@ -432,6 +432,7 @@ checks: postage-ttl: 24h postage-depth: 21 postage-label: test-label + chunks: 3 timeout: 10m type: gsoc ci-feed-v1: diff --git a/pkg/config/check.go b/pkg/config/check.go index 541afb7e5..78221b8c0 100644 --- a/pkg/config/check.go +++ b/pkg/config/check.go @@ -680,6 +680,7 @@ var Checks = map[string]CheckType{ PostageTTL *time.Duration `yaml:"postage-ttl"` PostageDepth *uint64 `yaml:"postage-depth"` PostageLabel *string `yaml:"postage-label"` + Chunks *int `yaml:"chunks"` }) if err := check.Options.Decode(checkOpts); err != nil { return nil, fmt.Errorf("decoding check %s options: %w", check.Type, err) From 157ac554b63a32813498aed35063bc05690e5b3f Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Thu, 27 Aug 2026 23:09:28 +0300 Subject: [PATCH 4/6] fix(check): improve gsoc neighborhood mining and pss prefix --- pkg/check/gsoc/gsoc.go | 88 +++++++------------------------------ pkg/check/gsoc/gsoc_test.go | 42 ++++++++++++++++++ pkg/check/pss/pss.go | 2 +- pkg/check/pss/pss_test.go | 21 +++++++++ 4 files changed, 79 insertions(+), 74 deletions(-) create mode 100644 pkg/check/gsoc/gsoc_test.go create mode 100644 pkg/check/pss/pss_test.go diff --git a/pkg/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index 0d6e21a45..c2e534a51 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/pkg/check/gsoc/gsoc.go @@ -6,8 +6,6 @@ import ( "encoding/binary" "encoding/hex" "fmt" - "strconv" - "strings" "sync" "time" @@ -126,7 +124,7 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client if err != nil { return err } - prefixMatchDepth := 6 + prefixMatchDepth := 8 logger.Infof("gsoc: mining resource id for overlay=%s, prefixMatchDepth=%d", addresses.Overlay, prefixMatchDepth) resourceId, socAddress, err := mineResourceId(ctx, addresses.Overlay, privKey, prefixMatchDepth) if err != nil { @@ -175,54 +173,21 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client return err } - // Wait for all messages to be received or timeout - ticker := time.NewTicker(30 * time.Second) - defer ticker.Stop() - timeout := time.After(3 * time.Minute) - - for { - select { - case <-done: - return nil - case <-timeout: - return fmt.Errorf("timeout: not all messages received") - case <-ticker.C: - receivedMtx.Lock() - if len(received) == numChunks { - receivedMtx.Unlock() - return nil - } - - var missing []int - for i := range numChunks { - want := fmt.Sprintf("data %d", i) - if !received[want] { - missing = append(missing, i) - } - } - receivedMtx.Unlock() - - if len(missing) == 0 { - continue - } - - logger.Infof("gsoc: still missing %d chunks: %v. retrying...", len(missing), missing) - - // Retry missing chunks sequentially to avoid flooding - for _, i := range missing { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - - payload := fmt.Sprintf("data %d", i) - logger.Infof("gsoc: retrying soc to node=%s, payload=%s", uploadClient.Name(), payload) - if err := uploadSoc(ctx, uploadClient, payload, resourceId, batches[i%2], privKey); err != nil { - logger.Errorf("gsoc: retry failed for %s: %v", payload, err) - } + select { + case <-done: + return nil + case <-time.After(3 * time.Minute): + receivedMtx.Lock() + defer receivedMtx.Unlock() + for i := range numChunks { + want := fmt.Sprintf("data %d", i) + if !received[want] { + return fmt.Errorf("message '%s' not received", want) } } + return fmt.Errorf("timeout: not all messages received") + case <-ctx.Done(): + return ctx.Err() } } @@ -265,31 +230,8 @@ func runInParallel(ctx context.Context, client *bee.Client, numChunks int, batch return errG.Wait() } -func getTargetNeighborhood(address swarm.Address, depth int) (string, error) { - var targetNeighborhood strings.Builder - for i := range depth { - hexChar := address.String()[i : i+1] - value, err := strconv.ParseUint(hexChar, 16, 4) - if err != nil { - return "", err - } - fmt.Fprintf(&targetNeighborhood, "%04b", value) - } - return targetNeighborhood.String(), nil -} - func mineResourceId(ctx context.Context, overlay swarm.Address, privKey *ecdsa.PrivateKey, depth int) ([]byte, swarm.Address, error) { - targetNeighborhood, err := getTargetNeighborhood(overlay, depth) - if err != nil { - return nil, swarm.ZeroAddress, err - } - - neighborhood, err := swarm.ParseBitStrAddress(targetNeighborhood) - if err != nil { - return nil, swarm.ZeroAddress, err - } nonce := make([]byte, 32) - prox := len(targetNeighborhood) owner, err := crypto.NewEthereumAddress(privKey.PublicKey) if err != nil { return nil, swarm.ZeroAddress, err @@ -309,7 +251,7 @@ func mineResourceId(ctx context.Context, overlay swarm.Address, privKey *ecdsa.P return nil, swarm.ZeroAddress, err } - if swarm.Proximity(address.Bytes(), neighborhood.Bytes()) >= uint8(prox) { + if swarm.Proximity(address.Bytes(), overlay.Bytes()) >= uint8(depth) { return nonce, address, nil } i++ diff --git a/pkg/check/gsoc/gsoc_test.go b/pkg/check/gsoc/gsoc_test.go new file mode 100644 index 000000000..94ed966b6 --- /dev/null +++ b/pkg/check/gsoc/gsoc_test.go @@ -0,0 +1,42 @@ +package gsoc + +import ( + "context" + "testing" + "time" + + "github.com/ethersphere/bee/v2/pkg/crypto" + "github.com/ethersphere/bee/v2/pkg/swarm" +) + +func TestMineResourceId(t *testing.T) { + t.Parallel() + + privKey, err := crypto.GenerateSecp256k1Key() + if err != nil { + t.Fatal(err) + } + + overlay, err := swarm.ParseHexAddress("aabbccddeeff00112233445566778899aabbccddeeff00112233445566778899") + if err != nil { + t.Fatal(err) + } + + depth := 8 + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + nonce, socAddress, err := mineResourceId(ctx, overlay, privKey, depth) + if err != nil { + t.Fatalf("mineResourceId error: %v", err) + } + + if len(nonce) != 32 { + t.Fatalf("expected 32-byte nonce, got %d bytes", len(nonce)) + } + + prox := swarm.Proximity(socAddress.Bytes(), overlay.Bytes()) + if prox < uint8(depth) { + t.Fatalf("expected proximity >= %d, got %d (soc: %s, overlay: %s)", depth, prox, socAddress, overlay) + } +} diff --git a/pkg/check/pss/pss.go b/pkg/check/pss/pss.go index dd502ee23..5800b0b62 100644 --- a/pkg/check/pss/pss.go +++ b/pkg/check/pss/pss.go @@ -30,7 +30,7 @@ type Options struct { func NewDefaultOptions() Options { return Options{ Count: 1, - AddressPrefix: 1, + AddressPrefix: 2, GasPrice: "", PostageTTL: 24 * time.Hour, PostageDepth: 22, diff --git a/pkg/check/pss/pss_test.go b/pkg/check/pss/pss_test.go new file mode 100644 index 000000000..5dbc964fe --- /dev/null +++ b/pkg/check/pss/pss_test.go @@ -0,0 +1,21 @@ +package pss + +import ( + "testing" + "time" +) + +func TestNewDefaultOptions(t *testing.T) { + t.Parallel() + + opts := NewDefaultOptions() + if opts.AddressPrefix != 2 { + t.Fatalf("expected AddressPrefix 2, got %d", opts.AddressPrefix) + } + if opts.PostageDepth != 22 { + t.Fatalf("expected PostageDepth 22, got %d", opts.PostageDepth) + } + if opts.PostageTTL != 24*time.Hour { + t.Fatalf("expected PostageTTL 24h, got %v", opts.PostageTTL) + } +} From 189242e920e4b754172bb01aa48fee5127ed5144 Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Mon, 31 Aug 2026 14:38:18 +0300 Subject: [PATCH 5/6] fix(check): address review comments on pss retry and gsoc error reporting --- pkg/check/gsoc/gsoc.go | 7 ++++++- pkg/check/pss/pss.go | 31 ++++++++++++++++++++++--------- 2 files changed, 28 insertions(+), 10 deletions(-) diff --git a/pkg/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index c2e534a51..db6d247d0 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/pkg/check/gsoc/gsoc.go @@ -5,6 +5,7 @@ import ( "crypto/ecdsa" "encoding/binary" "encoding/hex" + "errors" "fmt" "sync" "time" @@ -179,12 +180,16 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client case <-time.After(3 * time.Minute): receivedMtx.Lock() defer receivedMtx.Unlock() + var errs []error for i := range numChunks { want := fmt.Sprintf("data %d", i) if !received[want] { - return fmt.Errorf("message '%s' not received", want) + errs = append(errs, fmt.Errorf("message '%s' not received", want)) } } + if len(errs) > 0 { + return errors.Join(errs...) + } return fmt.Errorf("timeout: not all messages received") case <-ctx.Done(): return ctx.Err() diff --git a/pkg/check/pss/pss.go b/pkg/check/pss/pss.go index 5800b0b62..7db770a0f 100644 --- a/pkg/check/pss/pss.go +++ b/pkg/check/pss/pss.go @@ -2,6 +2,7 @@ package pss import ( "context" + "errors" "fmt" "math/rand" "time" @@ -132,32 +133,44 @@ func (c *Check) testPss(nodeAName, nodeBName string, clients map[string]*bee.Cli tStart := time.Now() c.metrics.SendAndReceiveGauge.WithLabelValues(nodeAName, nodeBName).Set(0) - for range 5 { + for i := range 5 { err = nodeA.SendPSSMessage(ctx, addrB.Overlay, addrB.PSSPublicKey, testTopic, o.AddressPrefix, testData, batchID) if err == nil { break } - // check if message is received + c.logger.Infof("pss: send failed: %v", err) + if i == 4 { + break + } + + c.logger.Infof("pss: waiting to retry in 1s") select { - case msg := <-ch: + case msg, ok := <-ch: + if !ok { + return errors.New("pss websocket closed") + } if msg == string(testData) { c.logger.Info("pss: message received despite send failure") + c.metrics.SendAndReceiveGauge.WithLabelValues(nodeAName, nodeBName).Set(time.Since(tStart).Seconds()) return nil } - default: - // continue + case <-time.After(1 * time.Second): + // Retry + case <-ctx.Done(): + return ctx.Err() } - - c.logger.Infof("pss: send failed, retrying in 1s: %v", err) - time.Sleep(1 * time.Second) } if err != nil { // check if message is received select { - case msg := <-ch: + case msg, ok := <-ch: + if !ok { + return errors.New("pss websocket closed") + } if msg == string(testData) { c.logger.Info("pss: message received despite send failure") + c.metrics.SendAndReceiveGauge.WithLabelValues(nodeAName, nodeBName).Set(time.Since(tStart).Seconds()) return nil } default: From e36dc893e09f381f96ce27a59582ce239d0eea87 Mon Sep 17 00:00:00 2001 From: Calin Martinconi Date: Wed, 2 Sep 2026 14:32:28 +0300 Subject: [PATCH 6/6] fix(check): format missing messages using strings.Join in gsoc check --- pkg/check/gsoc/gsoc.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/pkg/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index db6d247d0..3bce562f2 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/pkg/check/gsoc/gsoc.go @@ -5,8 +5,8 @@ import ( "crypto/ecdsa" "encoding/binary" "encoding/hex" - "errors" "fmt" + "strings" "sync" "time" @@ -180,15 +180,15 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client case <-time.After(3 * time.Minute): receivedMtx.Lock() defer receivedMtx.Unlock() - var errs []error + var missing []string for i := range numChunks { want := fmt.Sprintf("data %d", i) if !received[want] { - errs = append(errs, fmt.Errorf("message '%s' not received", want)) + missing = append(missing, want) } } - if len(errs) > 0 { - return errors.Join(errs...) + if len(missing) > 0 { + return fmt.Errorf("messages not received: %s", strings.Join(missing, ", ")) } return fmt.Errorf("timeout: not all messages received") case <-ctx.Done():