fix: wal deadlock (#375)

Co-authored-by: anrs <anders.hu@shopee.com>

Author: anrs

wal/hydro.go

@@ -44,21 +44,25 @@ func (h *Hydro) Register(handler EventHandler) {
 func (h *Hydro) Recover() {
 	ch, _ := h.kv.Scan([]byte(EventPrefix))
 
+	events := []HydroEvent{}
 	for ent := range ch {
-		event, err := h.decodeEvent(ent)
+		ev, err := h.decodeEvent(ent)
 		if err != nil {
 			log.Errorf("[Recover] decode event error: %v", err)
 			continue
 		}
+		events = append(events, ev)
+	}
 
-		handler, ok := h.getEventHandler(event.Type)
+	for _, ev := range events {
+		handler, ok := h.getEventHandler(ev.Type)
 		if !ok {
-			log.Errorf("[Recover] no such event handler for %s", event.Type)
+			log.Errorf("[Recover] no such event handler for %s", ev.Type)
 			continue
 		}
 
-		if err := h.recover(handler, event); err != nil {
-			log.Errorf("[Recover] handle event %d (%s) failed: %v", event.ID, event.Type, err)
+		if err := h.recover(handler, ev); err != nil {
+			log.Errorf("[Recover] handle event %d (%s) failed: %v", ev.ID, ev.Type, err)
 			continue
 		}
 	}
@@ -129,7 +133,6 @@ func (h *Hydro) decodeEvent(ent kv.ScanEntry) (event HydroEvent, err error) {
 	}
 
 	event.kv = h.kv
-
 	event.ID, err = parseHydroEventID(key)
 
 	return

wal/hydro_test.go

@@ -2,7 +2,11 @@ package wal
 
 import (
 	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -211,6 +215,41 @@ func TestHydroEventParseIDShouldRemovePadding(t *testing.T) {
 	require.Equal(t, uint64(15), id)
 }
 
+func TestHydroRecoverWithRealLithium(t *testing.T) {
+	dir, rmdir := tempdir(t)
+	defer rmdir()
+
+	hydro := NewHydro()
+	// Uses a real Lithium instance rather than a mocked one.
+	require.NoError(t, hydro.Open(filepath.Join(dir, "temp.wal"), time.Second))
+
+	handler := SimpleEventHandler{
+		event:  "create",
+		encode: func(interface{}) ([]byte, error) { return []byte("{}"), nil },
+		decode: func([]byte) (interface{}, error) { return struct{}{}, nil },
+		check:  func(interface{}) (bool, error) { return true, nil },
+		handle: func(interface{}) error { return nil },
+	}
+	hydro.Register(handler)
+
+	hydro.Log(handler.event, struct{}{})
+	hydro.Log(handler.event, struct{}{})
+	hydro.Log(handler.event, struct{}{})
+
+	hydro.Recover()
+
+	ch, _ := hydro.kv.Scan([]byte(EventPrefix))
+	for range ch {
+		require.FailNow(t, "expects no data")
+	}
+}
+
+func tempdir(t *testing.T) (string, func()) {
+	dir, err := ioutil.TempDir("", "temp.wal")
+	require.NoError(t, err)
+	return dir, func() { os.RemoveAll(dir) }
+}
+
 func newTestEventHandler(eventype string, checked, handled, encoded, decoded *bool) SimpleEventHandler {
 	check := func(interface{}) (bool, error) {
 		*checked = true

wal/kv/lithium.go

@@ -81,19 +81,13 @@ func (l *Lithium) close() error {
 
 // Put creates/updates a key/value pair.
 func (l *Lithium) Put(key []byte, value []byte) (err error) {
-	l.Lock()
-	defer l.Unlock()
-
 	return l.update(func(bkt *bolt.Bucket) error {
 		return bkt.Put(key, value)
 	})
 }
 
 // Get read a key's value.
 func (l *Lithium) Get(key []byte) (dst []byte, err error) {
-	l.Lock()
-	defer l.Unlock()
-
 	err = l.view(func(bkt *bolt.Bucket) error {
 		src := bkt.Get(key)
 		dst = make([]byte, len(src))
@@ -110,8 +104,6 @@ func (l *Lithium) Get(key []byte) (dst []byte, err error) {
 
 // Delete deletes a key.
 func (l *Lithium) Delete(key []byte) error {
-	l.Lock()
-	defer l.Unlock()
 	return l.update(func(bkt *bolt.Bucket) error {
 		return bkt.Delete(key)
 	})
@@ -130,32 +122,24 @@ func (l *Lithium) Scan(prefix []byte) (<-chan ScanEntry, func()) {
 	go func() {
 		defer close(ch)
 
-		l.Lock()
-		defer l.Unlock()
-
 		close(locked)
-		ent := &LithiumScanEntry{}
 
 		scan := func(bkt *bolt.Bucket) error {
 			c := bkt.Cursor()
 			for key, value := c.First(); key != nil && bytes.HasPrefix(key, prefix); key, value = c.Next() {
-				ent.key = key
-				ent.value = value
-
 				select {
 				case <-exit:
 					return nil
-				case ch <- *ent:
+				case ch <- LithiumScanEntry{key: key, value: value}:
 				}
 			}
 			return nil
 		}
 
 		if err := l.view(scan); err != nil {
-			ent.err = err
 			select {
 			case <-exit:
-			case ch <- *ent:
+			case ch <- LithiumScanEntry{err: err}:
 			}
 		}
 	}()
@@ -168,9 +152,6 @@ func (l *Lithium) Scan(prefix []byte) (<-chan ScanEntry, func()) {
 
 // NextSequence generates a new sequence.
 func (l *Lithium) NextSequence() (uint64, error) {
-	l.Lock()
-	defer l.Unlock()
-
 	var seq uint64
 	err := l.update(func(bkt *bolt.Bucket) (ue error) {
 		seq, ue = bkt.NextSequence()