feat: add precise task business key and creation time queries

2026-09-17 22:19:36 +08:00
parent a50d5faac7
commit 7641423f9d
32 changed files with 1105 additions and 93 deletions
--- a/internal/storage/mysql/application_resources_upgrade_integration_test.go
+++ b/internal/storage/mysql/application_resources_upgrade_integration_test.go
@@ -64,6 +64,7 @@ func TestIntegrationApplicationResourcesUpgrade(t *testing.T) {
 	retryFinish(t, g, finished, "SUCCEEDED", false)
 	assertApplicationCounts(t, f, 4, 2, 1)
 	// 还原后续新增表和字段，得到真实 v9 结构；现有任务内容保持不变。
+	f.exec(`ALTER TABLE pulse_tasks DROP INDEX ix_task_queue_business_created, DROP INDEX ix_task_queue_business_state_created, MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NULL COMMENT '业务侧关联键，用于检索追踪，不自动提供业务幂等'`)
 	f.exec(`ALTER TABLE pulse_tasks DROP CHECK ck_task_retry_not_self, DROP INDEX uq_task_retry_parent, ADD INDEX fk_pulse_tasks_3(namespace_id,retry_of)`)
 	f.exec(`DROP TABLE orbit_application_resources`)
 	f.exec(`DROP TABLE orbit_result_events,orbit_event_streams`)
--- a/internal/storage/mysql/migrations/000016_task_query_filters.up.sql
+++ b/internal/storage/mysql/migrations/000016_task_query_filters.up.sql
@@ -0,0 +1,6 @@
+-- 业务关联键按精确值查询；NO PAD二进制排序区分尾部空格，不规范化Unicode。
+-- 两种新索引覆盖已授权队列内的关联键及可选状态过滤，不增加外键。
+ALTER TABLE pulse_tasks
+    MODIFY COLUMN business_key VARCHAR(191) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin NULL COMMENT '可空业务关联键，精确匹配并区分大小写及尾部空格，不自动提供业务幂等',
+    ADD KEY ix_task_queue_business_created(namespace_id,queue_id,business_key,created_at,id) COMMENT '队列内按业务键及创建时间分页',
+    ADD KEY ix_task_queue_business_state_created(namespace_id,queue_id,business_key,state,created_at,id) COMMENT '队列内按业务键和状态及创建时间分页';
--- a/internal/storage/mysql/no_foreign_keys_integration_test.go
+++ b/internal/storage/mysql/no_foreign_keys_integration_test.go
@@ -89,7 +89,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
 	indexes := func() []string {
 		t.Helper()
 		// v15 将原 retry_of 非唯一索引替换为相同列的唯一索引，下方单独验证其定义。
-		rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
+		rows, err := db.QueryContext(ctx, `SELECT CONCAT_WS(':',TABLE_NAME,INDEX_NAME,NON_UNIQUE,SEQ_IN_INDEX,COLUMN_NAME,COALESCE(SUB_PART,0),INDEX_TYPE,IS_VISIBLE) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME NOT IN ('orbit_application_resources','orbit_event_streams','orbit_result_events') AND INDEX_NAME NOT IN ('ix_task_priority_dispatch','ix_delivery_subscription_due','ix_delivery_subscription_lease','ix_delivery_history','ix_delivery_state_history','uq_task_retry_parent','fk_pulse_tasks_3','ix_task_queue_business_created','ix_task_queue_business_state_created') ORDER BY TABLE_NAME,INDEX_NAME,SEQ_IN_INDEX`)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -113,6 +113,7 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
 	exec(`INSERT INTO orbit_namespaces(id,application_id,name,environment) VALUES (?,?,?,'test')`, ns, app, string(ns))
 	exec(`INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,policy) VALUES (?,?,?,4,10,'{}')`, ns, queue, string(queue))
 	exec(`INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,available_at,deadline_at,attempt_timeout_ms,max_attempts) VALUES (?,?,?,'object.prepare','v1','v1','{"n":9007199254740993}','{}',UTC_TIMESTAMP(6),DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 1 HOUR),5000,3)`, ns, task, queue)
+	exec(`UPDATE pulse_tasks SET business_key=? WHERE namespace_id=? AND id=?`, "关联键é尾部 ", ns, task)
 	taskSnapshot := func() []string {
 		t.Helper()
 		rows, err := db.QueryContext(ctx, `SELECT * FROM pulse_tasks WHERE namespace_id=? AND id=?`, ns, task)
@@ -172,6 +173,10 @@ func TestIntegrationNoForeignKeysUpgrade(t *testing.T) {
 	if err := db.QueryRowContext(ctx, `SELECT GROUP_CONCAT(CONCAT(COLUMN_NAME,':',NON_UNIQUE) ORDER BY SEQ_IN_INDEX) FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND INDEX_NAME='uq_task_retry_parent'`).Scan(&retryIndex); err != nil || retryIndex != "namespace_id:0,retry_of:0" {
 		t.Fatalf("retry uniqueness replacement=%q err=%v", retryIndex, err)
 	}
+	var collation string
+	if err := db.QueryRowContext(ctx, `SELECT COLLATION_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='pulse_tasks' AND COLUMN_NAME='business_key'`).Scan(&collation); err != nil || collation != "utf8mb4_0900_bin" {
+		t.Fatalf("business key collation=%q err=%v", collation, err)
+	}
 	if afterIndexes := indexes(); !reflect.DeepEqual(beforeIndexes, afterIndexes) {
 		before, after := map[string]bool{}, map[string]bool{}
 		for _, index := range beforeIndexes {
--- a/internal/storage/mysql/schedule_tick.go
+++ b/internal/storage/mysql/schedule_tick.go
@@ -224,7 +224,7 @@ func (s *ScheduleStore) saveOccurrence(ctx context.Context, tx *sql.Tx, v schedu
 		if err != nil {
 			return err
 		}
-		_, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,priority,owner_kind,owner_id,available_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,'schedule',?,?,?,?,?,?,?)`, v.NamespaceID, value, v.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, v.ID, now, now.Add(time.Duration(in.TaskTimeoutMS)*time.Millisecond), in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now)
+		_, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,priority,owner_kind,owner_id,available_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,'schedule',?,?,?,?,?,?,?)`, v.NamespaceID, value, v.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, businessKeyValue(in.BusinessKey), concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, v.ID, now, now.Add(time.Duration(in.TaskTimeoutMS)*time.Millisecond), in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now)
 		if err != nil {
 			return err
 		}
--- a/internal/storage/mysql/task_query.go
+++ b/internal/storage/mysql/task_query.go
@@ -0,0 +1,180 @@
+package mysql
+
+import (
+	"context"
+	"database/sql"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"time"
+
+	"git.yuntuo.hk/k2safe/orbit/internal/fault"
+	"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
+	core "git.yuntuo.hk/k2safe/orbit/model"
+	"git.yuntuo.hk/k2safe/orbit/platform/auth"
+	pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
+	"git.yuntuo.hk/k2safe/orbit/pulse/service"
+)
+
+// taskCursor 是已授权查询的排序位置，不能代替凭据；v2摘要绑定全部筛选。
+type taskCursor struct {
+	Version   int             `json:"v,omitempty"`
+	Namespace core.ID         `json:"n"`
+	Queue     core.ID         `json:"q"`
+	State     pulse.TaskState `json:"s"`
+	Created   time.Time       `json:"t"`
+	ID        core.ID         `json:"i"`
+	Filter    string          `json:"f,omitempty"`
+}
+
+func taskFilterDigest(in service.ListInput) (string, error) {
+	digest, err := requestDigest(struct {
+		Domain      string          `json:"domain"`
+		Queue       core.ID         `json:"queue"`
+		State       pulse.TaskState `json:"state"`
+		BusinessKey string          `json:"business_key"`
+		From        *time.Time      `json:"from"`
+		Before      *time.Time      `json:"before"`
+	}{"orbit.task-query.v2", in.QueueID, in.State, in.BusinessKey, in.CreatedFrom, in.CreatedBefore})
+	return hex.EncodeToString(digest[:]), err
+}
+
+func decodeTaskCursor(value string, ns core.ID, in service.ListInput, filter string) (taskCursor, error) {
+	var c taskCursor
+	invalid := fault.Invalid("cursor does not match this task query")
+	raw, err := base64.RawURLEncoding.Strict().DecodeString(value)
+	if err != nil || len(raw) > 768 {
+		return c, invalid
+	}
+	if _, err = jsonvalue.Canonical(raw); err != nil {
+		return c, invalid
+	}
+	var fields map[string]json.RawMessage
+	if json.Unmarshal(raw, &fields) != nil || fields == nil {
+		return c, invalid
+	}
+	for k, v := range fields {
+		switch k {
+		case "v", "n", "q", "s", "t", "i", "f":
+		default:
+			return c, invalid
+		}
+		if string(v) == "null" {
+			return c, invalid
+		}
+	}
+	for _, k := range []string{"n", "q", "s", "t", "i"} {
+		if _, ok := fields[k]; !ok {
+			return c, invalid
+		}
+	}
+	if json.Unmarshal(raw, &c) != nil || c.Namespace != ns || c.Queue != in.QueueID || c.State != in.State || c.ID.Validate() != nil || c.Created.Year() < 2000 || c.Created.Year() > 9998 || c.Created.Nanosecond()%1000 != 0 {
+		return c, invalid
+	}
+	if c.Version == 0 {
+		// 旧版未带版本的游标仅兼容原队列/状态查询，不能附加新筛选。
+		if _, ok := fields["v"]; ok {
+			return c, invalid
+		}
+		if _, ok := fields["f"]; ok {
+			return c, invalid
+		}
+		if in.BusinessKey != "" || in.CreatedFrom != nil || in.CreatedBefore != nil {
+			return c, invalid
+		}
+	} else if c.Version != 2 || c.Filter != filter {
+		return c, invalid
+	}
+	if in.CreatedFrom != nil && c.Created.Before(*in.CreatedFrom) || in.CreatedBefore != nil && !c.Created.Before(*in.CreatedBefore) {
+		return c, invalid
+	}
+	return c, nil
+}
+
+const taskSummaryColumns = `id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,deadline_at,attempt_count,created_at,updated_at,effect_uncertain`
+
+// taskListSQL 选择固定索引名，等值筛选全部位于创建时间前；游标不依赖状态更新时间。
+func taskListSQL(ns core.ID, in service.ListInput, filter string) (string, []any, error) {
+	index := "ix_task_created"
+	if in.State != "" {
+		index = "ix_task_state_created"
+	}
+	if in.BusinessKey != "" {
+		index = "ix_task_queue_business_created"
+		if in.State != "" {
+			index = "ix_task_queue_business_state_created"
+		}
+	}
+	q := `SELECT ` + taskSummaryColumns + ` FROM pulse_tasks FORCE INDEX (` + index + `) WHERE namespace_id=? AND queue_id=?`
+	args := []any{ns, in.QueueID}
+	if in.BusinessKey != "" {
+		q += " AND business_key=?"
+		args = append(args, in.BusinessKey)
+	}
+	if in.State != "" {
+		q += " AND state=?"
+		args = append(args, in.State)
+	}
+	if in.CreatedFrom != nil {
+		q += " AND created_at>=?"
+		args = append(args, *in.CreatedFrom)
+	}
+	if in.CreatedBefore != nil {
+		q += " AND created_at<?"
+		args = append(args, *in.CreatedBefore)
+	}
+	if in.Cursor != "" {
+		c, err := decodeTaskCursor(in.Cursor, ns, in, filter)
+		if err != nil {
+			return "", nil, err
+		}
+		q += " AND (created_at<? OR (created_at=? AND id<?))"
+		args = append(args, c.Created, c.Created, c.ID)
+	}
+	q += " ORDER BY created_at DESC,id DESC LIMIT ?"
+	args = append(args, in.Limit+1)
+	return q, args, nil
+}
+
+func (s *TaskStore) ListTasks(ctx context.Context, token auth.Token, in service.ListInput) (service.TaskPage, error) {
+	out := service.TaskPage{Items: []service.TaskSummary{}}
+	err := s.transaction(ctx, token, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error {
+		if !p.AllowsQueue(in.QueueID) {
+			return fault.Forbidden
+		}
+		filter, err := taskFilterDigest(in)
+		if err != nil {
+			return err
+		}
+		q, args, err := taskListSQL(token.NamespaceID, in, filter)
+		if err != nil {
+			return err
+		}
+		rows, err := tx.QueryContext(ctx, q, args...)
+		if err != nil {
+			return err
+		}
+		defer rows.Close()
+		for rows.Next() {
+			var v service.TaskSummary
+			if err := rows.Scan(&v.ID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.BusinessKey, &v.ConcurrencyKey, &v.Priority, &v.AvailableAt, &v.DeadlineAt, &v.AttemptCount, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain); err != nil {
+				return err
+			}
+			out.Items = append(out.Items, v)
+		}
+		if err := rows.Err(); err != nil {
+			return err
+		}
+		if len(out.Items) > in.Limit {
+			out.Items = out.Items[:in.Limit]
+			last := out.Items[len(out.Items)-1]
+			raw, err := json.Marshal(taskCursor{Version: 2, Namespace: token.NamespaceID, Queue: in.QueueID, State: in.State, Created: last.CreatedAt, ID: last.ID, Filter: filter})
+			if err != nil {
+				return err
+			}
+			out.NextCursor = base64.RawURLEncoding.EncodeToString(raw)
+		}
+		return nil
+	})
+	return out, err
+}
--- a/internal/storage/mysql/task_query_integration_test.go
+++ b/internal/storage/mysql/task_query_integration_test.go
@@ -0,0 +1,270 @@
+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)
+		}
+	})
+}
--- a/internal/storage/mysql/task_store.go
+++ b/internal/storage/mysql/task_store.go
@@ -5,7 +5,6 @@ import (
 	"crypto/sha256"
 	"crypto/subtle"
 	"database/sql"
-	"encoding/base64"
 	"encoding/json"
 	"errors"
 	"time"
@@ -208,6 +207,14 @@ func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in
 	return out, err
 }

+// businessKeyValue 仅把空字符串映射为NULL，不能由连接排序规则把空格折叠为空值。
+func businessKeyValue(key string) *string {
+	if key == "" {
+		return nil
+	}
+	return &key
+}
+
 // insertAcceptedTask 要求调用方已持有应用额度和队列锁，并将身份、事件及审计同事务提交。
 func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID) (service.TaskView, error) {
 	now, err := dbNow(ctx, tx)
@@ -233,7 +240,7 @@ func insertAcceptedTask(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in
 		return service.TaskView{}, err
 	}
 	_, err = tx.ExecContext(ctx, `INSERT INTO pulse_tasks(namespace_id,id,queue_id,handler,handler_version,payload_version,payload,policy_snapshot,recovery_safety,business_key,concurrency_key,concurrency_key_digest,priority,available_at,start_deadline_at,deadline_at,attempt_timeout_ms,max_attempts,created_at,updated_at,retry_of)
-            VALUES (?,?,?,?,?,?,?,?,?,NULLIF(?,''),?,?,?,?,?,?,?,?,?,?,?)`, ns, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, in.BusinessKey, concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now, retryOf)
+            VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, ns, taskID, in.QueueID, in.Handler, in.HandlerVersion, in.PayloadVersion, string(in.Payload), string(policy), in.RecoverySafety, businessKeyValue(in.BusinessKey), concurrencyKeyValue(in.ConcurrencyKey), concurrencyDigest(in.ConcurrencyKey), in.Priority, available, in.StartDeadlineAt, deadline, in.Retry.AttemptTimeoutMS, in.Retry.MaxAttempts, now, now, retryOf)
 	if err != nil {
 		return service.TaskView{}, err
 	}
@@ -283,65 +290,6 @@ func (s *TaskStore) GetTask(ctx context.Context, token auth.Token, id core.ID) (
 	return out, err
 }

-type taskCursor struct {
-	Namespace core.ID         `json:"n"`
-	Queue     core.ID         `json:"q"`
-	State     pulse.TaskState `json:"s"`
-	Created   time.Time       `json:"t"`
-	ID        core.ID         `json:"i"`
-}
-
-func (s *TaskStore) ListTasks(ctx context.Context, token auth.Token, in service.ListInput) (service.TaskPage, error) {
-	out := service.TaskPage{Items: []service.TaskSummary{}}
-	err := s.transaction(ctx, token, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error {
-		if !p.AllowsQueue(in.QueueID) {
-			return fault.Forbidden
-		}
-		q := `SELECT id,queue_id,state,revision,handler,handler_version,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,deadline_at,attempt_count,created_at,updated_at,effect_uncertain FROM pulse_tasks WHERE namespace_id=? AND queue_id=?`
-		args := []any{token.NamespaceID, in.QueueID}
-		if in.State != "" {
-			q += " AND state=?"
-			args = append(args, in.State)
-		}
-		if in.Cursor != "" {
-			raw, err := base64.RawURLEncoding.DecodeString(in.Cursor)
-			var c taskCursor
-			if err != nil || json.Unmarshal(raw, &c) != nil || c.Namespace != token.NamespaceID || c.Queue != in.QueueID || c.State != in.State || c.Created.IsZero() || c.ID.Validate() != nil {
-				return fault.Invalid("cursor does not match this query")
-			}
-			q += " AND (created_at < ? OR (created_at=? AND id<?))"
-			args = append(args, c.Created, c.Created, c.ID)
-		}
-		q += " ORDER BY created_at DESC,id DESC LIMIT ?"
-		args = append(args, in.Limit+1)
-		rows, err := tx.QueryContext(ctx, q, args...)
-		if err != nil {
-			return err
-		}
-		defer rows.Close()
-		for rows.Next() {
-			var v service.TaskSummary
-			if err := rows.Scan(&v.ID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.BusinessKey, &v.ConcurrencyKey, &v.Priority, &v.AvailableAt, &v.DeadlineAt, &v.AttemptCount, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain); err != nil {
-				return err
-			}
-			out.Items = append(out.Items, v)
-		}
-		if err := rows.Err(); err != nil {
-			return err
-		}
-		if len(out.Items) > in.Limit {
-			out.Items = out.Items[:in.Limit]
-			last := out.Items[len(out.Items)-1]
-			raw, err := json.Marshal(taskCursor{token.NamespaceID, in.QueueID, in.State, last.CreatedAt, last.ID})
-			if err != nil {
-				return err
-			}
-			out.NextCursor = base64.RawURLEncoding.EncodeToString(raw)
-		}
-		return nil
-	})
-	return out, err
-}
 func (s *TaskStore) CreateQueue(ctx context.Context, token auth.Token, key string, in service.QueueInput) (service.QueueResult, error) {
 	var out service.QueueResult
 	err := s.transaction(ctx, token, "queues:manage", func(tx *sql.Tx, p auth.Permissions) error {