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]>
116 lines
12 KiB
Go
116 lines
12 KiB
Go
package frontend
|
|
|
|
import (
|
|
"go.temporal.io/api/serviceerror"
|
|
)
|
|
|
|
var (
|
|
errInvalidTaskToken = serviceerror.NewInvalidArgument("Invalid TaskToken.")
|
|
errDeserializingToken = serviceerror.NewInvalidArgument("Error deserializing task token.")
|
|
errWorkflowIDNotSet = serviceerror.NewInvalidArgument("WorkflowId is not set on request.")
|
|
errExecutionNotSet = serviceerror.NewInvalidArgument("Execution is not set on request.")
|
|
errActivityIDNotSet = serviceerror.NewInvalidArgument("ActivityId is not set on request.")
|
|
errActivityIDOrTypeNotSet = serviceerror.NewInvalidArgument("Either Activity.Id or Activity.Type should be set on request.")
|
|
errSignalNameNotSet = serviceerror.NewInvalidArgument("SignalName is not set on request.")
|
|
errInvalidRunID = serviceerror.NewInvalidArgument("Invalid RunId.")
|
|
errQueryNotSet = serviceerror.NewInvalidArgument("WorkflowQuery is not set on request.")
|
|
errQueryTypeNotSet = serviceerror.NewInvalidArgument("QueryType is not set on request.")
|
|
errRequestNotSet = serviceerror.NewInvalidArgument("Request is nil.")
|
|
errRequestIDNotSet = serviceerror.NewInvalidArgument("RequestId is not set on request.")
|
|
errWorkflowTypeNotSet = serviceerror.NewInvalidArgument("WorkflowType is not set on request.")
|
|
errQueryDisallowedForNamespace = serviceerror.NewInvalidArgument("Namespace is not allowed to query, please contact temporal team to re-enable queries.")
|
|
errClusterNameNotSet = serviceerror.NewInvalidArgument("Cluster name is not set.")
|
|
errEmptyReplicationInfo = serviceerror.NewInvalidArgument("Replication task info is not set.")
|
|
errTaskRangeNotSet = serviceerror.NewInvalidArgument("Task range is not set")
|
|
errHistoryNotFound = serviceerror.NewInvalidArgument("Requested workflow history not found, may have passed retention period.")
|
|
errNamespaceTooLong = serviceerror.NewInvalidArgument("Namespace length exceeds limit.")
|
|
errWorkflowTypeTooLong = serviceerror.NewInvalidArgument("WorkflowType length exceeds limit.")
|
|
errWorkflowIDTooLong = serviceerror.NewInvalidArgument("WorkflowId length exceeds limit.")
|
|
errWorkflowRuleIDTooLong = serviceerror.NewInvalidArgument("Workflow Rule Id length exceeds limit.")
|
|
errSignalNameTooLong = serviceerror.NewInvalidArgument("SignalName length exceeds limit.")
|
|
errRequestIDTooLong = serviceerror.NewInvalidArgument("RequestId length exceeds limit.")
|
|
errIdentityTooLong = serviceerror.NewInvalidArgument("Identity length exceeds limit.")
|
|
errNotesTooLong = serviceerror.NewInvalidArgument("Schedule notes exceeds limit.")
|
|
errEarliestTimeIsGreaterThanLatestTime = serviceerror.NewInvalidArgument("EarliestTime in StartTimeFilter should not be larger than LatestTime.")
|
|
errClusterIsNotConfiguredForVisibilityArchival = serviceerror.NewInvalidArgument("Cluster is not configured for visibility archival.")
|
|
errClusterIsNotConfiguredForReadingArchivalVisibility = serviceerror.NewInvalidArgument("Cluster is not configured for reading archived visibility records.")
|
|
errNamespaceIsNotConfiguredForVisibilityArchival = serviceerror.NewInvalidArgument("Namespace is not configured for visibility archival.")
|
|
errSearchAttributesNotSet = serviceerror.NewInvalidArgument("SearchAttributes are not set on request.")
|
|
errInvalidPageSize = serviceerror.NewInvalidArgument("Invalid PageSize.") // DEPRECATED
|
|
errInvalidEventQueryRange = serviceerror.NewInvalidArgument("Invalid event query range.") // DEPRECATED
|
|
errDLQTypeIsNotSupported = serviceerror.NewInvalidArgument("The DLQ type is not supported.")
|
|
errFailureMustHaveApplicationFailureInfo = serviceerror.NewInvalidArgument("Failure must have ApplicationFailureInfo.")
|
|
errStatusFilterMustBeNotRunning = serviceerror.NewInvalidArgument("StatusFilter must be specified and must be not Running.")
|
|
errCronNotAllowed = serviceerror.NewInvalidArgument("Scheduled workflow must not contain CronSchedule")
|
|
errIDReusePolicyNotAllowed = serviceerror.NewInvalidArgument("Scheduled workflow must not contain WorkflowIDReusePolicy")
|
|
errBatchJobIDNotSet = serviceerror.NewInvalidArgument("JobId is not set on request.")
|
|
errScheduleIDNotSet = serviceerror.NewInvalidArgument("ScheduleId is not set on request.")
|
|
errIdentityNotSet = serviceerror.NewInvalidArgument("Identity is not set on request.")
|
|
errMigrationTargetNotSet = serviceerror.NewInvalidArgument("Target is not set on request.")
|
|
errNamespaceNotSet = serviceerror.NewInvalidArgument("Namespace is not set on request.")
|
|
errReasonNotSet = serviceerror.NewInvalidArgument("Reason is not set on request.")
|
|
errRaceConditionAddingSearchAttributes = serviceerror.NewUnavailable("Generated search attributes mapping unavailable.")
|
|
errUseVersioningWithoutBuildId = serviceerror.NewInvalidArgument("WorkerVersionStamp must be present if UseVersioning is true.")
|
|
errUseVersioningWithoutNormalName = serviceerror.NewInvalidArgument("NormalName must be set on sticky queue when UseVersioning is true or DeploymentOptions are set.")
|
|
errBuildIdTooLong = serviceerror.NewInvalidArgument("Build ID exceeds configured limit.workerBuildIdSize, use a shorter build ID.")
|
|
errDeploymentOptionsNotSet = serviceerror.NewInvalidArgument("Deployment name and build id must be set together.")
|
|
errUseEnhancedDescribeOnStickyQueue = serviceerror.NewInvalidArgument("Enhanced DescribeTaskQueue is not valid for a sticky queue, use api_mode=UNSPECIFIED or a normal queue.")
|
|
errUseEnhancedDescribeOnNonRootQueue = serviceerror.NewInvalidArgument("Enhanced DescribeTaskQueue is not valid for non-root queue partitions, use api_mode=UNSPECIFIED or a normal queue root name.")
|
|
errTaskQueuePartitionInvalid = serviceerror.NewInvalidArgument("Task Queue Partition invalid, use a different Task Queue or Task Queue Type")
|
|
errMultiOpWorkflowIdInconsistent = serviceerror.NewInvalidArgument("WorkflowId is not consistent with previous operation(s).")
|
|
errMultiOpStartCronSchedule = serviceerror.NewInvalidArgument("CronSchedule is not allowed.")
|
|
errMultiOpUpdateFirstExecutionRunId = serviceerror.NewInvalidArgument("FirstExecutionRunId is not allowed.")
|
|
errMultiOpUpdateExecutionRunId = serviceerror.NewInvalidArgument("RunId is not allowed.")
|
|
errMultiOpEagerWorkflow = serviceerror.NewInvalidArgument("RequestEagerExecution is not supported.")
|
|
errMultiOpStartDelay = serviceerror.NewInvalidArgument("WorkflowStartDelay is not supported.")
|
|
errMultiOpNamespaceMismatch = serviceerror.NewInvalidArgument("Operation namespace did not match request's namespace.")
|
|
errMultiOpNotStartAndUpdate = serviceerror.NewInvalidArgument("Operations have to be exactly [Start, Update].")
|
|
errMultiOpAborted = serviceerror.NewMultiOperationAborted("Operation was aborted.")
|
|
|
|
errUpdateMetaNotSet = serviceerror.NewInvalidArgument("Update meta is not set on request.")
|
|
errUpdateInputNotSet = serviceerror.NewInvalidArgument("Update input is not set on request.")
|
|
errUpdateNameNotSet = serviceerror.NewInvalidArgument("Update name is not set on request.")
|
|
errUpdateIDTooLong = serviceerror.NewInvalidArgument("UpdateId length exceeds limit.")
|
|
errUpdateRefNotSet = serviceerror.NewInvalidArgument("UpdateRef is not set on request.")
|
|
errSourceClusterNotSet = serviceerror.NewInvalidArgument("SourceCluster is not set on request.")
|
|
errTargetClusterNotSet = serviceerror.NewInvalidArgument("TargetCluster is not set on request.")
|
|
errInvalidDLQJobToken = serviceerror.NewInvalidArgument("Invalid DLQ job token.")
|
|
|
|
errPageSizeTooBigMessage = "PageSize is larger than allowed %d."
|
|
|
|
errSearchAttributeIsReservedMessage = "Search attribute %s is reserved by system."
|
|
errSearchAttributeAlreadyExistsMessage = "Search attribute %s already exists."
|
|
errSearchAttributeDoesntExistMessage = "Search attribute %s doesn't exist."
|
|
errUnknownSearchAttributeTypeMessage = "Unknown search attribute type: %v."
|
|
errUnableToGetSearchAttributesMessage = "Unable to get search attributes: %v."
|
|
errUnableToRemoveNonCustomSearchAttributesMessage = "Unable to remove non-custom search attributes: %v."
|
|
errUnableToSaveSearchAttributesMessage = "Unable to save search attributes: %v."
|
|
errUnableToStartWorkflowMessage = "Unable to start %s workflow: %v."
|
|
errWorkflowReturnedErrorMessage = "Workflow %s returned an error: %v."
|
|
errUnableConnectRemoteClusterMessage = "Unable connect to remote cluster %s with error: %v."
|
|
errInvalidRemoteClusterInfo = "Unable connect to remote cluster with invalid config: %v."
|
|
errUnableToStoreClusterInfo = "Unable to persist cluster info with error: %v."
|
|
errUnableToDeleteClusterInfo = "Unable to delete cluster info with error: %v."
|
|
errUnableToGetNamespaceInfoMessage = "Unable to get namespace %v info with error: %v"
|
|
errUnableToCreateFrontendClientMessage = "Unable to create frontend client with error: %v."
|
|
errTooManySearchAttributesMessage = "Unable to create search attributes: cannot have more than %d search attribute of type %s."
|
|
|
|
errListNotAllowed = serviceerror.NewPermissionDenied("List is disabled on this namespace.", "")
|
|
errSchedulesNotAllowed = serviceerror.NewPermissionDenied("Schedules are disabled on this namespace.", "")
|
|
|
|
errDeploymentVersionsNotAllowed = serviceerror.NewPermissionDenied("Worker Deployment Versions are disabled on this namespace.", "")
|
|
|
|
errBatchAPINotAllowed = serviceerror.NewPermissionDenied("Batch operation feature are disabled on this namespace.", "")
|
|
|
|
errUpdateWorkflowExecutionAPINotAllowed = serviceerror.NewPermissionDenied("UpdateWorkflowExecution operation is disabled on this namespace.", "")
|
|
errUpdateWorkflowExecutionAsyncAcceptedNotAllowed = serviceerror.NewPermissionDenied("UpdateWorkflowExecution issued asynchronously and waiting on update accepted is disabled on this namespace.", "")
|
|
errUpdateWorkflowExecutionAsyncAdmittedNotAllowed = serviceerror.NewPermissionDenied("UpdateWorkflowExecution issued asynchronously and waiting on update admitted is not supported.", "")
|
|
errMultiOperationAPINotAllowed = serviceerror.NewPermissionDenied("ExecuteMultiOperation API is disabled on this namespace.", "")
|
|
|
|
errWorkerVersioningV1_0NotAllowed = serviceerror.NewPermissionDenied("Worker versioning v0.1 (Version Set-based, deprecated) is disabled on this namespace.", "")
|
|
errWorkerVersioningV2_0NotAllowed = serviceerror.NewPermissionDenied("Worker versioning v0.2 (Rules-based, deprecated) is disabled on this namespace.", "")
|
|
errWorkerVersioningWorkflowAPIsNotAllowed = serviceerror.NewPermissionDenied("Worker versioning in workflow progress APIs is disabled on this namespace.", "")
|
|
|
|
errListHistoryTasksNotAllowed = serviceerror.NewPermissionDenied("ListHistoryTasks feature is disabled on this cluster.", "")
|
|
)
|