From e65cf0d9582f80c28c0abb86b09d9ce8367cb972 Mon Sep 17 00:00:00 2001 From: Denozordec Date: Mon, 25 May 2026 10:33:57 +0700 Subject: [PATCH] feat(jobs): persist job lifecycle to PostgreSQL job_audit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit UpsertQueued/Running/MarkTerminal через SetPersistHooks; исправлен deadlock fireEnqueued под Registry mutex. Co-authored-by: Cursor --- internal/httpapi/bootstrap.go | 40 ++++++++++++++++++------ internal/jobs/job.go | 53 +++++++++++++++++++++++++++++--- internal/jobs/worker.go | 3 ++ internal/repository/job_audit.go | 25 +++++++++++++-- 4 files changed, 106 insertions(+), 15 deletions(-) diff --git a/internal/httpapi/bootstrap.go b/internal/httpapi/bootstrap.go index ce6ea1e..28b02b3 100644 --- a/internal/httpapi/bootstrap.go +++ b/internal/httpapi/bootstrap.go @@ -56,18 +56,40 @@ func BootstrapWorkers(ctx context.Context, opts Options) (store.Backend, *jobs.R wk.Registry = reg if pool != nil { audit := repository.NewJobAuditWriter(pool) - reg.SetTerminalHook(func(j *jobs.Job) { + jobMeta := func(j *jobs.Job) map[string]any { if j == nil { - return + return nil } st := j.Snapshot() - status, _ := st["status"].(string) - var errMsg *string - if e, ok := st["error"].(string); ok && e != "" { - errMsg = &e - } - audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC()) - }) + meta, _ := st["meta"].(map[string]any) + return meta + } + reg.SetPersistHooks( + func(j *jobs.Job) { + if j == nil { + return + } + audit.UpsertQueued(context.Background(), j.TenantID, j.ID, j.Kind, j.IdempotencyKey, j.ModuleID, jobMeta(j)) + }, + func(j *jobs.Job) { + if j == nil { + return + } + audit.UpsertRunning(context.Background(), j.TenantID, j.ID, j.Kind, j.IdempotencyKey, jobMeta(j)) + }, + func(j *jobs.Job) { + if j == nil { + return + } + st := j.Snapshot() + status, _ := st["status"].(string) + var errMsg *string + if e, ok := st["error"].(string); ok && e != "" { + errMsg = &e + } + audit.MarkTerminal(context.Background(), j.TenantID, j.ID, status, errMsg, time.Now().UTC()) + }, + ) } observability.RegisterStoreBackend(backend) return backend, reg, pool, nil diff --git a/internal/jobs/job.go b/internal/jobs/job.go index 0628c65..513db7a 100644 --- a/internal/jobs/job.go +++ b/internal/jobs/job.go @@ -182,6 +182,8 @@ type Registry struct { workerStart func(j *Job) workerSem chan struct{} onTerminal func(j *Job) + onEnqueued func(j *Job) + onRunning func(j *Job) } type idempoKey struct { @@ -209,6 +211,44 @@ func (r *Registry) SetTerminalHook(fn func(j *Job)) { r.onTerminal = fn } +// SetPersistHooks registers best-effort callbacks for job lifecycle persistence. +func (r *Registry) SetPersistHooks(onEnqueued, onRunning, onTerminal func(j *Job)) { + if r == nil { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.onEnqueued = onEnqueued + r.onRunning = onRunning + if onTerminal != nil { + r.onTerminal = onTerminal + } +} + +func (r *Registry) fireEnqueued(j *Job) { + if r == nil || j == nil { + return + } + r.mu.RLock() + fn := r.onEnqueued + r.mu.RUnlock() + if fn != nil { + fn(j) + } +} + +func (r *Registry) fireRunning(j *Job) { + if r == nil || j == nil { + return + } + r.mu.RLock() + fn := r.onRunning + r.mu.RUnlock() + if fn != nil { + fn(j) + } +} + func (r *Registry) fireTerminal(j *Job) { if r == nil || j == nil { return @@ -271,8 +311,6 @@ func (r *Registry) pruneTerminalIfOver(maxJobs int) { // Enqueue creates a job or returns an existing one for the same idempotency key. func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) (*Job, bool, error) { r.mu.Lock() - defer r.mu.Unlock() - maxJobs := registryMaxJobsFromEnv() r.pruneTerminalIfOver(maxJobs) @@ -281,6 +319,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module if existing, ok := r.byIdempo[k]; ok { st := existing.statusLocked() if st == StatusQueued || st == StatusRunning { + r.mu.Unlock() return existing, false, nil } delete(r.byIdempo, k) @@ -302,8 +341,14 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module } r.byID[j.ID] = j r.pruneTerminalIfOver(maxJobs) + enqueuedHook := r.onEnqueued + workerStart := r.workerStart + r.mu.Unlock() - if r.workerStart != nil { + if enqueuedHook != nil { + enqueuedHook(j) + } + if workerStart != nil { go func() { r.workerSem <- struct{}{} active := len(r.workerSem) @@ -313,7 +358,7 @@ func (r *Registry) Enqueue(tenantID, kind string, idempotencyKey *string, module <-r.workerSem observability.RecordJobQueueDepth(len(r.workerSem), capacity) }() - r.workerStart(j) + workerStart(j) }() } return j, true, nil diff --git a/internal/jobs/worker.go b/internal/jobs/worker.go index a580a06..e274445 100644 --- a/internal/jobs/worker.go +++ b/internal/jobs/worker.go @@ -95,6 +95,9 @@ func (w *Worker) Process(j *Job) { return } j.MarkRunning() + if w != nil && w.Registry != nil { + w.Registry.fireRunning(j) + } if j.IsCancelRequested() { j.MarkCancelled() return diff --git a/internal/repository/job_audit.go b/internal/repository/job_audit.go index 58dfcd8..ff75b66 100644 --- a/internal/repository/job_audit.go +++ b/internal/repository/job_audit.go @@ -20,6 +20,28 @@ func NewJobAuditWriter(pool *pgxpool.Pool) *JobAuditWriter { return &JobAuditWriter{pool: pool} } +// UpsertQueued inserts a queued job row (best-effort). +func (w *JobAuditWriter) UpsertQueued(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, moduleID *string, meta map[string]any) { + if w == nil || w.pool == nil { + return + } + metaJSON, _ := json.Marshal(meta) + var idem any + if idempotencyKey != nil && *idempotencyKey != "" { + idem = *idempotencyKey + } + var mod any + if moduleID != nil && *moduleID != "" { + mod = *moduleID + } + _, _ = w.pool.Exec(ctx, ` + INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, module_id, meta_json, created_at) + VALUES ($1::uuid, $2::uuid, $3, 'queued', $4, $5::uuid, $6::jsonb, now()) + ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL + DO UPDATE SET status='queued', meta_json=EXCLUDED.meta_json, module_id=EXCLUDED.module_id`, + jobID, tenantID, kind, idem, mod, metaJSON) +} + // UpsertRunning inserts or updates a running job row (best-effort). func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kind string, idempotencyKey *string, meta map[string]any) { if w == nil || w.pool == nil { @@ -33,8 +55,7 @@ func (w *JobAuditWriter) UpsertRunning(ctx context.Context, tenantID, jobID, kin _, _ = w.pool.Exec(ctx, ` INSERT INTO job_audit (id, tenant_id, kind, status, idempotency_key, meta_json, created_at, started_at) VALUES ($1::uuid, $2::uuid, $3, 'running', $4, $5::jsonb, now(), now()) - ON CONFLICT (tenant_id, idempotency_key) WHERE idempotency_key IS NOT NULL - DO UPDATE SET status='running', started_at=now(), meta_json=EXCLUDED.meta_json`, + ON CONFLICT (id) DO UPDATE SET status='running', started_at=COALESCE(job_audit.started_at, now()), meta_json=EXCLUDED.meta_json`, jobID, tenantID, kind, idem, metaJSON) }