Vihren Atlas

test_engine.go ×20

Code frontier

unlabeled · c_cea340ae6958

6 tests · 3512 LOC · 131 files · introduces 0 tests · 142 LOC · 2 files

Introduces — evidence that enters the hierarchy at this concept

Code
23 ranges142 lines · 2 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
728 ranges3512 lines · 131 files · Browse complete extent
All tests (intent)
6 testsBrowse complete intent

Neighbourhood graph

The orange circle is the focus. Violet and green circles are every ancestor and descendant, broader and narrower, at any distance; blue squares and pink diamonds are the introduced files and exact introduced tests of every visible concept, not only the focus's. Arrows point from broader to narrower concepts and bridge only concepts omitted from this view. Undirected links show source or test introduction. Concept and file size follows LOC; exact test nodes use test-count units.

Introduced files, introduced tests, and structurally relevant concept specialization

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 native relationship evidence on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the native relationship evidence on this page while the interactive map is unavailable.

Native relationship evidence

Every exact file and test below is linked only from the concept that introduces it.

Introduced tests

Every collected test enters the hierarchy at exactly one concept.

No tests are introduced at this concept. Its intent tests are introduced by other concepts.

Introduced code

Every collected source range enters the hierarchy at exactly one concept.

2 files ranked by introduced lines: 142 introduced LOC across 23 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/chasm/chasmtest/test_engine.go 124 introduced LOC · 20 ranges

Open complete file

85 registry *chasm.Registry,
86 opts ...EngineOption,
87 > ) *Engine { test_engine.go
88 > t.Helper()
89 >
90 > ts := clock.NewEventTimeSource()
91 > ts.Update(time.Now())
92 > e := &Engine{
93 > t: t,
94 > registry: registry,
95 > logger: testlogger.NewTestLogger(t, testlogger.FailOnExpectedErrorOnly),
96 > metrics: metrics.NoopMetricsHandler,
97 > timeSource: ts,
98 > currentExecutions: make(map[businessKey]*execution),
99 > allExecutions: make(map[runKey]*execution),
100 > notifier: newExecutionNotifier(),
101 > }
102 >
103 > for _, opt := range opts {
104 opt(e)
105 }
106
107 > return e test_engine.go
108 }
109
128 startFn func(chasm.MutableContext) (chasm.RootComponent, error),
129 opts ...chasm.TransitionOption,
130 > ) (chasm.StartExecutionResult, error) { test_engine.go
131 > options := constructTransitionOptions(opts...)
132 > bKey := newBusinessKey(ref.ExecutionKey)
133 >
134 > current, hasCurrent := e.currentExecutions[bKey]
135 > if hasCurrent {
136 // if the requestID matches the original create request, return the existing run.
137 if options.RequestID != "" && options.RequestID == current.requestID {
159 }
160
161 > return e.startNew(ctx, ref.ExecutionKey, startFn, options.RequestID) test_engine.go
162 }
163
428 startFn func(chasm.MutableContext) (chasm.RootComponent, error),
429 requestID string,
430 > ) (chasm.StartExecutionResult, error) { test_engine.go
431 > exec := e.newExecution(key)
432 > exec.requestID = requestID
433 >
434 > mutableCtx := chasm.NewMutableContext(ctx, exec.node)
435 > root, err := startFn(mutableCtx)
436 > if err != nil {
437 return chasm.StartExecutionResult{}, err
438 }
439 > if err := exec.node.SetRootComponent(root); err != nil { test_engine.go
440 return chasm.StartExecutionResult{}, err
441 }
442 > if _, err = exec.node.CloseTransaction(); err != nil { test_engine.go
443 return chasm.StartExecutionResult{}, err
444 }
445
446 > exec.root = root test_engine.go
447 > e.currentExecutions[newBusinessKey(exec.key)] = exec
448 > e.allExecutions[newRunKey(exec.key)] = exec
449 >
450 > serializedRef, err := exec.node.Ref(root)
451 > if err != nil {
452 return chasm.StartExecutionResult{}, err
453 }
454
455 > return chasm.StartExecutionResult{ test_engine.go
456 > ExecutionKey: exec.key,
457 > ExecutionRef: serializedRef,
458 > Created: true,
459 > }, nil
460 }
461
503 }
504
505 > func (e *Engine) newExecution(key chasm.ExecutionKey) *execution { test_engine.go
506 > // bsMu (backend state mutex) guards transitionCount and execState, which are shared
507 > // across handler closures. It is separate from MockNodeBackend's internal mu to avoid deadlocks.
508 > var (
509 > bsMu sync.Mutex
510 > transitionCount int64 = 1
511 > execState = persistencespb.WorkflowExecutionState{
512 > State: enumsspb.WORKFLOW_EXECUTION_STATE_CREATED,
513 > Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
514 > }
515 > )
516 >
517 > backend := &chasm.MockNodeBackend{
518 > // NextTransitionCount increments on every CloseTransaction call, matching
519 > // the real engine's per transition monotonic counter.
520 > HandleNextTransitionCount: func() int64 {
521 > bsMu.Lock()
522 > defer bsMu.Unlock()
523 > transitionCount++
524 > return transitionCount
525 > },
526 // CurrentVersionedTransition reflects the latest committed transition count.
527 > HandleCurrentVersionedTransition: func() *persistencespb.VersionedTransition { test_engine.go
528 > bsMu.Lock()
529 > defer bsMu.Unlock()
530 > return &persistencespb.VersionedTransition{
531 > NamespaceFailoverVersion: 1,
532 > TransitionCount: transitionCount,
533 > }
534 > },
535 > HandleGetCurrentVersion: func() int64 { return 1 },
536 > HandleGetWorkflowKey: func() definition.WorkflowKey {
537 > return definition.NewWorkflowKey(key.NamespaceID, key.BusinessID, key.RunID)
538 > },
539 > HandleIsWorkflow: func() bool { return false },
540 // GetExecutionState returns the current lifecycle state, which CloseTransaction
541 // uses to decide whether to call UpdateWorkflowStateStatus on the backend.
542 > HandleGetExecutionState: func() *persistencespb.WorkflowExecutionState { test_engine.go
543 > bsMu.Lock()
544 > defer bsMu.Unlock()
545 > return &persistencespb.WorkflowExecutionState{
546 > State: execState.State,
547 > Status: execState.Status,
548 > }
549 > },
550 // UpdateWorkflowStateStatus is called by CloseTransaction when the root
551 // component's LifecycleState changes from Running to Completed, Failed, or Terminated.
552 > HandleUpdateWorkflowStateStatus: func(state enumsspb.WorkflowExecutionState, status enumspb.WorkflowExecutionStatus) (bool, error) { test_engine.go
553 > bsMu.Lock()
554 > defer bsMu.Unlock()
555 > changed := execState.State != state || execState.Status != status
556 > execState.State = state
557 > execState.Status = status
558 > return changed, nil
559 > },
560 }
561 > return &execution{ test_engine.go
562 > key: key,
563 > backend: backend,
564 > node: chasm.NewEmptyTree(
565 > e.registry,
566 > e.timeSource,
567 > backend,
568 > chasm.DefaultPathEncoder,
569 > e.logger,
570 > e.metrics,
571 > ),
572 > }
573 }
574
575 // executionForRef looks up an execution by the ref's RunID when present, or falls back
576 // to the current run for the business ID when RunID is empty.
577 > func (e *Engine) executionForRef(ref chasm.ComponentRef) (*execution, error) { test_engine.go
578 > if ref.RunID != "" {
579 exec, ok := e.allExecutions[newRunKey(ref.ExecutionKey)]
580 if !ok {
636 }
637
638 > func constructTransitionOptions(opts ...chasm.TransitionOption) chasm.TransitionOptions { test_engine.go
639 > options := defaultTransitionOptions
640 > for _, opt := range opts {
641 opt(&options)
642 }
643 // NOTE: TransitionOptions.Speculative is intentionally not implemented here. It is also
644 // unimplemented in the production engine (see the TODO in service/history/chasm_engine.go).
645 > return options test_engine.go
646 }
647
653 }
654
655 > func newBusinessKey(key chasm.ExecutionKey) businessKey { test_engine.go
656 > return businessKey{namespaceID: key.NamespaceID, businessID: key.BusinessID}
657 > }
658
659 > func newRunKey(key chasm.ExecutionKey) runKey { test_engine.go
660 > return runKey{namespaceID: key.NamespaceID, businessID: key.BusinessID, runID: key.RunID}
661 > }
662
663 // executionNotifier allows [PollComponent] callers to subscribe to state change
669 }
670
671 > func newExecutionNotifier() *executionNotifier { test_engine.go
672 > return &executionNotifier{
673 > subscribers: make(map[chasm.ExecutionKey][]chan struct{}),
674 > }
675 > }
676
677 // subscribe returns a channel that will be closed on the next notify call for key,
go.temporal.io/server/chasm/engine.go 18 introduced LOC · 3 ranges

Open complete file

248 input I,
249 opts ...TransitionOption,
250 > ) (StartExecutionResult, error) { engine.go
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 )
264 > if err != nil { engine.go
265 return StartExecutionResult{}, err
266 }
267
268 > return StartExecutionResult{ engine.go
269 > ExecutionKey: result.ExecutionKey,
270 > ExecutionRef: result.ExecutionRef,
271 > Created: result.Created,
272 > }, nil
273 }
274

Direct hierarchy

Parents

Children