Files
orbit/pulse/service/history.go

126 lines
7.1 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package service
import (
"context"
"encoding/json"
"time"
"git.yuntuo.hk/k2safe/orbit/internal/fault"
core "git.yuntuo.hk/k2safe/orbit/model"
"git.yuntuo.hk/k2safe/orbit/platform/auth"
pulse "git.yuntuo.hk/k2safe/orbit/pulse/model"
)
// HistoryPageInput 用于任务执行历史的有界分页;游标必须沿原父资源继续使用。
type HistoryPageInput struct {
Limit int // Limit 默认 20,最大 100;列表不返回业务结果正文。
Cursor string // Cursor 服务端不透明游标,最长 1024 字节;为空从最新记录开始。
}
// AttemptSummary 是执行尝试的查询快照,不提供当前执行授权。
type AttemptSummary struct {
ConcurrencyToken uint64 `json:"concurrency_token,omitempty,string"` // ConcurrencyToken 资源键的单调许可序号;无资源键时为零,不能代替当前授权。
ID core.ID `json:"id"` // ID 尝试标识,不是执行秘密。
TaskID core.ID `json:"task_id"` // TaskID 所属任务;查询始终限定在该任务下。
Epoch uint64 `json:"epoch,string"` // Epoch 单任务执行代次,JSON 为十进制字符串。
State pulse.AttemptState `json:"state"` // State 尝试状态,与整个任务状态分开。
Revision uint64 `json:"revision,string"` // Revision 当前记录版本,可因续租/恢复变化。
LeaseUntil time.Time `json:"lease_until"` // LeaseUntil 数据库记录的租约期限;查询不能授权业务执行。
AuthorizedAt *time.Time `json:"authorized_at,omitempty"` // AuthorizedAt 开始批准时间;空表示未批准,不证明副作用已经发生。
EndedAt *time.Time `json:"ended_at,omitempty"` // EndedAt 尝试结束时间;仍在执行时为空。
Retryable bool `json:"retryable,omitempty"` // Retryable 该明确失败报告是否声明可安全重试;不代表平台必定安排重试。
ErrorCode string `json:"error_code,omitempty"` // ErrorCode 稳定错误分类;无错误时为空。
CreatedAt time.Time `json:"created_at"` // CreatedAt 创建时间,UTC 微秒精度。
UpdatedAt time.Time `json:"updated_at"` // UpdatedAt 最新变更时间,不能代替版本校验。
}
// AttemptView 增加当前归属与已保存结果,不返回令牌、摘要或原始请求秘密。
type AttemptView struct {
AttemptSummary // AttemptSummary 尝试状态、执行代次与时间快照。
SessionID core.ID `json:"session_id"` // SessionID 当前或最后归属会话;重连迁移后可变化。
WorkerInstanceID core.ID `json:"worker_instance_id"` // WorkerInstanceID 对应进程身份,不是认证凭据。
Result json.RawMessage `json:"result,omitempty"` // Result 已保存的尝试结果,未保存时省略;不包含迟到证据。
}
// AttemptPage 返回一页尝试摘要及可选的后续位置。
type AttemptPage struct {
Items []AttemptSummary `json:"items"` // Items 按 Epoch 降序排列的摘要,空列表返回 []。
NextCursor string `json:"next_cursor,omitempty"` // NextCursor 后续页游标,空表示本次已读到末尾。
}
// EvidenceSummary 描述追加证据,原报告并未改变权威任务结果。
type EvidenceSummary struct {
ID core.ID `json:"id"` // ID 不可变证据标识。
TaskID core.ID `json:"task_id"` // TaskID 所属任务,从已验证的父关联确定。
AttemptID core.ID `json:"attempt_id"` // AttemptID 所属执行尝试。
Outcome string `json:"outcome"` // Outcome Worker 声明的 SUCCEEDED、FAILED 或 UNKNOWN,不代表平台认可。
ErrorCode string `json:"error_code,omitempty"` // ErrorCode 原报告的稳定分类,可为空。
Retryable bool `json:"retryable,omitempty"` // Retryable 原报告的重试声明;证据不触发重试。
Reason string `json:"reason"` // Reason LEASE_LOST、NOT_STARTED 或 RESULT_CONFLICT。
Disposition string `json:"disposition"` // Disposition 固定为 EVIDENCE_ONLY,表示采集时仅保存证据。
CreatedAt time.Time `json:"created_at"` // CreatedAt 首次收到时间,重复报告不改变它。
}
// EvidenceView 提供单条追加证据的原报告身份和正文。
type EvidenceView struct {
EvidenceSummary // EvidenceSummary 原报告分类及采集时间。
MessageID core.ID `json:"message_id"` // MessageID 原结果报告身份,用于追踪同一次重报。
Payload json.RawMessage `json:"payload"` // Payload 原始结果/证据引用 JSON,保留数字文本;不提升为任务结果。
}
// EvidencePage 返回一页证据摘要,正文通过详情单独读取。
type EvidencePage struct {
Items []EvidenceSummary `json:"items"` // Items 按 created_at、ID 降序的证据摘要,不含大正文。
NextCursor string `json:"next_cursor,omitempty"` // NextCursor 必须沿同一任务及尝试继续使用。
}
func normalizeHistoryPage(in HistoryPageInput) (HistoryPageInput, error) {
if in.Limit == 0 {
in.Limit = 20
}
if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 {
return in, fault.Invalid("invalid history page limit or cursor")
}
return in, nil
}
func validateHistoryIDs(ids ...core.ID) error {
for _, id := range ids {
if id.Validate() != nil {
return fault.Invalid("invalid history resource id")
}
}
return nil
}
func (s *Service) ListTaskAttempts(ctx context.Context, t auth.Token, task core.ID, in HistoryPageInput) (AttemptPage, error) {
if err := validateHistoryIDs(task); err != nil {
return AttemptPage{}, err
}
in, err := normalizeHistoryPage(in)
if err != nil {
return AttemptPage{}, err
}
return s.store.ListTaskAttempts(ctx, t, task, in)
}
func (s *Service) GetTaskAttempt(ctx context.Context, t auth.Token, task, attempt core.ID) (AttemptView, error) {
if err := validateHistoryIDs(task, attempt); err != nil {
return AttemptView{}, err
}
return s.store.GetTaskAttempt(ctx, t, task, attempt)
}
func (s *Service) ListAttemptEvidence(ctx context.Context, t auth.Token, task, attempt core.ID, in HistoryPageInput) (EvidencePage, error) {
if err := validateHistoryIDs(task, attempt); err != nil {
return EvidencePage{}, err
}
in, err := normalizeHistoryPage(in)
if err != nil {
return EvidencePage{}, err
}
return s.store.ListAttemptEvidence(ctx, t, task, attempt, in)
}
func (s *Service) GetAttemptEvidence(ctx context.Context, t auth.Token, task, attempt, evidence core.ID) (EvidenceView, error) {
if err := validateHistoryIDs(task, attempt, evidence); err != nil {
return EvidenceView{}, err
}
return s.store.GetAttemptEvidence(ctx, t, task, attempt, evidence)
}