fix: gsoc and pss checks

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:

pkg/check/gsoc/gsoc.go

@@ -6,7 +6,6 @@ import (
 	"encoding/binary"
 	"encoding/hex"
 	"fmt"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -29,14 +28,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 +96,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 +112,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 {
+		return fmt.Errorf("chunks must be greater than 0")
+	}
 	privKey, err := crypto.GenerateSecp256k1Key()
 	if err != nil {
 		return err
@@ -122,7 +125,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 {
@@ -172,20 +175,25 @@ func run(ctx context.Context, uploadClient *bee.Client, listenClient *bee.Client
 	}
 
 	select {
-	case <-time.After(3 * time.Minute):
-		return fmt.Errorf("timeout: not all messages received")
 	case <-done:
-	}
-
-	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 nil
+	case <-time.After(3 * time.Minute):
+		receivedMtx.Lock()
+		defer receivedMtx.Unlock()
+		var missing []string
+		for i := range numChunks {
+			want := fmt.Sprintf("data %d", i)
+			if !received[want] {
+				missing = append(missing, want)
+			}
+		}
+		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():
+		return ctx.Err()
 	}
-	return nil
 }
 
 func uploadSoc(ctx context.Context, client *bee.Client, payload string, resourceId []byte, batchID string, privKey *ecdsa.PrivateKey) error {
@@ -227,31 +235,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
@@ -271,7 +256,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++

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)
+	}
+}

pkg/check/pss/pss.go

@@ -2,6 +2,7 @@ package pss
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"time"
@@ -30,10 +31,10 @@ type Options struct {
 func NewDefaultOptions() Options {
 	return Options{
 		Count:          1,
-		AddressPrefix:  1,
+		AddressPrefix:  2,
 		GasPrice:       "",
 		PostageTTL:     24 * time.Hour,
-		PostageDepth:   16,
+		PostageDepth:   22,
 		PostageLabel:   "test-label",
 		RequestTimeout: 5 * time.Minute,
 		Seed:           random.Int64(),
@@ -131,8 +132,50 @@ 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 i := range 5 {
+		err = nodeA.SendPSSMessage(ctx, addrB.Overlay, addrB.PSSPublicKey, testTopic, o.AddressPrefix, testData, batchID)
+		if err == nil {
+			break
+		}
+
+		c.logger.Infof("pss: send failed: %v", err)
+		if i == 4 {
+			break
+		}
+
+		c.logger.Infof("pss: waiting to retry in 1s")
+		select {
+		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
+			}
+		case <-time.After(1 * time.Second):
+			// Retry
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
 	if err != nil {
+		// check if message is received
+		select {
+		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
+		}
 		return err
 	}
 	c.logger.Infof("pss: test data sent successfully to node %s. Waiting for response from node %s", nodeAName, nodeBName)

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)
+	}
+}

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)