Implement set reconciliation based sync (syncv2)

common/types/hashes.go

@@ -1,6 +1,7 @@
 package types
 
 import (
+	"bytes"
 	"encoding/hex"
 	"fmt"
 	"reflect"
@@ -9,11 +10,13 @@ import (
 
 	"github.com/spacemeshos/go-spacemesh/common/util"
 	"github.com/spacemeshos/go-spacemesh/hash"
+	"github.com/spacemeshos/go-spacemesh/log"
 )
 
 const (
 	Hash32Length = 32
 	Hash20Length = 20
+	Hash12Length = 12
 )
 
 var (
@@ -30,6 +33,9 @@ type Hash32 [Hash32Length]byte
 // Hash20 represents the 20-byte blake3 hash of arbitrary data.
 type Hash20 [Hash20Length]byte
 
+// Hash12 represents the 12-byte hash used for sync
+type Hash12 [Hash12Length]byte
+
 // Bytes gets the byte representation of the underlying hash.
 func (h Hash20) Bytes() []byte { return h[:] }
 
@@ -88,6 +94,15 @@ func (h Hash20) ToHash32() (h32 Hash32) {
 	return
 }
 
+// String implements the stringer interface and is used also by the logger when
+// doing full logging into a file.
+func (h Hash12) String() string {
+	return util.Encode(h[:5])
+}
+
+// Field returns a log field. Implements the LoggableField interface.
+func (h Hash12) Field() log.Field { return log.String("hash", hex.EncodeToString(h[:])) }
+
 // CalcProposalsHash32 returns the 32-byte blake3 sum of the IDs, sorted in lexicographic order. The pre-image is
 // prefixed with additionalBytes.
 func CalcProposalsHash32(view []ProposalID, additionalBytes []byte) Hash32 {
@@ -160,6 +175,16 @@ func (h Hash32) ShortString() string {
 	return hex.EncodeToString(h[:5])
 }
 
+// Compare compares a Hash32 to another hash and returns
+//
+//	-1 if this hash is less than the other
+//	 0 if the hashes are equal
+//	 1 if this hash is greater than the other
+func (h Hash32) Compare(other any) int {
+	oh := other.(Hash32)
+	return bytes.Compare(h[:], oh[:])
+}
+
 // Format implements fmt.Formatter, forcing the byte slice to be formatted as is,
 // without going through the stringer interface used for logging.
 func (h Hash32) Format(s fmt.State, c rune) {

fetch/fetch.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	corehost "github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
@@ -1006,3 +1007,11 @@ func (f *Fetch) SelectBestShuffled(n int) []p2p.Peer {
 	})
 	return peers
 }
+
+func (f *Fetch) Host() corehost.Host {
+	return f.host.(corehost.Host)
+}
+
+func (f *Fetch) Peers() *peers.Peers {
+	return f.peers
+}

fetch/peers/peers.go

@@ -54,6 +54,13 @@ type Peers struct {
 	globalLatency float64
 }
 
+func (p *Peers) Contains(id peer.ID) bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	_, exist := p.peers[id]
+	return exist
+}
+
 func (p *Peers) Add(id peer.ID) bool {
 	p.mu.Lock()
 	defer p.mu.Unlock()

go.mod

@@ -1,6 +1,6 @@
 module github.com/spacemeshos/go-spacemesh
 
-go 1.22.4
+go 1.23.0
 
 require (
 	cloud.google.com/go/storage v1.43.0
@@ -35,6 +35,7 @@ require (
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.55.0
 	github.com/quic-go/quic-go v0.46.0
+	github.com/rqlite/sql v0.0.0-20240312185922-ffac88a740bd
 	github.com/rs/cors v1.11.0
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/seehuhn/mt19937 v1.0.0

go.sum

@@ -152,6 +152,8 @@ github.com/go-openapi/swag v0.22.4/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
+github.com/go-test/deep v1.0.8 h1:TDsG77qcSprGbC6vTN8OuXp5g+J+b5Pcguhf7Zt61VM=
+github.com/go-test/deep v1.0.8/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE=
 github.com/go-yaml/yaml v2.1.0+incompatible/go.mod h1:w2MrLa16VYP0jy6N7M5kHaCkaLENm+P+Tv+MfurjSw0=
 github.com/godbus/dbus/v5 v5.0.3/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
@@ -551,6 +553,8 @@ github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzG
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
+github.com/rqlite/sql v0.0.0-20240312185922-ffac88a740bd h1:wW6BtayFoKaaDeIvXRE3SZVPOscSKlYD+X3bB749+zk=
+github.com/rqlite/sql v0.0.0-20240312185922-ffac88a740bd/go.mod h1:ib9zVtNgRKiGuoMyUqqL5aNpk+r+++YlyiVIkclVqPg=
 github.com/rs/cors v1.11.0 h1:0B9GE/r9Bc2UxRMMtymBkHTenPkHDv0CW4Y98GBY+po=
 github.com/rs/cors v1.11.0/go.mod h1:XyqrcTp5zjWr1wsJ8PIRZssZ8b/WMcMf71DJnit4EMU=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=

p2p/server/deadline_adjuster.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"sync"
 	"time"
 
 	"github.com/jonboulle/clockwork"
@@ -52,6 +53,8 @@ type deadlineAdjuster struct {
 	nextAdjustRead  int
 	nextAdjustWrite int
 	hardDeadline    time.Time
+	closeErr        error
+	close           sync.Once
 }
 
 var _ io.ReadWriteCloser = &deadlineAdjuster{}
@@ -85,11 +88,14 @@ func (dadj *deadlineAdjuster) augmentError(what string, err error) error {
 	}
 }
 
-// Close closes the stream. This method is safe to call multiple times.
+// Close closes the stream.
 func (dadj *deadlineAdjuster) Close() error {
-	// FIXME: unsure if this is really needed (inherited from the older Server code)
-	_ = dadj.peerStream.SetDeadline(time.Time{})
-	return dadj.peerStream.Close()
+	dadj.close.Do(func() {
+		// FIXME: unsure if this is really needed (inherited from the older Server code)
+		_ = dadj.peerStream.SetDeadline(time.Time{})
+		dadj.closeErr = dadj.peerStream.Close()
+	})
+	return dadj.closeErr
 }
 
 func (dadj *deadlineAdjuster) adjust() error {

p2p/server/interface.go

@@ -13,14 +13,18 @@ import (
 	"github.com/spacemeshos/go-spacemesh/p2p/peerinfo"
 )
 
-//go:generate mockgen -typed -package=mocks -destination=./mocks/mocks.go -source=./interface.go -exclude_interfaces Host
+//go:generate mockgen -typed -package=mocks -destination=./mocks/mocks.go -source=./interface.go -exclude_interfaces Host,PeerInfoHost
 
 // Host is a subset of libp2p Host interface that needs to be implemented to be usable with server.
 type Host interface {
 	SetStreamHandler(protocol.ID, network.StreamHandler)
 	NewStream(context.Context, peer.ID, ...protocol.ID) (network.Stream, error)
 	Network() network.Network
 	ConnManager() connmgr.ConnManager
+}
+
+type PeerInfoHost interface {
+	Host
 	PeerInfo() peerinfo.PeerInfo
 }
 

p2p/server/server.go

@@ -104,12 +104,30 @@ func WithRequestsPerInterval(n int, interval time.Duration) Opt {
 	}
 }
 
+// WithDecayingTag specifies P2P decaying tag that is applied to the peer when a request
+// is being served
 func WithDecayingTag(tag DecayingTagSpec) Opt {
 	return func(s *Server) {
 		s.decayingTagSpec = &tag
 	}
 }
 
+type peerIDKey = struct{}
+
+func withPeerID(ctx context.Context, peerID peer.ID) context.Context {
+	return context.WithValue(ctx, peerIDKey{}, peerID)
+}
+
+// ContextPeerID retrieves the ID of the peer being served from the context and a boolean
+// value indicating that the context contains peer ID. If there's no peer ID associated
+// with the context, the function returns an empty peer ID and false.
+func ContextPeerID(ctx context.Context) (peer.ID, bool) {
+	if v := ctx.Value(peerIDKey{}); v != nil {
+		return v.(peer.ID), true
+	}
+	return peer.ID(""), false
+}
+
 // Handler is a handler to be defined by the application.
 type Handler func(context.Context, []byte) ([]byte, error)
 
@@ -237,6 +255,13 @@ type request struct {
 	received time.Time
 }
 
+func (s *Server) peerInfo() peerinfo.PeerInfo {
+	if h, ok := s.h.(PeerInfoHost); ok {
+		return h.PeerInfo()
+	}
+	return nil
+}
+
 func (s *Server) Run(ctx context.Context) error {
 	var eg errgroup.Group
 	for {
@@ -257,7 +282,8 @@ func (s *Server) Run(ctx context.Context) error {
 				eg.Wait()
 				return nil
 			}
-			ctx, cancel := context.WithCancel(ctx)
+			peer := req.stream.Conn().RemotePeer()
+			ctx, cancel := context.WithCancel(withPeerID(ctx, peer))
 			eg.Go(func() error {
 				<-ctx.Done()
 				s.sem.Release(1)
@@ -268,12 +294,12 @@ func (s *Server) Run(ctx context.Context) error {
 				defer cancel()
 				conn := req.stream.Conn()
 				if s.decayingTag != nil {
-					s.decayingTag.Bump(conn.RemotePeer(), s.decayingTagSpec.Inc)
+					s.decayingTag.Bump(peer, s.decayingTagSpec.Inc)
 				}
 				ok := s.queueHandler(ctx, req.stream)
 				duration := time.Since(req.received)
-				if s.h.PeerInfo() != nil {
-					info := s.h.PeerInfo().EnsurePeerInfo(conn.RemotePeer())
+				if s.peerInfo() != nil {
+					info := s.peerInfo().EnsurePeerInfo(conn.RemotePeer())
 					info.ServerStats.RequestDone(duration, ok)
 				}
 				if s.metrics != nil {
@@ -448,8 +474,8 @@ func (s *Server) streamRequest(
 	if err != nil {
 		return nil, nil, err
 	}
-	if s.h.PeerInfo() != nil {
-		info = s.h.PeerInfo().EnsurePeerInfo(stream.Conn().RemotePeer())
+	if s.peerInfo() != nil {
+		info = s.peerInfo().EnsurePeerInfo(stream.Conn().RemotePeer())
 	}
 	dadj := newDeadlineAdjuster(stream, s.timeout, s.hardTimeout)
 	defer func() {

p2p/server/server_test.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/peer"
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	"github.com/spacemeshos/go-scale/tester"
 	"github.com/stretchr/testify/assert"
@@ -45,8 +46,10 @@ func TestServer(t *testing.T) {
 	request := []byte("test request")
 	testErr := errors.New("test error")
 
-	handler := func(_ context.Context, msg []byte) ([]byte, error) {
-		return msg, nil
+	handler := func(ctx context.Context, msg []byte) ([]byte, error) {
+		peerID, found := ContextPeerID(ctx)
+		require.True(t, found)
+		return append(msg, []byte(peerID)...), nil
 	}
 	errhandler := func(_ context.Context, _ []byte) ([]byte, error) {
 		return nil, testErr
@@ -81,6 +84,9 @@ func TestServer(t *testing.T) {
 		append(opts, WithRequestSizeLimit(limit))...,
 	)
 	ctx, cancel := context.WithCancel(context.Background())
+	noPeerID, found := ContextPeerID(ctx)
+	require.Equal(t, peer.ID(""), noPeerID)
+	require.False(t, found)
 	var eg errgroup.Group
 	eg.Go(func() error {
 		return srv1.Run(ctx)
@@ -109,16 +115,17 @@ func TestServer(t *testing.T) {
 		srvID := mesh.Hosts()[1].ID()
 		response, err := client.Request(ctx, srvID, request)
 		require.NoError(t, err)
-		require.Equal(t, request, response)
+		expResponse := append(request, []byte(mesh.Hosts()[0].ID())...)
+		require.Equal(t, expResponse, response)
 		srvConns := mesh.Hosts()[1].Network().ConnsToPeer(mesh.Hosts()[0].ID())
 		require.NotEmpty(t, srvConns)
 		require.Equal(t, n+1, srv1.NumAcceptedRequests())
 
-		clientInfo := client.h.PeerInfo().EnsurePeerInfo(srvID)
+		clientInfo := client.peerInfo().EnsurePeerInfo(srvID)
 		require.Equal(t, 1, clientInfo.ClientStats.SuccessCount())
 		require.Zero(t, clientInfo.ClientStats.FailureCount())
 
-		serverInfo := srv1.h.PeerInfo().EnsurePeerInfo(mesh.Hosts()[0].ID())
+		serverInfo := srv1.peerInfo().EnsurePeerInfo(mesh.Hosts()[0].ID())
 		require.Eventually(t, func() bool {
 			return serverInfo.ServerStats.SuccessCount() == 1
 		}, 10*time.Second, 10*time.Millisecond)
@@ -129,7 +136,8 @@ func TestServer(t *testing.T) {
 		srvID := mesh.Hosts()[3].ID()
 		response, err := client.Request(ctx, srvID, request)
 		require.NoError(t, err)
-		require.Equal(t, request, response)
+		expResponse := append(request, []byte(mesh.Hosts()[0].ID())...)
+		require.Equal(t, expResponse, response)
 		srvConns := mesh.Hosts()[3].Network().ConnsToPeer(mesh.Hosts()[0].ID())
 		require.NotEmpty(t, srvConns)
 		require.Equal(t, n+1, srv1.NumAcceptedRequests())
@@ -144,11 +152,11 @@ func TestServer(t *testing.T) {
 		require.ErrorContains(t, err, testErr.Error())
 		require.Equal(t, n+1, srv1.NumAcceptedRequests())
 
-		clientInfo := client.h.PeerInfo().EnsurePeerInfo(srvID)
+		clientInfo := client.peerInfo().EnsurePeerInfo(srvID)
 		require.Zero(t, clientInfo.ClientStats.SuccessCount())
 		require.Equal(t, 1, clientInfo.ClientStats.FailureCount())
 
-		serverInfo := srv2.h.PeerInfo().EnsurePeerInfo(mesh.Hosts()[0].ID())
+		serverInfo := srv2.peerInfo().EnsurePeerInfo(mesh.Hosts()[0].ID())
 		require.Eventually(t, func() bool {
 			return serverInfo.ServerStats.FailureCount() == 1
 		}, 10*time.Second, 10*time.Millisecond)

sql/database.go

@@ -11,6 +11,7 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
+	"testing"
 	"time"
 
 	sqlite "github.com/go-llsqlite/crawshaw"
@@ -233,6 +234,15 @@ func InMemory(opts ...Opt) *sqliteDatabase {
 	return db
 }
 
+// InMemoryTest returns an in-mem database for testing and ensures database is closed during `tb.Cleanup`.
+func InMemoryTest(tb testing.TB, opts ...Opt) *sqliteDatabase {
+	// When using empty DB schema, we don't want to check for schema drift due to
+	// "PRAGMA user_version = 0;" in the initial schema retrieved from the DB.
+	db := InMemory(append(opts, WithNoCheckSchemaDrift())...)
+	tb.Cleanup(func() { db.Close() })
+	return db
+}
+
 // Open database with options.
 //
 // Database is opened in WAL mode and pragma synchronous=normal.