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.
This commit is contained in:
John O'Keefe committed 2026-10-18 18:02:11 -04:00
1 parent d74db03cf3
commit b2220756e2
3 files changed
+265 -25

No files matched your search

+107 -25
View File
@@ -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() {