timeout: take the minimum expiration time of pathSegments

pkg/networkservice/common/refresh/client.go

@@ -1,6 +1,6 @@
-// Copyright (c) 2020 Cisco Systems, Inc.
+// Copyright (c) 2020-2023 Cisco Systems, Inc.
 //
-// Copyright (c) 2020-2022 Doc.ai and/or its affiliates.
+// Copyright (c) 2020-2023 Doc.ai and/or its affiliates.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -58,6 +58,7 @@ func (t *refreshClient) Request(ctx context.Context, request *networkservice.Net
 
 	// Compute refreshAfter
 	refreshAfter := after(ctx, conn)
+	log.FromContext(ctx).Infof("refresh after %s", refreshAfter.String())
 
 	// Create a cancel context.
 	cancelCtx, cancel := context.WithCancel(t.chainCtx)
@@ -102,8 +103,10 @@ func after(ctx context.Context, conn *networkservice.Connection) time.Duration {
 	clockTime := clock.FromContext(ctx)
 
 	var minTimeout *time.Duration
-	var expireTime time.Time
 	for _, segment := range conn.GetPath().GetPathSegments() {
+		if segment.GetExpires() == nil {
+			continue
+		}
 		expTime := segment.GetExpires().AsTime()
 
 		timeout := clockTime.Until(expTime)
@@ -114,14 +117,9 @@ func after(ctx context.Context, conn *networkservice.Connection) time.Duration {
 			}
 
 			*minTimeout = timeout
-			expireTime = expTime
 		}
 	}
 
-	if minTimeout != nil {
-		log.FromContext(ctx).Infof("expiration after %s at %s", minTimeout.String(), expireTime.UTC())
-	}
-
 	if minTimeout == nil || *minTimeout <= 0 {
 		return 1
 	}

pkg/networkservice/common/timeout/server.go

@@ -1,6 +1,6 @@
 // Copyright (c) 2020-2021 Doc.ai and/or its affiliates.
 //
-// Copyright (c) 2020-2022 Cisco Systems, Inc.
+// Copyright (c) 2020-2023 Cisco Systems, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -26,11 +26,11 @@ import (
 	iserror "errors"
 
 	"github.com/golang/protobuf/ptypes/empty"
-	"github.com/pkg/errors"
 
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/begin"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/metadata"
 	"github.com/networkservicemesh/sdk/pkg/tools/clock"
+	"github.com/networkservicemesh/sdk/pkg/tools/log"
 
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 
@@ -64,18 +64,13 @@ func (s *timeoutServer) Request(ctx context.Context, request *networkservice.Net
 		return nil, err
 	}
 
-	expirationTimestamp := conn.GetPrevPathSegment().GetExpires()
-	if expirationTimestamp == nil {
-		return nil, errors.Errorf("expiration for the previous path segment cannot be nil: %+v", conn)
-	}
-	expirationTime := expirationTimestamp.AsTime()
 	cancelCtx, cancel := context.WithCancel(s.chainCtx)
 	if oldCancel, loaded := loadAndDelete(ctx, metadata.IsClient(s)); loaded {
 		oldCancel()
 	}
 	store(ctx, metadata.IsClient(s), cancel)
 	eventFactory := begin.FromContext(ctx)
-	afterCh := timeClock.After(timeClock.Until(expirationTime) - requestTimeout)
+	afterCh := timeClock.After(minTokenTimeout(ctx, conn) - requestTimeout)
 
 	go func(cancelCtx context.Context, afterCh <-chan time.Time) {
 		select {
@@ -97,3 +92,21 @@ func (s *timeoutServer) Close(ctx context.Context, conn *networkservice.Connecti
 	}
 	return &empty.Empty{}, err
 }
+
+func minTokenTimeout(ctx context.Context, conn *networkservice.Connection) time.Duration {
+	clockTime := clock.FromContext(ctx)
+	var minTimeout time.Duration = 0
+	for _, segment := range conn.GetPath().GetPathSegments() {
+		if segment.GetExpires() == nil {
+			continue
+		}
+		timeout := clockTime.Until(segment.GetExpires().AsTime())
+
+		if timeout < minTimeout || minTimeout == 0 {
+			minTimeout = timeout
+		}
+	}
+	log.FromContext(ctx).Infof("expiration after %s", minTimeout.String())
+
+	return minTimeout
+}

pkg/networkservice/common/timeout/server_test.go

@@ -1,5 +1,7 @@
 // Copyright (c) 2020-2022 Doc.ai and/or its affiliates.
 //
+// Copyright (c) 2023 Cisco and/or its affiliates.
+//
 // SPDX-License-Identifier: Apache-2.0
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -27,7 +29,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
-	"google.golang.org/grpc/credentials"
+	"google.golang.org/protobuf/types/known/timestamppb"
 
 	"github.com/networkservicemesh/api/pkg/api/networkservice"
 	kernelmech "github.com/networkservicemesh/api/pkg/api/networkservice/mechanisms/kernel"
@@ -39,7 +41,6 @@ import (
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/refresh"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/timeout"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/common/updatepath"
-	"github.com/networkservicemesh/sdk/pkg/networkservice/common/updatetoken"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/adapters"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/core/next"
 	"github.com/networkservicemesh/sdk/pkg/networkservice/utils/inject/injectclock"
@@ -61,7 +62,8 @@ func testClient(
 	ctx context.Context,
 	client networkservice.NetworkServiceClient,
 	server networkservice.NetworkServiceServer,
-	duration time.Duration,
+	clientDuration time.Duration,
+	serverDuration time.Duration,
 	clk clock.Clock,
 ) networkservice.NetworkServiceClient {
 	return next.NewNetworkServiceClient(
@@ -72,13 +74,11 @@ func testClient(
 		client,
 		adapters.NewServerToClient(
 			next.NewNetworkServiceServer(
-				updatetoken.NewServer(func(_ credentials.AuthInfo) (string, time.Time, error) {
-					return "token", clock.FromContext(ctx).Now().Add(duration), nil
-				}),
-				begin.NewServer(),
-				metadata.NewServer(),
 				new(remoteServer), // <-- GRPC invocation
 				updatepath.NewServer(serverName),
+				begin.NewServer(),
+				newInjectExpirationServer(clientDuration, serverDuration),
+				metadata.NewServer(),
 				timeout.NewServer(ctx),
 				server,
 			),
@@ -101,6 +101,38 @@ func TestTimeoutServer_Request(t *testing.T) {
 			kernelmech.MECHANISM: connServer,
 		}),
 		tokenTimeout,
+		tokenTimeout,
+		clockMock,
+	)
+
+	_, err := client.Request(ctx, &networkservice.NetworkServiceRequest{})
+	require.NoError(t, err)
+
+	require.Eventually(t, connServer.validator(1, 0), testWait, testTick)
+
+	clockMock.Add(tokenTimeout / 2)
+	require.Never(t, connServer.validator(0, 1), testWait, testTick)
+
+	clockMock.Add(tokenTimeout / 2)
+	require.Eventually(t, connServer.validator(0, 1), testWait, testTick)
+}
+
+func TestTimeoutServer_DifferentTimeouts(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	clockMock := clockmock.New(ctx)
+	ctx = clock.WithClock(ctx, clockMock)
+
+	connServer := newConnectionsServer(t)
+
+	client := testClient(ctx,
+		kernel.NewClient(),
+		mechanisms.NewServer(map[string]networkservice.NetworkServiceServer{
+			kernelmech.MECHANISM: connServer,
+		}),
+		tokenTimeout*2,
+		tokenTimeout,
 		clockMock,
 	)
 
@@ -131,6 +163,7 @@ func TestTimeoutServer_CloseBeforeTimeout(t *testing.T) {
 			kernelmech.MECHANISM: connServer,
 		}),
 		tokenTimeout,
+		tokenTimeout,
 		clockMock,
 	)
 
@@ -164,6 +197,7 @@ func TestTimeoutServer_CloseAfterTimeout(t *testing.T) {
 			kernelmech.MECHANISM: connServer,
 		}),
 		tokenTimeout,
+		tokenTimeout,
 		clockMock,
 	)
 
@@ -204,7 +238,7 @@ func TestTimeoutServer_RaceTest(t *testing.T) {
 
 	connServer := newConnectionsServer(t)
 
-	client := testClient(ctx, null.NewClient(), connServer, 0, clock.FromContext(ctx))
+	client := testClient(ctx, null.NewClient(), connServer, 0, 0, clock.FromContext(ctx))
 
 	var wg sync.WaitGroup
 	for i := 0; i < 1000; i++ {
@@ -235,22 +269,16 @@ func TestTimeoutServer_RefreshFailure(t *testing.T) {
 
 	client := testClient(
 		ctx,
-		next.NewNetworkServiceClient(
-			begin.NewClient(),
-			metadata.NewClient(),
-			refresh.NewClient(ctx),
-		),
+		refresh.NewClient(ctx),
 		next.NewNetworkServiceServer(
 			injecterror.NewServer(
 				injecterror.WithRequestErrorTimes(1, -1),
 				injecterror.WithCloseErrorTimes(),
 			),
-			updatetoken.NewServer(func(_ credentials.AuthInfo) (string, time.Time, error) {
-				return "test", clock.FromContext(ctx).Now().Add(tokenTimeout), nil
-			}),
 			connServer,
 		),
 		tokenTimeout,
+		tokenTimeout,
 		clockMock,
 	)
 
@@ -289,6 +317,7 @@ func TestTimeoutServer_CloseFailure(t *testing.T) {
 			connServer,
 		),
 		tokenTimeout,
+		tokenTimeout,
 		clockMock,
 	)
 
@@ -371,3 +400,30 @@ func (s *connectionsServer) Close(ctx context.Context, conn *networkservice.Conn
 
 	return next.Server(ctx).Close(ctx, conn)
 }
+
+// injectExpirationServer sets an expiration time for previous and current path segments
+type injectExpirationServer struct {
+	clientTimeout time.Duration
+	serverTimeout time.Duration
+}
+
+func newInjectExpirationServer(clientTimeout, serverTimeout time.Duration) *injectExpirationServer {
+	return &injectExpirationServer{
+		clientTimeout: clientTimeout,
+		serverTimeout: serverTimeout,
+	}
+}
+
+func (s *injectExpirationServer) Request(ctx context.Context, request *networkservice.NetworkServiceRequest) (*networkservice.Connection, error) {
+	if prev := request.GetConnection().GetPrevPathSegment(); prev != nil {
+		prev.Expires = timestamppb.New(clock.FromContext(ctx).Now().Add(s.clientTimeout).Local())
+	}
+	if curr := request.GetConnection().GetCurrentPathSegment(); curr != nil {
+		curr.Expires = timestamppb.New(clock.FromContext(ctx).Now().Add(s.serverTimeout).Local())
+	}
+	return next.Server(ctx).Request(ctx, request)
+}
+
+func (s *injectExpirationServer) Close(ctx context.Context, conn *networkservice.Connection) (*empty.Empty, error) {
+	return next.Server(ctx).Close(ctx, conn)
+}