271 lines
11 KiB
Go
271 lines
11 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"net/url"
|
|
"os"
|
|
"sort"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/transport/httpserver"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
func TestIntegrationTaskQuery(t *testing.T) {
|
|
dsn := os.Getenv("ORBIT_TEST_MYSQL_DSN")
|
|
if dsn == "" {
|
|
t.Skip("requires disposable MySQL")
|
|
}
|
|
ctx, stop := context.WithTimeout(context.Background(), 90*time.Second)
|
|
defer stop()
|
|
db, err := Open(ctx, dsn, 12, 4)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer db.Close()
|
|
if err = Migrate(ctx, db); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f := newExecutionFixture(t, ctx, db)
|
|
g := newExecutionFixture(t, ctx, db)
|
|
from := time.Date(2026, 9, 17, 0, 0, 0, 123456000, time.UTC)
|
|
before := from.Add(time.Hour)
|
|
makeTask := func(owner *executionFixture, key string, at time.Time, state pulse.TaskState) core.ID {
|
|
t.Helper()
|
|
out, err := owner.tasks.Submit(ctx, owner.token, string(executionID(t)), service.SubmitInput{QueueID: owner.resources.QueueID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key})
|
|
if err != nil || out.Task.BusinessKey != key {
|
|
t.Fatal("business key changed on admission", err)
|
|
}
|
|
v := out.Task
|
|
owner.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, at, owner.resources.NamespaceID, v.ID)
|
|
if state == pulse.TaskCancelled {
|
|
if _, err := owner.tasks.CancelTask(ctx, owner.token, v.ID, string(executionID(t)), service.CancelInput{ExpectedRevision: v.Revision, Reason: "query regression"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
return v.ID
|
|
}
|
|
key := "任务 /+%? é "
|
|
ids := []core.ID{makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from, pulse.TaskWaiting), makeTask(f, key, from.Add(time.Minute), pulse.TaskWaiting)}
|
|
sort.Slice(ids[:2], func(i, j int) bool { return ids[i] > ids[j] })
|
|
expected := []core.ID{ids[2], ids[0], ids[1]}
|
|
makeTask(f, key, from.Add(-time.Microsecond), pulse.TaskWaiting)
|
|
makeTask(f, key, before, pulse.TaskWaiting)
|
|
makeTask(f, key, from, pulse.TaskCancelled)
|
|
for _, other := range []string{strings.TrimSuffix(key, " "), strings.TrimSuffix(key, " ") + " ", strings.Replace(key, "é", "É", 1), strings.Replace(key, "é", "e\u0301", 1), ""} {
|
|
makeTask(f, other, from, pulse.TaskWaiting)
|
|
}
|
|
makeTask(g, key, from, pulse.TaskWaiting)
|
|
spaceID := makeTask(f, " ", from, pulse.TaskWaiting)
|
|
spaces, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: " "})
|
|
if err != nil || len(spaces.Items) != 1 || spaces.Items[0].ID != spaceID {
|
|
t.Fatal("space key not preserved", spaces, err)
|
|
}
|
|
|
|
otherQueue, err := f.tasks.CreateQueue(ctx, f.token, string(executionID(t)), service.QueueInput{Name: "other-query", MaxConcurrency: 1, RatePerSecond: 10})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
otherTask, err := f.tasks.Submit(ctx, f.token, string(executionID(t)), service.SubmitInput{QueueID: otherQueue.Queue.ID, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`), BusinessKey: key})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
f.exec(`UPDATE pulse_tasks SET created_at=? WHERE namespace_id=? AND id=?`, from, f.resources.NamespaceID, otherTask.Task.ID)
|
|
longKey := strings.Repeat("界", 191)
|
|
makeTask(f, longKey, from, pulse.TaskWaiting)
|
|
makeTask(f, longKey, from, pulse.TaskWaiting)
|
|
longPage, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1})
|
|
if err != nil || longPage.NextCursor == "" || len(longPage.NextCursor) > 1024 {
|
|
t.Fatal("maximum key overflowed cursor", err)
|
|
}
|
|
lastLong, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, BusinessKey: longKey, Limit: 1, Cursor: longPage.NextCursor})
|
|
if err != nil || len(lastLong.Items) != 1 || lastLong.Items[0].ID == longPage.Items[0].ID || lastLong.NextCursor != "" {
|
|
t.Fatal("maximum key pagination", lastLong, err)
|
|
}
|
|
base := service.ListInput{QueueID: f.resources.QueueID, BusinessKey: key, State: pulse.TaskWaiting, CreatedFrom: &from, CreatedBefore: &before, Limit: 1}
|
|
var firstCursor string
|
|
t.Run("precise_key_time_and_same_timestamp_pages", func(t *testing.T) {
|
|
in := base
|
|
var got []core.ID
|
|
for {
|
|
page, err := f.tasks.ListTasks(ctx, f.token, in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, v := range page.Items {
|
|
got = append(got, v.ID)
|
|
if v.BusinessKey != key || v.State != pulse.TaskWaiting {
|
|
t.Fatal("wrong query result")
|
|
}
|
|
}
|
|
if firstCursor == "" {
|
|
firstCursor = page.NextCursor
|
|
}
|
|
if page.NextCursor == "" {
|
|
break
|
|
}
|
|
in.Cursor = page.NextCursor
|
|
}
|
|
if fmt.Sprint(got) != fmt.Sprint(expected) {
|
|
t.Fatal("page boundaries", got, expected)
|
|
}
|
|
in = base
|
|
in.State = ""
|
|
in.Limit = 100
|
|
page, err := f.tasks.ListTasks(ctx, f.token, in)
|
|
if err != nil || len(page.Items) != 4 {
|
|
t.Fatal(page, err)
|
|
}
|
|
in = base
|
|
in.CreatedFrom = nil
|
|
page, err = f.tasks.ListTasks(ctx, f.token, in)
|
|
if err != nil || len(page.Items) != 1 || page.NextCursor == "" {
|
|
t.Fatal(page, err)
|
|
}
|
|
})
|
|
t.Run("cursor_binds_filters_and_current_authority", func(t *testing.T) {
|
|
for name, change := range map[string]func(*service.ListInput){"key": func(v *service.ListInput) { v.BusinessKey = strings.TrimSuffix(key, " ") }, "remove_key": func(v *service.ListInput) { v.BusinessKey = "" }, "state": func(v *service.ListInput) { v.State = "" }, "lower": func(v *service.ListInput) { z := from.Add(-time.Second); v.CreatedFrom = &z }, "upper": func(v *service.ListInput) { z := before.Add(time.Second); v.CreatedBefore = &z }, "queue": func(v *service.ListInput) { v.QueueID = g.resources.QueueID }} {
|
|
t.Run(name, func(t *testing.T) {
|
|
in := base
|
|
in.Cursor = firstCursor
|
|
change(&in)
|
|
_, err := f.tasks.ListTasks(ctx, f.token, in)
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
})
|
|
}
|
|
in := base
|
|
in.Cursor = firstCursor
|
|
_, err := g.tasks.ListTasks(ctx, g.token, in)
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
zone := time.FixedZone("equivalent", 8*3600)
|
|
lo, hi := from.In(zone), before.In(zone)
|
|
in.CreatedFrom = &lo
|
|
in.CreatedBefore = &hi
|
|
in.Limit = 2
|
|
page, err := f.tasks.ListTasks(ctx, f.token, in)
|
|
if err != nil || len(page.Items) != 2 {
|
|
t.Fatal("equivalent UTC or page size not supported", page, err)
|
|
}
|
|
f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY(?)) WHERE namespace_id=? AND id=?`, string(g.resources.QueueID), f.resources.NamespaceID, f.resources.CredentialID)
|
|
_, err = f.tasks.ListTasks(ctx, f.token, in)
|
|
expectWorkerCode(t, err, "PERMISSION_DENIED")
|
|
f.exec(`UPDATE orbit_credentials SET permissions=JSON_SET(permissions,'$.queues',JSON_ARRAY('*')) WHERE namespace_id=? AND id=?`, f.resources.NamespaceID, f.resources.CredentialID)
|
|
})
|
|
t.Run("legacy_cursor_and_strict_cursor_validation", func(t *testing.T) {
|
|
legacy := taskCursor{Namespace: f.resources.NamespaceID, Queue: f.resources.QueueID, State: pulse.TaskWaiting, Created: from.Add(time.Minute), ID: ids[2]}
|
|
raw, _ := json.Marshal(legacy)
|
|
encoded := base64.RawURLEncoding.EncodeToString(raw)
|
|
_, err := f.tasks.ListTasks(ctx, f.token, service.ListInput{QueueID: f.resources.QueueID, State: pulse.TaskWaiting, Cursor: encoded})
|
|
if err != nil {
|
|
t.Fatal("old cursor rejected", err)
|
|
}
|
|
in := base
|
|
in.Cursor = encoded
|
|
_, err = f.tasks.ListTasks(ctx, f.token, in)
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
valid, _ := base64.RawURLEncoding.DecodeString(firstCursor)
|
|
for _, raw := range []string{strings.TrimSuffix(string(valid), "}") + `,"v":2}`, strings.Replace(string(valid), `"v":2`, `"v":3`, 1), strings.Replace(string(valid), `"v":2`, `"v":null`, 1), strings.Replace(string(valid), `"v":2`, `"V":2`, 1), strings.TrimSuffix(string(valid), "}") + `,"secret":1}`, strings.Replace(string(valid), ".123456Z", ".123456001Z", 1)} {
|
|
in.Cursor = base64.RawURLEncoding.EncodeToString([]byte(raw))
|
|
_, err = f.tasks.ListTasks(ctx, f.token, in)
|
|
expectWorkerCode(t, err, "INVALID_ARGUMENT")
|
|
}
|
|
})
|
|
t.Run("http_validation_and_encoded_unicode", func(t *testing.T) {
|
|
server := httptest.NewServer(httpserver.New("", func(context.Context) error { return nil }, f.tasks).Handler)
|
|
defer server.Close()
|
|
call := func(query string, status int) service.TaskPage {
|
|
t.Helper()
|
|
r, _ := http.NewRequestWithContext(ctx, "GET", server.URL+"/v1/tasks?queue_id="+string(f.resources.QueueID)+"&"+query, nil)
|
|
r.Header.Set("Authorization", "Bearer "+f.resources.Token)
|
|
resp, err := server.Client().Do(r)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer resp.Body.Close()
|
|
raw, _ := io.ReadAll(resp.Body)
|
|
if resp.StatusCode != status {
|
|
t.Fatalf("query status=%d want=%d body=%s", resp.StatusCode, status, raw)
|
|
}
|
|
var out struct {
|
|
Data service.TaskPage `json:"data"`
|
|
}
|
|
if err = json.Unmarshal(raw, &out); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return out.Data
|
|
}
|
|
q := url.Values{"business_key": {key}, "state": {"WAITING"}, "created_from": {from.Format(time.RFC3339Nano)}, "created_before": {before.Format(time.RFC3339Nano)}}
|
|
if page := call(q.Encode(), 200); len(page.Items) != 3 {
|
|
t.Fatal(page)
|
|
}
|
|
for _, q := range []string{"business_key=", "business_key=a&business_key=b", "business_key=%ff", "business_key=%00", "created_from=", "created_from=2026-02-30T00:00:00Z", "created_from=2026-09-17T00:00:00.000000001Z", "created_from=2026-09-17", "created_before=1999-01-01T00:00:00Z", "created_from=2026-09-17T00:00:00Z&created_before=2026-09-17T00:00:00Z", "CreatedFrom=1", "created_from=%zz", "created_before=a&created_before=b"} {
|
|
call(q, 400)
|
|
}
|
|
})
|
|
t.Run("all_filter_shapes_use_ordered_composite_indexes", func(t *testing.T) {
|
|
cases := []struct {
|
|
state pulse.TaskState
|
|
key, index string
|
|
}{{"", "", "ix_task_created"}, {pulse.TaskWaiting, "", "ix_task_state_created"}, {"", key, "ix_task_queue_business_created"}, {pulse.TaskWaiting, key, "ix_task_queue_business_state_created"}}
|
|
for _, tc := range cases {
|
|
in := base
|
|
in.State = tc.state
|
|
in.BusinessKey = tc.key
|
|
digest, err := taskFilterDigest(in)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
q, args, err := taskListSQL(f.resources.NamespaceID, in, digest)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var raw string
|
|
if err = db.QueryRowContext(ctx, "EXPLAIN FORMAT=JSON "+q, args...).Scan(&raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var plan map[string]any
|
|
if err = json.Unmarshal([]byte(raw), &plan); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
found := false
|
|
var walk func(any)
|
|
walk = func(v any) {
|
|
switch x := v.(type) {
|
|
case map[string]any:
|
|
if x["using_filesort"] == true {
|
|
t.Fatal("query uses filesort", raw)
|
|
}
|
|
if x["table_name"] == "pulse_tasks" {
|
|
found = true
|
|
if x["key"] != tc.index || (x["access_type"] != "range" && x["access_type"] != "ref") {
|
|
t.Fatal("unbounded query access", raw)
|
|
}
|
|
}
|
|
for _, v := range x {
|
|
walk(v)
|
|
}
|
|
case []any:
|
|
for _, v := range x {
|
|
walk(v)
|
|
}
|
|
}
|
|
}
|
|
walk(plan)
|
|
if !found {
|
|
t.Fatal("task query plan missing", raw)
|
|
}
|
|
t.Logf("state=%q business_key=%t index=%s no filesort", tc.state, tc.key != "", tc.index)
|
|
}
|
|
})
|
|
}
|