mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
Roey BermanandGitHub
b91c43893b
## 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
61 lines
1.6 KiB
Go
61 lines
1.6 KiB
Go
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination library_mock.go
|
|
|
|
package chasm
|
|
|
|
import (
|
|
"github.com/nexus-rpc/sdk-go/nexus"
|
|
"google.golang.org/grpc"
|
|
)
|
|
|
|
type (
|
|
Library interface {
|
|
Name() string
|
|
Components() []*RegistrableComponent
|
|
Tasks() []*RegistrableTask
|
|
RegisterServices(server *grpc.Server)
|
|
// NexusServices returns a list of nexus.Service instances to register with the __temporal_system Nexus endpoint.
|
|
NexusServices() []*nexus.Service
|
|
// NexusServiceProcessors returns a list of NexusServiceProcessor instances to register with the __temporal_system
|
|
// Nexus endpoint.
|
|
NexusServiceProcessors() []*NexusServiceProcessor
|
|
|
|
mustEmbedUnimplementedLibrary()
|
|
}
|
|
|
|
UnimplementedLibrary struct{}
|
|
|
|
namer interface {
|
|
Name() string
|
|
}
|
|
)
|
|
|
|
func (UnimplementedLibrary) Components() []*RegistrableComponent {
|
|
return nil
|
|
}
|
|
|
|
func (UnimplementedLibrary) Tasks() []*RegistrableTask {
|
|
return nil
|
|
}
|
|
|
|
// RegisterServices Registers the gRPC calls to the handlers of the library.
|
|
func (UnimplementedLibrary) RegisterServices(_ *grpc.Server) {
|
|
}
|
|
|
|
func (UnimplementedLibrary) NexusServices() []*nexus.Service {
|
|
return nil
|
|
}
|
|
|
|
func (UnimplementedLibrary) NexusServiceProcessors() []*NexusServiceProcessor {
|
|
return nil
|
|
}
|
|
|
|
func (UnimplementedLibrary) mustEmbedUnimplementedLibrary() {}
|
|
|
|
// FullyQualifiedName creates a fully qualified name (FQN) by combining a library name
|
|
// and a component or task name. The FQN is used to uniquely identify components and
|
|
// tasks within the CHASM framework.
|
|
// The format of the returned FQN is: "libName.name"
|
|
func FullyQualifiedName(libName, name string) string {
|
|
return libName + "." + name
|
|
}
|