Add reproduction test and fix the issue

Signed-off-by: Jakub Sztandera <kubuxu@protocol.ai>

chain/sync_manager.go

@@ -3,6 +3,7 @@ package chain
 import (
 	"context"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -153,6 +154,19 @@ func newSyncTargetBucket(tipsets ...*types.TipSet) *syncTargetBucket {
 	return &stb
 }
 
+func (sbs *syncBucketSet) String() string {
+	var bStrings []string
+	for _, b := range sbs.buckets {
+		var tsStrings []string
+		for _, t := range b.tips {
+			tsStrings = append(tsStrings, t.String())
+		}
+		bStrings = append(bStrings, "["+strings.Join(tsStrings, ",")+"]")
+	}
+
+	return "{" + strings.Join(bStrings, ";") + "}"
+}
+
 func (sbs *syncBucketSet) RelatedToAny(ts *types.TipSet) bool {
 	for _, b := range sbs.buckets {
 		if b.sameChainAs(ts) {
@@ -325,7 +339,7 @@ func (sm *syncManager) syncScheduler() {
 				return string(activeSyncs[i].Bytes()) < string(activeSyncs[j].Bytes())
 			})
 
-			log.Infof("activeSyncs: %v, ", activeSyncs)
+			log.Infof("activeSyncs: %v, activeSyncTips: %s ", activeSyncs, sm.activeSyncTips.String())
 		}
 	}
 }
@@ -341,7 +355,8 @@ func (sm *syncManager) scheduleIncoming(ts *types.TipSet) {
 	var relatedToActiveSync bool
 	for _, acts := range sm.activeSyncs {
 		if ts.Equals(acts) {
-			break
+			// ignore, we are already syncing it
+			return
 		}
 
 		if ts.Parents() == acts.Key() {

chain/sync_manager_test.go

@@ -67,6 +67,67 @@ func assertGetSyncOp(t *testing.T, c chan *syncOp, ts *types.TipSet) {
 	}
 }
 
+func TestSyncManagerEdgeCase(t *testing.T) {
+	ctx := context.Background()
+
+	a := mock.TipSet(mock.MkBlock(genTs, 1, 1))
+	t.Logf("a: %s", a)
+	b1 := mock.TipSet(mock.MkBlock(a, 1, 2))
+	t.Logf("b1: %s", b1)
+	b2 := mock.TipSet(mock.MkBlock(a, 2, 3))
+	t.Logf("b2: %s", b2)
+	c1 := mock.TipSet(mock.MkBlock(b1, 2, 4))
+	t.Logf("c1: %s", c1)
+	c2 := mock.TipSet(mock.MkBlock(b2, 1, 5))
+	t.Logf("c2: %s", c2)
+
+	runSyncMgrTest(t, "edgeCase", 1, func(t *testing.T, sm *syncManager, stc chan *syncOp) {
+		sm.SetPeerHead(ctx, "peer1", a)
+		assertGetSyncOp(t, stc, a)
+		time.Sleep(10 * time.Millisecond)
+		t.Logf("bootstate: %d", sm.bootstrapState)
+
+		sm.SetPeerHead(ctx, "peer1", b1)
+		sm.SetPeerHead(ctx, "peer1", b2)
+		// b1 and b2 are being processed
+
+		b1op := <-stc
+		b2op := <-stc
+		if !b1op.ts.Equals(b1) {
+			b1op, b2op = b2op, b1op
+		}
+		t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
+
+		sm.SetPeerHead(ctx, "peer2", c2) // c2 is put into activeSyncTips at index 0
+		sm.SetPeerHead(ctx, "peer2", c1) // c1 is put into activeSyncTips at index 1
+		sm.SetPeerHead(ctx, "peer3", b2) // b2 is related to c2 and even though it is actively synced it is put into activeSyncTips index 0
+		sm.SetPeerHead(ctx, "peer1", a)  // a is related to b2 and is put into activeSyncTips index 0
+
+		b1op.done() // b1 completes first, is related to a, so it pops activeSyncTips index 0
+		// even though correct one is index 1
+
+		time.Sleep(100 * time.Millisecond)
+		t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
+		b2op.done()
+		// b2 completes and is not related to c1, so it leaves activeSyncTips as it is
+		for i := 0; i < 10; {
+			select {
+			case so := <-stc:
+				so.done()
+			default:
+				i++
+				time.Sleep(10 * time.Millisecond)
+			}
+		}
+
+		time.Sleep(10 * time.Millisecond)
+		if len(sm.activeSyncTips.buckets) != 0 {
+			t.Errorf("activeSyncTips expected empty but got: %s", sm.activeSyncTips.String())
+		}
+		t.Logf("activeSyncs: %s: activeSyncTips: %s", sm.activeSyncs, sm.activeSyncTips.String())
+	})
+}
+
 func TestSyncManager(t *testing.T) {
 	ctx := context.Background()