go.temporal.io/server/chasm/engine.go
510 LOC · 88 covered · 422 uncovered · 22 ranges · 255 concepts · 13 introducers · 89 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 engine_mock.go
2
3
package chasm
4
5
import (
6
"context"
7
8
"go.temporal.io/server/common/log"
9
)
10
11
// NoValue is a sentinel type representing no value.
12
// Useful for accessing components using the engine methods (e.g., [GetComponent]) with a function that does not need to
13
// return any information.
14
type NoValue = *struct{}
15
16
type Engine interface {
17
StartExecution(
18
context.Context,
19
ComponentRef,
20
func(MutableContext) (RootComponent, error),
21
...TransitionOption,
22
) (StartExecutionResult, error)
23
UpdateWithStartExecution(
24
context.Context,
25
ComponentRef,
26
func(MutableContext) (RootComponent, error),
27
func(MutableContext, Component) error,
28
...TransitionOption,
29
) (EngineUpdateWithStartExecutionResult, error)
30
31
UpdateComponent(
32
context.Context,
33
ComponentRef,
34
func(MutableContext, Component) error,
35
...TransitionOption,
36
) ([]byte, error)
37
ReadComponent(
38
context.Context,
39
ComponentRef,
40
func(Context, Component) error,
41
...TransitionOption,
42
) error
43
44
PollComponent(
45
context.Context,
46
ComponentRef,
47
func(Context, Component) (bool, error),
48
...TransitionOption,
49
) ([]byte, error)
50
51
DeleteExecution(
52
context.Context,
53
ComponentRef,
54
DeleteExecutionRequest,
55
) error
56
57
// NotifyExecution notifies any PollComponent callers waiting on the execution.
58
NotifyExecution(ExecutionKey)
59
}
60
61
// DeleteExecutionRequest is the request for [DeleteExecution]. TerminateComponentRequest will only be
62
// used if the execution is still running. The actual deletion of the execution is async, and will return
63
// after creating the DeleteExecutionTask.
64
type DeleteExecutionRequest struct {
65
TerminateComponentRequest
66
}
67
68
type BusinessIDReusePolicy int
69
70
const (
71
BusinessIDReusePolicyAllowDuplicate BusinessIDReusePolicy = iota
72
BusinessIDReusePolicyAllowDuplicateFailedOnly
73
BusinessIDReusePolicyRejectDuplicate
74
)
75
76
type BusinessIDConflictPolicy int
77
78
const (
79
BusinessIDConflictPolicyFail BusinessIDConflictPolicy = iota
80
BusinessIDConflictPolicyTerminateExisting
81
BusinessIDConflictPolicyUseExisting
82
)
83
84
// RefConsistencyLevel controls how strictly a [ComponentRef] is validated when it is used to address a
85
// component in UpdateComponent. Each level selects which versioned transition the execution staleness check
86
// ([Node.IsStale]) keys off — i.e. how fresh the loaded mutable state must be — and, at the weakest level,
87
// whether the run ID is honored. It governs only the consistency-token / run resolution of the ref;
88
// archetype validation and access-intent (operation-intent) checks always apply.
89
//
90
// The levels form a ladder from strongest to weakest:
91
//
92
// - ExecutionLastUpdate: staleness is checked against the execution's last-update versioned transition —
93
// the loaded state must be at the exact transition the ref was taken at. Strongest; the default.
94
// - ComponentCreation: staleness is checked against the target component's initial (creation) versioned
95
// transition — the loaded state need only be at least as new as when the component was created (so it
96
// is guaranteed to know about the component), tolerating a stale execution transition. The creation
97
// transition is additionally matched in [Node.Component] so the same component instance must still
98
// exist at the path.
99
// - CurrentRun: the ref is resolved by component path on the current run, dropping the run ID and every
100
// versioned transition (no staleness check). Callers relying on this level must re-establish identity
101
// in component logic (e.g. by request ID). Weakest; note this resolves the current run only and does
102
// NOT verify the ref's run and the current run are in the same chain.
103
type RefConsistencyLevel int
104
105
const (
106
RefConsistencyLevelExecutionLastUpdate RefConsistencyLevel = iota
107
RefConsistencyLevelComponentCreation
108
RefConsistencyLevelCurrentRun
109
)
110
111
type TransitionOptions struct {
112
ReusePolicy BusinessIDReusePolicy
113
ConflictPolicy BusinessIDConflictPolicy
114
ConsistencyLevel RefConsistencyLevel
115
RequestID string
116
Speculative bool
117
}
118
119
type TransitionOption func(*TransitionOptions)
120
121
// StartExecutionResult contains the outcome of creating a new execution via [StartExecution].
122
//
123
// This struct provides information about whether a new execution was actually created,
124
// along with identifiers needed to reference the execution in subsequent operations.
125
//
126
// Fields:
127
// - ExecutionKey: The unique identifier for the execution. This key can be used to
128
// look up or reference the execution in future operations.
129
// - ExecutionRef: A serialized reference to the newly created root component.
130
// This can be passed to [UpdateComponent], [ReadComponent], or [PollComponent]
131
// to interact with the component. Use [DeserializeComponentRef] to convert this
132
// back to a [ComponentRef] if needed.
133
// - Created: Indicates whether a new execution was actually created. When false,
134
// the execution already existed (based on the [BusinessIDReusePolicy] and
135
// [BusinessIDConflictPolicy] configured via [WithBusinessIDPolicy]), and the
136
// existing execution was returned instead.
137
type StartExecutionResult struct {
138
ExecutionKey ExecutionKey
139
ExecutionRef []byte
140
Created bool
141
}
142
143
// UpdateWithStartExecutionResult is the result of a UpdateWithStartExecution operation.
144
//
145
// Fields:
146
// - ExecutionKey: The unique identifier for the execution. This key can be used to
147
// look up or reference the execution in future operations.
148
// - ExecutionRef: A serialized reference to the newly created root component.
149
// This can be passed to [UpdateComponent], [ReadComponent], or [PollComponent]
150
// to interact with the component. Use [DeserializeComponentRef] to convert this
151
// back to a [ComponentRef] if needed.
152
// - Created: Indicates whether a new execution was actually created. When false,
153
// the execution already existed (based on the [BusinessIDReusePolicy] and
154
// [BusinessIDConflictPolicy] configured via [WithBusinessIDPolicy]), and the
155
// existing execution was returned instead.
156
// - UpdateOutput: The output value returned by the update function.
157
type UpdateWithStartExecutionResult[O any] struct {
158
ExecutionKey ExecutionKey
159
ExecutionRef []byte
160
Created bool
161
UpdateOutput O
162
}
163
164
// EngineUpdateWithStartExecutionResult is a type alias for the result type returned by the UpdateWithStart Engine implementation.
165
type EngineUpdateWithStartExecutionResult = UpdateWithStartExecutionResult[struct{}]
166
167
// (only) this transition will not be persisted
168
// The next non-speculative transition will persist this transition as well.
169
// Compared to the ExecutionEphemeral() operation on RegistrableComponent,
170
// the scope of this operation is limited to a certain transition,
171
// while the ExecutionEphemeral() applies to all transitions.
172
// TODO: we need to figure out a way to run the tasks
173
// generated in a speculative transition
174
func WithSpeculative() TransitionOption {
175
return func(opts *TransitionOptions) {
176
opts.Speculative = true
177
}
178
}
179
180
// WithBusinessIDPolicy sets the businessID reuse and conflict policy
181
// used in the transition when creating a new execution.
182
// This option only applies to StartExecution() and UpdateWithStartExecution().
183
func WithBusinessIDPolicy(
184
reusePolicy BusinessIDReusePolicy,
185
conflictPolicy BusinessIDConflictPolicy,
187
>
return func(opts *TransitionOptions) {
188
>
opts.ReusePolicy = reusePolicy
189
>
opts.ConflictPolicy = conflictPolicy
190
>
}
191
}
192
193
// WithRequestID sets the requestID used when creating a new execution.
194
// This option only applies to StartExecution() and UpdateWithStartExecution().
195
func WithRequestID(
196
requestID string,
198
>
return func(opts *TransitionOptions) {
199
>
opts.RequestID = requestID
200
>
}
201
}
202
203
// WithRefConsistencyLevel sets the [RefConsistencyLevel] for the transition, controlling how strictly the
204
// supplied component ref is validated. Currently only UpdateComponent() honors it; it defaults to
205
// [RefConsistencyLevelExecutionLastUpdate].
206
func WithRefConsistencyLevel(level RefConsistencyLevel) TransitionOption {
207
return func(opts *TransitionOptions) {
208
opts.ConsistencyLevel = level
209
}
210
}
211
212
// Not needed for V1
213
// func WithEagerLoading(
214
// paths []ComponentPath,
215
// ) OperationOption {
216
// panic("not implemented")
217
// }
218
219
// StartExecution creates a new execution with a component initialized by the provided factory function.
220
//
221
// This is the primary entry point for starting a new execution in the CHASM engine. It handles
222
// the lifecycle of creating and persisting a new component within an execution context.
223
//
224
// Type Parameters:
225
// - C: The component type to create, must implement [RootComponent]
226
// - I: The input type passed to the factory function
227
// - O: The output type returned by the factory function
228
//
229
// Parameters:
230
// - ctx: Context containing the CHASM engine (must be created via [NewEngineContext])
231
// - key: Unique identifier for the execution, used for deduplication and lookup
232
// - startFn: Factory function that creates the component and produces output.
233
// Receives a [MutableContext] for accessing engine capabilities and the input value.
234
// - input: Application-specific data passed to startFn
235
// - opts: Optional [TransitionOption] functions to configure creation behavior:
236
// - [WithBusinessIDPolicy]: Controls duplicate handling and conflict resolution
237
// - [WithRequestID]: Sets a request ID for idempotency
238
// - [WithSpeculative]: Defers persistence until the next non-speculative transition
239
//
240
// Returns:
241
// - O: The output value produced by startFn
242
// - [NewExecutionResult]: Contains the execution key, serialized ref, and whether a new execution was created
243
// - error: Non-nil if creation failed or policy constraints were violated
244
func StartExecution[C RootComponent, I any](
245
ctx context.Context,
246
key ExecutionKey,
247
startFn func(MutableContext, I) (C, error),
248
input I,
249
opts ...TransitionOption,
251
>
result, err := engineFromContext(ctx).StartExecution(
252
>
ctx,
253
>
NewComponentRef[C](key),
254
>
func(mutableContext MutableContext) (_ RootComponent, retErr error) {
255
>
defer log.CapturePanic(mutableContext.Logger(), &retErr)
256
>
257
>
var c C
258
>
var err error
259
>
c, err = startFn(mutableContext, input)
260
>
return c, err
261
>
},
262
opts...,
263
)
266
>
}
267
269
>
ExecutionKey: result.ExecutionKey,
270
>
ExecutionRef: result.ExecutionRef,
271
>
Created: result.Created,
272
>
}, nil
273
}
274
275
func UpdateWithStartExecution[C RootComponent, I any, O any](
276
ctx context.Context,
277
key ExecutionKey,
278
startFn func(MutableContext, I) (C, error),
279
updateFn func(C, MutableContext, I) (O, error),
280
input I,
281
opts ...TransitionOption,
282
) (UpdateWithStartExecutionResult[O], error) {
283
var output O
284
result, err := engineFromContext(ctx).UpdateWithStartExecution(
285
ctx,
286
NewComponentRef[C](key),
287
func(mutableContext MutableContext) (_ RootComponent, retErr error) {
288
defer log.CapturePanic(mutableContext.Logger(), &retErr)
289
290
var c C
291
var err error
292
c, err = startFn(mutableContext, input)
293
return c, err
294
},
295
func(mutableContext MutableContext, c Component) (retErr error) {
296
defer log.CapturePanic(mutableContext.Logger(), &retErr)
297
298
var err error
299
output, err = updateFn(
300
c.(C),
301
mutableContext,
302
input,
303
)
304
return err
305
},
306
opts...,
307
)
308
if err != nil {
309
return UpdateWithStartExecutionResult[O]{
310
UpdateOutput: output,
311
}, err
312
}
313
return UpdateWithStartExecutionResult[O]{
314
ExecutionKey: result.ExecutionKey,
315
ExecutionRef: result.ExecutionRef,
316
Created: result.Created,
317
UpdateOutput: output,
318
}, nil
319
}
320
321
// TODO:
322
// - consider merge with ReadComponent
323
// - consider remove ComponentRef from the return value and allow components to get
324
// the ref in the transition function. There are some caveats there, check the
325
// comment of the NewRef method in MutableContext.
326
//
327
// UpdateComponent applies updateFn to the component identified by the supplied component reference.
328
//
329
// The only opts currently honored is [WithRefConsistencyLevel]; it selects the [RefConsistencyLevel] used to
330
// resolve and validate the ref (see that type for the ladder of levels). Other options are ignored.
331
//
332
// It returns the result, along with the new component reference. The returned reference may be
333
// nil when updateFn deletes the component in the same transaction and the component is not the
334
// root component.
335
func UpdateComponent[C any, R []byte | ComponentRef, I any, O any](
336
ctx context.Context,
337
r R,
338
updateFn func(C, MutableContext, I) (O, error),
339
input I,
340
opts ...TransitionOption,
342
>
var output O
343
>
344
>
ref, err := convertComponentRef(r)
345
>
if err != nil {
346
return output, nil, err
347
}
348
350
>
for _, opt := range opts {
351
opt(&options)
352
}
354
>
if err != nil {
355
return output, nil, err
356
}
357
359
>
ctx,
360
>
ref,
361
>
func(mutableContext MutableContext, c Component) (retErr error) {
362
>
defer log.CapturePanic(mutableContext.Logger(), &retErr)
363
>
364
>
var err error
365
>
output, err = updateFn(
366
>
c.(C),
367
>
mutableContext,
368
>
input,
369
>
)
370
>
return err
371
>
},
372
opts...,
373
)
374
377
>
}
379
}
380
381
// ReadComponent returns the result of evaluating readFn against the component identified by the
382
// component reference. opts are currently ignored.
383
func ReadComponent[C any, R []byte | ComponentRef, I any, O any](
384
ctx context.Context,
385
r R,
386
readFn func(C, Context, I) (O, error),
387
input I,
388
opts ...TransitionOption,
390
>
var output O
391
>
392
>
ref, err := convertComponentRef(r)
393
>
if err != nil {
394
return output, err
395
}
396
398
>
ctx,
399
>
ref,
400
>
func(chasmContext Context, c Component) (retErr error) {
401
>
defer log.CapturePanic(chasmContext.Logger(), &retErr)
402
>
403
>
var err error
404
>
output, err = readFn(
405
>
c.(C),
406
>
chasmContext,
407
>
input,
408
>
)
409
>
return err
410
>
},
411
opts...,
412
)
414
}
415
416
// PollComponent waits until the predicate is true when evaluated against the component identified
417
// by the supplied component reference. If this times out due to a server-imposed long-poll timeout
418
// then it returns (nil, nil, nil), as an indication that the caller should continue long-polling.
419
// Otherwise it returns (output, ref, err), where output is the output of the predicate function,
420
// and ref is a component reference identifying the state at which the predicate was satisfied. The
421
// predicate must be monotonic: if it returns true at execution state transition s then it must
422
// return true at all transitions t > s. If the predicate is true at the outset then PollComponent
423
// returns immediately. opts are currently ignored.
424
func PollComponent[C any, R []byte | ComponentRef, I any, O any](
425
ctx context.Context,
426
r R,
427
monotonicPredicate func(C, Context, I) (O, bool, error),
428
input I,
429
opts ...TransitionOption,
430
) (O, []byte, error) {
431
var output O
432
433
ref, err := convertComponentRef(r)
434
if err != nil {
435
return output, nil, err
436
}
437
438
newSerializedRef, err := engineFromContext(ctx).PollComponent(
439
ctx,
440
ref,
441
func(chasmContext Context, c Component) (_ bool, retErr error) {
442
defer log.CapturePanic(chasmContext.Logger(), &retErr)
443
444
out, satisfied, err := monotonicPredicate(
445
c.(C),
446
chasmContext,
447
input,
448
)
449
if satisfied {
450
output = out
451
}
452
return satisfied, err
453
},
454
opts...,
455
)
456
if err != nil {
457
return output, nil, err
458
}
459
return output, newSerializedRef, err
460
}
461
462
// DeleteExecution deletes the execution identified by the supplied execution key.
463
// If the execution is still running, it is terminated first. A DeleteExecutionTask is
464
// then queued to remove all execution data from persistence.
465
func DeleteExecution[C RootComponent](
466
ctx context.Context,
467
key ExecutionKey,
468
request DeleteExecutionRequest,
469
) error {
470
return engineFromContext(ctx).DeleteExecution(
471
ctx,
472
NewComponentRef[C](key),
473
request,
474
)
475
}
476
477
func convertComponentRef[R []byte | ComponentRef](
478
r R,
480
>
if refToken, ok := any(r).([]byte); ok {
481
return DeserializeComponentRef(refToken)
482
}
483
484
//revive:disable-next-line:unchecked-type-assertion
486
}
487
488
type engineCtxKeyType string
489
490
const engineCtxKey engineCtxKeyType = "chasmEngine"
491
492
// this will be done by the nexus handler?
493
// alternatively the engine can be a global variable,
494
// but not a good practice in fx.
495
func NewEngineContext(
496
ctx context.Context,
497
engine Engine,
499
>
return context.WithValue(ctx, engineCtxKey, engine)
500
>
}
501
502
func engineFromContext(
503
ctx context.Context,
505
>
e, ok := ctx.Value(engineCtxKey).(Engine)
506
>
if !ok {
508
>
}
510
}