sync: hash resolution use atx cache (#5119)

## Motivation
Closes #5078

## Changes
use atx cache during hash resolution

syncer/state_syncer.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	"go.uber.org/zap/zapcore"
-	"golang.org/x/exp/maps"
 
 	"github.com/spacemeshos/go-spacemesh/common/types"
 	"github.com/spacemeshos/go-spacemesh/fetch"
@@ -284,35 +283,24 @@ func (s *Syncer) ensureMeshAgreement(
 			)
 			continue
 		}
-		missing := make(map[types.ATXID]struct{})
-		for _, id := range ed.AtxIDs {
-			if _, ok := missing[id]; ok {
-				continue
-			}
-			hdr, _ := s.cdb.GetAtxHeader(id)
-			if hdr == nil {
-				missing[id] = struct{}{}
-				continue
-			}
-		}
+		missing := s.asCache.GetMissingActiveSet(diffLayer.GetEpoch(), ed.AtxIDs)
 		if len(missing) > 0 {
-			toFetch := maps.Keys(missing)
 			s.logger.WithContext(ctx).With().Debug("fetching missing atxs from peer",
 				log.Stringer("peer", peer),
 				log.Array("missing_atxs", log.ArrayMarshalerFunc(func(encoder zapcore.ArrayEncoder) error {
-					for _, id := range toFetch {
+					for _, id := range missing {
 						encoder.AppendString(id.ShortString())
 					}
 					return nil
 				})),
 			)
 			// node and peer has different state. check if peer has valid ATXs to back up its opinions
-			if err = s.dataFetcher.GetAtxs(ctx, toFetch); err != nil {
+			if err = s.dataFetcher.GetAtxs(ctx, missing); err != nil {
 				// if the node cannot download the ATXs claimed by this peer, it does not trust this peer's mesh
 				s.logger.WithContext(ctx).With().Warning("failed to download missing ATX claimed by peer",
 					log.Stringer("peer", peer),
 					log.Array("missing_atxs", log.ArrayMarshalerFunc(func(encoder zapcore.ArrayEncoder) error {
-						for _, id := range toFetch {
+						for _, id := range missing {
 							encoder.AppendString(id.ShortString())
 						}
 						return nil

syncer/state_syncer_test.go

@@ -345,6 +345,9 @@ func TestProcessLayers_MeshHashDiverged(t *testing.T) {
 			ts.mForkFinder.EXPECT().NeedResync(instate.Sub(1), opns[i].PrevAggHash).Return(false)
 		} else {
 			ts.mForkFinder.EXPECT().NeedResync(instate.Sub(1), opns[i].PrevAggHash).Return(true)
+			if i != 4 {
+				ts.mAtxCache.EXPECT().GetMissingActiveSet(epoch, eds[i].AtxIDs).Return(eds[i].AtxIDs)
+			}
 		}
 	}
 

syncer/syncer.go

@@ -112,6 +112,7 @@ type Syncer struct {
 
 	cfg          Config
 	cdb          *datastore.CachedDB
+	asCache      activeSetCache
 	ticker       layerTicker
 	beacon       system.BeaconGetter
 	mesh         *mesh.Mesh
@@ -152,6 +153,7 @@ func NewSyncer(
 		logger:           log.NewNop(),
 		cfg:              DefaultConfig(),
 		cdb:              cdb,
+		asCache:          cache,
 		ticker:           ticker,
 		beacon:           beacon,
 		mesh:             mesh,

syncer/syncer_test.go

@@ -71,6 +71,7 @@ type testSyncer struct {
 	mTortoise    *smocks.MockTortoise
 	mCertHdr     *mocks.MockcertHandler
 	mForkFinder  *mocks.MockforkFinder
+	mAtxCache    *mocks.MockactiveSetCache
 }
 
 func newTestSyncer(t *testing.T, interval time.Duration) *testSyncer {
@@ -87,6 +88,7 @@ func newTestSyncer(t *testing.T, interval time.Duration) *testSyncer {
 		mTortoise:    smocks.NewMockTortoise(ctrl),
 		mCertHdr:     mocks.NewMockcertHandler(ctrl),
 		mForkFinder:  mocks.NewMockforkFinder(ctrl),
+		mAtxCache:    mocks.NewMockactiveSetCache(ctrl),
 	}
 	ts.cdb = datastore.NewCachedDB(sql.InMemory(), lg)
 	var err error
@@ -102,7 +104,7 @@ func newTestSyncer(t *testing.T, interval time.Duration) *testSyncer {
 		HareDelayLayers:          5,
 		OutOfSyncThresholdLayers: outOfSyncThreshold,
 	}
-	ts.syncer = NewSyncer(ts.cdb, ts.mTicker, ts.mBeacon, ts.msh, nil, nil, ts.mLyrPatrol, ts.mCertHdr,
+	ts.syncer = NewSyncer(ts.cdb, ts.mTicker, ts.mBeacon, ts.msh, ts.mAtxCache, nil, ts.mLyrPatrol, ts.mCertHdr,
 		WithConfig(cfg),
 		WithLogger(lg),
 		withDataFetcher(ts.mDataFetcher),