feat: worker strategy and sources in GET /workers/{name} (openapi 1.5.0)

CHANGELOG.md

@@ -31,6 +31,7 @@
 
 ### ✨ New Features
 
+- **Worker strategy and sources over REST** (`openapi.yaml` 1.5.0): `GET /workers/{name}` gains `strategy` (`priority`, `round_robin`, `max_len`, `min_len`) and `sources`, each bound queue or topic subscription sorted by priority (ties in bind order; the list the strategy picks from) as `{queue_id, kind, priority}` (`queue_id` as in `queue_ids`, `""` for a topic or storage with no ID; `kind` `queue` or `topic`; bind priority, absent when not set or set to `math.MaxInt`, which ranks the same), so observers draw how a worker pulls. Not in `GET /workers`. No job-path cost (`BenchmarkAdd`, `BenchmarkAddWithId`, `BenchmarkAddAll` allocs/op unchanged).
 - **Job read and cancel over REST** (`openapi.yaml` 1.4.0): `GET /queues/{id}/jobs/{jobId}` (200; 404 queue or job; 501 storage without new optional `JobReader`) and `PATCH /queues/{id}/jobs/{jobId}/actions/cancel` (202 requested; 409 ended; 404 queue, or unknown in-memory job; 501 no `JobCanceller`). `Job` gains `result` (one-job read only; list pages show `result_size`); `GET /queues/{id}/jobs?state=cancelled` lists cancelled jobs (empty for in-memory by default; see `WithEndedJobs`). `contract` v0.7.0: `JobReader`, `ErrJobNotFound`, `JobRecord.Result`, `JobRecord.ResultSize`. `QueueInspector` implementers should accept `JobCancelled` in `Jobs` (an empty page is valid); varmq answers an empty page when storage rejects it but still serves `failed`. In-memory queues read and cancel waiting and running jobs with an ID, whichever worker runs them (running payload `null`; a job starting mid-read can still be read while its worker writes `Data()`, as on the pending list: don't mutate `Data()` in place; a job closed during the read is 404; any closed waiting job is left off the pending page, `offset` counts waiting jobs only and a page fills to `limit`); ended ones are not kept (404) unless the queue is bound with new `WithEndedJobs(n)` (QueueConfigFunc, n clamped 0..200, default 0: nothing kept, no per-job cost): then its last n ended jobs with an ID (completed, failed, cancelled; one ring, oldest dropped) are read by ID, listed newest first under their own state (cancelled not under failed) and counted in `counts.completed`/`counts.failed`; a cancel of one is 409. Records: payload and result as JSON (null / absent over 64 KiB or on encode failure), `finished_at`, attempts, error (cut at 64 KiB); no `enqueued_at`/`started_at`. Kept only while the queue is listed and a mux exists. A job held in a batch behind a running one is not found until it starts. Runs are tracked only for jobs with an ID of a REST-listed queue (a named worker's), once `NewServerMux`/`Handler` was built: tracking costs a lock pair and a map entry per such run, no allocation (`BenchmarkAdd`, `BenchmarkAddAll` and adds with IDs on a named worker, with and without a mux: allocs/op and B/op unchanged). Shared-storage cancel reaches every consumer process (24 h mark, 202 for any ID with no ended record; a waiting workflow task still runs): protect `Handler`.
 - **Worker queue IDs over REST** (`openapi.yaml` 1.4.0): `GET /workers` and `GET /workers/{name}` gain `queue_ids`, the `GET /queues` IDs of the queues each worker consumes, in-memory ones included, so observers link a worker to its queues.
 - **Job cancel** (`redis-storage.md` 1.7.0): `job.Cancel()` on every job handle. A job not started never runs (`Err`/`Result`: `ErrJobCancelled`); a running one has its `j.Context()` cancelled and ends cancelled if it then returns an error or panics (a worker that ignores the context finishes as usual). Shared storage: `Cancel` on a producer handle or `OpenJob`/`OpenErrJob`/`OpenResultJob` reaches the job in any process (new optional `JobCanceller`, `CancelNotifier`, `CancelAcknowledgeable`; outcome `cancelled`, never retried); storage without it: `ErrCancelUnsupported`. `Job[T]` gains `Context()`, made only when called: jobs that don't use it allocate nothing more (`BenchmarkAdd` allocs/op and B/op unchanged).

docs/src/content/docs/guides/configuration.mdx

@@ -20,6 +20,12 @@ Tick options in the scene below and it builds the call. A result list says what
 | Queue options, which return a [`QueueConfigFunc`](https://pkg.go.dev/github.com/goptics/varmq#QueueConfigFunc): [`WithQueueCapacity`](https://pkg.go.dev/github.com/goptics/varmq#WithQueueCapacity) and [`WithQueuePriority`](https://pkg.go.dev/github.com/goptics/varmq#WithQueuePriority) | [`BindQueue`](https://pkg.go.dev/github.com/goptics/varmq#IWorkerBinder.BindQueue), [`BindPriorityQueue`](https://pkg.go.dev/github.com/goptics/varmq#IWorkerBinder.BindPriorityQueue) and the other binding methods, such as [`WithQueue`](https://pkg.go.dev/github.com/goptics/varmq#IWorkerBinder.WithQueue) |
 | [`WithTopicCapacity`](https://pkg.go.dev/github.com/goptics/varmq#WithTopicCapacity), which returns a [`TopicConfigFunc`](https://pkg.go.dev/github.com/goptics/varmq#TopicConfigFunc) | [`NewTopic`](https://pkg.go.dev/github.com/goptics/varmq#NewTopic) |
 | [`WithTopicPriority`](https://pkg.go.dev/github.com/goptics/varmq#WithTopicPriority), which also returns a [`TopicConfigFunc`](https://pkg.go.dev/github.com/goptics/varmq#TopicConfigFunc) | [`WithTopic`](https://pkg.go.dev/github.com/goptics/varmq#IWorkerBinder.WithTopic), when you subscribe a worker |
+| [`WithEndedJobs`](https://pkg.go.dev/github.com/goptics/varmq#WithEndedJobs) | 0: none kept | Queue option for an in-memory queue: keep the records of its last n ended jobs with an ID (completed, failed, cancelled), n clamped 0..200, so REST reads and lists them. Shared storage keeps its own; ignored there. | [REST API](/operations/rest-api/) |
+| [`WithWorkflowID`](https://pkg.go.dev/github.com/goptics/varmq#WithWorkflowID) | generated | Workflow option: the workflow ID, `[A-Za-z0-9_.-]{1,128}`, not `index`. `Submit` is idempotent per ID. | [`NewWorkflow`](https://pkg.go.dev/github.com/goptics/varmq#NewWorkflow) |
+| [`WithFailFast`](https://pkg.go.dev/github.com/goptics/varmq#WithFailFast) | off | Workflow option: the first unabsorbed failure cancels every waiting task, instead of letting unrelated branches finish. | [`NewWorkflow`](https://pkg.go.dev/github.com/goptics/varmq#NewWorkflow) |
+| [`WithUndoOnCancel`](https://pkg.go.dev/github.com/goptics/varmq#WithUndoOnCancel) | `true` | Workflow option: whether `Cancel` runs the undos. Stored with the workflow. | [`NewWorkflow`](https://pkg.go.dev/github.com/goptics/varmq#NewWorkflow) |
+| [`WithResultLimit`](https://pkg.go.dev/github.com/goptics/varmq#WithResultLimit) | 64 KiB | In-memory workflow store option: the largest task result. A bigger one fails the task with `ErrResultTooLarge`. | [`NewWorkflowStore`](https://pkg.go.dev/github.com/goptics/varmq#NewWorkflowStore) |
+| [`WithWorkflowRetention`](https://pkg.go.dev/github.com/goptics/varmq#WithWorkflowRetention) | 7 days | In-memory workflow store option: how long an ended workflow is kept. Running ones stay until they end. | [`NewWorkflowStore`](https://pkg.go.dev/github.com/goptics/varmq#NewWorkflowStore) |
 
 ## Every option
 

docs/src/content/docs/guides/errors.mdx

@@ -92,6 +92,14 @@ The next two sections list every exported error, grouped by how you receive them
 | [`ErrSameConcurrency`](https://pkg.go.dev/github.com/goptics/varmq#ErrSameConcurrency) | `worker already has the same concurrency` | `TunePool` with the limit the worker already has. A value below 1 is turned into the CPU count first, and that is what is compared. |
 | [`ErrJobProcessing`](https://pkg.go.dev/github.com/goptics/varmq#ErrJobProcessing) | `job is processing, you can't close processing job` | [`Close`](https://pkg.go.dev/github.com/goptics/varmq#EnqueuedJob.Close) on a job handle while the job's function is running. |
 | [`ErrJobAlreadyClosed`](https://pkg.go.dev/github.com/goptics/varmq#ErrJobAlreadyClosed) | `job is already closed` | `Close` on a job that is already closed, which includes one that has run, since the worker closes those. |
+| [`ErrJobCancelled`](https://pkg.go.dev/github.com/goptics/varmq#ErrJobCancelled) | `varmq: job cancelled` | `Err()` or `Result()` of a job cancelled before it ran, or one cancelled while it ran that then returned an error or panicked. |
+| [`ErrCancelUnsupported`](https://pkg.go.dev/github.com/goptics/varmq#ErrCancelUnsupported) | `varmq: storage can't cancel jobs` | `Cancel` on a job whose storage has no `JobCanceller`. |
+| [`ErrWorkflowExists`](https://pkg.go.dev/github.com/goptics/varmq#ErrWorkflowExists) | `varmq: workflow already exists` | `Submit` of a workflow whose ID the store holds with a different plan. The same plan again succeeds. |
+| [`ErrWorkflowNotFound`](https://pkg.go.dev/github.com/goptics/varmq#ErrWorkflowNotFound) | `varmq: workflow not found` | `Cancel`, `Status` or `Wait` on a workflow the store does not hold, or an invalid ID. |
+| [`ErrNoResult`](https://pkg.go.dev/github.com/goptics/varmq#ErrNoResult) | `varmq: task has no result` | `TaskContext.Result` for a name that is not a dependency, or a dependency with no result. |
+| [`ErrDepFailed`](https://pkg.go.dev/github.com/goptics/varmq#ErrDepFailed) | `varmq: dependency failed` | `TaskContext.Result` for a dependency that failed or was cancelled. |
+| [`ErrResultTooLarge`](https://pkg.go.dev/github.com/goptics/varmq#ErrResultTooLarge) | `varmq: task result too large` | A task's result is bigger than the store's limit (`WithResultLimit`): the task fails. |
+| [`ErrScopeMismatch`](https://pkg.go.dev/github.com/goptics/varmq#ErrScopeMismatch) | `varmq: queues are in different storage scopes` | `wf.Add` or `Submit` with a queue that can't take workflow tasks, or one in another storage than the workflow store. |
 
 The first six rows are errors that come from the worker's states. [Lifecycle](/concepts/lifecycle/#moving-between-states) lists, call by call, which states return which error. [Concurrency](/guides/concurrency/#changing-the-limit-with-tunepool) covers `TunePool`, and [jobs](/concepts/jobs/#closing-a-queued-job) covers closing a job.
 

docs/src/content/docs/operations/rest-api.mdx

@@ -164,6 +164,8 @@ The concurrency route calls [`TunePool`](https://pkg.go.dev/github.com/goptics/v
 | `num_processing` | [`NumProcessing`](https://pkg.go.dev/github.com/goptics/varmq#Worker.NumProcessing) |
 | `num_pending` | [`NumPending`](https://pkg.go.dev/github.com/goptics/varmq#Worker.NumPending) |
 | `num_idle_workers` | [`NumIdleWorkers`](https://pkg.go.dev/github.com/goptics/varmq#Worker.NumIdleWorkers) |
+| `strategy` (`/workers/{name}` only) | the worker's [`WithStrategy`](https://pkg.go.dev/github.com/goptics/varmq#WithStrategy): `priority`, `round_robin`, `max_len` or `min_len` |
+| `sources` (`/workers/{name}` only) | each bound queue or topic subscription, sorted by `priority` (lowest first, then those with none; ties in bind order), the list the worker's strategy picks from: `queue_id` (its `/queues` ID, `""` if none), `kind` (`queue` or `topic`) and `priority` (from [`WithQueuePriority`](https://pkg.go.dev/github.com/goptics/varmq#WithQueuePriority) or [`WithTopicPriority`](https://pkg.go.dev/github.com/goptics/varmq#WithTopicPriority); absent when not set or set to `math.MaxInt`, which ranks the same) |
 | `metrics.submitted`, `completed`, `successful`, `failed` | the four counters of [`Metrics()`](https://pkg.go.dev/github.com/goptics/varmq#Worker.Metrics) |
 
 [Waiting](/guides/waiting/) covers what `NumPending` and `NumProcessing` count, and [Metrics](/guides/metrics/) covers the four counters on in-memory queues. For a worker on a persistent or distributed queue, see [the overview](/persistence/overview/#metrics).

instance.go

@@ -1,7 +1,9 @@
 package varmq
 
 import (
+	"math"
 	"os"
+	"reflect"
 	"runtime/debug"
 	"slices"
 	"strings"
@@ -96,6 +98,64 @@ func (w *worker[T, JobType]) listedQueueIDs() []string {
 	return ids
 }
 
+// boundSources is the worker's bound sources sorted by priority (ties in
+// bind order), the list its strategy picks from, for GET /workers/{name}. A queue's ID is the one
+// GET /queues lists, "" if it has none; a topic subscription has none.
+func (w *worker[T, JobType]) boundSources() []workerSource {
+	out := []workerSource{}
+	w.manager().each(func(src source[JobType], priority int) {
+		s := workerSource{Kind: "topic"}
+		if priority != math.MaxInt {
+			s.Priority = &priority
+		}
+		if qb, ok := src.(queueBacked); ok {
+			s.Kind = "queue"
+			s.QueueID = w.queueIDOf(qb.storage())
+		}
+		out = append(out, s)
+	})
+	return out
+}
+
+// queueIDOf is q's Queue ID: its own as a QueueInspector, else the one of
+// the worker's in-memory QueueInspector over q (the smallest, if two
+// handles share q), else "".
+func (w *worker[T, JobType]) queueIDOf(q IBaseQueue) string {
+	if qi, ok := q.(QueueInspector); ok {
+		return qi.QueueID()
+	}
+	id := ""
+	w.memQueues.Range(func(k, m any) bool {
+		if sameStorage(m.(*memQueue).q, q) && (id == "" || k.(string) < id) {
+			id = k.(string)
+		}
+		return true
+	})
+	return id
+}
+
+// sameStorage reports a == b without the panic == gives on an uncomparable
+// value: its type, or a map or slice in an interface field of a comparable
+// type (as job_cancel.go's runKey check).
+func sameStorage(a, b IBaseQueue) bool {
+	return reflect.ValueOf(a).Comparable() && a == b
+}
+
+// strategyName is the worker's Strategy as GET /workers/{name} sends it.
+// A value past Priority picks as Priority does (strategyManager.next).
+func (w *worker[T, JobType]) strategyName() string {
+	switch w.configs().strategy {
+	case RoundRobin:
+		return "round_robin"
+	case MaxLen:
+		return "max_len"
+	case MinLen:
+		return "min_len"
+	default:
+		return "priority"
+	}
+}
+
 // varmqVersion is the varmq module version this binary was built with, or ""
 // when unknown (tests, builds without module info).
 func varmqVersion() string { return depVersion(debug.ReadBuildInfo()) }

internal/helpers/manager.go

@@ -205,6 +205,19 @@ func (m *Manager[T]) Only() (T, int, bool) {
 	return m.items[0].item, m.items[0].priority, true
 }
 
+// Each calls fn with every registered item and its priority, in the order
+// the items are kept: priority, ties in registration order. It copies under
+// the read lock and calls fn after.
+func (m *Manager[T]) Each(fn func(item T, priority int)) {
+	m.mx.RLock()
+	items := slices.Clone(m.items)
+	m.mx.RUnlock()
+
+	for _, r := range items {
+		fn(r.item, r.priority)
+	}
+}
+
 // Count returns the number of registered items
 func (m *Manager[T]) Count() int {
 	m.mx.RLock()

internal/helpers/manager_test.go

@@ -309,3 +309,39 @@ func TestManagerOnly(t *testing.T) {
 	_, _, ok = m.Only()
 	assert.False(t, ok, "two items")
 }
+
+func TestManagerEach(t *testing.T) {
+	m := NewManager[*testItem]()
+
+	var got []string
+	m.Each(func(it *testItem, _ int) { got = append(got, it.id) })
+	assert.Empty(t, got, "no items")
+
+	a, b, c := &testItem{id: "a"}, &testItem{id: "b"}, &testItem{id: "c"}
+	m.Register(a, 5)
+	m.Register(b, 1) // sorts first for GetPriorityItem
+	m.Register(c, 5)
+
+	var prios []int
+	m.Each(func(it *testItem, p int) {
+		got = append(got, it.id)
+		prios = append(prios, p)
+	})
+	assert.Equal(t, []string{"b", "a", "c"}, got, "priority order, ties in registration order")
+	assert.Equal(t, []int{1, 5, 5}, prios)
+
+	m.UnregisterItem(a)
+	got = nil
+	m.Each(func(it *testItem, _ int) { got = append(got, it.id) })
+	assert.Equal(t, []string{"b", "c"}, got)
+
+	// fn runs on a copy, lock released: it may change the manager (a held
+	// lock deadlocks here; the test timeout catches it).
+	var seen []string
+	m.Each(func(it *testItem, _ int) {
+		seen = append(seen, it.id)
+		m.UnregisterItem(it)
+		m.Register(&testItem{id: "n" + it.id}, 9)
+	})
+	assert.Equal(t, []string{"b", "c"}, seen, "the snapshot taken before fn ran")
+}

mux.go

@@ -116,6 +116,19 @@ type workerSummary struct {
 	QueueIDs []string `json:"queue_ids"`
 }
 
+// workerSource is one source a worker pulls from, in GET /workers/{name}.
+type workerSource struct {
+	// QueueID is the GET /queues ID, "" when the source has none (a topic
+	// subscription, or storage with no Queue ID).
+	QueueID string `json:"queue_id"`
+	// Kind is "queue" or "topic".
+	Kind string `json:"kind"`
+	// Priority is the bind priority (WithQueuePriority, WithTopicPriority):
+	// lower picks first under the Priority strategy. nil (omitted) when it
+	// is math.MaxInt, the default: JSON clients in JS cannot hold 2^63-1.
+	Priority *int `json:"priority,omitempty"`
+}
+
 // summarizeWorker is w's workerSummary.
 func summarizeWorker(w Worker) workerSummary {
 	s := workerSummary{
@@ -246,13 +259,24 @@ func handleGetWorker(w http.ResponseWriter, r *http.Request) {
 	metrics := worker.Metrics()
 	details := struct {
 		workerSummary
-		Metrics struct {
+		// Strategy is omitted for a Worker that is not varmq's: unknown.
+		Strategy string         `json:"strategy,omitempty"`
+		Sources  []workerSource `json:"sources"`
+		Metrics  struct {
 			Completed  uint64 `json:"completed"`
 			Successful uint64 `json:"successful"`
 			Failed     uint64 `json:"failed"`
 		} `json:"metrics"`
 	}{
 		workerSummary: summarizeWorker(worker),
+		Sources:       []workerSource{},
+	}
+	if p, ok := worker.(interface {
+		strategyName() string
+		boundSources() []workerSource
+	}); ok {
+		details.Strategy = p.strategyName()
+		details.Sources = p.boundSources()
 	}
 
 	details.Metrics.Completed = metrics.Completed()