fix puller test, add cleanups to kademlia (#397)

pkg/kademlia/kademlia_test.go

@@ -260,6 +260,7 @@ func TestNotifierHooks(t *testing.T) {
 		peer                     = test.RandomAddressAt(base, 3)
 		addr                     = test.RandomAddressAt(peer, 4) // address which is closer to peer
 	)
+	defer kad.Close()
 
 	connectOne(t, signer, kad, ab, peer)
 
@@ -289,6 +290,7 @@ func TestDiscoveryHooks(t *testing.T) {
 		_, kad, ab, disc, signer = newTestKademlia(&conns, nil, nil)
 		p1, p2, p3               = test.RandomAddress(), test.RandomAddress(), test.RandomAddress()
 	)
+	defer kad.Close()
 
 	// first add a peer from AddPeer, wait for the connection
 	addOne(t, signer, kad, ab, p1)
@@ -320,9 +322,9 @@ func TestBackoff(t *testing.T) {
 
 	var (
 		conns                    int32 // how many connect calls were made to the p2p mock
-
 		base, kad, ab, _, signer = newTestKademlia(&conns, nil, nil)
 	)
+	defer kad.Close()
 
 	// add one peer, wait for connection
 	addr := test.RandomAddressAt(base, 1)
@@ -361,6 +363,7 @@ func TestAddressBookPrune(t *testing.T) {
 		conns, failedConns       int32 // how many connect calls were made to the p2p mock
 		base, kad, ab, _, signer = newTestKademlia(&conns, &failedConns, nil)
 	)
+	defer kad.Close()
 
 	nonConnPeer, err := bzz.NewAddress(signer, nonConnectableAddress, test.RandomAddressAt(base, 1), 0)
 	if err != nil {
@@ -525,6 +528,7 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 
 	t.Run("single subscription", func(t *testing.T) {
 		base, kad, ab, _, sg := newTestKademlia(nil, nil, nil)
+		defer kad.Close()
 
 		c, u := kad.SubscribePeersChange()
 		defer u()
@@ -537,6 +541,7 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 
 	t.Run("single subscription, remove peer", func(t *testing.T) {
 		base, kad, ab, _, sg := newTestKademlia(nil, nil, nil)
+		defer kad.Close()
 
 		c, u := kad.SubscribePeersChange()
 		defer u()
@@ -552,6 +557,7 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 
 	t.Run("multiple subscriptions", func(t *testing.T) {
 		base, kad, ab, _, sg := newTestKademlia(nil, nil, nil)
+		defer kad.Close()
 
 		c1, u1 := kad.SubscribePeersChange()
 		defer u1()
@@ -569,6 +575,7 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 
 	t.Run("multiple changes", func(t *testing.T) {
 		base, kad, ab, _, sg := newTestKademlia(nil, nil, nil)
+		defer kad.Close()
 
 		c, u := kad.SubscribePeersChange()
 		defer u()
@@ -590,6 +597,7 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 
 	t.Run("no depth change", func(t *testing.T) {
 		_, kad, _, _, _ := newTestKademlia(nil, nil, nil)
+		defer kad.Close()
 
 		c, u := kad.SubscribePeersChange()
 		defer u()
@@ -607,9 +615,9 @@ func TestKademlia_SubscribePeersChange(t *testing.T) {
 }
 
 func TestMarshal(t *testing.T) {
-	var (
-		_, kad, ab, _, signer = newTestKademlia(nil, nil, nil)
-	)
+	_, kad, ab, _, signer := newTestKademlia(nil, nil, nil)
+	defer kad.Close()
+
 	a := test.RandomAddress()
 	addOne(t, signer, kad, ab, a)
 	_, err := kad.MarshalJSON()

pkg/puller/puller_test.go

@@ -201,7 +201,7 @@ func TestSyncFlow_PeerOutsideDepth_Historical(t *testing.T) {
 			kad.Trigger()
 
 			waitCursorsCalled(t, pullsync, addr, false)
-			waitSyncCalled(t, pullsync, addr, false)
+			waitSyncCalledTimes(t, pullsync, addr, len(tc.expCalls))
 
 			// check historical sync calls
 			checkCalls(t, tc.expCalls, pullsync.SyncCalls(addr))
@@ -479,7 +479,7 @@ func checkCallsUnordered(t *testing.T, expCalls []c, calls []mockps.SyncCall) {
 	}
 }
 
-// waitSyncCalled waits until SyncInterval is called on the address given.
+// waitCursorsCalled waits until GetCursors are called on the given address.
 func waitCursorsCalled(t *testing.T, ps *mockps.PullSyncMock, addr swarm.Address, invert bool) {
 	t.Helper()
 	for i := 0; i < 20; i++ {
@@ -541,6 +541,19 @@ func waitSyncCalled(t *testing.T, ps *mockps.PullSyncMock, addr swarm.Address, i
 	t.Fatal("timed out waiting for sync")
 }
 
+func waitSyncCalledTimes(t *testing.T, ps *mockps.PullSyncMock, addr swarm.Address, times int) {
+	t.Helper()
+	for i := 0; i < 15; i++ {
+		v := ps.SyncCalls(addr)
+		if len(v) == times {
+			return
+		}
+
+		time.Sleep(50 * time.Millisecond)
+	}
+	t.Fatal("timed out waiting for sync")
+}
+
 type opts struct {
 	pullSync []mockps.Option
 	kad      []mockk.Option