go.temporal.io/server/chasm/task.go
102 LOC · 7 covered · 95 uncovered · 2 ranges · 345 concepts · 2 introducers · 179 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
//go:generate mockgen -package $GOPACKAGE -source $GOFILE -destination task_mock.go
2
3
package chasm
4
5
import (
6
"context"
7
"errors"
8
"time"
9
)
10
11
// ErrTaskDiscarded is the error returned by the default [SideEffectTaskHandlerBase] Discard implementation,
12
// indicating that a side-effect task on a standby cluster has been pending past the discard delay.
13
var ErrTaskDiscarded = errors.New("standby task pending for too long")
14
15
type (
16
// TaskAttributes specifies scheduling metadata for a task, supplied by the component author when
17
// the task is added via [MutableContext.AddTask].
18
TaskAttributes struct {
19
// ScheduledTime is when the task should fire. Use [TaskScheduledTimeImmediate] (zero value)
20
// for tasks that should execute as soon as possible.
21
ScheduledTime time.Time
22
// Destination is an optional routing key for outbound tasks (e.g., a URL host for HTTP
23
// callbacks). When non-empty, the task is categorized as outbound; when empty, it is
24
// categorized as a transfer task. Destination must only be set on immediate tasks.
25
Destination string
26
}
27
28
// TaskInvocation is passed to a task's Validate callback. It carries the task's [TaskAttributes]
29
// together with framework-supplied state for the current processing attempt.
30
TaskInvocation struct {
31
TaskAttributes
32
// Attempt is the current processing attempt for this task, starting at 1. It comes from the
33
// task executable and is not persisted; it resets to 1 on shard reload and on active or
34
// standby failover. It is 0 when the task is validated outside of task processing, such as
35
// during transaction close. A best effort validator may compare it against a threshold and
36
// return false to give up on a task that would otherwise never become invalid on its own.
37
Attempt int
38
}
39
40
// SideEffectTaskHandler handles side effect tasks that run outside of the state lock and have access to a Go
41
// context to perform I/O and access chasm engine methods such as [UpdateComponent]. Implementations must embed
42
// [SideEffectTaskHandlerBase].
43
SideEffectTaskHandler[C any, T any] interface {
44
TaskValidator[C, T]
45
Execute(context.Context, ComponentRef, TaskAttributes, T) error
46
// Discard implements custom discard behavior on standby clusters. When a side-effect task has been
47
// pending on standby past the discard delay, the framework calls Discard instead of silently dropping
48
// the task. For example, the activity dispatch handler implements this to spill tasks to matching.
49
// The ctx carries engine access, but implementations must avoid mutating component state on standby
50
// clusters.
51
Discard(context.Context, ComponentRef, TaskAttributes, T) error
52
sideEffectTaskHandler()
53
}
54
55
// PureTaskHandler handles pure tasks that run while holding execution state write lock and should not do I/O.
56
// Implementations must embed [PureTaskHandlerBase].
57
PureTaskHandler[C any, T any] interface {
58
TaskValidator[C, T]
59
Execute(MutableContext, C, TaskAttributes, T) error
60
pureTaskHandler()
61
}
62
63
// TaskValidator is implemented by both [SideEffectTaskHandler] and [PureTaskHandler] to gate
64
// whether a task should proceed with execution.
65
TaskValidator[C any, T any] interface {
66
// Validate determines whether a task should proceed with execution based on the current context, component
67
// state, task attributes, and task data.
68
//
69
// This function serves as a gate to prevent unnecessary task execution in several scenarios:
70
// 1. Standby cluster deduplication: When state is replicated to standby clusters, tasks are also replicated.
71
// Validate allows standby clusters to check if a task was already completed on the active cluster and
72
// skip execution if so (e.g., checking if an activity already transitioned from scheduled to started state).
73
// 2. Task obsolescence: Tasks can become irrelevant when state changes invalidate them (e.g., when a scheduler
74
// is updated to run at a different time, making the previously scheduled task invalid for the new state).
75
// For pure tasks that can run in a single transaction, Validate is called before execution to avoid
76
// unnecessary work.
77
//
78
// The framework automatically calls Validate at key points, such as after closing transactions, to check all
79
// generated tasks before they execute.
80
//
81
// Returns:
82
// - (true, nil) if the task is valid and should be executed
83
// - (false, nil) if the task should be silently dropped (it's no longer relevant)
84
// - (anything, error) if validation fails with an error
85
Validate(Context, C, TaskInvocation, T) (bool, error)
86
}
87
)
88
89
// TaskScheduledTimeImmediate is the zero time value used to indicate that a task should execute immediately.
90
var TaskScheduledTimeImmediate = time.Time{}
91
92
// IsImmediate reports whether the task is scheduled for immediate execution (zero or unset scheduled time).
94
>
return a.ScheduledTime.IsZero() ||
95
>
a.ScheduledTime.Equal(TaskScheduledTimeImmediate)
96
>
}
97
98
// IsValid reports whether the task attributes are well-formed. A Destination may only be set on
99
// immediate tasks; deferred tasks with a Destination are invalid.
101
>
return a.Destination == "" || a.IsImmediate()
102
>
}