DenozordecandCursor
f8e0520817
Очередь job получила lease и fencing, телеметрия и прогресс deploy сохраняются, ревизия пишет manifest входов. Обновление префиксов не переписывает неизменённый снимок, большие наборы идут через COPY, а устаревшая ревизия не попадает в deploy. Бандл ноды кэшируется, prune и постановка job больше не держат горячий путь. Co-authored-by: Cursor <cursoragent@cursor.com>
84 lines
2.9 KiB
Go
84 lines
2.9 KiB
Go
package store
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"time"
|
|
)
|
|
|
|
// SystemJobTenantID owns global maintenance jobs outside user tenant scope.
|
|
const SystemJobTenantID = "00000000-0000-0000-0000-000000000000"
|
|
|
|
// JobRequestHash identifies the immutable requested operation, excluding result metadata.
|
|
func JobRequestHash(in JobEnqueue) (string, error) {
|
|
meta := in.Meta
|
|
if in.Request != nil {
|
|
meta = in.Request
|
|
}
|
|
b, err := json.Marshal(struct {
|
|
Kind string `json:"kind"`
|
|
ModuleID *string `json:"module_id"`
|
|
Meta map[string]any `json:"meta"`
|
|
}{in.Kind, in.ModuleID, meta})
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
h := sha256.Sum256(b)
|
|
return hex.EncodeToString(h[:]), nil
|
|
}
|
|
|
|
var (
|
|
ErrJobNotFound = errors.New("store: job not found")
|
|
ErrJobConflict = errors.New("store: idempotency key conflicts with another request")
|
|
ErrJobQueueFull = errors.New("store: job queue is full")
|
|
ErrJobLeaseLost = errors.New("store: job lease lost")
|
|
)
|
|
|
|
// JobRecord is the shared, tenant-scoped job lifecycle record.
|
|
type JobRecord struct {
|
|
RequestHash string
|
|
ID, TenantID, Kind, Status string
|
|
IdempotencyKey, ModuleID *string
|
|
Meta map[string]any
|
|
CreatedAt time.Time
|
|
StartedAt, FinishedAt *time.Time
|
|
LeaseUntil *time.Time
|
|
Error *string
|
|
OwnerID string
|
|
FencingToken int64
|
|
RequestSequence int64
|
|
Attempt int
|
|
CancelRequested bool
|
|
}
|
|
|
|
// JobEnqueue is the immutable request used for idempotent insertion.
|
|
type JobEnqueue struct {
|
|
TenantID, Kind string
|
|
IdempotencyKey, ModuleID *string
|
|
Meta map[string]any
|
|
// Request optionally supplies the original selector before dynamic resolution.
|
|
Request map[string]any
|
|
}
|
|
|
|
// JobQueue provides durable operations; PostgreSQL is authoritative in production.
|
|
type JobQueue interface {
|
|
CountJobsByStatus(context.Context) (map[string]int64, error)
|
|
FindJobByIdempotencyKey(context.Context, string, string) (JobRecord, error)
|
|
EnqueueJob(context.Context, JobEnqueue) (JobRecord, bool, error)
|
|
ClaimJobs(context.Context, string, int, time.Duration) ([]JobRecord, error)
|
|
// ReapExpiredJobs marks orphaned cancellations and exhausted retries as terminal.
|
|
// Production callers run it about every 15 seconds, not on every claim.
|
|
ReapExpiredJobs(context.Context) error
|
|
HeartbeatJob(context.Context, string, string, int64, time.Duration) (bool, error)
|
|
CheckpointJob(context.Context, JobRecord) error
|
|
FinishJob(context.Context, JobRecord) error
|
|
ReleaseJob(context.Context, string, string, int64) error
|
|
GetJob(context.Context, string, string) (JobRecord, error)
|
|
ListJobs(context.Context, string, string, string, string, int) ([]JobRecord, string, bool, error)
|
|
CancelJob(context.Context, string, string) (JobRecord, error)
|
|
RetryJob(context.Context, string, string) (JobRecord, error)
|
|
}
|