[Scheduled Actions V2] WIP Generator component #6905

lina-temporal · 2024-11-27T23:44:36Z

Based on #6904

What changed?

Added the WIP (but mostly implementation-complete) Generator component, a sub-state machine of the top-level Scheduler

Why?

The generator generates buffered actions, and we want those!

How did you test it?

I haven't, testing is next (this is a draft PR)

bergundy · 2024-11-28T18:38:59Z

components/scheduler2/generator/executors.go

+	"go.temporal.io/server/service/worker/scheduler"
+	scheduler1 "go.temporal.io/server/service/worker/scheduler"


Duplicate import.

bergundy · 2024-11-28T18:44:23Z

components/scheduler2/generator/executors.go

+	}
+
+	// process time range between last high water mark and system time
+	t1 := timestamp.TimeValue(generator.LastProcessedTime)


Why not this? (out of curiosity)

Suggested change

t1 := timestamp.TimeValue(generator.LastProcessedTime)

t1 := generator.LastProcessedTime.AsTime()

I'm not sure, artifact from original implementation I suppose. will fix :)

bergundy · 2024-11-28T18:45:24Z

components/scheduler2/generator/executors.go

+	t1 := timestamp.TimeValue(generator.LastProcessedTime)
+	t2 := time.Now()
+	if t2.Before(t1) {
+		e.Logger.Warn("Time went backwards",


Wondering how this log would be printed. Could be worth baking a logger with some relevant tags.

bergundy · 2024-11-28T18:53:09Z

components/scheduler2/generator/tasks.go

+	return common.ValidateTask(node, TransitionBuffer)
+}
+
+var TransitionSleep = hsm.NewTransition(


I typically put the transitions with the state machine. I think we should try and keep that consistent.

Sure, will do.

bergundy · 2024-11-28T19:19:16Z

components/scheduler2/generator/tasks.go

+	TaskTypeSleep  = "scheduler.generator.Sleep"
+	TaskTypeBuffer = "scheduler.generator.Buffer"


Wondering why you need two tasks here, why don't you buffer in the sleep task under a write lock? It's less DB writes this way. You should only need the buffer task if you plan to do any IO or other type of side effect.

Wondering why you need two tasks here, why don't you buffer in the sleep task under a write lock? It's less DB writes this way. You should only need the buffer task if you plan to do any IO or other type of side effect.

If that's an option (buffering under a write lock), that works for me, since there shouldn't be any regular contention - posted a question on your other comment, re: ConflictToken vs write locking.

I was thinking about WAITING versus EXECUTING in the context of the Backfiller, as well, where a user's backfill request will trigger an immediate Backfiller state transition and task to start backfilling. For Generator, if we don't have a separate EXECUTING task to transition to after a user updates, how would we signal the sleeping Generator sub-state machine? If re-entering the WAITING state works, then we can probably also get rid of EXECUTING on the Backfiller.

bergundy · 2024-11-28T19:20:23Z

components/scheduler2/generator/executors.go

+			return err
+		}
+		if refreshedScheduler.ConflictToken != scheduler.ConflictToken {
+			// schedule was updated while processing, retry


Do we actually want to retry the task here? I commented on the transition definition too but I think you can do all of this under a write lock since there's no IO involved AFAICT.

Maybe not - If we were doing this under the write lock on the generator node, is the expectation that we'd also attempt to acquire the generator node for write access during an update request? I've been assuming that env.Access scopes applied only to the given ref, do they/can they apply more broadly?

The lock is held for the entire mutable state record including the whole tree structure.

Ah! Okay, I'd been thinking that each component could potentially map to its own mutable state. One MS simplifies that. Will update to just fail out the task (I believe ErrStaleReference is for this purpose?)

bergundy

Let's start with a clean slate and delete the PoC scheduler component and all of its protos.

dnr · 2024-12-04T14:06:44Z

components/scheduler2/generator/statemachine.go

+	if !ok {
+		return 0, fmt.Errorf("%w: expected state1 to be a Generator instance, got %v", hsm.ErrIncompatibleType, s1)
+	}
+	s2, ok := a.(Generator)


Suggested change

s2, ok := a.(Generator)

s2, ok := b.(Generator)

dnr · 2024-12-04T14:08:44Z

components/scheduler2/generator/executors.go

+			// We're reprocessing since the most recent event after an update. Discard actions before
+			// the update time (which was just set to "now"). This doesn't have to be guarded with
+			// hasMinVersion because this condition couldn't happen in previous versions.


I imagine this comment should be rewritten somewhat. We don't have versions anymore, at least. Also I'm not sure how updates fit into this execution model, it may not be true that "we just set update time to 'now'"

Oops, yes, will rewrite this.

dnr · 2024-12-04T14:10:08Z

components/scheduler2/generator/executors.go

+
+// Returns the next time result, or an error if the schedule cannot be compiled.
+func (e taskExecutor) getNextTime(s core.Scheduler, after time.Time) (scheduler.GetNextTimeResult, error) {
+	spec, err := s.CompiledSpec(e.SpecBuilder)


just curious if these can get cached at all. it's not that expensive to construct so probably not worth it, except for backfills where it's doing this in a loop during one transition

oh, I see, the caching is elsewhere. nm

dnr · 2024-12-04T14:11:37Z

components/scheduler2/generator/statemachine.go

+	if s1.State() > s2.State() {
+		return 1, nil
+	} else if s1.State() < s2.State() {
+		return -1, nil
+	}
+
+	return 0, nil


Suggested change

if s1.State() > s2.State() {

return 1, nil

} else if s1.State() < s2.State() {

return -1, nil

}

return 0, nil

return cmp.Compare(s1.State(), s2.State()), nil

dnr · 2024-12-04T14:22:00Z

components/scheduler2/generator/executors.go

+		bufferedStarts = append(bufferedStarts, &schedpb.BufferedStart{
+			NominalTime:   timestamppb.New(next.Nominal),
+			ActualTime:    timestamppb.New(next.Next),
+			OverlapPolicy: scheduler.OverlapPolicy(),


Suggested change

OverlapPolicy: scheduler.OverlapPolicy(),

OverlapPolicy: overlapPolicy,

?

bergundy

Reviewed all the way to spec_processor.go, will defer to @dnr to verify the business logic for lack of context.

bergundy · 2024-12-18T05:12:57Z

components/scheduler2/generator.go

+	// to the schedule's specification. Manually requested actions (from an immediate
+	// request or backfill) are separately handled in the Backfiller sub state machine.
+	Generator struct {
+		*schedspb.GeneratorInternal


Consider removing the redundant pointer here and embedding the struct directly.

bergundy · 2024-12-18T05:15:57Z

components/scheduler2/generator_executors.go

+
+	// Process time range between last high water mark and system time.
+	t1 := generator.LastProcessedTime.AsTime()
+	t2 := time.Now().UTC()


Use env.Now() instead, this will potentially allow you to inject a fake time source and fast forward for testing purposes.

bergundy · 2024-12-18T05:17:05Z

components/scheduler2/generator_executors.go

+	res, err := e.SpecProcessor.ProcessTimeRange(scheduler, t1, t2, false, nil)
+	if err != nil {
+		// An error here should be impossible, send to the DLQ.
+		e.Logger.Error("Error processing time range", tag.Error(err))


Consider baking a logger with the relevant scheduler attributes as tags.

bergundy · 2024-12-18T05:17:48Z

components/scheduler2/generator_executors.go

+			err,
+			serviceerror.NewInternal("Scheduler's Generator failed to process a time range"),


These should typically be reversed:

Suggested change

err,

serviceerror.NewInternal("Scheduler's Generator failed to process a time range"),

serviceerror.NewInternal("Scheduler's Generator failed to process a time range"),

err,

bergundy · 2024-12-18T05:18:36Z

components/scheduler2/generator_executors.go

+	})
+	if err != nil {
+		return fmt.Errorf(
+			"%w: unable to transition Executor to Executing state",


Typically the wrapped error comes in the end.

Suggested change

"%w: unable to transition Executor to Executing state",

"unable to transition Executor to Executing state: %w",

bergundy · 2024-12-18T05:24:45Z

components/scheduler2/generator_executors.go

+	}
+
+	return Scheduler{
+		SchedulerInternal: servercommon.CloneProto(prevScheduler.SchedulerInternal),


bergundy · 2024-12-18T05:25:42Z

components/scheduler2/generator_tasks.go

+
+func (g Generator) tasks() ([]hsm.Task, error) {
+	switch g.State() { // nolint:exhaustive
+	case enumsspb.SCHEDULER_GENERATOR_STATE_BUFFERING:


Shouldn't we always be buffering?

bergundy · 2024-12-18T05:26:44Z

components/scheduler2/generator_tasks.go

+func (g Generator) tasks() ([]hsm.Task, error) {
+	switch g.State() { // nolint:exhaustive
+	case enumsspb.SCHEDULER_GENERATOR_STATE_BUFFERING:
+		return []hsm.Task{BufferTask{deadline: g.NextInvocationTime.AsTime()}}, nil


Isn't NextInvocationTime a computed property? Based on the first item in the buffer? Seems like a better representation. I may be commenting out of context so LMK if I'm off here.

bergundy · 2024-12-18T05:28:18Z

components/scheduler2/generator.go

+	panic("TODO: CompareState not yet implemented for Generator")
+}
+
+var TransitionBuffer = hsm.NewTransition(


I've mentioned this before but to reiterate, there's only one state, you probably don't need to define state transitions.

If you insist on defining this transition, you should accept a transition from BUFFERING to BUFFERING.

bergundy · 2024-12-18T05:33:16Z

components/scheduler2/generator_executors.go

+	err = hsm.MachineTransition(node, func(g Generator) (hsm.TransitionOutput, error) {
+		wakeupTime := res.NextWakeupTime
+		g.LastProcessedTime = timestamppb.New(res.LastActionTime)
+		g.NextInvocationTime = timestamppb.New(wakeupTime)


I'd be careful caching this property, it may be better to compute it on the fly and store it only on the generated task's deadline.

V2 scheduler Generator substate machine

b48c717

lina-temporal assigned bergundy and yycptt Nov 27, 2024

bergundy reviewed Nov 28, 2024

View reviewed changes

bergundy reviewed Dec 2, 2024

View reviewed changes

bergundy mentioned this pull request Dec 2, 2024

[Scheduled Actions V2] state machine protobufs WIP #6901

Draft

lina-temporal unassigned bergundy and yycptt Dec 3, 2024

dnr reviewed Dec 4, 2024

View reviewed changes

lina-temporal added 2 commits December 4, 2024 17:21

generator feedback

e72743d

generator feedback, testing

1efe96c

bergundy reviewed Dec 18, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Scheduled Actions V2] WIP Generator component #6905

[Scheduled Actions V2] WIP Generator component #6905

lina-temporal commented Nov 27, 2024

bergundy Nov 28, 2024

lina-temporal Dec 3, 2024

bergundy Nov 28, 2024

lina-temporal Dec 3, 2024

bergundy Nov 28, 2024

bergundy Nov 28, 2024

lina-temporal Dec 3, 2024

bergundy Nov 28, 2024

lina-temporal Dec 3, 2024

lina-temporal Dec 3, 2024 •

edited

Loading

bergundy Nov 28, 2024

lina-temporal Dec 3, 2024

bergundy Dec 3, 2024

lina-temporal Dec 4, 2024

bergundy left a comment

dnr Dec 4, 2024

dnr Dec 4, 2024

lina-temporal Dec 4, 2024

dnr Dec 4, 2024 •

edited

Loading

dnr Dec 4, 2024

dnr Dec 4, 2024

bergundy left a comment

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

bergundy Dec 18, 2024

		"go.temporal.io/server/service/worker/scheduler"
		scheduler1 "go.temporal.io/server/service/worker/scheduler"

	t1 := timestamp.TimeValue(generator.LastProcessedTime)
	t1 := generator.LastProcessedTime.AsTime()

		TaskTypeSleep = "scheduler.generator.Sleep"
		TaskTypeBuffer = "scheduler.generator.Buffer"

	OverlapPolicy: scheduler.OverlapPolicy(),
	OverlapPolicy: overlapPolicy,

		err,
		serviceerror.NewInternal("Scheduler's Generator failed to process a time range"),

	"%w: unable to transition Executor to Executing state",
	"unable to transition Executor to Executing state: %w",

[Scheduled Actions V2] WIP Generator component #6905

Are you sure you want to change the base?

[Scheduled Actions V2] WIP Generator component #6905

Conversation

lina-temporal commented Nov 27, 2024

What changed?

Why?

How did you test it?

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lina-temporal Dec 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bergundy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dnr Dec 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bergundy left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lina-temporal Dec 3, 2024 •

edited

Loading

dnr Dec 4, 2024 •

edited

Loading