45
memo *commonpb.Memo,
46
migrationTime time.Time,
47
>
) *schedulerpb.CreateFromMigrationStateRequest {
migration.go
48
>
// V2 computes RunningWorkflows/RecentActions on-demand from BufferedStarts
49
>
infoClone := common.CloneProto(info)
50
>
infoClone.RunningWorkflows = nil
51
>
infoClone.RecentActions = nil
52
>
53
>
schedulerState := &schedulerpb.SchedulerState{
54
>
Schedule: common.CloneProto(schedule),
55
>
Info: infoClone,
56
>
Namespace: state.Namespace,
57
>
NamespaceId: state.NamespaceId,
58
>
ScheduleId: state.ScheduleId,
59
>
ConflictToken: state.ConflictToken,
60
>
Closed: false,
61
>
}
62
>
63
>
generatorState := &schedulerpb.GeneratorState{
64
>
LastProcessedTime: common.CloneProto(state.LastProcessedTime),
65
>
FutureActionTimes: nil, // Regenerated by GeneratorTask
66
>
}
67
>
68
>
pendingBufferedStarts := convertBufferedStartsLegacyToCHASM(
69
>
state.BufferedStarts,
70
>
state.NamespaceId,
71
>
state.ScheduleId,
72
>
state.ConflictToken,
73
>
getWorkflowID(schedule),
74
>
)
75
>
76
>
runningBufferedStarts := convertRunningWorkflowsToBufferedStarts(
77
>
info.RunningWorkflows,
78
>
state.NamespaceId,
79
>
state.ScheduleId,
80
>
state.ConflictToken,
81
>
migrationTime,
82
>
)
83
>
84
>
recentActionsBufferedStarts := convertRecentActionsToBufferedStarts(
85
>
info.RecentActions,
86
>
info.RunningWorkflows,
87
>
state.NamespaceId,
88
>
state.ScheduleId,
89
>
state.ConflictToken,
90
>
migrationTime,
91
>
)
92
>
93
>
allBufferedStarts := append(pendingBufferedStarts, runningBufferedStarts...)
94
>
allBufferedStarts = append(allBufferedStarts, recentActionsBufferedStarts...)
95
>
96
>
invokerState := &schedulerpb.InvokerState{
97
>
BufferedStarts: allBufferedStarts,
98
>
LastProcessedTime: timestamppb.New(migrationTime),
99
>
}
100
>
101
>
backfillers := convertBackfillsLegacyToCHASM(state.OngoingBackfills)
102
>
lastCompletion := convertLastCompletionLegacyToCHASM(state.LastCompletionResult, state.ContinuedFailure)
103
>
104
>
return &schedulerpb.CreateFromMigrationStateRequest{
105
>
NamespaceId: state.NamespaceId,
106
>
State: &schedulerpb.SchedulerMigrationState{
107
>
SchedulerState: schedulerState,
108
>
GeneratorState: generatorState,
109
>
InvokerState: invokerState,
110
>
Backfillers: backfillers,
111
>
LastCompletionResult: lastCompletion,
112
>
SearchAttributes: customSearchAttributesForMigration(searchAttributes),
113
>
Memo: memo.GetFields(),
114
>
},
115
>
}
116
>
}
117
118
// customSearchAttributesForMigration returns only the user-defined search attributes
119
// from a V1 scheduler workflow, stripping any reserved/system SAs.
120
>
func customSearchAttributesForMigration(sa *commonpb.SearchAttributes) map[string]*commonpb.Payload {
migration.go
121
>
fields := sa.GetIndexedFields()
122
>
if len(fields) == 0 {
124
>
}
125
out := maps.Clone(fields)
126
maps.DeleteFunc(out, func(k string, _ *commonpb.Payload) bool {