From 77efd9a560dc9e94573826ce3649a48ba852f678 Mon Sep 17 00:00:00 2001 From: Paul Lorenz Date: Thu, 25 Aug 2022 09:56:28 -0400 Subject: [PATCH] Wait for changes request from follow to be applied to follower before returning. Fixes openziti/fabric#481 --- controller/raft/fsm.go | 83 ++++++++------ controller/raft/handlers.go | 34 +++++- controller/raft/hclog_adapter.go | 16 +++ controller/raft/index_tracker.go | 82 ++++++++++++++ controller/raft/index_tracker_test.go | 84 ++++++++++++++ controller/raft/member.go | 16 +++ controller/raft/mesh/mesh.go | 18 ++- controller/raft/mesh/peerconn.go | 16 +++ controller/raft/mesh/timeouts.go | 16 +++ controller/raft/raft.go | 151 +++++++++++++++++++++++--- 10 files changed, 457 insertions(+), 59 deletions(-) create mode 100644 controller/raft/index_tracker.go create mode 100644 controller/raft/index_tracker_test.go diff --git a/controller/raft/fsm.go b/controller/raft/fsm.go index bee6090b7..2dc60bb17 100644 --- a/controller/raft/fsm.go +++ b/controller/raft/fsm.go @@ -1,35 +1,52 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package raft import ( "github.com/hashicorp/raft" + "github.com/michaelquigley/pfxlog" "github.com/openziti/fabric/controller/command" "github.com/openziti/fabric/controller/db" "github.com/openziti/foundation/v2/concurrenz" "github.com/openziti/storage/boltz" "github.com/pkg/errors" "github.com/sirupsen/logrus" - "go.etcd.io/bbolt" "io" - "io/ioutil" "os" "sync/atomic" ) -func NewFsm(dataDir string, decoders command.Decoders) *BoltDbFsm { +func NewFsm(dataDir string, decoders command.Decoders, indexTracker IndexTracker) *BoltDbFsm { return &BoltDbFsm{ - dataDir: dataDir, - decoders: decoders, - dbPath: dataDir + "ctrl.db", + dataDir: dataDir, + decoders: decoders, + dbPath: dataDir + "ctrl.db", + indexTracker: indexTracker, } } type BoltDbFsm struct { - db boltz.Db - dataDir string - dbPath string - decoders command.Decoders - env atomic.Value - initialized concurrenz.AtomicBoolean + db boltz.Db + dataDir string + dbPath string + decoders command.Decoders + env atomic.Value + initialized concurrenz.AtomicBoolean + indexTracker IndexTracker } func (self *BoltDbFsm) Init() error { @@ -51,18 +68,21 @@ func (self *BoltDbFsm) GetDb() boltz.Db { } func (self *BoltDbFsm) Apply(log *raft.Log) interface{} { + logger := pfxlog.Logger() if log.Type == raft.LogCommand { + defer self.indexTracker.NotifyOfIndex(log.Index) + if len(log.Data) >= 4 { cmd, err := self.decoders.Decode(log.Data) if err != nil { - logrus.WithError(err).Error("failed to create command") + logger.WithError(err).Error("failed to create command") return err } - logrus.Infof("apply log with type %T", cmd) + logger.Infof("[%v] apply log with type %T", log.Index, cmd) if err = cmd.Apply(); err != nil { - logrus.WithError(err).Error("applying log resulted in error") + logger.WithError(err).Error("applying log resulted in error") } return err @@ -74,23 +94,19 @@ func (self *BoltDbFsm) Apply(log *raft.Log) interface{} { } func (self *BoltDbFsm) Snapshot() (raft.FSMSnapshot, error) { - logrus.Info("creating snapshot") + logrus.Debug("creating snapshot") - file, err := ioutil.TempFile(self.dataDir, "raft-*.snapshot.db") + id, data, err := self.db.SnapshotToMemory() if err != nil { return nil, err } - err = self.db.View(func(tx *bbolt.Tx) error { - _, err = tx.WriteTo(file) - return err - }) + logrus.WithField("id", id).WithField("index", self.indexTracker.Index()).Info("creating snapshot") - if err != nil { - return nil, err - } - - return &boltSnapshot{path: file.Name()}, nil + return &boltSnapshot{ + snapshotId: id, + snapshotData: data, + }, nil } func (self *BoltDbFsm) Restore(snapshot io.ReadCloser) error { @@ -99,6 +115,7 @@ func (self *BoltDbFsm) Restore(snapshot io.ReadCloser) error { return err } } + logrus.Info("restoring from snapshot") backup := self.dbPath + ".backup" @@ -126,21 +143,15 @@ func (self *BoltDbFsm) Restore(snapshot io.ReadCloser) error { } type boltSnapshot struct { - path string + snapshotId string + snapshotData []byte } func (self *boltSnapshot) Persist(sink raft.SnapshotSink) error { - file, err := os.Open(self.path) - if err != nil { - return err - } - defer func() { _ = file.Close() }() - _, err = io.Copy(sink, file) + _, err := sink.Write(self.snapshotData) return err } func (self *boltSnapshot) Release() { - if err := os.Remove(self.path); err != nil { - logrus.WithError(err).WithField("path", self.path).Error("failed to remove bolt snapshot") - } + self.snapshotData = nil } diff --git a/controller/raft/handlers.go b/controller/raft/handlers.go index bad5038cb..e6dbd2ddf 100644 --- a/controller/raft/handlers.go +++ b/controller/raft/handlers.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package raft import ( @@ -24,6 +40,7 @@ const ( RemoveRequestType = 2054 HeaderErrorCode = 1000 + IndexHeader = 1001 ErrorCodeBadMessage = 1 ErrorCodeNotLeader = 2 @@ -152,11 +169,12 @@ func (self *commandHandler) ContentType() int32 { func (self *commandHandler) HandleReceive(m *channel.Message, ch channel.Channel) { err := self.pool.Queue(func() { - if err := self.controller.ApplyEncodedCommand(m.Body); err != nil { + if idx, err := self.controller.ApplyEncodedCommand(m.Body); err != nil { sendErrorResponseCalculateType(m, ch, err) return + } else { + sendSuccessResponse(m, ch, idx) } - sendSuccessResponse(m, ch) }) if err != nil { @@ -183,7 +201,13 @@ func sendErrorResponse(m *channel.Message, ch channel.Channel, err error, errorC } func sendApiErrorResponse(m *channel.Message, ch channel.Channel, err *errorz.ApiError) { - buf, encodeErr := json.Marshal(err) + encodingMap := map[string]interface{}{} + encodingMap["code"] = err.Code + encodingMap["message"] = err.Message + encodingMap["status"] = err.Status + encodingMap["cause"] = err.Cause + + buf, encodeErr := json.Marshal(encodingMap) if encodeErr != nil { logrus.WithError(encodeErr).WithField("apiErr", err).Error("unable to encode api error") sendErrorResponse(m, ch, err, ErrorCodeGeneric) @@ -197,9 +221,11 @@ func sendApiErrorResponse(m *channel.Message, ch channel.Channel, err *errorz.Ap logrus.WithError(sendErr).Error("error while sending error response") } } -func sendSuccessResponse(m *channel.Message, ch channel.Channel) { + +func sendSuccessResponse(m *channel.Message, ch channel.Channel, index uint64) { resp := channel.NewMessage(SuccessResponseType, nil) resp.ReplyTo(m) + resp.PutUint64Header(IndexHeader, index) if sendErr := ch.Send(resp); sendErr != nil { logrus.WithError(sendErr).Error("error while sending success response") } diff --git a/controller/raft/hclog_adapter.go b/controller/raft/hclog_adapter.go index 1a52cebea..0e3e9b392 100644 --- a/controller/raft/hclog_adapter.go +++ b/controller/raft/hclog_adapter.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package raft import ( diff --git a/controller/raft/index_tracker.go b/controller/raft/index_tracker.go new file mode 100644 index 000000000..58977c81a --- /dev/null +++ b/controller/raft/index_tracker.go @@ -0,0 +1,82 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package raft + +import ( + "github.com/pkg/errors" + "sync" + "time" +) + +type IndexTracker interface { + Index() uint64 + WaitForIndex(index uint64, deadline time.Time) error + NotifyOfIndex(index uint64) +} + +func NewIndexTracker() IndexTracker { + return &indexTrackerImpl{ + notify: make(chan struct{}), + } +} + +type indexTrackerImpl struct { + sync.Mutex + index uint64 + notify chan struct{} +} + +func (self *indexTrackerImpl) Index() uint64 { + idx, _ := self.getState() + return idx +} + +func (self *indexTrackerImpl) getState() (uint64, <-chan struct{}) { + self.Lock() + defer self.Unlock() + return self.index, self.notify +} + +func (self *indexTrackerImpl) WaitForIndex(index uint64, deadline time.Time) error { + for { + currentIndex, notifier := self.getState() + if currentIndex >= index { + return nil + } + now := time.Now() + if !deadline.After(now) { + return errors.New("timed out") + } + + select { + case <-notifier: + case <-time.After(deadline.Sub(now)): + } + } +} + +func (self *indexTrackerImpl) NotifyOfIndex(index uint64) { + self.Lock() + defer self.Unlock() + + if index > self.index { + self.index = index + } + + close(self.notify) + self.notify = make(chan struct{}) +} diff --git a/controller/raft/index_tracker_test.go b/controller/raft/index_tracker_test.go new file mode 100644 index 000000000..695f1b74e --- /dev/null +++ b/controller/raft/index_tracker_test.go @@ -0,0 +1,84 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + +package raft + +import ( + "github.com/stretchr/testify/require" + "testing" + "time" +) + +func TestIndexTracker(t *testing.T) { + indexTracker := NewIndexTracker() + + req := require.New(t) + indexTracker.NotifyOfIndex(1) + req.NoError(indexTracker.WaitForIndex(1, time.Now())) // if it's already complete, should work + + // If it never completes, should fail + req.Error(indexTracker.WaitForIndex(2, time.Now().Add(20*time.Millisecond))) + + notifyAsync := func(index uint64, after time.Duration) { + go func() { + time.Sleep(after) + indexTracker.NotifyOfIndex(index) + }() + } + + notifyAsync(2, 20*time.Millisecond) + req.NoError(indexTracker.WaitForIndex(2, time.Now().Add(30*time.Millisecond))) + req.NoError(indexTracker.WaitForIndex(2, time.Now())) + + waitAsync := func(index uint64, timeout time.Duration) <-chan error { + result := make(chan error, 1) + go func() { + err := indexTracker.WaitForIndex(index, time.Now().Add(timeout)) + if err == nil { + close(result) + } else { + result <- err + close(result) + } + }() + return result + } + + notifyAsync(3, 20*time.Millisecond) + var results []<-chan error + for i := 0; i < 10; i++ { + results = append(results, waitAsync(3, 30*time.Millisecond)) + } + req.Error(indexTracker.WaitForIndex(3, time.Now().Add(10*time.Millisecond))) + time.Sleep(15 * time.Millisecond) + + for _, result := range results { + var err error + select { + case err = <-result: + default: + } + req.NoError(err) + + closed := false + select { + case <-result: + closed = true + default: + } + req.True(closed) + } +} diff --git a/controller/raft/member.go b/controller/raft/member.go index 297b5bacd..700285b1d 100644 --- a/controller/raft/member.go +++ b/controller/raft/member.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package raft import ( diff --git a/controller/raft/mesh/mesh.go b/controller/raft/mesh/mesh.go index de71fbb89..95df66e26 100644 --- a/controller/raft/mesh/mesh.go +++ b/controller/raft/mesh/mesh.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package mesh import ( @@ -161,7 +177,7 @@ func (self *impl) GetOrConnectPeer(address string, timeout time.Duration) (*Peer return nil, errors.New("cannot get peer for empty address") } if peer := self.GetPeer(raft.ServerAddress(address)); peer != nil { - logrus.Infof("existing peer found for %v, returning", address) + logrus.Debugf("existing peer found for %v, returning", address) return peer, nil } logrus.Infof("creating new peer for %v, returning", address) diff --git a/controller/raft/mesh/peerconn.go b/controller/raft/mesh/peerconn.go index 26b31117a..6e757621a 100644 --- a/controller/raft/mesh/peerconn.go +++ b/controller/raft/mesh/peerconn.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package mesh import ( diff --git a/controller/raft/mesh/timeouts.go b/controller/raft/mesh/timeouts.go index 24a72b7bf..43459845b 100644 --- a/controller/raft/mesh/timeouts.go +++ b/controller/raft/mesh/timeouts.go @@ -1,3 +1,19 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package mesh import ( diff --git a/controller/raft/raft.go b/controller/raft/raft.go index 89c4f2f47..a9ec54915 100644 --- a/controller/raft/raft.go +++ b/controller/raft/raft.go @@ -1,7 +1,24 @@ +/* + Copyright NetFoundry Inc. + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + https://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +*/ + package raft import ( "encoding/json" + "fmt" "github.com/google/uuid" "github.com/hashicorp/raft" raftboltdb "github.com/hashicorp/raft-boltdb" @@ -20,6 +37,7 @@ import ( "os" "path" "reflect" + "strconv" "sync" "time" ) @@ -41,6 +59,7 @@ func NewController(id *identity.TokenId, config *Config, metricsRegistry metrics Id: id, Config: config, metricsRegistry: metricsRegistry, + indexTracker: NewIndexTracker(), } if err := result.Init(); err != nil { return nil, err @@ -61,6 +80,7 @@ type Controller struct { servers []raft.Server metricsRegistry metrics.Registry closeNotify <-chan struct{} + indexTracker IndexTracker } // GetRaft returns the managed raft instance @@ -92,6 +112,7 @@ func (self *Controller) GetLeaderAddr() string { // Dispatch dispatches the given command to the current leader. If the current node is the leader, the command // will be applied and the result returned func (self *Controller) Dispatch(cmd command.Command) error { + log := pfxlog.Logger() if validatable, ok := cmd.(command.Validatable); ok { if err := validatable.Validate(); err != nil { return err @@ -99,9 +120,12 @@ func (self *Controller) Dispatch(cmd command.Command) error { } if self.IsLeader() { - return self.applyCommand(cmd) + _, err := self.applyCommand(cmd) + return err } + log.WithField("cmd", reflect.TypeOf(cmd)).WithField("dest", self.GetLeaderAddr()).Info("forwarding command") + peer, err := self.GetMesh().GetOrConnectPeer(self.GetLeaderAddr(), 5*time.Second) if err != nil { return err @@ -119,17 +143,19 @@ func (self *Controller) Dispatch(cmd command.Command) error { } if result.ContentType == SuccessResponseType { + idx, found := result.GetUint64Header(IndexHeader) + if found { + if err = self.indexTracker.WaitForIndex(idx, time.Now().Add(5*time.Second)); err != nil { + return err + } + } return nil } if result.ContentType == ErrorResponseType { errCode, found := result.GetUint32Header(HeaderErrorCode) if found && errCode == ErrorCodeApiError { - apiErr := &errorz.ApiError{} - if err = json.Unmarshal(result.Body, apiErr); err != nil { - return errors.Wrap(err, "unable to decode api error response") - } - return apiErr + return self.decodeApiError(result.Body) } return errors.New(string(result.Body)) } @@ -137,42 +163,131 @@ func (self *Controller) Dispatch(cmd command.Command) error { return errors.Errorf("unexpected response type %v", result.ContentType) } +func (self *Controller) decodeApiError(data []byte) error { + m := map[string]interface{}{} + if err := json.Unmarshal(data, &m); err != nil { + pfxlog.Logger().Warnf("invalid api error encoding, unable to decode: %v", string(data)) + return errors.New(string(data)) + } + + apiErr := &errorz.ApiError{} + + if code, ok := m["code"]; ok { + if apiErr.Code, ok = code.(string); !ok { + pfxlog.Logger().Warnf("invalid api error encoding, invalid code, not string: %v", string(data)) + return errors.New(string(data)) + } + } else { + pfxlog.Logger().Warnf("invalid api error encoding, no code: %v", string(data)) + return errors.New(string(data)) + } + + if status, ok := m["status"]; ok { + statusStr := fmt.Sprintf("%v", status) + statusInt, err := strconv.Atoi(statusStr) + if err != nil { + pfxlog.Logger().Warnf("invalid api error encoding, invalid code, not int: %v", string(data)) + return errors.New(string(data)) + } + apiErr.Status = statusInt + } else { + pfxlog.Logger().Warnf("invalid api error encoding, no status: %v", string(data)) + return errors.New(string(data)) + } + + if message, ok := m["message"]; ok { + if apiErr.Message, ok = message.(string); !ok { + pfxlog.Logger().Warnf("invalid api error encoding, no message: %v", string(data)) + return errors.New(string(data)) + } + } else { + pfxlog.Logger().Warnf("invalid api error encoding, invalid message, not string: %v", string(data)) + return errors.New(string(data)) + } + + if cause, ok := m["cause"]; ok { + if strCause, ok := cause.(string); ok { + apiErr.Cause = errors.New(strCause) + } else if objCause, ok := cause.(map[string]interface{}); ok { + apiErr.Cause = self.parseFieldError(objCause) + if apiErr.Cause == nil { + if b, err := json.Marshal(objCause); err == nil { + apiErr.Cause = errors.New(string(b)) + } else { + apiErr.Cause = errors.New(fmt.Sprintf("%+v", objCause)) + } + } + } else { + pfxlog.Logger().Warnf("invalid api error encoding, no cause: %v", string(data)) + return errors.New(string(data)) + } + } + + return apiErr +} + +func (self *Controller) parseFieldError(m map[string]any) *errorz.FieldError { + var fieldError *errorz.FieldError + field, ok := m["field"] + if !ok { + return nil + } + + fieldStr, ok := field.(string) + if !ok { + return nil + } + + fieldError = &errorz.FieldError{ + FieldName: fieldStr, + FieldValue: m["value"], + } + + if reason, ok := m["reason"]; ok { + if reasonStr, ok := reason.(string); ok { + fieldError.Reason = reasonStr + } + } + + return fieldError +} + // applyCommand encodes the command and passes it to ApplyEncodedCommand -func (self *Controller) applyCommand(cmd command.Command) error { +func (self *Controller) applyCommand(cmd command.Command) (uint64, error) { encoded, err := cmd.Encode() if err != nil { - return err + return 0, err } return self.ApplyEncodedCommand(encoded) } // ApplyEncodedCommand applies the command to the RAFT distributed log -func (self *Controller) ApplyEncodedCommand(encoded []byte) error { - val, err := self.ApplyWithTimeout(encoded, 5*time.Second) +func (self *Controller) ApplyEncodedCommand(encoded []byte) (uint64, error) { + val, idx, err := self.ApplyWithTimeout(encoded, 5*time.Second) if err != nil { - return err + return 0, err } if err, ok := val.(error); ok { - return err + return 0, err } if val != nil { cmd, err := self.Fsm.decoders.Decode(encoded) if err != nil { logrus.WithError(err).Error("failed to unmarshal command which returned non-nil, non-error value") - return err + return 0, err } pfxlog.Logger().WithField("cmdType", reflect.TypeOf(cmd)).Error("command return non-nil, non-error value") } - return nil + return idx, nil } // ApplyWithTimeout applies the given command to the RAFT distributed log with the given timeout -func (self *Controller) ApplyWithTimeout(log []byte, timeout time.Duration) (interface{}, error) { +func (self *Controller) ApplyWithTimeout(log []byte, timeout time.Duration) (interface{}, uint64, error) { f := self.Raft.Apply(log, timeout) if err := f.Error(); err != nil { - return nil, err + return nil, 0, err } - return f.Response(), nil + return f.Response(), f.Index(), nil } // Init sets up the Mesh and Raft instances @@ -226,7 +341,7 @@ func (self *Controller) Init() error { transport := raft.NewNetworkTransportWithLogger(self.Mesh, 3, 10*time.Second, hclLogger) - self.Fsm = NewFsm(raftConfig.DataDir, command.GetDefaultDecoders()) + self.Fsm = NewFsm(raftConfig.DataDir, command.GetDefaultDecoders(), self.indexTracker) if err = self.Fsm.Init(); err != nil { return errors.Wrap(err, "failed to init FSM")