Pass lmdb.Env to Syncer, fix flaky sync tests

Explicitly pass the lmdb.Env to the syncer, so that we have more control
over its lifetime and avoid opening and closing it multiple times, which
can cause crashes in the LMDB C code.

Rewrite the sync tests to wait for expected values, instead of relying
on arbitrary sleeps that make the test flaky and slow.

Do not close the lmdb.Env in the sync tests, because for some reason it
causes random SEGFAULTs on Linux (not on macOS).

Author: wojas

cmd/lightningstream/commands/sync.go

@@ -63,23 +63,33 @@ func runSync(receiveOnly bool) error {
 
 	eg, ctx := errgroup.WithContext(ctx)
 	for name, lc := range conf.LMDBs {
+		l := logrus.WithField("db", name)
+		env, err := syncer.OpenEnv(l, lc)
+		if err != nil {
+			return err
+		}
+
 		opt := syncer.Options{
 			ReceiveOnly: receiveOnly,
 		}
-		s, err := syncer.New(name, st, conf, lc, opt)
+		s, err := syncer.New(name, env, st, conf, lc, opt)
 		if err != nil {
 			return err
 		}
 
-		name := name
 		eg.Go(func() error {
+			defer func() {
+				if err := env.Close(); err != nil {
+					l.WithError(err).Error("Env close failed")
+				}
+			}()
 			err := s.Sync(ctx)
 			if err != nil {
 				if err == context.Canceled {
-					logrus.WithField("db", name).Error("Sync cancelled")
+					l.Error("Sync cancelled")
 					return err
 				}
-				logrus.WithError(err).WithField("db", name).Error("Sync failed")
+				l.WithError(err).Error("Sync failed")
 			}
 			return err
 		})

syncer/env.go

@@ -0,0 +1,33 @@
+package syncer
+
+import (
+	"github.com/PowerDNS/lmdb-go/lmdb"
+	"github.com/c2h5oh/datasize"
+	"github.com/sirupsen/logrus"
+	"powerdns.com/platform/lightningstream/config"
+	"powerdns.com/platform/lightningstream/lmdbenv"
+)
+
+// OpenEnv opens the LMDB env with the right options
+func OpenEnv(l logrus.FieldLogger, lc config.LMDB) (env *lmdb.Env, err error) {
+	l.WithField("lmdbpath", lc.Path).Info("Opening LMDB")
+	env, err = lmdbenv.NewWithOptions(lc.Path, lc.Options)
+	if err != nil {
+		return nil, err
+	}
+
+	// Print some env info
+	info, err := env.Info()
+	if err != nil {
+		return nil, err
+	}
+	l.WithFields(logrus.Fields{
+		"MapSize":   datasize.ByteSize(info.MapSize).HumanReadable(),
+		"LastTxnID": info.LastTxnID,
+	}).Info("Env info")
+
+	// TODO: Perhaps check data if SchemaTracksChanges is set. Check if
+	//       the timestamp is in a reasonable range or 0.
+
+	return env, nil
+}

syncer/send_test.go

@@ -37,27 +37,28 @@ func doBenchmarkSyncerSendOnce(b *testing.B, native, dupsort bool) {
 		extraDBIFlags = lmdb.DupSort
 	}
 
-	syncer, err := New("test", memory.New(), config.Config{}, config.LMDB{
-		SchemaTracksChanges: native,
-		DupSortHack:         dupsort,
-	}, Options{})
-	require.NoError(t, err)
-
 	l, hook := test.NewNullLogger()
 	_ = hook
-	syncer.l = l
+	lc := config.LMDB{
+		SchemaTracksChanges: native,
+		DupSortHack:         dupsort,
+	}
 
 	// Fixed value
 	// We add a header, but we can also benchmark this as all app value
 	val := make([]byte, header.MinHeaderSize, 50)
 	header.PutBasic(val, header.TimestampFromTime(now), 42, header.NoFlags)
 	val = append(val, "TESTING-123456789"...)
 
-	err = lmdbenv.TestEnv(func(env *lmdb.Env) error {
+	err := lmdbenv.TestEnv(func(env *lmdb.Env) error {
 		info, err := env.Info()
 		require.NoError(t, err)
 		t.Logf("env info: %+v", info)
 
+		syncer, err := New("test", env, memory.New(), config.Config{}, lc, Options{})
+		require.NoError(t, err)
+		syncer.l = l
+
 		// Fill some data to dump
 		err = env.Update(func(txn *lmdb.Txn) error {
 			// First insert the initial data into the main database

syncer/shadow_test.go

@@ -39,12 +39,12 @@ func TestSyncer_shadow(t *testing.T) {
 	ts2 := testTS(2)
 	ts3 := testTS(3)
 
-	s, err := New("test", nil, config.Config{}, config.LMDB{}, Options{})
-	assert.NoError(t, err)
+	err := lmdbenv.TestEnv(func(env *lmdb.Env) error {
+		s, err := New("test", env, nil, config.Config{}, config.LMDB{}, Options{})
+		assert.NoError(t, err)
 
-	err = lmdbenv.TestEnv(func(env *lmdb.Env) error {
 		// Initial data
-		err := env.Update(func(txn *lmdb.Txn) error {
+		err = env.Update(func(txn *lmdb.Txn) error {
 			// First insert the initial data into the main database
 			dbi, err := txn.OpenDBI("foo", lmdb.Create)
 			assert.NoError(t, err)

syncer/sync.go

@@ -27,13 +27,7 @@ const (
 
 // Sync opens the env and starts the two-way sync loop.
 func (s *Syncer) Sync(ctx context.Context) error {
-	// Open the env
-	env, err := s.openEnv()
-	if err != nil {
-		return err
-	}
-	defer s.closeEnv(env)
-
+	env := s.env
 	status.AddLMDBEnv(s.name, env)
 	defer status.RemoveLMDBEnv(s.name)
 

syncer/sync_test.go

@@ -13,7 +13,7 @@ import (
 	"github.com/PowerDNS/simpleblob"
 	"github.com/PowerDNS/simpleblob/backends/memory"
 	"github.com/sirupsen/logrus"
-	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"powerdns.com/platform/lightningstream/config"
 	"powerdns.com/platform/lightningstream/config/logger"
 	"powerdns.com/platform/lightningstream/lmdbenv"
@@ -23,7 +23,7 @@ import (
 
 const testLMDBName = "default"
 const testDBIName = "test"
-const tick = 100 * time.Millisecond
+const tick = 10 * time.Millisecond
 
 func TestSyncer_Sync_startup(t *testing.T) {
 	t.Run("with-timestamped-schema", func(t *testing.T) {
@@ -44,6 +44,13 @@ func doTest(t *testing.T, withHeader bool) {
 	syncerA, envA := createInstance(t, "a", st, withHeader)
 	syncerB, envB := createInstance(t, "b", st, withHeader)
 
+	// For some reason trying to close the envs in this test segfaults on Linux (not on macOS).
+	// It appears like this is caused by the syncer still running after cancellation
+	// (and thus after the env is closed), but I did not get to the bottom of this yet.
+	// [signal SIGSEGV: segmentation violation code=0x1 addr=0x7fd015132090 pc=0x8dc942]
+	//defer envA.Close()
+	//defer envB.Close()
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	ctxA, cancelA := context.WithCancel(ctx)
@@ -55,46 +62,36 @@ func doTest(t *testing.T, withHeader bool) {
 	t.Log("Starting syncer A")
 	go runSync(ctxA, syncerA)
 
-	t.Log("----------")
-	time.Sleep(2 * tick)
 	t.Log("----------")
 
 	// Expecting one snapshot on startup, because not empty
-	logSnapshotList(t, st)
-	entries := listInstanceSnapshots(st, "a")
-	assert.Len(t, entries, 1, "A")
+	requireSnapshotsLenWait(t, st, 1, "A")
 
 	// Start syncer B with an empty LMDB
+	// Starting with an empty LMDB is a special case that will not trigger any
+	// local snapshot.
 	t.Log("Starting syncer B")
 	go runSync(ctxB, syncerB)
 
-	t.Log("----------")
-	time.Sleep(2 * tick)
 	t.Log("----------")
 
-	// No snapshot, because empty
-	logSnapshotList(t, st)
-	entries = listInstanceSnapshots(st, "b")
-	assert.Len(t, entries, 0, "B")
+	// Wait until the data from A was synced to B
+	assertKeyWait(t, envB, "foo", "v1", withHeader)
 
-	assertKey(t, envB, "foo", "v1", withHeader)
+	// No snapshot made by B, because we started empty
+	requireSnapshotsLenWait(t, st, 0, "B")
 
 	// Now set something in B
 	setKey(t, envB, "foo", "v2", withHeader)
 
-	t.Log("----------")
-	time.Sleep(3 * tick)
 	t.Log("----------")
 
 	// New snapshot in B, no new one in A
-	logSnapshotList(t, st)
-	entries = listInstanceSnapshots(st, "b")
-	assert.Len(t, entries, 1, "B")
-	entries = listInstanceSnapshots(st, "a")
-	assert.Len(t, entries, 1, "A")
+	requireSnapshotsLenWait(t, st, 1, "B")
+	requireSnapshotsLenWait(t, st, 1, "A")
 
 	// New value should be present in A
-	assertKey(t, envB, "foo", "v2", withHeader)
+	assertKeyWait(t, envB, "foo", "v2", withHeader)
 
 	// Restart syncer for A
 	t.Log("Restarting syncer A")
@@ -103,16 +100,13 @@ func doTest(t *testing.T, withHeader bool) {
 	t.Log("----------")
 	go runSync(ctxA, syncerA)
 
-	t.Log("----------")
-	time.Sleep(3 * tick)
 	t.Log("----------")
 
 	// Check is the contents of A are still correct after restart
-	assertKey(t, envA, "foo", "v2", withHeader)
-	entries = listInstanceSnapshots(st, "a")
+	assertKeyWait(t, envA, "foo", "v2", withHeader)
 	// A new snapshot should always be created on startup, in case the LMDB
 	// was modified while it was down.
-	assert.Len(t, entries, 2, "A")
+	requireSnapshotsLenWait(t, st, 2, "A")
 
 	// Stopping syncer for A
 	t.Log("Stopping syncer A")
@@ -127,15 +121,12 @@ func doTest(t *testing.T, withHeader bool) {
 	ctxA, cancelA = context.WithCancel(ctx)
 	go runSync(ctxA, syncerA)
 	t.Log("----------")
-	time.Sleep(6 * tick)
-	t.Log("----------")
 
+	// New value in A should get synced to B
+	assertKeyWait(t, envB, "new", "hello", withHeader)
 	// Check if the contents of A are still correct after restart
-	assertKey(t, envA, "new", "hello", withHeader)
-	// It should also be synced to B
-	assertKey(t, envB, "new", "hello", withHeader)
-	entries = listInstanceSnapshots(st, "a")
-	assert.Len(t, entries, 3, "A")
+	assertKeyWait(t, envA, "new", "hello", withHeader)
+	requireSnapshotsLenWait(t, st, 3, "A")
 
 	cancelA()
 	cancelB()
@@ -144,16 +135,16 @@ func doTest(t *testing.T, withHeader bool) {
 
 func createInstance(t *testing.T, name string, st simpleblob.Interface, timestamped bool) (*Syncer, *lmdb.Env) {
 	env, tmp, err := createLMDB(t)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 
 	c := createConfig(name, tmp, timestamped)
-	syncer, err := New("default", st, c, c.LMDBs[testLMDBName], Options{})
-	assert.NoError(t, err)
+	syncer, err := New("default", env, st, c, c.LMDBs[testLMDBName], Options{})
+	require.NoError(t, err)
 
 	return syncer, env
 }
 
-func logSnapshotList(t *testing.T, st simpleblob.Interface) {
+func LogSnapshotList(t *testing.T, st simpleblob.Interface) {
 	ctx := context.Background()
 	entries, _ := st.List(ctx, "")
 	var lines []string
@@ -178,17 +169,57 @@ func listInstanceSnapshots(st simpleblob.Interface, instance string) simpleblob.
 	return entries
 }
 
+func requireSnapshotsLenWait(t *testing.T, st simpleblob.Interface, expLen int, instance string) {
+	var list simpleblob.BlobList
+	// Retry until it succeeds
+	var i int
+	const maxIter = 150
+	const sleepTime = 10 * time.Millisecond
+	defer func() {
+		t.Logf("Waited %d/%d iterations for the expected snapshot length", i, maxIter)
+	}()
+	for i = 0; i < maxIter; i++ {
+		list = listInstanceSnapshots(st, strings.ToLower(instance))
+		l := len(list)
+		if l == expLen {
+			return
+		}
+		time.Sleep(sleepTime)
+	}
+	// This one is actually expected to fail, call it for the formatting
+	t.Logf("Gave up on waiting for the expected snapshot length")
+	require.Len(t, list, expLen, instance)
+}
+
 func runSync(ctx context.Context, syncer *Syncer) {
 	err := syncer.Sync(ctx)
 	if err != nil && err != context.Canceled {
 		logrus.WithError(err).WithField("syncer", syncer.name).Error("Syncer Sync error")
 	}
 }
 
-func assertKey(t *testing.T, env *lmdb.Env, key, val string, withHeader bool) {
-	kv, err := dumpData(env, withHeader)
-	assert.NoError(t, err)
-	assert.Equal(t, val, kv[key])
+func assertKeyWait(t *testing.T, env *lmdb.Env, key, val string, withHeader bool) {
+	var kv map[string]string
+	var err error
+	var i int
+	const maxIter = 150
+	const sleepTime = 10 * time.Millisecond
+	defer func() {
+		t.Logf("Waited %d/%d iterations for the expected key", i, maxIter)
+	}()
+	for i = 0; i < maxIter; i++ {
+		kv, err = dumpData(env, withHeader)
+		if err != nil && !lmdb.IsNotFound(err) {
+			require.NoError(t, err)
+		}
+		if kv[key] == val {
+			return
+		}
+		time.Sleep(sleepTime)
+	}
+	// Expected to fail now, called for formatting
+	t.Logf("Gave up on waiting for the expected key")
+	require.Equal(t, val, kv[key])
 }
 
 func setKey(t *testing.T, env *lmdb.Env, key, val string, withHeader bool) {
@@ -210,13 +241,13 @@ func setKey(t *testing.T, env *lmdb.Env, key, val string, withHeader bool) {
 		err = txn.Put(dbi, []byte(key), valb, 0)
 		return err
 	})
-	assert.NoError(t, err)
+	require.NoError(t, err)
 }
 
 func dumpData(env *lmdb.Env, withHeader bool) (map[string]string, error) {
 	data := make(map[string]string)
 	err := env.View(func(txn *lmdb.Txn) error {
-		dbi, err := txn.OpenDBI(testDBIName, lmdb.Create)
+		dbi, err := txn.OpenDBI(testDBIName, 0)
 		if err != nil {
 			return err
 		}