Origin/ljubisa radius v3

cmd/beekeeper/cmd/check.go

@@ -22,6 +22,7 @@ func (c *command) initCheckCmd() error {
 		optionNameSeed                 = "seed"
 		optionNameTimeout              = "timeout"
 		optionNameMetricsPusherAddress = "metrics-pusher-address"
+		optionNameParallel             = "parallel"
 	)
 
 	cmd := &cobra.Command{
@@ -103,7 +104,7 @@ Use --metrics-enabled to collect and push metrics to Prometheus.`,
 
 				checkRunner := check.NewCheckRunner(checkGlobalConfig, c.config.Checks, cluster, metricsPusher, tracer, c.log)
 
-				return checkRunner.Run(ctx, checks)
+				return checkRunner.Run(ctx, checks, c.globalConfig.GetBool(optionNameParallel))
 			})
 		},
 		PreRunE: c.preRunE,
@@ -113,6 +114,7 @@ Use --metrics-enabled to collect and push metrics to Prometheus.`,
 	cmd.Flags().String(optionNameMetricsPusherAddress, "pushgateway.staging.internal", "prometheus metrics pusher address")
 	cmd.Flags().Bool(optionNameCreateCluster, false, "creates cluster before executing checks")
 	cmd.Flags().StringSlice(optionNameChecks, []string{"pingpong"}, "list of checks to execute")
+	cmd.Flags().Bool(optionNameParallel, false, "run the --checks list concurrently instead of sequentially (default sequential)")
 	cmd.Flags().Bool(optionNameMetricsEnabled, true, "enable metrics")
 	cmd.Flags().Int64(optionNameSeed, -1, "seed, -1 for random")
 	cmd.Flags().Duration(optionNameTimeout, 30*time.Minute, "timeout")

config/local.yaml

@@ -40,12 +40,12 @@ clusters:
       bee:
         bee-config: bee-local-dns
         config: local-dns
-        count: 3
+        count: 8
         mode: node
       light:
         bee-config: bee-local-light
         config: local-light
-        count: 2
+        count: 0
         mode: node
   local-dns-autotls:
     _inherit: "local"
@@ -101,7 +101,7 @@ clusters:
 node-groups:
   local:
     _inherit: ""
-    image: k3d-registry.localhost:5000/ethersphere/bee:latest
+    image: k3d-registry.localhost:5002/ethersphere/bee:latest
     image-pull-policy: Always
     ingress-annotations:
       nginx.ingress.kubernetes.io/affinity: "cookie"
@@ -374,6 +374,61 @@ checks:
         - light
     timeout: 30m
     type: load
+  # Self-driving radius check: upload random data to random full nodes (own postage label) to
+  # drive the storage radius up to target-radius (3); then wait for pull-sync to go idle
+  # (pullsyncRate==0 — the reserve worker only decreases while SyncRate()==0), dilute one batch
+  # down to the chain's minimum validity so it expires soon (commitment drop), and verify the
+  # radius ticks back down below its peak (3 -> 2). Needs the patched bee image (small reserve).
+  # Locally the reserve-worker idle tick drops the radius in ~10 min (before any expiry); on
+  # testnet the drop is expiry-driven (~minimumValidityBlocks after diluting, a few hours) —
+  # there, raise decrease-timeout to ~6h and timeout to ~8h (a chain postage price must be set).
+  ci-reserve-radius:
+    options:
+      postage-ttl: 48h # must exceed the 24h min-validity floor (24h lands on the boundary and is rejected)
+      postage-depth: 22
+      postage-label: reserve-radius
+      blob-size: 8388608 # 8 MiB — bigger blobs reach the higher target radius faster
+      max-uploads: 200
+      target-radius: 3
+      force-decrease: true # dilute one batch to minimum validity to hasten its expiry (commitment drop)
+      dilute-step: 1 # +1 per dilution ~halves TTL; larger jumps revert at the min-validity floor
+      max-dilutions: 20
+      warmup-wait: 15m
+      increase-timeout: 30m
+      sync-settle-wait: 10m # wait for pullsyncRate==0 before diluting/observing (the decrease gate)
+      decrease-timeout: 30m # local: the idle tick drops it in ~10min. testnet (expiry-driven): ~6h
+      poll-interval: 15s
+      upload-groups:
+        - bee
+    timeout: 45m
+    type: reserve-radius
+  # Simplified reserve-radius scenario (v2, WIP): fill the reserve with small, fixed-size
+  # postage batches until fill-percent crosses target-fill-percent, wait for pull-sync to
+  # settle. NOTE: as of this writing fill()/driveDecrease in reserveradiusv2.go are under
+  # active rework — keep this config in sync with pkg/check/reserveradiusv2/reserveradiusv2.go.
+  ci-reserve-radius-v2:
+    options:
+      postage-amount: 1000000000
+      batch-depth: 18
+      postage-label: reserve-radius-v2
+      blob-size: 1048576 # 1 MiB per upload
+      data-per-batch: 4194304 # 4 MiB = ~1024 chunks per batch — granular fill control
+      max-batches: 50
+      reserve-capacity: 4000 # patched bee reserve capacity in chunks
+      target-fill-percent: 0.6
+      dilute: true
+      dilute-step: 1
+      dilute-interval: 1m # one dilution per minute — distinct events next to radius changes
+      max-dilution-rounds: 5
+      warmup-wait: 15m
+      fill-timeout: 30m
+      sync-settle-wait: 10m
+      decrease-timeout: 30m
+      poll-interval: 10s
+      upload-groups:
+        - bee
+    timeout: 90m
+    type: reserve-radius-v2
   ci-soc:
     options:
       postage-ttl: 24h

pkg/bee/api/api.go

@@ -42,22 +42,23 @@ type Client struct {
 	service    service      // Reuse a single struct instead of allocating one for each service on the heap.
 
 	// Services that API provides.
-	Act         *ActService
-	Bytes       *BytesService
-	Chunks      *ChunksService
-	Dirs        *DirsService
-	Feed        *FeedService
-	Files       *FilesService
-	Node        *NodeService
-	PingPong    *PingPongService
-	Pinning     *PinningService
-	Postage     *PostageService
-	PSS         *PSSService
-	SOC         *SOCService
-	Stake       *StakingService
-	Status      *StatusService
-	Stewardship *StewardshipService
-	Tags        *TagsService
+	Act            *ActService
+	Bytes          *BytesService
+	Chunks         *ChunksService
+	Dirs           *DirsService
+	Feed           *FeedService
+	Files          *FilesService
+	Node           *NodeService
+	PingPong       *PingPongService
+	Pinning        *PinningService
+	Postage        *PostageService
+	PSS            *PSSService
+	Redistribution *RedistributionService
+	SOC            *SOCService
+	Stake          *StakingService
+	Status         *StatusService
+	Stewardship    *StewardshipService
+	Tags           *TagsService
 }
 
 // NewClient constructs a new Client.
@@ -103,6 +104,7 @@ func newClient(apiURL *url.URL, httpClient *http.Client) (c *Client) {
 	c.Pinning = (*PinningService)(&c.service)
 	c.Postage = (*PostageService)(&c.service)
 	c.PSS = (*PSSService)(&c.service)
+	c.Redistribution = (*RedistributionService)(&c.service)
 	c.SOC = (*SOCService)(&c.service)
 	c.Stake = (*StakingService)(&c.service)
 	c.Status = (*StatusService)(&c.service)

pkg/bee/api/redistribution.go

@@ -0,0 +1,35 @@
+package api
+
+import (
+	"context"
+	"net/http"
+)
+
+type RedistributionService service
+
+// RedistributionState is the subset of the node's /redistributionstate response
+// relevant to radius / pull-sync validation: whether the node is frozen, fully
+// synced, healthy, and how far the redistribution round has progressed. These are
+// the signals that distinguish a recovering node from a halted one after a radius
+// change (a node that cannot reach fullySynced or that freezes skips rounds).
+type RedistributionState struct {
+	IsFrozen                  bool    `json:"isFrozen"`
+	IsFullySynced             bool    `json:"isFullySynced"`
+	IsHealthy                 bool    `json:"isHealthy"`
+	HasSufficientFunds        bool    `json:"hasSufficientFunds"`
+	Phase                     string  `json:"phase"`
+	Round                     uint64  `json:"round"`
+	LastWonRound              uint64  `json:"lastWonRound"`
+	LastPlayedRound           uint64  `json:"lastPlayedRound"`
+	LastFrozenRound           uint64  `json:"lastFrozenRound"`
+	LastSelectedRound         uint64  `json:"lastSelectedRound"`
+	LastSampleDurationSeconds float64 `json:"lastSampleDurationSeconds"`
+	Block                     uint64  `json:"block"`
+}
+
+// RedistributionState returns the node's redistribution-game state. This endpoint
+// is full-mode only; it errors on nodes without the redistribution agent enabled.
+func (s *RedistributionService) RedistributionState(ctx context.Context) (resp *RedistributionState, err error) {
+	err = s.client.requestJSON(ctx, http.MethodGet, "/redistributionstate", nil, &resp)
+	return resp, err
+}

pkg/bee/api/status.go

@@ -22,10 +22,29 @@ type StatusResponse struct {
 	IsReachable             bool    `json:"isReachable"`
 	LastSyncedBlock         uint64  `json:"lastSyncedBlock"`
 	CommittedDepth          uint8   `json:"committedDepth"`
+	IsWarmingUp             bool    `json:"isWarmingUp"`
 }
 
 // Ping pings given node
 func (s *StatusService) Status(ctx context.Context) (resp *StatusResponse, err error) {
 	err = s.client.requestJSON(ctx, http.MethodGet, "/status", nil, &resp)
 	return resp, err
 }
+
+type NeighborhoodStat struct {
+	Neighborhood            string `json:"neighborhood"`
+	ReserveSizeWithinRadius int    `json:"reserveSizeWithinRadius"`
+	Proximity               uint8  `json:"proximity"`
+}
+
+type NeighborhoodsResponse struct {
+	Neighborhoods []NeighborhoodStat `json:"neighborhoods"`
+}
+
+// Neighborhoods returns per-neighborhood reserve stats, computed synchronously (uncached)
+// on every call — unlike Status's ReserveSizeWithinRadius, which is refreshed on a fixed
+// interval and can lag a just-written chunk by that interval.
+func (s *StatusService) Neighborhoods(ctx context.Context) (resp *NeighborhoodsResponse, err error) {
+	err = s.client.requestJSON(ctx, http.MethodGet, "/status/neighborhoods", nil, &resp)
+	return resp, err
+}

pkg/bee/client.go

@@ -1089,3 +1089,14 @@ func (c *Client) FindFeedUpdate(ctx context.Context, signer crypto.Signer, topic
 func (c *Client) Status(ctx context.Context) (*api.StatusResponse, error) {
 	return c.api.Status.Status(ctx)
 }
+
+// Neighborhoods returns per-neighborhood reserve stats, computed synchronously (uncached)
+// on every call — use this instead of Status's ReserveSizeWithinRadius when freshness matters.
+func (c *Client) Neighborhoods(ctx context.Context) (*api.NeighborhoodsResponse, error) {
+	return c.api.Status.Neighborhoods(ctx)
+}
+
+// RedistributionState returns the node's redistribution-game state (full-mode only).
+func (c *Client) RedistributionState(ctx context.Context) (*api.RedistributionState, error) {
+	return c.api.Redistribution.RedistributionState(ctx)
+}

pkg/check/load/load.go

@@ -37,6 +37,7 @@ type Options struct {
 	DownloadGroups          []string
 	MaxCommittedDepth       uint8
 	CommittedDepthCheckWait time.Duration
+	DecreaseHold            time.Duration // once target reached and committedDepth drops, pause uploads this long before re-filling (0 = disabled)
 	IterationWait           time.Duration
 }
 
@@ -57,6 +58,7 @@ func NewDefaultOptions() Options {
 		DownloadGroups:          []string{},
 		MaxCommittedDepth:       2,
 		CommittedDepthCheckWait: 5 * time.Minute,
+		DecreaseHold:            0,
 		IterationWait:           5 * time.Minute,
 	}
 }
@@ -70,6 +72,9 @@ var _ beekeeper.Action = (*Check)(nil)
 type Check struct {
 	metrics metrics
 	logger  logging.Logger
+
+	mu            sync.Mutex
+	reachedTarget bool // set once committedDepth reaches max; consumed on the following decrease
 }
 
 func NewCheck(log logging.Logger) beekeeper.Action {
@@ -185,7 +190,7 @@ func (c *Check) run(ctx context.Context, cluster orchestration.Cluster, o Option
 					default:
 					}
 
-					if !c.checkCommittedDepth(ctx, uploader, o.MaxCommittedDepth, o.CommittedDepthCheckWait) {
+					if !c.checkCommittedDepth(ctx, uploader, o.MaxCommittedDepth, o.CommittedDepthCheckWait, o.DecreaseHold) {
 						return
 					}
 
@@ -299,7 +304,7 @@ func (c *Check) run(ctx context.Context, cluster orchestration.Cluster, o Option
 	return nil
 }
 
-func (c *Check) checkCommittedDepth(ctx context.Context, client *bee.Client, maxDepth uint8, wait time.Duration) bool {
+func (c *Check) checkCommittedDepth(ctx context.Context, client *bee.Client, maxDepth uint8, wait, decreaseHold time.Duration) bool {
 	for {
 		statusResp, err := client.Status(ctx)
 		if err != nil {
@@ -308,8 +313,21 @@ func (c *Check) checkCommittedDepth(ctx context.Context, client *bee.Client, max
 		}
 
 		if statusResp.CommittedDepth < maxDepth {
+			// Below target. If committedDepth just dropped from the target and a
+			// decrease-hold is set, pause uploads for that window before re-filling
+			// so the radius decrease + resync can play out without interference.
+			if decreaseHold > 0 && c.consumeReachedTarget() {
+				c.logger.Infof("committedDepth dropped to %d (max %d); holding uploads for %v before re-filling", statusResp.CommittedDepth, maxDepth, decreaseHold)
+				select {
+				case <-ctx.Done():
+					return false
+				case <-time.After(decreaseHold):
+				}
+			}
 			return true
 		}
+
+		c.markReachedTarget()
 		c.logger.Infof("waiting %v for CommittedDepth to decrease. Current: %d, Max: %d", wait, statusResp.CommittedDepth, maxDepth)
 
 		select {
@@ -321,6 +339,24 @@ func (c *Check) checkCommittedDepth(ctx context.Context, client *bee.Client, max
 	}
 }
 
+func (c *Check) markReachedTarget() {
+	c.mu.Lock()
+	c.reachedTarget = true
+	c.mu.Unlock()
+}
+
+// consumeReachedTarget reports whether the target was reached since the last call,
+// resetting the flag so the decrease-hold fires once per fill→drop cycle (re-arming).
+func (c *Check) consumeReachedTarget() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.reachedTarget {
+		c.reachedTarget = false
+		return true
+	}
+	return false
+}
+
 func (c *Check) Report() []prometheus.Collector {
 	return c.metrics.Report()
 }