orbit/sdk/go/schedule_integration_test.go

package orbit

import (
	"context"
	"encoding/json"
	"errors"
	"testing"
	"time"
)

func TestIntegrationScheduleLifecycle(t *testing.T) {
	api, queue, _ := integrationWorkerClient(t)
	ctx, cancel := context.WithTimeout(context.Background(), 25*time.Second)
	defer cancel()
	name := "scheduled." + workerTestID(t)
	w, err := NewWorker(api, WorkerOptions{QueueID: queue, Capacity: 1, Handlers: []WorkerHandler{runtimeHandler(name, func(ctx context.Context, e WorkerExecution) (json.RawMessage, error) { return e.Task.Payload, nil })}})
	if err != nil {
		t.Fatal(err)
	}
	run, stop := context.WithCancel(ctx)
	defer stop()
	done := make(chan error, 1)
	go func() { done <- w.Run(run) }()
	ended := false
	defer func() {
		stop()
		if !ended {
			<-done
		}
	}()
	start := time.Now().UTC().Truncate(time.Microsecond).Add(1500 * time.Millisecond)
	in := CreateScheduleRequest{Name: name, Paused: true, Definition: ScheduleDefinition{Rule: ScheduleRule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: start, IntervalMS: 1000}, Policy: SchedulePolicy{Misfire: "COALESCE", Overlap: "SERIAL", CatchupLimit: 2, CatchupWindowMS: 60000, GraceMS: 1000}, Task: SubmitTaskRequest{QueueID: queue, Handler: name, HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"n":9007199254740993}`)}}}
	key := "schedule-" + workerTestID(t)
	accepted, err := api.CreateSchedule(ctx, key, in)
	if err != nil {
		t.Fatal(err)
	}
	duplicate, err := api.CreateSchedule(ctx, key, in)
	if err != nil || !duplicate.Deduplicated || duplicate.Schedule.ID != accepted.Schedule.ID {
		t.Fatal("schedule submit replay", err)
	}
	resumed, err := api.ResumeSchedule(ctx, accepted.Schedule.ID, workerTestID(t), ControlScheduleRequest{ExpectedRevision: accepted.Schedule.Revision, Reason: "run integration schedule"})
	if err != nil {
		t.Fatal(err)
	}
	var occurrence ScheduleOccurrence
	for occurrence.TaskID == nil {
		page, err := api.ListScheduleOccurrences(ctx, resumed.Schedule.ID, SchedulePageRequest{})
		if err != nil {
			t.Fatal(err)
		}
		for _, item := range page.Items {
			if item.TaskID != nil {
				occurrence = item
				break
			}
		}
		if occurrence.TaskID != nil {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatal(ctx.Err())
		case <-time.After(80 * time.Millisecond):
		}
	}
	task := waitRuntimeState(t, ctx, api, *occurrence.TaskID, TaskSucceeded)
	if string(task.Result) != `{"n":9007199254740993}` || task.AttemptCount != 1 {
		t.Fatal("scheduled Worker result corrupted")
	}
	var paused ScheduleResult
	for {
		current, err := api.GetSchedule(ctx, resumed.Schedule.ID)
		if err != nil {
			t.Fatal(err)
		}
		paused, err = api.PauseSchedule(ctx, current.ID, workerTestID(t), ControlScheduleRequest{ExpectedRevision: current.Revision, Reason: "publish next definition"})
		if err == nil {
			break
		}
		var apiErr *APIError
		if !errors.As(err, &apiErr) || apiErr.Code != "STATE_CONFLICT" {
			t.Fatal(err)
		}
	}
	in.Definition.Rule = ScheduleRule{Kind: "CRON", Timezone: "Australia/Lord_Howe", StartAt: start, Cron: "*/5 * * * *"}
	publish := PublishScheduleRequest{ExpectedRevision: paused.Schedule.Revision, Reason: "test immutable cron version", Definition: in.Definition}
	publishKey := workerTestID(t)
	updated, err := api.PublishSchedule(ctx, paused.Schedule.ID, publishKey, publish)
	if err != nil || updated.Schedule.CurrentVersion != 2 || !updated.Schedule.Paused {
		t.Fatal("publish schedule", err)
	}
	old, err := api.GetScheduleVersion(ctx, paused.Schedule.ID, 1)
	if err != nil || old.Definition.Rule.Kind != "FIXED_RATE" {
		t.Fatal("old version lost", err)
	}
	current, err := api.GetScheduleVersion(ctx, paused.Schedule.ID, 2)
	if err != nil || current.Definition.Rule.Kind != "CRON" {
		t.Fatal("new cron version lost", err)
	}
	duplicate, err = api.PublishSchedule(ctx, paused.Schedule.ID, publishKey, publish)
	if err != nil || !duplicate.Deduplicated || duplicate.Schedule.Revision != updated.Schedule.Revision {
		t.Fatal("publish replay changed revision")
	}
	list, err := api.ListSchedules(ctx, queue, SchedulePageRequest{Limit: 100})
	if err != nil {
		t.Fatal(err)
	}
	found := false
	for _, v := range list.Items {
		found = found || v.ID == updated.Schedule.ID
	}
	if !found {
		t.Fatal("schedule missing from list")
	}
	before, err := api.ListScheduleOccurrences(ctx, updated.Schedule.ID, SchedulePageRequest{})
	if err != nil {
		t.Fatal(err)
	}
	select {
	case <-ctx.Done():
		t.Fatal(ctx.Err())
	case <-time.After(1200 * time.Millisecond):
	}
	after, err := api.ListScheduleOccurrences(ctx, updated.Schedule.ID, SchedulePageRequest{})
	if err != nil || len(after.Items) != len(before.Items) {
		t.Fatal("paused version fired")
	}
	stop()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
	ended = true
}