Skip to content

Commit 2ee4a1f

Browse files
chaptersixchaptersix
committed
feat: add memo update support for schedules
Add a Memo field to ScheduleUpdate that allows updating the schedule-level memo via UpdateScheduleRequest. Uses pointer-to-map semantics: nil means don't update, non-nil empty map clears, non-nil with entries replaces. This feature is only supported on CHASM-backed schedules. Attempting to update memo on a workflow-backed schedule returns a FailedPrecondition error. Also enables the chasm-scheduler experiment in dev server and CI dynamic config so integration tests can create CHASM-backed schedules via the temporal-experiment gRPC header.
1 parent 0d41672 commit 2ee4a1f

6 files changed

Lines changed: 342 additions & 1 deletion

File tree

‎.github/workflows/docker/dynamic-config-custom.yaml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,3 +54,7 @@ frontend.WorkerHeartbeatsEnabled:
5454
- value: true
5555
frontend.ListWorkersEnabled:
5656
- value: true
57+
# Will no longer be necessary when CHASM schedule creation is enabled by default.
58+
frontend.allowedExperiments:
59+
- value:
60+
- "chasm-scheduler"

‎internal/cmd/build/main.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,7 @@ func (b *builder) integrationTest() error {
161161
"--dynamic-config-value", `component.nexusoperations.useSystemCallbackURL=false`,
162162
"--dynamic-config-value", `component.nexusoperations.callback.endpoint.template="http://localhost:7243/namespaces/{{.NamespaceName}}/nexus/callback"`,
163163
"--dynamic-config-value", "frontend.ListWorkersEnabled=true",
164+
"--dynamic-config-value", `frontend.allowedExperiments=["chasm-scheduler"]`, // Will no longer be necessary when CHASM schedule creation is enabled by default
164165
},
165166
})
166167
if err != nil {

‎internal/internal_schedule_client.go‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -297,6 +297,22 @@ func (scheduleHandle *scheduleHandleImpl) Update(ctx context.Context, options Sc
297297
}
298298
}
299299

300+
var newMemo *commonpb.Memo
301+
if newSchedule.Memo != nil {
302+
dataConverter := WithContext(ctx, scheduleHandle.client.dataConverter)
303+
if dataConverter == nil {
304+
dataConverter = converter.GetDefaultDataConverter()
305+
}
306+
newMemo, err = getWorkflowMemo(*newSchedule.Memo, dataConverter, sdkFlagsAllowed[SDKFlagMemoUserDCEncode])
307+
if err != nil {
308+
return err
309+
}
310+
if newMemo == nil {
311+
// An empty but non-nil map should clear the memo.
312+
newMemo = &commonpb.Memo{}
313+
}
314+
}
315+
300316
updateRequest := &workflowservice.UpdateScheduleRequest{
301317
Namespace: scheduleHandle.client.namespace,
302318
ScheduleId: scheduleHandle.ID,
@@ -305,6 +321,7 @@ func (scheduleHandle *scheduleHandleImpl) Update(ctx context.Context, options Sc
305321
Identity: scheduleHandle.client.identity,
306322
RequestId: uuid.NewString(),
307323
SearchAttributes: newSA,
324+
Memo: newMemo,
308325
}
309326

310327
storeCtx := extstore.WithStorageTarget(ctx, extstore.StorageDriverWorkflowInfo{

‎internal/internal_schedule_client_test.go‎

Lines changed: 190 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,19 @@ package internal
22

33
import (
44
"context"
5-
iconverter "go.temporal.io/sdk/internal/converter"
65
"testing"
76

87
"github.com/golang/mock/gomock"
98
"github.com/stretchr/testify/suite"
9+
commonpb "go.temporal.io/api/common/v1"
1010
schedulepb "go.temporal.io/api/schedule/v1"
1111
"go.temporal.io/api/serviceerror"
12+
taskqueuepb "go.temporal.io/api/taskqueue/v1"
13+
workflowpb "go.temporal.io/api/workflow/v1"
1214
"go.temporal.io/api/workflowservice/v1"
1315
"go.temporal.io/api/workflowservicemock/v1"
1416
"go.temporal.io/sdk/converter"
17+
iconverter "go.temporal.io/sdk/internal/converter"
1518
)
1619

1720
const (
@@ -221,6 +224,192 @@ func (s *scheduleClientTestSuite) TestIteratorError() {
221224
s.NotNil(err)
222225
}
223226

227+
func (s *scheduleClientTestSuite) TestUpdateScheduleWithMemo() {
228+
wf := func(ctx Context) string {
229+
panic("this is just a stub")
230+
}
231+
232+
// Create the schedule first
233+
createResp := &workflowservice.CreateScheduleResponse{}
234+
s.service.EXPECT().CreateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(createResp, nil).Times(1)
235+
236+
handle, err := s.client.ScheduleClient().Create(context.Background(), ScheduleOptions{
237+
ID: scheduleID,
238+
Spec: ScheduleSpec{
239+
CronExpressions: []string{"*"},
240+
},
241+
Action: &ScheduleWorkflowAction{
242+
Workflow: wf,
243+
ID: workflowID,
244+
TaskQueue: taskqueue,
245+
WorkflowExecutionTimeout: timeoutInSeconds,
246+
WorkflowTaskTimeout: timeoutInSeconds,
247+
},
248+
})
249+
s.NoError(err)
250+
251+
// Mock Describe and Update for the update call
252+
describeResp := &workflowservice.DescribeScheduleResponse{
253+
Schedule: &schedulepb.Schedule{
254+
Spec: &schedulepb.ScheduleSpec{},
255+
Action: &schedulepb.ScheduleAction{
256+
Action: &schedulepb.ScheduleAction_StartWorkflow{
257+
StartWorkflow: s.createWorkflowExecutionInfo(),
258+
},
259+
},
260+
Policies: &schedulepb.SchedulePolicies{},
261+
State: &schedulepb.ScheduleState{},
262+
},
263+
Info: &schedulepb.ScheduleInfo{},
264+
ConflictToken: nil,
265+
}
266+
s.service.EXPECT().DescribeSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(describeResp, nil).Times(1)
267+
268+
s.service.EXPECT().UpdateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).
269+
DoAndReturn(func(_ context.Context, req *workflowservice.UpdateScheduleRequest, _ ...interface{}) (*workflowservice.UpdateScheduleResponse, error) {
270+
s.NotNil(req.Memo)
271+
s.Len(req.Memo.Fields, 1)
272+
s.Contains(req.Memo.Fields, "key1")
273+
return &workflowservice.UpdateScheduleResponse{}, nil
274+
}).Times(1)
275+
276+
memo := map[string]interface{}{
277+
"key1": "value1",
278+
}
279+
err = handle.Update(context.Background(), ScheduleUpdateOptions{
280+
DoUpdate: func(input ScheduleUpdateInput) (*ScheduleUpdate, error) {
281+
return &ScheduleUpdate{
282+
Schedule: &input.Description.Schedule,
283+
Memo: &memo,
284+
}, nil
285+
},
286+
})
287+
s.NoError(err)
288+
}
289+
290+
func (s *scheduleClientTestSuite) TestUpdateScheduleWithNilMemoDoesNotSetMemo() {
291+
wf := func(ctx Context) string {
292+
panic("this is just a stub")
293+
}
294+
295+
createResp := &workflowservice.CreateScheduleResponse{}
296+
s.service.EXPECT().CreateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(createResp, nil).Times(1)
297+
298+
handle, err := s.client.ScheduleClient().Create(context.Background(), ScheduleOptions{
299+
ID: scheduleID,
300+
Spec: ScheduleSpec{
301+
CronExpressions: []string{"*"},
302+
},
303+
Action: &ScheduleWorkflowAction{
304+
Workflow: wf,
305+
ID: workflowID,
306+
TaskQueue: taskqueue,
307+
WorkflowExecutionTimeout: timeoutInSeconds,
308+
WorkflowTaskTimeout: timeoutInSeconds,
309+
},
310+
})
311+
s.NoError(err)
312+
313+
describeResp := &workflowservice.DescribeScheduleResponse{
314+
Schedule: &schedulepb.Schedule{
315+
Spec: &schedulepb.ScheduleSpec{},
316+
Action: &schedulepb.ScheduleAction{
317+
Action: &schedulepb.ScheduleAction_StartWorkflow{
318+
StartWorkflow: s.createWorkflowExecutionInfo(),
319+
},
320+
},
321+
Policies: &schedulepb.SchedulePolicies{},
322+
State: &schedulepb.ScheduleState{},
323+
},
324+
Info: &schedulepb.ScheduleInfo{},
325+
ConflictToken: nil,
326+
}
327+
s.service.EXPECT().DescribeSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(describeResp, nil).Times(1)
328+
329+
s.service.EXPECT().UpdateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).
330+
DoAndReturn(func(_ context.Context, req *workflowservice.UpdateScheduleRequest, _ ...interface{}) (*workflowservice.UpdateScheduleResponse, error) {
331+
s.Nil(req.Memo)
332+
return &workflowservice.UpdateScheduleResponse{}, nil
333+
}).Times(1)
334+
335+
err = handle.Update(context.Background(), ScheduleUpdateOptions{
336+
DoUpdate: func(input ScheduleUpdateInput) (*ScheduleUpdate, error) {
337+
return &ScheduleUpdate{
338+
Schedule: &input.Description.Schedule,
339+
// Memo is nil, should not be set on the request
340+
}, nil
341+
},
342+
})
343+
s.NoError(err)
344+
}
345+
346+
func (s *scheduleClientTestSuite) TestUpdateScheduleWithEmptyMemoClears() {
347+
wf := func(ctx Context) string {
348+
panic("this is just a stub")
349+
}
350+
351+
createResp := &workflowservice.CreateScheduleResponse{}
352+
s.service.EXPECT().CreateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(createResp, nil).Times(1)
353+
354+
handle, err := s.client.ScheduleClient().Create(context.Background(), ScheduleOptions{
355+
ID: scheduleID,
356+
Spec: ScheduleSpec{
357+
CronExpressions: []string{"*"},
358+
},
359+
Action: &ScheduleWorkflowAction{
360+
Workflow: wf,
361+
ID: workflowID,
362+
TaskQueue: taskqueue,
363+
WorkflowExecutionTimeout: timeoutInSeconds,
364+
WorkflowTaskTimeout: timeoutInSeconds,
365+
},
366+
})
367+
s.NoError(err)
368+
369+
describeResp := &workflowservice.DescribeScheduleResponse{
370+
Schedule: &schedulepb.Schedule{
371+
Spec: &schedulepb.ScheduleSpec{},
372+
Action: &schedulepb.ScheduleAction{
373+
Action: &schedulepb.ScheduleAction_StartWorkflow{
374+
StartWorkflow: s.createWorkflowExecutionInfo(),
375+
},
376+
},
377+
Policies: &schedulepb.SchedulePolicies{},
378+
State: &schedulepb.ScheduleState{},
379+
},
380+
Info: &schedulepb.ScheduleInfo{},
381+
ConflictToken: nil,
382+
}
383+
s.service.EXPECT().DescribeSchedule(gomock.Any(), gomock.Any(), gomock.Any()).Return(describeResp, nil).Times(1)
384+
385+
s.service.EXPECT().UpdateSchedule(gomock.Any(), gomock.Any(), gomock.Any()).
386+
DoAndReturn(func(_ context.Context, req *workflowservice.UpdateScheduleRequest, _ ...interface{}) (*workflowservice.UpdateScheduleResponse, error) {
387+
// Empty map should produce an empty Memo (not nil), to signal "clear"
388+
s.NotNil(req.Memo)
389+
s.Empty(req.Memo.Fields)
390+
return &workflowservice.UpdateScheduleResponse{}, nil
391+
}).Times(1)
392+
393+
emptyMemo := map[string]interface{}{}
394+
err = handle.Update(context.Background(), ScheduleUpdateOptions{
395+
DoUpdate: func(input ScheduleUpdateInput) (*ScheduleUpdate, error) {
396+
return &ScheduleUpdate{
397+
Schedule: &input.Description.Schedule,
398+
Memo: &emptyMemo,
399+
}, nil
400+
},
401+
})
402+
s.NoError(err)
403+
}
404+
405+
func (s *scheduleClientTestSuite) createWorkflowExecutionInfo() *workflowpb.NewWorkflowExecutionInfo {
406+
return &workflowpb.NewWorkflowExecutionInfo{
407+
WorkflowId: workflowID,
408+
WorkflowType: &commonpb.WorkflowType{Name: "test-workflow"},
409+
TaskQueue: &taskqueuepb.TaskQueue{Name: taskqueue},
410+
}
411+
}
412+
224413
func (s *scheduleClientTestSuite) TestCreateScheduleWorkflowMemoDataConverter() {
225414
testFn := func() {
226415
dc := iconverter.NewTestDataConverter()

‎internal/schedule_client.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -540,6 +540,16 @@ type (
540540
// attributes present: replace any and all pre-existing assigned search attributes with the defined search
541541
// attributes, i.e. upsert
542542
TypedSearchAttributes *SearchAttributes
543+
544+
// Memo - Non-indexed user supplied information to replace on the schedule.
545+
// If set, replaces the entire memo. If nil, leaves the existing memo intact.
546+
// An initialized but empty map will clear the memo.
547+
//
548+
// NOTE: Memo updates are only supported on CHASM-backed schedules.
549+
// Attempting to update memo on a workflow-backed schedule will return an error.
550+
//
551+
// NOTE: Experimental
552+
Memo *map[string]interface{}
543553
}
544554

545555
// ScheduleUpdateInput describes the current state of the schedule to be updated.

0 commit comments

Comments
 (0)