From 8e79d78d1ca901b7c806dffde68bada86f1260fd Mon Sep 17 00:00:00 2001 From: lilydoar Date: Mon, 26 Oct 2026 13:50:26 -0700 Subject: [PATCH 1/3] Reconcile bounded scheduler operations after transient RPC failures --- docs/scheduler-operation-recovery.md | 43 ++ scenarios/scheduler_recovery.go | 316 ++++++++++++++ scenarios/scheduler_recovery_budget_test.go | 101 +++++ scenarios/scheduler_recovery_rpc_test.go | 439 ++++++++++++++++++++ scenarios/scheduler_recovery_server_test.go | 132 ++++++ scenarios/scheduler_stress.go | 67 ++- 6 files changed, 1087 insertions(+), 11 deletions(-) create mode 100644 docs/scheduler-operation-recovery.md create mode 100644 scenarios/scheduler_recovery.go create mode 100644 scenarios/scheduler_recovery_budget_test.go create mode 100644 scenarios/scheduler_recovery_rpc_test.go create mode 100644 scenarios/scheduler_recovery_server_test.go diff --git a/docs/scheduler-operation-recovery.md b/docs/scheduler-operation-recovery.md new file mode 100644 index 000000000..949b8472a --- /dev/null +++ b/docs/scheduler-operation-recovery.md @@ -0,0 +1,43 @@ +# Scheduler operation recovery + +`--option recover-schedule-operations=true` enables operation recovery for the +whole scheduler workload. It defaults to false. It does not change GenericExecutor or activity retries. + +Each operation shares a budget of four logical calls, including probes, and +five seconds including SDK calls and cancelable 100ms inter-call waits. Create, +read and update are also bounded by the original EndAt; delete may clean up +after EndAt while the parent remains live. SDK Update itself performs Describe +and Update, and SDK transport retries may send multiple physical RPCs per logical +call. This budget is not four RPCs or four independent five-second timeouts. + +Create freezes its ID, spec, action, overlap and EndAt before the first call. +An ownership Memo identifies the create intent. After an ambiguous error, a +matching owned object is recorded as observed creation, not an acknowledged +Create RPC. An absent object permits another Create with the same ID and frozen +intent; unresolved or foreign objects remain failures. + +Update gives each intended mutation a fresh State.Note witness, preserving the +schedule/action window and payload. A failure in SDK's initial Describe can retry +the SDK call. After a mutation may have been sent, recovery only probes: the +fresh witness must be observed with the owned intent. Matching the old payload +or passing EndAt is not proof of update execution. If the witness is not observed +within budget, the outcome remains unknown and fails. No ambiguous Update is +replayed by this layer. A server overwrite of Note can prevent reconciliation. + +Delete probes ownership before writing and after ambiguous failure. Only the +same owned ID can be retried. Observed absence and direct NotFound keep existing +compatibility without claiming a Delete or Update acknowledgment. Unknown +objects remain the native namespace cleanup owner's responsibility; cancellation +never starts detached cleanup. + +Ownership assumes Omes' fresh random IDs are not reused by another writer. +Memo, action identity/payload, overlap and EndAt are checked; canonicalized cron +specs are not compared byte-for-byte. The SDK has no conditional Delete fence +against concurrent replacement. Default payload encoding is used for validation; +incompatible encodings fail closed. This is not a general multi-writer schedule +manager or an exactly-once protocol. + +Info-level operation summaries distinguish acknowledgments, observed effects, +compatible NotFound and unresolved errors, with schedule/mutation identity, +logical-call count, error codes and elapsed time. No per-ID metrics are added. Fault settings, iteration +acceptance, unavailable-progress handling and coverage remain strict. diff --git a/scenarios/scheduler_recovery.go b/scenarios/scheduler_recovery.go new file mode 100644 index 000000000..2921e039e --- /dev/null +++ b/scenarios/scheduler_recovery.go @@ -0,0 +1,316 @@ +package scenarios + +import ( + "bytes" + "context" + "errors" + "time" + + "github.com/google/uuid" + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/converter" + "go.temporal.io/sdk/temporal" + "go.uber.org/zap" + "google.golang.org/grpc/codes" +) + +const ( + scheduleOwnerMemo = "omes-schedule-owner" + scheduleOperationCalls = 4 + scheduleOperationTimeout = 5 * time.Second + scheduleOperationBackoff = 100 * time.Millisecond +) + +var ( + errScheduleRecoveryBudget = errors.New("schedule operation recovery budget exhausted") + errScheduleIntentMismatch = errors.New("schedule ownership or intent mismatch") + errScheduleUpdateUnknown = errors.New("schedule update outcome unknown") +) + +type scheduleRecovery struct { + api client.ScheduleClient + options client.ScheduleOptions + owner string + logger *zap.SugaredLogger +} + +type scheduleOperation struct { + ctx context.Context + calls int + lastErr error + firstErr error + started time.Time + outcome string + name string + logger *zap.SugaredLogger +} + +func (r *scheduleRecovery) operation(ctx context.Context, name string) (*scheduleOperation, context.CancelFunc) { + ctx, cancel := context.WithTimeout(ctx, scheduleOperationTimeout) + // Deletion may clean up after EndAt, but cannot outlive the caller. + if name != "delete" { + var endCancel context.CancelFunc + ctx, endCancel = context.WithDeadline(ctx, r.options.Spec.EndAt) + parentCancel := cancel + cancel = func() { endCancel(); parentCancel() } + } + return &scheduleOperation{ctx: ctx, name: name, logger: r.logger.With("schedule_id", r.options.ID), outcome: "unresolved", started: time.Now()}, cancel +} + +func (b *scheduleOperation) call(fn func(context.Context) error) error { + if err := b.ctx.Err(); err != nil { + return errors.Join(b.lastErr, err) + } + if b.calls == scheduleOperationCalls { + return errors.Join(b.lastErr, errScheduleRecoveryBudget) + } + if b.calls > 0 { + timer := time.NewTimer(scheduleOperationBackoff) + defer timer.Stop() + select { + case <-b.ctx.Done(): + return errors.Join(b.lastErr, b.ctx.Err()) + case <-timer.C: + } + } + if err := b.ctx.Err(); err != nil { + return errors.Join(b.lastErr, err) + } + b.calls++ + err := fn(b.ctx) + if contextErr := b.ctx.Err(); contextErr != nil { + err = errors.Join(err, contextErr) + } + if err != nil { + if b.firstErr == nil { + b.firstErr = err + } + b.lastErr = err + } + return err +} + +func (b *scheduleOperation) finish(err error) { + if b.calls > 1 || err != nil || b.outcome != "acknowledged" { + b.logger.Infow("Schedule operation recovery", "operation", b.name, "logical_calls", b.calls, + "outcome", b.outcome, "error_code", serviceerror.ToStatus(err).Code().String(), + "initial_error_code", serviceerror.ToStatus(b.firstErr).Code().String(), "elapsed", time.Since(b.started)) + } +} + +func transientScheduleError(err error) bool { + if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, errScheduleRecoveryBudget) { + return false + } + switch serviceerror.ToStatus(err).Code() { + case codes.DeadlineExceeded, codes.Unavailable, codes.ResourceExhausted, codes.Aborted: + return true + default: + return false + } +} + +func scheduleMissing(err error) bool { return serviceerror.ToStatus(err).Code() == codes.NotFound } + +func scheduleExists(err error) bool { + return errors.Is(err, temporal.ErrScheduleAlreadyRunning) || serviceerror.ToStatus(err).Code() == codes.AlreadyExists +} + +func (r *scheduleRecovery) lookup(b *scheduleOperation) (*client.ScheduleDescription, error) { + for { + var d *client.ScheduleDescription + err := b.call(func(ctx context.Context) (err error) { + d, err = r.api.GetHandle(ctx, r.options.ID).Describe(ctx) + return err + }) + if err == nil { + err = r.check(d) + } + if !transientScheduleError(err) { + return d, err + } + } +} + +func (r *scheduleRecovery) check(d *client.ScheduleDescription) error { + if d == nil || d.Memo == nil || d.Schedule.Spec == nil { + return errScheduleIntentMismatch + } + var owner string + if err := converter.GetDefaultDataConverter().FromPayload(d.Memo.Fields[scheduleOwnerMemo], &owner); err != nil || owner != r.owner { + return errScheduleIntentMismatch + } + action, ok := d.Schedule.Action.(*client.ScheduleWorkflowAction) + expected := r.options.Action.(*client.ScheduleWorkflowAction) + if !ok || action.ID != expected.ID || action.Workflow != expected.Workflow || action.TaskQueue != expected.TaskQueue || !d.Schedule.Spec.EndAt.Equal(r.options.Spec.EndAt) { + return errScheduleIntentMismatch + } + if !schedulePayloadMatches(action.Args, expected.Args) { + return errScheduleIntentMismatch + } + if d.Schedule.Policy == nil || d.Schedule.Policy.Overlap != r.options.Overlap { + return errScheduleIntentMismatch + } + return nil +} + +func schedulePayloadMatches(actual, expected []any) bool { + if len(actual) != len(expected) { + return false + } + for i, value := range actual { + var data []byte + switch v := value.(type) { + case []byte: + data = v + case *commonpb.Payload: + if converter.GetDefaultDataConverter().FromPayload(v, &data) != nil { + return false + } + default: + return false + } + want, ok := expected[i].([]byte) + if !ok || !bytes.Equal(data, want) { + return false + } + } + return true +} + +func (r *scheduleRecovery) create(ctx context.Context) (err error) { + b, cancel := r.operation(ctx, "create") + defer cancel() + defer func() { b.finish(err) }() + for { + err = b.call(func(ctx context.Context) error { _, err := r.api.Create(ctx, r.options); return err }) + if err == nil { + b.outcome = "acknowledged" + return nil + } + if !transientScheduleError(err) && !scheduleExists(err) { + return err + } + writeErr := err + _, err = r.lookup(b) + if err == nil { + b.outcome = "observed_owned_create" + return nil + } + if !scheduleMissing(err) || scheduleExists(writeErr) { + return errors.Join(writeErr, err) + } + if b.calls == scheduleOperationCalls { + return errors.Join(writeErr, errScheduleRecoveryBudget, err) + } + // A late first create and its retry contend for the same unique ID and frozen intent. + } +} + +func (r *scheduleRecovery) describe(ctx context.Context) (err error) { + b, cancel := r.operation(ctx, "describe") + defer cancel() + defer func() { b.finish(err) }() + _, err = r.lookup(b) + if scheduleMissing(err) { + b.outcome = "not_found_compatible" + return nil + } + if err == nil { + b.outcome = "acknowledged" + } + return err +} + +func (r *scheduleRecovery) update(ctx context.Context) (err error) { + b, cancel := r.operation(ctx, "update") + defer cancel() + defer func() { b.finish(err) }() + witness := "omes-update-" + uuid.NewString() + b.logger = b.logger.With("mutation_id", witness) + for { + prepared := false + err = b.call(func(ctx context.Context) error { + return r.api.GetHandle(ctx, r.options.ID).Update(ctx, client.ScheduleUpdateOptions{DoUpdate: func(in client.ScheduleUpdateInput) (*client.ScheduleUpdate, error) { + if err := r.check(&in.Description); err != nil { + return nil, err + } + schedule := in.Description.Schedule + if schedule.State == nil { + return nil, errScheduleIntentMismatch + } + state := *schedule.State + // A fresh witness distinguishes this update's effect from an identical old payload. + state.Note = witness + schedule.State = &state + action := *schedule.Action.(*client.ScheduleWorkflowAction) + expected := r.options.Action.(*client.ScheduleWorkflowAction) + action.Args = expected.Args + schedule.Action = &action + prepared = true + return &client.ScheduleUpdate{Schedule: &schedule}, nil + }}) + }) + if err == nil { + b.outcome = "acknowledged" + return nil + } + if scheduleMissing(err) { + b.outcome = "not_found_compatible" + return nil + } + if !transientScheduleError(err) { + return err + } + // Before DoUpdate, the SDK has only described the schedule. + if !prepared { + continue + } + writeErr := err + for { + var d *client.ScheduleDescription + d, err = r.lookup(b) + if err != nil { + return errors.Join(writeErr, errScheduleUpdateUnknown, err) + } + if d.Schedule.State != nil && d.Schedule.State.Note == witness { + b.outcome = "observed_update_witness" + return nil + } + // The write may still be in flight. Probe within this budget; never replay it. + if b.calls == scheduleOperationCalls { + return errors.Join(writeErr, errScheduleUpdateUnknown, errScheduleRecoveryBudget) + } + } + } +} + +func (r *scheduleRecovery) delete(ctx context.Context) (err error) { + b, cancel := r.operation(ctx, "delete") + defer cancel() + defer func() { b.finish(err) }() + for { + _, err = r.lookup(b) + if scheduleMissing(err) { + b.outcome = "observed_absence" + return nil + } + if err != nil { + return err + } + err = b.call(func(ctx context.Context) error { return r.api.GetHandle(ctx, r.options.ID).Delete(ctx) }) + if err == nil { + b.outcome = "acknowledged" + return nil + } + if scheduleMissing(err) { + b.outcome = "not_found_compatible" + return nil + } + if !transientScheduleError(err) { + return err + } + } +} diff --git a/scenarios/scheduler_recovery_budget_test.go b/scenarios/scheduler_recovery_budget_test.go new file mode 100644 index 000000000..b682a5e8d --- /dev/null +++ b/scenarios/scheduler_recovery_budget_test.go @@ -0,0 +1,101 @@ +package scenarios + +import ( + "context" + "errors" + "fmt" + "testing" + "testing/synctest" + "time" + + "github.com/stretchr/testify/require" + "go.temporal.io/api/serviceerror" + "go.temporal.io/sdk/client" + "go.uber.org/zap" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestScheduleRecoveryTransientCategory(t *testing.T) { + for _, tc := range []struct { + name string + err error + retry bool + }{ + {"typed_deadline", serviceerror.NewDeadlineExceeded("persistence timeout"), true}, + {"wrapped_typed_deadline", fmt.Errorf("read: %w", serviceerror.NewDeadlineExceeded("timeout")), true}, + {"unavailable", serviceerror.NewUnavailable("unavailable"), true}, + {"resource_exhausted", serviceerror.NewResourceExhausted(0, "overload"), true}, + {"raw_grpc_aborted", status.Error(codes.Aborted, "conflict"), true}, + {"raw_context_deadline", context.DeadlineExceeded, false}, + {"canceled", context.Canceled, false}, + {"permission", serviceerror.NewPermissionDenied("denied", ""), false}, + {"invalid", serviceerror.NewInvalidArgument("invalid"), false}, + {"not_found", serviceerror.NewNotFound("absent"), false}, + {"internal", serviceerror.NewInternal("invariant"), false}, + {"budget", errors.Join(serviceerror.NewDeadlineExceeded("timeout"), errScheduleRecoveryBudget), false}, + } { + t.Run(tc.name, func(t *testing.T) { require.Equal(t, tc.retry, transientScheduleError(tc.err)) }) + } +} + +func TestScheduleRecoveryBudget(t *testing.T) { + t.Run("logical_calls_and_backoff", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + r := &scheduleRecovery{options: client.ScheduleOptions{Spec: client.ScheduleSpec{EndAt: time.Now().Add(time.Minute)}}, logger: zap.NewNop().Sugar()} + b, cancel := r.operation(t.Context(), "describe") + defer cancel() + start := time.Now() + calls := 0 + for range 5 { + err := b.call(func(context.Context) error { calls++; return serviceerror.NewDeadlineExceeded("timeout") }) + require.Error(t, err) + } + require.Equal(t, 4, calls) + require.Equal(t, 4, b.calls) + require.Equal(t, 300*time.Millisecond, time.Since(start)) + require.ErrorIs(t, b.call(func(context.Context) error { t.Fatal("extra call"); return nil }), errScheduleRecoveryBudget) + }) + }) + t.Run("one_wallclock_budget_including_calls", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + r := &scheduleRecovery{options: client.ScheduleOptions{Spec: client.ScheduleSpec{EndAt: time.Now().Add(time.Minute)}}, logger: zap.NewNop().Sugar()} + b, cancel := r.operation(t.Context(), "describe") + defer cancel() + start := time.Now() + require.Error(t, b.call(func(context.Context) error { + time.Sleep(4 * time.Second) + return serviceerror.NewDeadlineExceeded("timeout") + })) + require.ErrorIs(t, b.call(func(ctx context.Context) error { <-ctx.Done(); return ctx.Err() }), context.DeadlineExceeded) + require.Equal(t, 5*time.Second, time.Since(start)) + require.Equal(t, 2, b.calls) + }) + }) + t.Run("cancellation_during_backoff", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + r := &scheduleRecovery{options: client.ScheduleOptions{Spec: client.ScheduleSpec{EndAt: time.Now().Add(time.Minute)}}, logger: zap.NewNop().Sugar()} + b, stop := r.operation(ctx, "describe") + defer stop() + require.Error(t, b.call(func(context.Context) error { return serviceerror.NewDeadlineExceeded("timeout") })) + go func() { time.Sleep(50 * time.Millisecond); cancel() }() + require.ErrorIs(t, b.call(func(context.Context) error { t.Fatal("call after cancellation"); return nil }), context.Canceled) + require.Equal(t, 1, b.calls) + }) + }) + t.Run("expired_endat_not_success_or_renewal", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + r := &scheduleRecovery{options: client.ScheduleOptions{Spec: client.ScheduleSpec{EndAt: time.Now().Add(-time.Second)}}, logger: zap.NewNop().Sugar()} + for _, op := range []string{"create", "describe", "update"} { + b, cancel := r.operation(t.Context(), op) + require.ErrorIs(t, b.call(func(context.Context) error { t.Fatal("expired action window"); return nil }), context.DeadlineExceeded) + cancel() + } + b, cancel := r.operation(t.Context(), "delete") + defer cancel() + require.NoError(t, b.call(func(context.Context) error { return nil })) + }) + }) +} diff --git a/scenarios/scheduler_recovery_rpc_test.go b/scenarios/scheduler_recovery_rpc_test.go new file mode 100644 index 000000000..c4395a017 --- /dev/null +++ b/scenarios/scheduler_recovery_rpc_test.go @@ -0,0 +1,439 @@ +package scenarios + +import ( + "context" + "errors" + "net" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/temporalio/omes/loadgen" + commonpb "go.temporal.io/api/common/v1" + schedulepb "go.temporal.io/api/schedule/v1" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" +) + +type scheduleFault struct { + code codes.Code + apply bool + delayedEffect time.Duration + foreign bool + cancel context.CancelFunc + block bool + vanish bool +} + +type scheduleRecord struct { + schedule *schedulepb.Schedule + memo *commonpb.Memo +} + +type scheduleRecoveryService struct { + workflowservice.UnimplementedWorkflowServiceServer + mu sync.Mutex + effects sync.WaitGroup + records map[string]scheduleRecord + faults map[string]map[int]scheduleFault + calls map[string]int + creates []*workflowservice.CreateScheduleRequest + updates []*workflowservice.UpdateScheduleRequest + deletes []*workflowservice.DeleteScheduleRequest + mutateDescribe func(*schedulepb.Schedule) + mutationAt int +} + +func (s *scheduleRecoveryService) fault(op string) scheduleFault { + s.calls[op]++ + return s.faults[op][s.calls[op]] +} +func (s *scheduleRecoveryService) GetSystemInfo(context.Context, *workflowservice.GetSystemInfoRequest) (*workflowservice.GetSystemInfoResponse, error) { + return &workflowservice.GetSystemInfoResponse{}, nil +} +func (s *scheduleRecoveryService) CreateSchedule(ctx context.Context, r *workflowservice.CreateScheduleRequest) (*workflowservice.CreateScheduleResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.creates = append(s.creates, proto.Clone(r).(*workflowservice.CreateScheduleRequest)) + f := s.fault("create") + if f.block { + <-ctx.Done() + return nil, status.Error(codes.DeadlineExceeded, "slow create") + } + if f.cancel != nil { + f.cancel() + return nil, status.Error(codes.Canceled, "parent canceled") + } + if f.code == codes.OK || f.apply { + if _, exists := s.records[r.ScheduleId]; exists { + return nil, status.Error(codes.AlreadyExists, "exists") + } + var memo *commonpb.Memo + if r.Memo != nil { + memo = proto.Clone(r.Memo).(*commonpb.Memo) + } + if f.foreign { + memo = &commonpb.Memo{} + } + s.records[r.ScheduleId] = scheduleRecord{proto.Clone(r.Schedule).(*schedulepb.Schedule), memo} + } + if f.code != codes.OK { + return nil, status.Error(f.code, "scripted create response loss or rejection") + } + return &workflowservice.CreateScheduleResponse{}, nil +} +func (s *scheduleRecoveryService) DescribeSchedule(_ context.Context, r *workflowservice.DescribeScheduleRequest) (*workflowservice.DescribeScheduleResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + f := s.fault("describe") + if s.mutateDescribe != nil && s.calls["describe"] == s.mutationAt { + record := s.records[r.ScheduleId] + s.mutateDescribe(record.schedule) + s.records[r.ScheduleId] = record + } + if f.code == codes.NotFound && f.apply { + delete(s.records, r.ScheduleId) + } + if f.code != codes.OK { + return nil, status.Error(f.code, "scripted describe rejection") + } + record, ok := s.records[r.ScheduleId] + if !ok { + return nil, status.Error(codes.NotFound, "absent") + } + var memo *commonpb.Memo + if record.memo != nil { + memo = proto.Clone(record.memo).(*commonpb.Memo) + } + return &workflowservice.DescribeScheduleResponse{Schedule: proto.Clone(record.schedule).(*schedulepb.Schedule), Memo: memo, Info: &schedulepb.ScheduleInfo{}}, nil +} +func (s *scheduleRecoveryService) UpdateSchedule(_ context.Context, r *workflowservice.UpdateScheduleRequest) (*workflowservice.UpdateScheduleResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.updates = append(s.updates, proto.Clone(r).(*workflowservice.UpdateScheduleRequest)) + f := s.fault("update") + record, ok := s.records[r.ScheduleId] + if !ok { + return nil, status.Error(codes.NotFound, "absent") + } + if f.code == codes.OK || f.apply { + updated := proto.Clone(r.Schedule).(*schedulepb.Schedule) + if f.delayedEffect == 0 { + record.schedule = updated + s.records[r.ScheduleId] = record + } else { + s.effects.Add(1) + go func() { + defer s.effects.Done() + time.Sleep(f.delayedEffect) + s.mu.Lock() + defer s.mu.Unlock() + if current, ok := s.records[r.ScheduleId]; ok { + current.schedule = updated + s.records[r.ScheduleId] = current + } + }() + } + } + if f.vanish { + delete(s.records, r.ScheduleId) + } + if f.cancel != nil { + f.cancel() + return nil, status.Error(codes.Canceled, "parent canceled after update") + } + if f.code != codes.OK { + return nil, status.Error(f.code, "scripted update response loss or rejection") + } + return &workflowservice.UpdateScheduleResponse{}, nil +} +func (s *scheduleRecoveryService) DeleteSchedule(_ context.Context, r *workflowservice.DeleteScheduleRequest) (*workflowservice.DeleteScheduleResponse, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.deletes = append(s.deletes, proto.Clone(r).(*workflowservice.DeleteScheduleRequest)) + f := s.fault("delete") + if _, ok := s.records[r.ScheduleId]; !ok { + return nil, status.Error(codes.NotFound, "absent") + } + if f.code == codes.OK || f.apply { + delete(s.records, r.ScheduleId) + } + if f.code != codes.OK { + return nil, status.Error(f.code, "scripted delete response loss or rejection") + } + return &workflowservice.DeleteScheduleResponse{}, nil +} + +func scheduleRecoveryRPCFixture(t *testing.T) (client.Client, *scheduleRecoveryService, *observer.ObservedLogs, *zap.SugaredLogger) { + t.Helper() + s := &scheduleRecoveryService{records: map[string]scheduleRecord{}, faults: map[string]map[int]scheduleFault{}, calls: map[string]int{}} + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + server := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(server, s) + done := make(chan error, 1) + go func() { done <- server.Serve(listener) }() + t.Cleanup(func() { server.Stop(); require.NoError(t, <-done); s.effects.Wait() }) + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + c, err := client.DialContext(ctx, client.Options{HostPort: listener.Addr().String(), Namespace: "scheduler-recovery"}) + require.NoError(t, err) + t.Cleanup(c.Close) + core, logs := observer.New(zap.InfoLevel) + return c, s, logs, zap.New(core).Sugar() +} + +func scheduleRecoveryInfo(c client.Client, logger *zap.SugaredLogger, enabled bool) loadgen.ScenarioInfo { + options := loadgen.MustResolveScenarioOptions("scheduler_stress", map[string]string{ + "schedule-creation-per-iteration": "1", "schedule-reads-per-creation": "1", "schedule-updates-per-creation": "1", + "scheduler-duration-per-iteration": "1s", "wait-time-before-cleanup": "0s", "operation-interval": "1ms", + }) + // Exercise faults independently of option registration. + if options.Lookup("recover-schedule-operations") == nil { + options.Bool("recover-schedule-operations", false, "") + } + if enabled { + if err := options.Set("recover-schedule-operations", "true"); err != nil { + panic(err) + } + } + return loadgen.ScenarioInfo{RunID: "crud-recovery", Client: c, Logger: logger, Options: options} +} + +func TestSchedulerOperationRecoverySDK(t *testing.T) { + cases := []struct { + name string + faults map[string]map[int]scheduleFault + wantErr string + creates, updates, deletes, remaining int + outcome string + sdkRetry bool + }{ + {name: "healthy", creates: 1, updates: 1, deletes: 1}, + {name: "read_permanent_rejection", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.PermissionDenied}}}, wantErr: "describe rejection", creates: 1, updates: 1, deletes: 1}, + {name: "create_last_probe_not_found_no_fifth_call", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}}}, wantErr: "budget exhausted", creates: 2}, + {name: "sdk_unavailable_retries_same_rpc_request", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.Unavailable}}}, creates: 2, updates: 1, deletes: 1, sdkRetry: true}, + {name: "sdk_accepted_create_then_exists", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.Unavailable, apply: true}}}, creates: 2, updates: 1, deletes: 1, sdkRetry: true, outcome: "observed_owned_create"}, + {name: "describe_not_found_compatible", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.NotFound, apply: true}}}, creates: 1, outcome: "not_found_compatible"}, + {name: "update_describe_not_found_compatible", faults: map[string]map[int]scheduleFault{"describe": {2: {code: codes.NotFound, apply: true}}}, creates: 1, outcome: "not_found_compatible"}, + {name: "delete_not_found_compatible", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.NotFound, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "not_found_compatible"}, + {name: "create_accepted_response_lost", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_owned_create"}, + {name: "create_not_accepted_same_frozen_intent", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded}}}, creates: 2, updates: 1, deletes: 1, outcome: "acknowledged"}, + {name: "read_deadline", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 1, outcome: "acknowledged"}, + {name: "update_internal_describe_deadline", faults: map[string]map[int]scheduleFault{"describe": {2: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 1, outcome: "acknowledged"}, + {name: "update_accepted_response_lost", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_update_witness"}, + {name: "update_late_effect_no_replay", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, delayedEffect: 150 * time.Millisecond}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_update_witness"}, + {name: "update_identical_old_payload_not_success", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded}}}, wantErr: "outcome unknown", creates: 1, updates: 1, deletes: 1}, + {name: "update_response_lost_then_absent_is_unknown", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, vanish: true}}}, wantErr: "outcome unknown", creates: 1, updates: 1}, + {name: "update_permanent_rejection", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.PermissionDenied}}}, wantErr: "update response loss or rejection", creates: 1, updates: 1, deletes: 1}, + {name: "delete_accepted_response_lost", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_absence"}, + {name: "delete_not_accepted_retry_after_owned_probe", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 2, outcome: "acknowledged"}, + {name: "delete_last_response_lost_no_fifth_probe", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded, apply: true}}}, wantErr: "budget exhausted", creates: 1, updates: 1, deletes: 2}, + {name: "foreign_owner_never_adopted_or_deleted", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.AlreadyExists, apply: true, foreign: true}}}, wantErr: "ownership or intent mismatch", creates: 1, remaining: 1}, + {name: "create_probe_exhausted_namespace_cleanup_owned", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}, "describe": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}, 3: {code: codes.DeadlineExceeded}}}, wantErr: "budget exhausted", creates: 1, remaining: 1}, + {name: "permanent_create_rejection", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.InvalidArgument}}}, wantErr: "create response loss or rejection", creates: 1}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + c, s, logs, logger := scheduleRecoveryRPCFixture(t) + if tc.faults != nil { + s.faults = tc.faults + } + info := scheduleRecoveryInfo(c, logger, true) + err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) + if tc.wantErr == "" { + require.NoError(t, err) + } else { + require.ErrorContains(t, err, tc.wantErr) + } + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.creates, tc.creates) + require.Len(t, s.updates, tc.updates) + require.Len(t, s.deletes, tc.deletes) + require.Len(t, s.records, tc.remaining) + if tc.creates == 2 { + require.Equal(t, s.creates[0].ScheduleId, s.creates[1].ScheduleId) + require.True(t, proto.Equal(s.creates[0].Schedule, s.creates[1].Schedule)) + require.True(t, proto.Equal(s.creates[0].Memo, s.creates[1].Memo)) + if tc.sdkRetry { + require.Equal(t, s.creates[0].RequestId, s.creates[1].RequestId) + } else { + require.NotEqual(t, s.creates[0].RequestId, s.creates[1].RequestId) + } + } + for _, u := range s.updates { + require.Equal(t, s.creates[0].ScheduleId, u.ScheduleId) + require.True(t, proto.Equal(s.creates[0].Schedule.Spec.EndTime, u.Schedule.Spec.EndTime)) + before, after := s.creates[0].Schedule.Action.GetStartWorkflow(), u.Schedule.Action.GetStartWorkflow() + require.Equal(t, before.WorkflowId, after.WorkflowId) + require.True(t, proto.Equal(before.WorkflowType, after.WorkflowType)) + require.Equal(t, before.TaskQueue.Name, after.TaskQueue.Name) + require.True(t, proto.Equal(before.Input, after.Input)) + require.NotEmpty(t, u.Schedule.State.Notes) + } + if tc.outcome != "" { + found := false + for _, e := range logs.All() { + require.NotEmpty(t, e.ContextMap()["schedule_id"]) + if e.ContextMap()["operation"] == "update" && len(s.updates) > 0 { + require.Equal(t, s.updates[0].Schedule.State.Notes, e.ContextMap()["mutation_id"]) + } + if e.ContextMap()["outcome"] == tc.outcome { + found = true + } + } + require.True(t, found, "recovery outcome must be visible at Info") + } + }) + } +} + +func TestSchedulerOperationRecoveryDefaultUnchanged(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + s.faults = map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}} + info := scheduleRecoveryInfo(c, logger, false) + err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) + require.ErrorContains(t, err, "scripted create") + require.Len(t, s.creates, 1) + require.Nil(t, s.creates[0].Memo) + require.Zero(t, s.calls["describe"]) + require.Len(t, s.records, 1) +} + +func TestSchedulerOperationRecoveryParentCancellation(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + s.faults = map[string]map[int]scheduleFault{"create": {1: {cancel: cancel}}} + start := time.Now() + info := scheduleRecoveryInfo(c, logger, true) + err := new(SchedulerExecutor).Execute(ctx, info.NewRun(1)) + require.ErrorIs(t, err, context.Canceled) + require.Less(t, time.Since(start), time.Second) + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.creates, 1) + require.Empty(t, s.updates) + require.Empty(t, s.deletes) +} + +func TestSchedulerOperationRecoveryIntentMismatch(t *testing.T) { + for _, field := range []string{"action_id", "end_at", "payload", "overlap"} { + t.Run(field, func(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + s.mutationAt = 1 + s.mutateDescribe = func(schedule *schedulepb.Schedule) { + switch field { + case "action_id": + schedule.Action.GetStartWorkflow().WorkflowId = "another-owner" + case "end_at": + schedule.Spec.EndTime.Seconds++ + case "payload": + schedule.Action.GetStartWorkflow().Input.Payloads[0].Data = []byte("changed") + case "overlap": + schedule.Policies.OverlapPolicy = 0 + } + } + info := scheduleRecoveryInfo(c, logger, true) + err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) + require.ErrorContains(t, err, "ownership or intent mismatch") + require.Empty(t, s.updates) + require.Empty(t, s.deletes) + require.Len(t, s.records, 1) + }) + } +} + +func TestSchedulerOperationRecoveryCancellationAtTick(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + info := scheduleRecoveryInfo(c, logger, true) + require.NoError(t, info.Options.Set("operation-interval", "1h")) + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + start := time.Now() + err := new(SchedulerExecutor).Execute(ctx, info.NewRun(1)) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Less(t, time.Since(start), time.Second) + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.creates, 1) + require.Empty(t, s.deletes) + require.Len(t, s.records, 1) +} + +func TestSchedulerOperationRecoveryPreparedUpdateCancellation(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + s.faults = map[string]map[int]scheduleFault{"update": {1: {apply: true, cancel: cancel}}} + info := scheduleRecoveryInfo(c, logger, true) + err := new(SchedulerExecutor).Execute(ctx, info.NewRun(1)) + require.ErrorIs(t, err, context.Canceled) + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.updates, 1) + require.Empty(t, s.deletes) + require.Len(t, s.records, 1) +} + +func TestSchedulerOperationRecoverySlowRPCBudget(t *testing.T) { + c, s, logs, logger := scheduleRecoveryRPCFixture(t) + s.faults = map[string]map[int]scheduleFault{"create": {1: {block: true}, 2: {block: true}}} + info := scheduleRecoveryInfo(c, logger, true) + require.NoError(t, info.Options.Set("scheduler-duration-per-iteration", "30s")) + start := time.Now() + err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) + require.True(t, errors.Is(err, errScheduleRecoveryBudget) || errors.Is(err, context.DeadlineExceeded), "unexpected recovery result: %v", err) + require.Less(t, time.Since(start), 2*scheduleOperationTimeout) + entries := logs.FilterMessage("Schedule operation recovery").All() + require.Len(t, entries, 1) + require.LessOrEqual(t, entries[0].ContextMap()["logical_calls"].(int64), int64(scheduleOperationCalls)) + s.mu.Lock() + defer s.mu.Unlock() + require.NotEmpty(t, s.creates) + require.Empty(t, s.updates) + require.Empty(t, s.deletes) +} + +func TestSchedulerOperationRecoveryFreshUpdateWitness(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + s.faults = map[string]map[int]scheduleFault{"update": {2: {code: codes.DeadlineExceeded}}} + info := scheduleRecoveryInfo(c, logger, true) + require.NoError(t, info.Options.Set("schedule-updates-per-creation", "2")) + err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) + require.ErrorContains(t, err, "outcome unknown") + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.updates, 2) + require.NotEqual(t, s.updates[0].Schedule.State.Notes, s.updates[1].Schedule.State.Notes) + require.Len(t, s.deletes, 1) + require.Empty(t, s.records) +} + +func TestSchedulerOperationRecoveryParentDeadlineDuringRPC(t *testing.T) { + c, s, _, logger := scheduleRecoveryRPCFixture(t) + s.faults = map[string]map[int]scheduleFault{"create": {1: {block: true}}} + info := scheduleRecoveryInfo(c, logger, true) + require.NoError(t, info.Options.Set("scheduler-duration-per-iteration", "30s")) + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + start := time.Now() + err := new(SchedulerExecutor).Execute(ctx, info.NewRun(1)) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded) + require.Less(t, time.Since(start), time.Second) + s.mu.Lock() + defer s.mu.Unlock() + require.Len(t, s.creates, 1) + require.Empty(t, s.updates) + require.Empty(t, s.deletes) + require.Zero(t, s.calls["describe"]) +} diff --git a/scenarios/scheduler_recovery_server_test.go b/scenarios/scheduler_recovery_server_test.go new file mode 100644 index 000000000..23e33e192 --- /dev/null +++ b/scenarios/scheduler_recovery_server_test.go @@ -0,0 +1,132 @@ +package scenarios + +import ( + "context" + "net" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/temporalio/omes/internal/workertest" + commonpb "go.temporal.io/api/common/v1" + "go.temporal.io/api/serviceerror" + "go.temporal.io/api/workflowservice/v1" + "go.temporal.io/sdk/client" + "go.uber.org/zap" + "go.uber.org/zap/zaptest/observer" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" +) + +type scheduleResponseLossProxy struct { + workflowservice.UnimplementedWorkflowServiceServer + backend workflowservice.WorkflowServiceClient + mu sync.Mutex + calls map[string]int + backendType string +} + +func scheduleProxyContext(ctx context.Context) context.Context { + md, _ := metadata.FromIncomingContext(ctx) + return metadata.NewOutgoingContext(ctx, metadata.MD{"temporal-experiment": md.Get("temporal-experiment")}) +} +func (p *scheduleResponseLossProxy) lose(op string) bool { + p.mu.Lock() + defer p.mu.Unlock() + p.calls[op]++ + return p.calls[op] == 1 +} +func (p *scheduleResponseLossProxy) GetSystemInfo(ctx context.Context, r *workflowservice.GetSystemInfoRequest) (*workflowservice.GetSystemInfoResponse, error) { + v, err := p.backend.GetSystemInfo(scheduleProxyContext(ctx), r) + return v, serviceerror.ToStatus(err).Err() +} +func (p *scheduleResponseLossProxy) CreateSchedule(ctx context.Context, r *workflowservice.CreateScheduleRequest) (*workflowservice.CreateScheduleResponse, error) { + v, err := p.backend.CreateSchedule(scheduleProxyContext(ctx), r) + if err != nil { + return nil, serviceerror.ToStatus(err).Err() + } + d, err := p.backend.DescribeWorkflowExecution(scheduleProxyContext(ctx), &workflowservice.DescribeWorkflowExecutionRequest{Namespace: r.Namespace, Execution: &commonpb.WorkflowExecution{WorkflowId: "temporal-sys-scheduler:" + r.ScheduleId}}) + if err == nil { + p.mu.Lock() + p.backendType = d.WorkflowExecutionInfo.Type.Name + p.mu.Unlock() + } + if p.lose("create") { + return nil, status.Error(codes.DeadlineExceeded, "response lost after real server create") + } + return v, nil +} +func (p *scheduleResponseLossProxy) DescribeSchedule(ctx context.Context, r *workflowservice.DescribeScheduleRequest) (*workflowservice.DescribeScheduleResponse, error) { + v, err := p.backend.DescribeSchedule(scheduleProxyContext(ctx), r) + return v, serviceerror.ToStatus(err).Err() +} +func (p *scheduleResponseLossProxy) UpdateSchedule(ctx context.Context, r *workflowservice.UpdateScheduleRequest) (*workflowservice.UpdateScheduleResponse, error) { + v, err := p.backend.UpdateSchedule(scheduleProxyContext(ctx), r) + if err != nil { + return nil, serviceerror.ToStatus(err).Err() + } + if p.lose("update") { + return nil, status.Error(codes.DeadlineExceeded, "response lost after real server update") + } + return v, nil +} +func (p *scheduleResponseLossProxy) DeleteSchedule(ctx context.Context, r *workflowservice.DeleteScheduleRequest) (*workflowservice.DeleteScheduleResponse, error) { + v, err := p.backend.DeleteSchedule(scheduleProxyContext(ctx), r) + if err != nil { + return nil, serviceerror.ToStatus(err).Err() + } + if p.lose("delete") { + return nil, status.Error(codes.DeadlineExceeded, "response lost after real server delete") + } + return v, nil +} + +func TestSchedulerOperationRecoveryLocalServer(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 2*time.Minute) + defer cancel() + server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{"frontend.allowedExperiments": []string{"*"}})) + backend, err := client.DialContext(ctx, client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) + require.NoError(t, err) + t.Cleanup(backend.Close) + for _, experiment := range []string{"false", "true"} { + t.Run("chasm_requested_"+experiment, func(t *testing.T) { + p := &scheduleResponseLossProxy{backend: backend.WorkflowService(), calls: map[string]int{}} + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + rpc := grpc.NewServer() + workflowservice.RegisterWorkflowServiceServer(rpc, p) + done := make(chan error, 1) + go func() { done <- rpc.Serve(listener) }() + t.Cleanup(func() { rpc.Stop(); require.NoError(t, <-done) }) + c, err := client.DialContext(ctx, client.Options{HostPort: listener.Addr().String(), Namespace: "default"}) + require.NoError(t, err) + t.Cleanup(c.Close) + core, logs := observer.New(zap.InfoLevel) + info := scheduleRecoveryInfo(c, zap.New(core).Sugar(), true) + require.NoError(t, info.Options.Set("enable-chasm-scheduler", experiment)) + require.NoError(t, info.Options.Set("scheduler-duration-per-iteration", "30s")) + require.NoError(t, new(SchedulerExecutor).Execute(ctx, info.NewRun(1))) + p.mu.Lock() + defer p.mu.Unlock() + require.Equal(t, map[string]int{"create": 1, "update": 1, "delete": 1}, p.calls) + t.Logf("actual scheduler backend workflow type %q", p.backendType) + if experiment == "true" { + require.Equal(t, "temporal-sys-dummy-workflow", p.backendType) + } else { + require.Equal(t, "temporal-sys-scheduler-workflow", p.backendType) + } + outcomes := map[string]bool{} + for _, row := range logs.All() { + if s, ok := row.ContextMap()["outcome"].(string); ok { + outcomes[s] = true + } + } + for _, outcome := range []string{"observed_owned_create", "observed_update_witness", "observed_absence"} { + require.True(t, outcomes[outcome], outcome) + } + }) + } +} diff --git a/scenarios/scheduler_stress.go b/scenarios/scheduler_stress.go index 81e2d1fdd..abb32f394 100644 --- a/scenarios/scheduler_stress.go +++ b/scenarios/scheduler_stress.go @@ -35,6 +35,7 @@ func init() { o.String(ScheduledWorkflowTypeFlag, DefaultScheduledWorkflowType, "Workflow type to schedule: "+NoopScheduledWorkflowType+" or "+SleepScheduleWorkflowType+".") o.Bool(EnableChasmSchedulerFlag, DefaultEnableChasmScheduler, "Route schedule creation to the CHASM scheduler.") + o.Bool(RecoverScheduleOperationsFlag, false, "Recover transient schedule CRUD operations with bounded calls and observed effects.") }, ExecutorFn: func() loadgen.Executor { return &loadgen.GenericExecutor{ @@ -64,6 +65,7 @@ type schedulerExecutorConfig struct { OverlapPolicy []enums.ScheduleOverlapPolicy ScheduledWorkflowType string EnableChasmScheduler bool + RecoverScheduleOperations bool } var _ loadgen.Configurable = (*SchedulerExecutor)(nil) @@ -90,6 +92,7 @@ const ( OverlapPolicyFlag = "overlap-policy" ScheduledWorkflowTypeFlag = "scheduled-workflow-type" EnableChasmSchedulerFlag = "enable-chasm-scheduler" + RecoverScheduleOperationsFlag = "recover-schedule-operations" ) const ( @@ -124,6 +127,7 @@ func (s *SchedulerExecutor) Configure(info loadgen.ScenarioInfo) error { OverlapPolicy: parseOverlapPolicy(info.OptionString(OverlapPolicyFlag)), ScheduledWorkflowType: info.OptionString(ScheduledWorkflowTypeFlag), EnableChasmScheduler: info.OptionBool(EnableChasmSchedulerFlag), + RecoverScheduleOperations: info.OptionBool(RecoverScheduleOperationsFlag), } if config.ScheduleCreationPerIteration <= 0 { @@ -181,6 +185,18 @@ func (s *SchedulerExecutor) Execute(ctx context.Context, run *loadgen.Run) error wg.Go(func() { ticker := time.NewTicker(s.config.OperationInterval) defer ticker.Stop() + wait := func() bool { + if !s.config.RecoverScheduleOperations { + <-ticker.C + return true + } + select { + case <-ticker.C: + return true + case <-ctx.Done(): + return false + } + } start := time.Now() scheduleID := fmt.Sprintf("sched-%s-%d-%d-%s", run.RunID, run.Iteration, i, uuid.New()) @@ -189,30 +205,53 @@ func (s *SchedulerExecutor) Execute(ctx context.Context, run *loadgen.Run) error record(fmt.Errorf("creating schedule %s: %w", scheduleID, err)) return } - <-ticker.C - // Read and update failures fall through to the delete below; an - // abandoned schedule would outlive the iteration until its EndAt. + if !wait() { + return + } + // Read and update failures still fall through to deletion. for range s.config.ScheduleReadsPerCreation { - if err := s.describeSchedule(ctx, client, sc.ScheduleID, logger); err != nil { + var err error + if sc.recovery != nil { + err = sc.recovery.describe(ctx) + } else { + err = s.describeSchedule(ctx, client, sc.ScheduleID, logger) + } + if err != nil { record(fmt.Errorf("describing schedule %s: %w", sc.ScheduleID, err)) } - <-ticker.C // Wait between read operations + if !wait() { + return + } } for range s.config.ScheduleUpdatesPerCreation { - if err := s.updateSchedule(ctx, client, sc.ScheduleID, logger); err != nil { + var err error + if sc.recovery != nil { + err = sc.recovery.update(ctx) + } else { + err = s.updateSchedule(ctx, client, sc.ScheduleID, logger) + } + if err != nil { record(fmt.Errorf("updating schedule %s: %w", sc.ScheduleID, err)) } - <-ticker.C // Wait between update operations + if !wait() { + return + } } dur := time.Until(start.Add(s.config.WaitTimeBeforeCleanup)) select { case <-time.After(dur): - if err := s.deleteSchedule(ctx, client, sc.ScheduleID, logger); err != nil { + var err error + if sc.recovery != nil { + err = sc.recovery.delete(ctx) + } else { + err = s.deleteSchedule(ctx, client, sc.ScheduleID, logger) + } + if err != nil { record(fmt.Errorf("deleting schedule %s: %w", sc.ScheduleID, err)) } case <-ctx.Done(): - // Left undeleted; EndAt bounds how long it survives. + // Cancellation can leave the schedule object behind after EndAt. } }) } @@ -230,6 +269,7 @@ func (s *SchedulerExecutor) Execute(ctx context.Context, run *loadgen.Run) error type ScheduleState struct { ScheduleID string DeleteAfter time.Duration + recovery *scheduleRecovery } func (s *SchedulerExecutor) createSchedule(ctx context.Context, c client.Client, scheduleID string, taskQueue string, logger *zap.SugaredLogger) (ScheduleState, error) { @@ -256,13 +296,18 @@ func (s *SchedulerExecutor) createSchedule(ctx context.Context, c client.Client, ID: scheduleID, Spec: client.ScheduleSpec{ CronExpressions: []string{s.config.CronExpression}, - // defining an end at ensures all schedules will be removed - // regardless of errors from this executor + // EndAt stops scheduled actions; it does not delete the schedule. EndAt: endTime, }, Action: action, Overlap: pickOverlap(s.config.OverlapPolicy, logger), } + if s.config.RecoverScheduleOperations { + owner := uuid.NewString() + options.Memo = map[string]any{scheduleOwnerMemo: owner} + sc.recovery = &scheduleRecovery{api: c.ScheduleClient(), options: options, owner: owner, logger: logger} + return sc, sc.recovery.create(ctx) + } _, err := c.ScheduleClient().Create(ctx, options) return sc, err } From 0054f4600a4d98f3589251203c16e2f6039b57c2 Mon Sep 17 00:00:00 2001 From: lilydoar Date: Mon, 26 Oct 2026 14:06:12 -0700 Subject: [PATCH 2/3] Make scheduler recovery tests deterministic and assert typed outcomes --- scenarios/scheduler_recovery_rpc_test.go | 90 ++++++++++++------------ 1 file changed, 45 insertions(+), 45 deletions(-) diff --git a/scenarios/scheduler_recovery_rpc_test.go b/scenarios/scheduler_recovery_rpc_test.go index c4395a017..6ad1e8dd7 100644 --- a/scenarios/scheduler_recovery_rpc_test.go +++ b/scenarios/scheduler_recovery_rpc_test.go @@ -12,6 +12,7 @@ import ( "github.com/temporalio/omes/loadgen" commonpb "go.temporal.io/api/common/v1" schedulepb "go.temporal.io/api/schedule/v1" + "go.temporal.io/api/serviceerror" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/client" "go.uber.org/zap" @@ -25,7 +26,7 @@ import ( type scheduleFault struct { code codes.Code apply bool - delayedEffect time.Duration + delayedEffect bool foreign bool cancel context.CancelFunc block bool @@ -33,14 +34,15 @@ type scheduleFault struct { } type scheduleRecord struct { - schedule *schedulepb.Schedule - memo *commonpb.Memo + schedule *schedulepb.Schedule + memo *commonpb.Memo + pending *schedulepb.Schedule + probesUntilApplied int } type scheduleRecoveryService struct { workflowservice.UnimplementedWorkflowServiceServer mu sync.Mutex - effects sync.WaitGroup records map[string]scheduleRecord faults map[string]map[int]scheduleFault calls map[string]int @@ -82,7 +84,7 @@ func (s *scheduleRecoveryService) CreateSchedule(ctx context.Context, r *workflo if f.foreign { memo = &commonpb.Memo{} } - s.records[r.ScheduleId] = scheduleRecord{proto.Clone(r.Schedule).(*schedulepb.Schedule), memo} + s.records[r.ScheduleId] = scheduleRecord{schedule: proto.Clone(r.Schedule).(*schedulepb.Schedule), memo: memo} } if f.code != codes.OK { return nil, status.Error(f.code, "scripted create response loss or rejection") @@ -108,6 +110,15 @@ func (s *scheduleRecoveryService) DescribeSchedule(_ context.Context, r *workflo if !ok { return nil, status.Error(codes.NotFound, "absent") } + if record.pending != nil { + if record.probesUntilApplied == 0 { + record.schedule = record.pending + record.pending = nil + } else { + record.probesUntilApplied-- + } + s.records[r.ScheduleId] = record + } var memo *commonpb.Memo if record.memo != nil { memo = proto.Clone(record.memo).(*commonpb.Memo) @@ -125,22 +136,13 @@ func (s *scheduleRecoveryService) UpdateSchedule(_ context.Context, r *workflows } if f.code == codes.OK || f.apply { updated := proto.Clone(r.Schedule).(*schedulepb.Schedule) - if f.delayedEffect == 0 { - record.schedule = updated - s.records[r.ScheduleId] = record + if f.delayedEffect { + record.pending = updated + record.probesUntilApplied = 1 } else { - s.effects.Add(1) - go func() { - defer s.effects.Done() - time.Sleep(f.delayedEffect) - s.mu.Lock() - defer s.mu.Unlock() - if current, ok := s.records[r.ScheduleId]; ok { - current.schedule = updated - s.records[r.ScheduleId] = current - } - }() + record.schedule = updated } + s.records[r.ScheduleId] = record } if f.vanish { delete(s.records, r.ScheduleId) @@ -180,7 +182,7 @@ func scheduleRecoveryRPCFixture(t *testing.T) (client.Client, *scheduleRecoveryS workflowservice.RegisterWorkflowServiceServer(server, s) done := make(chan error, 1) go func() { done <- server.Serve(listener) }() - t.Cleanup(func() { server.Stop(); require.NoError(t, <-done); s.effects.Wait() }) + t.Cleanup(func() { server.Stop(); require.NoError(t, <-done) }) ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) defer cancel() c, err := client.DialContext(ctx, client.Options{HostPort: listener.Addr().String(), Namespace: "scheduler-recovery"}) @@ -191,19 +193,14 @@ func scheduleRecoveryRPCFixture(t *testing.T) (client.Client, *scheduleRecoveryS } func scheduleRecoveryInfo(c client.Client, logger *zap.SugaredLogger, enabled bool) loadgen.ScenarioInfo { - options := loadgen.MustResolveScenarioOptions("scheduler_stress", map[string]string{ + values := map[string]string{ "schedule-creation-per-iteration": "1", "schedule-reads-per-creation": "1", "schedule-updates-per-creation": "1", "scheduler-duration-per-iteration": "1s", "wait-time-before-cleanup": "0s", "operation-interval": "1ms", - }) - // Exercise faults independently of option registration. - if options.Lookup("recover-schedule-operations") == nil { - options.Bool("recover-schedule-operations", false, "") } if enabled { - if err := options.Set("recover-schedule-operations", "true"); err != nil { - panic(err) - } + values[RecoverScheduleOperationsFlag] = "true" } + options := loadgen.MustResolveScenarioOptions("scheduler_stress", values) return loadgen.ScenarioInfo{RunID: "crud-recovery", Client: c, Logger: logger, Options: options} } @@ -211,14 +208,15 @@ func TestSchedulerOperationRecoverySDK(t *testing.T) { cases := []struct { name string faults map[string]map[int]scheduleFault - wantErr string + wantErr error + wantCode codes.Code creates, updates, deletes, remaining int outcome string sdkRetry bool }{ {name: "healthy", creates: 1, updates: 1, deletes: 1}, - {name: "read_permanent_rejection", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.PermissionDenied}}}, wantErr: "describe rejection", creates: 1, updates: 1, deletes: 1}, - {name: "create_last_probe_not_found_no_fifth_call", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}}}, wantErr: "budget exhausted", creates: 2}, + {name: "read_permanent_rejection", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.PermissionDenied}}}, wantCode: codes.PermissionDenied, creates: 1, updates: 1, deletes: 1}, + {name: "create_last_probe_not_found_no_fifth_call", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}}}, wantErr: errScheduleRecoveryBudget, creates: 2}, {name: "sdk_unavailable_retries_same_rpc_request", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.Unavailable}}}, creates: 2, updates: 1, deletes: 1, sdkRetry: true}, {name: "sdk_accepted_create_then_exists", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.Unavailable, apply: true}}}, creates: 2, updates: 1, deletes: 1, sdkRetry: true, outcome: "observed_owned_create"}, {name: "describe_not_found_compatible", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.NotFound, apply: true}}}, creates: 1, outcome: "not_found_compatible"}, @@ -229,16 +227,16 @@ func TestSchedulerOperationRecoverySDK(t *testing.T) { {name: "read_deadline", faults: map[string]map[int]scheduleFault{"describe": {1: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 1, outcome: "acknowledged"}, {name: "update_internal_describe_deadline", faults: map[string]map[int]scheduleFault{"describe": {2: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 1, outcome: "acknowledged"}, {name: "update_accepted_response_lost", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_update_witness"}, - {name: "update_late_effect_no_replay", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, delayedEffect: 150 * time.Millisecond}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_update_witness"}, - {name: "update_identical_old_payload_not_success", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded}}}, wantErr: "outcome unknown", creates: 1, updates: 1, deletes: 1}, - {name: "update_response_lost_then_absent_is_unknown", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, vanish: true}}}, wantErr: "outcome unknown", creates: 1, updates: 1}, - {name: "update_permanent_rejection", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.PermissionDenied}}}, wantErr: "update response loss or rejection", creates: 1, updates: 1, deletes: 1}, + {name: "update_late_effect_no_replay", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, delayedEffect: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_update_witness"}, + {name: "update_identical_old_payload_not_success", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded}}}, wantErr: errScheduleUpdateUnknown, creates: 1, updates: 1, deletes: 1}, + {name: "update_response_lost_then_absent_is_unknown", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.DeadlineExceeded, apply: true, vanish: true}}}, wantErr: errScheduleUpdateUnknown, creates: 1, updates: 1}, + {name: "update_permanent_rejection", faults: map[string]map[int]scheduleFault{"update": {1: {code: codes.PermissionDenied}}}, wantCode: codes.PermissionDenied, creates: 1, updates: 1, deletes: 1}, {name: "delete_accepted_response_lost", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded, apply: true}}}, creates: 1, updates: 1, deletes: 1, outcome: "observed_absence"}, {name: "delete_not_accepted_retry_after_owned_probe", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded}}}, creates: 1, updates: 1, deletes: 2, outcome: "acknowledged"}, - {name: "delete_last_response_lost_no_fifth_probe", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded, apply: true}}}, wantErr: "budget exhausted", creates: 1, updates: 1, deletes: 2}, - {name: "foreign_owner_never_adopted_or_deleted", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.AlreadyExists, apply: true, foreign: true}}}, wantErr: "ownership or intent mismatch", creates: 1, remaining: 1}, - {name: "create_probe_exhausted_namespace_cleanup_owned", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}, "describe": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}, 3: {code: codes.DeadlineExceeded}}}, wantErr: "budget exhausted", creates: 1, remaining: 1}, - {name: "permanent_create_rejection", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.InvalidArgument}}}, wantErr: "create response loss or rejection", creates: 1}, + {name: "delete_last_response_lost_no_fifth_probe", faults: map[string]map[int]scheduleFault{"delete": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded, apply: true}}}, wantErr: errScheduleRecoveryBudget, creates: 1, updates: 1, deletes: 2}, + {name: "foreign_owner_never_adopted_or_deleted", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.AlreadyExists, apply: true, foreign: true}}}, wantErr: errScheduleIntentMismatch, creates: 1, remaining: 1}, + {name: "create_probe_exhausted_namespace_cleanup_owned", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}, "describe": {1: {code: codes.DeadlineExceeded}, 2: {code: codes.DeadlineExceeded}, 3: {code: codes.DeadlineExceeded}}}, wantErr: errScheduleRecoveryBudget, creates: 1, remaining: 1}, + {name: "permanent_create_rejection", faults: map[string]map[int]scheduleFault{"create": {1: {code: codes.InvalidArgument}}}, wantCode: codes.InvalidArgument, creates: 1}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -248,10 +246,12 @@ func TestSchedulerOperationRecoverySDK(t *testing.T) { } info := scheduleRecoveryInfo(c, logger, true) err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) - if tc.wantErr == "" { - require.NoError(t, err) + if tc.wantErr != nil { + require.ErrorIs(t, err, tc.wantErr) + } else if tc.wantCode != codes.OK { + require.Equal(t, tc.wantCode, serviceerror.ToStatus(err).Code()) } else { - require.ErrorContains(t, err, tc.wantErr) + require.NoError(t, err) } s.mu.Lock() defer s.mu.Unlock() @@ -301,7 +301,7 @@ func TestSchedulerOperationRecoveryDefaultUnchanged(t *testing.T) { s.faults = map[string]map[int]scheduleFault{"create": {1: {code: codes.DeadlineExceeded, apply: true}}} info := scheduleRecoveryInfo(c, logger, false) err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) - require.ErrorContains(t, err, "scripted create") + require.Equal(t, codes.DeadlineExceeded, serviceerror.ToStatus(err).Code()) require.Len(t, s.creates, 1) require.Nil(t, s.creates[0].Memo) require.Zero(t, s.calls["describe"]) @@ -344,7 +344,7 @@ func TestSchedulerOperationRecoveryIntentMismatch(t *testing.T) { } info := scheduleRecoveryInfo(c, logger, true) err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) - require.ErrorContains(t, err, "ownership or intent mismatch") + require.ErrorIs(t, err, errScheduleIntentMismatch) require.Empty(t, s.updates) require.Empty(t, s.deletes) require.Len(t, s.records, 1) @@ -409,7 +409,7 @@ func TestSchedulerOperationRecoveryFreshUpdateWitness(t *testing.T) { info := scheduleRecoveryInfo(c, logger, true) require.NoError(t, info.Options.Set("schedule-updates-per-creation", "2")) err := new(SchedulerExecutor).Execute(t.Context(), info.NewRun(1)) - require.ErrorContains(t, err, "outcome unknown") + require.ErrorIs(t, err, errScheduleUpdateUnknown) s.mu.Lock() defer s.mu.Unlock() require.Len(t, s.updates, 2) From 5c7c2011999712324dcff22ef182c0cb22bb254f Mon Sep 17 00:00:00 2001 From: lilydoar Date: Thu, 29 Oct 2026 10:55:21 -0700 Subject: [PATCH 3/3] Start scheduler protocol deadline after server setup --- scenarios/scheduler_recovery_server_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scenarios/scheduler_recovery_server_test.go b/scenarios/scheduler_recovery_server_test.go index 23e33e192..f09596002 100644 --- a/scenarios/scheduler_recovery_server_test.go +++ b/scenarios/scheduler_recovery_server_test.go @@ -85,9 +85,9 @@ func (p *scheduleResponseLossProxy) DeleteSchedule(ctx context.Context, r *workf } func TestSchedulerOperationRecoveryLocalServer(t *testing.T) { + server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{"frontend.allowedExperiments": []string{"*"}})) ctx, cancel := context.WithTimeout(t.Context(), 2*time.Minute) defer cancel() - server := workertest.StartDevServer(t, workertest.WithDynamicConfig(map[string]any{"frontend.allowedExperiments": []string{"*"}})) backend, err := client.DialContext(ctx, client.Options{HostPort: server.FrontendHostPort(), Namespace: "default"}) require.NoError(t, err) t.Cleanup(backend.Close)