style(lint): fix lint errors

allocator/allocator_test.go

@@ -319,7 +319,8 @@ func TestAllocator(t *testing.T) {
 					select {
 					case <-allocated:
 					default:
-						pending = append(pending, pendingResultWithChan{pendingResult{op.p, op.amount}, allocated})
+
+						pending = append(pending, pendingResultWithChan{pendingResult{p: op.p, amount: op.amount}, allocated}) //nolint:gosimple
 					}
 				case releasePeer:
 					err := allocator.ReleasePeerMemory(op.p)

benchmarks/benchmark_test.go

@@ -29,17 +29,15 @@ import (
 	basicnode "github.com/ipld/go-ipld-prime/node/basic"
 	ipldselector "github.com/ipld/go-ipld-prime/traversal/selector"
 	"github.com/ipld/go-ipld-prime/traversal/selector/builder"
-	peer "github.com/libp2p/go-libp2p-core/peer"
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/ipfs/go-graphsync/benchmarks/testinstance"
 	tn "github.com/ipfs/go-graphsync/benchmarks/testnet"
 	graphsync "github.com/ipfs/go-graphsync/impl"
 )
 
-const stdBlockSize = 8000
-
 type runStats struct {
 	Time time.Duration
 	Name string
@@ -96,34 +94,36 @@ func benchmarkRepeatedDisconnects(ctx context.Context, b *testing.B, numnodes in
 	b.ReportAllocs()
 	fetcher := instances[0]
 	for i := 0; i < b.N; i++ {
-		var wg sync.WaitGroup
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		require.NoError(b, err)
 		start := time.Now()
+		errgrp, grpctx := errgroup.WithContext(ctx)
 		for j := 0; j < numnodes; j++ {
 			instance := instances[j+1]
-			_, errChan := fetcher.Exchange.Request(ctx, instance.Peer, cidlink.Link{Cid: allCids[i][j]}, allSelector)
+			_, errChan := fetcher.Exchange.Request(grpctx, instance.Peer, cidlink.Link{Cid: allCids[i][j]}, allSelector)
+			other := instance.Peer
 
-			wg.Add(1)
-			go func(other peer.ID) {
+			errgrp.Go(func() error {
 				defer func() {
-					mn.DisconnectPeers(fetcher.Peer, other)
-					wg.Done()
+					_ = mn.DisconnectPeers(fetcher.Peer, other)
 				}()
 				for {
 					select {
-					case <-ctx.Done():
-						return
+					case <-grpctx.Done():
+						return nil
 					case err, ok := <-errChan:
 						if !ok {
-							return
+							return nil
 						}
-						b.Fatalf("received error on request: %s", err.Error())
+						return err
 					}
 				}
-			}(instance.Peer)
+			})
+
+		}
+		if err := errgrp.Wait(); err != nil {
+			b.Fatalf("received error on request: %s", err.Error())
 		}
-		wg.Wait()
 		result := runStats{
 			Time: time.Since(start),
 			Name: b.Name(),
@@ -163,24 +163,24 @@ func p2pStrestTest(ctx context.Context, b *testing.B, numfiles int, df distFunc,
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
 		fetcher := instances[i+1]
-		var wg sync.WaitGroup
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		require.NoError(b, err)
 		start := time.Now()
+		errgrp, grpctx := errgroup.WithContext(ctx)
 		for j := 0; j < numfiles; j++ {
-			responseChan, errChan := fetcher.Exchange.Request(ctx, instances[0].Peer, cidlink.Link{Cid: allCids[j]}, allSelector)
-
-			wg.Add(1)
-			go func(j int) {
-				defer wg.Done()
-				for _ = range responseChan {
+			responseChan, errChan := fetcher.Exchange.Request(grpctx, instances[0].Peer, cidlink.Link{Cid: allCids[j]}, allSelector)
+			errgrp.Go(func() error {
+				for range responseChan {
 				}
 				for err := range errChan {
-					b.Fatalf("received error on request: %s", err.Error())
+					return err
 				}
-			}(j)
+				return nil
+			})
+		}
+		if err := errgrp.Wait(); err != nil {
+			b.Fatalf("received error on request: %s", err.Error())
 		}
-		wg.Wait()
 		result := runStats{
 			Time: time.Since(start),
 			Name: b.Name(),
@@ -216,31 +216,31 @@ func subtestDistributeAndFetch(ctx context.Context, b *testing.B, numnodes int,
 	b.ReportAllocs()
 	for i := 0; i < b.N; i++ {
 		fetcher := instances[i+numnodes]
-		var wg sync.WaitGroup
 		ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
 		require.NoError(b, err)
 		start := time.Now()
+		errgrp, grpctx := errgroup.WithContext(ctx)
 		for j := 0; j < numnodes; j++ {
 			instance := instances[j]
-			_, errChan := fetcher.Exchange.Request(ctx, instance.Peer, cidlink.Link{Cid: destCids[j]}, allSelector)
+			_, errChan := fetcher.Exchange.Request(grpctx, instance.Peer, cidlink.Link{Cid: destCids[j]}, allSelector)
 
-			wg.Add(1)
-			go func() {
-				defer wg.Done()
+			errgrp.Go(func() error {
 				for {
 					select {
 					case <-ctx.Done():
-						return
+						return err
 					case err, ok := <-errChan:
 						if !ok {
-							return
+							return nil
 						}
-						b.Fatalf("received error on request: %s", err.Error())
+						return err
 					}
 				}
-			}()
+			})
+		}
+		if err := errgrp.Wait(); err != nil {
+			b.Fatalf("received error on request: %s", err.Error())
 		}
-		wg.Wait()
 		result := runStats{
 			Time: time.Since(start),
 			Name: b.Name(),

benchmarks/testnet/peernet.go

@@ -24,7 +24,10 @@ func (pn *peernet) Adapter(p tnet.Identity) gsnet.GraphSyncNetwork {
 	if err != nil {
 		panic(err.Error())
 	}
-	pn.Mocknet.LinkAll()
+	err = pn.Mocknet.LinkAll()
+	if err != nil {
+		panic(err.Error())
+	}
 	return gsnet.NewFromLibp2pHost(client)
 }
 

go.mod

@@ -43,6 +43,7 @@ require (
 	github.com/smartystreets/assertions v1.0.1 // indirect
 	github.com/stretchr/testify v1.6.1
 	github.com/whyrusleeping/cbor-gen v0.0.0-20210219115102-f37d292932f2
+	golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9
 	golang.org/x/tools v0.1.1-0.20210225150353-54dc8c5edb56 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1
 	google.golang.org/protobuf v1.25.0

impl/graphsync_test.go

@@ -610,8 +610,8 @@ func TestNetworkDisconnect(t *testing.T) {
 	testutil.AssertChannelEmpty(t, networkError, "no network errors so far")
 
 	// unlink peers so they cannot communicate
-	td.mn.DisconnectPeers(td.host1.ID(), td.host2.ID())
-	td.mn.UnlinkPeers(td.host1.ID(), td.host2.ID())
+	require.NoError(t, td.mn.DisconnectPeers(td.host1.ID(), td.host2.ID()))
+	require.NoError(t, td.mn.UnlinkPeers(td.host1.ID(), td.host2.ID()))
 	requestID := <-requestIDChan
 	err := responder.UnpauseResponse(td.host1.ID(), requestID)
 	require.NoError(t, err)
@@ -639,8 +639,8 @@ func TestConnectFail(t *testing.T) {
 	defer requestCancel()
 
 	// unlink peers so they cannot communicate
-	td.mn.DisconnectPeers(td.host1.ID(), td.host2.ID())
-	td.mn.UnlinkPeers(td.host1.ID(), td.host2.ID())
+	require.NoError(t, td.mn.DisconnectPeers(td.host1.ID(), td.host2.ID()))
+	require.NoError(t, td.mn.UnlinkPeers(td.host1.ID(), td.host2.ID()))
 
 	reqNetworkError := make(chan error, 1)
 	requestor.RegisterNetworkErrorListener(func(p peer.ID, request graphsync.RequestData, err error) {

message/message.go

@@ -304,6 +304,9 @@ func (gsm GraphSyncMessage) ToProto() (*pb.Message, error) {
 
 func (gsm GraphSyncMessage) ToNet(w io.Writer) error {
 	msg, err := gsm.ToProto()
+	if err != nil {
+		return err
+	}
 	size := proto.Size(msg)
 	buf := pool.Get(size + binary.MaxVarintLen64)
 	defer pool.Put(buf)

messagequeue/messagequeue.go

@@ -133,7 +133,7 @@ func (mq *MessageQueue) Shutdown() {
 
 func (mq *MessageQueue) runQueue() {
 	defer func() {
-		mq.allocator.ReleasePeerMemory(mq.p)
+		_ = mq.allocator.ReleasePeerMemory(mq.p)
 		mq.eventPublisher.Shutdown()
 	}()
 	mq.eventPublisher.Startup()
@@ -308,12 +308,12 @@ func (mp *messagePublisher) publishQueued() {
 
 func (mp *messagePublisher) publishSent() {
 	mp.mq.eventPublisher.Publish(mp.topic, Event{Name: Sent})
-	mp.mq.allocator.ReleaseBlockMemory(mp.mq.p, mp.msgSize)
+	_ = mp.mq.allocator.ReleaseBlockMemory(mp.mq.p, mp.msgSize)
 }
 
 func (mp *messagePublisher) publishError(err error) {
 	mp.mq.eventPublisher.Publish(mp.topic, Event{Name: Error, Err: err})
-	mp.mq.allocator.ReleaseBlockMemory(mp.mq.p, mp.msgSize)
+	_ = mp.mq.allocator.ReleaseBlockMemory(mp.mq.p, mp.msgSize)
 }
 
 func (mp *messagePublisher) close() {

messagequeue/messagequeue_test.go

@@ -332,7 +332,7 @@ func TestSendsResponsesMemoryPressure(t *testing.T) {
 	defer cancel()
 
 	p := testutil.GeneratePeers(1)[0]
-	messagesSent := make(chan gsmsg.GraphSyncMessage, 0)
+	messagesSent := make(chan gsmsg.GraphSyncMessage)
 	resetChan := make(chan struct{}, 1)
 	fullClosedChan := make(chan struct{}, 1)
 	messageSender := &fakeMessageSender{nil, fullClosedChan, resetChan, messagesSent}
@@ -363,6 +363,7 @@ func TestSendsResponsesMemoryPressure(t *testing.T) {
 
 	// assert transaction does not complete within 200ms because it is waiting on memory
 	ctx2, cancel2 := context.WithTimeout(ctx, 200*time.Millisecond)
+	defer cancel2()
 	select {
 	case <-finishes:
 		t.Fatal("transaction failed to wait on memory")

notifications/data_subscriber.go

@@ -34,9 +34,7 @@ func (m *TopicDataSubscriber) getData(id Topic) []TopicData {
 		return []TopicData{}
 	}
 	newData := make([]TopicData, len(data))
-	for i, d := range data {
-		newData[i] = d
-	}
+	copy(newData, data)
 	return newData
 }
 

responsemanager/responseassembler/responseassembler_test.go

@@ -444,17 +444,3 @@ func (fph *fakePeerHandler) sendResponse(p peer.ID, responses []gsmsg.GraphSyncR
 	fph.lastBlocks = blks
 	fph.lastNotifiees = notifees
 }
-
-type fakeAllocator struct {
-	response chan error
-}
-
-func newFakeAllocator() *fakeAllocator {
-	return &fakeAllocator{
-		response: make(chan error, 0),
-	}
-}
-
-func (fa *fakeAllocator) AllocateBlockMemory(p peer.ID, amount uint64) <-chan error {
-	return fa.response
-}

testutil/chaintypes/gen/main.go

@@ -28,5 +28,5 @@ func main() {
 		schema.SpawnStructRepresentationMap(nil),
 	))
 	gengo.Generate(".", pkgName, ts, adjCfg)
-	exec.Command("go", "fmt").Run()
+	_ = exec.Command("go", "fmt").Run()
 }

testutil/testnotifications.go

@@ -11,7 +11,6 @@ import (
 )
 
 type TestSubscriber struct {
-	expectedTopic  notifications.Topic
 	receivedEvents chan DispatchedEvent
 	closed         chan notifications.Topic
 }