176 lines
7.8 KiB
Go
176 lines
7.8 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"math"
|
|
"time"
|
|
|
|
"git.yuntuo.hk/k2safe/orbit/internal/fault"
|
|
core "git.yuntuo.hk/k2safe/orbit/model"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/auth"
|
|
"git.yuntuo.hk/k2safe/orbit/platform/events"
|
|
"git.yuntuo.hk/k2safe/orbit/pulse/service"
|
|
)
|
|
|
|
// requireActiveQueue 对创建/恢复工作取得共享锁，与退役排他锁串行；历史读取不使用此检查。
|
|
func requireActiveQueue(ctx context.Context, tx *sql.Tx, ns, id core.ID) error {
|
|
var retired *time.Time
|
|
err := tx.QueryRowContext(ctx, `SELECT retired_at FROM pulse_queues WHERE namespace_id=? AND id=? FOR SHARE`, ns, id).Scan(&retired)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return fault.NotFound
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if retired != nil {
|
|
return fault.Retired
|
|
}
|
|
return nil
|
|
}
|
|
func retirementBusy(message string) error { return fault.New("RESOURCE_BUSY", message) }
|
|
|
|
// RetireSubscription 锁定提交水位后锁订阅，既不丢未物化结果，也不丢过期交付。
|
|
// 与创建共用流锁串行化活动名额；消费只锁订阅，不反向锁流/队列。
|
|
func (s *EventStore) RetireSubscription(ctx context.Context, t auth.Token, id core.ID, key string, in events.RetireInput) (events.SubscriptionResult, error) {
|
|
var out events.SubscriptionResult
|
|
err := s.transaction(ctx, t, "events:manage", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if _, err := readSubscription(ctx, tx, t, p, id, false); err != nil {
|
|
return err
|
|
}
|
|
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "subscription.retire", eventScope(t.CredentialID, id), key, "subscription", 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
|
|
}
|
|
head, retained, err := lockEventStream(ctx, tx, t.NamespaceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
before, err := readSubscription(ctx, tx, t, p, id, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if before.RetiredAt != nil {
|
|
return fault.Retired
|
|
}
|
|
if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 {
|
|
return fault.Conflict
|
|
}
|
|
if before.PendingCount != 0 {
|
|
return retirementBusy("subscription has unacknowledged deliveries; drain or resolve expired deliveries first")
|
|
}
|
|
var pending bool
|
|
if err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM orbit_deliveries WHERE namespace_id=? AND subscription_id=? AND state IN ('PENDING','LEASED','EXPIRED'))`, t.NamespaceID, id).Scan(&pending); err != nil {
|
|
return err
|
|
}
|
|
if pending {
|
|
return retirementBusy("subscription has unacknowledged deliveries")
|
|
}
|
|
if before.CursorSequence < retained-1 {
|
|
return events.HistoryExpired
|
|
}
|
|
filter, err := json.Marshal(before.Types)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// 缺失来源事件也拒绝退役，不能将损坏记录误当成不匹配过滤器。
|
|
if err = tx.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM orbit_result_events r LEFT JOIN orbit_events e ON e.namespace_id=r.namespace_id AND e.id=r.id WHERE r.namespace_id=? AND r.queue_id=? AND r.sequence>? AND r.sequence<=? AND (e.id IS NULL OR JSON_CONTAINS(?,JSON_QUOTE(e.type))))`, t.NamespaceID, before.QueueID, before.CursorSequence, head, string(filter)).Scan(&pending); err != nil {
|
|
return err
|
|
}
|
|
if pending {
|
|
return retirementBusy("subscription has unconsumed results; resume consumption and acknowledge them before retirement")
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE orbit_subscriptions SET enabled=FALSE,retired_at=UTC_TIMESTAMP(6),retired_sequence=?,cursor_sequence=?,revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, head, head, t.NamespaceID, id); err != nil {
|
|
return err
|
|
}
|
|
out.Subscription, err = readSubscription(ctx, tx, t, p, id, false)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAudit(ctx, tx, t, identity, id, "subscription", "subscription.retire", in.Reason, before, out.Subscription); err != nil {
|
|
return err
|
|
}
|
|
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
|
})
|
|
return out, err
|
|
}
|
|
|
|
// RetireQueue 按应用额度→身份→队列持锁，拒绝尚未结清的工作和未来工作来源。
|
|
// 暂停计划可以保留；其恢复/发布与本操作共用队列锁，退役后无法重新激活。
|
|
func (s *TaskStore) RetireQueue(ctx context.Context, t auth.Token, id core.ID, key string, in service.RetireQueueInput) (service.QueueResult, error) {
|
|
var out service.QueueResult
|
|
err := s.transaction(ctx, t, "queues:manage", func(tx *sql.Tx, p auth.Permissions) error {
|
|
if !p.AllowsQueue(id) {
|
|
return fault.NotFound
|
|
}
|
|
if _, err := lockApplicationResources(ctx, tx, t.NamespaceID, false); err != nil {
|
|
return err
|
|
}
|
|
identity, cached, err := reserveIdentity(ctx, tx, t.NamespaceID, "queue.retire", string(id), 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
|
|
}
|
|
if _, _, err = lockQueueAfterApplication(ctx, tx, t.NamespaceID, id); err != nil {
|
|
return err
|
|
}
|
|
before, err := readQueue(ctx, tx, t.NamespaceID, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if before.Revision != in.ExpectedRevision || before.Revision == math.MaxUint64 {
|
|
return fault.Conflict
|
|
}
|
|
if before.UnfinishedTasks != 0 {
|
|
return retirementBusy("queue has unfinished tasks, including tasks needing review")
|
|
}
|
|
checks := []struct {
|
|
query, message string
|
|
args []any
|
|
}{
|
|
{`SELECT EXISTS(SELECT 1 FROM pulse_tasks WHERE namespace_id=? AND queue_id=? AND state IN ('WAITING','LEASED','RUNNING','NEEDS_REVIEW'))`, "queue has unfinished or uncertain tasks", []any{t.NamespaceID, id}},
|
|
{`SELECT EXISTS(SELECT 1 FROM pulse_attempts a JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.id=a.task_id WHERE t.namespace_id=? AND t.queue_id=? AND a.state IN ('LEASED','RUNNING'))`, "queue has active execution attempts", []any{t.NamespaceID, id}},
|
|
{`SELECT EXISTS(SELECT 1 FROM pulse_permits p JOIN pulse_attempts a ON a.namespace_id=p.namespace_id AND a.id=p.attempt_id JOIN pulse_tasks t ON t.namespace_id=a.namespace_id AND t.id=a.task_id WHERE t.namespace_id=? AND t.queue_id=? AND p.released_at IS NULL)`, "queue has unreleased execution permits", []any{t.NamespaceID, id}},
|
|
{`SELECT EXISTS(SELECT 1 FROM pulse_schedules WHERE namespace_id=? AND queue_id=? AND paused=FALSE)`, "queue has enabled schedules; pause them first", []any{t.NamespaceID, id}},
|
|
{`SELECT EXISTS(SELECT 1 FROM weave_instances i LEFT JOIN weave_version_scopes v ON v.namespace_id=i.namespace_id AND v.definition_version_id=i.definition_version_id WHERE i.namespace_id=? AND i.state IN ('RUNNING','WAITING','CANCELLING','NEEDS_REVIEW') AND (v.definition_version_id IS NULL OR JSON_CONTAINS(v.queue_ids,JSON_QUOTE(?))))`, "queue is referenced by unfinished workflows", []any{t.NamespaceID, string(id)}},
|
|
{`SELECT EXISTS(SELECT 1 FROM orbit_subscriptions WHERE namespace_id=? AND queue_id=? AND retired_at IS NULL)`, "queue has subscriptions; retire them first", []any{t.NamespaceID, id}},
|
|
}
|
|
for _, check := range checks {
|
|
var busy bool
|
|
if err = tx.QueryRowContext(ctx, check.query, check.args...).Scan(&busy); err != nil {
|
|
return err
|
|
}
|
|
if busy {
|
|
return retirementBusy(check.message)
|
|
}
|
|
}
|
|
if _, err = tx.ExecContext(ctx, `UPDATE pulse_queues SET paused=TRUE,retired_at=UTC_TIMESTAMP(6),revision=revision+1,updated_at=UTC_TIMESTAMP(6) WHERE namespace_id=? AND id=?`, t.NamespaceID, id); err != nil {
|
|
return err
|
|
}
|
|
out.Queue, err = readQueue(ctx, tx, t.NamespaceID, id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAudit(ctx, tx, t, identity, id, "queue", "queue.retire", in.Reason, before, out.Queue); err != nil {
|
|
return err
|
|
}
|
|
return saveIdentity(ctx, tx, t.NamespaceID, identity, out)
|
|
})
|
|
return out, err
|
|
}
|