go/worker/storage: Move availability nudger to separate worker

Same issues as with checkpointer. Possibly we would want
to accept WatchFinalizedRounds as part of an interface.

Co-authored-by: Peter Nose <peter.nose@gmail.com>

Author: martintomazic

go/worker/storage/availabilitynudger/availability.go

@@ -0,0 +1,144 @@
+// Package availabilitynudger defines logic for updating the role providers.
+package availabilitynudger
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/eapache/channels"
+
+	"github.com/oasisprotocol/oasis-core/go/common"
+	"github.com/oasisprotocol/oasis-core/go/common/logging"
+	"github.com/oasisprotocol/oasis-core/go/common/node"
+	"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
+	"github.com/oasisprotocol/oasis-core/go/worker/registration"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/statesync"
+)
+
+const (
+	// The maximum number of rounds the worker can be behind the chain before it's sensible for
+	// it to register as available.
+	maximumRoundDelayForAvailability = uint64(10)
+
+	// The minimum number of rounds the worker can be behind the chain before it's sensible for
+	// it to stop advertising availability.
+	minimumRoundDelayForUnavailability = uint64(15)
+)
+
+// Worker tracks the progress of last and last synced rounds
+// and “nudges” role providers to mark themselves available or unavailable
+// based on how closely the node is keeping up with consensus.
+type Worker struct {
+	roleProvider    registration.RoleProvider
+	rpcRoleProvider registration.RoleProvider
+	roleAvailable   bool
+
+	lastRound       uint64
+	lastSyncedRound uint64
+
+	blockCh   *channels.InfiniteChannel
+	stateSync *statesync.Worker
+
+	logger *logging.Logger
+}
+
+// New creates a new worker that updates the availability to role providers.
+func New(localProvider, rpcProvider registration.RoleProvider, blockCh *channels.InfiniteChannel, stateSync *statesync.Worker, runtimeID common.Namespace) *Worker {
+	return &Worker{
+		roleProvider:    localProvider,
+		rpcRoleProvider: rpcProvider,
+		lastRound:       math.MaxUint64,
+		lastSyncedRound: math.MaxUint64,
+		blockCh:         blockCh,
+		stateSync:       stateSync,
+		logger:          logging.GetLogger("worker/storage/availabilitynudger").With("runtime_id", runtimeID),
+	}
+}
+
+// Serve starts the worker.
+func (w *Worker) Serve(ctx context.Context) error {
+	w.logger.Info("started")
+	defer w.logger.Info("stopped")
+
+	finalizeCh, sub, err := w.stateSync.WatchFinalizedRounds()
+	if err != nil {
+		return fmt.Errorf("failed to watch finalized rounds: %w", err)
+	}
+	defer sub.Close()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case inBlk := <-w.blockCh.Out():
+			blk := inBlk.(*block.Block)
+			w.setLastRound(blk.Header.Round)
+		case round := <-finalizeCh:
+			w.setLastSyncedRound(round)
+		}
+		w.updateAvailability()
+	}
+}
+
+// setLastRound updates the last round number.
+func (w *Worker) setLastRound(round uint64) {
+	w.lastRound = round
+}
+
+// setLastSyncedRound updates the most recently synced round number.
+func (w *Worker) setLastSyncedRound(round uint64) {
+	w.lastSyncedRound = round
+}
+
+// updateAvailability updates the role's availability based on the gap
+// between the last round and the last synced round.
+func (w *Worker) updateAvailability() {
+	if w.lastRound == math.MaxUint64 || w.lastSyncedRound == math.MaxUint64 {
+		return
+	}
+	// if w.lastRound > w.lastSyncedRound {
+	// 	return
+	// } not sure what was intent here given this we are looking for the gap.
+
+	switch roundLag := w.lastRound - w.lastSyncedRound; {
+	case roundLag < maximumRoundDelayForAvailability:
+		w.markAvailable()
+	case roundLag > minimumRoundDelayForUnavailability:
+		w.markUnavailable()
+	}
+}
+
+// markAvailable sets the role as available if it is not already.
+func (w *Worker) markAvailable() {
+	if w.roleAvailable {
+		return
+	}
+	w.roleAvailable = true
+
+	w.logger.Info("marking as available")
+
+	if w.roleProvider != nil {
+		w.roleProvider.SetAvailable(func(*node.Node) error { return nil })
+	}
+	if w.rpcRoleProvider != nil {
+		w.rpcRoleProvider.SetAvailable(func(*node.Node) error { return nil })
+	}
+}
+
+// markUnavailable sets the role as unavailable if it is currently available.
+func (w *Worker) markUnavailable() {
+	if !w.roleAvailable {
+		return
+	}
+	w.roleAvailable = false
+
+	w.logger.Info("marking as unavailable")
+
+	if w.roleProvider != nil {
+		w.roleProvider.SetUnavailable()
+	}
+	if w.rpcRoleProvider != nil {
+		w.rpcRoleProvider.SetUnavailable()
+	}
+}

go/worker/storage/runtime_worker.go

@@ -15,16 +15,20 @@ import (
 	committeeCommon "github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 	"github.com/oasisprotocol/oasis-core/go/worker/registration"
 	storageAPI "github.com/oasisprotocol/oasis-core/go/worker/storage/api"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/availabilitynudger"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/checkpointer"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/statesync"
 )
 
 // Worker is handling storage operations for a single runtime.
 type worker struct {
-	logger         *logging.Logger
-	stateSync      *statesync.Worker
-	checkpointer   *checkpointer.Worker
-	stateSyncBlkCh *channels.InfiniteChannel
+	commonNode         *committeeCommon.Node
+	logger             *logging.Logger
+	stateSync          *statesync.Worker
+	checkpointer       *checkpointer.Worker
+	availabilityNudger *availabilitynudger.Worker
+	stateSyncBlkCh     *channels.InfiniteChannel
+	availabilityBlkCh  *channels.InfiniteChannel
 }
 
 func newRuntimeWorker(
@@ -36,14 +40,14 @@ func newRuntimeWorker(
 	checkpointerEnabled bool,
 ) (*worker, error) {
 	worker := &worker{
-		logger:         logging.GetLogger("worker/storage").With("runtimeID", commonNode.Runtime.ID()),
-		stateSyncBlkCh: channels.NewInfiniteChannel(),
+		commonNode:        commonNode,
+		logger:            logging.GetLogger("worker/storage").With("runtimeID", commonNode.Runtime.ID()),
+		stateSyncBlkCh:    channels.NewInfiniteChannel(),
+		availabilityBlkCh: channels.NewInfiniteChannel(),
 	}
 
 	stateSync, err := statesync.New(
 		commonNode,
-		rp,
-		rpRPC,
 		localStorage,
 		worker.stateSyncBlkCh,
 		checkpointSyncCfg,
@@ -64,6 +68,8 @@ func newRuntimeWorker(
 	}
 	worker.checkpointer = checkpointer
 
+	worker.availabilityNudger = availabilitynudger.New(rp, rpRPC, worker.availabilityBlkCh, stateSync, commonNode.Runtime.ID())
+
 	return worker, nil
 }
 
@@ -76,8 +82,9 @@ func (w *worker) HandleNewBlockEarlyLocked(*runtimeAPI.BlockInfo) {
 
 // HandleNewBlockLocked is guarded by CrossNode.
 func (w *worker) HandleNewBlockLocked(bi *runtimeAPI.BlockInfo) {
-	// Notify the state syncer that there is a new block.
+	// Notify the state syncer and availability nudger that there is a new block.
 	w.stateSyncBlkCh.In() <- bi.RuntimeBlock
+	w.availabilityBlkCh.In() <- bi.RuntimeBlock
 }
 
 // HandleRuntimeHostEventLocked is guarded by CrossNode.
@@ -117,7 +124,16 @@ func (w *worker) serve(ctx context.Context) error {
 
 	g, ctx := errgroup.WithContext(ctx)
 	g.Go(func() error {
-		return w.stateSync.Serve(ctx)
+		if err := w.stateSync.Serve(ctx); err != nil {
+			return fmt.Errorf("state sync worker failed: %w", err)
+		}
+		return nil
+	})
+	g.Go(func() error {
+		if err := w.availabilityNudger.Serve(ctx); err != nil {
+			return fmt.Errorf("availability nudger failed: %w", err)
+		}
+		return nil
 	})
 	return g.Wait()
 }

go/worker/storage/statesync/state_sync.go

@@ -14,7 +14,6 @@ import (
 	"github.com/eapache/channels"
 
 	"github.com/oasisprotocol/oasis-core/go/common/logging"
-	"github.com/oasisprotocol/oasis-core/go/common/node"
 	"github.com/oasisprotocol/oasis-core/go/common/pubsub"
 	"github.com/oasisprotocol/oasis-core/go/common/workerpool"
 	"github.com/oasisprotocol/oasis-core/go/config"
@@ -26,11 +25,9 @@ import (
 	storageApi "github.com/oasisprotocol/oasis-core/go/storage/api"
 	dbApi "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/api"
 	"github.com/oasisprotocol/oasis-core/go/worker/common/committee"
-	"github.com/oasisprotocol/oasis-core/go/worker/registration"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/api"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/checkpointsync"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/diffsync"
-	storagePub "github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/pub"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/synclegacy"
 )
 
@@ -45,14 +42,6 @@ const (
 
 	checkpointSyncRetryDelay = 10 * time.Second
 
-	// The maximum number of rounds the worker can be behind the chain before it's sensible for
-	// it to register as available.
-	maximumRoundDelayForAvailability = uint64(10)
-
-	// The minimum number of rounds the worker can be behind the chain before it's sensible for
-	// it to stop advertising availability.
-	minimumRoundDelayForUnavailability = uint64(15)
-
 	// maxInFlightRounds is the maximum number of rounds that should be fetched before waiting
 	// for them to be applied.
 	maxInFlightRounds = 100
@@ -114,17 +103,10 @@ type finalizeResult struct {
 // In addition this worker is responsible for:
 //  1. Initializing the runtime state, possibly using checkpoints (if configured).
 //  2. Pruning the state as specified by the configuration.
-//  3. Optionally creating runtime state checkpoints (used by other nodes) for the state sync.
-//  4. Creating (and optionally advertising) statesync p2p protocol clients and servers.
-//  5. Registering node availability when it has synced sufficiently close to
-//     the latest known block header.
+//  3. Creating (and optionally advertising) statesync p2p protocol clients and servers.
 type Worker struct {
 	commonNode *committee.Node
 
-	roleProvider    registration.RoleProvider
-	rpcRoleProvider registration.RoleProvider
-	roleAvailable   bool
-
 	logger *logging.Logger
 
 	localStorage storageApi.LocalBackend
@@ -156,8 +138,6 @@ type Worker struct {
 // New creates a new state sync worker.
 func New(
 	commonNode *committee.Node,
-	roleProvider registration.RoleProvider,
-	rpcRoleProvider registration.RoleProvider,
 	localStorage storageApi.LocalBackend,
 	blockCh *channels.InfiniteChannel,
 	checkpointSyncCfg *CheckpointSyncConfig,
@@ -167,9 +147,6 @@ func New(
 	w := &Worker{
 		commonNode: commonNode,
 
-		roleProvider:    roleProvider,
-		rpcRoleProvider: rpcRoleProvider,
-
 		logger: logging.GetLogger("worker/storage/statesync").With("runtime_id", commonNode.Runtime.ID()),
 
 		localStorage: localStorage,
@@ -202,9 +179,6 @@ func New(
 	if config.GlobalConfig.Storage.Checkpointer.Enabled {
 		commonNode.P2P.RegisterProtocolServer(checkpointsync.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
 	}
-	if rpcRoleProvider != nil {
-		commonNode.P2P.RegisterProtocolServer(storagePub.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
-	}
 
 	// Create p2p protocol clients.
 	w.legacyStorageSync = synclegacy.NewClient(commonNode.P2P, commonNode.ChainContext, commonNode.Runtime.ID())
@@ -493,31 +467,6 @@ func (w *Worker) flushSyncedState(summary *blockSummary) (uint64, error) {
 	return w.syncedState.Round, nil
 }
 
-// This is only called from the main worker goroutine, so no locking should be necessary.
-func (w *Worker) nudgeAvailability(lastSynced, latest uint64) {
-	if lastSynced == w.undefinedRound || latest == w.undefinedRound {
-		return
-	}
-	if latest-lastSynced < maximumRoundDelayForAvailability && !w.roleAvailable {
-		w.roleProvider.SetAvailable(func(_ *node.Node) error {
-			return nil
-		})
-		if w.rpcRoleProvider != nil {
-			w.rpcRoleProvider.SetAvailable(func(_ *node.Node) error {
-				return nil
-			})
-		}
-		w.roleAvailable = true
-	}
-	if latest-lastSynced > minimumRoundDelayForUnavailability && w.roleAvailable {
-		w.roleProvider.SetUnavailable()
-		if w.rpcRoleProvider != nil {
-			w.rpcRoleProvider.SetUnavailable()
-		}
-		w.roleAvailable = false
-	}
-}
-
 // Serve runs the state sync worker.
 func (w *Worker) Serve(ctx context.Context) error { // nolint: gocyclo
 	defer close(w.diffCh)
@@ -950,7 +899,6 @@ mainLoop:
 
 			// Check if we're far enough to reasonably register as available.
 			latestBlockRound = blk.Header.Round
-			w.nudgeAvailability(lastFinalizedRound, latestBlockRound)
 
 			if _, ok := summaryCache[lastFullyAppliedRound]; !ok && lastFullyAppliedRound == w.undefinedRound {
 				dummy := blockSummary{
@@ -1043,9 +991,6 @@ mainLoop:
 			}
 			storageWorkerLastFinalizedRound.With(w.getMetricLabels()).Set(float64(finalized.summary.Round))
 
-			// Check if we're far enough to reasonably register as available.
-			w.nudgeAvailability(lastFinalizedRound, latestBlockRound)
-
 		case <-ctx.Done():
 			err = ctx.Err()
 			break mainLoop

go/worker/storage/worker.go

@@ -15,6 +15,7 @@ import (
 	committeeCommon "github.com/oasisprotocol/oasis-core/go/worker/common/committee"
 	"github.com/oasisprotocol/oasis-core/go/worker/registration"
 	storageWorkerAPI "github.com/oasisprotocol/oasis-core/go/worker/storage/api"
+	"github.com/oasisprotocol/oasis-core/go/worker/storage/p2p/pub"
 	"github.com/oasisprotocol/oasis-core/go/worker/storage/statesync"
 )
 
@@ -79,6 +80,11 @@ func (w *Worker) registerRuntime(commonNode *committeeCommon.Node) error {
 		"runtime_id", id,
 	)
 
+	localStorage, err := NewLocalBackend(commonNode.Runtime.DataDir(), id)
+	if err != nil {
+		return fmt.Errorf("can't create local storage backend: %w", err)
+	}
+
 	// Since the storage node is always coupled with another role, make sure to not add any
 	// particular role here. Instead this only serves to prevent registration until the storage node
 	// is synced by making the role provider unavailable.
@@ -93,10 +99,8 @@ func (w *Worker) registerRuntime(commonNode *committeeCommon.Node) error {
 			return fmt.Errorf("failed to create rpc role provider: %w", err)
 		}
 	}
-
-	localStorage, err := NewLocalBackend(commonNode.Runtime.DataDir(), id)
-	if err != nil {
-		return fmt.Errorf("can't create local storage backend: %w", err)
+	if rpRPC != nil {
+		commonNode.P2P.RegisterProtocolServer(pub.NewServer(commonNode.ChainContext, commonNode.Runtime.ID(), localStorage))
 	}
 
 	worker, err := newRuntimeWorker(