Vihren Atlas

workflow_task_state_machine.go ×11

Code frontier

unlabeled · c_fecb5077da66

294 tests · 4484 LOC · 172 files · introduces 0 tests · 50 LOC · 1 file

Introduces — evidence that enters the hierarchy at this concept

Code
11 ranges50 lines · 1 files
Tests
0 tests

Contains — complete concept membership

All code (extent)
751 ranges4484 lines · 172 files · Browse complete extent
All tests (intent)
294 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.

1 file ranked by introduced lines: 50 introduced LOC across 11 ranges. Expand a file to inspect source; the > gutter marks introduced lines.

go.temporal.io/server/service/history/workflow/workflow_task_state_machine.go 50 introduced LOC · 11 ranges

Open complete file

308 originalScheduledTimestamp *timestamppb.Timestamp,
309 workflowTaskType enumsspb.WorkflowTaskType,
310 > ) (*historyi.WorkflowTaskInfo, error) { workflow_task_state_machine.go
311 > opTag := tag.WorkflowActionWorkflowTaskScheduled
312 > if m.HasPendingWorkflowTask() {
313 m.ms.logger.Warn(mutableStateInvalidHistoryActionMsg, opTag,
314 tag.WorkflowEventID(m.ms.GetNextEventID()),
322 // and
323 // - is not speculative.
324 > createWorkflowTaskScheduledEvent := !m.ms.IsTransientWorkflowTask() && workflowTaskType != enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE workflow_task_state_machine.go
325 >
326 > // If while scheduling a workflow task and new events has come, then this workflow task cannot be a transient/speculative.
327 > // Flush any buffered events before creating the workflow task, otherwise it will result in invalid IDs for
328 > // transient/speculative workflow task and will cause in timeout processing to not work for transient workflow tasks.
329 > if m.ms.HasBufferedEvents() {
330 m.ms.executionInfo.WorkflowTaskAttempt = 1
331 workflowTaskType = enumsspb.WORKFLOW_TASK_TYPE_NORMAL
333 m.ms.updatePendingEventIDs(m.ms.hBuilder.FlushBufferToCurrentBatch())
334 }
335 > if m.ms.IsTransientWorkflowTask() { workflow_task_state_machine.go
336 // TODO: ideally this should be the version of the last started workflow task.
337 // but we are using the last event version here instead since there's no other
351 }
352
353 > scheduleTime := m.ms.timeSource.Now().UTC() workflow_task_state_machine.go
354 > attempt := m.ms.executionInfo.WorkflowTaskAttempt
355 > // TaskQueue should already be set from workflow execution started event.
356 > taskQueue := m.ms.CurrentTaskQueue()
357 > // DefaultWorkflowTaskTimeout should already be set from workflow execution started event.
358 > startToCloseTimeout := m.getStartToCloseTimeout(m.ms.executionInfo.DefaultWorkflowTaskTimeout, attempt)
359 >
360 > var scheduledEvent *historypb.HistoryEvent
361 > var scheduledEventID int64
362 >
363 > if createWorkflowTaskScheduledEvent {
364 scheduledEvent = m.ms.hBuilder.AddWorkflowTaskScheduledEvent(
365 taskQueue,
369 )
370 scheduledEventID = scheduledEvent.GetEventId()
371 > } else { workflow_task_state_machine.go
372 // WorkflowTaskScheduledEvent will be created later.
373 scheduledEventID = m.ms.GetNextEventID()
374 }
375
376 > workflowTask, err := m.ApplyWorkflowTaskScheduledEvent( workflow_task_state_machine.go
377 > m.ms.GetCurrentVersion(),
378 > scheduledEventID,
379 > taskQueue,
380 > startToCloseTimeout,
381 > attempt,
382 > timestamppb.New(scheduleTime),
383 > originalScheduledTimestamp,
384 > workflowTaskType,
385 > )
386 > if err != nil {
387 return nil, err
388 }
389
390 // TODO merge active & passive task generation
391 > if !bypassTaskGeneration { workflow_task_state_machine.go
392 if workflowTask.Type == enumsspb.WORKFLOW_TASK_TYPE_SPECULATIVE {
393 err = m.ms.taskGenerator.GenerateScheduleSpeculativeWorkflowTaskTasks(workflowTask)
408 bypassTaskGeneration bool,
409 workflowTaskType enumsspb.WorkflowTaskType,
410 > ) (*historyi.WorkflowTaskInfo, error) { workflow_task_state_machine.go
411 > return m.AddWorkflowTaskScheduledEventAsHeartbeat(bypassTaskGeneration, timestamppb.New(m.ms.timeSource.Now()), workflowTaskType)
412 > }
413
414 // AddFirstWorkflowTaskScheduled adds the first workflow task scheduled event unless it should be delayed as indicated
1466 defaultTimeout *durationpb.Duration,
1467 attempt int32,
1468 > ) *durationpb.Duration { workflow_task_state_machine.go
1469 > // This util function is only for calculating active workflow task timeout.
1470 > // Transient workflow task in passive cluster won't call this function and
1471 > // always use default timeout as it will either be completely overwritten by
1472 > // a replicated workflow schedule event from active cluster, or if used, it's
1473 > // attempt will be reset to 1.
1474 > // Check ApplyTransientWorkflowTaskScheduled for details.
1475 >
1476 > if defaultTimeout == nil {
1477 defaultTimeout = durationpb.New(0)
1478 }
1479
1480 > if attempt <= workflowTaskRetryBackoffMinAttempts { workflow_task_state_machine.go
1481 > return defaultTimeout
1482 > }
1483
1484 policy := backoff.NewExponentialRetryPolicy(workflowTaskRetryInitialInterval).

Direct hierarchy

Parents

Children