feat: redistribution toggle

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,38 @@ 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.
+        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:
+        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

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. 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:
@@ -846,6 +849,21 @@ components:
         fees:
           $ref: "#/components/schemas/BigInt"
 
+    RedistributionEnableRequest:
+      type: object
+      required:
+        - enabled
+      properties:
+        enabled:
+          type: boolean
+          description: Whether the node should commit in new redistribution rounds. Sampling still runs while disabled.
+
+    RedistributionEnableResponse:
+      type: object
+      properties:
+        enabled:
+          type: boolean
+
     PendingTransactionsResponse:
       type: object
       properties:

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)

pkg/api/export_test.go

@@ -98,6 +98,8 @@ type (
 	StakeTransactionReponse           = stakeTransactionReponse
 	StatusSnapshotResponse            = statusSnapshotResponse
 	StatusResponse                    = statusResponse
+	RedistributionStatusResponse      = redistributionStatusResponse
+	RedistributionToggleResponse      = redistributionToggleResponse
 )
 
 var (

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

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) {
@@ -76,3 +83,154 @@ func TestRedistributionStatus(t *testing.T) {
 		)
 	})
 }
+
+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,
+			}),
+		)
+	})
+}