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/check/gsoc/gsoc.go b/pkg/check/gsoc/gsoc.go index c19b95ab5..3bce562f2 100644 --- a/pkg/check/gsoc/gsoc.go +++ b/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++ 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 6856ad67f..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" @@ -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) 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) + } +} 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)