[Scheduled Actions V2] WIP Common/util Package

components/scheduler2/common/config.go

@@ -0,0 +1,53 @@
+package common
+
+import (
+	"time"
+
+	"go.temporal.io/server/common/dynamicconfig"
+)
+
+type (
+	Tweakables struct {
+		DefaultCatchupWindow              time.Duration // Default for catchup window
+		MinCatchupWindow                  time.Duration // Minimum for catchup window
+		MaxBufferSize                     int           // MaxBufferSize limits the number of buffered actions pending execution in total
+		BackfillsPerIteration             int           // How many backfilled actions to buffer per iteration (implies rate limit since min sleep is 1s)
+		CanceledTerminatedCountAsFailures bool          // Whether cancelled+terminated count for pause-on-failure
+
+		// TODO - incomplete tweakables list
+	}
+
+	// V2 Scheduler dynamic config, shared among all substate machines.
+	Config struct {
+		Tweakables       dynamicconfig.TypedPropertyFnWithNamespaceFilter[Tweakables]
+		ExecutionTimeout dynamicconfig.DurationPropertyFnWithNamespaceFilter
+	}
+)
+
+var (
+	CurrentTweakables = dynamicconfig.NewNamespaceTypedSetting(
+		"component.scheduler.tweakables",
+		DefaultTweakables,
+		"A set of tweakable parameters for the V2 scheduler")
+
+	ExecutionTimeout = dynamicconfig.NewNamespaceDurationSetting(
+		"component.scheduler.executionTimeout",
+		time.Second*10,
+		`ExecutionTimeout is the timeout for executing a single scheduler task.`,
+	)
+
+	DefaultTweakables = Tweakables{
+		DefaultCatchupWindow:              365 * 24 * time.Hour,
+		MinCatchupWindow:                  10 * time.Second,
+		MaxBufferSize:                     1000,
+		BackfillsPerIteration:             10,
+		CanceledTerminatedCountAsFailures: false,
+	}
+)
+
+func ConfigProvider(dc *dynamicconfig.Collection) *Config {
+	return &Config{
+		Tweakables:       CurrentTweakables.Get(dc),
+		ExecutionTimeout: ExecutionTimeout.Get(dc),
+	}
+}

components/scheduler2/common/util.go

@@ -0,0 +1,77 @@
+package common
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	servercommon "go.temporal.io/server/common"
+	"go.temporal.io/server/components/scheduler2/core"
+	"go.temporal.io/server/service/history/consts"
+	"go.temporal.io/server/service/history/hsm"
+)
+
+func ValidateTask[
+	S comparable,
+	SM hsm.StateMachine[S],
+	E any](node *hsm.Node, transition hsm.Transition[S, SM, E]) error {
+	if err := node.CheckRunning(); err != nil {
+		return err
+	}
+	sm, err := hsm.MachineData[SM](node)
+	if err != nil {
+		return err
+	}
+	if !transition.Possible(sm) {
+		return fmt.Errorf(
+			"%w: %w: cannot transition from state %v to %v",
+			consts.ErrStaleReference,
+			hsm.ErrInvalidTransition,
+			sm.State(),
+			transition.Destination,
+		)
+	}
+	return nil
+}
+
+// Intended to be called with a substate machine's node. Returns a cloned copy
+// of the top-level Scheduler from its persisted state.
+func LoadSchedulerFromParent(
+	ctx context.Context,
+	env hsm.Environment,
+	ref hsm.Ref) (scheduler core.Scheduler, err error) {
+	err = env.Access(ctx, ref, hsm.AccessRead, func(node *hsm.Node) error {
+		prevScheduler, err := hsm.MachineData[core.Scheduler](node.Parent)
+		if err != nil {
+			return err
+		}
+		scheduler = core.Scheduler{
+			HsmSchedulerV2State: servercommon.CloneProto(prevScheduler.HsmSchedulerV2State),
+		}
+		return nil
+	})
+	return
+}
+
+// Generates a deterministic request ID for a buffered action's time. The request
+// ID is deterministic because the jittered actual time (as well as the spec's
+// nominal time) is, in turn, also deterministic.
+//
+// backfillID should be left blank for actions that are being started
+// automatically, based on the schedule spec. It must be set for backfills,
+// as backfills may generate buffered actions that overlap with both
+// automatically-buffered actions, as well as other requested backfills.
+func GenerateRequestID(scheduler core.Scheduler, backfillID string, nominal, actual time.Time) string {
+	if backfillID == "" {
+		backfillID = "auto"
+	}
+	return fmt.Sprintf(
+		"sched-%s-%s-%s-%d-%d-%d",
+		backfillID,
+		scheduler.NamespaceId,
+		scheduler.ScheduleId,
+		scheduler.ConflictToken,
+		nominal.UnixMilli(),
+		actual.UnixMilli(),
+	)
+}