CHASM Visibility support minor fixes (#8690)

## What changed?
- Addresses a few follow-ups from the initial PR after merge:
  - Makes `Memo` provider use deterministic proto marshalling
  - Fixes a few style nitpicks
  - Removes some unneeded TODOs

Author: lina-temporal

chasm/lib/scheduler/generator_tasks.go

@@ -101,7 +101,7 @@ func (g *GeneratorTaskExecutor) Execute(
 
 	// Write the new high water mark and future action times.
 	generator.LastProcessedTime = timestamppb.New(result.LastActionTime)
-	if err := g.updateFutureActionTimes(ctx, generator, scheduler); err != nil {
+	if err := g.updateFutureActionTimes(generator, scheduler); err != nil {
 		return queueerrors.NewUnprocessableTaskError(
 			fmt.Sprintf("failed to update future action times: %s", err.Error()))
 	}
@@ -143,12 +143,11 @@ func (g *GeneratorTaskExecutor) logSchedule(logger log.Logger, msg string, sched
 }
 
 func (g *GeneratorTaskExecutor) updateFutureActionTimes(
-	ctx chasm.MutableContext,
 	generator *Generator,
 	scheduler *Scheduler,
 ) error {
 	nextTime := func(t time.Time) (time.Time, error) {
-		res, err := g.SpecProcessor.GetNextTime(scheduler, t)
+		res, err := g.SpecProcessor.NextTime(scheduler, t)
 		return res.Next, err
 	}
 

chasm/lib/scheduler/invoker_tasks.go

@@ -20,9 +20,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
-	"go.temporal.io/server/common/payload"
 	"go.temporal.io/server/common/resource"
-	"go.temporal.io/server/common/searchattribute/sadefs"
 	"go.temporal.io/server/common/util"
 	queueerrors "go.temporal.io/server/service/history/queues/errors"
 	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
@@ -516,26 +514,6 @@ func (e *InvokerProcessBufferTaskExecutor) startWorkflowDeadline(
 	return start.ActualTime.AsTime().Add(timeout)
 }
 
-// startWorkflowSearchAttributes returns the search attributes to be applied to
-// workflows kicked off. Includes custom search attributes and Temporal-managed.
-func startWorkflowSearchAttributes(
-	scheduler *Scheduler,
-	nominal time.Time,
-) *commonpb.SearchAttributes {
-	attributes := scheduler.Schedule.GetAction().GetStartWorkflow().GetSearchAttributes()
-
-	fields := util.CloneMapNonNil(attributes.GetIndexedFields())
-	if p, err := payload.Encode(nominal); err == nil {
-		fields[sadefs.TemporalScheduledStartTime] = p
-	}
-	if p, err := payload.Encode(scheduler.ScheduleId); err == nil {
-		fields[sadefs.TemporalScheduledById] = p
-	}
-	return &commonpb.SearchAttributes{
-		IndexedFields: fields,
-	}
-}
-
 func (e *InvokerExecuteTaskExecutor) startWorkflow(
 	ctx context.Context,
 	scheduler *Scheduler,
@@ -574,7 +552,7 @@ func (e *InvokerExecuteTaskExecutor) startWorkflow(
 		Namespace:                scheduler.Namespace,
 		RequestId:                start.RequestId,
 		RetryPolicy:              requestSpec.RetryPolicy,
-		SearchAttributes:         startWorkflowSearchAttributes(scheduler, start.NominalTime.AsTime()),
+		SearchAttributes:         scheduler.startWorkflowSearchAttributes(start.NominalTime.AsTime()),
 		TaskQueue:                requestSpec.TaskQueue,
 		UserMetadata:             requestSpec.UserMetadata,
 		WorkflowExecutionTimeout: requestSpec.WorkflowExecutionTimeout,

chasm/lib/scheduler/scheduler.go

@@ -18,8 +18,11 @@ import (
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/payload"
+	"go.temporal.io/server/common/searchattribute/sadefs"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/worker/scheduler"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -63,7 +66,7 @@ const (
 
 var (
 	ErrConflictTokenMismatch = serviceerror.NewFailedPrecondition("mismatched conflict token")
-	ErrClosed                = serviceerror.NewInvalidArgument("schedule closed")
+	ErrClosed                = serviceerror.NewFailedPrecondition("schedule closed")
 	ErrUnprocessable         = serviceerror.NewInternal("unprocessable schedule")
 )
 
@@ -615,33 +618,25 @@ func (s *Scheduler) SearchAttributes(chasm.Context) []chasm.SearchAttributeKeyVa
 func (s *Scheduler) Memo(
 	ctx chasm.Context,
 ) map[string]chasm.VisibilityValue {
-	newInfo, err := s.ListInfo(ctx)
-	if err != nil {
-		// Unable to retrieve list info. Return nil to skip memo update.
-		// Error will be logged once loggers are available in CHASM.
-		return nil
-	}
+	newInfo := s.ListInfo(ctx)
 
-	payload, err := newInfo.Marshal()
+	infoPayload, err := proto.MarshalOptions{
+		Deterministic: true,
+	}.Marshal(newInfo)
 	if err != nil {
-		// Unable to marshal list info. Return nil to skip memo update.
-		// Error will be logged once loggers are available in CHASM.
 		return nil
 	}
 
-	// TODO - Memo provider is being updated to support protobufs, byte slices
-	// may cause oscillating writes to visibility given that proto serialization is
-	// non-deterministic.
 	return map[string]chasm.VisibilityValue{
-		visibilityMemoFieldInfo: chasm.VisibilityValueByteSlice(payload),
+		visibilityMemoFieldInfo: chasm.VisibilityValueByteSlice(infoPayload),
 	}
 }
 
 // ListInfo returns the ScheduleListInfo, used as the visibility memo, and to
 // answer List queries.
 func (s *Scheduler) ListInfo(
 	ctx chasm.Context,
-) (*schedulepb.ScheduleListInfo, error) {
+) *schedulepb.ScheduleListInfo {
 	spec := common.CloneProto(s.Schedule.Spec)
 
 	// Clear fields that are too large/not useful for the list view.
@@ -661,5 +656,24 @@ func (s *Scheduler) ListInfo(
 		Paused:            s.Schedule.State.Paused,
 		RecentActions:     util.SliceTail(s.Info.RecentActions, recentActionCount),
 		FutureActionTimes: generator.FutureActionTimes,
-	}, nil
+	}
+}
+
+// startWorkflowSearchAttributes returns the search attributes to be applied to
+// workflows kicked off. Includes custom search attributes and Temporal-managed.
+func (s *Scheduler) startWorkflowSearchAttributes(
+	nominal time.Time,
+) *commonpb.SearchAttributes {
+	attributes := s.Schedule.GetAction().GetStartWorkflow().GetSearchAttributes()
+
+	fields := util.CloneMapNonNil(attributes.GetIndexedFields())
+	if p, err := payload.Encode(nominal); err == nil {
+		fields[sadefs.TemporalScheduledStartTime] = p
+	}
+	if p, err := payload.Encode(s.ScheduleId); err == nil {
+		fields[sadefs.TemporalScheduledById] = p
+	}
+	return &commonpb.SearchAttributes{
+		IndexedFields: fields,
+	}
 }

chasm/lib/scheduler/scheduler_suite_test.go

@@ -93,11 +93,11 @@ func (s *schedulerSuite) SetupTest() {
 
 	// Set up future action times.
 	futureTime := now.Add(time.Hour)
-	s.specProcessor.EXPECT().GetNextTime(s.scheduler, gomock.Any()).Return(legacyscheduler.GetNextTimeResult{
+	s.specProcessor.EXPECT().NextTime(s.scheduler, gomock.Any()).Return(legacyscheduler.GetNextTimeResult{
 		Next:    futureTime,
 		Nominal: futureTime,
 	}, nil).MaxTimes(1)
-	s.specProcessor.EXPECT().GetNextTime(s.scheduler, gomock.Any()).Return(legacyscheduler.GetNextTimeResult{}, nil).AnyTimes()
+	s.specProcessor.EXPECT().NextTime(s.scheduler, gomock.Any()).Return(legacyscheduler.GetNextTimeResult{}, nil).AnyTimes()
 }
 
 // hasTask returns true if the given task type was added at the end of the

chasm/lib/scheduler/scheduler_test.go

@@ -16,8 +16,7 @@ func TestListInfo(t *testing.T) {
 	expectedFutureTimes := []*timestamppb.Timestamp{timestamppb.Now(), timestamppb.Now()}
 	generator.FutureActionTimes = expectedFutureTimes
 
-	listInfo, err := scheduler.ListInfo(ctx)
-	require.NoError(t, err)
+	listInfo := scheduler.ListInfo(ctx)
 
 	// Should return a populated info block.
 	require.NotNil(t, listInfo)

chasm/lib/scheduler/spec_processor.go

@@ -37,8 +37,8 @@ type (
 			limit *int,
 		) (*ProcessedTimeRange, error)
 
-		// GetNextTime provides a peek at the next time in the spec following 'after'.
-		GetNextTime(scheduler *Scheduler, after time.Time) (legacyscheduler.GetNextTimeResult, error)
+		// NextTime provides a peek at the next time in the spec following 'after'.
+		NextTime(scheduler *Scheduler, after time.Time) (legacyscheduler.GetNextTimeResult, error)
 	}
 
 	SpecProcessorImpl struct {
@@ -94,7 +94,7 @@ func (s *SpecProcessorImpl) ProcessTimeRange(
 	// Manual (backfill/patch) runs are always buffered here.
 	if !scheduler.useScheduledAction(false) && !manual {
 		// Use end as last action time so that we don't reprocess time spent paused.
-		next, err := s.GetNextTime(scheduler, end)
+		next, err := s.NextTime(scheduler, end)
 		if err != nil {
 			return nil, err
 		}
@@ -117,7 +117,7 @@ func (s *SpecProcessorImpl) ProcessTimeRange(
 	var next legacyscheduler.GetNextTimeResult
 	var err error
 	var bufferedStarts []*schedulespb.BufferedStart
-	for next, err = s.GetNextTime(scheduler, start); err == nil && (!next.Next.IsZero() && !next.Next.After(end)); next, err = s.GetNextTime(scheduler, next.Next) {
+	for next, err = s.NextTime(scheduler, start); err == nil && (!next.Next.IsZero() && !next.Next.After(end)); next, err = s.NextTime(scheduler, next.Next) {
 		lastAction = next.Next
 
 		if scheduler.Info.UpdateTime.AsTime().After(next.Next) {
@@ -172,8 +172,8 @@ func catchupWindow(s *Scheduler, tweakables Tweakables) time.Duration {
 	return max(cw.AsDuration(), tweakables.MinCatchupWindow)
 }
 
-// GetNextTime returns the next time result, or an error if the schedule cannot be compiled.
-func (s *SpecProcessorImpl) GetNextTime(scheduler *Scheduler, after time.Time) (legacyscheduler.GetNextTimeResult, error) {
+// NextTime returns the next time result, or an error if the schedule cannot be compiled.
+func (s *SpecProcessorImpl) NextTime(scheduler *Scheduler, after time.Time) (legacyscheduler.GetNextTimeResult, error) {
 	spec, err := scheduler.getCompiledSpec(s.specBuilder)
 	if err != nil {
 		s.logger.Error("Invalid schedule", tag.Error(err))