Add additional test for RequestInterval

p2p/server/server.go

@@ -343,6 +343,10 @@ func (s *Server) Request(ctx context.Context, pid peer.ID, req []byte, extraProt
 	if err := s.StreamRequest(ctx, pid, req, func(ctx context.Context, stream io.ReadWriter) error {
 		rd := bufio.NewReader(stream)
 		if _, err := codec.DecodeFrom(rd, &r); err != nil {
+			if errors.Is(err, io.ErrClosedPipe) && ctx.Err() != nil {
+				// ensure that a canceled context is returned as the right error
+				return ctx.Err()
+			}
 			return fmt.Errorf("peer %s: %w", pid, err)
 		}
 		if r.Error != "" {

p2p/server/server_test.go

@@ -194,8 +194,6 @@ func Test_Queued(t *testing.T) {
 			return msg, nil
 		}),
 		WithQueueSize(queueSize),
-		WithRequestsPerInterval(50, time.Second),
-		WithMetrics(),
 	)
 	var (
 		eg          errgroup.Group
@@ -228,6 +226,48 @@ func Test_Queued(t *testing.T) {
 	require.NoError(t, reqEq.Wait())
 }
 
+func Test_RequestInterval(t *testing.T) {
+	mesh, err := mocknet.FullMeshConnected(2)
+	require.NoError(t, err)
+
+	var (
+		maxReqPerMin = 10
+		proto        = "test"
+	)
+
+	client := New(wrapHost(t, mesh.Hosts()[0]), proto, nil)
+	srv := New(
+		wrapHost(t, mesh.Hosts()[1]),
+		proto,
+		WrapHandler(func(_ context.Context, msg []byte) ([]byte, error) {
+			return msg, nil
+		}),
+		WithRequestsPerInterval(maxReqPerMin, time.Minute),
+	)
+	var (
+		eg          errgroup.Group
+		ctx, cancel = context.WithCancel(context.Background())
+	)
+	defer cancel()
+	eg.Go(func() error {
+		return srv.Run(ctx)
+	})
+	t.Cleanup(func() {
+		assert.NoError(t, eg.Wait())
+	})
+
+	for i := 0; i < maxReqPerMin; i++ {
+		resp, err := client.Request(ctx, mesh.Hosts()[1].ID(), []byte("ping"))
+		require.NoError(t, err)
+		require.Equal(t, []byte("ping"), resp)
+	}
+
+	ctx, cancel = context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+	_, err = client.Request(ctx, mesh.Hosts()[1].ID(), []byte("ping"))
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+}
+
 func FuzzResponseConsistency(f *testing.F) {
 	tester.FuzzConsistency[Response](f)
 }