Fix raftwal tests

raftwal/fsm_test.go

@@ -96,7 +96,8 @@ func TestAddAndRestoreSnapshot(t *testing.T) {
 	_, trans := raft.NewInmemTransport(raft.NewInmemAddr())
 	sink, _ := snap.Create(raft.SnapshotVersionMax, 10, 3, configuration, 2, trans)
 
-	fsmsnap.Persist(sink)
+	err = fsmsnap.Persist(sink)
+	assert.NoError(t, err)
 	// fsm.Close()
 
 	// Read the latest snapshot
@@ -110,7 +111,8 @@ func TestAddAndRestoreSnapshot(t *testing.T) {
 	fsm2, err := NewBalloonFSM(store2, hashing.NewSha256Hasher, make(chan *protocol.Snapshot, 100))
 	assert.NoError(t, err)
 
-	fsm2.Restore(r)
+	err = fsm2.Restore(r)
+	assert.NoError(t, err)
 
 	// Error: Command already applied
 	e := fsm2.Apply(newRaftLog(0, 0)).(*fsmAddResponse)
@@ -119,6 +121,6 @@ func TestAddAndRestoreSnapshot(t *testing.T) {
 
 func newRaftLog(index, term uint64) *raft.Log {
 	event := []byte("All's right with the world")
-	data, _ := commands.Encode(commands.AddEventCommandType, &commands.AddEventCommand{event})
-	return &raft.Log{index, term, raft.LogCommand, data}
+	data, _ := commands.Encode(commands.AddEventCommandType, &commands.AddEventCommand{Event: event})
+	return &raft.Log{Index: index, Term: term, Type: raft.LogCommand, Data: data}
 }

raftwal/raft.go

@@ -35,7 +35,6 @@ import (
 
 const (
 	retainSnapshotCount = 2
-	raftTimeout         = 10 * time.Second
 	leaderWaitDelay     = 100 * time.Millisecond
 	raftLogCacheSize    = 512
 )
@@ -373,7 +372,7 @@ func (b *RaftBalloon) remove(id string) error {
 		return f.Error()
 	}
 
-	cmd := &commands.MetadataDeleteCommand{id}
+	cmd := &commands.MetadataDeleteCommand{Id: id}
 	_, err := b.raftApply(commands.MetadataDeleteCommandType, cmd)
 
 	return err
@@ -397,7 +396,7 @@ func (b *RaftBalloon) raftApply(t commands.CommandType, cmd interface{}) (interf
 */
 
 func (b *RaftBalloon) Add(event []byte) (*balloon.Snapshot, error) {
-	cmd := &commands.AddEventCommand{event}
+	cmd := &commands.AddEventCommand{Event: event}
 	resp, err := b.raftApply(commands.AddEventCommandType, cmd)
 	if err != nil {
 		return nil, err

raftwal/raft_test.go

@@ -38,25 +38,25 @@ import (
 
 func init() {
 	log.SetLogger("testRaft", log.DEBUG)
+
 }
 
 func raftAddr(id int) string {
 	return fmt.Sprintf(":830%d", id)
 }
-func joinAddr(id int) string {
-	return fmt.Sprintf(":840%d", id)
-}
 
 func newNode(t *testing.T, id int) (*RaftBalloon, func()) {
 	badgerPath := fmt.Sprintf("/var/tmp/raft-test/node%d/badger", id)
 
-	os.MkdirAll(badgerPath, os.FileMode(0755))
+	err := os.MkdirAll(badgerPath, os.FileMode(0755))
+	require.NoError(t, err)
 	badger, err := badger.NewBadgerStore(badgerPath)
 	require.NoError(t, err)
 
 	raftPath := fmt.Sprintf("/var/tmp/raft-test/node%d/raft", id)
-	os.MkdirAll(raftPath, os.FileMode(0755))
-	r, err := NewRaftBalloon(raftPath, raftAddr(id), fmt.Sprintf("%d", id), badger, make(chan *protocol.Snapshot, 100))
+	err = os.MkdirAll(raftPath, os.FileMode(0755))
+	require.NoError(t, err)
+	r, err := NewRaftBalloon(raftPath, raftAddr(id), fmt.Sprintf("%d", id), badger, make(chan *protocol.Snapshot, 25000))
 	require.NoError(t, err)
 
 	return r, func() {
@@ -355,12 +355,14 @@ func mustTempDir() string {
 func newNodeBench(b *testing.B, id int) (*RaftBalloon, func()) {
 	badgerPath := fmt.Sprintf("/var/tmp/raft-test/node%d/badger", id)
 
-	os.MkdirAll(badgerPath, os.FileMode(0755))
+	err := os.MkdirAll(badgerPath, os.FileMode(0755))
+	require.NoError(b, err)
 	badger, err := badger.NewBadgerStore(badgerPath)
 	require.NoError(b, err)
 
 	raftPath := fmt.Sprintf("/var/tmp/raft-test/node%d/raft", id)
-	os.MkdirAll(raftPath, os.FileMode(0755))
+	err = os.MkdirAll(raftPath, os.FileMode(0755))
+	require.NoError(b, err)
 	r, err := NewRaftBalloon(raftPath, raftAddr(id), fmt.Sprintf("%d", id), badger, make(chan *protocol.Snapshot, 100))
 	require.NoError(b, err)
 

raftwal/snapshot.go

@@ -37,7 +37,7 @@ func (f *fsmSnapshot) Persist(sink raft.SnapshotSink) error {
 		return sink.Close()
 	}()
 	if err != nil {
-		sink.Cancel()
+		_ = sink.Cancel()
 	}
 	return err
 }