mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07: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
175 lines
5.3 KiB
Go
175 lines
5.3 KiB
Go
// Code generated by MockGen. DO NOT EDIT.
|
|
// Source: library.go
|
|
//
|
|
// Generated by this command:
|
|
//
|
|
// mockgen -package chasm -source library.go -destination library_mock.go
|
|
//
|
|
|
|
// Package chasm is a generated GoMock package.
|
|
package chasm
|
|
|
|
import (
|
|
reflect "reflect"
|
|
|
|
nexus "github.com/nexus-rpc/sdk-go/nexus"
|
|
gomock "go.uber.org/mock/gomock"
|
|
grpc "google.golang.org/grpc"
|
|
)
|
|
|
|
// MockLibrary is a mock of Library interface.
|
|
type MockLibrary struct {
|
|
ctrl *gomock.Controller
|
|
recorder *MockLibraryMockRecorder
|
|
isgomock struct{}
|
|
}
|
|
|
|
// MockLibraryMockRecorder is the mock recorder for MockLibrary.
|
|
type MockLibraryMockRecorder struct {
|
|
mock *MockLibrary
|
|
}
|
|
|
|
// NewMockLibrary creates a new mock instance.
|
|
func NewMockLibrary(ctrl *gomock.Controller) *MockLibrary {
|
|
mock := &MockLibrary{ctrl: ctrl}
|
|
mock.recorder = &MockLibraryMockRecorder{mock}
|
|
return mock
|
|
}
|
|
|
|
// EXPECT returns an object that allows the caller to indicate expected use.
|
|
func (m *MockLibrary) EXPECT() *MockLibraryMockRecorder {
|
|
return m.recorder
|
|
}
|
|
|
|
// Components mocks base method.
|
|
func (m *MockLibrary) Components() []*RegistrableComponent {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "Components")
|
|
ret0, _ := ret[0].([]*RegistrableComponent)
|
|
return ret0
|
|
}
|
|
|
|
// Components indicates an expected call of Components.
|
|
func (mr *MockLibraryMockRecorder) Components() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Components", reflect.TypeOf((*MockLibrary)(nil).Components))
|
|
}
|
|
|
|
// Name mocks base method.
|
|
func (m *MockLibrary) Name() string {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "Name")
|
|
ret0, _ := ret[0].(string)
|
|
return ret0
|
|
}
|
|
|
|
// Name indicates an expected call of Name.
|
|
func (mr *MockLibraryMockRecorder) Name() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockLibrary)(nil).Name))
|
|
}
|
|
|
|
// NexusServiceProcessors mocks base method.
|
|
func (m *MockLibrary) NexusServiceProcessors() []*NexusServiceProcessor {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "NexusServiceProcessors")
|
|
ret0, _ := ret[0].([]*NexusServiceProcessor)
|
|
return ret0
|
|
}
|
|
|
|
// NexusServiceProcessors indicates an expected call of NexusServiceProcessors.
|
|
func (mr *MockLibraryMockRecorder) NexusServiceProcessors() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NexusServiceProcessors", reflect.TypeOf((*MockLibrary)(nil).NexusServiceProcessors))
|
|
}
|
|
|
|
// NexusServices mocks base method.
|
|
func (m *MockLibrary) NexusServices() []*nexus.Service {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "NexusServices")
|
|
ret0, _ := ret[0].([]*nexus.Service)
|
|
return ret0
|
|
}
|
|
|
|
// NexusServices indicates an expected call of NexusServices.
|
|
func (mr *MockLibraryMockRecorder) NexusServices() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NexusServices", reflect.TypeOf((*MockLibrary)(nil).NexusServices))
|
|
}
|
|
|
|
// RegisterServices mocks base method.
|
|
func (m *MockLibrary) RegisterServices(server *grpc.Server) {
|
|
m.ctrl.T.Helper()
|
|
m.ctrl.Call(m, "RegisterServices", server)
|
|
}
|
|
|
|
// RegisterServices indicates an expected call of RegisterServices.
|
|
func (mr *MockLibraryMockRecorder) RegisterServices(server any) *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterServices", reflect.TypeOf((*MockLibrary)(nil).RegisterServices), server)
|
|
}
|
|
|
|
// Tasks mocks base method.
|
|
func (m *MockLibrary) Tasks() []*RegistrableTask {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "Tasks")
|
|
ret0, _ := ret[0].([]*RegistrableTask)
|
|
return ret0
|
|
}
|
|
|
|
// Tasks indicates an expected call of Tasks.
|
|
func (mr *MockLibraryMockRecorder) Tasks() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Tasks", reflect.TypeOf((*MockLibrary)(nil).Tasks))
|
|
}
|
|
|
|
// mustEmbedUnimplementedLibrary mocks base method.
|
|
func (m *MockLibrary) mustEmbedUnimplementedLibrary() {
|
|
m.ctrl.T.Helper()
|
|
m.ctrl.Call(m, "mustEmbedUnimplementedLibrary")
|
|
}
|
|
|
|
// mustEmbedUnimplementedLibrary indicates an expected call of mustEmbedUnimplementedLibrary.
|
|
func (mr *MockLibraryMockRecorder) mustEmbedUnimplementedLibrary() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "mustEmbedUnimplementedLibrary", reflect.TypeOf((*MockLibrary)(nil).mustEmbedUnimplementedLibrary))
|
|
}
|
|
|
|
// Mocknamer is a mock of namer interface.
|
|
type Mocknamer struct {
|
|
ctrl *gomock.Controller
|
|
recorder *MocknamerMockRecorder
|
|
isgomock struct{}
|
|
}
|
|
|
|
// MocknamerMockRecorder is the mock recorder for Mocknamer.
|
|
type MocknamerMockRecorder struct {
|
|
mock *Mocknamer
|
|
}
|
|
|
|
// NewMocknamer creates a new mock instance.
|
|
func NewMocknamer(ctrl *gomock.Controller) *Mocknamer {
|
|
mock := &Mocknamer{ctrl: ctrl}
|
|
mock.recorder = &MocknamerMockRecorder{mock}
|
|
return mock
|
|
}
|
|
|
|
// EXPECT returns an object that allows the caller to indicate expected use.
|
|
func (m *Mocknamer) EXPECT() *MocknamerMockRecorder {
|
|
return m.recorder
|
|
}
|
|
|
|
// Name mocks base method.
|
|
func (m *Mocknamer) Name() string {
|
|
m.ctrl.T.Helper()
|
|
ret := m.ctrl.Call(m, "Name")
|
|
ret0, _ := ret[0].(string)
|
|
return ret0
|
|
}
|
|
|
|
// Name indicates an expected call of Name.
|
|
func (mr *MocknamerMockRecorder) Name() *gomock.Call {
|
|
mr.mock.ctrl.T.Helper()
|
|
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*Mocknamer)(nil).Name))
|
|
}
|