go.temporal.io/server/chasm/context.go
292 LOC · 96 covered · 196 uncovered · 25 ranges · 933 concepts · 23 introducers · 387 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
package chasm
2
3
import (
4
"context"
5
"errors"
6
"time"
7
8
commonpb "go.temporal.io/api/common/v1"
9
sdkpb "go.temporal.io/api/sdk/v1"
10
persistencespb "go.temporal.io/server/api/persistence/v1"
11
"go.temporal.io/server/common/log"
12
"go.temporal.io/server/common/metrics"
13
"go.temporal.io/server/common/namespace"
14
"google.golang.org/grpc/metadata"
15
)
16
17
type Context interface {
18
// Context is not bound to any component,
19
// so all methods needs to take in component as a parameter
20
21
// NOTE: component created in the current transaction won't have a ref
22
// this is a Ref to the component state at the start of the transition
23
Ref(Component) ([]byte, error)
24
// Now returns the current time in the context of the given component.
25
// In a context of a transaction, this time must be used to allow for framework support of pause and time skipping.
26
Now(Component) time.Time
27
// ExecutionKey returns the execution key for the execution the context is operating on.
28
ExecutionKey() ExecutionKey
29
// ExecutionInfo returns metadata information about the execution.
30
ExecutionInfo() ExecutionInfo
31
// Logger returns a logger tagged with execution key and other chasm framework internal information.
32
Logger() log.Logger
33
// NamespaceEntry returns the namespace entry for the execution.
34
NamespaceEntry() *namespace.Namespace
35
// EndpointByName resolves a nexus endpoint entry.
36
EndpointByName(endpointName string) (*persistencespb.NexusEndpointEntry, error)
37
// MetricsHandler returns a metrics handler with namespace tag.
38
MetricsHandler() metrics.Handler
39
// Value returns the value associated with this context for key. The behavior is the same as context.Context.Value().
40
// Use WithContextValues RegistrableComponentOption to set key values pair for a component upon registration.
41
// Registered key-value pairs will automatically be added to the Context whenever framework accesses the component.
42
// Alternatively, use ContextWithValue() to manually set values on Context which will take precedence over registered ones.
43
Value(key any) any
44
// RequestHeader returns the first value of the named gRPC metadata header from the inbound request context, or ""
45
// if absent.
46
//
47
// Only available when this Context was constructed from an inbound gRPC request, i.e. inside the start/update/read
48
// callbacks invoked by the chasm engine. In other contexts, such as pure tasks executed at the end of a transaction
49
// or background task handlers, the underlying ctx has no gRPC metadata and this method always returns "".
50
RequestHeader(key string) string
51
// Links returns the union of links attached to the given component across all requests.
52
// Returns nil for components that are not (yet) registered as tree nodes.
53
Links(Component) []*commonpb.Link
54
// RequestLinks returns the links attached to the given component for the specific requestID.
55
// Returns nil if no entry exists for that requestID. Empty requestID is rejected.
56
RequestLinks(Component, string) ([]*commonpb.Link, error)
57
// UserMetadata returns the user metadata attached to the given component, or nil if none.
58
UserMetadata(Component) *sdkpb.UserMetadata
59
60
// Intent() OperationIntent
61
// ComponentOptions(Component) []ComponentOption
62
63
// withValue should only be used by ContextWithValue() function, do NOT call it directly.
64
// For structs implementing this method, although the returned value has type Context,
65
// the concrete type MUST be the same concrete type as the receiver.
66
withValue(key any, value any) Context
67
structuredRef(Component) (ComponentRef, error)
68
goContext() context.Context
69
}
70
71
type ExecutionInfo struct {
72
// StateTransitionCount is the number of create/update transactions in the history of this execution.
73
StateTransitionCount int64
74
// ApproximateStateSize is the approximate size in bytes of the persisted execution state of this execution.
75
ApproximateStateSize int
76
// CloseTime is the time when the execution was closed.
77
// An execution is closed when its root component reaches a terminal state in its lifecycle.
78
// If the component is still running (not yet closed), it returns a zero time.Time value.
79
CloseTime time.Time
80
}
81
82
type EndpointRegistry interface {
83
GetByName(ctx context.Context, namespaceID namespace.ID, endpointName string) (*persistencespb.NexusEndpointEntry, error)
84
}
85
86
type MutableContext interface {
87
Context
88
89
// AddTask adds a task to be emitted as part of the current transaction.
90
// The task is associated with the given component and will be invoked via the registered handler for the given task
91
// referencing the component.
92
AddTask(Component, TaskAttributes, any)
93
// SetRequestLinks records the links contributed by the given request on the
94
// component, replacing any prior entry for the same request ID. Passing
95
// nil/empty links removes the entry.
96
SetRequestLinks(Component, string, []*commonpb.Link) error
97
// SetUserMetadata replaces the user metadata attached to the given component.
98
SetUserMetadata(Component, *sdkpb.UserMetadata) error
99
100
// Get a Ref for the component
101
// This ref to the component state at the end of the transition
102
// Same as Ref(Component) method in Context,
103
// this only works for components that already exists at the start of the transition
104
//
105
// If we provide this method, then the method on the engine doesn't need to
106
// return a Ref
107
// NewRef(Component) (ComponentRef, bool)
108
}
109
110
type immutableCtx struct {
111
// The context here is not really used today.
112
// But it will be when we support partial loading later,
113
// and the framework potentially needs to go to persistence to load some fields.
114
ctx context.Context
115
// now is constant for this context; child contexts inherit the same value.
116
now time.Time
117
118
executionKey ExecutionKey
119
120
// Not embedding the Node here to avoid exposing AddTask() method on Node,
121
// so that ContextImpl won't implement MutableContext interface.
122
root *Node
123
}
124
125
type mutableCtx struct {
126
*immutableCtx
127
}
128
129
// NewContext creates a new Context from an existing Context and root Node.
130
//
131
// NOTE: Library authors should not invoke this constructor directly, and instead use [ReadComponent].
132
func NewContext(
133
ctx context.Context,
134
node *Node,
136
>
return newContext(ctx, node)
137
>
}
138
139
// newContext creates a new immutableCtx from an existing Context and root Node.
140
// This is similar to NewContext, but returns *immutableCtx instead of Context interface.
141
func newContext(
142
ctx context.Context,
143
node *Node,
145
>
root := node.root()
146
>
workflowKey := node.backend.GetWorkflowKey()
147
>
return &immutableCtx{
148
>
ctx: ctx,
149
>
now: root.Now(nil),
150
>
root: root,
151
>
executionKey: ExecutionKey{
152
>
NamespaceID: workflowKey.NamespaceID,
153
>
BusinessID: workflowKey.WorkflowID,
154
>
RunID: workflowKey.RunID,
155
>
},
156
>
}
157
>
}
158
160
>
return c.root.Ref(component)
161
>
}
162
164
>
return c.root.componentLinks(component)
165
>
}
166
167
>
func (c *immutableCtx) RequestLinks(component Component, requestID string) ([]*commonpb.Link, error) {
context.go ×1
168
>
return c.root.componentRequestLinks(component, requestID)
169
>
}
170
172
>
return c.root.componentUserMetadata(component)
173
>
}
174
176
>
return c.now
177
>
}
178
180
>
return c.executionKey
181
>
}
182
184
>
executionInfo := c.root.backend.GetExecutionInfo()
185
>
186
>
var closeTime time.Time
187
>
closeTimestamp := executionInfo.GetCloseTime()
188
>
if closeTimestamp != nil {
190
>
}
191
193
>
StateTransitionCount: executionInfo.GetStateTransitionCount(),
194
>
ApproximateStateSize: c.root.backend.GetApproximatePersistedSize(),
195
>
CloseTime: closeTime,
196
>
}
197
}
198
200
>
return c.root.logger
201
>
}
202
204
>
return c.root.metricsHandler
205
>
}
206
208
>
if v := c.goContext().Value(key); v != nil {
210
>
}
211
213
}
214
216
>
return &immutableCtx{
217
>
ctx: context.WithValue(c.goContext(), key, value),
218
>
now: c.now,
219
>
root: c.root,
220
>
executionKey: c.executionKey,
221
>
}
222
>
}
223
224
func (c *immutableCtx) structuredRef(component Component) (ComponentRef, error) {
225
return c.root.structuredRef(component)
226
}
227
229
>
return c.root.backend.GetNamespaceEntry()
230
>
}
231
233
>
return c.ctx
234
>
}
235
236
func (c *immutableCtx) RequestHeader(key string) string {
237
if values := metadata.ValueFromIncomingContext(c.ctx, key); len(values) > 0 {
238
return values[0]
239
}
240
return ""
241
}
242
243
func (c *immutableCtx) EndpointByName(name string) (*persistencespb.NexusEndpointEntry, error) {
244
reg := c.root.backend.EndpointRegistry()
245
if reg == nil {
246
return nil, errors.New("endpoint registry not available")
247
}
248
return reg.GetByName(c.ctx, c.NamespaceEntry().ID(), name)
249
}
250
251
// NewMutableContext creates a new MutableContext from an existing Context and root Node.
252
//
253
// NOTE: Library authors should not invoke this constructor directly, and instead use the [UpdateComponent],
254
// [UpdateWithStartExecution], or [StartExecution] APIs.
255
func NewMutableContext(
256
ctx context.Context,
257
node *Node,
259
>
return &mutableCtx{
260
>
immutableCtx: newContext(ctx, node),
261
>
}
262
>
}
263
264
func (c *mutableCtx) AddTask(
265
component Component,
266
attributes TaskAttributes,
267
payload any,
269
>
c.root.AddTask(component, attributes, payload)
270
>
}
271
272
>
func (c *mutableCtx) SetRequestLinks(component Component, requestID string, links []*commonpb.Link) error {
context.go ×1
273
>
return c.root.setComponentRequestLinks(component, requestID, links)
274
>
}
275
276
>
func (c *mutableCtx) SetUserMetadata(component Component, md *sdkpb.UserMetadata) error {
context.go ×1
277
>
return c.root.setComponentUserMetadata(component, md)
278
>
}
279
281
>
return &mutableCtx{
282
>
immutableCtx: ContextWithValue(c.immutableCtx, key, value),
283
>
}
284
>
}
285
286
// ContextWithValue returns a new Context with the given key-value pair added.
287
// Added key-value pairs will be accessible via the Value() method on the returned Context,
288
// and the behavior of the key-value pair is the same as context.Context.WithValue().
290
>
//nolint:revive // unchecked-type-assertion
291
>
return any(c.withValue(key, value)).(C)
292
>
}