Propagate Nexus Operation-Timeout header #6676

pdoerner · 2024-10-18T19:33:36Z

What changed?

Added support to propagate the new Operation-Timeout header. If this timeout is shorter than the context timeout, the context timeout will be reduced.

Why?

How did you test it?

New tests

pdoerner · 2024-10-18T19:35:48Z

components/nexusoperations/workflow/commands.go

@@ -110,6 +111,14 @@ func (ch *commandHandler) HandleScheduleCommand(
 		}
 	}

+	if err := timestamp.ValidateProtoDuration(attrs.ScheduleToCloseTimeout); err != nil {


Not exactly related to this PR but I realized I missed this validation in #6631

components/nexusoperations/executors.go

bergundy · 2024-10-28T20:03:29Z

components/nexusoperations/executors.go

+	// The following logic handles the operation ScheduleToCloseTimout parameter and Operation-Timeout header.
+	// The minimum of the resolved operation timeout and the configured request timeout is used for the context timeout
+	// when making the StartOperation request, according to the following logic:
+	// (ScheduleToClose set, Operation-Timeout set) -> no changes, use ScheduleToClose


Suggested change

// (ScheduleToClose set, Operation-Timeout set) -> no changes, use ScheduleToClose

// (ScheduleToClose set, Operation-Timeout set) -> Use Operation-Timeout

I thought we discussed that ScheduleToClose should always be used if it is set.

I mean just the Operation-Timeout header, not the context deadline.

bergundy · 2024-10-28T20:04:36Z

components/nexusoperations/executors.go

+	// (ScheduleToClose unset, Operation-Timeout set) -> no changes, use Operation-Timeout
+	header := nexus.Header(args.header)
+	opTimeout := args.scheduleToCloseTimeout
+	opTimeoutHeader, set := header["Operation-Timeout"]


Use .Get() which is case insensitive.

bergundy · 2024-10-28T20:08:28Z

components/nexusoperations/executors.go

+		if header == nil {
+			header = make(nexus.Header, 1)
+		}
+		header["Operation-Timeout"] = args.scheduleToCloseTimeout.String()


We need to export this from the Nexus SDK and use it here and for Request-Timeout: https://github.com/nexus-rpc/sdk-go/pull/20/files#diff-98737c3c4b252474548805ea1be712acf49b5514a96e5ffdbc642c366482f282R363

Also please use time left, not time since the operation was originally scheduled by the workflow. If this is the 10th attempt and an hour has already gone by, it should be reflected in the header.

bergundy · 2024-10-28T20:11:10Z

components/nexusoperations/executors.go

+			header = make(nexus.Header, 1)
+		}
+		header["Operation-Timeout"] = args.scheduleToCloseTimeout.String()
+	} else if set && args.scheduleToCloseTimeout == 0 {


Let's just ignore the user header and not enforce this. It's not technically supported and doesn't affect what the caller machinery scheduleToClose timeout.

bergundy · 2024-10-28T20:15:51Z

service/matching/physical_task_queue_manager.go

@@ -423,7 +423,18 @@ func (c *physicalTaskQueueManagerImpl) DispatchNexusTask(
 	request *matchingservice.DispatchNexusTaskRequest,
 ) (*matchingservice.DispatchNexusTaskResponse, error) {
 	deadline, _ := ctx.Deadline() // If not set by user, our client will set a default.
-	task := newInternalNexusTask(taskId, deadline, request)
+	var opDeadline time.Time
+	opTimeoutHeader, set := request.GetRequest().GetHeader()["Operation-Timeout"]


What if header is nil here? What if the header is user provided (e.g. via curl) and is in lower case?

Also note that this is only applicable for StartOperation requests but I don't think it hurts to manipulate on the other requests.

bergundy · 2024-10-28T20:16:39Z

service/matching/physical_task_queue_manager.go

+		opTimeout, err := time.ParseDuration(opTimeoutHeader)
+		if err != nil {
+			// Operation-Timeout header is not required so don't fail request on parsing errors.
+			c.logger.Warn(fmt.Sprintf("unable to parse Operation-Timeout header: %v", opTimeoutHeader), tag.Error(err))


Maybe log the request ID here if one is present?

bergundy

I think we need another short round here but this is very close.

bergundy · 2024-11-20T19:53:33Z

components/nexusoperations/executors.go

+			if header == nil {
+				header = make(nexus.Header, 1)
+			}
+			header[nexus.HeaderOperationTimeout] = opTimeout.String()


I realized we need a Set method on nexus.Header and a doc clarification that headers set directly should be in lower case. Not blocking this PR though.

bergundy · 2024-11-20T19:55:42Z

components/nexusoperations/executors.go

+	header := nexus.Header(args.header)
+	callTimeout := e.Config.RequestTimeout(ns.Name().String(), task.EndpointName)
+	if args.scheduleToCloseTimeout > 0 {
+		opTimeout := max(time.Duration(0), args.scheduleToCloseTimeout-time.Since(args.scheduledTime))


Let's add a dynamic config to control the minimum time allowed for a single operation call. The default will be 0 and instead of doing this max here, you'd basically fail the request with a non-retryable timeout and avoid sending the RPC.

Also, for the cancel request, set the context deadline to the minimum and fail the cancelation with a non-retryable timeout and avoid sending the RPC.

And add test coverage for this please.

bergundy · 2024-11-20T19:59:39Z

components/nexusoperations/executors_test.go

+			name:                "sync start",
+			requestTimeout:      time.Hour,
+			schedToCloseTimeout: time.Hour,
+			headers:             nexus.Header{nexus.HeaderOperationTimeout: time.Microsecond.String()}, // to test this value is ignored when ScheduleToCloseTimeout is set


No need to test this IMHO, but you can check that we don't override this user header if they set it.

bergundy · 2024-11-20T19:59:52Z

components/nexusoperations/executors_test.go

@@ -105,11 +105,13 @@ func TestProcessInvocationTask(t *testing.T) {
 		endpointNotFound           bool
 		eventHasNoEndpointID       bool
 		operationIsCanceled        bool
+		headers                    nexus.Header


Suggested change

headers nexus.Header

header nexus.Header

bergundy · 2024-11-20T20:02:01Z

components/nexusoperations/executors_test.go

+			checkOutcome: func(t *testing.T, op nexusoperations.Operation, events []*historypb.HistoryEvent) {
+				require.Equal(t, enumsspb.NEXUS_OPERATION_STATE_BACKING_OFF, op.State())
+				require.NotNil(t, op.LastAttemptFailure.GetApplicationFailureInfo())
+				require.Regexp(t, "Post \"http://localhost:\\d+/service/operation\\?callback=http%3A%2F%2Flocalhost%2Fcallback\": context deadline exceeded", op.LastAttemptFailure.Message)


Hmm... we should probably not show the HTTP endpoint in the error message. But that's out of scope for this PR.

bergundy · 2024-11-20T20:03:59Z

service/matching/physical_task_queue_manager.go

@@ -423,7 +424,19 @@ func (c *physicalTaskQueueManagerImpl) DispatchNexusTask(
 	request *matchingservice.DispatchNexusTaskRequest,
 ) (*matchingservice.DispatchNexusTaskResponse, error) {
 	deadline, _ := ctx.Deadline() // If not set by user, our client will set a default.
-	task := newInternalNexusTask(taskId, deadline, request)
+	var opDeadline time.Time
+	if headers := nexus.Header(request.GetRequest().GetHeader()); headers != nil {


nit: call this header as we do in the rest of the codebase.

Propagate Nexus Operation-Timeout header

bf4882c

pdoerner commented Oct 18, 2024

View reviewed changes

tests

0ba0bf6

pdoerner requested a review from bergundy October 28, 2024 17:15

pdoerner marked this pull request as ready for review October 28, 2024 17:15

pdoerner requested a review from a team as a code owner October 28, 2024 17:15

Merge branch 'main' into propagate-operation-timeout-header

0ab7a3e

pdoerner commented Oct 28, 2024

View reviewed changes

components/nexusoperations/executors.go Outdated Show resolved Hide resolved

bergundy reviewed Oct 28, 2024

View reviewed changes

pdoerner added 3 commits November 19, 2024 08:55

Merge branch 'main' into propagate-operation-timeout-header

e5cbb47

feedback

94168b5

lint

7f0e9fa

pdoerner requested a review from bergundy November 19, 2024 18:47

bergundy reviewed Nov 20, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Propagate Nexus Operation-Timeout header #6676

Propagate Nexus Operation-Timeout header #6676

pdoerner commented Oct 18, 2024 •

edited

Loading

pdoerner Oct 18, 2024

bergundy Oct 28, 2024

pdoerner Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy Oct 28, 2024

bergundy left a comment

bergundy Nov 20, 2024

bergundy Nov 20, 2024

bergundy Nov 20, 2024

bergundy Nov 20, 2024

bergundy Nov 20, 2024

bergundy Nov 20, 2024

bergundy Nov 20, 2024

	// (ScheduleToClose set, Operation-Timeout set) -> no changes, use ScheduleToClose
	// (ScheduleToClose set, Operation-Timeout set) -> Use Operation-Timeout

Propagate Nexus Operation-Timeout header #6676

Are you sure you want to change the base?

Propagate Nexus Operation-Timeout header #6676

Conversation

pdoerner commented Oct 18, 2024 • edited Loading

What changed?

Why?

How did you test it?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bergundy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pdoerner commented Oct 18, 2024 •

edited

Loading