diff --git a/README.md b/README.md index a8ddb46..69cd9b2 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ ForgeQueue is a Go and PostgreSQL job queue focused on durable execution, concur Implemented: - Durable job submission and lookup API with idempotency keys +- Scheduled job availability using an optional `available_at` timestamp - Transactional job claiming with `FOR UPDATE SKIP LOCKED` - Worker identity, execution leases, and lease renewal - Expired-lease recovery and attempt-count increments @@ -54,6 +55,16 @@ curl -X POST http://localhost:8080/jobs \ -d '{"queue_name":"default","job_type":"noop","payload":{"message":"hello ForgeQueue"},"priority":10}' ``` +### Schedule a job + +Include `available_at` when submitting a job to make it ineligible for workers until that timestamp. The value must be an RFC 3339 timestamp; omit it to make the job available immediately. The worker claim query enforces this timestamp in PostgreSQL. + +```sh +curl -X POST http://localhost:8080/jobs \ + -H 'Content-Type: application/json' \ + -d '{"queue_name":"default","job_type":"report","payload":{},"available_at":"2026-11-02T09:00:00Z"}' +``` + ### Retrieve a job ```sh @@ -161,6 +172,6 @@ Publish benchmark numbers only alongside the environment and exact command used. - **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 5 — Scheduling + Dependencies:** scheduled job availability implemented; job dependencies 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/internal/job/model.go b/internal/job/model.go index 75e95b2..bde2911 100644 --- a/internal/job/model.go +++ b/internal/job/model.go @@ -38,4 +38,5 @@ type CreateRequest struct { Priority int `json:"priority"` MaxAttempts int `json:"max_attempts"` IdempotencyKey string `json:"idempotency_key,omitempty"` + AvailableAt *time.Time `json:"available_at,omitempty"` } diff --git a/internal/job/postgres_cancel_integration_test.go b/internal/job/postgres_cancel_integration_test.go index 04dc3a1..7b56ce3 100644 --- a/internal/job/postgres_cancel_integration_test.go +++ b/internal/job/postgres_cancel_integration_test.go @@ -89,16 +89,21 @@ func TestPostgresRepositoryCancelPendingJob(t *testing.T) { } repo := NewPostgresRepository(pool) + scheduledAt := time.Now().UTC().Add(2 * time.Hour).Truncate(time.Microsecond) pending, err := repo.Create(ctx, CreateRequest{ IdempotencyKey: fmt.Sprintf("cancel-pending-%d", time.Now().UnixNano()), QueueName: "cancel-test", JobType: "test", Payload: []byte(`{}`), MaxAttempts: 3, + AvailableAt: &scheduledAt, }) if err != nil { t.Fatalf("create pending job: %v", err) } + if !pending.AvailableAt.Equal(scheduledAt) { + t.Fatalf("available_at = %s; want %s", pending.AvailableAt, scheduledAt) + } cancelled, err := repo.Cancel(ctx, pending.ID) if err != nil { diff --git a/internal/job/postgres_repository.go b/internal/job/postgres_repository.go index c5887cb..7c8f69c 100644 --- a/internal/job/postgres_repository.go +++ b/internal/job/postgres_repository.go @@ -25,8 +25,8 @@ func scanJob(row pgx.Row) (Job, error) { } 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)) + q := "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::timestamptz,NOW())) 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, req.AvailableAt)) } func (r *PostgresRepository) Cancel(ctx context.Context, id string) (Job, error) { diff --git a/internal/job/service.go b/internal/job/service.go index c0f9d55..568273a 100644 --- a/internal/job/service.go +++ b/internal/job/service.go @@ -1,7 +1,61 @@ 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") + } + if req.AvailableAt != nil && req.AvailableAt.IsZero() { + return Job{}, errors.New("available_at must be a valid timestamp") + } + + 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/job/service_schedule_test.go b/internal/job/service_schedule_test.go new file mode 100644 index 0000000..0cce826 --- /dev/null +++ b/internal/job/service_schedule_test.go @@ -0,0 +1,70 @@ +package job + +import ( + "context" + "encoding/json" + "testing" + "time" +) + +type scheduledRequestRepository struct { + request CreateRequest + called bool +} + +func (r *scheduledRequestRepository) Create(_ context.Context, req CreateRequest) (Job, error) { + r.called = true + r.request = req + return Job{}, nil +} + +func (r *scheduledRequestRepository) Get(context.Context, string) (Job, error) { + return Job{}, nil +} + +func (r *scheduledRequestRepository) Cancel(context.Context, string) (Job, error) { + return Job{}, nil +} + +func TestCreatePassesAvailableAtToRepository(t *testing.T) { + availableAt := time.Date(2026, time.November, 2, 9, 0, 0, 0, time.UTC) + repo := &scheduledRequestRepository{} + service := NewService(repo) + + _, err := service.Create(context.Background(), CreateRequest{ + QueueName: "reports", + JobType: "daily-report", + Payload: json.RawMessage(`{}`), + MaxAttempts: 3, + AvailableAt: &availableAt, + }) + if err != nil { + t.Fatalf("create scheduled job: %v", err) + } + if !repo.called { + t.Fatal("expected repository to be called") + } + if repo.request.AvailableAt == nil || !repo.request.AvailableAt.Equal(availableAt) { + t.Fatalf("repository available_at = %v; want %s", repo.request.AvailableAt, availableAt) + } +} + +func TestCreateRejectsZeroAvailableAt(t *testing.T) { + repo := &scheduledRequestRepository{} + service := NewService(repo) + zero := time.Time{} + + _, err := service.Create(context.Background(), CreateRequest{ + QueueName: "reports", + JobType: "daily-report", + Payload: json.RawMessage(`{}`), + MaxAttempts: 3, + AvailableAt: &zero, + }) + if err == nil { + t.Fatal("expected zero available_at to be rejected") + } + if repo.called { + t.Fatal("repository must not be called for an invalid available_at") + } +} diff --git a/internal/queue/postgres_scheduling_integration_test.go b/internal/queue/postgres_scheduling_integration_test.go new file mode 100644 index 0000000..086e50e --- /dev/null +++ b/internal/queue/postgres_scheduling_integration_test.go @@ -0,0 +1,45 @@ +package queue + +import ( + "context" + "errors" + "fmt" + "testing" + "time" + + "github.com/nooby6/forgequeue/internal/job" +) + +func TestPostgresQueueDoesNotClaimScheduledJobBeforeAvailableAt(t *testing.T) { + pool := integrationPool(t) + ctx := context.Background() + queueName := fmt.Sprintf("scheduled-%d", time.Now().UnixNano()) + availableAt := time.Now().UTC().Add(time.Hour) + + created, err := job.NewPostgresRepository(pool).Create(ctx, job.CreateRequest{ + QueueName: queueName, + JobType: "scheduled-test", + Payload: []byte(`{}`), + MaxAttempts: 3, + AvailableAt: &availableAt, + }) + if err != nil { + t.Fatalf("create scheduled job: %v", err) + } + + q := NewPostgresQueue(pool) + if _, err := q.ClaimNext(ctx, queueName, "scheduled-worker", time.Minute); !errors.Is(err, ErrNoJob) { + t.Fatalf("claim before available_at error = %v; want ErrNoJob", err) + } + + if _, err := pool.Exec(ctx, "UPDATE jobs SET available_at = NOW() WHERE id = $1", created.ID); err != nil { + t.Fatalf("make scheduled job available: %v", err) + } + claimed, err := q.ClaimNext(ctx, queueName, "scheduled-worker", time.Minute) + if err != nil { + t.Fatalf("claim job after available_at: %v", err) + } + if claimed.ID != created.ID { + t.Fatalf("claimed job = %s; want %s", claimed.ID, created.ID) + } +}