283
}
284
285
>
createRequest := &workflowservice.StartWorkflowExecutionRequest{
retry.go
286
>
RequestId: uuid.NewString(),
287
>
Namespace: newMutableState.GetNamespaceEntry().Name().String(),
288
>
WorkflowId: newExecution.WorkflowId,
289
>
TaskQueue: tq,
290
>
WorkflowType: wType,
291
>
WorkflowExecutionTimeout: previousExecutionInfo.WorkflowExecutionTimeout,
292
>
WorkflowRunTimeout: runTimeout,
293
>
WorkflowTaskTimeout: taskTimeout,
294
>
Input: startAttr.Input,
295
>
Header: startAttr.Header,
296
>
RetryPolicy: startAttr.RetryPolicy,
297
>
CronSchedule: startAttr.CronSchedule,
298
>
Memo: startAttr.Memo,
299
>
SearchAttributes: startAttr.SearchAttributes,
300
>
CompletionCallbacks: completionCallbacks,
301
>
Links: startLinks,
302
>
Priority: startAttr.Priority,
303
>
VersioningOverride: pinnedOverride,
304
>
}
305
>
306
>
tsc, stateProp := propagateTimeSkippingToNextRun(previousExecutionInfo)
307
>
createRequest.TimeSkippingConfig = tsc
308
>
309
>
attempt := int32(1)
310
>
if initiator == enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY {
311
>
attempt = previousExecutionInfo.Attempt + 1
312
>
}
313
314
>
var sourceVersionStamp *commonpb.WorkerVersionStamp
retry.go
315
>
if previousExecutionInfo.AssignedBuildId == "" && GetEffectiveVersioningBehavior(previousExecutionInfo.GetVersioningInfo()) == enumspb.VERSIONING_BEHAVIOR_UNSPECIFIED {
316
>
// TODO: only keeping this part for old versioning. The desired logic seem to be the same for both cron and
317
>
// retry: keep originally-inherited build ID. [cleanup-old-wv]
318
>
// For retry: propagate build-id version info to new workflow.
319
>
// For cron: do not propagate (always start on latest version).
320
>
if initiator == enumspb.CONTINUE_AS_NEW_INITIATOR_RETRY {
321
>
sourceVersionStamp = worker_versioning.StampIfUsingVersioning(previousMutableState.GetMostRecentWorkerVersionStamp())
322
>
}
323
}
324
325
>
req := &historyservice.StartWorkflowExecutionRequest{
retry.go
326
>
NamespaceId: newMutableState.GetNamespaceEntry().ID().String(),
327
>
StartRequest: createRequest,
328
>
ParentExecutionInfo: parentInfo,
329
>
LastCompletionResult: lastCompletionResult,
330
>
ContinuedFailure: failure,
331
>
ContinueAsNewInitiator: initiator,
332
>
// enforce minimal interval between runs to prevent tight loop continue as new spin.
333
>
FirstWorkflowTaskBackoff: previousMutableState.ContinueAsNewMinBackoff(durationpb.New(backoffInterval)),
334
>
Attempt: attempt,
335
>
SourceVersionStamp: sourceVersionStamp,
336
>
RootExecutionInfo: rootInfo,
337
>
InheritedBuildId: startAttr.InheritedBuildId, //nolint:staticcheck
338
>
InheritedPinnedVersion: inheritedPinnedVersion,
339
>
InheritedAutoUpgradeInfo: inheritedAutoUpgradeInfo,
340
>
// For retries, pass through the declined value from the started event directly.
341
>
DeclinedTargetVersionUpgrade: startAttr.GetDeclinedTargetVersionUpgrade(),
342
>
// Carry the previous run's accumulated skip and fast-forward target forward.
343
>
TimeSkippingStatePropagation: stateProp,
344
>
}
345
>
workflowTimeoutTime := timestamp.TimeValue(previousExecutionInfo.WorkflowExecutionExpirationTime)
346
>
if !workflowTimeoutTime.IsZero() {
347
>
req.WorkflowExecutionExpirationTime = timestamppb.New(workflowTimeoutTime)
348
>
}
349
350
>
event, err := newMutableState.AddWorkflowExecutionStartedEventWithOptions(
retry.go
351
>
&newExecution,
352
>
req,
353
>
previousExecutionInfo.AutoResetPoints,
354
>
previousMutableState.GetExecutionState().GetRunId(),
355
>
firstRunID,
356
>
)
357
>
if err != nil {
358
return serviceerror.NewInternal("Failed to add workflow execution started event.")
359
}
360
>
var parentClock *clockspb.VectorClock
retry.go
361
>
if parentInfo != nil {
362
parentClock = parentInfo.Clock
363
}
364
>
if _, err = newMutableState.AddFirstWorkflowTaskScheduled(parentClock, event, false); err != nil {
retry.go
365
return err
366
}
367
369
}