VAULT-23553: Revert "Don't panic on unknown raft ops"

physical/raft/fsm.go

@@ -179,7 +179,6 @@ type FSM struct {
 
 	localID         string
 	desiredSuffrage string
-	unknownOpTypes  sync.Map
 }
 
 // NewFSM constructs a FSM using the given directory
@@ -678,24 +677,24 @@ func (f *FSM) ApplyBatch(logs []*raft.Log) []interface{} {
 	// Do the unmarshalling first so we don't hold locks
 	var latestConfiguration *ConfigurationValue
 	commands := make([]interface{}, 0, numLogs)
-	for _, l := range logs {
-		switch l.Type {
+	for _, log := range logs {
+		switch log.Type {
 		case raft.LogCommand:
 			command := &LogData{}
 
 			// explicitly check for zero length Data, which will be the case for verifier no-ops
-			if len(l.Data) > 0 {
-				err := proto.Unmarshal(l.Data, command)
+			if len(log.Data) > 0 {
+				err := proto.Unmarshal(log.Data, command)
 				if err != nil {
-					f.logger.Error("error proto unmarshaling log data", "error", err, "data", l.Data)
+					f.logger.Error("error proto unmarshaling log data", "error", err, "data", log.Data)
 					panic("error proto unmarshaling log data")
 				}
 			}
 
 			commands = append(commands, command)
 		case raft.LogConfiguration:
-			configuration := raft.DecodeConfiguration(l.Data)
-			config := raftConfigurationToProtoConfiguration(l.Index, configuration)
+			configuration := raft.DecodeConfiguration(log.Data)
+			config := raftConfigurationToProtoConfiguration(log.Index, configuration)
 
 			commands = append(commands, config)
 
@@ -704,7 +703,7 @@ func (f *FSM) ApplyBatch(logs []*raft.Log) []interface{} {
 			latestConfiguration = config
 
 		default:
-			panic(fmt.Sprintf("got unexpected log type: %d", l.Type))
+			panic(fmt.Sprintf("got unexpected log type: %d", log.Type))
 		}
 	}
 
@@ -763,10 +762,7 @@ func (f *FSM) ApplyBatch(logs []*raft.Log) []interface{} {
 							go f.restoreCb(context.Background())
 						}
 					default:
-						if _, ok := f.unknownOpTypes.Load(op.OpType); !ok {
-							f.logger.Error("unsupported transaction operation", "op", op.OpType)
-							f.unknownOpTypes.Store(op.OpType, struct{}{})
-						}
+						return fmt.Errorf("%q is not a supported transaction operation", op.OpType)
 					}
 					if err != nil {
 						return err

physical/raft/fsm_test.go

@@ -6,9 +6,7 @@ package raft
 import (
 	"context"
 	"fmt"
-	"io/ioutil"
 	"math/rand"
-	"os"
 	"sort"
 	"testing"
 
@@ -17,13 +15,11 @@ import (
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/raft"
 	"github.com/hashicorp/vault/sdk/physical"
+	"github.com/stretchr/testify/require"
 )
 
-func getFSM(t testing.TB) (*FSM, string) {
-	raftDir, err := ioutil.TempDir("", "vault-raft-")
-	if err != nil {
-		t.Fatal(err)
-	}
+func getFSM(t testing.TB) *FSM {
+	raftDir := t.TempDir()
 	t.Logf("raft dir: %s", raftDir)
 
 	logger := hclog.New(&hclog.LoggerOptions{
@@ -36,12 +32,11 @@ func getFSM(t testing.TB) (*FSM, string) {
 		t.Fatal(err)
 	}
 
-	return fsm, raftDir
+	return fsm
 }
 
 func TestFSM_Batching(t *testing.T) {
-	fsm, dir := getFSM(t)
-	defer func() { _ = os.RemoveAll(dir) }()
+	fsm := getFSM(t)
 
 	var index uint64
 	var term uint64 = 1
@@ -133,8 +128,7 @@ func TestFSM_Batching(t *testing.T) {
 }
 
 func TestFSM_List(t *testing.T) {
-	fsm, dir := getFSM(t)
-	defer func() { _ = os.RemoveAll(dir) }()
+	fsm := getFSM(t)
 
 	ctx := context.Background()
 	count := 100
@@ -162,3 +156,41 @@ func TestFSM_List(t *testing.T) {
 		t.Fatal(diff)
 	}
 }
+
+// TestFSM_UnknownOperation calls ApplyBatch with a batch that has an unknown
+// command operation type. The test verifies that the call panics
+func TestFSM_UnknownOperation(t *testing.T) {
+	fsm := getFSM(t)
+	command := &LogData{
+		Operations: make([]*LogOperation, 5),
+	}
+
+	for i := range command.Operations {
+		op := putOp
+		if i == 4 {
+			// the last operation has an invalid op type
+			op = 0
+		}
+		command.Operations[i] = &LogOperation{
+			OpType: op,
+			Key:    fmt.Sprintf("key-%d", i),
+			Value:  []byte(fmt.Sprintf("value-%d", i)),
+		}
+	}
+	commandBytes, err := proto.Marshal(command)
+	require.NoError(t, err)
+
+	defer func() {
+		r := recover()
+		require.NotNil(t, r)
+		require.Contains(t, r, "failed to store data")
+	}()
+	fsm.ApplyBatch([]*raft.Log{{
+		Index: 0,
+		Term:  1,
+		Type:  raft.LogCommand,
+		Data:  commandBytes,
+	}})
+
+	require.Fail(t, "failed to panic")
+}