From b2220756e2264581dcc280421f9e44e4e70569aa Mon Sep 17 00:00:00 2001 From: John O'Keefe Date: Sun, 18 Oct 2026 18:02:11 -0400 Subject: [PATCH] feat(worker): real per-job cancellation for the background job pool CancelJob existed but was dead code: it only flipped a status in the in-memory results map - it never cancelled anything, and processJob's unconditional completion write overwrote the marker anyway. - derive a per-job context in processJob (parent is the handler ctx for manual scans, Background for auto-scans) and register its cancel func under the job id; previously every manual scan shared the handler's process-lifetime context, so no targeted cancel was possible without killing everything - skip execution at dequeue for jobs cancelled while still queued - record a pending JobResult on BOTH enqueue paths (Enqueue used by auto-scans, EnqueueJob used by the API): queued jobs were invisible to status polling (404 until a worker picked them up) and could not be cancelled before execution started - cancelled jobs keep their last real progress instead of a fake 100% and drop the meaningless "context canceled" error text - broadcast the new scan_cancelled websocket event for cancelled scan jobs instead of scan_complete, so clients reset their scanning indicators rather than reading the abort as a finished scan - sentinel errors (ErrJobNotFound, ErrJobNotCancellable) so HTTP handlers map outcomes to 404/409 without string matching Unit tests (worker_cancel_test.go) pin: queued-cancel never executes, running-cancel invokes the context, error mapping, pending recording on both enqueue paths. --- internal/services/worker.go | 132 ++++++++++++++++---- internal/services/worker_cancel_test.go | 157 ++++++++++++++++++++++++ internal/sync/websocket.go | 1 + 3 files changed, 265 insertions(+), 25 deletions(-) create mode 100644 internal/services/worker_cancel_test.go diff --git a/internal/services/worker.go b/internal/services/worker.go index c960faee..83ef59e8 100644 --- a/internal/services/worker.go +++ b/internal/services/worker.go @@ -4,6 +4,7 @@ import ( "bookhoard/internal/database" wsync "bookhoard/internal/sync" "context" + "errors" "fmt" "net/http" "path/filepath" @@ -27,6 +28,13 @@ const ( JobStatusCancelled JobStatus = "cancelled" ) +// Sentinel errors surfaced by CancelJob so HTTP handlers can map them +// to 404/409 without string matching. +var ( + ErrJobNotFound = errors.New("job not found") + ErrJobNotCancellable = errors.New("job cannot be cancelled") +) + type JobType string const ( @@ -45,11 +53,29 @@ var WorkerInstance *Worker func (w *Worker) Enqueue(job *Job) { select { case w.jobQueue <- job: + w.recordPending(job) default: fmt.Printf("Worker queue full, rejecting job: %s\n", job.ID) } } +// recordPending registers a queued job's pending result so the status +// endpoint and GetActiveScanJobs can report jobs still waiting for a +// worker, and so CancelJob can cancel them before execution starts. +// The entry is replaced when processJob dequeues the job. +func (w *Worker) recordPending(job *Job) { + libID, _ := job.Params["library_id"].(string) + w.mu.Lock() + defer w.mu.Unlock() + if _, exists := w.results[job.ID]; !exists { + w.results[job.ID] = &JobResult{ + JobID: job.ID, + Status: JobStatusPending, + LibraryID: libID, + } + } +} + type Job struct { ID string Type JobType @@ -146,8 +172,14 @@ type AnalyticsJobResult struct { ReportPath string `json:"report_path"` } type Worker struct { - jobQueue chan *Job - results map[string]*JobResult + jobQueue chan *Job + results map[string]*JobResult + // cancels maps a running job's id to its per-job context cancel + // func, registered by processJob and removed at completion. Lets + // CancelJob stop a running job without disturbing siblings (all + // manual scans used to share the handler's process-lifetime + // context, so no targeted cancellation was possible). + cancels map[string]context.CancelFunc connManager *wsync.ConnectionManager mu sync.RWMutex wg sync.WaitGroup @@ -192,6 +224,7 @@ func NewWorkerWithConfig(numWorkers, queueCap int, connManager *wsync.Connection w := &Worker{ jobQueue: make(chan *Job, queueCap), results: make(map[string]*JobResult), + cancels: make(map[string]context.CancelFunc), ctx: ctx, cancel: cancel, } @@ -243,22 +276,40 @@ func (w *Worker) worker() { func (w *Worker) processJob(job *Job) { libID, _ := job.Params["library_id"].(string) w.mu.Lock() + // A job cancelled while queued must never execute: honour the + // pending-cancel marker and leave its result untouched. + if result, exists := w.results[job.ID]; exists && result.Status == JobStatusCancelled { + w.mu.Unlock() + return + } w.results[job.ID] = &JobResult{ JobID: job.ID, Status: JobStatusRunning, LibraryID: libID, } + + // Derive a per-job context so cancelling this job (CancelJob) stops + // only this job. The parent stays whatever the caller passed (the + // handler's context for manual scans, Background for auto-scans). + parent := job.Context + if parent == nil { + parent = context.Background() + } + jobCtx, cancel := context.WithCancel(parent) + job.Context = jobCtx + w.cancels[job.ID] = cancel w.mu.Unlock() + defer func() { + w.mu.Lock() + delete(w.cancels, job.ID) + w.mu.Unlock() + cancel() + }() + started := time.Now() job.StartedAt = &started - w.mu.Lock() - if result, exists := w.results[job.ID]; exists { - result.Status = JobStatusRunning - } - w.mu.Unlock() - var err error var result interface{} @@ -318,27 +369,43 @@ func (w *Worker) processJob(job *Job) { } } + // A cancelled job stopped mid-flight: keep its last real progress + // instead of claiming 100%, and drop the meaningless "context + // canceled" processing error. + finalProgress := 1.0 + finalError := "" + if status == JobStatusCancelled { + finalError = "" + if old, exists := w.results[job.ID]; exists { + finalProgress = old.Progress + } + } else if err != nil { + finalError = err.Error() + } + w.results[job.ID] = &JobResult{ JobID: job.ID, Status: status, LibraryID: libID, - Error: func() string { - if err != nil { - return err.Error() - } else { - return "" - } - }(), - Result: result, - Progress: 1.0, + Error: finalError, + Result: result, + Progress: finalProgress, + FilesScanned: filesScanned, NewItems: newItems, Errors: errors, } if job.Type == JobTypeScan && w.connManager != nil && job.UserID != "" { + // Cancelled scans announce scan_cancelled so clients reset + // their scanning indicators; a "scan_complete" would read as + // a finished scan. + msgType := wsync.MessageTypeScanComplete + if status == JobStatusCancelled { + msgType = wsync.MessageTypeScanCancelled + } w.connManager.BroadcastToUser(job.UserID, wsync.BroadcastMessage{ - Type: wsync.MessageTypeScanComplete, + Type: msgType, Data: map[string]interface{}{ "job_id": job.ID, "files_scanned": filesScanned, @@ -972,6 +1039,7 @@ func (w *Worker) EnqueueJob(job *Job) error { select { case w.jobQueue <- job: + w.recordPending(job) return nil case <-w.ctx.Done(): return fmt.Errorf("worker is shutting down") @@ -1005,19 +1073,33 @@ func (w *Worker) GetActiveScanJobs() []JobResult { return active } +// CancelJob cancels a pending or running job by id. Pending jobs are +// marked cancelled so processJob skips them at dequeue (no context +// exists yet); running jobs have their per-job context cancelled — the +// job stops at its next context check and processJob finalises the +// status as cancelled. func (w *Worker) CancelJob(jobID string) error { w.mu.Lock() defer w.mu.Unlock() - if result, exists := w.results[jobID]; exists { - if result.Status == JobStatusRunning || result.Status == JobStatusPending { - result.Status = JobStatusCancelled - return nil - } - return fmt.Errorf("job cannot be cancelled") + result, exists := w.results[jobID] + if !exists { + return ErrJobNotFound } - return fmt.Errorf("job not found") + switch result.Status { + case JobStatusPending: + result.Status = JobStatusCancelled + return nil + case JobStatusRunning: + result.Status = JobStatusCancelled + if cancel, ok := w.cancels[jobID]; ok { + cancel() + } + return nil + default: + return ErrJobNotCancellable + } } func (w *Worker) Shutdown() { diff --git a/internal/services/worker_cancel_test.go b/internal/services/worker_cancel_test.go new file mode 100644 index 00000000..61722620 --- /dev/null +++ b/internal/services/worker_cancel_test.go @@ -0,0 +1,157 @@ +package services + +import ( + "context" + "errors" + "testing" +) + +// A job cancelled while still queued must never execute. processJob +// sees the pending-cancel marker at dequeue and leaves the result +// untouched; a scan without params would otherwise fail with +// "library_id required" and flip the status to failed — so an empty +// error field proves the job body never ran. +func TestCancelQueuedJobNeverExecutes(t *testing.T) { + // Zero workers: the job stays queued until processJob is driven by + // hand below. + w := NewWorkerWithConfig(0, 10, nil) + + job := &Job{ + ID: "queued-cancel", + Type: JobTypeScan, + Context: context.Background(), + Params: map[string]interface{}{}, + } + if err := w.EnqueueJob(job); err != nil { + t.Fatalf("enqueue failed: %v", err) + } + + // The pending result must exist before any worker picks the job up, + // so status polling can see (and cancel) queued jobs. + result, ok := w.GetJobStatus("queued-cancel") + if !ok { + t.Fatal("queued job has no result entry") + } + if result.Status != JobStatusPending { + t.Fatalf("expected pending status, got %q", result.Status) + } + + if err := w.CancelJob("queued-cancel"); err != nil { + t.Fatalf("CancelJob failed: %v", err) + } + + // Simulate the dequeue that races the cancellation. + w.processJob(job) + + result, _ = w.GetJobStatus("queued-cancel") + if result.Status != JobStatusCancelled { + t.Fatalf("expected cancelled status after dequeue, got %q", result.Status) + } + if result.Error != "" { + t.Errorf("skipped job must carry no processing error, got %q", result.Error) + } +} + +// Cancelling a running job must invoke its per-job context cancel func +// — the old CancelJob only flipped a map status and the job kept +// running to completion. +func TestCancelRunningJobCancelsContext(t *testing.T) { + w := NewWorker(1, nil) + + ctx, cancel := context.WithCancel(context.Background()) + job := &Job{ID: "running-cancel", Type: JobTypeScan, Context: ctx} + + // Seed the state processJob establishes while a job runs. + w.mu.Lock() + w.results[job.ID] = &JobResult{JobID: job.ID, Status: JobStatusRunning} + w.cancels[job.ID] = cancel + w.mu.Unlock() + + if err := w.CancelJob(job.ID); err != nil { + t.Fatalf("CancelJob failed: %v", err) + } + + if ctx.Err() == nil { + t.Fatal("CancelJob must cancel the job's context") + } + result, _ := w.GetJobStatus(job.ID) + if result.Status != JobStatusCancelled { + t.Fatalf("expected cancelled status, got %q", result.Status) + } +} + +func TestCancelJobErrors(t *testing.T) { + w := NewWorker(1, nil) + + if err := w.CancelJob("missing-job"); !errors.Is(err, ErrJobNotFound) { + t.Errorf("unknown job: expected ErrJobNotFound, got %v", err) + } + + w.mu.Lock() + w.results["done-job"] = &JobResult{JobID: "done-job", Status: JobStatusCompleted} + w.results["failed-job"] = &JobResult{JobID: "failed-job", Status: JobStatusFailed} + w.results["cancelled-job"] = &JobResult{JobID: "cancelled-job", Status: JobStatusCancelled} + w.mu.Unlock() + + for _, id := range []string{"done-job", "failed-job", "cancelled-job"} { + if err := w.CancelJob(id); !errors.Is(err, ErrJobNotCancellable) { + t.Errorf("terminal job %q: expected ErrJobNotCancellable, got %v", id, err) + } + } +} + +// Both enqueue paths must register a pending result: auto-scans arrive +// via Enqueue (fire-and-forget) and manual scans via EnqueueJob, and a +// queued job must be visible to status polling and cancellable before +// a worker dequeues it. +func TestEnqueueRecordsPendingResult(t *testing.T) { + w := NewWorkerWithConfig(0, 10, nil) + + fireAndForget := &Job{ + ID: "via-enqueue", + Type: JobTypeScan, + Context: context.Background(), + Params: map[string]interface{}{"library_id": "lib-auto"}, + } + w.Enqueue(fireAndForget) + + tracked := &Job{ + ID: "via-enqueue-job", + Type: JobTypeScan, + Context: context.Background(), + Params: map[string]interface{}{"library_id": "lib-manual"}, + } + if err := w.EnqueueJob(tracked); err != nil { + t.Fatalf("EnqueueJob failed: %v", err) + } + + for id, libID := range map[string]string{ + "via-enqueue": "lib-auto", + "via-enqueue-job": "lib-manual", + } { + result, ok := w.GetJobStatus(id) + if !ok { + t.Fatalf("%s: no pending result recorded", id) + } + if result.Status != JobStatusPending { + t.Errorf("%s: expected pending, got %q", id, result.Status) + } + if result.LibraryID != libID { + t.Errorf("%s: expected library_id %q, got %q", id, libID, result.LibraryID) + } + } +} + +// A scan whose context is already cancelled must abort immediately +// with context.Canceled — before counting, walking, or running the +// archive pass. processJob classifies that error as a cancelled job. +func TestScanFoldersCancelledContext(t *testing.T) { + s := NewMediaScanner(nil) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if err := s.ScanFolders(ctx); !errors.Is(err, context.Canceled) { + t.Fatalf("expected context.Canceled, got %v", err) + } +} diff --git a/internal/sync/websocket.go b/internal/sync/websocket.go index 19bcf71e..5d861713 100644 --- a/internal/sync/websocket.go +++ b/internal/sync/websocket.go @@ -20,6 +20,7 @@ const ( MessageTypeScanProgress = "scan_progress" MessageTypeScanComplete = "scan_complete" MessageTypeScanError = "scan_error" + MessageTypeScanCancelled = "scan_cancelled" ) // BroadcastMessage represents a message to broadcast to connected clients