feat(web): worker sources graph, Workers on queue list, drop Submitted

DESIGN.md

@@ -242,7 +242,10 @@ Brand (20px three-line SVG mark in accent, wordmark 16px/650). Nav: Dashboard, I
 `h1` + `.sub` scope line on one baseline, wrapping. Worker, queue and job pages add breadcrumbs (`/` separator, 13px, `subtle`).
 
 ### Stat strip
-One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairlines (each tile draws left+top edge pulled 1px under the frame). Tile: label heading, display number, sparkline at foot. `.hot` (failed > 0): number and onset line in `bad`. Onset line: "since 15:04" under the failed number. Status tile lists status counts as status dots. Worker page counters use the same strip without sparklines. Above them, a "Queues" line: each queue the worker's `queue_ids` hold, as a queue name link; hidden when stale or without `queue_ids`.
+One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairlines (each tile draws left+top edge pulled 1px under the frame). Tile: label heading, display number, sparkline at foot. `.hot` (failed > 0): number and onset line in `bad`. Onset line: "since 15:04" under the failed number. Status tile lists status counts as status dots. Worker page counters use the same strip without sparklines: Completed, Successful, Failed, Success rate (no Submitted: varmq API 1.1.0 dropped it; a queue's is its `totals.submitted`).
+
+### Worker sources
+Under the worker's counters, `h2.section` "Sources" + a fan (`.fan`, `bg` well, hairline, 8px; no JS): left, one node per source in priority order (the list the strategy picks from) (`.srcn`: `panel`, `line-strong` 1px, 8px, as a DAG task box): `qname` (glyph, name, kind badge) + priority chip "p N" (`.prio`, 12px `subtle` pill) only under the `priority` strategy and when set; a link to the queue page. Topic: broadcast glyph + "topic"; storage with no Queue ID: italic `subtle` "custom storage"; neither links. Edges (1.5px `subtle` 70%, as DAG edges) join one bus, then a 7px arrow into the worker node (`.wnode`: `panel`, 1.5px `accent`, status dot + name 600, strategy in 12px `subtle`: "Priority: lowest number first", "Round robin: each source in turn", "Max length: longest queue first", "Min length: shortest non-empty first"; "Pulls directly" with one source). Under 640px the bus runs down the left, the worker below. Pods merge by kind + Queue ID (first pod's order, then what others add; first pod's strategy). Older varmq (no `sources`): nodes from `queue_ids`, no label, a `subtle` note. Hidden when stale.
 - **Gap:** tile sparklines are fixed 160x32 (row trends 120x24); they do not stretch with the tile.
 
 ### Queue name
@@ -251,7 +254,7 @@ One `panel` (8px, hairline) holding auto-fit tiles (min 150px) split by hairline
 - **DAG node:** the name, cut at 12 chars, after the state.
 
 ### Tables
-Inside `.table-wrap` panel. Head `raised`, 12px/500 `subtle`; cells 8px 12px, hairline rows, no zebra. Row hover `raised` 60%; selected row accent 8%. Names `fg`/500, accent + underline on hover. Under 640px worker and queue rows stack into a labelled grid (head moved off-screen, `data-label` captions, status on its own line).
+Inside `.table-wrap` panel. Head `raised`, 12px/500 `subtle`; cells 8px 12px, hairline rows, no zebra. Row hover `raised` 60%; selected row accent 8%. Names `fg`/500, accent + underline on hover. Queue list "Workers" column: a chip per worker whose `queue_ids` hold the queue (link to its page, title "runs on <instances>"); none known: a 12px `subtle` "instances" then the instance chips. Under 640px worker and queue rows stack into a labelled grid (head moved off-screen, `data-label` captions, status on its own line).
 - **Gap:** no column sorting; needs a server-side sort param (no new JS).
 
 ### Status dot
@@ -293,7 +296,7 @@ Default: `raised`, `line-strong` border, 13px/500, 5px 10px, hover border `subtl
 - **Waiting:** 16px `fg` line with a 10px teal dot; page meta-refreshes every 2s.
 
 ### Job page
-- **Header:** `h1.qtitle`: queue name (20px/600, 18px glyph, kind badge); under it the full Queue ID in mono 12px `subtle` with a "Copy ID" button (`queue.js`, `data-copy`; hidden where `navigator.clipboard` is missing); "Namespace" chip (redisq) and "Reported by" chips. Breadcrumb ends in the queue name.
+- **Header:** `h1.qtitle`: queue name (20px/600, 18px glyph, kind badge); under it the full Queue ID in mono 12px `subtle` with a "Copy ID" button (`queue.js`, `data-copy`; hidden where `navigator.clipboard` is missing); "Namespace" chip (redisq); "Read from instance-1, …" in 12px `subtle` (the workers table names the workers). Breadcrumb ends in the queue name.
 - **Workers on this queue:** `h2.section` + table (instance, worker link to its page, status dot, concurrency) of the workers whose `queue_ids` hold the queue; stacks under 640px like queue rows. None with an older varmq (no `queue_ids`).
 - **Queue controls** (`#qstate`, under the header): paused status dot ("paused since"), default buttons "Pause queue" / "Resume queue" (Resume alone when every instance says paused, Pause alone when none does, both when mixed or unknown); hidden, with a `subtle` line, when no instance reports `paused` (varmq < API 1.2.0). Confirm panel lists instance x sent, then results; `#qstate` refreshed out of band.
 - **State tabs:** one sticky strip, five equal columns split by hairlines (scrolls sideways under 640px); label (capitalized, state dot: processing warn, completed ok, failed bad, cancelled hollow `bad` ring, pending `subtle`) over 20px count, `—` in `subtle` when varmq does not count the state (cancelled); current `raised` + 2px teal underline.

PRODUCT.md

@@ -34,7 +34,7 @@ Live dashboard and control panel for varmq worker fleets. Talks to varmq's REST
 
 ## Capabilities and Constraints
 
-- Pages: Dashboard (fleet totals + per-worker rows with 10-min trend), Instance grid (every worker on every instance), Worker page (its queues, totals, success rate, completed/failed charts with 1h | 24h | 7d | 30d), Queues list (by name: store icon, FIFO/Priority, namespace; paused badge), Queue job page (full ID with copy, workers on the queue, jobs by state incl. cancelled, paged, payload, per-attempt errors; queue pause/resume), One-job page (varmq API 1.4.0: state, timeline with waited / ran / total timers, live while it waits or runs; per-attempt errors, payload, result; cancel; linked from queue lists and workflow tasks), Workflows list (varmq API 1.3.0: merged across instances, one row per workflow, state tabs, progress, duration), Workflow page (task DAG laid out server-side as SVG: pan/zoom, upstream/downstream trace, task panel, task table; live while it runs; cancel). Confirm dialogs and action results.
+- Pages: Dashboard (fleet totals + per-worker rows with 10-min trend), Instance grid (every worker on every instance), Worker page (its sources fanning into it in priority order, with store glyphs, priorities and the pick strategy (varmq API 1.5.0; `queue_ids` before), totals, success rate, completed/failed charts with 1h | 24h | 7d | 30d), Queues list (by name: store icon, FIFO/Priority, namespace; paused badge; the workers consuming each), Queue job page (full ID with copy, workers on the queue, jobs by state incl. cancelled, paged, payload, per-attempt errors; queue pause/resume), One-job page (varmq API 1.4.0: state, timeline with waited / ran / total timers, live while it waits or runs; per-attempt errors, payload, result; cancel; linked from queue lists and workflow tasks), Workflows list (varmq API 1.3.0: merged across instances, one row per workflow, state tabs, progress, duration), Workflow page (task DAG laid out server-side as SVG: pan/zoom, upstream/downstream trace, task panel, task table; live while it runs; cancel). Confirm dialogs and action results.
 - Actions: pause, resume, restart, stop, concurrency; bulk pause/resume; queue pause/resume (varmq API 1.2.0+, storage-wide: every process reading the queue); workflow cancel, with or without undo jobs (API 1.3.0, stored with the workflow: every process working on it); job cancel (API 1.4.0, cooperative: sent to the first instance that has the job). No job is retried, removed or purged.
 - **Stack (binding):** one Go binary. Server-rendered `html/template` + vendored htmx + plain CSS/JS, embedded with `go:embed`. No Node toolchain, no build step, no front-end framework.
 - Safety behavior (product function, not styling): loopback only; bearer token never reaches the browser; actions refused and nothing sent while data is stale or from another configuration; stale data called out with its age and reason.

README.md

@@ -44,7 +44,7 @@ There is no agent to deploy and no database to run: vartop is one binary on your
 
 - **Fleet dashboard.** Completed per second, failed per minute, pending jobs and workers by status, with a 10-minute trend per worker.
 - **Instance grid.** Every worker on every instance at a glance: status, busy/concurrency, pending.
-- **Worker page.** Totals and success rate, and two charts (completed /s, failed /s), each with its own **1h | 24h** switch, plus **7d** and **30d** as far as the server's `history` reaches, axes and a hover readout.
+- **Worker page.** The queues and topics it pulls from, fanning into it in the order it picks them, with priorities and its strategy (varmq API 1.5.0). Totals and success rate, and two charts (completed /s, failed /s), each with its own **1h | 24h** switch, plus **7d** and **30d** as far as the server's `history` reaches, axes and a hover readout.
 - **Workflows** (varmq API 1.3.0). Every workflow, newest first, once however many instances report it, with its state, tasks ended and duration; its task graph laid out left to right: pan and zoom, hover a task to trace what it waits for and what waits on it, click it for its queue, job, reason, result size and undo job. A running workflow's page follows it live.
 - **One job** (varmq API 1.4.0). Its state, a timeline with how long it waited and ran (ticking while it does), each failed attempt's error, its payload and result. Linked from a queue's job list and a workflow's tasks.
 - **History that survives restarts.** 24 hours of polls, and up to 30 days of each worker's rates per server, in a local SQLite store.
@@ -128,7 +128,7 @@ The page above. One row per worker, merged across every instance (or one instanc
 
 <picture>
   <source media="(prefers-color-scheme: dark)" srcset="docs/images/worker-dark.png">
-  <img alt="A worker page: submitted, completed, successful, failed and success rate; a 7-day completed/s chart with a daily rhythm; a 24-hour failed/s chart; the concurrency slider and the pause, resume, restart and stop buttons" src="docs/images/worker-light.png">
+  <img alt="A worker page: completed, successful, failed and success rate; a 7-day completed/s chart with a daily rhythm; a 24-hour failed/s chart; the concurrency slider and the pause, resume, restart and stop buttons" src="docs/images/worker-light.png">
 </picture>
 
 Click a worker for its totals and success rate, its two charts, and its controls. Each chart has its own span: **1h** from the stored polls, **24h** from 5-minute buckets, and **7d**, **30d** (or the server's own [`history`](#schema), like `14d`) merged into longer steps. Hovering a chart shows how many instances answered and, on 1h, the poll's time and value; on longer spans, the exact window, its average and its peak:

internal/api/client.go

@@ -47,17 +47,35 @@ type WorkerSummary struct {
 	Instance string   `json:"-"` // varmq instance that served it (InstanceHeader), if reported
 }
 
-// WorkerDetails matches the extended API response of GET /workers/{name}
+// WorkerDetails matches the extended API response of GET /workers/{name}.
+// Metrics has no submitted: varmq API 1.1.0 dropped it (a queue's is its
+// totals.submitted under GET /queues).
 type WorkerDetails struct {
 	WorkerSummary
+	// Strategy: how the worker picks the next of its Sources (varmq API
+	// 1.5.0): priority, round_robin, max_len, min_len, or one varmq adds
+	// later; "" from an older varmq.
+	Strategy string `json:"strategy,omitempty"`
+	// Sources: the queues and topic subscriptions it pulls from, in pick
+	// order (varmq API 1.5.0); nil from an older varmq.
+	Sources []WorkerSource `json:"sources,omitempty"`
 	Metrics struct {
-		Submitted  uint64 `json:"submitted"`
 		Completed  uint64 `json:"completed"`
 		Successful uint64 `json:"successful"`
 		Failed     uint64 `json:"failed"`
 	} `json:"metrics"`
 }
 
+// WorkerSource is one source a worker pulls from (spec WorkerSource):
+// QueueID "" for a topic subscription or storage with no Queue ID; Kind
+// queue or topic (or one varmq adds later); Priority nil when none was set
+// (the lowest).
+type WorkerSource struct {
+	QueueID  string `json:"queue_id"`
+	Kind     string `json:"kind"`
+	Priority *int64 `json:"priority,omitempty"`
+}
+
 // APIError represents an error response from the varmq server.
 type APIError struct {
 	StatusCode int

internal/api/client_test.go

@@ -3,6 +3,7 @@ package api
 import (
 	"context"
 	"errors"
+	"math"
 	"net"
 	"net/http"
 	"net/http/httptest"
@@ -64,7 +65,7 @@ func recordingServer(t *testing.T, status int, header map[string]string, body st
 
 func TestGetWorker(t *testing.T) {
 	const body = `{"name":"emails","status":"running","concurrency":4,"num_processing":2,"num_pending":7,"num_idle_workers":2,
-		"queue_ids":["redisq:fifo:{emails}"],"metrics":{"submitted":10,"completed":8,"successful":6,"failed":2}}`
+		"queue_ids":["redisq:fifo:{emails}"],"metrics":{"completed":8,"successful":6,"failed":2}}`
 
 	tests := []struct {
 		name         string
@@ -95,7 +96,7 @@ func TestGetWorker(t *testing.T) {
 			if !slices.Equal(w.QueueIDs, []string{"redisq:fifo:{emails}"}) {
 				t.Errorf("QueueIDs = %q", w.QueueIDs)
 			}
-			if w.Metrics.Submitted != 10 || w.Metrics.Completed != 8 || w.Metrics.Successful != 6 || w.Metrics.Failed != 2 {
+			if w.Metrics.Completed != 8 || w.Metrics.Successful != 6 || w.Metrics.Failed != 2 {
 				t.Errorf("metrics decoded wrong: %+v", w.Metrics)
 			}
 			if w.Origin != srv.URL {
@@ -602,3 +603,23 @@ func TestFreshMethodsEscapeNothing(t *testing.T) {
 		t.Fatalf("paths = %v, want %v", paths, want)
 	}
 }
+
+// strategy and sources (varmq API 1.5.0) decode in priority order, a priority
+// absent as nil; an older varmq sends neither: "" and nil.
+func TestGetWorkerSources(t *testing.T) {
+	srv, _ := recordingServer(t, http.StatusOK, nil, `{"name":"w","status":"Running","concurrency":1,"num_processing":0,"num_pending":0,"num_idle_workers":1,
+		"strategy":"priority","sources":[{"queue_id":"redisq:fifo:{a}","kind":"queue","priority":-2},{"queue_id":"","kind":"topic"},{"queue_id":"","kind":"queue","priority":9223372036854775807}],
+		"metrics":{"completed":0,"successful":0,"failed":0}}`)
+	w, err := NewClient(srv.URL, "").GetWorkerCtx(context.Background(), "w")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if w.Strategy != "priority" || len(w.Sources) != 3 || w.Sources[0].QueueID != "redisq:fifo:{a}" || *w.Sources[0].Priority != -2 ||
+		w.Sources[1].Kind != "topic" || w.Sources[1].Priority != nil || *w.Sources[2].Priority != math.MaxInt64 {
+		t.Errorf("worker = %+v", w)
+	}
+	srv, _ = recordingServer(t, http.StatusOK, nil, `{"name":"w","status":"Running","concurrency":1,"num_processing":0,"num_pending":0,"num_idle_workers":1,"metrics":{"submitted":3,"completed":0,"successful":0,"failed":0}}`)
+	if w, err := NewClient(srv.URL, "").GetWorkerCtx(context.Background(), "w"); err != nil || w.Strategy != "" || w.Sources != nil {
+		t.Errorf("older varmq: %+v, %v", w, err)
+	}
+}

internal/fleet/fleettest/fakevarmq.go

@@ -17,9 +17,13 @@ import (
 
 // Worker is one worker as a Varmq serves it.
 type Worker struct {
-	Status                                   string
-	Concurrency, Busy, Pending, Idle         int
-	Submitted, Completed, Successful, Failed uint64
+	Status                           string
+	Concurrency, Busy, Pending, Idle int
+	Completed, Successful, Failed    uint64
+	// Strategy and Sources, served by GET /workers/{name} when Sources is
+	// non-nil (varmq API 1.5.0); nil: an older varmq.
+	Strategy string
+	Sources  []api.WorkerSource
 	// QueueIDs, served as queue_ids when non-nil; nil: an older varmq.
 	QueueIDs []string
 	// ActionStatus, when non-zero, makes every PATCH on this worker reply
@@ -166,8 +170,10 @@ func (f *Varmq) get(w http.ResponseWriter, r *http.Request) {
 	if fw != nil {
 		failWith = fw.GetStatus
 		body = fakeSummary(name, fw)
+		if fw.Sources != nil {
+			body["strategy"], body["sources"] = fw.Strategy, fw.Sources
+		}
 		body["metrics"] = map[string]uint64{
-			"submitted":  fw.Submitted,
 			"completed":  fw.Completed,
 			"successful": fw.Successful,
 			"failed":     fw.Failed,
@@ -575,3 +581,24 @@ func NewInstances(t *testing.T, ids ...string) ([]*Varmq, []*httptest.Server) {
 	}
 	return fakes, srvs
 }
+
+// SourceWorkers are workers on queues redis, sqlite and memory (Queue IDs)
+// as varmq API 1.5.0 serves them, for tests and screenshots: "orders"
+// picks by priority from all three (1, 5, none set), "emails" pulls from
+// redis alone, "fanout" takes turns between sqlite, a topic and storage
+// with no Queue ID, and "legacy" is an older varmq's: queue_ids only.
+func SourceWorkers(redis, sqlite, memory string) map[string]*Worker {
+	p := func(n int64) *int64 { return &n }
+	q := func(id string, prio *int64) api.WorkerSource {
+		return api.WorkerSource{QueueID: id, Kind: "queue", Priority: prio}
+	}
+	return map[string]*Worker{
+		"orders": {Status: "Running", Concurrency: 8, Busy: 2, Idle: 6, QueueIDs: []string{memory, redis, sqlite},
+			Strategy: "priority", Sources: []api.WorkerSource{q(redis, p(1)), q(sqlite, p(5)), q(memory, nil)}},
+		"emails": {Status: "Running", Concurrency: 2, Idle: 2, QueueIDs: []string{redis},
+			Strategy: "priority", Sources: []api.WorkerSource{q(redis, nil)}},
+		"fanout": {Status: "Idle", Concurrency: 1, Idle: 1, QueueIDs: []string{sqlite},
+			Strategy: "round_robin", Sources: []api.WorkerSource{q(sqlite, p(2)), {Kind: "topic"}, {Kind: "queue"}}},
+		"legacy": {Status: "Running", Concurrency: 1, Idle: 1, QueueIDs: []string{redis, sqlite}},
+	}
+}

internal/fleet/poller_test.go

@@ -70,10 +70,10 @@ func countCalls(calls []string, call string) int {
 func TestPollOnceBuildsSnapshot(t *testing.T) {
 	_, a := fleettest.NewVarmq(t, "host", map[string]*fleettest.Worker{
 		"orders": {Status: "Running", Concurrency: 8, Busy: 6, Pending: 80, Idle: 2,
-			Submitted: 100, Completed: 90, Successful: 85, Failed: 5},
+			Completed: 90, Successful: 85, Failed: 5},
 	})
 	_, b := fleettest.NewVarmq(t, "host", map[string]*fleettest.Worker{
-		"orders":  {Status: "Paused", Concurrency: 4, Pending: 7, Idle: 4, Submitted: 3, Completed: 2, Successful: 2},
+		"orders":  {Status: "Paused", Concurrency: 4, Pending: 7, Idle: 4, Completed: 2, Successful: 2},
 		"billing": {Status: "Idle", Concurrency: 1, Idle: 1},
 	})
 	wantEPs := []string{a.URL, b.URL}
@@ -121,7 +121,7 @@ func TestPollOnceBuildsSnapshot(t *testing.T) {
 		Name: "orders", Status: "Running", Concurrency: 8, NumProcessing: 6, NumPending: 80, NumIdleWorkers: 2,
 		Origin: a.URL, Instance: "host",
 	}}
-	want.Metrics.Submitted, want.Metrics.Completed, want.Metrics.Successful, want.Metrics.Failed = 100, 90, 85, 5
+	want.Metrics.Completed, want.Metrics.Successful, want.Metrics.Failed = 90, 85, 5
 	if !reflect.DeepEqual(sm.WorkerDetails, want) {
 		t.Errorf("sample = %+v\nwant     %+v", sm.WorkerDetails, want)
 	}