fix potential leaks of nse/ns streams in case of lost close

Signed-off-by: denis-tingaikin <[email protected]>

pkg/networkservice/common/netsvcmonitor/server.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2023 Cisco Systems, Inc.
+// Copyright (c) 2023-2024 Cisco Systems, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -61,12 +61,24 @@ func (m *monitorServer) Request(ctx context.Context, request *networkservice.Net
 
 	var monitorCtx, cancel = context.WithCancel(m.chainCtx)
 
+	if len(resp.GetPath().GetPathSegments()) > 0 {
+		var minT = resp.GetPath().GetPathSegments()[0].Expires.AsTime().Local()
+		for _, seg := range resp.GetPath().GetPathSegments() {
+			var t = seg.Expires.AsTime().Local()
+			if minT.Before(t) {
+				minT = t
+			}
+		}
+		cancel()
+		monitorCtx, cancel = context.WithTimeout(m.chainCtx, time.Until(minT))
+	}
+
 	storeCancelFunction(ctx, cancel)
 
 	var logger = log.FromContext(ctx).WithField("monitorServer", "Find")
 
 	var monitorNetworkServiceGoroutine = func() {
-		for ; monitorCtx.Err() == nil; time.Sleep(time.Millisecond * 100) {
+		for ; monitorCtx.Err() == nil; time.Sleep(time.Millisecond * 10) {
 			// nolint:govet
 			var stream, err = m.nsClient.Find(monitorCtx, &registry.NetworkServiceQuery{
 				Watch: true,

pkg/networkservice/common/netsvcmonitor/server_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2023 Cisco Systems, Inc.
+// Copyright (c) 2023-2024 Cisco Systems, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -25,6 +25,8 @@ import (
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	"github.com/networkservicemesh/api/pkg/api/registry"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/begin"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/netsvcmonitor"
@@ -36,6 +38,10 @@ import (
 )
 
 func Test_Netsvcmonitor_And_GroupOfSimilarNetworkServices(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+
 	var testCtx, cancel = context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 
@@ -91,3 +97,75 @@ func Test_Netsvcmonitor_And_GroupOfSimilarNetworkServices(t *testing.T) {
 		return counter.Closes() > 0
 	}, time.Millisecond*300, time.Millisecond*50)
 }
+
+func Test_NetsvcMonitor_ShouldNotLeakWithoutClose(t *testing.T) {
+	t.Cleanup(func() {
+		goleak.VerifyNone(t)
+	})
+
+	var testCtx, cancel = context.WithCancel(context.Background())
+	defer cancel()
+
+	var nsServer = memory.NewNetworkServiceRegistryServer()
+	var nseServer = memory.NewNetworkServiceEndpointRegistryServer()
+	var counter count.Server
+
+	_, _ = nsServer.Register(context.Background(), &registry.NetworkService{
+		Name: "service-1",
+	})
+
+	_, _ = nseServer.Register(context.Background(), &registry.NetworkServiceEndpoint{
+		Name:                "endpoint-1",
+		NetworkServiceNames: []string{"service-1"},
+	})
+
+	var server = chain.NewNetworkServiceServer(
+		metadata.NewServer(),
+		begin.NewServer(),
+		netsvcmonitor.NewServer(
+			testCtx,
+			adapters.NetworkServiceServerToClient(nsServer),
+			adapters.NetworkServiceEndpointServerToClient(nseServer),
+		),
+		&counter,
+	)
+
+	var n = time.Now().Add(time.Second)
+
+	var request = &networkservice.NetworkServiceRequest{
+		Connection: &networkservice.Connection{
+			Id:                         "1",
+			NetworkService:             "service-1",
+			NetworkServiceEndpointName: "endpoint-1",
+			Path: &networkservice.Path{
+				PathSegments: []*networkservice.PathSegment{
+					{
+						Expires: timestamppb.New(n),
+					},
+				},
+			},
+		},
+	}
+
+	var _, err = server.Request(testCtx, request)
+	require.NoError(t, err)
+
+	for end := time.Now().Add(time.Second); time.Now().Before(end); time.Sleep(time.Millisecond * 100) {
+		if goleak.Find() != nil {
+			break
+		}
+	}
+	if goleak.Find() == nil {
+		require.FailNow(t, "netsvc goroutine must be created")
+	}
+
+	for end := time.Now().Add(time.Second * 2); time.Now().Before(end); time.Sleep(time.Millisecond * 100) {
+		if goleak.Find() == nil {
+			break
+		}
+	}
+	if e := goleak.Find(); e != nil {
+		require.FailNow(t, "netsvc goroutine must be stopped, but it's found"+e.Error())
+	}
+
+}