From 3c0cea4218581ea7ce9d13d5ff88516ea01a0dba Mon Sep 17 00:00:00 2001 From: Aric Camarata Date: Thu, 29 Oct 2026 07:19:19 -0400 Subject: [PATCH] fix(ci): clear the red checks on main (hygiene, registry, security scan, coverage) - job-queue: move the worker loop out of api/handlers.go into worker.go; the file was 425 lines against a 356 budget. The budget row is dropped (file is now 284). - registry.json: workflows port 3712, matching free/workflows/plugin.json (the projection was left stale by the port fix); plugin-counts --check passes. - ci/exec: container_facts_test.go builds its Executor with BaselineIsolator, as the other planned-container tests do; E611 now requires an explicit isolation policy. This was the only failure behind Go Build & Test and the coverage gate. - gitleaks: allowlist the exact synthetic literal guess-0123456789 (signin_test.go fixture in branch history that the CI fetch scans). --- .github/file-size-budget.txt | 1 - .github/gitleaks.toml | 3 + free/ci/internal/exec/container_facts_test.go | 6 +- free/job-queue/go/api/handlers.go | 141 ---------------- free/job-queue/go/api/worker.go | 151 ++++++++++++++++++ registry.json | 2 +- 6 files changed, 158 insertions(+), 146 deletions(-) create mode 100644 free/job-queue/go/api/worker.go diff --git a/.github/file-size-budget.txt b/.github/file-size-budget.txt index aa2814e4f..bf5b31e57 100644 --- a/.github/file-size-budget.txt +++ b/.github/file-size-budget.txt @@ -55,7 +55,6 @@ free/github-runner/third_party/sdk/identity.go 338 free/idme/internal/handlers.go 536 free/invitations/third_party/sdk/devkit/cmd/new-plugin/main.go 510 free/invitations/third_party/sdk/identity.go 338 -free/job-queue/go/api/handlers.go 356 free/link-preview/third_party/sdk/devkit/cmd/new-plugin/main.go 510 free/link-preview/third_party/sdk/identity.go 338 free/linkedin/internal/handlers.go 346 diff --git a/.github/gitleaks.toml b/.github/gitleaks.toml index 857efc6c1..ab7d28021 100644 --- a/.github/gitleaks.toml +++ b/.github/gitleaks.toml @@ -47,6 +47,9 @@ regexes = [ '''^PKCS8-wrapped$''', # free/ci/test/security/redteam/c_pairing_test.go (P7-TRUST-18): NodeKeyID fixture of the pairing prompt test, a counting sequence (history commits carry it; later ones use node-key-1) '''^0123456789abcdef$''', + # free/ci/internal/roles/signin_test.go (D-1276): the fake token prefix the rate-limit test guesses with + # (a counting sequence, the whole matched value); present in branch history that the CI fetch scans + '''^guess-0123456789$''', '''wJalrXUtnFEMI''', '''bPxRfiCYEXAMPLEKEY''', diff --git a/free/ci/internal/exec/container_facts_test.go b/free/ci/internal/exec/container_facts_test.go index 780cdb97d..f0b040cfe 100644 --- a/free/ci/internal/exec/container_facts_test.go +++ b/free/ci/internal/exec/container_facts_test.go @@ -36,7 +36,7 @@ func TestExit137IsVerifiedAgainstTheRuntime(t *testing.T) { t.Run(name, func(t *testing.T) { fakeRuntime(t, "", "137") fakeEvents(t, tc.events...) - res := (&Executor{}).Run(context.Background(), plannedSpec(t, "plan-137-"+name, hardPlanner{})) + res := (&Executor{Isolation: BaselineIsolator{}}).Run(context.Background(), plannedSpec(t, "plan-137-"+name, hardPlanner{})) if res.Reason != tc.reason { t.Fatalf("reason %q want %q (%+v)", res.Reason, tc.reason, res) } @@ -48,7 +48,7 @@ func TestExit125IsVerifiedAgainstTheRuntime(t *testing.T) { t.Run("jobs own exit", func(t *testing.T) { fakeRuntime(t, "", "125") fakeEvents(t, "create", "start", "die") - res := (&Executor{}).Run(context.Background(), plannedSpec(t, "plan-125-job", hardPlanner{})) + res := (&Executor{Isolation: BaselineIsolator{}}).Run(context.Background(), plannedSpec(t, "plan-125-job", hardPlanner{})) if res.Err != nil || res.ExitCode != 125 || res.Attempt.FailureClass != "code" { t.Fatalf("a job that ran and exited 125 is a code failure, not a runtime one: %+v", res) } @@ -56,7 +56,7 @@ func TestExit125IsVerifiedAgainstTheRuntime(t *testing.T) { t.Run("runtime failed", func(t *testing.T) { fakeRuntime(t, "", "125") fakeEvents(t, "create", "destroy") // created, never started - res := (&Executor{}).Run(context.Background(), plannedSpec(t, "plan-125-rt", hardPlanner{})) + res := (&Executor{Isolation: BaselineIsolator{}}).Run(context.Background(), plannedSpec(t, "plan-125-rt", hardPlanner{})) var e *Error if !errors.As(res.Err, &e) || e.Code != "E613" { t.Fatalf("a container that never started is an E613 runtime failure: %+v", res) diff --git a/free/job-queue/go/api/handlers.go b/free/job-queue/go/api/handlers.go index 145a48fcf..15d3d5ef0 100644 --- a/free/job-queue/go/api/handlers.go +++ b/free/job-queue/go/api/handlers.go @@ -9,7 +9,6 @@ import ( "encoding/json" "fmt" "log/slog" - "math" "net/http" "strings" "sync" @@ -78,146 +77,6 @@ func (s *Server) routes() { s.mux.HandleFunc("/metrics", s.handleMetrics) } -// StartWorkers launches one goroutine per queue × concurrency worker. -func (s *Server) StartWorkers(ctx context.Context) error { - for _, q := range s.cfg.Queues { - for i := 0; i < s.cfg.Concurrency; i++ { - go s.worker(ctx, q) - } - s.log.Info("workers started", "queue", q, "concurrency", s.cfg.Concurrency) - } - return nil -} - -// worker blocks on BRPOPLPUSH, processes each job, and updates np_jobqueue_jobs status. -func (s *Server) worker(ctx context.Context, queue string) { - qKey := "queue:" + queue - processingKey := "queue:" + queue + ":processing" - for { - select { - case <-ctx.Done(): - return - default: - } - - // BRPOPLPUSH: atomically move job from tail of qKey to head of processingKey. - result, err := s.rdb.BRPopLPush(ctx, qKey, processingKey, 5*time.Second).Result() - if err == redis.Nil { - continue - } - if err != nil { - if ctx.Err() != nil { - return - } - s.log.Error("BRPOPLPUSH error", "queue", queue, "err", err) - time.Sleep(time.Second) - continue - } - - s.processJob(ctx, queue, processingKey, result) - } -} - -type jobPayload struct { - ID string `json:"id"` - JobType string `json:"job_type"` - Payload json.RawMessage `json:"payload"` - AttemptCount int `json:"attempt_count"` -} - -func (s *Server) processJob(ctx context.Context, queue, processingKey, raw string) { - var job jobPayload - if err := json.Unmarshal([]byte(raw), &job); err != nil { - s.log.Error("invalid job payload", "raw", raw, "err", err) - s.rdb.LRem(ctx, processingKey, 1, raw) - return - } - - s.log.Info("processing job", "id", job.ID, "type", job.JobType, "queue", queue, "attempt", job.AttemptCount+1) - s.updateJobStatus(ctx, job.ID, "active", "") - - // Dispatch to registered handlers (stub: log only — real handlers registered via SDK). - err := s.dispatch(ctx, job.JobType, job.Payload) - - if err != nil { - job.AttemptCount++ - if job.AttemptCount >= s.cfg.MaxAttempts { - s.log.Warn("job exhausted retries, moving to DLQ", "id", job.ID, "attempts", job.AttemptCount) - s.updateJobStatus(ctx, job.ID, "dead", err.Error()) - s.moveToDLQ(ctx, job.ID, err.Error()) - } else { - delay := retryDelay(job.AttemptCount) - s.log.Info("job failed, scheduling retry", "id", job.ID, "delay", delay, "attempt", job.AttemptCount) - s.updateJobStatus(ctx, job.ID, "pending", err.Error()) - // Preserve the incremented attempt count in the Redis execution payload. - retryRaw, _ := json.Marshal(job) - // Re-enqueue after delay - go func(raw string, delay time.Duration) { - time.Sleep(delay) - s.rdb.LPush(ctx, "queue:"+queue, raw) - }(string(retryRaw), delay) - } - } else { - s.updateJobStatus(ctx, job.ID, "completed", "") - } - - // Remove from processing set regardless of outcome. - s.rdb.LRem(ctx, processingKey, 1, raw) -} - -// dispatch calls the registered handler for jobType. -// Returns an error for unknown job types so jobs fail fast and route through -// the retry/DLQ path instead of being silently marked completed. -// Wire real handlers here or use the SDK RegisterHandler bridge once available. -func (s *Server) dispatch(_ context.Context, jobType string, _ json.RawMessage) error { - s.log.Warn("no handler registered for job type; failing job so it routes to retry/DLQ", - "job_type", jobType) - return fmt.Errorf("no handler registered for job_type %q", jobType) -} - -// retryDelay returns exponential backoff: 2^attempt seconds, capped at 1h. -func retryDelay(attempt int) time.Duration { - secs := math.Pow(2, float64(attempt)) - if secs > 3600 { - secs = 3600 - } - return time.Duration(secs) * time.Second -} - -func (s *Server) updateJobStatus(ctx context.Context, id, status, errMsg string) { - if s.db == nil { - return - } - now := time.Now().UTC() - var q string - var args []interface{} - switch status { - case "active": - q = `UPDATE public.np_jobqueue_jobs SET status=$1, started_at=$2, attempt_count=attempt_count+1 WHERE id=$3 AND source_account_id='primary'` - args = []interface{}{status, now, id} - case "completed": - q = `UPDATE public.np_jobqueue_jobs SET status=$1, completed_at=$2 WHERE id=$3 AND source_account_id='primary'` - args = []interface{}{status, now, id} - default: - q = `UPDATE public.np_jobqueue_jobs SET status=$1, error=$2 WHERE id=$3 AND source_account_id='primary'` - args = []interface{}{status, nullStr(errMsg), id} - } - if _, err := s.db.Exec(ctx, q, args...); err != nil { - s.log.Warn("failed to update job status in Postgres", "id", id, "err", err) - } -} - -func (s *Server) moveToDLQ(ctx context.Context, jobID, finalError string) { - if s.db == nil { - return - } - _, err := s.db.Exec(ctx, - `INSERT INTO public.np_job_dlq (job_id, final_error) SELECT id, $2 FROM public.np_jobqueue_jobs WHERE id=$1 AND source_account_id='primary'`, jobID, finalError) - if err != nil { - s.log.Warn("failed to insert into np_job_dlq", "job_id", jobID, "err", err) - } -} - // HTTP handlers (handleHealth lives in health.go) func (s *Server) handleEnqueue(w http.ResponseWriter, r *http.Request) { diff --git a/free/job-queue/go/api/worker.go b/free/job-queue/go/api/worker.go new file mode 100644 index 000000000..7fa622eb8 --- /dev/null +++ b/free/job-queue/go/api/worker.go @@ -0,0 +1,151 @@ +package api + +import ( + "context" + "encoding/json" + "fmt" + "math" + "time" + + "github.com/redis/go-redis/v9" +) + +// StartWorkers launches one goroutine per queue × concurrency worker. +func (s *Server) StartWorkers(ctx context.Context) error { + for _, q := range s.cfg.Queues { + for i := 0; i < s.cfg.Concurrency; i++ { + go s.worker(ctx, q) + } + s.log.Info("workers started", "queue", q, "concurrency", s.cfg.Concurrency) + } + return nil +} + +// worker blocks on BRPOPLPUSH, processes each job, and updates np_jobqueue_jobs status. +func (s *Server) worker(ctx context.Context, queue string) { + qKey := "queue:" + queue + processingKey := "queue:" + queue + ":processing" + for { + select { + case <-ctx.Done(): + return + default: + } + + // BRPOPLPUSH: atomically move job from tail of qKey to head of processingKey. + result, err := s.rdb.BRPopLPush(ctx, qKey, processingKey, 5*time.Second).Result() + if err == redis.Nil { + continue + } + if err != nil { + if ctx.Err() != nil { + return + } + s.log.Error("BRPOPLPUSH error", "queue", queue, "err", err) + time.Sleep(time.Second) + continue + } + + s.processJob(ctx, queue, processingKey, result) + } +} + +type jobPayload struct { + ID string `json:"id"` + JobType string `json:"job_type"` + Payload json.RawMessage `json:"payload"` + AttemptCount int `json:"attempt_count"` +} + +func (s *Server) processJob(ctx context.Context, queue, processingKey, raw string) { + var job jobPayload + if err := json.Unmarshal([]byte(raw), &job); err != nil { + s.log.Error("invalid job payload", "raw", raw, "err", err) + s.rdb.LRem(ctx, processingKey, 1, raw) + return + } + + s.log.Info("processing job", "id", job.ID, "type", job.JobType, "queue", queue, "attempt", job.AttemptCount+1) + s.updateJobStatus(ctx, job.ID, "active", "") + + // Dispatch to registered handlers (stub: log only — real handlers registered via SDK). + err := s.dispatch(ctx, job.JobType, job.Payload) + + if err != nil { + job.AttemptCount++ + if job.AttemptCount >= s.cfg.MaxAttempts { + s.log.Warn("job exhausted retries, moving to DLQ", "id", job.ID, "attempts", job.AttemptCount) + s.updateJobStatus(ctx, job.ID, "dead", err.Error()) + s.moveToDLQ(ctx, job.ID, err.Error()) + } else { + delay := retryDelay(job.AttemptCount) + s.log.Info("job failed, scheduling retry", "id", job.ID, "delay", delay, "attempt", job.AttemptCount) + s.updateJobStatus(ctx, job.ID, "pending", err.Error()) + // Preserve the incremented attempt count in the Redis execution payload. + retryRaw, _ := json.Marshal(job) + // Re-enqueue after delay + go func(raw string, delay time.Duration) { + time.Sleep(delay) + s.rdb.LPush(ctx, "queue:"+queue, raw) + }(string(retryRaw), delay) + } + } else { + s.updateJobStatus(ctx, job.ID, "completed", "") + } + + // Remove from processing set regardless of outcome. + s.rdb.LRem(ctx, processingKey, 1, raw) +} + +// dispatch calls the registered handler for jobType. +// Returns an error for unknown job types so jobs fail fast and route through +// the retry/DLQ path instead of being silently marked completed. +// Wire real handlers here or use the SDK RegisterHandler bridge once available. +func (s *Server) dispatch(_ context.Context, jobType string, _ json.RawMessage) error { + s.log.Warn("no handler registered for job type; failing job so it routes to retry/DLQ", + "job_type", jobType) + return fmt.Errorf("no handler registered for job_type %q", jobType) +} + +// retryDelay returns exponential backoff: 2^attempt seconds, capped at 1h. +func retryDelay(attempt int) time.Duration { + secs := math.Pow(2, float64(attempt)) + if secs > 3600 { + secs = 3600 + } + return time.Duration(secs) * time.Second +} + +func (s *Server) updateJobStatus(ctx context.Context, id, status, errMsg string) { + if s.db == nil { + return + } + now := time.Now().UTC() + var q string + var args []interface{} + switch status { + case "active": + q = `UPDATE public.np_jobqueue_jobs SET status=$1, started_at=$2, attempt_count=attempt_count+1 WHERE id=$3 AND source_account_id='primary'` + args = []interface{}{status, now, id} + case "completed": + q = `UPDATE public.np_jobqueue_jobs SET status=$1, completed_at=$2 WHERE id=$3 AND source_account_id='primary'` + args = []interface{}{status, now, id} + default: + q = `UPDATE public.np_jobqueue_jobs SET status=$1, error=$2 WHERE id=$3 AND source_account_id='primary'` + args = []interface{}{status, nullStr(errMsg), id} + } + if _, err := s.db.Exec(ctx, q, args...); err != nil { + s.log.Warn("failed to update job status in Postgres", "id", id, "err", err) + } +} + +func (s *Server) moveToDLQ(ctx context.Context, jobID, finalError string) { + if s.db == nil { + return + } + _, err := s.db.Exec(ctx, + `INSERT INTO public.np_job_dlq (job_id, final_error) SELECT id, $2 FROM public.np_jobqueue_jobs WHERE id=$1 AND source_account_id='primary'`, jobID, finalError) + if err != nil { + s.log.Warn("failed to insert into np_job_dlq", "job_id", jobID, "err", err) + } +} diff --git a/registry.json b/registry.json index 5ddd5cfd7..0874bc0ff 100644 --- a/registry.json +++ b/registry.json @@ -7976,7 +7976,7 @@ "minNselfVersion": "1.0.0", "min_nself_version": "1.0.0", "name": "workflows", - "port": 3733, + "port": 3712, "releaseTag": "v1.2.13", "release_signature": { "alg": "ed25519",