Bound CHASM scheduler request IDs

chasm/lib/scheduler/backfiller.go

@@ -7,7 +7,7 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
-	schedulescommon "go.temporal.io/server/common/schedules"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -36,7 +36,7 @@ func addBackfiller(
 	ctx chasm.MutableContext,
 	scheduler *Scheduler,
 ) *Backfiller {
-	id := schedulescommon.GenerateBackfillerID()
+	id := schedulerinternal.GenerateBackfillerID()
 	backfiller := newBackfillerWithState(ctx, &schedulerpb.BackfillerState{
 		BackfillId:        id,
 		LastProcessedTime: timestamppb.New(ctx.Now(scheduler)),

chasm/lib/scheduler/backfiller_tasks.go

@@ -7,10 +7,10 @@ import (
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	"go.temporal.io/server/chasm"
 	"go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
-	schedulescommon "go.temporal.io/server/common/schedules"
 	queueerrors "go.temporal.io/server/service/history/queues/errors"
 	"go.uber.org/fx"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -219,7 +219,7 @@ func (b *BackfillerTaskHandler) processTrigger(
 	nowpb := backfiller.GetLastProcessedTime()
 	now := nowpb.AsTime()
 	requestID := generateRequestID(scheduler, backfiller.GetBackfillId(), now, now)
-	workflowID := schedulescommon.GenerateWorkflowID(scheduler.WorkflowID(), now)
+	workflowID := schedulerinternal.GenerateWorkflowID(scheduler.WorkflowID(), now)
 	result.BufferedStarts = []*schedulespb.BufferedStart{
 		{
 			NominalTime:   nowpb,

common/schedules/id.go → chasm/lib/scheduler/internal/request_id.go

@@ -1,4 +1,4 @@
-package schedules
+package internal
 
 import (
 	"fmt"
@@ -7,9 +7,13 @@ import (
 	"github.com/google/uuid"
 )
 
+const lowestKnownSchemaRequestIDColumnLimit = 255
+
 // GenerateRequestID 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.
+// Its total length must not exceed SQLite's request ID VARCHAR size, the
+// smallest known persistence limit.
 //
 // backfillID should be left blank for actions that are being started
 // automatically, based on the schedule spec. It must be set for backfills,
@@ -26,11 +30,14 @@ func GenerateRequestID(
 	if backfillID == "" {
 		backfillID = "auto"
 	}
+
+	// Keep request IDs bounded and deterministic even when schedule IDs are long.
+	scheduleIDUUID := uuid.NewSHA1(uuid.Nil, []byte(scheduleID))
 	return fmt.Sprintf(
 		"sched-%s-%s-%s-%d-%d-%d",
 		backfillID,
 		namespaceID,
-		scheduleID,
+		scheduleIDUUID,
 		conflictToken,
 		nominal.UnixMilli(),
 		actual.UnixMilli(),

chasm/lib/scheduler/internal/request_id_test.go

@@ -0,0 +1,85 @@
+package internal
+
+import (
+	"fmt"
+	"math"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+)
+
+func TestGenerateWorkflowID(t *testing.T) {
+	baseWorkflowID := "my-workflow"
+	nominalTime := time.Date(2024, 6, 15, 10, 30, 45, 123456789, time.UTC)
+
+	actual := GenerateWorkflowID(baseWorkflowID, nominalTime)
+	require.Equal(t, "my-workflow-2024-06-15T10:30:45Z", actual)
+}
+
+func TestGenerateRequestID(t *testing.T) {
+	nominalTime := time.Now()
+	actualTime := time.Now()
+	scheduleID := "mysched"
+	scheduleIDUUID := uuid.NewSHA1(uuid.Nil, []byte(scheduleID))
+
+	actual := GenerateRequestID("nsid", scheduleID, 10, "", nominalTime, actualTime)
+	expected := fmt.Sprintf(
+		"sched-auto-nsid-%s-10-%d-%d",
+		scheduleIDUUID,
+		nominalTime.UnixMilli(),
+		actualTime.UnixMilli(),
+	)
+	require.Equal(t, expected, actual)
+
+	actual = GenerateRequestID("nsid", scheduleID, 10, "backfillid", nominalTime, actualTime)
+	expected = fmt.Sprintf(
+		"sched-backfillid-nsid-%s-10-%d-%d",
+		scheduleIDUUID,
+		nominalTime.UnixMilli(),
+		actualTime.UnixMilli(),
+	)
+	require.Equal(t, expected, actual)
+}
+
+func TestGenerateRequestIDUsesUUIDv5ForScheduleID(t *testing.T) {
+	nominalTime := time.UnixMilli(1_700_000_000_000)
+	actualTime := time.UnixMilli(1_700_000_000_001)
+	scheduleID := strings.Repeat("a", 1000)
+	scheduleIDUUID := uuid.NewSHA1(uuid.Nil, []byte(scheduleID))
+	require.Equal(t, uuid.Version(5), scheduleIDUUID.Version())
+
+	for _, tc := range []struct {
+		name       string
+		backfillID string
+		prefix     string
+	}{
+		{name: "automatic", prefix: "auto"},
+		{name: "backfill", backfillID: "backfill-id", prefix: "backfill-id"},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			requestID := GenerateRequestID("nsid", scheduleID, 1, tc.backfillID, nominalTime, actualTime)
+			require.Contains(t, requestID, fmt.Sprintf("sched-%s-nsid-%s-1-", tc.prefix, scheduleIDUUID))
+			require.NotContains(t, requestID, scheduleID)
+			require.LessOrEqual(t, len(requestID), lowestKnownSchemaRequestIDColumnLimit)
+
+			otherRequestID := GenerateRequestID("nsid", scheduleID+"b", 1, tc.backfillID, nominalTime, actualTime)
+			require.NotEqual(t, requestID, otherRequestID)
+		})
+	}
+}
+
+func TestGenerateRequestIDFitsSQLColumn(t *testing.T) {
+	requestID := GenerateRequestID(
+		strings.Repeat("a", 36),
+		"schedule-id",
+		math.MinInt64,
+		strings.Repeat("b", 36),
+		time.UnixMilli(math.MinInt64),
+		time.UnixMilli(math.MaxInt64),
+	)
+
+	require.LessOrEqual(t, len(requestID), lowestKnownSchemaRequestIDColumnLimit)
+}

chasm/lib/scheduler/migration/migration.go

@@ -10,8 +10,8 @@ import (
 	schedulepb "go.temporal.io/api/schedule/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
 	schedulerpb "go.temporal.io/server/chasm/lib/scheduler/gen/schedulerpb/v1"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common"
-	schedulescommon "go.temporal.io/server/common/schedules"
 	"go.temporal.io/server/common/searchattribute/sadefs"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -216,7 +216,7 @@ func convertBufferedStartsLegacyToCHASM(
 		v2Start := common.CloneProto(v1Start)
 
 		if v2Start.RequestId == "" {
-			v2Start.RequestId = schedulescommon.GenerateRequestID(
+			v2Start.RequestId = schedulerinternal.GenerateRequestID(
 				namespaceID,
 				scheduleID,
 				conflictToken,
@@ -227,7 +227,7 @@ func convertBufferedStartsLegacyToCHASM(
 		}
 
 		if v2Start.WorkflowId == "" {
-			v2Start.WorkflowId = schedulescommon.GenerateWorkflowID(
+			v2Start.WorkflowId = schedulerinternal.GenerateWorkflowID(
 				baseWorkflowID,
 				v1Start.GetNominalTime().AsTime(),
 			)
@@ -268,7 +268,7 @@ func convertRunningWorkflowsToBufferedStarts(
 			// Include the RunId in the tag to ensure each running workflow
 			// gets a unique RequestId (important for ALLOW_ALL overlap
 			// policy where multiple workflows may be running concurrently).
-			RequestId: schedulescommon.GenerateRequestID(
+			RequestId: schedulerinternal.GenerateRequestID(
 				namespaceID,
 				scheduleID,
 				conflictToken,
@@ -332,7 +332,7 @@ func convertRecentActionsToBufferedStarts(
 			StartTime:   action.ActualTime,
 			WorkflowId:  action.StartWorkflowResult.WorkflowId,
 			RunId:       action.StartWorkflowResult.RunId,
-			RequestId: schedulescommon.GenerateRequestID(
+			RequestId: schedulerinternal.GenerateRequestID(
 				namespaceID,
 				scheduleID,
 				conflictToken,
@@ -359,7 +359,7 @@ func convertBackfillsLegacyToCHASM(
 
 	backfillers := make(map[string]*schedulerpb.BackfillerState, len(legacyBackfills))
 	for _, v1Backfill := range legacyBackfills {
-		backfillID := schedulescommon.GenerateBackfillerID()
+		backfillID := schedulerinternal.GenerateBackfillerID()
 
 		backfillers[backfillID] = &schedulerpb.BackfillerState{
 			Request: &schedulerpb.BackfillerState_BackfillRequest{

chasm/lib/scheduler/spec_processor.go

@@ -6,10 +6,10 @@ import (
 
 	enumspb "go.temporal.io/api/enums/v1"
 	schedulespb "go.temporal.io/server/api/schedule/v1"
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
-	schedulescommon "go.temporal.io/server/common/schedules"
 	legacyscheduler "go.temporal.io/server/service/worker/scheduler"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
@@ -179,7 +179,7 @@ func (s *SpecProcessorImpl) ProcessTimeRange(
 			OverlapPolicy: overlapPolicy,
 			Manual:        manual,
 			RequestId:     generateRequestID(scheduler, backfillID, next.Nominal, next.Next),
-			WorkflowId:    schedulescommon.GenerateWorkflowID(workflowID, next.Nominal),
+			WorkflowId:    schedulerinternal.GenerateWorkflowID(workflowID, next.Nominal),
 		})
 
 		if limit != nil {

chasm/lib/scheduler/util.go

@@ -4,17 +4,17 @@ import (
 	"encoding/binary"
 	"time"
 
+	schedulerinternal "go.temporal.io/server/chasm/lib/scheduler/internal"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/metrics"
-	schedulescommon "go.temporal.io/server/common/schedules"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 func generateRequestID(scheduler *Scheduler, backfillID string, nominal, actual time.Time) string {
-	return schedulescommon.GenerateRequestID(
+	return schedulerinternal.GenerateRequestID(
 		scheduler.NamespaceId,
 		scheduler.ScheduleId,
 		scheduler.ConflictToken,

tests/schedule_test.go

@@ -416,6 +416,7 @@ func TestScheduleCHASM(t *testing.T) {
 	t.Run("TestMigrationCallbackAttach", func(t *testing.T) { t.Parallel(); testMigrationCallbackAttach(t, newContext) })
 	t.Run("TestCreatesWorkflowSentinel", func(t *testing.T) { t.Parallel(); testCreatesWorkflowSentinel(t, newContext) })
 	t.Run("TestSkipsWorkflowSentinelWhenDisabled", func(t *testing.T) { t.Parallel(); testSkipsWorkflowSentinelWhenDisabled(t, newContext) })
+	t.Run("TestLargeScheduleID", func(t *testing.T) { t.Parallel(); testLargeScheduleID(t, newContext) })
 	t.Run("TestUpdateScheduleMemo", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemo(t, newContext) })
 	t.Run("TestUpdateScheduleMemoOnly", func(t *testing.T) { t.Parallel(); testUpdateScheduleMemoOnly(t, newContext) })
 	t.Run("TestStateSizeBytesReported", func(t *testing.T) { t.Parallel(); testStateSizeBytesReported(t, newContext) })
@@ -4604,6 +4605,30 @@ func testUpdateScheduleRequestIDTooLong(t *testing.T, newContext contextFactory)
 	require.ErrorAs(t, err, &invalidArgReqID)
 }
 
+func testLargeScheduleID(t *testing.T, newContext contextFactory) {
+	s := newScheduleEnv(t, scheduleCommonOpts(t)...)
+	ctx := newContext(testcore.NewContext())
+
+	// The V1 sentinel shares the SQL workflow ID limit with the schedule ID
+	// prefix, so this is the largest schedule ID supported by every SQL backend.
+	const workflowIDColumnLimit = 255
+	scheduleIDLength := workflowIDColumnLimit - len(scheduler.WorkflowIDPrefix)
+	sid := strings.Repeat("a", scheduleIDLength)
+	wid := testcore.RandomizeStr("sched-large-id-wf")
+	wt := testcore.RandomizeStr("sched-large-id-wt")
+
+	var runs atomic.Int32
+	registerCountingWorkflow(s, wt, &runs)
+
+	createSchedule(ctx, t, s, sid, &schedulepb.Schedule{
+		Spec:   intervalSpec(fastInterval),
+		Action: startWorkflowAction(s, wid, wt),
+	})
+
+	await.RequireTruef(t, func() bool { return runs.Load() > 0 }, awaitTimeout, pollInterval,
+		"schedule ID of length %d should start a workflow", scheduleIDLength)
+}
+
 func testUpdateScheduleBlobSizeLimit(t *testing.T, newContext contextFactory) {
 	s := newScheduleEnv(t,
 		append(scheduleCommonOpts(t),