From 8103ed1125c5e754f0e5d4d04ace0237ed491283 Mon Sep 17 00:00:00 2001 From: sbackend Date: Wed, 26 Aug 2026 14:02:21 +0200 Subject: [PATCH 1/7] feat: draft implementation --- openapi/Swarm.yaml | 33 ++- openapi/SwarmCommon.yaml | 18 ++ pkg/api/api_test.go | 38 ++-- pkg/api/export_test.go | 2 + pkg/api/redistribution.go | 35 +++ pkg/api/redistribution_test.go | 172 +++++++++++++++ pkg/api/router.go | 16 +- pkg/api/router_test.go | 4 + pkg/storageincentives/agent.go | 26 ++- pkg/storageincentives/agent_test.go | 326 +++++++++++++++++++++++++++- 10 files changed, 643 insertions(+), 27 deletions(-) diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index c320410b4ee..3b7fe6abe68 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -1,7 +1,7 @@ openapi: 3.0.3 info: - version: 8.1.1 + version: 8.2.0 title: Bee API description: "API endpoints for interacting with the Swarm network, supporting file operations, messaging, and node management" @@ -2426,6 +2426,37 @@ paths: default: description: Default response + "/redistribution": + put: + summary: Enable or disable participation in new redistribution rounds + description: > + Controls whether the node will commit in a new redistribution round. + Disabling does not abort an in-flight commit and does not skip reveal or claim + for a round that already committed. Re-enabling during a commit phase that was + already skipped does not retry that round; participation resumes on the next + sample/commit cycle. After a node restart participation is enabled again. + tags: + - RedistributionState + requestBody: + required: true + content: + application/json: + schema: + $ref: "SwarmCommon.yaml#/components/schemas/RedistributionEnableRequest" + responses: + "200": + description: Participation flag updated + content: + application/json: + schema: + $ref: "SwarmCommon.yaml#/components/schemas/RedistributionEnableResponse" + "400": + $ref: "SwarmCommon.yaml#/components/responses/400" + "500": + $ref: "SwarmCommon.yaml#/components/responses/500" + default: + description: Default response + "/redistributionstate": get: summary: Get the node's redistribution game status diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index ffcbbac3b8e..ccdad854fc3 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -825,6 +825,9 @@ components: type: boolean isHealthy: type: boolean + enabled: + type: boolean + description: Whether the node will commit in new redistribution rounds. A disabled node still finishes a round that already has an on-chain commit. Re-enabling during a commit phase that was already skipped does not retry that round; participation resumes on the next sample/commit cycle. phase: type: string round: @@ -846,6 +849,21 @@ components: fees: $ref: "#/components/schemas/BigInt" + RedistributionEnableRequest: + type: object + required: + - enabled + properties: + enabled: + type: boolean + description: Whether the node should enter new redistribution rounds. + + RedistributionEnableResponse: + type: object + properties: + enabled: + type: boolean + PendingTransactionsResponse: type: object properties: diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index 23782bcbc6b..e6be53ea4e4 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -22,6 +22,9 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/gorilla/websocket" + "resenje.org/web" + "github.com/ethersphere/bee/v2/pkg/accesscontrol" mockac "github.com/ethersphere/bee/v2/pkg/accesscontrol/mock" accountingmock "github.com/ethersphere/bee/v2/pkg/accounting/mock" @@ -70,8 +73,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/transaction/backendmock" transactionmock "github.com/ethersphere/bee/v2/pkg/transaction/mock" "github.com/ethersphere/bee/v2/pkg/util/testutil" - "github.com/gorilla/websocket" - "resenje.org/web" ) var ( @@ -126,17 +127,18 @@ type testServerOptions struct { BatchStore postage.Storer SyncStatus func() (bool, error) - BackendOpts []backendmock.Option - Erc20Opts []erc20mock.Option - BeeMode api.BeeNodeMode - RedistributionAgent *storageincentives.Agent - NodeStatus *status.Service - PinIntegrity api.PinIntegrity - WhitelistedAddr string - FullAPIDisabled bool - ChequebookDisabled bool - SwapDisabled bool - Erc20ServiceNil bool + BackendOpts []backendmock.Option + Erc20Opts []erc20mock.Option + BeeMode api.BeeNodeMode + RedistributionAgent *storageincentives.Agent + RedistributionAgentDisabled bool + NodeStatus *status.Service + PinIntegrity api.PinIntegrity + WhitelistedAddr string + FullAPIDisabled bool + ChequebookDisabled bool + SwapDisabled bool + Erc20ServiceNil bool } func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.Conn, string, *chanStorer) { @@ -223,11 +225,13 @@ func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket. s.SetP2P(o.P2P) - if o.RedistributionAgent == nil { - o.RedistributionAgent, _ = createRedistributionAgentService(t, o.Overlay, o.StateStorer, erc20, transaction, backend, o.BatchStore) - s.SetRedistributionAgent(o.RedistributionAgent) + if !o.RedistributionAgentDisabled { + if o.RedistributionAgent == nil { + o.RedistributionAgent, _ = createRedistributionAgentService(t, o.Overlay, o.StateStorer, erc20, transaction, backend, o.BatchStore) + s.SetRedistributionAgent(o.RedistributionAgent) + } + testutil.CleanupCloser(t, o.RedistributionAgent) } - testutil.CleanupCloser(t, o.RedistributionAgent) s.SetSwarmAddress(&o.Overlay) s.SetProbe(o.Probe) diff --git a/pkg/api/export_test.go b/pkg/api/export_test.go index 5bda912a3e9..39dcc20356c 100644 --- a/pkg/api/export_test.go +++ b/pkg/api/export_test.go @@ -98,6 +98,8 @@ type ( StakeTransactionReponse = stakeTransactionReponse StatusSnapshotResponse = statusSnapshotResponse StatusResponse = statusResponse + RedistributionStatusResponse = redistributionStatusResponse + RedistributionToggleResponse = redistributionToggleResponse ) var ( diff --git a/pkg/api/redistribution.go b/pkg/api/redistribution.go index bce920e1d72..a3973ca76bd 100644 --- a/pkg/api/redistribution.go +++ b/pkg/api/redistribution.go @@ -5,6 +5,7 @@ package api import ( + "encoding/json" "net/http" "github.com/ethersphere/bee/v2/pkg/bigint" @@ -28,6 +29,15 @@ type redistributionStatusResponse struct { Reward *bigint.BigInt `json:"reward"` Fees *bigint.BigInt `json:"fees"` IsHealthy bool `json:"isHealthy"` + Enabled bool `json:"enabled"` +} + +type redistributionToggleRequest struct { + Enabled *bool `json:"enabled"` +} + +type redistributionToggleResponse struct { + Enabled bool `json:"enabled"` } func (s *Service) redistributionStatusHandler(w http.ResponseWriter, r *http.Request) { @@ -70,5 +80,30 @@ func (s *Service) redistributionStatusHandler(w http.ResponseWriter, r *http.Req Reward: bigint.Wrap(status.Reward), Fees: bigint.Wrap(status.Fees), IsHealthy: status.IsHealthy, + Enabled: s.redistributionAgent.IsEnabled(), }) } + +func (s *Service) redistributionToggleHandler(w http.ResponseWriter, r *http.Request) { + logger := tracing.NewLoggerWithTraceID(r.Context(), s.logger.WithName("put_redistribution").Build()) + + if s.beeMode != FullMode { + jsonhttp.BadRequest(w, errOperationSupportedOnlyInFullMode) + return + } + + var body redistributionToggleRequest + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + logger.Debug("decode body failed", "error", err) + logger.Error(nil, "decode body failed") + jsonhttp.BadRequest(w, "invalid request body") + return + } + if body.Enabled == nil { + jsonhttp.BadRequest(w, "enabled is required") + return + } + + s.redistributionAgent.SetEnabled(*body.Enabled) + jsonhttp.OK(w, redistributionToggleResponse{Enabled: *body.Enabled}) +} diff --git a/pkg/api/redistribution_test.go b/pkg/api/redistribution_test.go index eef01977866..2483d49c37a 100644 --- a/pkg/api/redistribution_test.go +++ b/pkg/api/redistribution_test.go @@ -5,12 +5,14 @@ package api_test import ( + "bytes" "context" "math/big" "net/http" "testing" "github.com/ethereum/go-ethereum/common" + "github.com/ethersphere/bee/v2/pkg/api" "github.com/ethersphere/bee/v2/pkg/jsonhttp" "github.com/ethersphere/bee/v2/pkg/jsonhttp/jsonhttptest" @@ -51,9 +53,14 @@ func TestRedistributionStatus(t *testing.T) { }), }, }) + var got api.RedistributionStatusResponse jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusOK, jsonhttptest.WithRequestHeader(api.ContentTypeHeader, "application/json; charset=utf-8"), + jsonhttptest.WithUnmarshalJSONResponse(&got), ) + if !got.Enabled { + t.Fatal("expected redistribution to be enabled by default") + } }) t.Run("bad request", func(t *testing.T) { @@ -75,4 +82,169 @@ func TestRedistributionStatus(t *testing.T) { }), ) }) + + t.Run("forbidden when agent missing", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, testServerOptions{ + RedistributionAgentDisabled: true, + }) + jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusForbidden, + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "Storage incentives are disabled. This endpoint is unavailable.", + Code: http.StatusForbidden, + }), + ) + }) +} + +func redistributionTestOpts(t *testing.T) testServerOptions { + t.Helper() + + store := statestore.NewStateStore() + if err := store.Put("redistribution_state", storageincentives.Status{ + Phase: storageincentives.PhaseType(1), + Round: 1, + Block: 12, + }); err != nil { + t.Fatal(err) + } + + return testServerOptions{ + StateStorer: store, + TransactionOpts: []mock.Option{ + mock.WithTransactionFeeFunc(func(ctx context.Context, txHash common.Hash) (*big.Int, error) { + return big.NewInt(1000), nil + }), + }, + BackendOpts: []backendmock.Option{ + backendmock.WithBalanceAt(func(ctx context.Context, address common.Address, block *big.Int) (*big.Int, error) { + return big.NewInt(100000000), nil + }), + backendmock.WithSuggestedFeeAndTipFunc(func(ctx context.Context, gasPrice *big.Int, boostPercent int) (*big.Int, *big.Int, error) { + return big.NewInt(1), big.NewInt(2), nil + }), + }, + } +} + +func TestRedistributionToggle(t *testing.T) { + t.Parallel() + + t.Run("put false then true", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) + + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusOK, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), + jsonhttptest.WithExpectedJSONResponse(api.RedistributionToggleResponse{Enabled: false}), + ) + + var got api.RedistributionStatusResponse + jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusOK, + jsonhttptest.WithUnmarshalJSONResponse(&got), + ) + if got.Enabled { + t.Fatal("expected redistribution to be disabled") + } + + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusOK, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": true}), + jsonhttptest.WithExpectedJSONResponse(api.RedistributionToggleResponse{Enabled: true}), + ) + + jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusOK, + jsonhttptest.WithUnmarshalJSONResponse(&got), + ) + if !got.Enabled { + t.Fatal("expected redistribution to be enabled") + } + }) + + t.Run("missing enabled", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.WithJSONRequestBody(map[string]any{}), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "enabled is required", + Code: http.StatusBadRequest, + }), + ) + }) + + t.Run("null enabled", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": nil}), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "enabled is required", + Code: http.StatusBadRequest, + }), + ) + }) + + t.Run("malformed json", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.WithRequestHeader(api.ContentTypeHeader, "application/json"), + jsonhttptest.WithRequestBody(bytes.NewReader([]byte("{invalid"))), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "invalid request body", + Code: http.StatusBadRequest, + }), + ) + }) + + t.Run("light mode", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, testServerOptions{ + BeeMode: api.LightMode, + StateStorer: statestore.NewStateStore(), + }) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: api.ErrOperationSupportedOnlyInFullMode.Error(), + Code: http.StatusBadRequest, + }), + ) + }) + + t.Run("forbidden when agent missing", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, testServerOptions{ + RedistributionAgentDisabled: true, + }) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusForbidden, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "Storage incentives are disabled. This endpoint is unavailable.", + Code: http.StatusForbidden, + }), + ) + }) + + t.Run("unavailable when full api disabled", func(t *testing.T) { + t.Parallel() + + srv, _, _, _ := newTestServer(t, testServerOptions{ + FullAPIDisabled: true, + }) + jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusServiceUnavailable, + jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), + jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ + Message: "Node is syncing. This endpoint is unavailable. Try again later.", + Code: http.StatusServiceUnavailable, + }), + ) + }) } diff --git a/pkg/api/router.go b/pkg/api/router.go index 941c63fc89e..cb9cdd3ae25 100644 --- a/pkg/api/router.go +++ b/pkg/api/router.go @@ -11,15 +11,16 @@ import ( "net/http/pprof" "strings" - "github.com/ethersphere/bee/v2/pkg/jsonhttp" - "github.com/ethersphere/bee/v2/pkg/log/httpaccess" - "github.com/ethersphere/bee/v2/pkg/swarm" - "github.com/ethersphere/bee/v2/pkg/transaction/backendnoop" "github.com/felixge/fgprof" "github.com/gorilla/handlers" "github.com/gorilla/mux" "github.com/prometheus/client_golang/prometheus/promhttp" "resenje.org/web" + + "github.com/ethersphere/bee/v2/pkg/jsonhttp" + "github.com/ethersphere/bee/v2/pkg/log/httpaccess" + "github.com/ethersphere/bee/v2/pkg/swarm" + "github.com/ethersphere/bee/v2/pkg/transaction/backendnoop" ) const ( @@ -679,6 +680,13 @@ func (s *Service) mountBusinessDebug() { })), ) + handle("/redistribution", web.ChainHandlers( + s.checkStorageIncentivesAvailability, + web.FinalHandler(jsonhttp.MethodHandler{ + "PUT": http.HandlerFunc(s.redistributionToggleHandler), + })), + ) + handle("/status", jsonhttp.MethodHandler{ "GET": web.ChainHandlers( httpaccess.NewHTTPAccessSuppressLogHandler(), diff --git a/pkg/api/router_test.go b/pkg/api/router_test.go index db70b645446..a53596898f2 100644 --- a/pkg/api/router_test.go +++ b/pkg/api/router_test.go @@ -114,6 +114,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, + {"/redistribution", []string{"PUT"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, @@ -209,6 +210,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/{amount}", nil, http.StatusServiceUnavailable}, {"/stake", nil, http.StatusServiceUnavailable}, {"/redistributionstate", nil, http.StatusServiceUnavailable}, + {"/redistribution", nil, http.StatusServiceUnavailable}, {"/status", nil, http.StatusServiceUnavailable}, {"/status/peers", nil, http.StatusServiceUnavailable}, {"/status/neighborhoods", nil, http.StatusServiceUnavailable}, @@ -304,6 +306,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, + {"/redistribution", []string{"PUT"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, @@ -399,6 +402,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, + {"/redistribution", []string{"PUT"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, diff --git a/pkg/storageincentives/agent.go b/pkg/storageincentives/agent.go index 4a1c0a6e994..7797847f939 100644 --- a/pkg/storageincentives/agent.go +++ b/pkg/storageincentives/agent.go @@ -12,10 +12,13 @@ import ( "io" "math/big" "sync" + "sync/atomic" "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core/types" + "resenje.org/singleflight" + "github.com/ethersphere/bee/v2/pkg/crypto" "github.com/ethersphere/bee/v2/pkg/log" "github.com/ethersphere/bee/v2/pkg/postage" @@ -28,7 +31,6 @@ import ( "github.com/ethersphere/bee/v2/pkg/storer" "github.com/ethersphere/bee/v2/pkg/swarm" "github.com/ethersphere/bee/v2/pkg/transaction" - "resenje.org/singleflight" ) const loggerName = "storageincentives" @@ -73,6 +75,7 @@ type Agent struct { commitLock sync.Mutex health Health sampleFlight singleflight.Group[string, sampleResult] + disabled atomic.Bool } func New(overlay swarm.Address, @@ -269,6 +272,11 @@ func (a *Agent) handleCommit(ctx context.Context, round uint64) error { a.commitLock.Lock() defer a.commitLock.Unlock() + if !a.IsEnabled() { + a.logger.Info("skipping commit because redistribution is disabled", "round", round) + return nil + } + if _, exists := a.state.CommitKey(round); exists { // already committed on this round, phase is skipped return nil @@ -403,6 +411,11 @@ func (a *Agent) handleSample(ctx context.Context, round uint64) (bool, error) { return false, nil } + if !a.IsEnabled() { + a.logger.Info("skipping round because redistribution is disabled", "round", round) + return false, nil + } + isPlaying, err := a.contract.IsPlaying(ctx, committedDepth) if err != nil { a.metrics.ErrCheckIsPlaying.Inc() @@ -581,6 +594,17 @@ func (a *Agent) Status() (*Status, error) { return a.state.Status() } +// SetEnabled controls whether the node may enter new redistribution rounds. +// Disabling does not abort an in-flight commit or skip reveal/claim of a round +// that already has a commit key. +func (a *Agent) SetEnabled(enabled bool) { + a.disabled.Store(!enabled) +} + +func (a *Agent) IsEnabled() bool { + return !a.disabled.Load() +} + type SampleWithProofs struct { Hash swarm.Address `json:"hash"` Proofs redistribution.ChunkInclusionProofs `json:"proofs"` diff --git a/pkg/storageincentives/agent_test.go b/pkg/storageincentives/agent_test.go index 6449ede9059..79135048730 100644 --- a/pkg/storageincentives/agent_test.go +++ b/pkg/storageincentives/agent_test.go @@ -160,6 +160,296 @@ func TestAgent(t *testing.T) { } } +func TestAgentEnabledDefault(t *testing.T) { + t.Parallel() + + synctest.Test(t, func(t *testing.T) { + backend := &mockchainBackend{ + incrementBy: 1, + block: 9, + limit: 18, + balance: big.NewInt(4_000_000_000), + } + contract := &mockContract{t: t, expectedRadius: 8} + service, err := createService(t, swarm.RandAddress(t), backend, contract, 9, 3, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + if !service.IsEnabled() { + t.Fatal("expected redistribution to be enabled by default") + } + + service.SetEnabled(false) + if service.IsEnabled() { + t.Fatal("expected redistribution to be disabled") + } + + service.SetEnabled(true) + if !service.IsEnabled() { + t.Fatal("expected redistribution to be enabled") + } + }) +} + +func TestAgentParticipationToggle(t *testing.T) { + t.Parallel() + + const ( + blocksPerRound = uint64(9) + blocksPerPhase = uint64(3) + limit = uint64(108) + ) + bigBalance := big.NewInt(4_000_000_000) + + t.Run("disabled before sample skips playing and commit", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wait := make(chan struct{}, 1) + backend := &mockchainBackend{ + limit: limit, + limitCallback: func() { + wait <- struct{}{} + }, + incrementBy: 1, + block: blocksPerRound, + balance: bigBalance, + } + contract := &mockContract{t: t, expectedRadius: 8} + service, err := createService(t, swarm.RandAddress(t), backend, contract, blocksPerRound, blocksPerPhase, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + service.SetEnabled(false) + + <-wait + synctest.Wait() + + if got := contract.playingCount(); got != 0 { + t.Fatalf("expected no isPlaying calls, got %d", got) + } + if got := contract.countCalls(commitCall); got != 0 { + t.Fatalf("expected no commit calls, got %d", got) + } + if got := contract.countCalls(revealCall); got != 0 { + t.Fatalf("expected no reveal calls, got %d", got) + } + }) + }) + + t.Run("disable after sample skips commit", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wait := make(chan struct{}, 1) + started := make(chan struct{}) + unblock := make(chan struct{}) + var once sync.Once + backend := &mockchainBackend{ + limit: limit, + limitCallback: func() { + wait <- struct{}{} + }, + incrementBy: 1, + block: blocksPerRound, + balance: bigBalance, + } + contract := &mockContract{ + t: t, + expectedRadius: 8, + beforeIsPlaying: func() { + once.Do(func() { close(started) }) + <-unblock + }, + } + service, err := createService(t, swarm.RandAddress(t), backend, contract, blocksPerRound, blocksPerPhase, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + <-started + service.SetEnabled(false) + close(unblock) + + <-wait + synctest.Wait() + + if got := contract.playingCount(); got == 0 { + t.Fatal("expected isPlaying to run before disable") + } + if got := contract.countCalls(commitCall); got != 0 { + t.Fatalf("expected no commit calls, got %d", got) + } + }) + }) + + t.Run("disable after commit still reveals", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wait := make(chan struct{}, 1) + committed := make(chan struct{}) + var once sync.Once + backend := &mockchainBackend{ + limit: limit, + limitCallback: func() { + wait <- struct{}{} + }, + incrementBy: 1, + block: blocksPerRound, + balance: bigBalance, + } + contract := &mockContract{ + t: t, + expectedRadius: 8, + beforeCommit: func() { + once.Do(func() { close(committed) }) + }, + } + service, err := createService(t, swarm.RandAddress(t), backend, contract, blocksPerRound, blocksPerPhase, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + <-committed + service.SetEnabled(false) + + <-wait + synctest.Wait() + + if got := contract.countCalls(commitCall); got != 1 { + t.Fatalf("expected exactly one commit, got %d", got) + } + if got := contract.countCalls(revealCall); got != 1 { + t.Fatalf("expected reveal after commit, got %d", got) + } + if got := contract.countCalls(isWinnerCall); got != 1 { + t.Fatalf("expected claim-phase winner check, got %d", got) + } + }) + }) + + t.Run("disable during commit still reveals", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wait := make(chan struct{}, 1) + started := make(chan struct{}) + unblock := make(chan struct{}) + var once sync.Once + backend := &mockchainBackend{ + limit: limit, + limitCallback: func() { + wait <- struct{}{} + }, + incrementBy: 1, + block: blocksPerRound, + balance: bigBalance, + } + contract := &mockContract{ + t: t, + expectedRadius: 8, + beforeCommit: func() { + once.Do(func() { close(started) }) + <-unblock + }, + } + service, err := createService(t, swarm.RandAddress(t), backend, contract, blocksPerRound, blocksPerPhase, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + <-started + service.SetEnabled(false) + close(unblock) + + <-wait + synctest.Wait() + + if got := contract.countCalls(commitCall); got != 1 { + t.Fatalf("expected in-flight commit to finish, got %d", got) + } + if got := contract.countCalls(revealCall); got != 1 { + t.Fatalf("expected reveal after in-flight commit, got %d", got) + } + }) + }) + + t.Run("re-enable in same commit phase does not retry", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + wait := make(chan struct{}, 1) + started := make(chan struct{}) + unblock := make(chan struct{}) + var once sync.Once + backend := &mockchainBackend{ + limit: limit, + limitCallback: func() { + wait <- struct{}{} + }, + incrementBy: 1, + block: blocksPerRound, + balance: bigBalance, + } + contract := &mockContract{ + t: t, + expectedRadius: 8, + beforeIsPlaying: func() { + once.Do(func() { close(started) }) + <-unblock + }, + } + service, err := createService(t, swarm.RandAddress(t), backend, contract, blocksPerRound, blocksPerPhase, 8, 0) + if err != nil { + t.Fatal(err) + } + testutil.CleanupCloser(t, service) + + <-started + service.SetEnabled(false) + close(unblock) + + waitUntilStatus(t, service, func(status *storageincentives.Status) bool { + return status.Phase.String() == "commit" && status.Round >= 2 + }) + synctest.Wait() + + if got := contract.countCalls(commitCall); got != 0 { + t.Fatalf("expected skipped commit, got %d", got) + } + + service.SetEnabled(true) + time.Sleep(200 * time.Millisecond) + synctest.Wait() + + if got := contract.countCalls(commitCall); got != 0 { + t.Fatalf("re-enable in the same commit phase should not commit, got %d", got) + } + + <-wait + synctest.Wait() + + if got := contract.countCalls(commitCall); got == 0 { + t.Fatal("expected commit after the next sample cycle") + } + }) + }) +} + +func waitUntilStatus(t *testing.T, agent *storageincentives.Agent, pred func(*storageincentives.Status) bool) { + t.Helper() + + deadline := time.Now().Add(time.Second) + for { + status, err := agent.Status() + if err == nil && pred(status) { + return + } + if time.Now().After(deadline) { + t.Fatal("timeout waiting for redistribution status") + } + time.Sleep(time.Millisecond) + } +} + func createService( t *testing.T, addr swarm.Address, @@ -271,10 +561,13 @@ const ( ) type mockContract struct { - callsList []contractCall - mtx sync.Mutex - expectedRadius uint8 - t *testing.T + callsList []contractCall + mtx sync.Mutex + expectedRadius uint8 + t *testing.T + isPlayingCount int + beforeIsPlaying func() + beforeCommit func() } // getCalls returns a snapshot of the calls list @@ -289,14 +582,36 @@ func (m *mockContract) getCalls() []contractCall { return calls } +func (m *mockContract) countCalls(call contractCall) int { + n := 0 + for _, c := range m.getCalls() { + if c == call { + n++ + } + } + return n +} + +func (m *mockContract) playingCount() int { + m.mtx.Lock() + defer m.mtx.Unlock() + return m.isPlayingCount +} + func (m *mockContract) ReserveSalt(context.Context) ([]byte, error) { return nil, nil } func (m *mockContract) IsPlaying(_ context.Context, r uint8) (bool, error) { + if m.beforeIsPlaying != nil { + m.beforeIsPlaying() + } if r != m.expectedRadius { m.t.Fatalf("isPlaying: expected radius %d, got %d", m.expectedRadius, r) } + m.mtx.Lock() + m.isPlayingCount++ + m.mtx.Unlock() return true, nil } @@ -315,6 +630,9 @@ func (m *mockContract) Claim(context.Context, redistribution.ChunkInclusionProof } func (m *mockContract) Commit(context.Context, []byte, uint64) (common.Hash, error) { + if m.beforeCommit != nil { + m.beforeCommit() + } m.mtx.Lock() defer m.mtx.Unlock() m.callsList = append(m.callsList, commitCall) From a41eea0c5d8e13ed7547378efa0b25a201921a65 Mon Sep 17 00:00:00 2001 From: sbackend Date: Wed, 26 Aug 2026 19:26:04 +0200 Subject: [PATCH 2/7] fix: clean up --- pkg/api/redistribution_test.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/pkg/api/redistribution_test.go b/pkg/api/redistribution_test.go index 2483d49c37a..b7f83da0c23 100644 --- a/pkg/api/redistribution_test.go +++ b/pkg/api/redistribution_test.go @@ -82,20 +82,6 @@ func TestRedistributionStatus(t *testing.T) { }), ) }) - - t.Run("forbidden when agent missing", func(t *testing.T) { - t.Parallel() - - srv, _, _, _ := newTestServer(t, testServerOptions{ - RedistributionAgentDisabled: true, - }) - jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusForbidden, - jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ - Message: "Storage incentives are disabled. This endpoint is unavailable.", - Code: http.StatusForbidden, - }), - ) - }) } func redistributionTestOpts(t *testing.T) testServerOptions { From 19c906ea9791bd6128b8ef4116af881eba6724b0 Mon Sep 17 00:00:00 2001 From: sbackend Date: Fri, 28 Aug 2026 12:28:04 +0200 Subject: [PATCH 3/7] fix: clean up --- openapi/Swarm.yaml | 9 +++++---- openapi/SwarmCommon.yaml | 4 ++-- pkg/storageincentives/agent.go | 5 ----- pkg/storageincentives/agent_test.go | 6 +++--- 4 files changed, 10 insertions(+), 14 deletions(-) diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index 3b7fe6abe68..52d3e2e5e6b 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -2431,10 +2431,11 @@ paths: summary: Enable or disable participation in new redistribution rounds description: > Controls whether the node will commit in a new redistribution round. - Disabling does not abort an in-flight commit and does not skip reveal or claim - for a round that already committed. Re-enabling during a commit phase that was - already skipped does not retry that round; participation resumes on the next - sample/commit cycle. After a node restart participation is enabled again. + Sampling still runs while disabled so that a later re-enable can commit + in the following commit phase. Disabling does not abort an in-flight + commit and does not skip reveal or claim for a round that already committed. + Re-enabling during a commit phase that was already skipped does not retry + that round. After a node restart participation is enabled again. tags: - RedistributionState requestBody: diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index ccdad854fc3..18be2d6e860 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -827,7 +827,7 @@ components: type: boolean enabled: type: boolean - description: Whether the node will commit in new redistribution rounds. A disabled node still finishes a round that already has an on-chain commit. Re-enabling during a commit phase that was already skipped does not retry that round; participation resumes on the next sample/commit cycle. + description: Whether the node will commit in new redistribution rounds. Sampling still runs while disabled. A disabled node still finishes a round that already has an on-chain commit. Re-enabling during a commit phase that was already skipped does not retry that round. phase: type: string round: @@ -856,7 +856,7 @@ components: properties: enabled: type: boolean - description: Whether the node should enter new redistribution rounds. + description: Whether the node should commit in new redistribution rounds. Sampling still runs while disabled. RedistributionEnableResponse: type: object diff --git a/pkg/storageincentives/agent.go b/pkg/storageincentives/agent.go index 7797847f939..7f464f63e92 100644 --- a/pkg/storageincentives/agent.go +++ b/pkg/storageincentives/agent.go @@ -411,11 +411,6 @@ func (a *Agent) handleSample(ctx context.Context, round uint64) (bool, error) { return false, nil } - if !a.IsEnabled() { - a.logger.Info("skipping round because redistribution is disabled", "round", round) - return false, nil - } - isPlaying, err := a.contract.IsPlaying(ctx, committedDepth) if err != nil { a.metrics.ErrCheckIsPlaying.Inc() diff --git a/pkg/storageincentives/agent_test.go b/pkg/storageincentives/agent_test.go index 79135048730..8da336e992e 100644 --- a/pkg/storageincentives/agent_test.go +++ b/pkg/storageincentives/agent_test.go @@ -203,7 +203,7 @@ func TestAgentParticipationToggle(t *testing.T) { ) bigBalance := big.NewInt(4_000_000_000) - t.Run("disabled before sample skips playing and commit", func(t *testing.T) { + t.Run("disabled before sample still samples but skips commit", func(t *testing.T) { synctest.Test(t, func(t *testing.T) { wait := make(chan struct{}, 1) backend := &mockchainBackend{ @@ -227,8 +227,8 @@ func TestAgentParticipationToggle(t *testing.T) { <-wait synctest.Wait() - if got := contract.playingCount(); got != 0 { - t.Fatalf("expected no isPlaying calls, got %d", got) + if got := contract.playingCount(); got == 0 { + t.Fatal("expected sampling to run while disabled") } if got := contract.countCalls(commitCall); got != 0 { t.Fatalf("expected no commit calls, got %d", got) From 78137c5b72c837870bea994b0f37e262adf738ca Mon Sep 17 00:00:00 2001 From: sbackend Date: Thu, 10 Sep 2026 14:26:10 +0200 Subject: [PATCH 4/7] fix: address comments --- openapi/Swarm.yaml | 40 ++++++----- openapi/SwarmCommon.yaml | 6 ++ pkg/api/api.go | 2 +- pkg/api/cors_test.go | 9 +++ pkg/api/redistribution.go | 7 +- pkg/api/redistribution_test.go | 114 ++++++++++++++++++++++++++++--- pkg/api/router.go | 10 +-- pkg/api/router_test.go | 10 +-- pkg/node/node.go | 6 ++ pkg/storageincentives/agent.go | 6 ++ pkg/storageincentives/metrics.go | 7 ++ 11 files changed, 169 insertions(+), 48 deletions(-) diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index 52d3e2e5e6b..deed70b0d67 100644 --- a/openapi/Swarm.yaml +++ b/openapi/Swarm.yaml @@ -2426,8 +2426,25 @@ paths: default: description: Default response - "/redistribution": - put: + "/redistributionstate": + get: + summary: Get the node's redistribution game status + tags: + - RedistributionState + responses: + "200": + description: Redistribution status info + content: + application/json: + schema: + $ref: "SwarmCommon.yaml#/components/schemas/RedistributionStatusResponse" + "400": + $ref: "SwarmCommon.yaml#/components/responses/400" + "500": + $ref: "SwarmCommon.yaml#/components/responses/500" + default: + description: Default response + patch: summary: Enable or disable participation in new redistribution rounds description: > Controls whether the node will commit in a new redistribution round. @@ -2457,25 +2474,6 @@ paths: $ref: "SwarmCommon.yaml#/components/responses/500" default: description: Default response - - "/redistributionstate": - get: - summary: Get the node's redistribution game status - tags: - - RedistributionState - responses: - "200": - description: Redistribution status info - content: - application/json: - schema: - $ref: "SwarmCommon.yaml#/components/schemas/RedistributionStatusResponse" - "400": - $ref: "SwarmCommon.yaml#/components/responses/400" - "500": - $ref: "SwarmCommon.yaml#/components/responses/500" - default: - description: Default response "/wallet": get: summary: Get wallet balance for BZZ and xDAI diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index 18be2d6e860..de8d858d052 100644 --- a/openapi/SwarmCommon.yaml +++ b/openapi/SwarmCommon.yaml @@ -828,6 +828,12 @@ components: enabled: type: boolean description: Whether the node will commit in new redistribution rounds. Sampling still runs while disabled. A disabled node still finishes a round that already has an on-chain commit. Re-enabling during a commit phase that was already skipped does not retry that round. + hasCommittedThisRound: + type: boolean + description: Whether the node has already committed in the current round. Operators can shut down without freeze risk when this is false. + hasRevealedThisRound: + type: boolean + description: Whether the node has revealed in the current round. After a commit, the node must reveal before it is safe to shut down. phase: type: string round: diff --git a/pkg/api/api.go b/pkg/api/api.go index 63a04c390ff..a6b74837ef4 100644 --- a/pkg/api/api.go +++ b/pkg/api/api.go @@ -615,7 +615,7 @@ func (s *Service) corsHandler(h http.Handler) http.Handler { w.Header().Set("Access-Control-Allow-Credentials", "true") w.Header().Set("Access-Control-Allow-Origin", o) w.Header().Set("Access-Control-Allow-Headers", allowedHeadersStr) - w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, OPTIONS, POST, PUT, DELETE") + w.Header().Set("Access-Control-Allow-Methods", "GET, HEAD, OPTIONS, POST, PUT, PATCH, DELETE") w.Header().Set("Access-Control-Max-Age", "3600") } h.ServeHTTP(w, r) diff --git a/pkg/api/cors_test.go b/pkg/api/cors_test.go index 9a45fd5fade..715987edf42 100644 --- a/pkg/api/cors_test.go +++ b/pkg/api/cors_test.go @@ -130,6 +130,10 @@ func TestCors(t *testing.T) { endpoint: "tags", expectedMethods: "GET, POST", }, + { + endpoint: "redistributionstate", + expectedMethods: "GET, PATCH", + }, { endpoint: "bzz", expectedMethods: "POST", @@ -184,6 +188,11 @@ func TestCorsStatus(t *testing.T) { notAllowedMethods: http.MethodDelete, allowedMethods: "GET, POST", }, + { + endpoint: "redistributionstate", + notAllowedMethods: http.MethodPut, + allowedMethods: "GET, PATCH", + }, { endpoint: "bzz", notAllowedMethods: http.MethodDelete, diff --git a/pkg/api/redistribution.go b/pkg/api/redistribution.go index a3973ca76bd..2edd0087ad0 100644 --- a/pkg/api/redistribution.go +++ b/pkg/api/redistribution.go @@ -30,6 +30,8 @@ type redistributionStatusResponse struct { Fees *bigint.BigInt `json:"fees"` IsHealthy bool `json:"isHealthy"` Enabled bool `json:"enabled"` + HasCommittedThisRound bool `json:"hasCommittedThisRound"` + HasRevealedThisRound bool `json:"hasRevealedThisRound"` } type redistributionToggleRequest struct { @@ -64,6 +66,7 @@ func (s *Service) redistributionStatusHandler(w http.ResponseWriter, r *http.Req return } + rd := status.RoundData[status.Round] jsonhttp.OK(w, redistributionStatusResponse{ MinimumGasFunds: bigint.Wrap(minGasFunds), HasSufficientFunds: hasSufficientFunds, @@ -81,11 +84,13 @@ func (s *Service) redistributionStatusHandler(w http.ResponseWriter, r *http.Req Fees: bigint.Wrap(status.Fees), IsHealthy: status.IsHealthy, Enabled: s.redistributionAgent.IsEnabled(), + HasCommittedThisRound: rd.CommitKey != nil, + HasRevealedThisRound: rd.HasRevealed, }) } func (s *Service) redistributionToggleHandler(w http.ResponseWriter, r *http.Request) { - logger := tracing.NewLoggerWithTraceID(r.Context(), s.logger.WithName("put_redistribution").Build()) + logger := tracing.NewLoggerWithTraceID(r.Context(), s.logger.WithName("patch_redistributionstate").Build()) if s.beeMode != FullMode { jsonhttp.BadRequest(w, errOperationSupportedOnlyInFullMode) diff --git a/pkg/api/redistribution_test.go b/pkg/api/redistribution_test.go index b7f83da0c23..879657c9e1d 100644 --- a/pkg/api/redistribution_test.go +++ b/pkg/api/redistribution_test.go @@ -12,7 +12,6 @@ import ( "testing" "github.com/ethereum/go-ethereum/common" - "github.com/ethersphere/bee/v2/pkg/api" "github.com/ethersphere/bee/v2/pkg/jsonhttp" "github.com/ethersphere/bee/v2/pkg/jsonhttp/jsonhttptest" @@ -61,6 +60,101 @@ func TestRedistributionStatus(t *testing.T) { if !got.Enabled { t.Fatal("expected redistribution to be enabled by default") } + if got.HasCommittedThisRound { + t.Fatal("expected no commit in the current round") + } + if got.HasRevealedThisRound { + t.Fatal("expected no reveal in the current round") + } + }) + + t.Run("committed and revealed flags", func(t *testing.T) { + t.Parallel() + + store := statestore.NewStateStore() + err := store.Put("redistribution_state", storageincentives.Status{ + Phase: storageincentives.PhaseType(1), + Round: 1, + Block: 12, + RoundData: map[uint64]storageincentives.RoundData{ + 1: { + CommitKey: []byte{1, 2, 3}, + HasRevealed: true, + }, + }, + }) + if err != nil { + t.Errorf("redistribution put state: %v", err) + } + srv, _, _, _ := newTestServer(t, testServerOptions{ + StateStorer: store, + TransactionOpts: []mock.Option{ + mock.WithTransactionFeeFunc(func(ctx context.Context, txHash common.Hash) (*big.Int, error) { + return big.NewInt(1000), nil + }), + }, + BackendOpts: []backendmock.Option{ + backendmock.WithBalanceAt(func(ctx context.Context, address common.Address, block *big.Int) (*big.Int, error) { + return big.NewInt(100000000), nil + }), + backendmock.WithSuggestedFeeAndTipFunc(func(ctx context.Context, gasPrice *big.Int, boostPercent int) (*big.Int, *big.Int, error) { + return big.NewInt(1), big.NewInt(2), nil + }), + }, + }) + var got api.RedistributionStatusResponse + jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusOK, + jsonhttptest.WithUnmarshalJSONResponse(&got), + ) + if !got.HasCommittedThisRound { + t.Fatal("expected commit in the current round") + } + if !got.HasRevealedThisRound { + t.Fatal("expected reveal in the current round") + } + }) + + t.Run("committed but not revealed", func(t *testing.T) { + t.Parallel() + + store := statestore.NewStateStore() + err := store.Put("redistribution_state", storageincentives.Status{ + Phase: storageincentives.PhaseType(1), + Round: 1, + Block: 12, + RoundData: map[uint64]storageincentives.RoundData{ + 1: {CommitKey: []byte{1, 2, 3}}, + }, + }) + if err != nil { + t.Errorf("redistribution put state: %v", err) + } + srv, _, _, _ := newTestServer(t, testServerOptions{ + StateStorer: store, + TransactionOpts: []mock.Option{ + mock.WithTransactionFeeFunc(func(ctx context.Context, txHash common.Hash) (*big.Int, error) { + return big.NewInt(1000), nil + }), + }, + BackendOpts: []backendmock.Option{ + backendmock.WithBalanceAt(func(ctx context.Context, address common.Address, block *big.Int) (*big.Int, error) { + return big.NewInt(100000000), nil + }), + backendmock.WithSuggestedFeeAndTipFunc(func(ctx context.Context, gasPrice *big.Int, boostPercent int) (*big.Int, *big.Int, error) { + return big.NewInt(1), big.NewInt(2), nil + }), + }, + }) + var got api.RedistributionStatusResponse + jsonhttptest.Request(t, srv, http.MethodGet, "/redistributionstate", http.StatusOK, + jsonhttptest.WithUnmarshalJSONResponse(&got), + ) + if !got.HasCommittedThisRound { + t.Fatal("expected commit in the current round") + } + if got.HasRevealedThisRound { + t.Fatal("expected no reveal in the current round") + } }) t.Run("bad request", func(t *testing.T) { @@ -117,12 +211,12 @@ func redistributionTestOpts(t *testing.T) testServerOptions { func TestRedistributionToggle(t *testing.T) { t.Parallel() - t.Run("put false then true", func(t *testing.T) { + t.Run("patch false then true", func(t *testing.T) { t.Parallel() srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusOK, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusOK, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), jsonhttptest.WithExpectedJSONResponse(api.RedistributionToggleResponse{Enabled: false}), ) @@ -135,7 +229,7 @@ func TestRedistributionToggle(t *testing.T) { t.Fatal("expected redistribution to be disabled") } - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusOK, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusOK, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": true}), jsonhttptest.WithExpectedJSONResponse(api.RedistributionToggleResponse{Enabled: true}), ) @@ -152,7 +246,7 @@ func TestRedistributionToggle(t *testing.T) { t.Parallel() srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusBadRequest, jsonhttptest.WithJSONRequestBody(map[string]any{}), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ Message: "enabled is required", @@ -165,7 +259,7 @@ func TestRedistributionToggle(t *testing.T) { t.Parallel() srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusBadRequest, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": nil}), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ Message: "enabled is required", @@ -178,7 +272,7 @@ func TestRedistributionToggle(t *testing.T) { t.Parallel() srv, _, _, _ := newTestServer(t, redistributionTestOpts(t)) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusBadRequest, jsonhttptest.WithRequestHeader(api.ContentTypeHeader, "application/json"), jsonhttptest.WithRequestBody(bytes.NewReader([]byte("{invalid"))), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ @@ -195,7 +289,7 @@ func TestRedistributionToggle(t *testing.T) { BeeMode: api.LightMode, StateStorer: statestore.NewStateStore(), }) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusBadRequest, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusBadRequest, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ Message: api.ErrOperationSupportedOnlyInFullMode.Error(), @@ -210,7 +304,7 @@ func TestRedistributionToggle(t *testing.T) { srv, _, _, _ := newTestServer(t, testServerOptions{ RedistributionAgentDisabled: true, }) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusForbidden, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusForbidden, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ Message: "Storage incentives are disabled. This endpoint is unavailable.", @@ -225,7 +319,7 @@ func TestRedistributionToggle(t *testing.T) { srv, _, _, _ := newTestServer(t, testServerOptions{ FullAPIDisabled: true, }) - jsonhttptest.Request(t, srv, http.MethodPut, "/redistribution", http.StatusServiceUnavailable, + jsonhttptest.Request(t, srv, http.MethodPatch, "/redistributionstate", http.StatusServiceUnavailable, jsonhttptest.WithJSONRequestBody(map[string]any{"enabled": false}), jsonhttptest.WithExpectedJSONResponse(jsonhttp.StatusResponse{ Message: "Node is syncing. This endpoint is unavailable. Try again later.", diff --git a/pkg/api/router.go b/pkg/api/router.go index cb9cdd3ae25..de65efabc2c 100644 --- a/pkg/api/router.go +++ b/pkg/api/router.go @@ -676,14 +676,8 @@ func (s *Service) mountBusinessDebug() { handle("/redistributionstate", web.ChainHandlers( s.checkStorageIncentivesAvailability, web.FinalHandler(jsonhttp.MethodHandler{ - "GET": http.HandlerFunc(s.redistributionStatusHandler), - })), - ) - - handle("/redistribution", web.ChainHandlers( - s.checkStorageIncentivesAvailability, - web.FinalHandler(jsonhttp.MethodHandler{ - "PUT": http.HandlerFunc(s.redistributionToggleHandler), + "GET": http.HandlerFunc(s.redistributionStatusHandler), + "PATCH": http.HandlerFunc(s.redistributionToggleHandler), })), ) diff --git a/pkg/api/router_test.go b/pkg/api/router_test.go index a53596898f2..4aaf1388aa2 100644 --- a/pkg/api/router_test.go +++ b/pkg/api/router_test.go @@ -113,8 +113,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/withdrawable", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, - {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, - {"/redistribution", []string{"PUT"}, http.StatusNoContent}, + {"/redistributionstate", []string{"GET", "PATCH"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, @@ -210,7 +209,6 @@ func TestEndpointOptions(t *testing.T) { {"/stake/{amount}", nil, http.StatusServiceUnavailable}, {"/stake", nil, http.StatusServiceUnavailable}, {"/redistributionstate", nil, http.StatusServiceUnavailable}, - {"/redistribution", nil, http.StatusServiceUnavailable}, {"/status", nil, http.StatusServiceUnavailable}, {"/status/peers", nil, http.StatusServiceUnavailable}, {"/status/neighborhoods", nil, http.StatusServiceUnavailable}, @@ -305,8 +303,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/withdrawable", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, - {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, - {"/redistribution", []string{"PUT"}, http.StatusNoContent}, + {"/redistributionstate", []string{"GET", "PATCH"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, @@ -401,8 +398,7 @@ func TestEndpointOptions(t *testing.T) { {"/stake/withdrawable", []string{"GET", "DELETE"}, http.StatusNoContent}, {"/stake/{amount}", []string{"POST"}, http.StatusNoContent}, {"/stake", []string{"GET", "DELETE"}, http.StatusNoContent}, - {"/redistributionstate", []string{"GET"}, http.StatusNoContent}, - {"/redistribution", []string{"PUT"}, http.StatusNoContent}, + {"/redistributionstate", []string{"GET", "PATCH"}, http.StatusNoContent}, {"/status", []string{"GET"}, http.StatusNoContent}, {"/status/peers", []string{"GET"}, http.StatusNoContent}, {"/status/neighborhoods", []string{"GET"}, http.StatusNoContent}, diff --git a/pkg/node/node.go b/pkg/node/node.go index c05517ad5be..a83d9710280 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -1325,6 +1325,12 @@ func NewBee( isFullySynced := func() bool { reserveThreshold := reserveCapacity * 5 / 10 + // Local beekeeper clusters use networkID 0 and never fill half of + // DefaultReserveCapacity (~2M). Lower the bar so storage incentives + // can be exercised locally without uploading millions of chunks. + if networkID == 0 { + reserveThreshold = 1 + } logger.Debug("Sync status check evaluated", "stabilized", detector.IsStabilized()) return localStore.ReserveSize() >= reserveThreshold && pullerService.SyncRate() == 0 && detector.IsStabilized() } diff --git a/pkg/storageincentives/agent.go b/pkg/storageincentives/agent.go index 7f464f63e92..1473de8698e 100644 --- a/pkg/storageincentives/agent.go +++ b/pkg/storageincentives/agent.go @@ -118,6 +118,7 @@ func New(overlay swarm.Address, } a.state = state + a.metrics.Enabled.Set(1) a.wg.Add(1) go a.start(blockTime, a.blocksPerRound, blocksPerPhase) @@ -594,6 +595,11 @@ func (a *Agent) Status() (*Status, error) { // that already has a commit key. func (a *Agent) SetEnabled(enabled bool) { a.disabled.Store(!enabled) + if enabled { + a.metrics.Enabled.Set(1) + } else { + a.metrics.Enabled.Set(0) + } } func (a *Agent) IsEnabled() bool { diff --git a/pkg/storageincentives/metrics.go b/pkg/storageincentives/metrics.go index b376d9d20b2..77a6153c4eb 100644 --- a/pkg/storageincentives/metrics.go +++ b/pkg/storageincentives/metrics.go @@ -19,6 +19,7 @@ type metrics struct { NeighborhoodSelected prometheus.Counter SampleDuration prometheus.Gauge Round prometheus.Gauge + Enabled prometheus.Gauge InsufficientFundsToPlay prometheus.Counter // total calls to chain backend @@ -91,6 +92,12 @@ func newMetrics() metrics { Name: "round", Help: "Current round calculated from the block height.", }), + Enabled: prometheus.NewGauge(prometheus.GaugeOpts{ + Namespace: m.Namespace, + Subsystem: subsystem, + Name: "redistribution_enabled", + Help: "Whether the node will commit in new redistribution rounds (1 enabled, 0 disabled).", + }), // total call BackendCalls: prometheus.NewCounter(prometheus.CounterOpts{ From 6c4aefda549fa70bb7965ff3986d2c4325fda790 Mon Sep 17 00:00:00 2001 From: sbackend Date: Thu, 10 Sep 2026 14:28:09 +0200 Subject: [PATCH 5/7] fix: revert local cluster changes --- pkg/node/node.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/pkg/node/node.go b/pkg/node/node.go index a83d9710280..c05517ad5be 100644 --- a/pkg/node/node.go +++ b/pkg/node/node.go @@ -1325,12 +1325,6 @@ func NewBee( isFullySynced := func() bool { reserveThreshold := reserveCapacity * 5 / 10 - // Local beekeeper clusters use networkID 0 and never fill half of - // DefaultReserveCapacity (~2M). Lower the bar so storage incentives - // can be exercised locally without uploading millions of chunks. - if networkID == 0 { - reserveThreshold = 1 - } logger.Debug("Sync status check evaluated", "stabilized", detector.IsStabilized()) return localStore.ReserveSize() >= reserveThreshold && pullerService.SyncRate() == 0 && detector.IsStabilized() } From f97b6c21899e1d184bd75f7b5361ad54f3570f5f Mon Sep 17 00:00:00 2001 From: sbackend Date: Thu, 10 Sep 2026 15:17:14 +0200 Subject: [PATCH 6/7] fix: make linter happy --- pkg/api/api_test.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index 8f36dbd568b..58edd046f80 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -127,18 +127,18 @@ type testServerOptions struct { BatchStore postage.Storer SyncStatus func() (bool, error) - BackendOpts []backendmock.Option - Erc20Opts []erc20mock.Option - BeeMode api.BeeNodeMode - RedistributionAgent *storageincentives.Agent + BackendOpts []backendmock.Option + Erc20Opts []erc20mock.Option + BeeMode api.BeeNodeMode + RedistributionAgent *storageincentives.Agent RedistributionAgentDisabled bool - NodeStatus *status.Service - PinIntegrity api.PinIntegrity - WhitelistedAddr string - FullAPIDisabled bool - ChequebookDisabled bool - SwapDisabled bool - Erc20ServiceNil bool + NodeStatus *status.Service + PinIntegrity api.PinIntegrity + WhitelistedAddr string + FullAPIDisabled bool + ChequebookDisabled bool + SwapDisabled bool + Erc20ServiceNil bool // ServiceOut, when set, receives the constructed *api.Service so tests // can drive it directly (e.g. via a custom net.Listener) instead of // through the httptest.Server this function also sets up. From f4e3f2fa0e50115cf090249b933a5dc9fd0b0ebf Mon Sep 17 00:00:00 2001 From: sbackend Date: Sun, 13 Sep 2026 13:21:43 +0200 Subject: [PATCH 7/7] fix: revert changes from master --- pkg/api/api_test.go | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/pkg/api/api_test.go b/pkg/api/api_test.go index 0024e98804d..e6be53ea4e4 100644 --- a/pkg/api/api_test.go +++ b/pkg/api/api_test.go @@ -127,18 +127,18 @@ type testServerOptions struct { BatchStore postage.Storer SyncStatus func() (bool, error) - BackendOpts []backendmock.Option - Erc20Opts []erc20mock.Option - BeeMode api.BeeNodeMode - RedistributionAgent *storageincentives.Agent + BackendOpts []backendmock.Option + Erc20Opts []erc20mock.Option + BeeMode api.BeeNodeMode + RedistributionAgent *storageincentives.Agent RedistributionAgentDisabled bool - NodeStatus *status.Service - PinIntegrity api.PinIntegrity - WhitelistedAddr string - FullAPIDisabled bool - ChequebookDisabled bool - SwapDisabled bool - Erc20ServiceNil bool + NodeStatus *status.Service + PinIntegrity api.PinIntegrity + WhitelistedAddr string + FullAPIDisabled bool + ChequebookDisabled bool + SwapDisabled bool + Erc20ServiceNil bool } func newTestServer(t *testing.T, o testServerOptions) (*http.Client, *websocket.Conn, string, *chanStorer) {