go/worker/storage: Create new runtime storage worker

This worker will be responsible for orchestrating storage
operations for the given runtime. This includes:
1. Registering availability
2. Creating checkpoints if configured
3. Pruning state
4. Syncing state (internally init, checkpoint sync and diff sync)

Subsequent commit will do the outlined refactor above.

Author: martintomazic

go/worker/storage/runtime_worker.go

@@ -0,0 +1,96 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/eapache/channels"
+	"golang.org/x/sync/errgroup"
+
+	"github.com/oasisprotocol/oasis-core/go/common/logging"
+	runtimeAPI "github.com/oasisprotocol/oasis-core/go/runtime/api"
+	"github.com/oasisprotocol/oasis-core/go/runtime/host"
+	"github.com/oasisprotocol/oasis-core/go/storage/api"
+	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/statesync"
+)
+
+// Worker is handling storage operations for a single runtime.
+type worker struct {
+	logger         *logging.Logger
+	stateSync      *statesync.Worker
+	stateSyncBlkCh *channels.InfiniteChannel
+}
+
+func newRuntimeWorker(
+	commonNode *committeeCommon.Node,
+	rp registration.RoleProvider,
+	rpRPC registration.RoleProvider,
+	localStorage api.LocalBackend,
+	checkpointerCfg *statesync.CheckpointSyncConfig,
+) (*worker, error) {
+	worker := &worker{
+		logger:         logging.GetLogger("worker/storage").With("runtimeID", commonNode.Runtime.ID()),
+		stateSyncBlkCh: channels.NewInfiniteChannel(),
+	}
+
+	stateSync, err := statesync.New(
+		commonNode,
+		rp,
+		rpRPC,
+		localStorage,
+		worker.stateSyncBlkCh,
+		checkpointerCfg,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create state sync worker: %w", err)
+	}
+
+	worker.stateSync = stateSync
+
+	return worker, nil
+}
+
+// NodeHooks implementation.
+
+// HandleNewBlockEarlyLocked is guarded by CrossNode.
+func (w *worker) HandleNewBlockEarlyLocked(*runtimeAPI.BlockInfo) {
+	// Nothing to do here.
+}
+
+// HandleNewBlockLocked is guarded by CrossNode.
+func (w *worker) HandleNewBlockLocked(bi *runtimeAPI.BlockInfo) {
+	// Notify the state syncer that there is a new block.
+	w.stateSyncBlkCh.In() <- bi.RuntimeBlock
+}
+
+// HandleRuntimeHostEventLocked is guarded by CrossNode.
+func (w *worker) HandleRuntimeHostEventLocked(*host.Event) {
+	// Nothing to do here.
+}
+
+// Initialized returns a channel that will be closed once the worker finished starting up.
+func (w *worker) Initialized() <-chan struct{} {
+	return w.stateSync.Initialized()
+}
+
+func (w *worker) GetStatus(ctx context.Context) (*storageAPI.Status, error) {
+	return w.stateSync.GetStatus(ctx)
+}
+
+func (w *worker) PauseCheckpointer(pause bool) error {
+	return w.stateSync.PauseCheckpointer(pause)
+}
+
+func (w *worker) serve(ctx context.Context) error {
+	w.logger.Info("started")
+	defer w.logger.Info("stopped")
+
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		return w.stateSync.Serve(ctx)
+	})
+	return g.Wait()
+}

go/worker/storage/service_internal.go

@@ -14,7 +14,7 @@ func (w *Worker) GetLastSyncedRound(_ context.Context, request *api.GetLastSynce
 		return nil, api.ErrRuntimeNotFound
 	}
 
-	round, ioRoot, stateRoot := node.GetLastSynced()
+	round, ioRoot, stateRoot := node.stateSync.GetLastSynced()
 	return &api.GetLastSyncedRoundResponse{
 		Round:     round,
 		IORoot:    ioRoot,

go/worker/storage/statesync/state_sync.go

@@ -24,12 +24,9 @@ import (
 	registryApi "github.com/oasisprotocol/oasis-core/go/registry/api"
 	roothashApi "github.com/oasisprotocol/oasis-core/go/roothash/api"
 	"github.com/oasisprotocol/oasis-core/go/roothash/api/block"
-	runtime "github.com/oasisprotocol/oasis-core/go/runtime/api"
-	"github.com/oasisprotocol/oasis-core/go/runtime/host"
 	storageApi "github.com/oasisprotocol/oasis-core/go/storage/api"
 	"github.com/oasisprotocol/oasis-core/go/storage/mkvs/checkpoint"
 	dbApi "github.com/oasisprotocol/oasis-core/go/storage/mkvs/db/api"
-	workerCommon "github.com/oasisprotocol/oasis-core/go/worker/common"
 	"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"
@@ -40,8 +37,6 @@ import (
 )
 
 var (
-	_ committee.NodeHooks = (*Worker)(nil)
-
 	// ErrNonLocalBackend is the error returned when the storage backend doesn't implement the LocalBackend interface.
 	ErrNonLocalBackend = errors.New("storage: storage backend doesn't support local storage")
 )
@@ -150,8 +145,6 @@ type Worker struct {
 
 	undefinedRound uint64
 
-	workerCommonCfg workerCommon.Config
-
 	checkpointer         checkpoint.Checkpointer
 	checkpointSyncCfg    *CheckpointSyncConfig
 	checkpointSyncForced bool
@@ -174,8 +167,8 @@ func New(
 	commonNode *committee.Node,
 	roleProvider registration.RoleProvider,
 	rpcRoleProvider registration.RoleProvider,
-	workerCommonCfg workerCommon.Config,
 	localStorage storageApi.LocalBackend,
+	blockCh *channels.InfiniteChannel,
 	checkpointSyncCfg *CheckpointSyncConfig,
 ) (*Worker, error) {
 	initMetrics()
@@ -188,15 +181,13 @@ func New(
 
 		logger: logging.GetLogger("worker/storage/statesync").With("runtime_id", commonNode.Runtime.ID()),
 
-		workerCommonCfg: workerCommonCfg,
-
 		localStorage: localStorage,
 
 		checkpointSyncCfg: checkpointSyncCfg,
 
 		status: api.StatusInitializing,
 
-		blockCh:    channels.NewInfiniteChannel(),
+		blockCh:    blockCh,
 		diffCh:     make(chan *fetchedDiff),
 		finalizeCh: make(chan finalizeResult),
 
@@ -327,24 +318,6 @@ func (w *Worker) GetLocalStorage() storageApi.LocalBackend {
 	return w.localStorage
 }
 
-// NodeHooks implementation.
-
-// HandleNewBlockEarlyLocked is guarded by CrossNode.
-func (w *Worker) HandleNewBlockEarlyLocked(*runtime.BlockInfo) {
-	// Nothing to do here.
-}
-
-// HandleNewBlockLocked is guarded by CrossNode.
-func (w *Worker) HandleNewBlockLocked(bi *runtime.BlockInfo) {
-	// Notify the state syncer that there is a new block.
-	w.blockCh.In() <- bi.RuntimeBlock
-}
-
-// HandleRuntimeHostEventLocked is guarded by CrossNode.
-func (w *Worker) HandleRuntimeHostEventLocked(*host.Event) {
-	// Nothing to do here.
-}
-
 // Watcher implementation.
 
 // GetLastSynced returns the height, IORoot hash and StateRoot hash of the last block that was fully synced to.

go/worker/storage/worker.go

@@ -29,7 +29,7 @@ type Worker struct {
 	initCh chan struct{}
 	quitCh chan struct{}
 
-	runtimes map[common.Namespace]*statesync.Worker
+	runtimes map[common.Namespace]*worker
 
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -51,7 +51,7 @@ func New(
 		logger:       logging.GetLogger("worker/storage"),
 		initCh:       make(chan struct{}),
 		quitCh:       make(chan struct{}),
-		runtimes:     make(map[common.Namespace]*statesync.Worker),
+		runtimes:     make(map[common.Namespace]*worker),
 		ctx:          ctx,
 		cancel:       cancel,
 	}
@@ -99,11 +99,10 @@ func (w *Worker) registerRuntime(commonNode *committeeCommon.Node) error {
 		return fmt.Errorf("can't create local storage backend: %w", err)
 	}
 
-	worker, err := statesync.New(
+	worker, err := newRuntimeWorker(
 		commonNode,
 		rp,
 		rpRPC,
-		w.commonWorker.GetConfig(),
 		localStorage,
 		&statesync.CheckpointSyncConfig{
 			Disabled:          config.GlobalConfig.Storage.CheckpointSyncDisabled,
@@ -191,7 +190,7 @@ func (w *Worker) serve(ctx context.Context) error {
 	g, ctx := errgroup.WithContext(ctx)
 	for id, r := range w.runtimes {
 		g.Go(func() error {
-			err := r.Serve(ctx)
+			err := r.serve(ctx)
 			if err != nil {
 				return fmt.Errorf("state sync worker failed (runtimeID: %s): %w", id, err)
 			}
@@ -222,9 +221,11 @@ func (w *Worker) Quit() <-chan struct{} {
 func (w *Worker) Cleanup() {
 }
 
-// GetRuntime returns a storage committee node for the given runtime (if available).
+// GetRuntime returns the state sync for the given runtime (if available).
 //
 // In case the runtime with the specified id was not configured for this node it returns nil.
+//
+// Sugggestion: This is only used to get status, how about making this GetRuntimeStatus?
 func (w *Worker) GetRuntime(id common.Namespace) *statesync.Worker {
-	return w.runtimes[id]
+	return w.runtimes[id].stateSync
 }