401 lines
15 KiB
Go
401 lines
15 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"os"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/schedule"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
type scheduleFixture struct {
|
|
*executionFixture
|
|
store *ScheduleStore
|
|
api *schedule.Service
|
|
}
|
|
|
|
func newScheduleFixture(t *testing.T, f *executionFixture) *scheduleFixture {
|
|
t.Helper()
|
|
raw, _ := json.Marshal(auth.Permissions{Actions: []string{"tasks:submit", "tasks:read", "tasks:cancel", "queues:read", "schedules:read", "schedules:manage", "workers:connect", "workers:consume"}, Queues: []string{"*"}})
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.resources.NamespaceID, f.resources.CredentialID)
|
|
store, err := NewScheduleStore(f.db, executionID(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return &scheduleFixture{f, store, schedule.New(store)}
|
|
}
|
|
func (f *scheduleFixture) definition() schedule.Definition {
|
|
return schedule.Definition{Rule: schedule.Rule{Kind: "FIXED_RATE", Timezone: "UTC", StartAt: time.Now().UTC().Truncate(time.Microsecond).Add(-time.Hour), IntervalMS: 1000}, Policy: schedule.Policy{Misfire: "CATCH_UP", Overlap: "PARALLEL", CatchupLimit: 3, CatchupWindowMS: 30000, GraceMS: 1000}, Task: service.SubmitInput{QueueID: f.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"number":9007199254740993}`)}}
|
|
}
|
|
func (f *scheduleFixture) create(in schedule.Definition) schedule.View {
|
|
f.t.Helper()
|
|
out, err := f.api.Create(f.ctx, f.token, string(executionID(f.t)), schedule.CreateInput{Name: "plan-" + string(executionID(f.t)), Definition: in})
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out.Schedule
|
|
}
|
|
func (f *scheduleFixture) current(id core.ID) schedule.View {
|
|
f.t.Helper()
|
|
out, err := f.api.Get(f.ctx, f.token, id)
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out
|
|
}
|
|
func (f *scheduleFixture) occurrences(id core.ID) []schedule.OccurrenceView {
|
|
f.t.Helper()
|
|
out, err := f.api.Occurrences(f.ctx, f.token, id, schedule.PageInput{Limit: 100})
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out.Items
|
|
}
|
|
|
|
// 等待真实时间形成积压，不将游标改到计划实际生效之前。
|
|
func (f *scheduleFixture) due(v schedule.View, age time.Duration) {
|
|
f.t.Helper()
|
|
if delay := time.Until(v.NextFireAt.Add(age)); delay > 0 {
|
|
timer := time.NewTimer(delay)
|
|
defer timer.Stop()
|
|
select {
|
|
case <-timer.C:
|
|
case <-f.ctx.Done():
|
|
f.t.Fatal(f.ctx.Err())
|
|
}
|
|
}
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
}
|
|
func (f *scheduleFixture) tick() ScheduleStats {
|
|
f.t.Helper()
|
|
out, err := f.store.Tick(f.ctx, 100)
|
|
if err != nil {
|
|
f.t.Fatal(err)
|
|
}
|
|
return out
|
|
}
|
|
func TestIntegrationSchedules(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer cancel()
|
|
db, err := Open(ctx, dsn, 32, 8)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err := Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
fixture := func(t *testing.T) *scheduleFixture { return newScheduleFixture(t, newExecutionFixture(t, ctx, db)) }
|
|
// 连续发布经过“下一版本尚待初始化”和空版本时，不能丢失积压或重新开始已结束版本。
|
|
for _, emptyMiddle := range []bool{false, true} {
|
|
name := "publish_preserves_uninitialized_version"
|
|
if emptyMiddle {
|
|
name = "empty_version_then_finished_latest_stays_finished"
|
|
}
|
|
t.Run(name, func(t *testing.T) {
|
|
f := fixture(t)
|
|
finite := func(delay time.Duration) schedule.Definition {
|
|
in := f.definition()
|
|
in.Rule.StartAt = time.Now().UTC().Truncate(time.Microsecond).Add(delay)
|
|
end := in.Rule.StartAt.Add(time.Millisecond)
|
|
in.Rule.EndAt = &end
|
|
return in
|
|
}
|
|
publish := func(id core.ID, in schedule.Definition) schedule.View {
|
|
t.Helper()
|
|
current := f.current(id)
|
|
out, err := f.api.Publish(ctx, f.token, id, string(executionID(t)), schedule.PublishInput{ExpectedRevision: current.Revision, Reason: "version boundary regression", Definition: in})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out.Schedule
|
|
}
|
|
waitUntil := func(at time.Time) {
|
|
t.Helper()
|
|
timer := time.NewTimer(time.Until(at.Add(10 * time.Millisecond)))
|
|
defer timer.Stop()
|
|
select {
|
|
case <-timer.C:
|
|
case <-ctx.Done():
|
|
t.Fatal(ctx.Err())
|
|
}
|
|
}
|
|
v := f.create(finite(300 * time.Millisecond))
|
|
waitUntil(*v.NextFireAt)
|
|
delay := 300 * time.Millisecond
|
|
if emptyMiddle {
|
|
delay = 30 * time.Second
|
|
}
|
|
middle := finite(delay)
|
|
publish(v.ID, middle)
|
|
var latest schedule.Definition
|
|
if emptyMiddle {
|
|
latest = finite(300 * time.Millisecond)
|
|
publish(v.ID, latest)
|
|
}
|
|
f.tick() // 第一版本结束，第二版本尚未计算第一个触发点。
|
|
pending := f.current(v.ID)
|
|
if pending.ProcessingVersion != 2 || pending.NextFireAt != nil || pending.CheckAt == nil {
|
|
t.Fatal("expected pending initialization of version 2")
|
|
}
|
|
if !emptyMiddle {
|
|
waitUntil(middle.Rule.StartAt)
|
|
latest = finite(300 * time.Millisecond)
|
|
published := publish(v.ID, latest)
|
|
if published.ProcessingVersion != 2 {
|
|
t.Fatal("publishing discarded an uninitialized version")
|
|
}
|
|
}
|
|
waitUntil(latest.Rule.StartAt)
|
|
for i := 0; i < 3 && f.current(v.ID).CheckAt != nil; i++ {
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
}
|
|
finished := f.current(v.ID)
|
|
if finished.ProcessingVersion != 3 || finished.CheckAt != nil || finished.NextFireAt != nil {
|
|
t.Fatal("finished latest version retained scan work")
|
|
}
|
|
got := f.occurrences(v.ID)
|
|
want := map[uint64]time.Time{1: *v.NextFireAt, 3: latest.Rule.StartAt}
|
|
if !emptyMiddle {
|
|
want[2] = middle.Rule.StartAt
|
|
}
|
|
if len(got) != len(want) {
|
|
t.Fatalf("got %d occurrences, want %d", len(got), len(want))
|
|
}
|
|
for _, o := range got {
|
|
at, ok := want[o.ScheduleVersion]
|
|
if !ok || !o.ScheduledAt.Equal(at) || o.Decision != "FIRED" || o.TaskID == nil {
|
|
t.Fatal("version occurrence lost or duplicated")
|
|
}
|
|
delete(want, o.ScheduleVersion)
|
|
}
|
|
f.tick()
|
|
if f.current(v.ID).Revision != finished.Revision {
|
|
t.Fatal("finished schedule advanced again")
|
|
}
|
|
})
|
|
}
|
|
t.Run("multi_node_atomic_trigger_and_rate", func(t *testing.T) {
|
|
f := fixture(t)
|
|
v := f.create(f.definition())
|
|
f.due(v, 5*time.Second)
|
|
other, err := NewScheduleStore(db, executionID(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var wg sync.WaitGroup
|
|
failures := make(chan error, 2)
|
|
for _, s := range []*ScheduleStore{f.store, other} {
|
|
wg.Add(1)
|
|
go func(s *ScheduleStore) { defer wg.Done(); _, err := s.Tick(ctx, 100); failures <- err }(s)
|
|
}
|
|
wg.Wait()
|
|
close(failures)
|
|
for err := range failures {
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
got := f.occurrences(v.ID)
|
|
if len(got) != 3 || f.count("pulse_tasks", "owner_kind='schedule' AND owner_id=?", v.ID) != 3 {
|
|
t.Fatal("duplicate or partial trigger")
|
|
}
|
|
current := f.current(v.ID)
|
|
if current.CheckAt == nil || !current.CheckAt.After(current.UpdatedAt) {
|
|
t.Fatal("catchup rate not durable")
|
|
}
|
|
f.tick()
|
|
if len(f.occurrences(v.ID)) != 3 {
|
|
t.Fatal("repeat scan bypassed cooldown")
|
|
}
|
|
for _, o := range got {
|
|
if o.TaskID == nil || o.ScheduleVersion != 1 {
|
|
t.Fatal("missing ownership")
|
|
}
|
|
task := f.get(*o.TaskID)
|
|
if string(task.Payload) != `{"number":9007199254740993}` || task.AttemptCount != 0 {
|
|
t.Fatal("template snapshot damaged")
|
|
}
|
|
}
|
|
f.exec(`UPDATE orbit_namespaces SET enabled=FALSE WHERE id=?`, f.resources.NamespaceID)
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
if f.count("pulse_tasks", "owner_kind='schedule' AND owner_id=?", v.ID) != 3 {
|
|
t.Fatal("disabled namespace generated tasks")
|
|
}
|
|
f.exec(`UPDATE orbit_namespaces SET enabled=TRUE WHERE id=?`, f.resources.NamespaceID)
|
|
})
|
|
t.Run("idempotency_pause_scope_and_version", func(t *testing.T) {
|
|
f := fixture(t)
|
|
in := schedule.CreateInput{Name: "stable", Definition: f.definition(), Paused: true}
|
|
key := string(executionID(t))
|
|
out, err := f.api.Create(ctx, f.token, key, in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
again, err := f.api.Create(ctx, f.token, key, in)
|
|
if err != nil || !again.Deduplicated || again.Schedule.ID != out.Schedule.ID {
|
|
t.Fatal("creation replay")
|
|
}
|
|
in.Paused = false
|
|
_, err = f.api.Create(ctx, f.token, key, in)
|
|
expectWorkerCode(t, err, "IDEMPOTENCY_CONFLICT")
|
|
v := out.Schedule
|
|
f.due(v, 4*time.Second)
|
|
f.tick()
|
|
if len(f.occurrences(v.ID)) != 0 {
|
|
t.Fatal("paused schedule fired")
|
|
}
|
|
next := f.definition()
|
|
next.Task.Payload = json.RawMessage(`{"version":2}`)
|
|
published, err := f.api.Publish(ctx, f.token, v.ID, string(executionID(t)), schedule.PublishInput{ExpectedRevision: v.Revision, Reason: "new template", Definition: next})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if published.Schedule.ProcessingVersion != 1 || published.Schedule.CurrentVersion != 2 {
|
|
t.Fatal("publish erased old backlog")
|
|
}
|
|
old, err := f.api.Version(ctx, f.token, v.ID, 1)
|
|
if err != nil || string(old.Definition.Task.Payload) != `{"number":9007199254740993}` {
|
|
t.Fatal("old version changed")
|
|
}
|
|
resumeKey := string(executionID(t))
|
|
resume := schedule.ControlInput{ExpectedRevision: published.Schedule.Revision, Reason: "resume work"}
|
|
resumed, err := f.api.Control(ctx, f.token, v.ID, resumeKey, resume, false)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
repeated, err := f.api.Control(ctx, f.token, v.ID, resumeKey, resume, false)
|
|
if err != nil || !repeated.Deduplicated || repeated.Schedule.Revision != resumed.Schedule.Revision {
|
|
t.Fatal("control replay")
|
|
}
|
|
f.tick()
|
|
for _, o := range f.occurrences(v.ID) {
|
|
if o.ScheduleVersion != 1 || !o.ScheduledAt.Before(published.Schedule.Current.EffectiveAt) {
|
|
t.Fatal("historical version boundary lost")
|
|
}
|
|
}
|
|
for i := 0; i < 3 && f.current(v.ID).ProcessingVersion == 1; i++ {
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
}
|
|
if f.current(v.ID).ProcessingVersion != 2 {
|
|
t.Fatal("old backlog never advanced to published version")
|
|
}
|
|
for _, o := range f.occurrences(v.ID) {
|
|
if o.ScheduleVersion == 1 && !o.ScheduledAt.Before(published.Schedule.Current.EffectiveAt) {
|
|
t.Fatal("old version crossed effective boundary")
|
|
}
|
|
}
|
|
// 另一命名空间知道 ID 也不可读或管理。
|
|
stranger := fixture(t)
|
|
_, err = stranger.api.Get(ctx, stranger.token, v.ID)
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
raw, _ := json.Marshal(auth.Permissions{Actions: []string{"schedules:read", "schedules:manage"}, Queues: []string{string(executionID(t))}})
|
|
f.exec(`UPDATE orbit_credentials SET permissions=? WHERE namespace_id=? AND id=?`, string(raw), f.resources.NamespaceID, f.resources.CredentialID)
|
|
_, err = f.api.Get(ctx, f.token, v.ID)
|
|
expectWorkerCode(t, err, "NOT_FOUND")
|
|
})
|
|
t.Run("persisted_skip_and_coalesce_ranges", func(t *testing.T) {
|
|
f := fixture(t)
|
|
in := f.definition()
|
|
in.Policy.CatchupWindowMS = 2000
|
|
in.Policy.Misfire = "COALESCE"
|
|
coalesce := f.create(in)
|
|
in.Policy.Misfire = "SKIP"
|
|
skip := f.create(in)
|
|
f.due(coalesce, 4*time.Second)
|
|
f.tick()
|
|
for _, v := range []schedule.View{coalesce, skip} {
|
|
items := f.occurrences(v.ID)
|
|
skipped, merged := false, false
|
|
for _, o := range items {
|
|
if o.Decision == "SKIPPED" {
|
|
skipped = true
|
|
if o.TaskID != nil || o.CoveredUntil == nil || !o.CoveredUntil.After(o.ScheduledAt) {
|
|
t.Fatal("invalid skipped range")
|
|
}
|
|
}
|
|
if o.Decision == "COALESCED" {
|
|
merged = true
|
|
if o.TaskID == nil || o.CoveredUntil == nil {
|
|
t.Fatal("coalesced range lost task")
|
|
}
|
|
}
|
|
}
|
|
if !skipped || (v.ID == coalesce.ID && !merged) {
|
|
t.Fatal("misfire decisions not persisted")
|
|
}
|
|
}
|
|
})
|
|
t.Run("overlap_serial_unknown_and_waiting_cancel", func(t *testing.T) {
|
|
f := fixture(t)
|
|
in := f.definition()
|
|
in.Policy.Overlap = "SERIAL"
|
|
v := f.create(in)
|
|
f.due(v, 5*time.Second)
|
|
f.tick()
|
|
items := f.occurrences(v.ID)
|
|
if len(items) != 1 || items[0].TaskID == nil {
|
|
t.Fatal("serial did not start one")
|
|
}
|
|
first := f.current(v.ID)
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
if len(f.occurrences(v.ID)) != 1 || !f.current(v.ID).NextFireAt.Equal(*first.NextFireAt) {
|
|
t.Fatal("serial skipped waiting task")
|
|
}
|
|
task := f.get(*items[0].TaskID)
|
|
if _, err := f.tasks.CancelTask(ctx, f.token, task.ID, string(executionID(t)), service.CancelInput{ExpectedRevision: task.Revision, Reason: "cancel pending scheduled task"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
if len(f.occurrences(v.ID)) != 2 {
|
|
t.Fatal("serial did not continue after terminal task")
|
|
}
|
|
retryFinish(t, f.executionFixture, retryStart(t, f.executionFixture), "UNKNOWN", false)
|
|
waiting := f.current(v.ID)
|
|
f.exec(`UPDATE pulse_schedules SET check_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, v.ID)
|
|
f.tick()
|
|
if len(f.occurrences(v.ID)) != 2 || !f.current(v.ID).NextFireAt.Equal(*waiting.NextFireAt) {
|
|
t.Fatal("unknown outcome did not block serial schedule")
|
|
}
|
|
})
|
|
t.Run("transaction_rollback_and_read_pagination", func(t *testing.T) {
|
|
f := fixture(t)
|
|
v := f.create(f.definition())
|
|
f.due(v, 6*time.Second)
|
|
prior := f.current(v.ID)
|
|
f.exec(`CREATE TRIGGER orbit_test_schedule_event BEFORE INSERT ON pulse_schedule_occurrences FOR EACH ROW SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='controlled trigger failure'`)
|
|
_, err := f.store.Tick(ctx, 100)
|
|
f.exec(`DROP TRIGGER orbit_test_schedule_event`)
|
|
if err == nil || len(f.occurrences(v.ID)) != 0 || f.count("pulse_tasks", "owner_id=?", v.ID) != 0 || !f.current(v.ID).NextFireAt.Equal(*prior.NextFireAt) {
|
|
t.Fatal("partial schedule transaction")
|
|
}
|
|
f.tick()
|
|
page, err := f.api.Occurrences(ctx, f.token, v.ID, schedule.PageInput{Limit: 1})
|
|
if err != nil || len(page.Items) != 1 || page.NextCursor == "" {
|
|
t.Fatal("missing page")
|
|
}
|
|
second, err := f.api.Occurrences(ctx, f.token, v.ID, schedule.PageInput{Limit: 1, Cursor: page.NextCursor})
|
|
if err != nil || len(second.Items) != 1 || second.Items[0].ID == page.Items[0].ID {
|
|
t.Fatal("duplicate page")
|
|
}
|
|
another := f.create(f.definition())
|
|
_, err = f.api.Occurrences(ctx, f.token, another.ID, schedule.PageInput{Cursor: page.NextCursor})
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
})
|
|
}
|