feat: deliver authenticated task APIs, SDK and P1 regression
This commit is contained in:
156
pulse/service/service.go
Normal file
156
pulse/service/service.go
Normal file
@@ -0,0 +1,156 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
"unicode/utf8"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
type Service struct{ store Store }
|
||||
|
||||
func New(store Store) *Service { return &Service{store: store} }
|
||||
|
||||
var namePattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9_.:-]*$`)
|
||||
|
||||
func ValidName(s string, max int) bool {
|
||||
return len(s) > 0 && len(s) <= max && namePattern.MatchString(s)
|
||||
}
|
||||
func ValidateKey(key string) error {
|
||||
if len(key) < 1 || len(key) > 128 {
|
||||
return fault.Invalid("Idempotency-Key must contain 1 to 128 printable ASCII characters")
|
||||
}
|
||||
for _, c := range key {
|
||||
if c < 33 || c > 126 {
|
||||
return fault.Invalid("Idempotency-Key must contain printable ASCII characters without spaces")
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func NormalizeSubmit(in SubmitInput) (SubmitInput, error) {
|
||||
if in.QueueID.Validate() != nil {
|
||||
return in, fault.Invalid("queue_id is invalid")
|
||||
}
|
||||
if !ValidName(in.Handler, 128) || !ValidName(in.HandlerVersion, 64) || !ValidName(in.PayloadVersion, 32) {
|
||||
return in, fault.Invalid("handler and version names are required ASCII identifiers")
|
||||
}
|
||||
if len(in.Payload) == 0 || len(in.Payload) > 256<<10 || bytes.Equal(bytes.TrimSpace(in.Payload), []byte("null")) {
|
||||
return in, fault.Invalid("payload must be a non-null JSON value up to 256 KiB")
|
||||
}
|
||||
if _, err := jsonvalue.Canonical(in.Payload); err != nil {
|
||||
return in, fault.Invalid("payload is invalid, duplicated, too deep or outside numeric limits")
|
||||
}
|
||||
if !utf8.ValidString(in.BusinessKey) || utf8.RuneCountInString(in.BusinessKey) > 191 || strings.ContainsAny(in.BusinessKey, "\x00\r\n") {
|
||||
return in, fault.Invalid("business_key must be at most 191 characters without control newlines")
|
||||
}
|
||||
if in.TaskTimeoutMS == 0 {
|
||||
in.TaskTimeoutMS = 24 * 60 * 60 * 1000
|
||||
}
|
||||
if in.TaskTimeoutMS > 30*24*60*60*1000 {
|
||||
return in, fault.Invalid("task_timeout_ms exceeds 30 days")
|
||||
}
|
||||
if in.RecoverySafety == "" {
|
||||
in.RecoverySafety = pulse.RecoveryReconcileOnly
|
||||
}
|
||||
if in.RecoverySafety != pulse.RecoveryReconcileOnly && in.RecoverySafety != pulse.RecoveryIdempotent {
|
||||
return in, fault.Invalid("invalid recovery_safety")
|
||||
}
|
||||
if in.Retry.MaxAttempts == 0 {
|
||||
in.Retry.MaxAttempts = 1
|
||||
}
|
||||
if in.Retry.AttemptTimeoutMS == 0 {
|
||||
in.Retry.AttemptTimeoutMS = 30000
|
||||
}
|
||||
if in.Retry.MaxAttempts > 100 || in.Retry.AttemptTimeoutMS > in.TaskTimeoutMS {
|
||||
return in, fault.Invalid("retry attempts or timeout exceed limits")
|
||||
}
|
||||
if in.Retry.MaxAttempts > 1 {
|
||||
if in.Retry.InitialDelayMS == 0 {
|
||||
in.Retry.InitialDelayMS = 1000
|
||||
}
|
||||
if in.Retry.MaxDelayMS == 0 {
|
||||
in.Retry.MaxDelayMS = 30000
|
||||
}
|
||||
}
|
||||
if err := in.Retry.Validate(); err != nil {
|
||||
return in, fault.Invalid("invalid retry policy")
|
||||
}
|
||||
for _, t := range []**time.Time{&in.AvailableAt, &in.StartDeadlineAt} {
|
||||
if *t != nil {
|
||||
v := (**t).UTC()
|
||||
if v.Year() < 2000 || v.Year() > 9998 || v.Nanosecond()%1000 != 0 {
|
||||
return in, fault.Invalid("times must use microsecond precision and supported years")
|
||||
}
|
||||
*t = &v
|
||||
}
|
||||
}
|
||||
return in, nil
|
||||
}
|
||||
func (s *Service) Submit(ctx context.Context, token auth.Token, key string, in SubmitInput) (SubmitResult, error) {
|
||||
if err := ValidateKey(key); err != nil {
|
||||
return SubmitResult{}, err
|
||||
}
|
||||
in, err := NormalizeSubmit(in)
|
||||
if err != nil {
|
||||
return SubmitResult{}, err
|
||||
}
|
||||
return s.store.Submit(ctx, token, key, in)
|
||||
}
|
||||
func (s *Service) GetTask(ctx context.Context, token auth.Token, id core.ID) (TaskView, error) {
|
||||
if id.Validate() != nil {
|
||||
return TaskView{}, fault.Invalid("task id is invalid")
|
||||
}
|
||||
return s.store.GetTask(ctx, token, id)
|
||||
}
|
||||
func (s *Service) ListTasks(ctx context.Context, token auth.Token, in ListInput) (TaskPage, error) {
|
||||
if in.QueueID.Validate() != nil {
|
||||
return TaskPage{}, fault.Invalid("queue_id is required")
|
||||
}
|
||||
if in.Limit == 0 {
|
||||
in.Limit = 20
|
||||
}
|
||||
if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 {
|
||||
return TaskPage{}, fault.Invalid("invalid page limit or cursor")
|
||||
}
|
||||
switch in.State {
|
||||
case "", pulse.TaskWaiting, pulse.TaskLeased, pulse.TaskRunning, pulse.TaskNeedsReview, pulse.TaskSucceeded, pulse.TaskFailed, pulse.TaskCancelled:
|
||||
default:
|
||||
return TaskPage{}, fault.Invalid("invalid task state")
|
||||
}
|
||||
return s.store.ListTasks(ctx, token, in)
|
||||
}
|
||||
func (s *Service) CancelTask(ctx context.Context, token auth.Token, id core.ID, key string, in CancelInput) (CancelResult, error) {
|
||||
if id.Validate() != nil {
|
||||
return CancelResult{}, fault.Invalid("task id is invalid")
|
||||
}
|
||||
if err := ValidateKey(key); err != nil {
|
||||
return CancelResult{}, err
|
||||
}
|
||||
if strings.TrimSpace(in.Reason) == "" || !utf8.ValidString(in.Reason) || utf8.RuneCountInString(in.Reason) > 1024 || in.ExpectedRevision == 0 {
|
||||
return CancelResult{}, fault.Invalid("reason and nonzero expected_revision are required")
|
||||
}
|
||||
return s.store.CancelTask(ctx, token, id, key, in)
|
||||
}
|
||||
func (s *Service) CreateQueue(ctx context.Context, token auth.Token, key string, in QueueInput) (QueueResult, error) {
|
||||
if err := ValidateKey(key); err != nil {
|
||||
return QueueResult{}, err
|
||||
}
|
||||
if !ValidName(in.Name, 128) || in.MaxConcurrency == 0 || in.MaxConcurrency > 100000 || in.RatePerSecond == 0 || in.RatePerSecond > 1000000 {
|
||||
return QueueResult{}, fault.Invalid("queue name, concurrency or rate is invalid")
|
||||
}
|
||||
return s.store.CreateQueue(ctx, token, key, in)
|
||||
}
|
||||
func (s *Service) GetQueue(ctx context.Context, token auth.Token, id core.ID) (QueueView, error) {
|
||||
if id.Validate() != nil {
|
||||
return QueueView{}, fault.Invalid("queue id is invalid")
|
||||
}
|
||||
return s.store.GetQueue(ctx, token, id)
|
||||
}
|
||||
42
pulse/service/service_test.go
Normal file
42
pulse/service/service_test.go
Normal file
@@ -0,0 +1,42 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||||
)
|
||||
|
||||
func TestSubmitNormalizationAndBounds(t *testing.T) {
|
||||
base := SubmitInput{QueueID: core.ID(strings.Repeat("1", 32)), Handler: "file.convert", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{"v":9007199254740993}`)}
|
||||
out, err := NormalizeSubmit(base)
|
||||
if err != nil || out.RecoverySafety != pulse.RecoveryReconcileOnly || out.Retry.MaxAttempts != 1 || out.TaskTimeoutMS != 86400000 {
|
||||
t.Fatalf("defaults %+v %v", out, err)
|
||||
}
|
||||
for name, change := range map[string]func(*SubmitInput){
|
||||
"missing queue": func(v *SubmitInput) { v.QueueID = "" },
|
||||
"missing handler version": func(v *SubmitInput) { v.HandlerVersion = "" },
|
||||
"null payload": func(v *SubmitInput) { v.Payload = json.RawMessage("null") },
|
||||
"oversize payload": func(v *SubmitInput) { v.Payload = json.RawMessage(`"` + strings.Repeat("a", 256<<10) + `"`) },
|
||||
"unbounded attempts": func(v *SubmitInput) { v.Retry.MaxAttempts = 101 },
|
||||
"unknown recovery policy": func(v *SubmitInput) { v.RecoverySafety = "UNKNOWN" },
|
||||
"excess timeout": func(v *SubmitInput) { v.TaskTimeoutMS = 30*86400000 + 1 },
|
||||
"sub-microsecond time": func(v *SubmitInput) { now := time.Date(2026, 1, 1, 0, 0, 0, 1, time.UTC); v.AvailableAt = &now },
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
in := base
|
||||
change(&in)
|
||||
if _, err := NormalizeSubmit(in); err == nil {
|
||||
t.Fatal("accepted invalid input")
|
||||
}
|
||||
})
|
||||
}
|
||||
for _, key := range []string{"", strings.Repeat("a", 129), "with space", "带中文", "line\n"} {
|
||||
if ValidateKey(key) == nil {
|
||||
t.Fatal("accepted invalid key")
|
||||
}
|
||||
}
|
||||
}
|
||||
186
pulse/service/types.go
Normal file
186
pulse/service/types.go
Normal file
@@ -0,0 +1,186 @@
|
||||
// Package service 定义任务应用层契约与校验，不直接操作 HTTP 或 SQL。
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||||
)
|
||||
|
||||
// SubmitInput 是任务受理命令；可变配置在受理时固定为快照。
|
||||
type SubmitInput struct {
|
||||
// QueueID 目标队列，必须位于凭据所属命名空间。
|
||||
QueueID core.ID `json:"queue_id"`
|
||||
// Handler 业务 Worker 的处理器名称。
|
||||
Handler string `json:"handler"`
|
||||
// HandlerVersion 固定处理器版本。
|
||||
HandlerVersion string `json:"handler_version"`
|
||||
// PayloadVersion 输入参数结构版本。
|
||||
PayloadVersion string `json:"payload_version"`
|
||||
// Payload 业务无关 JSON 文档，最大 256 KiB，保留精确数值文本。
|
||||
Payload json.RawMessage `json:"payload"`
|
||||
// BusinessKey 可选关联键，不提供业务效果幂等保证。
|
||||
BusinessKey string `json:"business_key,omitempty"`
|
||||
// AvailableAt 最早执行时间；为空表示受理后即可进入派发范围。
|
||||
AvailableAt *time.Time `json:"available_at,omitempty"`
|
||||
// StartDeadlineAt 最晚开始时间；为空表示使用任务总截止时间。
|
||||
StartDeadlineAt *time.Time `json:"start_deadline_at,omitempty"`
|
||||
// TaskTimeoutMS 从受理时计算的总时限，单位毫秒。
|
||||
TaskTimeoutMS uint64 `json:"task_timeout_ms"`
|
||||
// RecoverySafety 外部效果未知时的恢复安全等级。
|
||||
RecoverySafety pulse.RecoverySafety `json:"recovery_safety"`
|
||||
// Retry 每次执行的有限尝试策略，包含首次尝试。
|
||||
Retry pulse.RetryPolicy `json:"retry"`
|
||||
}
|
||||
|
||||
// TaskView 是对外任务快照，不包含 Worker 令牌或凭据摘要。
|
||||
type TaskView struct {
|
||||
// ID 资源标识，32 位小写十六进制字符串。
|
||||
ID core.ID `json:"id"`
|
||||
// NamespaceID 命名空间标识，由凭据确定隔离范围。
|
||||
NamespaceID core.ID `json:"namespace_id"`
|
||||
// QueueID 队列标识，必须属于当前命名空间。
|
||||
QueueID core.ID `json:"queue_id"`
|
||||
// State 当前任务状态；终态不会被重复请求重开。
|
||||
State pulse.TaskState `json:"state"`
|
||||
// Revision 并发版本，JSON 使用十进制字符串避免跨语言精度丢失。
|
||||
Revision uint64 `json:"revision,string"`
|
||||
// Handler Worker 处理器名称。
|
||||
Handler string `json:"handler"`
|
||||
// HandlerVersion 受理时固定的处理器版本。
|
||||
HandlerVersion string `json:"handler_version"`
|
||||
// PayloadVersion 受理时固定的参数契约版本。
|
||||
PayloadVersion string `json:"payload_version"`
|
||||
// Payload 业务无关 JSON 输入，保留数字精度。
|
||||
Payload json.RawMessage `json:"payload"`
|
||||
// Result 业务 JSON 结果；未取得结果时为空并省略。
|
||||
Result json.RawMessage `json:"result,omitempty"`
|
||||
// BusinessKey 可选业务关联键，不自动提供业务效果幂等。
|
||||
BusinessKey string `json:"business_key,omitempty"`
|
||||
// AvailableAt 最早可执行时间，UTC。
|
||||
AvailableAt time.Time `json:"available_at"`
|
||||
// StartDeadlineAt 最晚开始时间；为空时使用任务总截止时间。
|
||||
StartDeadlineAt *time.Time `json:"start_deadline_at,omitempty"`
|
||||
// DeadlineAt 任务总截止时间，UTC。
|
||||
DeadlineAt time.Time `json:"deadline_at"`
|
||||
// RecoverySafety 未知外部效果的恢复安全等级。
|
||||
RecoverySafety pulse.RecoverySafety `json:"recovery_safety"`
|
||||
// Retry 受理时冻结的尝试次数与时限策略。
|
||||
Retry pulse.RetryPolicy `json:"retry"`
|
||||
// AttemptCount 已创建的尝试次数，包含首次尝试。
|
||||
AttemptCount uint32 `json:"attempt_count"`
|
||||
// CancelRequestedAt 取消意图时间；为空表示未请求取消。
|
||||
CancelRequestedAt *time.Time `json:"cancel_requested_at,omitempty"`
|
||||
// CompletedAt 终结时间；非终态为空。
|
||||
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
||||
// CreatedAt 创建时间，UTC 微秒精度。
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
// UpdatedAt 最后变更时间，不能替代并发版本。
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
type SubmitResult struct {
|
||||
// Task 本次操作的提交回执快照；当前状态应另行查询。
|
||||
Task TaskView `json:"task"`
|
||||
// Deduplicated 是否复用了同一幂等身份的原始回执。
|
||||
Deduplicated bool `json:"deduplicated"`
|
||||
}
|
||||
type CancelInput struct {
|
||||
// Reason 管理操作原因，写入审计历史。
|
||||
Reason string `json:"reason"`
|
||||
// ExpectedRevision 调用者预期的资源版本，JSON 为十进制字符串。
|
||||
ExpectedRevision uint64 `json:"expected_revision,string"`
|
||||
}
|
||||
type CancelResult struct {
|
||||
// Task 本次操作的提交回执快照；当前状态应另行查询。
|
||||
Task TaskView `json:"task"`
|
||||
// Deduplicated 是否复用了同一幂等身份的原始回执。
|
||||
Deduplicated bool `json:"deduplicated"`
|
||||
}
|
||||
type ListInput struct {
|
||||
// QueueID 队列标识，必须属于当前命名空间。
|
||||
QueueID core.ID
|
||||
// State 当前任务状态；终态不会被重复请求重开。
|
||||
State pulse.TaskState
|
||||
// Limit 分页大小，默认 20，上限 100。
|
||||
Limit int
|
||||
// Cursor 上页返回的游标，必须复用相同查询条件。
|
||||
Cursor string
|
||||
}
|
||||
|
||||
// TaskSummary 用于列表，避免在分页中重复传输大参数和结果。
|
||||
type TaskSummary struct {
|
||||
// ID 资源标识，32 位小写十六进制字符串。
|
||||
ID core.ID `json:"id"`
|
||||
// QueueID 队列标识，必须属于当前命名空间。
|
||||
QueueID core.ID `json:"queue_id"`
|
||||
// State 当前任务状态；终态不会被重复请求重开。
|
||||
State pulse.TaskState `json:"state"`
|
||||
// Revision 并发版本，JSON 使用十进制字符串避免跨语言精度丢失。
|
||||
Revision uint64 `json:"revision,string"`
|
||||
// Handler Worker 处理器名称。
|
||||
Handler string `json:"handler"`
|
||||
// HandlerVersion 受理时固定的处理器版本。
|
||||
HandlerVersion string `json:"handler_version"`
|
||||
// BusinessKey 可选业务关联键，不自动提供业务效果幂等。
|
||||
BusinessKey string `json:"business_key,omitempty"`
|
||||
// AvailableAt 最早可执行时间，UTC。
|
||||
AvailableAt time.Time `json:"available_at"`
|
||||
// DeadlineAt 任务总截止时间，UTC。
|
||||
DeadlineAt time.Time `json:"deadline_at"`
|
||||
// AttemptCount 已创建的尝试次数，包含首次尝试。
|
||||
AttemptCount uint32 `json:"attempt_count"`
|
||||
// CreatedAt 创建时间，UTC 微秒精度。
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
// UpdatedAt 最后变更时间，不能替代并发版本。
|
||||
UpdatedAt time.Time `json:"updated_at"`
|
||||
}
|
||||
type TaskPage struct {
|
||||
// Items 当前页任务摘要，不包含大体积参数与结果。
|
||||
Items []TaskSummary `json:"items"`
|
||||
// NextCursor 下一页游标；为空表示当前查询已无下一页。
|
||||
NextCursor string `json:"next_cursor,omitempty"`
|
||||
}
|
||||
type QueueInput struct {
|
||||
// Name 队列名称，同一命名空间内唯一。
|
||||
Name string `json:"name"`
|
||||
// MaxConcurrency 最大同时执行数量，供执行阶段使用。
|
||||
MaxConcurrency uint32 `json:"max_concurrency"`
|
||||
// RatePerSecond 每秒开始数量上限，供执行阶段使用。
|
||||
RatePerSecond uint32 `json:"rate_per_second"`
|
||||
}
|
||||
type QueueView struct {
|
||||
// ID 资源标识，32 位小写十六进制字符串。
|
||||
ID core.ID `json:"id"`
|
||||
// NamespaceID 命名空间标识，由凭据确定隔离范围。
|
||||
NamespaceID core.ID `json:"namespace_id"`
|
||||
// Name 队列名称，同一命名空间内唯一。
|
||||
Name string `json:"name"`
|
||||
// Paused 是否停止新执行；不禁止受理新任务。
|
||||
Paused bool `json:"paused"`
|
||||
// MaxConcurrency 最大同时执行数量，供执行阶段使用。
|
||||
MaxConcurrency uint32 `json:"max_concurrency"`
|
||||
// RatePerSecond 每秒开始数量上限，供执行阶段使用。
|
||||
RatePerSecond uint32 `json:"rate_per_second"`
|
||||
// Revision 并发版本，JSON 使用十进制字符串避免跨语言精度丢失。
|
||||
Revision uint64 `json:"revision,string"`
|
||||
}
|
||||
type QueueResult struct {
|
||||
// Queue 本次操作返回的队列快照。
|
||||
Queue QueueView `json:"queue"`
|
||||
// Deduplicated 是否复用了同一幂等身份的原始回执。
|
||||
Deduplicated bool `json:"deduplicated"`
|
||||
}
|
||||
|
||||
// Store 实现命名空间隔离、事务授权与持久化，不在事务外产生业务副作用。
|
||||
type Store interface {
|
||||
Submit(context.Context, auth.Token, string, SubmitInput) (SubmitResult, error)
|
||||
GetTask(context.Context, auth.Token, core.ID) (TaskView, error)
|
||||
ListTasks(context.Context, auth.Token, ListInput) (TaskPage, error)
|
||||
CancelTask(context.Context, auth.Token, core.ID, string, CancelInput) (CancelResult, error)
|
||||
CreateQueue(context.Context, auth.Token, string, QueueInput) (QueueResult, error)
|
||||
GetQueue(context.Context, auth.Token, core.ID) (QueueView, error)
|
||||
}
|
||||
Reference in New Issue
Block a user