[Scheduled Actions] CHASM scheduled actions visibility support

chasm/lib/scheduler/backfiller_tasks_test.go

@@ -241,13 +241,9 @@ func (s *backfillerTasksSuite) runTestCase(c *backfillTestCase) {
 	// Either type of request will spawn a Backfiller and schedule an immediate pure task.
 	_, err = s.node.CloseTransaction()
 	s.NoError(err)
-	s.GreaterOrEqual(1, len(s.addedTasks))
-	task, ok := s.addedTasks[0].(*tasks.ChasmTaskPure)
-	s.True(ok)
-	s.Equal(chasm.TaskScheduledTimeImmediate, task.GetVisibilityTime())
+	s.True(s.hasTask(&tasks.ChasmTaskPure{}, chasm.TaskScheduledTimeImmediate))
 
 	// Run a backfill task.
-	s.addedTasks = make([]tasks.Task, 0) // Clear old tasks.
 	err = s.executor.Execute(ctx, backfiller, chasm.TaskAttributes{}, &schedulerpb.BackfillerTask{})
 	s.NoError(err)
 	_, err = s.node.CloseTransaction() // TODO - remove this when CHASM has unit testing hooks for task generation
@@ -260,7 +256,11 @@ func (s *backfillerTasksSuite) runTestCase(c *backfillTestCase) {
 		s.NoError(err)
 		s.Nil(res)
 	} else {
-		s.GreaterOrEqual(1, len(s.addedTasks))
+		// TODO - check that a pure task to continue driving backfill exists here. Because
+		// a pure task in the tree already has the physically-created status, closing the
+		// transaction won't call our backend mock for AddTasks twice. Fix this when CHASM
+		// offers unit testing hooks for task generation.
+
 		s.Equal(int64(c.ExpectedAttempt), backfiller.GetAttempt())
 		s.Equal(c.ExpectedLastProcessedTime.UTC(), backfiller.GetLastProcessedTime().AsTime())
 	}

chasm/lib/scheduler/generator_tasks.go

@@ -77,9 +77,10 @@ func (g *GeneratorTaskExecutor) Execute(
 	result, err := g.SpecProcessor.ProcessTimeRange(scheduler, t1, t2, scheduler.overlapPolicy(), "", false, nil)
 	if err != nil {
 		// An error here should be impossible, send to the DLQ.
-		logger.Error("error processing time range", tag.Error(err))
+		msg := "failed to process a time range"
+		logger.Error(msg, tag.Error(err))
 		return fmt.Errorf("%w: %w",
-			serviceerror.NewInternalf("failed to process a time range"),
+			serviceerror.NewInternal(msg),
 			err)
 	}
 
@@ -91,6 +92,16 @@ func (g *GeneratorTaskExecutor) Execute(
 	// Write the new high water mark.
 	generator.LastProcessedTime = timestamppb.New(result.LastActionTime)
 
+	// Update visibility, since future action times may have changed.
+	err = scheduler.UpdateVisibility(ctx, g.SpecProcessor, nil)
+	if err != nil {
+		msg := "failed to update visibility"
+		logger.Error(msg, tag.Error(err))
+		return fmt.Errorf("%w: %w",
+			serviceerror.NewInternal(msg),
+			err)
+	}
+
 	// Check if the schedule has gone idle.
 	idleTimeTotal := g.Config.Tweakables(scheduler.Namespace).IdleTime
 	idleExpiration, isIdle := scheduler.getIdleExpiration(ctx, idleTimeTotal, result.NextWakeupTime)

chasm/lib/scheduler/generator_tasks_test.go

@@ -82,8 +82,5 @@ func (s *generatorTasksSuite) TestExecuteBufferTask_Basic() {
 	// Ensure we scheduled an immediate physical pure task on the tree.
 	_, err = s.node.CloseTransaction()
 	s.NoError(err)
-	s.Equal(1, len(s.addedTasks))
-	task, ok := s.addedTasks[0].(*tasks.ChasmTaskPure)
-	s.True(ok)
-	s.Equal(chasm.TaskScheduledTimeImmediate, task.GetVisibilityTime())
+	s.True(s.hasTask(&tasks.ChasmTaskPure{}, chasm.TaskScheduledTimeImmediate))
 }

chasm/lib/scheduler/helper_test.go

@@ -3,6 +3,7 @@ package scheduler_test
 import (
 	"time"
 
+	commonpb "go.temporal.io/api/common/v1"
 	schedulepb "go.temporal.io/api/schedule/v1"
 	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/server/chasm/lib/scheduler"
@@ -34,7 +35,8 @@ func defaultSchedule() *schedulepb.Schedule {
 		Action: &schedulepb.ScheduleAction{
 			Action: &schedulepb.ScheduleAction_StartWorkflow{
 				StartWorkflow: &workflowpb.NewWorkflowExecutionInfo{
-					WorkflowId: "scheduled-wf",
+					WorkflowId:   "scheduled-wf",
+					WorkflowType: &commonpb.WorkflowType{Name: "scheduled-wf-type"},
 				},
 			},
 		},

chasm/lib/scheduler/invoker_execute_task_test.go

@@ -16,7 +16,6 @@ import (
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/testing/mockapi/workflowservicemock/v1"
-	"go.temporal.io/server/service/history/tasks"
 	"go.uber.org/mock/gomock"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -43,6 +42,7 @@ func (s *invokerExecuteTaskSuite) SetupTest() {
 		Config:         defaultConfig(),
 		MetricsHandler: metrics.NoopMetricsHandler,
 		BaseLogger:     s.logger,
+		SpecProcessor:  s.specProcessor,
 		HistoryClient:  s.mockHistoryClient,
 		FrontendClient: s.mockFrontendClient,
 	})
@@ -340,9 +340,6 @@ func (s *invokerExecuteTaskSuite) runExecuteTestCase(c *executeTestCase) {
 	s.ExpectReadComponent(invoker)
 	s.ExpectUpdateComponent(invoker)
 
-	// Clear old tasks and run the execute task.
-	s.addedTasks = make([]tasks.Task, 0)
-
 	// Create engine context for side effect task execution
 	engineCtx := s.newEngineContext()
 	err = s.executor.Execute(engineCtx, chasm.ComponentRef{}, chasm.TaskAttributes{}, &schedulerpb.InvokerExecuteTask{})

chasm/lib/scheduler/invoker_process_buffer_task_test.go

@@ -13,7 +13,6 @@ import (
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/util"
-	"go.temporal.io/server/service/history/tasks"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -33,6 +32,7 @@ func (s *invokerProcessBufferTaskSuite) SetupTest() {
 		Config:         defaultConfig(),
 		MetricsHandler: metrics.NoopMetricsHandler,
 		BaseLogger:     s.logger,
+		SpecProcessor:  s.specProcessor,
 	})
 }
 
@@ -313,9 +313,6 @@ func (s *invokerProcessBufferTaskSuite) runProcessBufferTestCase(c *processBuffe
 	// Set LastProcessedTime to current time to ensure time checks pass
 	invoker.LastProcessedTime = timestamppb.New(s.timeSource.Now())
 
-	// Clear old tasks and run the process buffer task
-	s.addedTasks = make([]tasks.Task, 0)
-
 	err = s.executor.Execute(ctx, invoker, chasm.TaskAttributes{}, &schedulerpb.InvokerProcessBufferTask{})
 	s.NoError(err)
 	_, err = s.node.CloseTransaction()

chasm/lib/scheduler/invoker_tasks.go

@@ -34,6 +34,7 @@ type (
 		Config         *Config
 		MetricsHandler metrics.Handler
 		BaseLogger     log.Logger
+		SpecProcessor  SpecProcessor
 
 		HistoryClient resource.HistoryClient
 
@@ -176,6 +177,12 @@ func (e *InvokerExecuteTaskExecutor) Execute(
 			i.recordExecuteResult(ctx, &result)
 			s.recordActionResult(&schedulerActionResult{starts: startResults})
 
+			// Update visibility, since RecentActions may have been updated.
+			err = s.UpdateVisibility(ctx, e.SpecProcessor, nil)
+			if err != nil {
+				return struct{}{}, err
+			}
+
 			return struct{}{}, nil
 		},
 		nil,

chasm/lib/scheduler/scheduler.go

@@ -1,13 +1,19 @@
 package scheduler
 
 import (
+	"bytes"
 	"fmt"
 	"time"
 
+	commonpb "go.temporal.io/api/common/v1"
 	enumspb "go.temporal.io/api/enums/v1"
 	schedulepb "go.temporal.io/api/schedule/v1"
 	"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/primitives/timestamp"
+	"go.temporal.io/server/common/searchattribute"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/service/worker/scheduler"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -29,6 +35,8 @@ type Scheduler struct {
 	Invoker     chasm.Field[*Invoker]
 	Backfillers chasm.Map[string, *Backfiller] // Backfill ID => *Backfiller
 
+	Visibility chasm.Field[*chasm.Visibility]
+
 	// Locally-cached state, invalidated whenever cacheConflictToken != ConflictToken.
 	cacheConflictToken int64
 	compiledSpec       *scheduler.CompiledSpec // compiledSpec is only ever replaced whole, not mutated.
@@ -37,6 +45,12 @@ type Scheduler struct {
 const (
 	// How many recent actions to keep on the Info.RecentActions list.
 	recentActionCount = 10
+
+	// Item limit per spec field on the ScheduleInfo memo.
+	listInfoSpecFieldLimit = 10
+
+	// Field in which the schedule's memo is stored.
+	visibilityMemoFieldInfo = "ScheduleInfo"
 )
 
 // NewScheduler returns an initialized CHASM scheduler root component.
@@ -66,10 +80,14 @@ func NewScheduler(
 	}
 
 	invoker := NewInvoker(ctx, sched)
-	generator := NewGenerator(ctx, sched, invoker)
 	sched.Invoker = chasm.NewComponentField(ctx, invoker)
+
+	generator := NewGenerator(ctx, sched, invoker)
 	sched.Generator = chasm.NewComponentField(ctx, generator)
 
+	visibility := chasm.NewVisibility(ctx)
+	sched.Visibility = chasm.NewComponentField(ctx, visibility)
+
 	return sched
 }
 
@@ -290,3 +308,165 @@ func (s *Scheduler) recordActionResult(result *schedulerActionResult) {
 		}
 	}
 }
+
+// UpdateVisibility updates the schedule's visibility record, including memo
+// and search attributes. customSearchAttributes is optional, and custom search
+// attributes will be left as-is with it unset.
+//
+// See mergeCustomSearchAttributes for how custom search attributes are merged.
+func (s *Scheduler) UpdateVisibility(
+	ctx chasm.MutableContext,
+	specProcessor SpecProcessor,
+	customSearchAttributes *commonpb.SearchAttributes,
+) error {
+	needsTask := false // Set to true if we need to write anything to Visibility.
+	visibility, err := s.Visibility.Get(ctx)
+	if err != nil {
+		return err
+	}
+
+	// Update the schedule's search attributes. This includes both Temporal-managed
+	// fields (paused state), as well as upserts for customer-specified fields.
+	upsertAttrs := make(map[string]any)
+	currentAttrs, err := visibility.GetSearchAttributes(ctx)
+	if err != nil {
+		return err
+	}
+
+	// Only attempt to merge additional search attributes if any are given, otherwise
+	// we'll unset existing attributes.
+	if customSearchAttributes != nil &&
+		len(customSearchAttributes.GetIndexedFields()) > 0 {
+		mergeCustomSearchAttributes(
+			currentAttrs,
+			customSearchAttributes.GetIndexedFields(),
+			upsertAttrs,
+		)
+	}
+
+	// Update Paused status.
+	var currentPaused bool
+	currentPausedPayload, ok := currentAttrs[searchattribute.TemporalSchedulePaused]
+	if ok {
+		err = payload.Decode(currentPausedPayload, &currentPaused)
+		if err != nil {
+			return err
+		}
+	}
+	if !ok || currentPaused != s.Schedule.State.Paused {
+		upsertAttrs[searchattribute.TemporalSchedulePaused] = s.Schedule.State.Paused
+	}
+
+	if len(upsertAttrs) > 0 {
+		err = visibility.UpsertSearchAttributes(ctx, upsertAttrs)
+		if err != nil {
+			return err
+		}
+		needsTask = true
+	}
+
+	newInfo, err := s.GetListInfo(ctx, specProcessor)
+	if err != nil {
+		return err
+	}
+	newInfoPayload, err := newInfo.Marshal()
+	if err != nil {
+		return err
+	}
+	currentMemo, err := visibility.GetMemo(ctx)
+	if err != nil {
+		return err
+	}
+	currentInfoPayload := currentMemo[visibilityMemoFieldInfo]
+
+	// Update visibility if the memo is out-of-date or absent.
+	if currentInfoPayload == nil ||
+		!bytes.Equal(currentInfoPayload.Data, newInfoPayload) {
+		newMemo := map[string]any{visibilityMemoFieldInfo: newInfoPayload}
+		err = visibility.UpsertMemo(ctx, newMemo)
+		if err != nil {
+			return err
+		}
+		needsTask = true
+	}
+
+	if needsTask {
+		visibility.GenerateTask(ctx)
+	}
+
+	return nil
+}
+
+// GetListInfo returns the ScheduleListInfo, used as the visibility memo, and to
+// answer List queries.
+func (s *Scheduler) GetListInfo(
+	ctx chasm.Context,
+	specProcessor SpecProcessor,
+) (*schedulepb.ScheduleListInfo, error) {
+	spec := common.CloneProto(s.Schedule.Spec)
+
+	// Clear fields that are too large/not useful for the list view.
+	spec.TimezoneData = nil
+
+	// Limit the number of specs and exclusions stored on the memo.
+	spec.ExcludeStructuredCalendar = util.SliceHead(spec.ExcludeStructuredCalendar, listInfoSpecFieldLimit)
+	spec.Interval = util.SliceHead(spec.Interval, listInfoSpecFieldLimit)
+	spec.StructuredCalendar = util.SliceHead(spec.StructuredCalendar, listInfoSpecFieldLimit)
+
+	futureActionTimes, err := s.getFutureActionTimes(ctx, specProcessor)
+	if err != nil {
+		return nil, err
+	}
+
+	return &schedulepb.ScheduleListInfo{
+		Spec:              spec,
+		WorkflowType:      s.Schedule.Action.GetStartWorkflow().GetWorkflowType(),
+		Notes:             s.Schedule.State.Notes,
+		Paused:            s.Schedule.State.Paused,
+		RecentActions:     util.SliceTail(s.Info.RecentActions, recentActionCount),
+		FutureActionTimes: futureActionTimes,
+	}, nil
+}
+
+// getFutureActionTimes returns up to min(`recentActionCount`, `RemainingActions`)
+// future action times. Future action times that precede the schedule's UpdateTime
+// are not included.
+func (s *Scheduler) getFutureActionTimes(
+	ctx chasm.Context,
+	specProcessor SpecProcessor,
+) ([]*timestamppb.Timestamp, error) {
+	generator, err := s.Generator.Get(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	nextTime := func(t time.Time) (time.Time, error) {
+		res, err := specProcessor.GetNextTime(s, t)
+		return res.Next, err
+	}
+
+	count := recentActionCount
+	if s.Schedule.State.LimitedActions {
+		count = min(int(s.Schedule.State.RemainingActions), recentActionCount)
+	}
+	out := make([]*timestamppb.Timestamp, 0, count)
+	t := timestamp.TimeValue(generator.LastProcessedTime)
+	for len(out) < count {
+		t, err = nextTime(t)
+		if err != nil {
+			return nil, err
+		}
+		if t.IsZero() {
+			break
+		}
+
+		if s.Info.UpdateTime.AsTime().After(t) {
+			// Skip action times whose nominal times are prior to the schedule's update time.
+			continue
+		}
+
+		out = append(out, timestamppb.New(t))
+	}
+
+	return out, nil
+}