mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
## 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
209 lines
7.4 KiB
Go
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
|
|
}
|