[Merged by Bors] - sync2: multipeer: fix edge cases

sql/database.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"maps"
+	"math/rand/v2"
 	"net/url"
 	"os"
 	"strings"
@@ -223,8 +224,12 @@ type Opt func(c *conf)
 
 // OpenInMemory creates an in-memory database.
 func OpenInMemory(opts ...Opt) (*sqliteDatabase, error) {
-	opts = append(opts, WithConnections(1), withForceFresh())
-	return Open("file::memory:?mode=memory", opts...)
+	opts = append(opts, withForceFresh())
+	// Unique uri is needed to avoid sharing the same in-memory database,
+	// while allowing multiple connections to the same database.
+	uri := fmt.Sprintf("file:mem-%d-%d?mode=memory&cache=shared",
+		rand.Uint64(), rand.Uint64())
+	return Open(uri, opts...)
 }
 
 // InMemory creates an in-memory database for testing and panics if

sync2/dbset/dbset.go

@@ -1,6 +1,7 @@
 package dbset
 
 import (
+	"context"
 	"fmt"
 	"maps"
 	"sync"
@@ -49,7 +50,16 @@ func (d *DBSet) handleIDfromDB(stmt *sql.Statement) bool {
 	return true
 }
 
+// Loaded returns true if the DBSet is loaded.
+// Implements rangesync.OrderedSet.
+func (d *DBSet) Loaded() bool {
+	d.loadMtx.Lock()
+	defer d.loadMtx.Unlock()
+	return d.ft != nil
+}
+
 // EnsureLoaded ensures that the DBSet is loaded and ready to be used.
+// Implements rangesync.OrderedSet.
 func (d *DBSet) EnsureLoaded() error {
 	d.loadMtx.Lock()
 	defer d.loadMtx.Unlock()
@@ -65,7 +75,7 @@ func (d *DBSet) EnsureLoaded() error {
 	if err != nil {
 		return fmt.Errorf("error loading count: %w", err)
 	}
-	d.dbStore = fptree.NewDBBackedStore(d.db, d.snapshot, count, d.keyLen)
+	d.dbStore = fptree.NewDBBackedStore(d.db, d.snapshot, d.keyLen)
 	d.ft = fptree.NewFPTree(count, d.dbStore, d.keyLen, d.maxDepth)
 	return d.snapshot.Load(d.db, d.handleIDfromDB)
 }
@@ -217,28 +227,39 @@ func (d *DBSet) Advance() error {
 	return d.snapshot.LoadSinceSnapshot(d.db, oldSnapshot, d.handleIDfromDB)
 }
 
-// Copy creates a copy of the DBSet.
+// WithCopy invokes the specified function, passing it a temporary copy of the DBSet.
 // Implements rangesync.OrderedSet.
-func (d *DBSet) Copy(syncScope bool) rangesync.OrderedSet {
-	d.loadMtx.Lock()
-	defer d.loadMtx.Unlock()
-	if d.ft == nil {
-		// FIXME
-		panic("BUG: can't copy the DBItemStore before it's loaded")
+func (d *DBSet) WithCopy(ctx context.Context, toCall func(rangesync.OrderedSet) error) error {
+	if err := d.EnsureLoaded(); err != nil {
+		return fmt.Errorf("loading DBSet: %w", err)
 	}
+	d.loadMtx.Lock()
 	ft := d.ft.Clone().(*fptree.FPTree)
-	return &DBSet{
+	ds := &DBSet{
 		db:       d.db,
 		ft:       ft,
 		st:       d.st,
+		snapshot: d.snapshot,
 		keyLen:   d.keyLen,
 		maxDepth: d.maxDepth,
 		dbStore:  d.dbStore,
 		received: maps.Clone(d.received),
 	}
+	d.loadMtx.Unlock()
+	defer ds.release()
+	db, ok := d.db.(sql.Database)
+	if ok {
+		return db.WithConnection(ctx, func(ex sql.Executor) error {
+			ds.db = ex
+			return toCall(ds)
+		})
+	} else {
+		return toCall(ds)
+	}
 }
 
 // Has returns true if the DBSet contains the given item.
+// Implements rangesync.OrderedSet.
 func (d *DBSet) Has(k rangesync.KeyBytes) (bool, error) {
 	if err := d.EnsureLoaded(); err != nil {
 		return false, err
@@ -258,17 +279,14 @@ func (d *DBSet) Has(k rangesync.KeyBytes) (bool, error) {
 }
 
 // Recent returns a sequence of items that have been added to the DBSet since the given time.
+// Implements rangesync.OrderedSet.
 func (d *DBSet) Recent(since time.Time) (rangesync.SeqResult, int) {
 	return d.dbStore.Since(make(rangesync.KeyBytes, d.keyLen), since.UnixNano())
 }
 
-// Release releases resources associated with the DBSet.
-func (d *DBSet) Release() error {
-	d.loadMtx.Lock()
-	defer d.loadMtx.Unlock()
+func (d *DBSet) release() {
 	if d.ft != nil {
 		d.ft.Release()
 		d.ft = nil
 	}
-	return nil
 }

sync2/dbset/dbset_test.go

@@ -1,11 +1,13 @@
 package dbset_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/spacemeshos/go-spacemesh/sql"
 	"github.com/spacemeshos/go-spacemesh/sync2/dbset"
 	"github.com/spacemeshos/go-spacemesh/sync2/rangesync"
 	"github.com/spacemeshos/go-spacemesh/sync2/sqlstore"
@@ -36,7 +38,6 @@ func TestDBSet_Empty(t *testing.T) {
 		IDColumn:  "id",
 	}
 	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
 	empty, err := s.Empty()
 	require.NoError(t, err)
 	require.True(t, empty)
@@ -80,7 +81,6 @@ func TestDBSet(t *testing.T) {
 		IDColumn:  "id",
 	}
 	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
 	require.Equal(t, "0000000000000000000000000000000000000000000000000000000000000000",
 		firstKey(t, s.Items()).String())
 	has, err := s.Has(
@@ -184,7 +184,6 @@ func TestDBSet_Receive(t *testing.T) {
 		IDColumn:  "id",
 	}
 	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
 	require.Equal(t, "0000000000000000000000000000000000000000000000000000000000000000",
 		firstKey(t, s.Items()).String())
 
@@ -216,40 +215,42 @@ func TestDBSet_Copy(t *testing.T) {
 		IDColumn:  "id",
 	}
 	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
 	require.Equal(t, "0000000000000000000000000000000000000000000000000000000000000000",
 		firstKey(t, s.Items()).String())
 
-	copy := s.Copy(false)
-
-	info, err := copy.GetRangeInfo(ids[2], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 2, info.Count)
-	require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[2], firstKey(t, info.Items))
-
-	newID := rangesync.MustParseHexKeyBytes("abcdef1234567890000000000000000000000000000000000000000000000000")
-	require.NoError(t, copy.Receive(newID))
-
-	info, err = s.GetRangeInfo(ids[2], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 2, info.Count)
-	require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[2], firstKey(t, info.Items))
-
-	items, err := s.Received().FirstN(100)
-	require.NoError(t, err)
-	require.Empty(t, items)
-
-	info, err = s.GetRangeInfo(ids[2], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 2, info.Count)
-	require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[2], firstKey(t, info.Items))
-
-	items, err = copy.(*dbset.DBSet).Received().FirstN(100)
-	require.NoError(t, err)
-	require.Equal(t, []rangesync.KeyBytes{newID}, items)
+	require.NoError(t, s.WithCopy(context.Background(), func(copy rangesync.OrderedSet) error {
+		info, err := copy.GetRangeInfo(ids[2], ids[0])
+		require.NoError(t, err)
+		require.Equal(t, 2, info.Count)
+		require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
+		require.Equal(t, ids[2], firstKey(t, info.Items))
+
+		newID := rangesync.MustParseHexKeyBytes(
+			"abcdef1234567890000000000000000000000000000000000000000000000000")
+		require.NoError(t, copy.Receive(newID))
+
+		info, err = s.GetRangeInfo(ids[2], ids[0])
+		require.NoError(t, err)
+		require.Equal(t, 2, info.Count)
+		require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
+		require.Equal(t, ids[2], firstKey(t, info.Items))
+
+		items, err := s.Received().FirstN(100)
+		require.NoError(t, err)
+		require.Empty(t, items)
+
+		info, err = s.GetRangeInfo(ids[2], ids[0])
+		require.NoError(t, err)
+		require.Equal(t, 2, info.Count)
+		require.Equal(t, "dddddddddddddddddddddddd", info.Fingerprint.String())
+		require.Equal(t, ids[2], firstKey(t, info.Items))
+
+		items, err = copy.(*dbset.DBSet).Received().FirstN(100)
+		require.NoError(t, err)
+		require.Equal(t, []rangesync.KeyBytes{newID}, items)
+
+		return nil
+	}))
 }
 
 func TestDBItemStore_Advance(t *testing.T) {
@@ -259,64 +260,86 @@ func TestDBItemStore_Advance(t *testing.T) {
 		rangesync.MustParseHexKeyBytes("5555555555555555555555555555555555555555555555555555555555555555"),
 		rangesync.MustParseHexKeyBytes("8888888888888888888888888888888888888888888888888888888888888888"),
 	}
-	db := sqlstore.PopulateDB(t, testKeyLen, ids)
+
 	st := &sqlstore.SyncedTable{
 		TableName: "foo",
 		IDColumn:  "id",
 	}
-	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
-	require.NoError(t, s.EnsureLoaded())
 
-	copy := s.Copy(false)
+	verifyDS := func(db sql.Database, os rangesync.OrderedSet) {
+		require.NoError(t, os.EnsureLoaded())
 
-	info, err := s.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 4, info.Count)
-	require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+		require.NoError(t, os.WithCopy(context.Background(), func(copy rangesync.OrderedSet) error {
+			info, err := os.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 4, info.Count)
+			require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	info, err = copy.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 4, info.Count)
-	require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			info, err = copy.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 4, info.Count)
+			require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	sqlstore.InsertDBItems(t, db, []rangesync.KeyBytes{
-		rangesync.MustParseHexKeyBytes("abcdef1234567890000000000000000000000000000000000000000000000000"),
-	})
+			sqlstore.InsertDBItems(t, db, []rangesync.KeyBytes{
+				rangesync.MustParseHexKeyBytes(
+					"abcdef1234567890000000000000000000000000000000000000000000000000"),
+			})
 
-	info, err = s.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 4, info.Count)
-	require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			info, err = os.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 4, info.Count)
+			require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	info, err = copy.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 4, info.Count)
-	require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			info, err = copy.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 4, info.Count)
+			require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	require.NoError(t, s.Advance())
+			require.NoError(t, os.Advance())
 
-	info, err = s.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 5, info.Count)
-	require.Equal(t, "642464b773377bbddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			info, err = os.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 5, info.Count)
+			require.Equal(t, "642464b773377bbddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	info, err = copy.GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 4, info.Count)
-	require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			info, err = copy.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 4, info.Count)
+			require.Equal(t, "cfe98ba54761032ddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
 
-	info, err = s.Copy(false).GetRangeInfo(ids[0], ids[0])
-	require.NoError(t, err)
-	require.Equal(t, 5, info.Count)
-	require.Equal(t, "642464b773377bbddddddddd", info.Fingerprint.String())
-	require.Equal(t, ids[0], firstKey(t, info.Items))
+			return nil
+		}))
+
+		require.NoError(t, os.WithCopy(context.Background(), func(copy rangesync.OrderedSet) error {
+			info, err := copy.GetRangeInfo(ids[0], ids[0])
+			require.NoError(t, err)
+			require.Equal(t, 5, info.Count)
+			require.Equal(t, "642464b773377bbddddddddd", info.Fingerprint.String())
+			require.Equal(t, ids[0], firstKey(t, info.Items))
+			return nil
+		}))
+	}
+
+	t.Run("original DBSet", func(t *testing.T) {
+		db := sqlstore.PopulateDB(t, testKeyLen, ids)
+		dbSet := dbset.NewDBSet(db, st, testKeyLen, testDepth)
+		verifyDS(db, dbSet)
+	})
+
+	t.Run("DBSet copy", func(t *testing.T) {
+		db := sqlstore.PopulateDB(t, testKeyLen, ids)
+		origSet := dbset.NewDBSet(db, st, testKeyLen, testDepth)
+		require.NoError(t, origSet.WithCopy(context.Background(), func(copy rangesync.OrderedSet) error {
+			verifyDS(db, copy)
+			return nil
+		}))
+	})
 }
 
 func TestDBSet_Added(t *testing.T) {
@@ -333,7 +356,6 @@ func TestDBSet_Added(t *testing.T) {
 		IDColumn:  "id",
 	}
 	s := dbset.NewDBSet(db, st, testKeyLen, testDepth)
-	defer s.Release()
 	requireEmpty(t, s.Received())
 
 	add := []rangesync.KeyBytes{
@@ -353,7 +375,10 @@ func TestDBSet_Added(t *testing.T) {
 		rangesync.MustParseHexKeyBytes("4444444444444444444444444444444444444444444444444444444444444444"),
 	}, added)
 
-	added1, err := s.Copy(false).(*dbset.DBSet).Received().FirstN(3)
-	require.NoError(t, err)
-	require.ElementsMatch(t, added, added1)
+	require.NoError(t, s.WithCopy(context.Background(), func(copy rangesync.OrderedSet) error {
+		added1, err := copy.(*dbset.DBSet).Received().FirstN(3)
+		require.NoError(t, err)
+		require.ElementsMatch(t, added, added1)
+		return nil
+	}))
 }