feat: support one-time scheduled jobs

README.md

@@ -1,40 +1,45 @@
 # 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
 
 ## Architecture
 
 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

db/migrations/000006_job_dependencies.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS job_dependencies;

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);

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"`
 }

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)
 	}

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
+}