Fix storage worker

.changelog/3820.bugfix.md

@@ -0,0 +1 @@
+go/worker/storage: Fix Finalize and Apply failure handling

.changelog/3820.internal.1.md

@@ -0,0 +1 @@
+go/worker/storage: Add independent heartbeat for syncing retries

.changelog/3820.internal.2.md

@@ -0,0 +1 @@
+go/storage/client: Implement node blacklisting

go/oasis-node/cmd/debug/byzantine/byzantine.go

@@ -279,7 +279,8 @@ func init() {
 
 	storageFlags.Uint64(CfgNumStorageFailApplyBatch, 0, "Number of ApplyBatch requests to fail")
 	storageFlags.Uint64(CfgNumStorageFailApply, 0, "Number of Apply requests to fail")
-	storageFlags.Bool(CfgFailReadRequests, false, "If storage worker should fail read requests")
+	storageFlags.Bool(CfgFailReadRequests, false, "Whether the storage node should fail read requests")
+	storageFlags.Bool(CfgCorruptGetDiff, false, "Whether the storage node should corrupt GetDiff responses")
 	_ = viper.BindPFlags(storageFlags)
 	byzantineCmd.PersistentFlags().AddFlagSet(storageFlags)
 

go/oasis-node/cmd/debug/byzantine/storage_node.go

@@ -26,8 +26,10 @@ const (
 	// CfgNumStorageFailApplyBatch configures how many apply-batch requests
 	// the storage node should fail.
 	CfgNumStorageFailApplyBatch = "num_storage_fail_apply_batch"
-	// CfgFailReadRequests configures if storage node should fail read requests.
+	// CfgFailReadRequests configures whether the storage node should fail read requests.
 	CfgFailReadRequests = "fail_read_requests"
+	// CfgCorruptGetDiff configures whether the storage node should corrupt GetDiff responses.
+	CfgCorruptGetDiff = "corrupt_get_diff"
 )
 
 var (
@@ -48,6 +50,7 @@ type storageWorker struct {
 	numFailApply      uint64
 	numFailApplyBatch uint64
 	failReadRequests  bool
+	corruptGetDiff    bool
 }
 
 func newStorageNode(id *identity.Identity, namespace common.Namespace, datadir string) (*storageWorker, error) {
@@ -74,6 +77,7 @@ func newStorageNode(id *identity.Identity, namespace common.Namespace, datadir s
 		numFailApply:      viper.GetUint64(CfgNumStorageFailApply),
 		numFailApplyBatch: viper.GetUint64(CfgNumStorageFailApplyBatch),
 		failReadRequests:  viper.GetBool(CfgFailReadRequests),
+		corruptGetDiff:    viper.GetBool(CfgCorruptGetDiff),
 	}, nil
 }
 
@@ -125,12 +129,46 @@ func (w *storageWorker) ApplyBatch(ctx context.Context, request *storage.ApplyBa
 	return w.backend.ApplyBatch(ctx, request)
 }
 
+type corruptIterator struct {
+	it        storage.WriteLogIterator
+	corrupted bool
+}
+
+// Implements storage.WriteLogIterator.
+func (ci *corruptIterator) Next() (bool, error) {
+	return ci.it.Next()
+}
+
+// Implements storage.WriteLogIterator.
+func (ci *corruptIterator) Value() (storage.LogEntry, error) {
+	v, err := ci.it.Value()
+	if err != nil {
+		return storage.LogEntry{}, err
+	}
+
+	// Corrupt the first entry.
+	if !ci.corrupted {
+		v.Value = []byte("corrupted")
+		ci.corrupted = true
+	}
+	return v, nil
+}
+
 func (w *storageWorker) GetDiff(ctx context.Context, request *storage.GetDiffRequest) (storage.WriteLogIterator, error) {
 	if w.failReadRequests {
 		return nil, errByzantine
 	}
 
-	return w.backend.GetDiff(ctx, request)
+	wl, err := w.backend.GetDiff(ctx, request)
+	if err != nil {
+		return nil, err
+	}
+
+	modifiedWl := wl
+	if w.corruptGetDiff {
+		modifiedWl = &corruptIterator{it: wl}
+	}
+	return modifiedWl, nil
 }
 
 func (w *storageWorker) GetCheckpoints(ctx context.Context, request *checkpoint.GetCheckpointsRequest) ([]*checkpoint.Metadata, error) {

go/oasis-node/cmd/node/node.go

@@ -309,6 +309,7 @@ func (n *Node) initRuntimeWorkers() error {
 
 	// Initialize the common worker.
 	n.CommonWorker, err = workerCommon.New(
+		n,
 		dataDir,
 		compute.Enabled() || workerStorage.Enabled() || workerKeymanager.Enabled(),
 		n.Identity,

go/oasis-test-runner/scenario/e2e/runtime/byzantine.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"time"
 
 	"github.com/oasisprotocol/oasis-core/go/common/quantity"
 	consensus "github.com/oasisprotocol/oasis-core/go/consensus/api"
@@ -14,7 +15,9 @@ import (
 	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/scenario"
 	"github.com/oasisprotocol/oasis-core/go/oasis-test-runner/scenario/e2e"
 	registry "github.com/oasisprotocol/oasis-core/go/registry/api"
+	runtimeClient "github.com/oasisprotocol/oasis-core/go/runtime/client/api"
 	staking "github.com/oasisprotocol/oasis-core/go/staking/api"
+	"github.com/oasisprotocol/oasis-core/go/storage/mkvs"
 )
 
 var (
@@ -38,6 +41,7 @@ var (
 		"executor",
 		nil,
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		nil,
 	)
@@ -47,6 +51,7 @@ var (
 		"executor",
 		nil,
 		oasis.ByzantineSlot1IdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgSchedulerRoleExpected,
@@ -63,6 +68,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		// Byzantine node entity should be slashed once for submitting incorrect commitment.
 		map[staking.SlashReason]uint64{
 			staking.SlashRuntimeIncorrectResults: 1,
@@ -83,6 +89,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineSlot1IdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgSchedulerRoleExpected,
@@ -100,6 +107,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgExecutorMode, byzantine.ModeExecutorStraggler.String(),
@@ -117,6 +125,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineSlot1IdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgSchedulerRoleExpected,
@@ -135,6 +144,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgExecutorMode, byzantine.ModeExecutorFailureIndicating.String(),
@@ -152,6 +162,7 @@ var (
 			oasis.LogAssertExecutionDiscrepancyDetected(),
 		},
 		oasis.ByzantineSlot1IdentitySeed,
+		false,
 		nil,
 		[]string{
 			"--" + byzantine.CfgSchedulerRoleExpected,
@@ -164,6 +175,7 @@ var (
 		"storage",
 		nil,
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		nil,
 	)
@@ -176,6 +188,7 @@ var (
 		// should keep retrying proposing a batch until it succeeds.
 		nil,
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		[]string{
 			// Fail first 5 ApplyBatch requests.
@@ -194,6 +207,7 @@ var (
 			oasis.LogAssertRoundFailures(),
 		},
 		oasis.ByzantineDefaultIdentitySeed,
+		false,
 		nil,
 		[]string{
 			// Fail first 3 ApplyBatch requests - from the 2 executor workers and 1 backup node.
@@ -207,12 +221,33 @@ var (
 		// There should be no discrepancy or round failures.
 		nil,
 		oasis.ByzantineDefaultIdentitySeed,
+		// Hack to work around the way the storage client selects nodes.
+		// It can happen that for this test, the client will keep selecting the byzantine
+		// node instead of the other storage nodes and so would never be able to fully sync.
+		// It can take up to two minutes for storage-0 to sync in this scenario with the
+		// current shuffling method in the storage client. See also #1815.
+		true,
 		nil,
 		[]string{
 			// Fail all read requests.
 			"--" + byzantine.CfgFailReadRequests,
 		},
 	)
+	// ByzantineStorageCorruptGetDiff is the byzantine storage node scenario that corrupts GetDiff
+	// responses.
+	ByzantineStorageCorruptGetDiff scenario.Scenario = newByzantineImpl(
+		"storage-corrupt-getdiff",
+		"storage",
+		// There should be no discrepancy or round failures.
+		nil,
+		oasis.ByzantineDefaultIdentitySeed,
+		false,
+		nil,
+		[]string{
+			// Corrupt all GetDiff responses.
+			"--" + byzantine.CfgCorruptGetDiff,
+		},
+	)
 )
 
 type byzantineImpl struct {
@@ -221,6 +256,7 @@ type byzantineImpl struct {
 	script    string
 	extraArgs []string
 
+	skipStorageSyncWait        bool
 	identitySeed               string
 	logWatcherHandlerFactories []log.WatcherHandlerFactory
 
@@ -235,6 +271,7 @@ func newByzantineImpl(
 	script string,
 	logWatcherHandlerFactories []log.WatcherHandlerFactory,
 	identitySeed string,
+	skipStorageWait bool,
 	expectedSlashes map[staking.SlashReason]uint64,
 	extraArgs []string,
 ) scenario.Scenario {
@@ -246,6 +283,7 @@ func newByzantineImpl(
 		),
 		script:                     script,
 		extraArgs:                  extraArgs,
+		skipStorageSyncWait:        skipStorageWait,
 		identitySeed:               identitySeed,
 		logWatcherHandlerFactories: logWatcherHandlerFactories,
 		expectedSlashes:            expectedSlashes,
@@ -257,6 +295,7 @@ func (sc *byzantineImpl) Clone() scenario.Scenario {
 		runtimeImpl:                *sc.runtimeImpl.Clone().(*runtimeImpl),
 		script:                     sc.script,
 		extraArgs:                  sc.extraArgs,
+		skipStorageSyncWait:        sc.skipStorageSyncWait,
 		identitySeed:               sc.identitySeed,
 		logWatcherHandlerFactories: sc.logWatcherHandlerFactories,
 		expectedSlashes:            sc.expectedSlashes,
@@ -372,5 +411,72 @@ func (sc *byzantineImpl) Run(childEnv *env.Env) error {
 		return fmt.Errorf("expected entity stake: %v got: %v", expectedStake, acc.General.Balance)
 	}
 
+	if sc.skipStorageSyncWait {
+		return nil
+	}
+
+	// Wait for all storage nodes to be synced.
+	blk, err := sc.Net.ClientController().RuntimeClient.GetBlock(ctx, &runtimeClient.GetBlockRequest{
+		RuntimeID: runtimeID,
+		Round:     runtimeClient.RoundLatest,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to fetch latest block: %w", err)
+	}
+
+	sc.Logger.Info("waiting for storage nodes to be synced",
+		"target_round", blk.Header.Round,
+	)
+
+	syncedNodes := make(map[string]bool)
+	storageCtx, cancelFn := context.WithTimeout(ctx, 60*time.Second)
+	defer cancelFn()
+StorageWorkerSyncLoop:
+	for {
+		if storageCtx.Err() != nil {
+			return fmt.Errorf("failed to wait for storage nodes to be synced: %w", storageCtx.Err())
+		}
+
+		for _, n := range sc.Net.StorageWorkers() {
+			if syncedNodes[n.Name] {
+				continue
+			}
+
+			ctrl, err := oasis.NewController(n.SocketPath())
+			if err != nil {
+				return fmt.Errorf("failed to create storage node controller: %w", err)
+			}
+
+			// Iterate over the roots to confirm they have been synced.
+			for _, root := range blk.Header.StorageRoots() {
+				state := mkvs.NewWithRoot(ctrl.Storage, nil, root)
+				it := state.NewIterator(storageCtx)
+
+				for it.Rewind(); it.Valid(); it.Next() {
+				}
+				err = it.Err()
+				it.Close()
+				state.Close()
+
+				if err != nil {
+					// Failed to iterate over the root.
+					sc.Logger.Warn("storage node is still not synced",
+						"node", n.Name,
+						"root", root,
+						"err", err,
+					)
+					time.Sleep(1 * time.Second)
+					continue StorageWorkerSyncLoop
+				}
+			}
+
+			sc.Logger.Warn("storage node is synced",
+				"node", n.Name,
+			)
+			syncedNodes[n.Name] = true
+		}
+		break
+	}
+
 	return nil
 }

go/oasis-test-runner/scenario/e2e/runtime/runtime.go

@@ -580,6 +580,7 @@ func RegisterScenarios() error {
 		ByzantineStorageFailApply,
 		ByzantineStorageFailApplyBatch,
 		ByzantineStorageFailRead,
+		ByzantineStorageCorruptGetDiff,
 		// Storage sync test.
 		StorageSync,
 		StorageSyncFromRegistered,