mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Fred TzengandGitHub
a36f8d3148
## What changed? Added a SideEffectDiscardHandler interface to the CHASM framework that allows side-effect task executors to define custom discard behavior on standby clusters. When a side-effect task has been pending on standby past the configurable discard delay (ChasmStandbyTaskDiscardDelay), the framework now calls HandleDiscard instead of silently discarding the task. Implemented the handler for ActivityDispatchTask to spill tasks to matching on standby clusters. ## Why? On standby clusters, CHASM side-effect tasks (e.g., ActivityDispatchTask) were silently discarded after the discard delay. For standalone activities, this means the activity task never gets pushed to the standby cluster's matching queue, so workers on the standby cluster can't pick up work even after a failover. The HandleDiscard hook lets task executors define spillover behavior — the activity dispatch executor uses it to push to matching on standby, enabling faster recovery after namespace failover. ## How did you test it? - [X] built - [ ] run locally and tested manually - [ ] covered by existing tests - [X] added new unit test(s) - [X] added new functional test(s)
54 lines
1.6 KiB
Go
54 lines
1.6 KiB
Go
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination chasm_tree_mock.go
|
|
|
|
package interfaces
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/chasm"
|
|
"go.temporal.io/server/service/history/tasks"
|
|
)
|
|
|
|
var _ ChasmTree = (*chasm.Node)(nil)
|
|
|
|
// TODO: Remove this interface and use *chasm.Node directly
|
|
// when chasm/tree.go implementation completes.
|
|
type ChasmTree interface {
|
|
CloseTransaction() (chasm.NodesMutation, error)
|
|
Snapshot(*persistencespb.VersionedTransition) chasm.NodesSnapshot
|
|
ApplySystemMutation(chasm.NodesMutation) error
|
|
ApplyMutation(chasm.NodesMutation) error
|
|
ApplySnapshot(chasm.NodesSnapshot) error
|
|
RefreshTasks() error
|
|
IsStateDirty() bool
|
|
IsDirty() bool
|
|
Terminate(chasm.TerminateComponentRequest) error
|
|
Archetype() (chasm.Archetype, error)
|
|
ArchetypeID() chasm.ArchetypeID
|
|
EachPureTask(
|
|
deadline time.Time,
|
|
callback func(executor chasm.NodePureTask, taskAttributes chasm.TaskAttributes, task any) (bool, error),
|
|
) error
|
|
ExecuteSideEffectTask(
|
|
ctx context.Context,
|
|
executionKey chasm.ExecutionKey,
|
|
task *tasks.ChasmTask,
|
|
validate func(chasm.NodeBackend, chasm.Context, chasm.Component) error,
|
|
) error
|
|
ExecuteSideEffectDiscardTask(
|
|
ctx context.Context,
|
|
executionKey chasm.ExecutionKey,
|
|
task *tasks.ChasmTask,
|
|
validate func(chasm.NodeBackend, chasm.Context, chasm.Component) error,
|
|
) error
|
|
ValidateSideEffectTask(
|
|
ctx context.Context,
|
|
task *tasks.ChasmTask,
|
|
) (bool, error)
|
|
IsStale(chasm.ComponentRef) error
|
|
Component(chasm.Context, chasm.ComponentRef) (chasm.Component, error)
|
|
ComponentByPath(chasm.Context, []string) (chasm.Component, error)
|
|
}
|