Implement Nexus async completion endpoint #5735

bergundy · 2024-04-16T21:00:09Z

What changed?

Implemented the async Nexus operation completion endpoint with functional tests
Refactored frontend fx to provide a mux.Router instead of collecting handlers so I can inject routes in the component (plugin) code
Refactored task_executor.go to statemachine_environment.go and adapted it to support handling API requests
Added a StateMachineEnvironment() to history Engine. In the future we'll want to have the environment on the shard controller instead of the engine but because I'm waiting until we drop support for shard level workflow cache first to make that change
Refactored some helpers from nexus_handler.go into common/nexus for reusability

cmd/tools/rpcwrappers/main.go

service/history/consts/const.go

service/matching/forwarder.go

service/history/workflow/state_transition_history.go

yycptt · 2024-04-18T00:24:12Z

service/history/queues/executable.go

+		// The task is stale and is safe to be dropped.
+		// Even though ErrStaleReference is castable to serviceerror.NotFound, we give this error special treatment
+		// because we're interested in the metric.
+		metrics.TaskSkipped.With(e.taggedMetricsHandler).Record(1)


we should change the other place where TaskSkipped is emitted to return the StaleStateError

I'm not sure what the semantics are there.
When would the task event ID be >= than the mutable state next event ID?

It does seem like we want to put the task in the DLQ in that case instead of dropping it. Do you want to change that behavior in this PR that goes into a feature branch?

My understanding it that it should never happen and basically the same as the StaleStateError.

goes into a feature branch

I am fine either way. That dropping logic is there for a long time so not an urgent thing to fix.
I commented here because it might impact the interpretation of the TaskSkipped metric.

proto/internal/temporal/server/api/historyservice/v1/request_response.proto

proto/internal/temporal/server/api/token/v1/message.proto

service/frontend/fx.go

plugins/nexusoperations/executors.go

yycptt · 2024-04-18T01:12:46Z

plugins/nexusoperations/frontend/handler.go

+	Config                               *Config
+	CallbackTokenGenerator               *commonnexus.CallbackTokenGenerator
+	HistoryClient                        resource.HistoryClient
+	NamespaceValidationInterceptor       *interceptor.NamespaceValidatorInterceptor


I didn't read the code very carefully. I thought those interceptors are taken care of by the existing NexusHTTPHandler in frontend? I guess I am confused by all those handlers now 🤦

Yeah, I was debating whether to put this functionality in the frontend package.
I'm trying to see how far I can take these components.

There's definitely some duplication but I'll take care of that later. I have it tracked.

Yeah, I'm also a little confused why there is a separate Nexus frontend handler just for completions.

I think we could move the nexus http handler into this package as part of later refactoring work.

## What changed? Added pending nexus operations to describe response. I was debating if I want to allow components to register a describe hook but decided to just inline this for now.

bergundy · 2024-04-19T13:58:36Z

Note to reviewers, I squashed the stacked PR #5749 on top of this. You can ignore that commit.

yycptt · 2024-04-19T17:50:04Z

cmd/tools/rpcwrappers/main.go

+			panic(fmt.Sprintf("expected at least one namespace ID field in request with nesting of 2 in %s", t))
+		} else {
+			// There's more than one, assume there's a top level one (e.g.
+			// historyservice.GetWorkflowExecutionHistoryRequest)


I only see one NamespaceId field in historyservice.GetWorkflowExecutionHistoryRequest.

I think ideally we should panic here as well since it's ambiguous in this case.

Oh sorry it's GetWorkflowExecutionRawHistoryRequest.

yycptt · 2024-04-19T18:03:58Z

service/history/queues/executable.go

+		// The task is stale and is safe to be dropped.
+		// Even though ErrStaleReference is castable to serviceerror.NotFound, we give this error special treatment
+		// because we're interested in the metric.
+		metrics.TaskSkipped.With(e.taggedMetricsHandler).Record(1)


My understanding it that it should never happen and basically the same as the StaleStateError.

goes into a feature branch

I am fine either way. That dropping logic is there for a long time so not an urgent thing to fix.
I commented here because it might impact the interpretation of the TaskSkipped metric.

rodrigozhou · 2024-04-17T15:36:29Z

common/metrics/metric_defs.go

+		"nexus_completion_requests",
+		WithDescription("The number of Nexus completion (callback) requests received by the service."),
+	)
+	NexusCompletionLatencyHistogram = NewCounterDef(


Don't you want a NewTimerDef instead? Ditto L686 above.

Suggested change

NexusCompletionLatencyHistogram = NewCounterDef(

NexusCompletionLatency = NewTimerDef(

rodrigozhou · 2024-04-17T15:43:13Z

common/nexus/callback_token.go

+	}
+
+	completion := &tokenspb.NexusOperationCompletion{}
+	return completion, proto.Unmarshal(plaintext, completion)


Just wanna double check, proto.Unmarshal works with completion instead of &completion?

Can you add unit tests for this function and DecodeCallbackToken below?

rodrigozhou · 2024-04-17T15:46:04Z

common/nexus/callback_token.go

@@ -69,3 +71,26 @@ func (g *CallbackTokenGenerator) Tokenize(completion *tokenspb.NexusOperationCom
 	}
 	return string(b), nil
 }
+
+// DecodeCompletion decodes a callback token unwrapping the contained NexusOperationCompletion proto struct.
+func (g *CallbackTokenGenerator) DecodeCompletion(token *CallbackToken) (*tokenspb.NexusOperationCompletion, error) {


This function doesn't seem to need be a method of CallbackTokenGenerator

rodrigozhou · 2024-04-17T15:54:34Z

common/nexus/failure.go

+	if !exposeDetails {
+		return nexus.HandlerErrorf(nexus.HandlerErrorTypeInternal, "internal error")
+	}
+	return err


Can you move this to the default case inside the switch?
Also, don't you need to wrap the last return in nexus.HandlerError?

rodrigozhou · 2024-04-17T17:34:07Z

plugins/nexusoperations/completion.go

+func handleSuccessfulOperationResult(node *hsm.Node, operation Operation,
+	result *commonpb.Payload,
+	attemptTime *time.Time,


nit

Suggested change

func handleSuccessfulOperationResult(node *hsm.Node, operation Operation,

result *commonpb.Payload,

attemptTime *time.Time,

func handleSuccessfulOperationResult(

node *hsm.Node,

operation Operation,

result *commonpb.Payload,

attemptTime *time.Time,

rodrigozhou · 2024-04-17T21:42:05Z

plugins/nexusoperations/frontend/handler.go

+		tag.WorkflowRunID(completion.GetRunId()),
+	)
+	if completion.GetNamespaceId() != ns.ID().String() {
+		logger.Error("namespace ID in token doesn't match the token", tag.WorkflowNamespaceID(ns.ID().String()), tag.Error(err), tag.NewStringTag("completion-namespace-id", completion.GetNamespaceId()))


nit: avoid very long lines

Suggested change

logger.Error("namespace ID in token doesn't match the token", tag.WorkflowNamespaceID(ns.ID().String()), tag.Error(err), tag.NewStringTag("completion-namespace-id", completion.GetNamespaceId()))

logger.Error(

"namespace ID in token doesn't match the token",

tag.WorkflowNamespaceID(ns.ID().String()),

tag.Error(err),

tag.NewStringTag("completion-namespace-id", completion.GetNamespaceId()),

)

rodrigozhou · 2024-04-18T15:10:30Z

plugins/nexusoperations/frontend/handler.go

+	if !h.Config.Enabled() {
+		h.MetricsHandler.Counter(metrics.NexusCompletionRequestPreProcessErrors.Name()).Record(1)


Suggested change

if !h.Config.Enabled() {

h.MetricsHandler.Counter(metrics.NexusCompletionRequestPreProcessErrors.Name()).Record(1)

reqPreProcessErrorsCounter := h.MetricsHandler.Counter(metrics.NexusCompletionRequestPreProcessErrors.Name())

if !h.Config.Enabled() {

reqPreProcessErrorsCounter.Record(1)

rodrigozhou · 2024-04-18T15:12:00Z

plugins/nexusoperations/frontend/handler.go

+			metrics.OperationTag(apiName),
+			metrics.NamespaceTag(nsName),
+		),
+		requestStartTime: time.Now(),


Wouldn't it be more precise if you recorded the start time at the top of this function?

rodrigozhou · 2024-04-19T17:47:55Z

plugins/nexusoperations/frontend/handler.go

+	} else {
+		if c.outcomeTag != nil {
+			c.metricsHandler = c.metricsHandler.WithTags(c.outcomeTag)
+		} else {
+			var he *nexus.HandlerError
+			if errors.As(*errPtr, &he) {
+				c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_" + strings.ToLower(string(he.Type))))
+			} else {
+				c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_internal"))
+			}
+		}
+	}


Suggested change

} else {

if c.outcomeTag != nil {

c.metricsHandler = c.metricsHandler.WithTags(c.outcomeTag)

} else {

var he *nexus.HandlerError

if errors.As(*errPtr, &he) {

c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_" + strings.ToLower(string(he.Type))))

} else {

c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_internal"))

}

}

}

} else if c.outcomeTag != nil {

c.metricsHandler = c.metricsHandler.WithTags(c.outcomeTag)

} else {

var he *nexus.HandlerError

if errors.As(*errPtr, &he) {

c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_" + strings.ToLower(string(he.Type))))

} else {

c.metricsHandler = c.metricsHandler.WithTags(metrics.NexusOutcomeTag("error_internal"))

}

}

rodrigozhou · 2024-04-19T19:23:46Z

service/history/workflow/state_transition_history.go

-// not referencing a stale state or that the task itself is not stale. For example, if the state has a history of
-// `[{v: 1, t: 3}, {v: 2, t: 5}]`, task A `{v: 2, t: 4}` **is not** referencing stale state because for version `2`
+// When a task or API request is being processed, the history is compared with the imprinted state reference to verify
+// that the state is not stale or that the task or request itself is not stale. For example, if the state has a history


nit

Suggested change

// that the state is not stale or that the task or request itself is not stale. For example, if the state has a history

// that the state is not stale or that the task/request itself is not stale. For example, if the state has a history

## What changed? [Operation state machine](#5545) [Nexus command processing](#5546) [Implement Nexus start operation task executors](#5686) [Implement Nexus async completion endpoint](#5735) Upgrade API to support full E2E flow temporalio/api#363

bergundy requested a review from a team as a code owner April 16, 2024 21:00

bergundy force-pushed the bergundy/async-operation-completion branch from 4a1ba13 to 4bb9cb4 Compare April 16, 2024 21:27

Implement Nexus async completion endpoint

7c8d04c

bergundy force-pushed the bergundy/async-operation-completion branch from 4bb9cb4 to 7c8d04c Compare April 16, 2024 23:06

yycptt reviewed Apr 18, 2024

View reviewed changes

Address review comments

4d20fd8

bergundy force-pushed the bergundy/async-operation-completion branch from f2367fd to 4d20fd8 Compare April 19, 2024 00:15

Add pending nexus operations to describe response (#5749)

eb72cda

## What changed? Added pending nexus operations to describe response. I was debating if I want to allow components to register a describe hook but decided to just inline this for now.

yycptt approved these changes Apr 19, 2024

View reviewed changes

Fix comment

c57e3e1

bergundy merged commit 8083c39 into nexus Apr 19, 2024
5 checks passed

bergundy deleted the bergundy/async-operation-completion branch April 19, 2024 18:11

rodrigozhou reviewed Apr 19, 2024

View reviewed changes

bergundy mentioned this pull request Apr 23, 2024

Nexus E2E implementation #5779

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement Nexus async completion endpoint #5735

Implement Nexus async completion endpoint #5735

bergundy commented Apr 16, 2024

yycptt Apr 18, 2024

bergundy Apr 18, 2024

yycptt Apr 19, 2024

yycptt Apr 18, 2024 •

edited

Loading

bergundy Apr 18, 2024

pdoerner Apr 18, 2024

bergundy Apr 18, 2024

bergundy commented Apr 19, 2024

yycptt Apr 19, 2024

bergundy Apr 19, 2024

yycptt Apr 19, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 17, 2024

rodrigozhou Apr 18, 2024

rodrigozhou Apr 18, 2024

rodrigozhou Apr 19, 2024

rodrigozhou Apr 19, 2024

	NexusCompletionLatencyHistogram = NewCounterDef(
	NexusCompletionLatency = NewTimerDef(

		if !h.Config.Enabled() {
		h.MetricsHandler.Counter(metrics.NexusCompletionRequestPreProcessErrors.Name()).Record(1)

	// that the state is not stale or that the task or request itself is not stale. For example, if the state has a history
	// that the state is not stale or that the task/request itself is not stale. For example, if the state has a history

Implement Nexus async completion endpoint #5735

Implement Nexus async completion endpoint #5735

Conversation

bergundy commented Apr 16, 2024

What changed?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yycptt Apr 18, 2024 • edited Loading

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 commented Apr 19, 2024

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yycptt Apr 18, 2024 •

edited

Loading