Nexus Standalone: Poll

Author: stephanos

chasm/lib/nexusoperation/config.go

@@ -10,6 +10,20 @@ import (
 	"go.temporal.io/server/common/rpc/interceptor"
 )
 
+var LongPollTimeout = dynamicconfig.NewNamespaceDurationSetting(
+	"nexusoperation.longPollTimeout",
+	20*time.Second,
+	`Timeout for nexus operation long-poll requests.`,
+)
+
+var LongPollBuffer = dynamicconfig.NewNamespaceDurationSetting(
+	"nexusoperation.longPollBuffer",
+	time.Second,
+	`A buffer used to adjust the nexus operation long-poll timeouts.
+ Specifically, nexus operation long-poll requests are timed out at a time which leaves at least the buffer's duration
+ remaining before the caller's deadline, if permitted by the caller's deadline.`,
+)
+
 var Enabled = dynamicconfig.NewNamespaceBoolSetting(
 	"nexusoperation.enableStandalone",
 	false,
@@ -165,6 +179,8 @@ type Config struct {
 	Enabled                             dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	ChasmEnabled                        dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	ChasmNexusEnabled                   dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	LongPollBuffer                      dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	LongPollTimeout                     dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	RequestTimeout                      dynamicconfig.DurationPropertyFnWithDestinationFilter
 	MinRequestTimeout                   dynamicconfig.DurationPropertyFnWithNamespaceFilter
 	MaxConcurrentOperations             dynamicconfig.IntPropertyFnWithNamespaceFilter
@@ -190,6 +206,8 @@ func configProvider(dc *dynamicconfig.Collection) *Config {
 		Enabled:                            Enabled.Get(dc),
 		ChasmEnabled:                       dynamicconfig.EnableChasm.Get(dc),
 		ChasmNexusEnabled:                  ChasmNexusEnabled.Get(dc),
+		LongPollBuffer:                     LongPollBuffer.Get(dc),
+		LongPollTimeout:                    LongPollTimeout.Get(dc),
 		RequestTimeout:                     RequestTimeout.Get(dc),
 		MinRequestTimeout:                  MinRequestTimeout.Get(dc),
 		MaxConcurrentOperations:            MaxConcurrentOperations.Get(dc),

chasm/lib/nexusoperation/frontend.go

@@ -121,11 +121,29 @@ func (h *frontendHandler) DescribeNexusOperationExecution(
 	return resp.GetFrontendResponse(), err
 }
 
-func (h *frontendHandler) PollNexusOperationExecution(_ context.Context, req *workflowservice.PollNexusOperationExecutionRequest) (*workflowservice.PollNexusOperationExecutionResponse, error) {
+// PollNexusOperationExecution long-polls for a Nexus operation to reach a specific stage.
+func (h *frontendHandler) PollNexusOperationExecution(
+	ctx context.Context,
+	req *workflowservice.PollNexusOperationExecutionRequest,
+) (*workflowservice.PollNexusOperationExecutionResponse, error) {
 	if !h.isStandaloneNexusOperationEnabled(req.GetNamespace()) {
 		return nil, ErrStandaloneNexusOperationDisabled
 	}
-	return nil, serviceerror.NewUnimplemented("PollNexusOperationExecution not implemented")
+
+	if err := validateAndNormalizePollRequest(req, h.config); err != nil {
+		return nil, err
+	}
+
+	namespaceID, err := h.namespaceRegistry.GetNamespaceID(namespace.Name(req.GetNamespace()))
+	if err != nil {
+		return nil, err
+	}
+
+	resp, err := h.client.PollNexusOperation(ctx, &nexusoperationpb.PollNexusOperationRequest{
+		NamespaceId:     namespaceID.String(),
+		FrontendRequest: req,
+	})
+	return resp.GetFrontendResponse(), err
 }
 
 func (h *frontendHandler) ListNexusOperationExecutions(