diff --git a/README.md b/README.md index a8ddb46..043dfef 100644 --- a/README.md +++ b/README.md @@ -1,25 +1,27 @@ # ForgeQueue -ForgeQueue is a Go and PostgreSQL job queue focused on durable execution, concurrent workers, lease ownership, retries, cancellation, and failure recovery. +ForgeQueue is a Go and PostgreSQL job queue focused on durable execution, concurrent workers, lease ownership, retries, scheduling, cancellation, and failure recovery. ## Current status -**Milestone 4 — Reliability and Recovery (verification in progress)** +**Milestone 5 — Scheduling + Dependencies (in progress)** Implemented: - Durable job submission and lookup API with idempotency keys +- One-time scheduled jobs using an RFC 3339 `scheduled_at` timestamp +- Durable job dependencies: workers wait for all prerequisites to succeed; dependents are cancelled if a prerequisite fails or is cancelled +- PostgreSQL-backed readiness times; scheduled jobs remain pending and cannot be claimed before their availability time - Transactional job claiming with `FOR UPDATE SKIP LOCKED` - Worker identity, execution leases, and lease renewal - Expired-lease recovery and attempt-count increments - Exponential retry backoff and maximum-attempt exhaustion - Dead-letter inspection and replay - Graceful worker shutdown +- Safe cancellation of pending jobs and cooperative cancellation of running jobs - Lease renewal for long-running handlers - Active-lease guards for completion, failure, and retry - Idempotent repeated successful completion -- Safe cancellation of pending jobs -- Cooperative cancellation requests for running jobs -- PostgreSQL integration tests for concurrency, lease ownership, cancellation, and recovery +- PostgreSQL integration tests for concurrency, lease ownership, cancellation, scheduling, and recovery - Queue statistics endpoint for operational visibility - CI against PostgreSQL @@ -27,14 +29,17 @@ Implemented: Client → HTTP API → Job service → PostgreSQL → Queue runtime → Worker pool → Handler -Workers claim jobs in a database transaction. PostgreSQL row locking prevents two workers from claiming the same pending row. Leases provide time-bounded ownership; a worker cannot renew or acknowledge a job after its lease expires. Lease identities include a random process-instance token and a worker-slot number so restarted processes and concurrent slots do not reuse an owner identity. +Workers claim jobs in a database transaction. PostgreSQL row locking prevents two workers from claiming the same pending row. Leases provide time-bounded ownership; a worker cannot renew or acknowledge a job after that lease expires. Lease identities include a random process-instance token and a worker-slot number so restarted processes and concurrent slots do not reuse an owner identity. -## Job lifecycle and cancellation +## Job lifecycle and scheduling - Jobs are selected by queue, readiness time, and priority. +- Jobs submitted without `scheduled_at` become available immediately. +- A future `scheduled_at` timestamp is persisted as `available_at`; the worker's existing database claim query ignores the job until that timestamp has passed. No in-memory timer or separate scheduler process is required. +- Timestamps must be RFC 3339, for example `2027-02-03T12:30:00Z`. A timestamp in the past is treated as immediately eligible. - Each claim increments the attempt count. - Failed handlers are retried with capped exponential backoff until `max_attempts` is reached. -- Exhausted jobs become `failed` and are recorded in `dead_letter_jobs`. +- Exhausted jobs become `failed` and are recorded as dead letters. - Expired running jobs are recovered when a worker next claims from that queue. - Pending jobs can be cancelled immediately. - A running job receives a durable `cancellation_requested_at` timestamp. Its handler context is cancelled on the next lease-renewal check. @@ -45,7 +50,7 @@ Workers claim jobs in a database transaction. PostgreSQL row locking prevents tw ## HTTP API -### Submit a job +### Submit an immediate job ```sh curl -X POST http://localhost:8080/jobs \ @@ -54,7 +59,20 @@ curl -X POST http://localhost:8080/jobs \ -d '{"queue_name":"default","job_type":"noop","payload":{"message":"hello ForgeQueue"},"priority":10}' ``` -### Retrieve a job +### Schedule a job + +Use `scheduled_at` as an RFC 3339 timestamp. The job is persisted immediately but cannot be claimed by a worker before its availability time. + +```sh +curl -X POST http://localhost:8080/jobs \ + -H 'Content-Type: application/json' \ + -H 'Idempotency-Key: daily-report-2027-02-03' \ + -d '{"queue_name":"default","job_type":"noop","payload":{"message":"scheduled report"},"scheduled_at":"2027-02-03T12:30:00Z"}' +``` + +The response's `available_at` field confirms the persisted time. Omitting `scheduled_at` keeps the previous immediate-execution behaviour. Reusing an idempotency key returns the existing job and does not reschedule it. + +### Submit a job with dependencies\n\nPass existing job IDs in `depends_on`. The job is persisted immediately, but workers will not claim it until every prerequisite has succeeded. Dependencies are committed atomically with the job. Reusing an idempotency key does not change the original dependency graph. If any prerequisite fails or is cancelled, the dependent is marked cancelled during the next claim attempt for its queue.\n\n```sh\ncurl -X POST http://localhost:8080/jobs \\\n -H 'Content-Type: application/json' \\\n -d '{"queue_name":"default","job_type":"publish-report","payload":{"report_id":"rpt-42"},"depends_on":["PREPARATION_JOB_ID","EXPORT_JOB_ID"]}'\n```\n\n### Retrieve a job ```sh curl http://localhost:8080/jobs/JOB_ID @@ -160,7 +178,7 @@ Publish benchmark numbers only alongside the environment and exact command used. - **Milestone 1 — Foundation:** complete - **Milestone 2 — Durable Job API:** complete - **Milestone 3 — Queue + Worker Runtime:** complete -- **Milestone 4 — Reliability + Recovery:** final CI verification in progress -- **Milestone 5 — Scheduling + Dependencies:** planned +- **Milestone 4 — Reliability + Recovery:** complete +- **Milestone 5 — Scheduling + Dependencies:** in progress; one-time scheduling and dependency gating implemented; recurring schedules remain planned - **Milestone 6 — Observability + Load Testing:** initial queue statistics endpoint added; broader metrics and load testing planned - **Milestone 7 — Infrastructure + Production Operations:** planned diff --git a/db/migrations/000006_job_dependencies.down.sql b/db/migrations/000006_job_dependencies.down.sql new file mode 100644 index 0000000..c34dec7 --- /dev/null +++ b/db/migrations/000006_job_dependencies.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS job_dependencies; \ No newline at end of file diff --git a/db/migrations/000006_job_dependencies.up.sql b/db/migrations/000006_job_dependencies.up.sql new file mode 100644 index 0000000..78ca8b9 --- /dev/null +++ b/db/migrations/000006_job_dependencies.up.sql @@ -0,0 +1,8 @@ +CREATE TABLE job_dependencies ( + job_id UUID NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + depends_on_job_id UUID NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (job_id, depends_on_job_id), + CHECK (job_id <> depends_on_job_id) +); +CREATE INDEX idx_job_dependencies_prerequisite ON job_dependencies(depends_on_job_id, job_id); \ No newline at end of file diff --git a/internal/job/model.go b/internal/job/model.go index 75e95b2..c5cf154 100644 --- a/internal/job/model.go +++ b/internal/job/model.go @@ -8,34 +8,36 @@ import ( type Status string const ( - StatusPending Status = "pending" - StatusRunning Status = "running" + StatusPending Status = "pending" + StatusRunning Status = "running" StatusSucceeded Status = "succeeded" - StatusFailed Status = "failed" + StatusFailed Status = "failed" StatusCancelled Status = "cancelled" ) type Job struct { - ID string `json:"id"` - IdempotencyKey string `json:"idempotency_key,omitempty"` - QueueName string `json:"queue_name"` - JobType string `json:"job_type"` - Payload json.RawMessage `json:"payload"` - Status Status `json:"status"` - Priority int `json:"priority"` - Attempts int `json:"attempts"` - MaxAttempts int `json:"max_attempts"` - AvailableAt time.Time `json:"available_at"` - CancellationRequestedAt *time.Time `json:"cancellation_requested_at,omitempty"` - CreatedAt time.Time `json:"created_at"` - UpdatedAt time.Time `json:"updated_at"` + ID string `json:"id"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + QueueName string `json:"queue_name"` + JobType string `json:"job_type"` + Payload json.RawMessage `json:"payload"` + Status Status `json:"status"` + Priority int `json:"priority"` + Attempts int `json:"attempts"` + MaxAttempts int `json:"max_attempts"` + AvailableAt time.Time `json:"available_at"` + CancellationRequestedAt *time.Time `json:"cancellation_requested_at,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` } type CreateRequest struct { - QueueName string `json:"queue_name"` - JobType string `json:"job_type"` - Payload json.RawMessage `json:"payload"` - Priority int `json:"priority"` - MaxAttempts int `json:"max_attempts"` - IdempotencyKey string `json:"idempotency_key,omitempty"` + QueueName string `json:"queue_name"` + JobType string `json:"job_type"` + Payload json.RawMessage `json:"payload"` + Priority int `json:"priority"` + MaxAttempts int `json:"max_attempts"` + IdempotencyKey string `json:"idempotency_key,omitempty"` + ScheduledAt *time.Time `json:"scheduled_at,omitempty"` + DependsOn []string `json:"depends_on,omitempty"` } diff --git a/internal/job/postgres_cancel_integration_test.go b/internal/job/postgres_cancel_integration_test.go index 04dc3a1..e9a76e0 100644 --- a/internal/job/postgres_cancel_integration_test.go +++ b/internal/job/postgres_cancel_integration_test.go @@ -11,7 +11,7 @@ import ( "github.com/jackc/pgx/v5/pgxpool" ) -func TestPostgresRepositoryCancelPendingJob(t *testing.T) { +func TestPostgresRepositorySchedulingAndCancellation(t *testing.T) { dsn := os.Getenv("TEST_DATABASE_URL") if dsn == "" { t.Skip("TEST_DATABASE_URL is not set") @@ -144,6 +144,25 @@ func TestPostgresRepositoryCancelPendingJob(t *testing.T) { t.Fatalf("repeated cancellation should preserve the original request timestamp: first=%v second=%v", requested.CancellationRequestedAt, repeatedRequest.CancellationRequestedAt) } + runAt := time.Now().UTC().Add(2 * time.Hour).Truncate(time.Microsecond) + scheduled, err := repo.Create(ctx, CreateRequest{ + IdempotencyKey: fmt.Sprintf("scheduled-job-%d", time.Now().UnixNano()), + QueueName: "scheduled-test", + JobType: "report", + Payload: []byte(`{"kind":"daily"}`), + MaxAttempts: 3, + ScheduledAt: &runAt, + }) + if err != nil { + t.Fatalf("create scheduled job: %v", err) + } + if !scheduled.AvailableAt.Equal(runAt) { + t.Fatalf("available_at = %s; want scheduled time %s", scheduled.AvailableAt, runAt) + } + if scheduled.Status != StatusPending { + t.Fatalf("scheduled job status = %q; want pending", scheduled.Status) + } + if _, err := repo.Cancel(ctx, "00000000-0000-0000-0000-000000000000"); err != ErrNotFound { t.Fatalf("cancel missing job error = %v; want ErrNotFound", err) } diff --git a/internal/job/postgres_repository.go b/internal/job/postgres_repository.go index c5887cb..77f05a0 100644 --- a/internal/job/postgres_repository.go +++ b/internal/job/postgres_repository.go @@ -1,76 +1,41 @@ package job import ( - "context" - "errors" - "fmt" - - "github.com/jackc/pgx/v5" - "github.com/jackc/pgx/v5/pgxpool" + "context" + "errors" + "fmt" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" ) var ErrNotFound = errors.New("job not found") var ErrNotCancellable = errors.New("job cannot be cancelled in its current state") - -type PostgresRepository struct{ db *pgxpool.Pool } - -func NewPostgresRepository(db *pgxpool.Pool) *PostgresRepository { return &PostgresRepository{db: db} } - +type PostgresRepository struct { db *pgxpool.Pool } +func NewPostgresRepository(db *pgxpool.Pool) *PostgresRepository { return &PostgresRepository{db:db} } const jobColumns = "id,idempotency_key,queue_name,job_type,payload,status,priority,attempts,max_attempts,available_at,cancellation_requested_at,created_at,updated_at" - -func scanJob(row pgx.Row) (Job, error) { - var j Job - err := row.Scan(&j.ID, &j.IdempotencyKey, &j.QueueName, &j.JobType, &j.Payload, &j.Status, &j.Priority, &j.Attempts, &j.MaxAttempts, &j.AvailableAt, &j.CancellationRequestedAt, &j.CreatedAt, &j.UpdatedAt) - return j, err +func scanJob(row pgx.Row) (Job,error) { + var j Job + err:=row.Scan(&j.ID,&j.IdempotencyKey,&j.QueueName,&j.JobType,&j.Payload,&j.Status,&j.Priority,&j.Attempts,&j.MaxAttempts,&j.AvailableAt,&j.CancellationRequestedAt,&j.CreatedAt,&j.UpdatedAt) + return j,err } - -func (r *PostgresRepository) Create(ctx context.Context, req CreateRequest) (Job, error) { - q := "INSERT INTO jobs(idempotency_key,queue_name,job_type,payload,status,priority,max_attempts) VALUES($1,$2,$3,$4,'pending',$5,$6) ON CONFLICT(idempotency_key) DO UPDATE SET idempotency_key=EXCLUDED.idempotency_key RETURNING " + jobColumns - return scanJob(r.db.QueryRow(ctx, q, req.IdempotencyKey, req.QueueName, req.JobType, req.Payload, req.Priority, req.MaxAttempts)) +func (r *PostgresRepository) Create(ctx context.Context, req CreateRequest) (Job,error) { + tx,err:=r.db.Begin(ctx); if err!=nil{return Job{},fmt.Errorf("begin create job: %w",err)}; defer tx.Rollback(ctx) + j,err:=scanJob(tx.QueryRow(ctx,"INSERT INTO jobs(idempotency_key,queue_name,job_type,payload,status,priority,max_attempts,available_at) VALUES($1,$2,$3,$4,'pending',$5,$6,COALESCE($7,NOW())) ON CONFLICT(idempotency_key) DO UPDATE SET idempotency_key=EXCLUDED.idempotency_key RETURNING "+jobColumns,req.IdempotencyKey,req.QueueName,req.JobType,req.Payload,req.Priority,req.MaxAttempts,req.ScheduledAt)) + if err!=nil{return Job{},fmt.Errorf("insert job: %w",err)} + var inserted bool + if req.IdempotencyKey!="" { if err:=tx.QueryRow(ctx,"SELECT (xmax = 0) FROM jobs WHERE id=$1",j.ID).Scan(&inserted);err!=nil{return Job{},fmt.Errorf("inspect idempotent create: %w",err)} } else {inserted=true} + if inserted { for _,dependencyID:=range req.DependsOn { if _,err:=tx.Exec(ctx,"INSERT INTO job_dependencies(job_id,depends_on_job_id) VALUES($1,$2)",j.ID,dependencyID);err!=nil{return Job{},fmt.Errorf("add job dependency %s: %w",dependencyID,err)} } } + if err:=tx.Commit(ctx);err!=nil{return Job{},fmt.Errorf("commit create job: %w",err)} + return j,nil } - -func (r *PostgresRepository) Cancel(ctx context.Context, id string) (Job, error) { - cancelled, err := scanJob(r.db.QueryRow(ctx, - "UPDATE jobs SET status='cancelled',locked_by=NULL,locked_at=NULL,lease_expires_at=NULL,updated_at=NOW() WHERE id=$1 AND status='pending' RETURNING "+jobColumns, - id, - )) - if err == nil { - return cancelled, nil - } - if !errors.Is(err, pgx.ErrNoRows) { - return Job{}, fmt.Errorf("cancel job: %w", err) - } - - // Running work is not marked terminal until its handler acknowledges - // cancellation. This prevents the queue from reassigning a still-running job. - requested, err := scanJob(r.db.QueryRow(ctx, - "UPDATE jobs SET cancellation_requested_at=COALESCE(cancellation_requested_at,NOW()),updated_at=NOW() WHERE id=$1 AND status='running' RETURNING "+jobColumns, - id, - )) - if err == nil { - return requested, nil - } - if !errors.Is(err, pgx.ErrNoRows) { - return Job{}, fmt.Errorf("request job cancellation: %w", err) - } - - current, err := r.Get(ctx, id) - if err != nil { - return Job{}, err - } - if current.Status == StatusCancelled { - return current, nil - } - return Job{}, ErrNotCancellable -} - -func (r *PostgresRepository) Get(ctx context.Context, id string) (Job, error) { - j, err := scanJob(r.db.QueryRow(ctx, "SELECT "+jobColumns+" FROM jobs WHERE id=$1", id)) - if errors.Is(err, pgx.ErrNoRows) { - return Job{}, ErrNotFound - } - if err != nil { - return Job{}, fmt.Errorf("get job: %w", err) - } - return j, nil +func (r *PostgresRepository) Cancel(ctx context.Context,id string)(Job,error) { + cancelled,err:=scanJob(r.db.QueryRow(ctx,"UPDATE jobs SET status='cancelled',locked_by=NULL,locked_at=NULL,lease_expires_at=NULL,updated_at=NOW() WHERE id=$1 AND status='pending' RETURNING "+jobColumns,id)) + if err==nil{return cancelled,nil};if !errors.Is(err,pgx.ErrNoRows){return Job{},fmt.Errorf("cancel job: %w",err)} + requested,err:=scanJob(r.db.QueryRow(ctx,"UPDATE jobs SET cancellation_requested_at=COALESCE(cancellation_requested_at,NOW()),updated_at=NOW() WHERE id=$1 AND status='running' RETURNING "+jobColumns,id)) + if err==nil{return requested,nil};if !errors.Is(err,pgx.ErrNoRows){return Job{},fmt.Errorf("request job cancellation: %w",err)} + current,err:=r.Get(ctx,id);if err!=nil{return Job{},err};if current.Status==StatusCancelled{return current,nil};return Job{},ErrNotCancellable } +func (r *PostgresRepository) Get(ctx context.Context,id string)(Job,error) { + j,err:=scanJob(r.db.QueryRow(ctx,"SELECT "+jobColumns+" FROM jobs WHERE id=$1",id)) + if errors.Is(err,pgx.ErrNoRows){return Job{},ErrNotFound};if err!=nil{return Job{},fmt.Errorf("get job: %w",err)};return j,nil +} \ No newline at end of file diff --git a/internal/job/scheduled_jobs_test.go b/internal/job/scheduled_jobs_test.go new file mode 100644 index 0000000..84e01eb --- /dev/null +++ b/internal/job/scheduled_jobs_test.go @@ -0,0 +1,93 @@ +package job + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +type schedulingTestRepository struct { + request CreateRequest + job Job + err error +} + +func (r *schedulingTestRepository) Create(_ context.Context, req CreateRequest) (Job, error) { + r.request = req + return r.job, r.err +} + +func (r *schedulingTestRepository) Get(context.Context, string) (Job, error) { + return Job{}, ErrNotFound +} + +func (r *schedulingTestRepository) Cancel(context.Context, string) (Job, error) { + return Job{}, errors.New("not implemented") +} + +func TestCreatePreservesScheduledTime(t *testing.T) { + runAt := time.Date(2027, 2, 3, 12, 30, 0, 0, time.UTC) + repo := &schedulingTestRepository{} + service := NewService(repo) + + _, err := service.Create(context.Background(), CreateRequest{ + QueueName: " default ", + JobType: " report ", + Payload: json.RawMessage(`{"kind":"daily"}`), + ScheduledAt: &runAt, + }) + if err != nil { + t.Fatalf("create scheduled job: %v", err) + } + if repo.request.ScheduledAt == nil || !repo.request.ScheduledAt.Equal(runAt) { + t.Fatalf("scheduled_at = %v; want %v", repo.request.ScheduledAt, runAt) + } + if repo.request.QueueName != "default" || repo.request.JobType != "report" { + t.Fatalf("request was not normalized: %+v", repo.request) + } +} + +func TestCreateHandlerAcceptsScheduledAt(t *testing.T) { + runAt := time.Date(2027, 2, 3, 12, 30, 0, 0, time.UTC) + repo := &schedulingTestRepository{ + job: Job{ID: "job-1", QueueName: "reports", JobType: "daily-report", Status: StatusPending, AvailableAt: runAt}, + } + handler := NewHandler(NewService(repo)) + body := `{"queue_name":"reports","job_type":"daily-report","payload":{"kind":"daily"},"scheduled_at":"2027-02-03T12:30:00Z"}` + request := httptest.NewRequest(http.MethodPost, "/jobs", strings.NewReader(body)) + response := httptest.NewRecorder() + + handler.Create(response, request) + + if response.Code != http.StatusCreated { + t.Fatalf("status = %d; want %d; body=%s", response.Code, http.StatusCreated, response.Body.String()) + } + if repo.request.ScheduledAt == nil || !repo.request.ScheduledAt.Equal(runAt) { + t.Fatalf("scheduled_at = %v; want %v", repo.request.ScheduledAt, runAt) + } + if !strings.Contains(response.Body.String(), `"available_at":"2027-02-03T12:30:00Z"`) { + t.Fatalf("response does not expose the scheduled availability time: %s", response.Body.String()) + } +} + +func TestCreateHandlerRejectsInvalidScheduledAt(t *testing.T) { + repo := &schedulingTestRepository{} + handler := NewHandler(NewService(repo)) + body := `{"queue_name":"reports","job_type":"daily-report","payload":{},"scheduled_at":"next Tuesday"}` + request := httptest.NewRequest(http.MethodPost, "/jobs", strings.NewReader(body)) + response := httptest.NewRecorder() + + handler.Create(response, request) + + if response.Code != http.StatusBadRequest { + t.Fatalf("status = %d; want %d", response.Code, http.StatusBadRequest) + } + if repo.request.ScheduledAt != nil { + t.Fatalf("repository should not be called for invalid timestamp") + } +} diff --git a/internal/job/service.go b/internal/job/service.go index c0f9d55..65d9395 100644 --- a/internal/job/service.go +++ b/internal/job/service.go @@ -1,7 +1,44 @@ package job -import("context";"encoding/json";"errors";"strings") -type Service struct{repo Repository} -func NewService(repo Repository)*Service{return &Service{repo:repo}} -func(s *Service)Create(ctx context.Context,req CreateRequest)(Job,error){req.QueueName=strings.TrimSpace(req.QueueName);req.JobType=strings.TrimSpace(req.JobType);if req.QueueName==""{return Job{},errors.New("queue_name is required")};if req.JobType==""{return Job{},errors.New("job_type is required")};if len(req.Payload)==0||!json.Valid(req.Payload){return Job{},errors.New("payload must be valid JSON")};if req.Priority < -100||req.Priority>100{return Job{},errors.New("priority must be between -100 and 100")};if req.MaxAttempts==0{req.MaxAttempts=3};if req.MaxAttempts<1||req.MaxAttempts>20{return Job{},errors.New("max_attempts must be between 1 and 20")};return s.repo.Create(ctx,req)} -func(s *Service)Get(ctx context.Context,id string)(Job,error){id=strings.TrimSpace(id);if id==""{return Job{},errors.New("job id is required")};return s.repo.Get(ctx,id)} -func(s *Service)Cancel(ctx context.Context,id string)(Job,error){id=strings.TrimSpace(id);if id==""{return Job{},errors.New("job id is required")};return s.repo.Cancel(ctx,id)} + +import ( + "context" + "encoding/json" + "errors" + "strings" +) + +type Service struct { repo Repository } + +func NewService(repo Repository) *Service { return &Service{repo: repo} } + +func (s *Service) Create(ctx context.Context, req CreateRequest) (Job, error) { + req.QueueName = strings.TrimSpace(req.QueueName) + req.JobType = strings.TrimSpace(req.JobType) + if req.QueueName == "" { return Job{}, errors.New("queue_name is required") } + if req.JobType == "" { return Job{}, errors.New("job_type is required") } + if len(req.Payload) == 0 || !json.Valid(req.Payload) { return Job{}, errors.New("payload must be valid JSON") } + if req.Priority < -100 || req.Priority > 100 { return Job{}, errors.New("priority must be between -100 and 100") } + if req.MaxAttempts == 0 { req.MaxAttempts = 3 } + if req.MaxAttempts < 1 || req.MaxAttempts > 20 { return Job{}, errors.New("max_attempts must be between 1 and 20") } + seenDependencies := make(map[string]struct{}, len(req.DependsOn)) + for i, id := range req.DependsOn { + id = strings.TrimSpace(id) + if id == "" { return Job{}, errors.New("depends_on entries must be non-empty job IDs") } + if _, exists := seenDependencies[id]; exists { return Job{}, errors.New("depends_on must not contain duplicate job IDs") } + seenDependencies[id] = struct{}{} + req.DependsOn[i] = id + } + return s.repo.Create(ctx, req) +} + +func (s *Service) Get(ctx context.Context, id string) (Job, error) { + id = strings.TrimSpace(id) + if id == "" { return Job{}, errors.New("job id is required") } + return s.repo.Get(ctx, id) +} + +func (s *Service) Cancel(ctx context.Context, id string) (Job, error) { + id = strings.TrimSpace(id) + if id == "" { return Job{}, errors.New("job id is required") } + return s.repo.Cancel(ctx, id) +} diff --git a/internal/queue/postgres.go b/internal/queue/postgres.go index 129b9c1..5530d66 100644 --- a/internal/queue/postgres.go +++ b/internal/queue/postgres.go @@ -35,8 +35,19 @@ func (q *PostgresQueue) ClaimNext(ctx context.Context, queueName, workerID strin _, err = tx.Exec(ctx, "WITH exhausted AS (UPDATE jobs SET status='failed',locked_by=NULL,locked_at=NULL,lease_expires_at=NULL,updated_at=NOW() WHERE queue_name=$1 AND status='running' AND cancellation_requested_at IS NULL AND lease_expires_at<=NOW() AND attempts>=max_attempts RETURNING id,queue_name,job_type,payload,attempts,max_attempts) INSERT INTO dead_letter_jobs (job_id,queue_name,job_type,payload,attempts,max_attempts,failure_reason,failed_at,replayed_at) SELECT id,queue_name,job_type,payload,attempts,max_attempts,'lease expired after maximum attempts',NOW(),NULL FROM exhausted ON CONFLICT (job_id) DO UPDATE SET queue_name=EXCLUDED.queue_name,job_type=EXCLUDED.job_type,payload=EXCLUDED.payload,attempts=EXCLUDED.attempts,max_attempts=EXCLUDED.max_attempts,failure_reason=EXCLUDED.failure_reason,failed_at=NOW(),replayed_at=NULL", queueName) if err != nil { return job.Job{}, fmt.Errorf("dead-letter expired jobs: %w", err) } + _, err = tx.Exec(ctx, `UPDATE jobs AS candidate SET status='cancelled',updated_at=NOW() + WHERE candidate.queue_name=$1 AND candidate.status='pending' + AND EXISTS (SELECT 1 FROM job_dependencies d JOIN jobs prerequisite ON prerequisite.id=d.depends_on_job_id + WHERE d.job_id=candidate.id AND prerequisite.status IN ('failed','cancelled'))`, queueName) + if err != nil { return job.Job{}, fmt.Errorf("cancel jobs with failed prerequisites: %w", err) } + var id string - err = tx.QueryRow(ctx, "SELECT id FROM jobs WHERE queue_name=$1 AND status='pending' AND available_at<=NOW() ORDER BY priority DESC,available_at ASC,created_at ASC FOR UPDATE SKIP LOCKED LIMIT 1", queueName).Scan(&id) + err = tx.QueryRow(ctx, `SELECT candidate.id FROM jobs AS candidate + WHERE candidate.queue_name=$1 AND candidate.status='pending' AND candidate.available_at<=NOW() + AND NOT EXISTS (SELECT 1 FROM job_dependencies d JOIN jobs prerequisite ON prerequisite.id=d.depends_on_job_id + WHERE d.job_id=candidate.id AND prerequisite.status <> 'succeeded') + ORDER BY candidate.priority DESC,candidate.available_at ASC,candidate.created_at ASC + FOR UPDATE OF candidate SKIP LOCKED LIMIT 1`, queueName).Scan(&id) if errors.Is(err, pgx.ErrNoRows) { if err := tx.Commit(ctx); err != nil { return job.Job{}, fmt.Errorf("commit lease recovery: %w", err) } return job.Job{}, ErrNoJob diff --git a/internal/queue/postgres_integration_test.go b/internal/queue/postgres_integration_test.go index c2ecf66..83d02ad 100644 --- a/internal/queue/postgres_integration_test.go +++ b/internal/queue/postgres_integration_test.go @@ -83,6 +83,7 @@ func integrationPool(t testing.TB) *pgxpool.Pool { "../../db/migrations/000003_retries.up.sql", "../../db/migrations/000004_dead_letter_jobs.up.sql", "../../db/migrations/000005_running_job_cancellation.up.sql", + "../../db/migrations/000006_job_dependencies.up.sql", } { sql, err := os.ReadFile(path) if err != nil { @@ -448,3 +449,60 @@ func TestPostgresQueueClaimRollsBackExpiredLeaseRecoveryWhenDeadLetterWriteFails t.Fatalf("dead letters after recovery rollback = %d, expected 0", len(letters)) } } + + +func TestPostgresQueueDoesNotClaimScheduledJobEarly(t *testing.T) { + pool := integrationPool(t) + ctx := context.Background() + queueName := fmt.Sprintf("scheduled-%d", time.Now().UnixNano()) + runAt := time.Now().Add(2 * time.Minute).UTC() + var id string + err := pool.QueryRow(ctx, ` + INSERT INTO jobs (idempotency_key, queue_name, job_type, payload, status, priority, max_attempts, available_at) + VALUES (gen_random_uuid()::text, $1, 'scheduled-test', '{}', 'pending', 0, 3, $2) + RETURNING id + `, queueName, runAt).Scan(&id) + if err != nil { + t.Fatalf("insert scheduled job: %v", err) + } + + q := NewPostgresQueue(pool) + if _, err := q.ClaimNext(ctx, queueName, "early-worker", time.Minute); err != ErrNoJob { + t.Fatalf("claim before scheduled time returned %v; want ErrNoJob", err) + } + + if _, err := pool.Exec(ctx, "UPDATE jobs SET available_at=NOW() WHERE id=$1", id); err != nil { + t.Fatalf("make scheduled job ready: %v", err) + } + claimed, err := q.ClaimNext(ctx, queueName, "ready-worker", time.Minute) + if err != nil { + t.Fatalf("claim job after it becomes available: %v", err) + } + if claimed.ID != id { + t.Fatalf("claimed job %s; want %s", claimed.ID, id) + } +} + + +func TestPostgresQueueWaitsForDependencies(t *testing.T) { + pool:=integrationPool(t);ctx:=context.Background();queueName:=fmt.Sprintf("deps-%d",time.Now().UnixNano()) + prerequisite:=insertIntegrationJob(t,pool,queueName,0,3,"pending",time.Time{}) + dependent:=insertIntegrationJob(t,pool,queueName,0,3,"pending",time.Time{}) + if _,err:=pool.Exec(ctx,"INSERT INTO job_dependencies(job_id,depends_on_job_id) VALUES($1,$2)",dependent,prerequisite);err!=nil{t.Fatal(err)} + q:=NewPostgresQueue(pool);claimed,err:=q.ClaimNext(ctx,queueName,"worker",time.Minute) + if err!=nil||claimed.ID!=prerequisite{t.Fatalf("first claim = %v, %v; want prerequisite %s",claimed,err,prerequisite)} + if _,err:=q.ClaimNext(ctx,queueName,"worker-2",time.Minute);err!=ErrNoJob{t.Fatalf("claim with unresolved prerequisite = %v; want ErrNoJob",err)} + if err:=q.Complete(ctx,prerequisite,"worker");err!=nil{t.Fatal(err)} + claimed,err=q.ClaimNext(ctx,queueName,"worker-2",time.Minute) + if err!=nil||claimed.ID!=dependent{t.Fatalf("claim after prerequisite = %v, %v; want dependent %s",claimed,err,dependent)} +} + +func TestPostgresQueueCancelsJobsWithFailedDependencies(t *testing.T) { + pool:=integrationPool(t);ctx:=context.Background();queueName:=fmt.Sprintf("failed-deps-%d",time.Now().UnixNano()) + prerequisite:=insertIntegrationJob(t,pool,queueName,0,3,"failed",time.Time{}) + dependent:=insertIntegrationJob(t,pool,queueName,0,3,"pending",time.Time{}) + if _,err:=pool.Exec(ctx,"INSERT INTO job_dependencies(job_id,depends_on_job_id) VALUES($1,$2)",dependent,prerequisite);err!=nil{t.Fatal(err)} + q:=NewPostgresQueue(pool);if _,err:=q.ClaimNext(ctx,queueName,"worker",time.Minute);err!=ErrNoJob{t.Fatalf("claim = %v; want ErrNoJob",err)} + var status string;if err:=pool.QueryRow(ctx,"SELECT status FROM jobs WHERE id=$1",dependent).Scan(&status);err!=nil{t.Fatal(err)} + if status!="cancelled"{t.Fatalf("dependent status = %q; want cancelled",status)} +} diff --git a/internal/worker/worker_process_integration_test.go b/internal/worker/worker_process_integration_test.go index a867ccf..c2b6f50 100644 --- a/internal/worker/worker_process_integration_test.go +++ b/internal/worker/worker_process_integration_test.go @@ -72,7 +72,7 @@ func TestWorkerRecoversJobAfterAbruptProcessExit(t *testing.T) { pool, err := pgxpool.NewWithConfig(ctx, cfg) if err != nil { t.Fatal(err) } defer pool.Close() - for _, path := range []string{"../../db/migrations/000001_initial_schema.up.sql","../../db/migrations/000002_worker_leases.up.sql","../../db/migrations/000003_retries.up.sql","../../db/migrations/000004_dead_letter_jobs.up.sql","../../db/migrations/000005_running_job_cancellation.up.sql"} { + for _, path := range []string{"../../db/migrations/000001_initial_schema.up.sql","../../db/migrations/000002_worker_leases.up.sql","../../db/migrations/000003_retries.up.sql","../../db/migrations/000004_dead_letter_jobs.up.sql","../../db/migrations/000005_running_job_cancellation.up.sql","../../db/migrations/000006_job_dependencies.up.sql"} { sql, e := os.ReadFile(path) if e != nil { t.Fatal(e)