removed Startup function from ProviderQueryManager (#741)

* removed Startup function from ProviderQueryManager

Now `providerquerymanager.New` creates a `ProvicerQueryManager` that is already started.

There is no use case for starting PQM at a later time than it is created. Removing the need to call a `Statup` function separately from `New` is more convenient and reduces the opportunity for a problem if calling `Startup` is missed or if called multiple times.

* Remove flaky portion of test - requires synchronization to test sucessive timer delays

Author: gammazero

CHANGELOG.md

@@ -67,6 +67,7 @@ The following emojis are used to highlight certain changes:
 - `routing/http/server`: exposes Prometheus metrics on `prometheus.DefaultRegisterer` and a custom one can be provided via `WithPrometheusRegistry` [#722](https://github.com/ipfs/boxo/pull/722)
 - `gateway`: `NewCacheBlockStore` and `NewCarBackend` will use `prometheus.DefaultRegisterer` when a custom one is not specified via `WithPrometheusRegistry` [#722](https://github.com/ipfs/boxo/pull/722)
 - `filestore`: added opt-in `WithMMapReader` option to `FileManager` to enable memory-mapped file reads [#665](https://github.com/ipfs/boxo/pull/665)
+- `bitswap/routing` `ProviderQueryManager` does not require calling `Startup` separate from `New`. [#741](https://github.com/ipfs/boxo/pull/741)
 
 ### Changed
 

bitswap/client/bitswap_with_sessions_test.go

@@ -134,7 +134,6 @@ func TestCustomProviderQueryManager(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	pqm.Startup()
 	bs := bitswap.New(ctx, a.Adapter, pqm, a.Blockstore,
 		bitswap.WithClientOption(client.WithDefaultProviderQueryManager(false)))
 	a.Exchange.Close() // close old to be sure.

bitswap/client/client.go

@@ -190,7 +190,6 @@ func New(parent context.Context, network bsnet.BitSwapNetwork, providerFinder Pr
 			// Should not be possible to hit this
 			panic(err)
 		}
-		pqm.Startup()
 		bs.pqm = pqm
 	}
 

bitswap/client/internal/session/session_test.go

@@ -357,7 +357,6 @@ func TestSessionFailingToGetFirstBlock(t *testing.T) {
 	for _, block := range blks {
 		cids = append(cids, block.Cid())
 	}
-	startTick := time.Now()
 	_, err := session.GetBlocks(ctx, cids)
 	require.NoError(t, err, "error getting blocks")
 
@@ -389,7 +388,6 @@ func TestSessionFailingToGetFirstBlock(t *testing.T) {
 	case <-ctx.Done():
 		t.Fatal("Did not find more peers")
 	}
-	firstTickLength := time.Since(startTick)
 
 	// Wait for another broadcast to occur
 	select {
@@ -402,7 +400,6 @@ func TestSessionFailingToGetFirstBlock(t *testing.T) {
 	}
 
 	// Wait for another broadcast to occur
-	startTick = time.Now()
 	select {
 	case receivedWantReq := <-fpm.wantReqs:
 		if len(receivedWantReq.cids) < len(cids) {
@@ -412,14 +409,7 @@ func TestSessionFailingToGetFirstBlock(t *testing.T) {
 		t.Fatal("Never rebroadcast want list")
 	}
 
-	// Tick should take longer
-	consecutiveTickLength := time.Since(startTick)
-	if firstTickLength > consecutiveTickLength {
-		t.Fatal("Should have increased tick length after first consecutive tick")
-	}
-
 	// Wait for another broadcast to occur
-	startTick = time.Now()
 	select {
 	case receivedWantReq := <-fpm.wantReqs:
 		if len(receivedWantReq.cids) < len(cids) {
@@ -429,12 +419,6 @@ func TestSessionFailingToGetFirstBlock(t *testing.T) {
 		t.Fatal("Never rebroadcast want list")
 	}
 
-	// Tick should take longer
-	secondConsecutiveTickLength := time.Since(startTick)
-	if consecutiveTickLength > secondConsecutiveTickLength {
-		t.Fatal("Should have increased tick length after first consecutive tick")
-	}
-
 	// Should not have tried to find peers on consecutive ticks
 	select {
 	case <-fpf.findMorePeersRequested:

routing/providerquerymanager/providerquerymanager.go

@@ -150,12 +150,9 @@ func New(ctx context.Context, dialer ProviderQueryDialer, router ProviderQueryRo
 		}
 	}
 
-	return pqm, nil
-}
-
-// Startup starts processing for the ProviderQueryManager.
-func (pqm *ProviderQueryManager) Startup() {
 	go pqm.run()
+
+	return pqm, nil
 }
 
 type inProgressRequest struct {

routing/providerquerymanager/providerquerymanager_test.go

@@ -76,7 +76,6 @@ func TestNormalSimultaneousFetch(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn))
-	providerQueryManager.Startup()
 	keys := random.Cids(2)
 
 	sessionCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -114,7 +113,6 @@ func TestDedupingProviderRequests(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn))
-	providerQueryManager.Startup()
 	key := random.Cids(1)[0]
 
 	sessionCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -155,7 +153,6 @@ func TestCancelOneRequestDoesNotTerminateAnother(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn))
-	providerQueryManager.Startup()
 
 	key := random.Cids(1)[0]
 
@@ -202,7 +199,6 @@ func TestCancelManagerExitsGracefully(t *testing.T) {
 	managerCtx, managerCancel := context.WithTimeout(ctx, 5*time.Millisecond)
 	defer managerCancel()
 	providerQueryManager := mustNotErr(New(managerCtx, fpd, fpn))
-	providerQueryManager.Startup()
 
 	key := random.Cids(1)[0]
 
@@ -238,7 +234,6 @@ func TestPeersWithConnectionErrorsNotAddedToPeerList(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn))
-	providerQueryManager.Startup()
 
 	key := random.Cids(1)[0]
 
@@ -275,7 +270,6 @@ func TestRateLimitingRequests(t *testing.T) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxInProcessRequests(maxInProcessRequests)))
-	providerQueryManager.Startup()
 
 	keys := random.Cids(maxInProcessRequests + 1)
 	sessionCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -317,7 +311,6 @@ func TestUnlimitedRequests(t *testing.T) {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxInProcessRequests(0)))
-	providerQueryManager.Startup()
 
 	keys := random.Cids(inProcessRequests)
 	sessionCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -355,7 +348,6 @@ func TestFindProviderTimeout(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxTimeout(2*time.Millisecond)))
-	providerQueryManager.Startup()
 	keys := random.Cids(1)
 
 	sessionCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
@@ -379,7 +371,6 @@ func TestFindProviderPreCanceled(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxTimeout(100*time.Millisecond)))
-	providerQueryManager.Startup()
 	keys := random.Cids(1)
 
 	sessionCtx, cancel := context.WithCancel(ctx)
@@ -404,7 +395,6 @@ func TestCancelFindProvidersAfterCompletion(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxTimeout(100*time.Millisecond)))
-	providerQueryManager.Startup()
 	keys := random.Cids(1)
 
 	sessionCtx, cancel := context.WithCancel(ctx)
@@ -437,7 +427,6 @@ func TestLimitedProviders(t *testing.T) {
 	}
 	ctx := context.Background()
 	providerQueryManager := mustNotErr(New(ctx, fpd, fpn, WithMaxProviders(max), WithMaxTimeout(100*time.Millisecond)))
-	providerQueryManager.Startup()
 	keys := random.Cids(1)
 
 	providersChan := providerQueryManager.FindProvidersAsync(ctx, keys[0], 0)