[Merged by Bors] - sync2: fix Advance race

sync2/multipeer/interface.go

@@ -16,6 +16,9 @@ import (
 type SyncBase interface {
 	// Count returns the number of items in the set.
 	Count() (int, error)
+	// Advance advances the underlying OrderedSet, loading the items since the last Advance call
+	// or when OrderedSet was first loaded.
+	Advance() error
 	// Sync synchronizes the set with the peer.
 	// It returns a sequence of new keys that were received from the peer and the
 	// number of received items.

sync2/multipeer/setsyncbase.go

@@ -38,7 +38,8 @@ func NewSetSyncBase(
 
 // Count implements SyncBase.
 func (ssb *SetSyncBase) Count() (int, error) {
-	// TODO: don't lock on potentially db-bound operations
+	// In most cases ssb.os.SetInfo will not access the database, so we're not holding
+	// the lock for long here.
 	ssb.mtx.Lock()
 	defer ssb.mtx.Unlock()
 	info, err := ssb.os.SetInfo()
@@ -48,6 +49,13 @@ func (ssb *SetSyncBase) Count() (int, error) {
 	return info.Count, nil
 }
 
+// Advance implements SyncBase.
+func (ssb *SetSyncBase) Advance() error {
+	ssb.mtx.Lock()
+	defer ssb.mtx.Unlock()
+	return ssb.os.Advance()
+}
+
 func (ssb *SetSyncBase) syncPeer(
 	ctx context.Context,
 	p p2p.Peer,

sync2/p2p.go

@@ -168,7 +168,7 @@ func (s *P2PHashSync) start() (isWaiting bool) {
 					return ctx.Err()
 				case <-ticker.C:
 					s.logger.Debug("advancing OrderedSet on timer")
-					if err := s.os.Advance(); err != nil {
+					if err := s.syncBase.Advance(); err != nil {
 						s.logger.Error("error advancing the set", zap.Error(err))
 					}
 				}