128 lines
4.9 KiB
Go
128 lines
4.9 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
)
|
||
|
||
// MaxBatchTasks 限制一次请求的独立任务数；HTTP 总请求仍限制为512 KiB。
|
||
const MaxBatchTasks = 32
|
||
|
||
type BatchSubmitInput struct {
|
||
// Items 按输入顺序分别受理，不保证整批原子性。
|
||
Items []BatchSubmitItem `json:"items"`
|
||
}
|
||
|
||
type BatchSubmitItem struct {
|
||
// IdempotencyKey 必须由调用方持久化，与单项提交共用命名空间内的幂等身份。
|
||
IdempotencyKey string `json:"idempotency_key"`
|
||
// Task 与单项受理完全相同的输入，重报时保留原值。
|
||
Task SubmitInput `json:"task"`
|
||
}
|
||
|
||
type BatchItemStatus string
|
||
|
||
const (
|
||
BatchAccepted BatchItemStatus = "ACCEPTED"
|
||
BatchRejected BatchItemStatus = "REJECTED"
|
||
BatchUnknown BatchItemStatus = "UNKNOWN"
|
||
BatchNotAttempted BatchItemStatus = "NOT_ATTEMPTED"
|
||
)
|
||
|
||
// BatchTaskReceipt 是原始受理回执的精简视图，避免重复返回大参数。
|
||
type BatchTaskReceipt struct {
|
||
// TaskID 已持久化的任务标识，业务结果需另行查询或订阅。
|
||
TaskID core.ID `json:"task_id"`
|
||
// QueueID 原始回执所属的队列标识。
|
||
QueueID core.ID `json:"queue_id"`
|
||
// Revision 原始受理版本，十进制字符串，不代表当前任务版本。
|
||
Revision uint64 `json:"revision,string"`
|
||
// CreatedAt 原始受理时间，UTC微秒精度。
|
||
CreatedAt time.Time `json:"created_at"`
|
||
// Deduplicated 本项是否返回已有幂等回执。
|
||
Deduplicated bool `json:"deduplicated"`
|
||
}
|
||
|
||
type BatchItemError struct {
|
||
// Code 稳定错误码；UNKNOWN不能被视为确认回滚。
|
||
Code string `json:"code"`
|
||
// Message 对外说明，不包含数据库错误或任务输入。
|
||
Message string `json:"message"`
|
||
// RetryAfterMS 建议最小退避毫秒数，省略为未指定，不代表可更换幂等键。
|
||
RetryAfterMS uint32 `json:"retry_after_ms,omitempty"`
|
||
}
|
||
|
||
type BatchSubmitItemResult struct {
|
||
// Index 本次请求内从0开始的索引，结果始终与输入同序且数量相同。
|
||
Index int `json:"index"`
|
||
// Status 区分已受理、明确拒绝、结果未知及本次未尝试。
|
||
Status BatchItemStatus `json:"status"`
|
||
// Receipt 仅ACCEPTED包含原始受理回执。
|
||
Receipt *BatchTaskReceipt `json:"receipt,omitempty"`
|
||
// Error 仅非ACCEPTED包含稳定错误及退避提示。
|
||
Error *BatchItemError `json:"error,omitempty"`
|
||
}
|
||
|
||
type BatchSubmitResult struct {
|
||
// Items 每项独立结果，HTTP成功不能替代逐项检查。
|
||
Items []BatchSubmitItemResult `json:"items"`
|
||
}
|
||
|
||
// SubmitBatch 逐项重用单项事务；不建立整批幂等身份或跨项锁。
|
||
func (s *Service) SubmitBatch(ctx context.Context, token auth.Token, in BatchSubmitInput) (BatchSubmitResult, error) {
|
||
if len(in.Items) < 1 || len(in.Items) > MaxBatchTasks {
|
||
return BatchSubmitResult{}, fault.Invalid("items must contain 1 to 32 tasks")
|
||
}
|
||
keys := make(map[string]struct{}, len(in.Items))
|
||
for _, item := range in.Items {
|
||
if err := ValidateKey(item.IdempotencyKey); err != nil {
|
||
return BatchSubmitResult{}, err
|
||
}
|
||
if _, exists := keys[item.IdempotencyKey]; exists {
|
||
return BatchSubmitResult{}, fault.Invalid("item idempotency keys must be distinct within a batch")
|
||
}
|
||
keys[item.IdempotencyKey] = struct{}{}
|
||
}
|
||
// 即使全部任务参数无效，也必须先验证凭据和动作权限；每项写入会再次认证。
|
||
if err := s.store.AuthorizeSubmit(ctx, token); err != nil {
|
||
return BatchSubmitResult{}, err
|
||
}
|
||
out := BatchSubmitResult{Items: make([]BatchSubmitItemResult, len(in.Items))}
|
||
stopped := false
|
||
for i, item := range in.Items {
|
||
result := BatchSubmitItemResult{Index: i}
|
||
if stopped || ctx.Err() != nil {
|
||
result.Status = BatchNotAttempted
|
||
result.Error = &BatchItemError{Code: "BATCH_STOPPED", Message: "this item was not attempted in this request; retry using the original identity", RetryAfterMS: 1000}
|
||
} else {
|
||
accepted, err := s.Submit(ctx, token, item.IdempotencyKey, item.Task)
|
||
if err == nil {
|
||
result.Status = BatchAccepted
|
||
result.Receipt = &BatchTaskReceipt{TaskID: accepted.Task.ID, QueueID: accepted.Task.QueueID, Revision: accepted.Task.Revision, CreatedAt: accepted.Task.CreatedAt, Deduplicated: accepted.Deduplicated}
|
||
} else {
|
||
var public *fault.Error
|
||
if !errors.As(err, &public) || public.Code == "UNAVAILABLE" {
|
||
// 驱动超时、COMMIT响应丢失等不能证明未提交，停止继续占用数据库。
|
||
public = fault.Unavailable
|
||
result.Status, stopped = BatchUnknown, true
|
||
} else {
|
||
result.Status = BatchRejected
|
||
stopped = public.Code == "UNAUTHENTICATED"
|
||
}
|
||
result.Error = &BatchItemError{Code: public.Code, Message: public.Message}
|
||
switch public.Code {
|
||
case "QUEUE_FULL", "APPLICATION_FULL", "UNAVAILABLE":
|
||
result.Error.RetryAfterMS = 1000
|
||
}
|
||
}
|
||
}
|
||
out.Items[i] = result
|
||
}
|
||
return out, nil
|
||
}
|