386 lines
16 KiB
Go
386 lines
16 KiB
Go
package mysql
|
||
|
||
import (
|
||
"context"
|
||
"crypto/sha256"
|
||
"crypto/subtle"
|
||
"database/sql"
|
||
"encoding/json"
|
||
"errors"
|
||
"time"
|
||
|
||
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
|
||
"git.yuntuo.hk/k2safe/orbit/internal/jsonwire"
|
||
core "git.yuntuo.hk/k2safe/orbit/model"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
||
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
||
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
|
||
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
||
driver "github.com/go-sql-driver/mysql"
|
||
)
|
||
|
||
// TaskStore 在单个 MySQL 事务中执行权限、幂等与任务状态写入。
|
||
type TaskStore struct{ db *sql.DB }
|
||
|
||
func NewTaskStore(db *sql.DB) *TaskStore { return &TaskStore{db: db} }
|
||
|
||
var _ service.Store = (*TaskStore)(nil)
|
||
|
||
// AuthorizeSubmit 用于批量请求的只读预认证；每项事务仍独立校验凭据、动作与队列范围。
|
||
func (s *TaskStore) AuthorizeSubmit(ctx context.Context, token auth.Token) error {
|
||
return s.transaction(ctx, token, "tasks:submit", func(*sql.Tx, auth.Permissions) error { return nil })
|
||
}
|
||
|
||
func (s *TaskStore) transaction(ctx context.Context, token auth.Token, action string, fn func(*sql.Tx, auth.Permissions) error) error {
|
||
return WithTx(ctx, s.db, func(tx *sql.Tx) error {
|
||
p, err := authenticate(ctx, tx, token)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !p.Allows(action) {
|
||
return fault.Forbidden
|
||
}
|
||
return fn(tx, p)
|
||
})
|
||
}
|
||
func authenticate(ctx context.Context, tx *sql.Tx, token auth.Token) (auth.Permissions, error) {
|
||
var digest core.Digest
|
||
var raw []byte
|
||
err := tx.QueryRowContext(ctx, `SELECT c.secret_digest,c.permissions FROM orbit_credentials c
|
||
JOIN orbit_namespaces n ON n.id=c.namespace_id JOIN orbit_applications a ON a.id=n.application_id
|
||
WHERE c.namespace_id=? AND c.id=? AND c.revoked_at IS NULL
|
||
AND (c.expires_at IS NULL OR c.expires_at>UTC_TIMESTAMP(6)) AND n.enabled=TRUE AND a.enabled=TRUE FOR SHARE`, token.NamespaceID, token.CredentialID).Scan(&digest, &raw)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
return auth.Permissions{}, fault.Unauthorized
|
||
}
|
||
if err != nil {
|
||
return auth.Permissions{}, err
|
||
}
|
||
if subtle.ConstantTimeCompare(digest[:], token.Digest[:]) != 1 {
|
||
// 控制台会话使用独立的短期令牌摘要，不轮换也不覆盖业务 SDK
|
||
// 凭据。只有经过会话表、账号、命名空间和应用启停状态联合校验，
|
||
// 才允许沿用该业务凭据的权限快照。
|
||
var sessionPermissions []byte
|
||
sessionErr := tx.QueryRowContext(ctx, `SELECT c.permissions
|
||
FROM orbit_console_sessions s
|
||
JOIN orbit_console_users u ON u.id=s.user_id AND u.enabled=TRUE
|
||
JOIN orbit_credentials c ON c.namespace_id=s.namespace_id AND c.id=s.credential_id
|
||
JOIN orbit_namespaces n ON n.id=s.namespace_id AND n.enabled=TRUE
|
||
JOIN orbit_applications a ON a.id=n.application_id AND a.enabled=TRUE
|
||
WHERE s.namespace_id=? AND s.credential_id=? AND s.token_digest=?
|
||
AND s.revoked_at IS NULL AND s.expires_at>UTC_TIMESTAMP(6)
|
||
AND c.revoked_at IS NULL AND (c.expires_at IS NULL OR c.expires_at>UTC_TIMESTAMP(6))
|
||
FOR SHARE`, token.NamespaceID, token.CredentialID, token.Digest).Scan(&sessionPermissions)
|
||
if errors.Is(sessionErr, sql.ErrNoRows) {
|
||
return auth.Permissions{}, fault.Unauthorized
|
||
}
|
||
if sessionErr != nil {
|
||
return auth.Permissions{}, sessionErr
|
||
}
|
||
raw = sessionPermissions
|
||
}
|
||
var p auth.Permissions
|
||
if json.Unmarshal(raw, &p) != nil || p.Validate() != nil {
|
||
return p, fault.Unauthorized
|
||
}
|
||
return p, nil
|
||
}
|
||
func requestDigest(value any) (core.Digest, error) {
|
||
raw, err := json.Marshal(value)
|
||
if err != nil {
|
||
return core.Digest{}, err
|
||
}
|
||
raw, err = jsonvalue.Canonical(raw)
|
||
if err != nil {
|
||
return core.Digest{}, err
|
||
}
|
||
return core.Digest(sha256.Sum256(append([]byte("orbit-json-v1\x00"), raw...))), nil
|
||
}
|
||
|
||
// reserveIdentity 通过唯一键和行锁串行化同一请求。占位记录、资源及最终响应一起提交。
|
||
// 事务失败整体回滚；提交结果不明时调用方仍必须使用原键重报。
|
||
func reserveIdentity(ctx context.Context, tx *sql.Tx, ns core.ID, operation, scope, key, kind string, resource core.ID, value any) (core.ID, []byte, error) {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
digest, err := requestDigest(value)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
keyDigest := core.Digest(sha256.Sum256([]byte(key)))
|
||
_, err = tx.ExecContext(ctx, `INSERT INTO orbit_idempotency_keys(namespace_id,id,operation,scope_key,key_digest,request_digest,resource_kind,resource_id,response,expires_at)
|
||
VALUES (?,?,?,?,?,?,?,?, '{}',DATE_ADD(UTC_TIMESTAMP(6),INTERVAL 90 DAY))
|
||
ON DUPLICATE KEY UPDATE id=orbit_idempotency_keys.id`, ns, id, operation, scope, keyDigest, digest, kind, resource)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
var storedID core.ID
|
||
var storedDigest core.Digest
|
||
var response []byte
|
||
err = tx.QueryRowContext(ctx, `SELECT id,request_digest,response FROM orbit_idempotency_keys
|
||
WHERE namespace_id=? AND operation=? AND scope_key=? AND key_digest=? FOR UPDATE`, ns, operation, scope, keyDigest).Scan(&storedID, &storedDigest, &response)
|
||
if err != nil {
|
||
return "", nil, err
|
||
}
|
||
if subtle.ConstantTimeCompare(digest[:], storedDigest[:]) != 1 {
|
||
return "", nil, fault.IdempotencyConflict
|
||
}
|
||
if storedID == id {
|
||
return id, nil, nil
|
||
}
|
||
return storedID, response, nil
|
||
}
|
||
func saveIdentity(ctx context.Context, tx *sql.Tx, ns, id core.ID, value any) error {
|
||
raw, err := jsonwire.Marshal(value)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, "UPDATE orbit_idempotency_keys SET response=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?", string(raw), ns, id)
|
||
return err
|
||
}
|
||
func emitEvent(ctx context.Context, tx *sql.Tx, ns, resource core.ID, revision uint64, kind, event string, payload any) error {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
raw, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = tx.ExecContext(ctx, `INSERT INTO orbit_events(namespace_id,id,aggregate_kind,aggregate_id,aggregate_revision,type,payload,occurred_at) VALUES (?,?,?,?,?,?,?,UTC_TIMESTAMP(6))`, ns, id, kind, resource, revision, event, string(raw))
|
||
if err == nil && kind == "task" && events.ResultType(event) {
|
||
return appendResultEvent(ctx, tx, ns, id, resource, revision, event)
|
||
}
|
||
return err
|
||
}
|
||
func appendAudit(ctx context.Context, tx *sql.Tx, token auth.Token, operation, resource core.ID, kind, action, reason string, before, after any) error {
|
||
_, err := appendAuditWithEvidence(ctx, tx, token, operation, resource, kind, action, reason, "", before, after)
|
||
return err
|
||
}
|
||
|
||
func appendAuditWithEvidence(ctx context.Context, tx *sql.Tx, token auth.Token, operation, resource core.ID, kind, action, reason, evidence string, before, after any) (core.ID, error) {
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
old, err := json.Marshal(before)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
current, err := json.Marshal(after)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO orbit_audit_events(namespace_id,id,operation_id,actor_id,action,resource_kind,resource_id,reason,evidence_ref,`before`,`after`) VALUES (?,?,?,?,?,?,?,?,NULLIF(?,''),?,?)", token.NamespaceID, id, operation, string(token.CredentialID), action, kind, resource, reason, evidence, string(old), string(current))
|
||
return id, err
|
||
}
|
||
func dbNow(ctx context.Context, tx *sql.Tx) (time.Time, error) {
|
||
var now time.Time
|
||
err := tx.QueryRowContext(ctx, "SELECT UTC_TIMESTAMP(6)").Scan(&now)
|
||
return now, err
|
||
}
|
||
|
||
func (s *TaskStore) Submit(ctx context.Context, token auth.Token, key string, in service.SubmitInput) (service.SubmitResult, error) {
|
||
var out service.SubmitResult
|
||
err := s.transaction(ctx, token, "tasks:submit", func(tx *sql.Tx, p auth.Permissions) error {
|
||
if !p.AllowsQueue(in.QueueID) {
|
||
return fault.Forbidden
|
||
}
|
||
// 先按既有应用额度锁串行进入受理，再占用幂等唯一键。
|
||
// 满额回滚时，同键并发INSERT的重复检查锁不能与应用额度锁互相等待。
|
||
// 此处只取得锁，不检查/扣减额度；已有回执在满额时仍可返回。
|
||
if _, err := lockApplicationResources(ctx, tx, token.NamespaceID, false); err != nil {
|
||
return err
|
||
}
|
||
taskID, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
identity, cached, err := reserveIdentity(ctx, tx, token.NamespaceID, "task.submit", "", key, "task", taskID, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err := json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
if _, _, err := lockQueueAfterApplication(ctx, tx, token.NamespaceID, in.QueueID); err != nil {
|
||
return err
|
||
}
|
||
out.Task, err = insertAcceptedTask(ctx, tx, token.NamespaceID, taskID, in, nil)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := emitEvent(ctx, tx, token.NamespaceID, taskID, 1, "task", "task.accepted", map[string]any{"task_id": taskID, "state": pulse.TaskWaiting}); err != nil {
|
||
return err
|
||
}
|
||
if err := appendAudit(ctx, tx, token, identity, taskID, "task", "task.submit", "submit task", nil, map[string]any{"state": pulse.TaskWaiting, "queue_id": in.QueueID, "handler": in.Handler, "handler_version": in.HandlerVersion}); err != nil {
|
||
return err
|
||
}
|
||
return saveIdentity(ctx, tx, token.NamespaceID, identity, out)
|
||
})
|
||
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) {
|
||
return insertAcceptedTaskUntil(ctx, tx, ns, taskID, in, retryOf, nil)
|
||
}
|
||
|
||
// insertAcceptedTaskUntil 在原任务时限基础上收紧绝对父截止，不重写冻结重试策略。
|
||
func insertAcceptedTaskUntil(ctx context.Context, tx *sql.Tx, ns, taskID core.ID, in service.SubmitInput, retryOf *core.ID, parentDeadline *time.Time) (service.TaskView, error) {
|
||
now, err := dbNow(ctx, tx)
|
||
if err != nil {
|
||
return service.TaskView{}, err
|
||
}
|
||
// 相对时长保留在请求摘要中，只在首次受理时按数据库时钟转换；重报先返回原回执。
|
||
available := now.Add(time.Duration(in.DelayMS) * time.Millisecond)
|
||
if in.AvailableAt != nil {
|
||
available = *in.AvailableAt
|
||
}
|
||
deadline := now.Add(time.Duration(in.TaskTimeoutMS) * time.Millisecond)
|
||
if parentDeadline != nil {
|
||
deadline = minimumTime(deadline, *parentDeadline)
|
||
}
|
||
if available.Before(now.Add(-time.Second)) || !available.Before(deadline) {
|
||
return service.TaskView{}, fault.Invalid("available_at must be current/future and before the task deadline")
|
||
}
|
||
if in.StartDeadlineAt != nil && (!in.StartDeadlineAt.After(now) || in.StartDeadlineAt.Before(available) || in.StartDeadlineAt.After(deadline)) {
|
||
return service.TaskView{}, fault.Invalid("start_deadline_at must be after now, at or after available_at, and no later than the task deadline")
|
||
}
|
||
policy, err := json.Marshal(in.Retry)
|
||
if err != nil {
|
||
return service.TaskView{}, err
|
||
}
|
||
if err := reserveQueueTask(ctx, tx, ns, in.QueueID); err != nil {
|
||
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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, 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
|
||
}
|
||
return readTask(ctx, tx, ns, taskID, false)
|
||
}
|
||
|
||
const taskColumns = `id,namespace_id,queue_id,state,revision,handler,handler_version,payload_version,payload,result,COALESCE(business_key,''),COALESCE(concurrency_key,''),priority,available_at,start_deadline_at,deadline_at,recovery_safety,policy_snapshot,attempt_count,cancel_requested_at,completed_at,created_at,updated_at,effect_uncertain,retry_of`
|
||
|
||
type scanner interface{ Scan(...any) error }
|
||
|
||
func scanTask(row scanner) (service.TaskView, error) {
|
||
var v service.TaskView
|
||
var policy, payload, result []byte
|
||
err := row.Scan(&v.ID, &v.NamespaceID, &v.QueueID, &v.State, &v.Revision, &v.Handler, &v.HandlerVersion, &v.PayloadVersion, &payload, &result, &v.BusinessKey, &v.ConcurrencyKey, &v.Priority, &v.AvailableAt, &v.StartDeadlineAt, &v.DeadlineAt, &v.RecoverySafety, &policy, &v.AttemptCount, &v.CancelRequestedAt, &v.CompletedAt, &v.CreatedAt, &v.UpdatedAt, &v.EffectUncertain, &v.RetryOf)
|
||
if err != nil {
|
||
return v, err
|
||
}
|
||
v.Payload = payload
|
||
v.Result = result
|
||
err = json.Unmarshal(policy, &v.Retry)
|
||
return v, err
|
||
}
|
||
func readTask(ctx context.Context, tx *sql.Tx, ns, id core.ID, lock bool) (service.TaskView, error) {
|
||
q := "SELECT " + taskColumns + " FROM pulse_tasks WHERE namespace_id=? AND id=?"
|
||
if lock {
|
||
q += " FOR UPDATE"
|
||
}
|
||
v, err := scanTask(tx.QueryRowContext(ctx, q, ns, id))
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
err = fault.NotFound
|
||
}
|
||
return v, err
|
||
}
|
||
func (s *TaskStore) GetTask(ctx context.Context, token auth.Token, id core.ID) (service.TaskView, error) {
|
||
var out service.TaskView
|
||
err := s.transaction(ctx, token, "tasks:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||
var err error
|
||
out, err = readTask(ctx, tx, token.NamespaceID, id, false)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if !p.AllowsQueue(out.QueueID) {
|
||
return fault.NotFound
|
||
}
|
||
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 {
|
||
if !p.AllowsQueue("") {
|
||
return fault.Forbidden
|
||
}
|
||
id, err := core.NewID()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
identity, cached, err := reserveIdentity(ctx, tx, token.NamespaceID, "queue.create", "", key, "queue", id, in)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if cached != nil {
|
||
if err := json.Unmarshal(cached, &out); err != nil {
|
||
return err
|
||
}
|
||
out.Deduplicated = true
|
||
return nil
|
||
}
|
||
maxUnfinished := in.MaxUnfinishedTasks
|
||
if maxUnfinished == 0 {
|
||
maxUnfinished = 100000
|
||
}
|
||
_, err = tx.ExecContext(ctx, "INSERT INTO pulse_queues(namespace_id,id,name,max_concurrency,rate_per_second,max_unfinished_tasks,policy) VALUES (?,?,?,?,?,?,'{}')", token.NamespaceID, id, in.Name, in.MaxConcurrency, in.RatePerSecond, maxUnfinished)
|
||
var dbErr *driver.MySQLError
|
||
if errors.As(err, &dbErr) && dbErr.Number == 1062 {
|
||
return fault.Conflict
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
out.Queue, err = readQueue(ctx, tx, token.NamespaceID, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if err := appendAudit(ctx, tx, token, identity, id, "queue", "queue.create", "create queue", nil, out.Queue); err != nil {
|
||
return err
|
||
}
|
||
return saveIdentity(ctx, tx, token.NamespaceID, identity, out)
|
||
})
|
||
return out, err
|
||
}
|
||
func readQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) (service.QueueView, error) {
|
||
var q service.QueueView
|
||
err := tx.QueryRowContext(ctx, "SELECT id,namespace_id,name,paused,max_concurrency,rate_per_second,max_unfinished_tasks,unfinished_tasks,revision FROM pulse_queues WHERE namespace_id=? AND id=?", ns, id).Scan(&q.ID, &q.NamespaceID, &q.Name, &q.Paused, &q.MaxConcurrency, &q.RatePerSecond, &q.MaxUnfinishedTasks, &q.UnfinishedTasks, &q.Revision)
|
||
if errors.Is(err, sql.ErrNoRows) {
|
||
err = fault.NotFound
|
||
}
|
||
return q, err
|
||
}
|
||
func (s *TaskStore) GetQueue(ctx context.Context, token auth.Token, id core.ID) (service.QueueView, error) {
|
||
var out service.QueueView
|
||
err := s.transaction(ctx, token, "queues:read", func(tx *sql.Tx, p auth.Permissions) error {
|
||
if !p.AllowsQueue(id) {
|
||
return fault.NotFound
|
||
}
|
||
var err error
|
||
out, err = readQueue(ctx, tx, token.NamespaceID, id)
|
||
return err
|
||
})
|
||
return out, err
|
||
}
|