mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-30 18:41:49 -07:00
f1fe14b1f5
## 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
<!-- CURSOR_SUMMARY -->
---
> [!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](https://cursor.com/bugbot) for commit
4484fee104. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->
---------
Co-authored-by: long-nt-tran <long.tran@temporal.io>
64 lines
2.7 KiB
Go
64 lines
2.7 KiB
Go
package chasm
|
|
|
|
import (
|
|
"time"
|
|
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
historypb "go.temporal.io/api/history/v1"
|
|
"go.temporal.io/server/common/nexus/nexusrpc"
|
|
)
|
|
|
|
// MSPointer is a special CHASM type which components can use to access their Node's underlying backend (i.e. mutable
|
|
// state). It is used to expose methods needed from the mutable state without polluting the chasm.Context interface.
|
|
// When deserializing components with fields of this type, the CHASM engine will set the value to its NodeBackend.
|
|
// This should only be used by the Workflow component.
|
|
type MSPointer struct {
|
|
backend NodeBackend
|
|
}
|
|
|
|
// NewMSPointer creates a new MSPointer instance.
|
|
func NewMSPointer(backend NodeBackend) MSPointer {
|
|
return MSPointer{
|
|
backend: backend,
|
|
}
|
|
}
|
|
|
|
// WorkflowRunTimeout returns the workflow run timeout duration. Returns 0 if no timeout is set.
|
|
func (m MSPointer) WorkflowRunTimeout() time.Duration {
|
|
return m.backend.GetExecutionInfo().GetWorkflowRunTimeout().AsDuration()
|
|
}
|
|
|
|
// AddHistoryEvent adds a history event via the underlying mutable state.
|
|
func (m MSPointer) AddHistoryEvent(t enumspb.EventType, setAttributes func(*historypb.HistoryEvent)) *historypb.HistoryEvent {
|
|
return m.backend.AddHistoryEvent(t, setAttributes)
|
|
}
|
|
|
|
// HasAnyBufferedEvent returns true if there is at least one buffered event that matches the provided filter.
|
|
func (m MSPointer) HasAnyBufferedEvent(filter func(*historypb.HistoryEvent) bool) bool {
|
|
return m.backend.HasAnyBufferedEvent(filter)
|
|
}
|
|
|
|
func (m MSPointer) GenerateEventLoadToken(event *historypb.HistoryEvent) ([]byte, error) {
|
|
return m.backend.GenerateEventLoadToken(event)
|
|
}
|
|
|
|
// LoadHistoryEvent loads a history event from the underlying mutable state using the given token.
|
|
func (m MSPointer) LoadHistoryEvent(ctx Context, token []byte) (*historypb.HistoryEvent, error) {
|
|
return m.backend.LoadHistoryEvent(ctx.goContext(), token)
|
|
}
|
|
|
|
// GetNexusCompletion retrieves the Nexus operation completion data for the given request ID from the underlying mutable state.
|
|
func (m MSPointer) GetNexusCompletion(ctx Context, requestID string) (nexusrpc.CompleteOperationOptions, error) {
|
|
return m.backend.GetNexusCompletion(ctx.goContext(), requestID)
|
|
}
|
|
|
|
// GetWorkflowTypeName retrieves the workflow type name from the underlying mutable state.
|
|
func (m MSPointer) GetWorkflowTypeName() string {
|
|
return m.backend.GetExecutionInfo().GetWorkflowTypeName()
|
|
}
|
|
|
|
// GetNexusUpdateCompletion retrieves the Nexus operation completion data for the given update ID and request ID from the underlying mutable state.
|
|
func (m MSPointer) GetNexusUpdateCompletion(ctx Context, updateID string, requestID string) (nexusrpc.CompleteOperationOptions, error) {
|
|
return m.backend.GetNexusUpdateCompletion(ctx.goContext(), updateID, requestID)
|
|
}
|