denozord/EvoBGP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 131 Wiki Activity

feat(jobs): persist job lifecycle to PostgreSQL job_audit

Browse Source 
UpsertQueued/Running/MarkTerminal через SetPersistHooks; исправлен deadlock
fireEnqueued под Registry mutex.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Denozordec
2026-05-25 10:33:57 +07:00
co-authored by Cursor
parent 4a57c91e29
commit e65cf0d958
4 changed files with 106 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/httpapi/bootstrap.go
+31 -9
View File
@@ -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
internal/jobs/job.go
+49 -4
View File
@@ -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
internal/jobs/worker.go
+3
View File
@@ -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
internal/repository/job_audit.go
+23 -2
View File
@@ -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)
}