mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
f97a858310
## What changed? Added support for Nexus workflow update completion callbacks via CHASM. This allows a Nexus caller to be notified when a workflow update completes by attaching completion callbacks to the update request. ## Why? Nexus operations that target workflow updates need a way to receive completion notifications. Without this, a Nexus caller that sends an update has no async mechanism to learn when the update finishes. Completion callbacks enable the same async notification pattern that already exists for workflow-level Nexus operations. ## How did you test it? - [ ] 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 Touches speculative workflow updates, they are always hard to reason about. Tried to compensate with lots of test coverage. Note: Needs this API PR https://github.com/temporalio/api/pull/742/changes --- > [!NOTE] > **High Risk** > Touches workflow update state machine and mutable state event handling to persist/trigger per-update callbacks, including close/retry/reset paths, which is complex and can affect correctness of update outcomes and callback delivery. > > **Overview** > Adds **workflow update completion callbacks** via CHASM so Nexus callers can register callbacks on `UpdateWorkflowExecution` and have them fired on update completion or workflow close. > > This introduces a `WorkflowUpdate` CHASM component with new `UpdateState` protobuf (including persisted `rejection_failure`), stores update callbacks under `Workflow.Updates`, and extends callback processing to handle *update-level* callbacks on update completion, rejection (including reset/reapply), and on run transitions (retry/timeout/continue-as-new) where update callbacks must fire even if workflow-level callbacks are inherited. > > It also adds dynamic config gates/limits (`EnableWorkflowUpdateCallbacks`, `MaxCallbacksPerUpdateID`), updates `DescribeWorkflow` to surface update callback triggers, extends mutable state/history builder APIs to carry per-update callback options in `WorkflowExecutionOptionsUpdated`, and adds `Update.AttachCallbacks` logic to persist/flush callbacks (including buffering while `stateSent`, request-id dedup, and stricter validation requiring `request_id` when callbacks are present). > > <sup>Reviewed by [Cursor Bugbot]() for commit 4484fee104d73344ef5370c468008440c7df1c6a. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> --------- Co-authored-by: long-nt-tran <[email protected]>
68 lines
1.9 KiB
Go
68 lines
1.9 KiB
Go
package workflow
|
|
|
|
import (
|
|
"github.com/nexus-rpc/sdk-go/nexus"
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/chasm/lib/callback"
|
|
"go.temporal.io/server/chasm/lib/workflow/gen/workflowpb/v1"
|
|
commonnexus "go.temporal.io/server/common/nexus"
|
|
"go.temporal.io/server/common/nexus/nexusrpc"
|
|
)
|
|
|
|
type WorkflowUpdate struct {
|
|
chasm.UnimplementedComponent
|
|
|
|
*workflowpb.UpdateState
|
|
|
|
// MSPointer is a special in-memory field for accessing the underlying mutable state.
|
|
chasm.MSPointer
|
|
|
|
// Callbacks map is used to store the callbacks for the update.
|
|
Callbacks chasm.Map[string, *callback.Callback]
|
|
}
|
|
|
|
func NewWorkflowUpdate(
|
|
_ chasm.MutableContext, updateID string, msPointer chasm.MSPointer,
|
|
) *WorkflowUpdate {
|
|
return &WorkflowUpdate{
|
|
UpdateState: &workflowpb.UpdateState{
|
|
UpdateId: updateID,
|
|
},
|
|
MSPointer: msPointer,
|
|
}
|
|
}
|
|
|
|
func (u *WorkflowUpdate) LifecycleState(
|
|
_ chasm.Context,
|
|
) chasm.LifecycleState {
|
|
return chasm.LifecycleStateRunning
|
|
}
|
|
|
|
func (u *WorkflowUpdate) GetNexusCompletion(
|
|
ctx chasm.Context,
|
|
requestID string,
|
|
) (nexusrpc.CompleteOperationOptions, error) {
|
|
// If the update was rejected, return the rejection failure directly instead
|
|
// of looking up a completion event that doesn't exist.
|
|
if rf := u.GetRejectionFailure(); rf != nil {
|
|
f, err := commonnexus.TemporalFailureToNexusFailure(rf)
|
|
if err != nil {
|
|
return nexusrpc.CompleteOperationOptions{}, err
|
|
}
|
|
opErr := &nexus.OperationError{
|
|
Message: "update rejected",
|
|
State: nexus.OperationStateFailed,
|
|
Cause: &nexus.FailureError{Failure: f},
|
|
}
|
|
if err := nexusrpc.MarkAsWrapperError(nexusrpc.DefaultFailureConverter(), opErr); err != nil {
|
|
return nexusrpc.CompleteOperationOptions{}, err
|
|
}
|
|
return nexusrpc.CompleteOperationOptions{
|
|
Error: opErr,
|
|
}, nil
|
|
}
|
|
|
|
// Retrieve the completion data from the underlying mutable state via MSPointer
|
|
return u.GetNexusUpdateCompletion(ctx, u.UpdateId, requestID)
|
|
}
|