mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +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]>
97 lines
2.5 KiB
Go
97 lines
2.5 KiB
Go
package update
|
|
|
|
import (
|
|
"go.temporal.io/api/serviceerror"
|
|
updatepb "go.temporal.io/api/update/v1"
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
func notZero[T comparable](v T, label string, msg proto.Message) func() error {
|
|
return func() error {
|
|
var zero T
|
|
if v == zero {
|
|
return serviceerror.NewInvalidArgumentf("invalid %T: %v is not set", msg, label)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func eq[T comparable](
|
|
left T,
|
|
leftLbl string,
|
|
right T,
|
|
rightLbl string,
|
|
msg proto.Message,
|
|
) func() error {
|
|
return func() error {
|
|
if left != right {
|
|
return serviceerror.NewInvalidArgumentf("invalid %T: %v != %v", msg, leftLbl, rightLbl)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func validate(vs ...func() error) error {
|
|
for _, v := range vs {
|
|
if err := v(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func validateRequestMsg(updateID string, msg *updatepb.Request) error {
|
|
return validateRequestMsgPrefix(updateID, "", msg)
|
|
}
|
|
|
|
func callbacksRequireRequestID(msg *updatepb.Request) func() error {
|
|
return func() error {
|
|
if len(msg.GetCompletionCallbacks()) > 0 && msg.GetRequestId() == "" {
|
|
return serviceerror.NewInvalidArgumentf("invalid %T: request_id is required when completion_callbacks are set", msg)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func validateRequestMsgPrefix(
|
|
updateID string,
|
|
prefix string,
|
|
msg *updatepb.Request,
|
|
) error {
|
|
return validate(
|
|
notZero(msg, prefix+"body", msg),
|
|
notZero(msg.GetMeta(), prefix+"meta", msg),
|
|
notZero(msg.GetMeta().GetUpdateId(), prefix+"meta.update_id", msg),
|
|
eq(msg.GetMeta().GetUpdateId(), prefix+"meta.update_id", updateID, updateID, msg),
|
|
notZero(msg.GetInput(), prefix+"input", msg),
|
|
notZero(msg.GetInput().GetName(), prefix+"input.name", msg),
|
|
callbacksRequireRequestID(msg),
|
|
)
|
|
}
|
|
|
|
func validateAcceptanceMsg(msg *updatepb.Acceptance) error {
|
|
return validate(
|
|
notZero(msg, "body", msg),
|
|
notZero(msg.GetAcceptedRequestSequencingEventId(), "accepted_request_sequencing_event_id", msg),
|
|
notZero(msg.GetAcceptedRequestMessageId(), "accepted_request_message_id", msg),
|
|
)
|
|
}
|
|
|
|
func validateResponseMsg(updateID string, msg *updatepb.Response) error {
|
|
return validate(
|
|
notZero(msg, "body", msg),
|
|
notZero(msg.GetMeta(), "meta", msg),
|
|
notZero(msg.GetMeta().GetUpdateId(), "meta.update_id", msg),
|
|
eq(msg.GetMeta().GetUpdateId(), "meta.update_id", updateID, updateID, msg),
|
|
notZero(msg.GetOutcome(), "outcome", msg),
|
|
)
|
|
}
|
|
|
|
// ValidateWorkflowExecutionUpdateRejectionMessage validates the presence of
|
|
// expected fields on updatepb.Rejection messages.
|
|
func validateRejectionMsg(msg *updatepb.Rejection) error {
|
|
return validate(
|
|
notZero(msg, "body", msg),
|
|
)
|
|
}
|