Files
tasks/common/nexus/link_converter.go
Roey BermanandGitHub b91c43893b System nexus endpoint (#9002)
## Overview

This feature introduces a special internal Nexus endpoint called
`__temporal_system` that enables adding functionality to workflows
**without requiring new workflow commands and events**. Operations on
this endpoint are routed internally within Temporal's history service
rather than via external HTTP calls.

## Key Components

### 1. System Endpoint Infrastructure

- **Endpoint Name**: `__temporal_system` (constant in
`common/nexus/constants.go:6`)
- **Callback URL**: `temporal://system` for internal routing
- **New History Service RPCs**
(`proto/internal/temporal/server/api/historyservice/v1/service.proto:433-437`):
  - `StartNexusOperation` - Starts operations on the system endpoint
  - `CancelNexusOperation` - Cancels operations on the system endpoint

### 2. Operation Processor Framework

A new processor pattern (`chasm/nexus_operation_processor.go`) that
allows CHASM libraries to:

- **Validate and transform input**: Processors can validate operation
inputs and set default values
- **Determine routing**: Each processor returns a routing key that
determines which history shard handles the operation
- **Re-serialize input**: Mutated inputs can be re-serialized to persist
default values

**Routing strategies**:
- `NexusOperationRoutingKeyExecution` - Routes to the shard owning a
specific workflow execution
- `NexusOperationRoutingKeyRandom` - Routes to a random shard

### 3. CHASM Library Integration

CHASM libraries can now provide (`chasm/library.go:16-19`):
- **`NexusServices()`**: Regular Nexus service handlers (implement the
actual operation logic)
- **`NexusServiceProcessors()`**: Input processors for validation and
routing

Example from test library (`chasm/lib/tests/nexus_service.go`):

```go
// Service handler - implements the actual operation
TestOperation = nexus.NewSyncOperation("TestOperation",
    func(ctx context.Context, input string, options nexus.StartOperationOptions) (string, error) {
        return "Hello, " + input, nil
    })

// Processor - validates input and determines routing
func (o testOperationProcessor) ProcessInput(ctx chasm.NexusOperationProcessorContext, input string)
    (*chasm.NexusOperationProcessorResult, error) {
    return &chasm.NexusOperationProcessorResult{
        RoutingKey: chasm.NexusOperationRoutingKeyExecution{
            NamespaceID: ctx.Namespace.ID().String(),
            BusinessID:  input, // Route based on input
        },
    }, nil
}
```

### 4. Execution Flow

When a workflow schedules a Nexus operation on `__temporal_system`
(`components/nexusoperations/executors.go:233-238`):

1. **Input Processing**: The processor validates input and determines
routing
2. **Internal RPC**: Instead of HTTP, calls
`HistoryClient.StartNexusOperation` with the target shard ID
3. **Handler Execution**: The history service invokes the registered
Nexus handler (`service/history/handler.go:2707-2768`)
4. **Result Handling**: Supports both sync (immediate result) and async
(operation token) responses
5. **Workflow Completion**: Results flow back through the same
completion path as external Nexus operations

### 5. Benefits

✅ **No schema changes**: Add functionality without new commands/events
in workflow history
✅ **Consistent API**: Uses existing Nexus operation semantics
(sync/async, callbacks, links)
✅ **Proper routing**: Operations are intelligently routed to the correct
shard
✅ **Input validation**: Type-safe input validation and default value
handling
✅ **Future extensibility**: Foundation for direct client invocation (not
yet implemented)

### 6. Technical Details

- **Error handling** (`components/nexusoperations/executors.go:444`):
Non-retryable service errors are properly handled and fail operations
immediately
- **Metrics**: System operations are tracked separately with
`DestinationTag` set to the endpoint name
- **Link conversion**: Helper functions convert between Nexus SDK links
and protobuf links (`common/nexus/util.go:17-46`)
- **Operation token handling**: Moved link converters to common package
for reuse (`common/nexus/link_converter.go`)

### 7. Current Limitations

- Only accessible from workflows (via `ScheduleNexusOperation` command)
- Direct client invocation not yet implemented
- Headers not supported for system endpoint operations

## Test Coverage

New test (`tests/nexus_workflow_test.go:2763-2843`) demonstrates:
- Scheduling operation on `__temporal_system` endpoint
- Synchronous operation completion
- Result propagation back to workflow

## Architecture

This architecture provides a clean, extensible way to add internal
functionality while maintaining compatibility with Temporal's existing
workflow execution model. The system endpoint acts as a bridge between
workflows and internal CHASM components, enabling:

- **Extensibility**: New operations can be added by implementing CHASM
libraries
- **Type safety**: Input validation happens before operations are routed
- **Scalability**: Intelligent routing ensures operations land on the
correct shard
- **Consistency**: Same execution model as external Nexus operations
2026-02-16 15:53:51 -08:00

209 lines
7.4 KiB
Go

// The MIT License
//
// Copyright (c) 2024 Temporal Technologies Inc. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
// This file is duplicated in sdk-go/temporalnexus/link_converter.go.
// Any changes here or there must be replicated. This is temporary until the
// temporal repo updates to the most recent SDK version.
package nexus
import (
"errors"
"fmt"
"net/url"
"regexp"
"strconv"
"github.com/nexus-rpc/sdk-go/nexus"
commonpb "go.temporal.io/api/common/v1"
enumspb "go.temporal.io/api/enums/v1"
)
const (
urlSchemeTemporalKey = "temporal"
urlPathNamespaceKey = "namespace"
urlPathWorkflowIDKey = "workflowID"
urlPathRunIDKey = "runID"
urlPathTemplate = "/namespaces/%s/workflows/%s/%s/history"
linkWorkflowEventReferenceTypeKey = "referenceType"
linkEventIDKey = "eventID"
linkEventTypeKey = "eventType"
linkRequestIDKey = "requestID"
)
var (
rePatternNamespace = fmt.Sprintf(`(?P<%s>[^/]+)`, urlPathNamespaceKey)
rePatternWorkflowID = fmt.Sprintf(`(?P<%s>[^/]+)`, urlPathWorkflowIDKey)
rePatternRunID = fmt.Sprintf(`(?P<%s>[^/]+)`, urlPathRunIDKey)
urlPathRE = regexp.MustCompile(fmt.Sprintf(
`^/namespaces/%s/workflows/%s/%s/history$`,
rePatternNamespace,
rePatternWorkflowID,
rePatternRunID,
))
eventReferenceType = string((&commonpb.Link_WorkflowEvent_EventReference{}).ProtoReflect().Descriptor().Name())
requestIDReferenceType = string((&commonpb.Link_WorkflowEvent_RequestIdReference{}).ProtoReflect().Descriptor().Name())
)
// ConvertLinkWorkflowEventToNexusLink converts a Link_WorkflowEvent type to Nexus Link.
//
// NOTE: Experimental
func ConvertLinkWorkflowEventToNexusLink(we *commonpb.Link_WorkflowEvent) nexus.Link {
u := &url.URL{
Scheme: urlSchemeTemporalKey,
Path: fmt.Sprintf(urlPathTemplate, we.GetNamespace(), we.GetWorkflowId(), we.GetRunId()),
RawPath: fmt.Sprintf(
urlPathTemplate,
url.PathEscape(we.GetNamespace()),
url.PathEscape(we.GetWorkflowId()),
url.PathEscape(we.GetRunId()),
),
}
switch ref := we.GetReference().(type) {
case *commonpb.Link_WorkflowEvent_EventRef:
u.RawQuery = convertLinkWorkflowEventEventReferenceToURLQuery(ref.EventRef)
case *commonpb.Link_WorkflowEvent_RequestIdRef:
u.RawQuery = convertLinkWorkflowEventRequestIdReferenceToURLQuery(ref.RequestIdRef)
}
return nexus.Link{
URL: u,
Type: string(we.ProtoReflect().Descriptor().FullName()),
}
}
// ConvertNexusLinkToLinkWorkflowEvent converts a Nexus Link to Link_WorkflowEvent.
//
// NOTE: Experimental
func ConvertNexusLinkToLinkWorkflowEvent(link nexus.Link) (*commonpb.Link_WorkflowEvent, error) {
we := &commonpb.Link_WorkflowEvent{}
if link.Type != string(we.ProtoReflect().Descriptor().FullName()) {
return nil, fmt.Errorf(
"cannot parse link type %q to %q",
link.Type,
we.ProtoReflect().Descriptor().FullName(),
)
}
if link.URL.Scheme != urlSchemeTemporalKey {
return nil, fmt.Errorf(
"failed to parse link to Link_WorkflowEvent: invalid scheme: %s",
link.URL.Scheme,
)
}
matches := urlPathRE.FindStringSubmatch(link.URL.EscapedPath())
if len(matches) != 4 {
return nil, errors.New("failed to parse link to Link_WorkflowEvent: malformed URL path")
}
var err error
we.Namespace, err = url.PathUnescape(matches[urlPathRE.SubexpIndex(urlPathNamespaceKey)])
if err != nil {
return nil, fmt.Errorf("failed to parse link to Link_WorkflowEvent: %w", err)
}
we.WorkflowId, err = url.PathUnescape(matches[urlPathRE.SubexpIndex(urlPathWorkflowIDKey)])
if err != nil {
return nil, fmt.Errorf("failed to parse link to Link_WorkflowEvent: %w", err)
}
we.RunId, err = url.PathUnescape(matches[urlPathRE.SubexpIndex(urlPathRunIDKey)])
if err != nil {
return nil, fmt.Errorf("failed to parse link to Link_WorkflowEvent: %w", err)
}
switch refType := link.URL.Query().Get(linkWorkflowEventReferenceTypeKey); refType {
case eventReferenceType:
eventRef, err := convertURLQueryToLinkWorkflowEventEventReference(link.URL.Query())
if err != nil {
return nil, fmt.Errorf("failed to parse link to Link_WorkflowEvent: %w", err)
}
we.Reference = &commonpb.Link_WorkflowEvent_EventRef{
EventRef: eventRef,
}
case requestIDReferenceType:
requestIDRef, err := convertURLQueryToLinkWorkflowEventRequestIdReference(link.URL.Query())
if err != nil {
return nil, fmt.Errorf("failed to parse link to Link_WorkflowEvent: %w", err)
}
we.Reference = &commonpb.Link_WorkflowEvent_RequestIdRef{
RequestIdRef: requestIDRef,
}
default:
return nil, fmt.Errorf(
"failed to parse link to Link_WorkflowEvent: unknown reference type: %q",
refType,
)
}
return we, nil
}
func convertLinkWorkflowEventEventReferenceToURLQuery(eventRef *commonpb.Link_WorkflowEvent_EventReference) string {
values := url.Values{}
values.Set(linkWorkflowEventReferenceTypeKey, eventReferenceType)
if eventRef.GetEventId() > 0 {
values.Set(linkEventIDKey, strconv.FormatInt(eventRef.GetEventId(), 10))
}
values.Set(linkEventTypeKey, eventRef.GetEventType().String())
return values.Encode()
}
func convertURLQueryToLinkWorkflowEventEventReference(queryValues url.Values) (*commonpb.Link_WorkflowEvent_EventReference, error) {
var err error
eventRef := &commonpb.Link_WorkflowEvent_EventReference{}
eventIDValue := queryValues.Get(linkEventIDKey)
if eventIDValue != "" {
eventRef.EventId, err = strconv.ParseInt(queryValues.Get(linkEventIDKey), 10, 64)
if err != nil {
return nil, err
}
}
eventRef.EventType, err = enumspb.EventTypeFromString(queryValues.Get(linkEventTypeKey))
if err != nil {
return nil, err
}
return eventRef, nil
}
func convertLinkWorkflowEventRequestIdReferenceToURLQuery(requestIDRef *commonpb.Link_WorkflowEvent_RequestIdReference) string {
values := url.Values{}
values.Set(linkWorkflowEventReferenceTypeKey, requestIDReferenceType)
values.Set(linkRequestIDKey, requestIDRef.GetRequestId())
values.Set(linkEventTypeKey, requestIDRef.GetEventType().String())
return values.Encode()
}
func convertURLQueryToLinkWorkflowEventRequestIdReference(queryValues url.Values) (*commonpb.Link_WorkflowEvent_RequestIdReference, error) {
var err error
requestIDRef := &commonpb.Link_WorkflowEvent_RequestIdReference{
RequestId: queryValues.Get(linkRequestIDKey),
}
requestIDRef.EventType, err = enumspb.EventTypeFromString(queryValues.Get(linkEventTypeKey))
if err != nil {
return nil, err
}
return requestIDRef, nil
}