mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
7fec4a9a3f
## What changed? This PR adds `SignalWithStartWorkflowExecution` as a synchronous Nexus operation exposed via `__temporal_system endpoint`, allowing workflows to signal-with-start other workflows through the CHASM Nexus operation framework. Key changes: - `chasm/lib/workflow/nexus_service.go`: `workflowServiceNexusHandler` implements the `SignalWithStartWorkflowExecution` Nexus sync operation by resolving the namespace and delegating to the History service. A `SignalWithStartOperationProcessor` handles input enrichment (namespace, request ID, links) and routing via CHASM's `NexusOperationProcessorResult`. - `chasm/lib/workflow/library.go` — library now holds the `workflowServiceNexusHandler`, the workflow `Config`, the SA mapper provider, and the SA validator. `newLibrary` (used by fx) takes those dependencies, while the public `NewLibrary` keeps its old signature for external callers. Adds `NexusServices()` so the library registers its Nexus service via CHASM. - `chasm/lib/workflow/validator.go`: `RequestValidator` consolidates the `SignalWithStartWorkflowExecution` validation logic (previously inlined in `WorkflowHandler`) into a reusable, injectable struct. This same validator is used by both the frontend handler and the new CHASM processor. - common/dynamicconfig/constants.go — adds `EnableSignalWithStartFromWorkflow` (namespace-scoped, default false). - `service/frontend/workflow_handler.go`: Removed the `SignalWithStart` validation block - `service/history/fx.go`: Provides a `HistoryServiceServerProvider` so the CHASM workflow library can call the history handler directly. - `temporal/fx.go`: Removes the now-redundant `ChasmLibraryOptions` grouping; each service module registers its own CHASM libraries. - `components/nexusoperations/workflow/commands.go`: `NotFound` and `InvalidArgument` errors during Nexus command handling are now surfaced as workflow task failures instead of being treated as transient handler errors. - `common/payloads`: Adds `EncodeSingle`, `MustEncodeSingle`, and `MustEncode` helpers used in tests. - `cmd/tools/getproto`: Adds support for nexus-proto-annotations proto imports. - `tests/signal_with_start_from_workflow_test.go`: Functional test suite covering the happy path, duplicate detection, conflict policies, and validation rejection for the new Nexus operation. ## Why? This functionality is one of our most requested GitHub issues. ## How did you test it? - [X] built - [X] run locally and tested manually - [ ] covered by existing tests - [X] added new unit test(s) - [X] added new functional test(s) ## Potential risks The history service now directly exposes a `HistoryServiceServer` interface via `fx` for injection into the CHASM workflow library. This tight coupling between the CHASM workflow library and the history handler could complicate future layering — callers outside the history service should not adopt this pattern. The feature is gated by `history.enableSignalWithStartFromWorkflow` for rollout. --------- Co-authored-by: Roey Berman <[email protected]>
35 lines
989 B
Go
35 lines
989 B
Go
package workflow
|
|
|
|
import (
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/chasm/lib/nexusoperation"
|
|
"go.uber.org/fx"
|
|
)
|
|
|
|
var Module = fx.Module(
|
|
"chasm.lib.workflow",
|
|
fx.Provide(NewConfig),
|
|
fx.Provide(NewRegistry),
|
|
fx.Provide(newLibrary),
|
|
fx.Invoke(func(
|
|
chasmRegistry *chasm.Registry,
|
|
library *library,
|
|
config *nexusoperation.Config,
|
|
) error {
|
|
if err := library.registry.Register(
|
|
newNexusLibrary(config, chasmRegistry.NexusEndpointProcessor),
|
|
); err != nil {
|
|
return err
|
|
}
|
|
return chasmRegistry.Register(library)
|
|
}),
|
|
)
|
|
|
|
// HistoryHandlerModule wires the workflow library's Nexus handler to the
|
|
// history service. Only include this in services that provide
|
|
// historyservice.HistoryServiceServer (the history service).
|
|
var HistoryHandlerModule = fx.Invoke(func(library *library, historyHandler historyservice.HistoryServiceServer) {
|
|
library.workflowServiceNexusHandler.setHistoryHandler(historyHandler)
|
|
})
|