mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
b701b9bae2
## What changed?
Integrated chasm/lib/callback into MutableState and updated
callback-related APIs to use CHASM implementation. Added integration
tests following the scheduler pattern.
## Why?
Porting callback functionality from HSM to CHASM as part of the ongoing
HSM-to-CHASM migration effort.
## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [ ] added new unit test(s)
- [X] added new functional test(s)
## Potential risks
This will need to be tested more, but is not being turned on with this
PR
---
> [!NOTE]
> Integrates CHASM-based callbacks into mutable state and history APIs,
adds configs and error types, updates routing/state transitions, and
includes migration tests.
>
> - **CHASM Callback Integration**:
> - Add CHASM callback module (FX wiring, HTTP caller provider, request
routing) and metrics shared with HSM implementation.
> - Implement `ToAPICallback`, invocation flow changes, and refined
error wrapping.
> - **History/MutableState**:
> - Expose `ChasmEnabled`, `ChasmWorkflowComponent{,ReadOnly}`; use
CHASM for adding/processing completion callbacks when enabled.
> - Describe API now builds callback info from both CHASM and HSM trees.
> - Continue-as-new/retry path aggregates callbacks from both
implementations.
> - **State Machine/Executors**:
> - Switch to `queues/errors` types; move `NamespaceIDAndDestination` to
`queues/common`.
> - Adjust transitions: generate `BackoffTask` with scheduled time;
include destination on reschedule; no tasks on success/fail.
> - Simplify retry result (policy passed separately) and validation
signatures.
> - **Config**:
> - Add `EnableCHASMCallbacks` and `MaxCHASMCallbacksPerWorkflow`; wire
through history configs.
> - **Tests**:
> - Add migration and CHASM-enabled functional tests; update unit tests
to new error/types and behaviors.
>
> <sup>Written by [Cursor
Bugbot]() for commit
e3b51a931d3ea2dd7f18fb2f9c55fc880ba828d8. This will update automatically
on new commits. Configure
[here]().</sup>
---------
Co-authored-by: Roey Berman <[email protected]>
73 lines
2.1 KiB
Go
73 lines
2.1 KiB
Go
package queues
|
|
|
|
import (
|
|
"go.temporal.io/server/common/predicates"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
)
|
|
|
|
// Grouper groups tasks and constructs predicates for those groups.
|
|
type Grouper interface {
|
|
// Key returns the group key for a given task.
|
|
Key(task tasks.Task) (key any)
|
|
// Predicate constructs a prdicate from a slice of keys.
|
|
Predicate(keys []any) tasks.Predicate
|
|
}
|
|
|
|
type GrouperNamespaceID struct {
|
|
}
|
|
|
|
func (GrouperNamespaceID) Key(task tasks.Task) (key any) {
|
|
return task.GetNamespaceID()
|
|
}
|
|
|
|
func (GrouperNamespaceID) Predicate(keys []any) tasks.Predicate {
|
|
pendingNamespaceIDs := make([]string, len(keys))
|
|
for i, namespaceID := range keys {
|
|
// Assume predicate is only called with keys returned from GrouperNamespaceID.Key()
|
|
pendingNamespaceIDs[i] = namespaceID.(string)
|
|
}
|
|
return tasks.NewNamespacePredicate(pendingNamespaceIDs)
|
|
}
|
|
|
|
var _ Grouper = GrouperNamespaceID{}
|
|
|
|
type GrouperStateMachineNamespaceIDAndDestination struct {
|
|
}
|
|
|
|
func (g GrouperStateMachineNamespaceIDAndDestination) Key(task tasks.Task) (key any) {
|
|
return g.KeyTyped(task)
|
|
}
|
|
|
|
func (GrouperStateMachineNamespaceIDAndDestination) KeyTyped(task tasks.Task) (key tasks.TaskGroupNamespaceIDAndDestination) {
|
|
destGetter, ok := task.(tasks.HasDestination)
|
|
var dest string
|
|
if ok {
|
|
dest = destGetter.GetDestination()
|
|
}
|
|
smtGetter, ok := task.(tasks.HasStateMachineTaskType)
|
|
var smt string
|
|
if ok {
|
|
smt = smtGetter.StateMachineTaskType()
|
|
}
|
|
return tasks.TaskGroupNamespaceIDAndDestination{
|
|
TaskGroup: smt,
|
|
NamespaceID: task.GetNamespaceID(),
|
|
Destination: dest,
|
|
}
|
|
}
|
|
|
|
func (GrouperStateMachineNamespaceIDAndDestination) Predicate(keys []any) tasks.Predicate {
|
|
if len(keys) == 0 {
|
|
return predicates.Empty[tasks.Task]()
|
|
}
|
|
groups := make([]tasks.TaskGroupNamespaceIDAndDestination, len(keys))
|
|
for i, anyKey := range keys {
|
|
// Assume predicate is only called with keys returned from OutboundTaskGroupNamespaceIDAndDestination.Key()
|
|
key := anyKey.(tasks.TaskGroupNamespaceIDAndDestination)
|
|
groups[i] = key
|
|
}
|
|
return tasks.NewOutboundTaskPredicate(groups)
|
|
}
|
|
|
|
var _ Grouper = GrouperStateMachineNamespaceIDAndDestination{}
|