mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Chetan GowdaandGitHub
8e624df2aa
## What changed? **Note**: This depends on https://github.com/temporalio/api/pull/653. Sending it for some early feedback. - Made changes to `WorkflowPauseInfo` (in WorkflowExecutionInfo). Reusing unused proto fields. - Added mutablestate.IsWorkflowExecutionPaused() - Added mutablestate.AddWorkflowExecutionPausedEvent() ## Why? - These changes are needed to implement pause/unpause features. ## How did you test it? - [x] built - [ ] run locally and tested manually - [ ] covered by existing tests - [x] added new unit test(s) - [ ] added new functional test(s) --- > [!NOTE] > Adds workflow-level pause/unpause with new `WorkflowPauseInfo` fields, client endpoints, history event handling, state/status validation for `PAUSED`, and tests; removes activity-level pause info. > > - **Protocol/API**: > - Redefine `persistence.v1.WorkflowPauseInfo` to `{pause_time, identity, reason, request_id}`; remove `ActivityPauseInfo` and related helpers. > - Bump dependency `go.temporal.io/api` and adjust generated code indexes. > - Update RPC metadata, redirection maps, quotas, and log tags for `PauseWorkflowExecution`/`UnpauseWorkflowExecution`. > - **Frontend Clients**: > - Add `PauseWorkflowExecution` and `UnpauseWorkflowExecution` to `client_impl`, `metric_client`, `retryable_client`, and mocks. > - **History/State**: > - Add `WorkflowExecutionPaused` event creation (`Create/Add...PausedEvent`) with buffering rules (paused allowed to buffer; unpaused not buffered). > - Implement `ApplyWorkflowExecutionPausedEvent`: set status `PAUSED`, populate `executionInfo.PauseInfo`, invalidate pending activities and workflow task via stamps. > - Rebuilder applies paused event; state transition validation supports `PAUSED` across CREATED/RUNNING/ZOMBIE. > - **Validation/Tests**: > - Allow `WORKFLOW_EXECUTION_STATUS_PAUSED` in validators; expand unit tests for create/update state/status and paused behavior. > - **Tooling**: > - `buf.yaml` breaking ignore for `executions.proto`. > > <sup>Written by [Cursor Bugbot]() for commit 951c9c4a53473a319afc5e2d62ac3cc1c85f3770. This will update automatically on new commits. Configure [here]().</sup>
110 lines
3.5 KiB
Go
110 lines
3.5 KiB
Go
package persistence
|
|
|
|
import (
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
enumsspb "go.temporal.io/server/api/enums/v1"
|
|
)
|
|
|
|
var (
|
|
validWorkflowStates = map[enumsspb.WorkflowExecutionState]struct{}{
|
|
enumsspb.WORKFLOW_EXECUTION_STATE_CREATED: {},
|
|
enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING: {},
|
|
enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED: {},
|
|
enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE: {},
|
|
enumsspb.WORKFLOW_EXECUTION_STATE_CORRUPTED: {},
|
|
}
|
|
|
|
validWorkflowStatuses = map[enumspb.WorkflowExecutionStatus]struct{}{
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_FAILED: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_CANCELED: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_TERMINATED: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_TIMED_OUT: {},
|
|
enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED: {},
|
|
}
|
|
)
|
|
|
|
// ValidateCreateWorkflowStateStatus validate workflow state and close status
|
|
func ValidateCreateWorkflowStateStatus(
|
|
state enumsspb.WorkflowExecutionState,
|
|
status enumspb.WorkflowExecutionStatus,
|
|
) error {
|
|
|
|
if err := validateWorkflowState(state); err != nil {
|
|
return err
|
|
}
|
|
if err := validateWorkflowStatus(status); err != nil {
|
|
return err
|
|
}
|
|
|
|
// validate workflow state & status
|
|
if state == enumsspb.WORKFLOW_EXECUTION_STATE_COMPLETED {
|
|
if status == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING || status == enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED {
|
|
return serviceerror.NewInternalf("Create workflow with invalid state: %v or status: %v", state, status)
|
|
}
|
|
} else {
|
|
if status != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
|
|
return serviceerror.NewInternalf("Create workflow with invalid state: %v or status: %v", state, status)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ValidateUpdateWorkflowStateStatus validate workflow state and status
|
|
func ValidateUpdateWorkflowStateStatus(
|
|
state enumsspb.WorkflowExecutionState,
|
|
status enumspb.WorkflowExecutionStatus,
|
|
) error {
|
|
|
|
if err := validateWorkflowState(state); err != nil {
|
|
return err
|
|
}
|
|
if err := validateWorkflowStatus(status); err != nil {
|
|
return err
|
|
}
|
|
|
|
// validate workflow state & status
|
|
switch state {
|
|
case enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING, enumsspb.WORKFLOW_EXECUTION_STATE_ZOMBIE:
|
|
if status != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING && status != enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED {
|
|
return serviceerror.NewInternalf("Update workflow with invalid state: %v or status: %v", state, status)
|
|
}
|
|
case enumsspb.WORKFLOW_EXECUTION_STATE_CREATED:
|
|
if status != enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING {
|
|
return serviceerror.NewInternalf("Update workflow with invalid state: %v or status: %v", state, status)
|
|
}
|
|
default:
|
|
if status == enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING || status == enumspb.WORKFLOW_EXECUTION_STATUS_PAUSED {
|
|
return serviceerror.NewInternalf("Update workflow with invalid state: %v or status: %v", state, status)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// validateWorkflowState validate workflow state
|
|
func validateWorkflowState(
|
|
state enumsspb.WorkflowExecutionState,
|
|
) error {
|
|
|
|
if _, ok := validWorkflowStates[state]; !ok {
|
|
return serviceerror.NewInternalf("Invalid workflow state: %v", state)
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// validateWorkflowStatus validate workflow status
|
|
func validateWorkflowStatus(
|
|
status enumspb.WorkflowExecutionStatus,
|
|
) error {
|
|
|
|
if _, ok := validWorkflowStatuses[status]; !ok {
|
|
return serviceerror.NewInternalf("Invalid workflow status: %v", status)
|
|
}
|
|
|
|
return nil
|
|
}
|