diff --git a/openapi/Swarm.yaml b/openapi/Swarm.yaml index c320410b4ee..52d3e2e5e6b 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,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 diff --git a/openapi/SwarmCommon.yaml b/openapi/SwarmCommon.yaml index ffcbbac3b8e..18be2d6e860 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. 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: 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..b7f83da0c23 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) { @@ -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, + }), + ) + }) +} 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..7f464f63e92 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 @@ -581,6 +589,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..8da336e992e 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 still samples but skips 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.Fatal("expected sampling to run while disabled") + } + 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)