[Merged by Bors] - sync2: ATX integration

config/mainnet.go

@@ -24,6 +24,7 @@ import (
 	"github.com/spacemeshos/go-spacemesh/hare4"
 	"github.com/spacemeshos/go-spacemesh/miner"
 	"github.com/spacemeshos/go-spacemesh/p2p"
+	"github.com/spacemeshos/go-spacemesh/sync2"
 	"github.com/spacemeshos/go-spacemesh/syncer"
 	"github.com/spacemeshos/go-spacemesh/syncer/atxsync"
 	"github.com/spacemeshos/go-spacemesh/syncer/malsync"
@@ -77,6 +78,14 @@ func MainnetConfig() Config {
 
 	hare4conf := hare4.DefaultConfig()
 	hare4conf.Enable = false
+
+	oldAtxSyncCfg := sync2.DefaultConfig()
+	oldAtxSyncCfg.MultiPeerReconcilerConfig.SyncInterval = time.Hour
+	oldAtxSyncCfg.MaxDepth = 16
+	newAtxSyncCfg := sync2.DefaultConfig()
+	newAtxSyncCfg.MaxDepth = 21
+	newAtxSyncCfg.MultiPeerReconcilerConfig.SyncInterval = 5 * time.Minute
+
 	return Config{
 		BaseConfig: BaseConfig{
 			DataDirParent:         defaultDataDir,
@@ -212,6 +221,17 @@ func MainnetConfig() Config {
 			DisableMeshAgreement:     true,
 			AtxSync:                  atxsync.DefaultConfig(),
 			MalSync:                  malsync.DefaultConfig(),
+			ReconcSync: syncer.ReconcSyncConfig{
+				OldAtxSyncCfg:     oldAtxSyncCfg,
+				NewAtxSyncCfg:     newAtxSyncCfg,
+				ParallelLoadLimit: 10,
+				HardTimeout:       10 * time.Minute,
+				ServerConfig: fetch.ServerConfig{
+					Queue:    200,
+					Requests: 100,
+					Interval: time.Second,
+				},
+			},
 		},
 		Recovery: checkpoint.DefaultConfig(),
 		Cache:    datastore.DefaultConfig(),

config/presets/testnet.go

@@ -25,6 +25,7 @@ import (
 	"github.com/spacemeshos/go-spacemesh/hare4"
 	"github.com/spacemeshos/go-spacemesh/miner"
 	"github.com/spacemeshos/go-spacemesh/p2p"
+	"github.com/spacemeshos/go-spacemesh/sync2"
 	"github.com/spacemeshos/go-spacemesh/syncer"
 	"github.com/spacemeshos/go-spacemesh/syncer/atxsync"
 	"github.com/spacemeshos/go-spacemesh/syncer/malsync"
@@ -65,6 +66,13 @@ func testnet() config.Config {
 	hare4conf := hare4.DefaultConfig()
 	hare4conf.Enable = false
 	defaultdir := filepath.Join(home, "spacemesh-testnet", "/")
+
+	oldAtxSyncCfg := sync2.DefaultConfig()
+	oldAtxSyncCfg.MaxDepth = 16
+	newAtxSyncCfg := sync2.DefaultConfig()
+	newAtxSyncCfg.MaxDepth = 21
+	newAtxSyncCfg.MultiPeerReconcilerConfig.SyncInterval = 5 * time.Minute
+
 	return config.Config{
 		Preset: "testnet",
 		BaseConfig: config.BaseConfig{
@@ -163,6 +171,17 @@ func testnet() config.Config {
 			OutOfSyncThresholdLayers: 10,
 			AtxSync:                  atxsync.DefaultConfig(),
 			MalSync:                  malsync.DefaultConfig(),
+			ReconcSync: syncer.ReconcSyncConfig{
+				OldAtxSyncCfg:     oldAtxSyncCfg,
+				NewAtxSyncCfg:     newAtxSyncCfg,
+				ParallelLoadLimit: 10,
+				HardTimeout:       time.Minute,
+				ServerConfig: fetch.ServerConfig{
+					Queue:    200,
+					Requests: 100,
+					Interval: time.Second,
+				},
+			},
 		},
 		Recovery: checkpoint.DefaultConfig(),
 		Cache:    datastore.DefaultConfig(),

fetch/fetch.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 	"time"
 
+	corehost "github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/protocol"
 	"go.uber.org/zap"
@@ -116,7 +117,7 @@ type ServerConfig struct {
 	Interval time.Duration `mapstructure:"interval"`
 }
 
-func (s ServerConfig) toOpts() []server.Opt {
+func (s ServerConfig) ToOpts() []server.Opt {
 	opts := []server.Opt{}
 	if s.Queue != 0 {
 		opts = append(opts, server.WithQueueSize(s.Queue))
@@ -375,7 +376,7 @@ func (f *Fetch) registerServer(
 	if f.cfg.EnableServerMetrics {
 		opts = append(opts, server.WithMetrics())
 	}
-	opts = append(opts, f.cfg.getServerConfig(protocol).toOpts()...)
+	opts = append(opts, f.cfg.getServerConfig(protocol).ToOpts()...)
 	f.servers[protocol] = server.New(host, protocol, handler, opts...)
 }
 
@@ -1025,3 +1026,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/limiter.go

@@ -2,6 +2,8 @@ package fetch
 
 import (
 	"context"
+
+	"github.com/spacemeshos/go-spacemesh/common/types"
 )
 
 type limiter interface {
@@ -10,7 +12,8 @@ type limiter interface {
 }
 
 type getHashesOpts struct {
-	limiter limiter
+	limiter  limiter
+	callback func(types.Hash32, error)
 }
 
 type noLimit struct{}

fetch/mesh_data.go

@@ -31,7 +31,7 @@ func (f *Fetch) GetAtxs(ctx context.Context, ids []types.ATXID, opts ...system.G
 		return nil
 	}
 
-	options := system.GetAtxOpts{}
+	var options system.GetAtxOpts
 	for _, opt := range opts {
 		opt(&options)
 	}
@@ -42,10 +42,17 @@ func (f *Fetch) GetAtxs(ctx context.Context, ids []types.ATXID, opts ...system.G
 		zap.Bool("limiting", !options.LimitingOff),
 	)
 	hashes := types.ATXIDsToHashes(ids)
-	if options.LimitingOff {
-		return f.getHashes(ctx, hashes, datastore.ATXDB, f.validators.atx.HandleMessage)
+	handler := f.validators.atx.HandleMessage
+	var ghOpts []getHashesOpt
+	if !options.LimitingOff {
+		ghOpts = append(ghOpts, withLimiter(f.getAtxsLimiter))
 	}
-	return f.getHashes(ctx, hashes, datastore.ATXDB, f.validators.atx.HandleMessage, withLimiter(f.getAtxsLimiter))
+	if options.Callback != nil {
+		ghOpts = append(ghOpts, withHashCallback(func(hash types.Hash32, err error) {
+			options.Callback(types.ATXID(hash), err)
+		}))
+	}
+	return f.getHashes(ctx, hashes, datastore.ATXDB, handler, ghOpts...)
 }
 
 type dataReceiver func(context.Context, types.Hash32, p2p.Peer, []byte) error
@@ -58,6 +65,12 @@ func withLimiter(l limiter) getHashesOpt {
 	}
 }
 
+func withHashCallback(callback func(types.Hash32, error)) getHashesOpt {
+	return func(o *getHashesOpts) {
+		o.callback = callback
+	}
+}
+
 func (f *Fetch) getHashes(
 	ctx context.Context,
 	hashes []types.Hash32,
@@ -66,7 +79,8 @@ func (f *Fetch) getHashes(
 	opts ...getHashesOpt,
 ) error {
 	options := getHashesOpts{
-		limiter: noLimit{},
+		limiter:  noLimit{},
+		callback: func(types.Hash32, error) {},
 	}
 	for _, opt := range opts {
 		opt(&options)
@@ -83,18 +97,26 @@ func (f *Fetch) getHashes(
 	for i, hash := range hashes {
 		if err := options.limiter.Acquire(ctx, 1); err != nil {
 			pendingMetric.Add(float64(i - len(hashes)))
-			return fmt.Errorf("acquiring slot to get hash: %w", err)
+			err = fmt.Errorf("acquiring slot to get hash: %w", err)
+			for _, h := range hashes[i:] {
+				options.callback(h, err)
+			}
+			return err
 		}
 		p, err := f.getHash(ctx, hash, hint, receiver)
 		if err != nil {
 			options.limiter.Release(1)
 			pendingMetric.Add(float64(i - len(hashes)))
+			for _, h := range hashes[i:] {
+				options.callback(h, err)
+			}
 			return err
 		}
 		if p == nil {
 			// data is available locally
 			options.limiter.Release(1)
 			pendingMetric.Add(-1)
+			options.callback(hash, nil)
 			continue
 		}
 
@@ -103,6 +125,7 @@ func (f *Fetch) getHashes(
 			case <-ctx.Done():
 				options.limiter.Release(1)
 				pendingMetric.Add(-1)
+				options.callback(hash, ctx.Err())
 				return ctx.Err()
 			case <-p.completed:
 				options.limiter.Release(1)
@@ -118,6 +141,7 @@ func (f *Fetch) getHashes(
 					bfailure.Add(hash, p.err)
 					mu.Unlock()
 				}
+				options.callback(hash, p.err)
 				return nil
 			}
 		})

fetch/mesh_data_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"sync"
 	"testing"
 
 	p2phost "github.com/libp2p/go-libp2p/core/host"
@@ -87,7 +88,7 @@ func startTestLoop(tb testing.TB, f *Fetch, eg *errgroup.Group, stop chan struct
 			default:
 				f.mu.Lock()
 				for h, req := range f.unprocessed {
-					require.NoError(tb, req.validator(req.ctx, types.Hash32{}, p2p.NoPeer, []byte{}))
+					require.NoError(tb, req.validator(req.ctx, h, p2p.NoPeer, []byte{}))
 					close(req.promise.completed)
 					delete(f.unprocessed, h)
 				}
@@ -596,7 +597,7 @@ func genATXs(tb testing.TB, num uint32) []*types.ActivationTx {
 }
 
 func TestGetATXs(t *testing.T) {
-	atxs := genATXs(t, 2)
+	atxs := genATXs(t, 4)
 	f := createFetch(t)
 	f.mAtxH.EXPECT().
 		HandleMessage(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
@@ -607,10 +608,22 @@ func TestGetATXs(t *testing.T) {
 	var eg errgroup.Group
 	startTestLoop(t, f.Fetch, &eg, stop)
 
-	atxIDs := types.ToATXIDs(atxs)
-	require.NoError(t, f.GetAtxs(context.Background(), atxIDs))
+	atxIDs1 := types.ToATXIDs(atxs[:2])
+	require.NoError(t, f.GetAtxs(context.Background(), atxIDs1))
+
+	atxIDs2 := types.ToATXIDs(atxs[2:])
+	var recvIDs []types.ATXID
+	var mtx sync.Mutex
+	require.NoError(t, f.GetAtxs(context.Background(), atxIDs2,
+		system.WithATXCallback(func(id types.ATXID, err error) {
+			mtx.Lock()
+			defer mtx.Unlock()
+			require.NoError(t, err)
+			recvIDs = append(recvIDs, id)
+		})))
 	close(stop)
 	require.NoError(t, eg.Wait())
+	require.ElementsMatch(t, atxIDs2, recvIDs)
 }
 
 func TestGetActiveSet(t *testing.T) {