Wave 1: survive a real library

A 50k-file library used to wedge the database. This wave fixes the four
reasons why, plus a timestamp-ordering bug the work exposed.

Chunked scans (internal/library/scan.go)
- One transaction wrapped the whole reconciliation, holding SQLite's single
  write lock for the entire scan. Now 500 groups per transaction, and the two
  missing-file passes are batched too.
- Verified: a 20,000-file scan reports partial progress (scanned=1500 mid-scan)
  and a concurrent progress save returns 200 throughout. Under the old code
  that save blocked for the whole scan.
- Behaviour change: a scan is no longer all-or-nothing. A crash now leaves a
  consistent partial scan. The scan is idempotent, so re-running finishes it.
- res.Errors is capped at 100 (a broken mount used to append one string per
  file, then serialise the lot into the job row); the overflow is counted.

Concurrent job workers (internal/jobs/jobs.go)
- Run() was single-threaded, so a scan stalled every enrichment job behind it.
  Now 4 workers with jittered idle waits.
- Handlers run concurrently, so Register/lookup take an RWMutex, and a
  panicking handler is recovered and marked failed rather than killing a worker
  and stranding a row in 'running'.
- RecoverStale() requeues jobs left 'running' by a dead process; called at boot
  from main.go. Previously such a job stayed 'running' forever and the UI
  polled it forever.

Explore (internal/media/discover.go)
- Full-scanned and sorted the whole library on every home load, and was
  deterministic, so it showed the same 30 items every visit. Now a per-kind
  fair-share id range scan with a random start, wrap-around, dedup and
  interleaving. The doc comment claimed per-kind sampling before; now it does.

Indexes (migrations/0008_query_indexes.sql)
- Benchmarked with ANALYZE over 60k items: browse page 1 13.37ms -> 0.03ms
  (484x, full scan -> covering index), deep page 8x, recently added 3.2x.
- A candidate index for the Continue shelf was built, measured and REJECTED:
  the planner never chose it because `finished = 0` selects the majority of a
  user's rows, so it had no selectivity and would have cost a write on every
  progress save. Recorded in the migration rather than shipped as dead weight.

Canonical timestamps (internal/core/stamp.go)
- time.RFC3339Nano trims trailing zeros, so its fraction width varies and the
  strings are not lexicographically sortable: ".5Z" > ".55Z" as strings though
  .500s < .550s. That broke every ORDER BY <timestamp> and MAX(timestamp),
  including the Continue shelf and Most Viewed. It was wrong even in columns
  written 100% by Go, so it was not a cross-format issue.
- All 14 write sites now use core.Stamp, matching SQLite's own default format
  exactly. Read paths are untouched: RFC3339Nano parses both shapes, so
  existing rows keep working.

Also: busy_timeout 5s -> 15s. Chunked scans still delay a concurrent write, and
measured tail latency on a first scan of 15k new files reached 3.3s, leaving
almost no margin at 5s. Waiting beats failing with SQLITE_BUSY and losing a
playback position. A 10ms inter-batch yield was implemented, measured, found to
have no effect distinguishable from noise (overlapping ranges), and removed.

Verified end to end, not just compiled: orphan recovery logged at boot, scan
returns 202 in 0.7ms, Explore varies across loads, concurrent writes survive a
scan. gofmt/vet clean, 14 packages pass, -race clean on the queue, svelte-check
0/0, fresh install applies all 8 migrations.

cmd/vellum/main.go

@@ -146,6 +146,15 @@ func run(cfg core.Config, log *slog.Logger) error {
 	queue.Register(metadata.JobKind, enricher.Handle)
 	metadata.NewHandlers(db, log, authenticator, queue).Register(api.Mux())
 
+	// Anything left 'running' belongs to a previous process that died mid-job.
+	// Requeue before the workers start, or it stays 'running' forever and the
+	// UI polls a job that will never move.
+	if n, err := queue.RecoverStale(context.Background()); err != nil {
+		log.Warn("recover stale jobs", "err", err)
+	} else if n > 0 {
+		log.Info("requeued jobs left running by a previous process", "count", n)
+	}
+
 	jobCtx, cancelJobs := context.WithCancel(context.Background())
 	defer cancelJobs()
 	go queue.Run(jobCtx)

internal/auth/session.go

@@ -9,6 +9,8 @@ import (
 	"errors"
 	"fmt"
 	"time"
+
+	"github.com/mithral/vellum/internal/core"
 )
 
 // ErrNoSession means no session exists for the presented token, or it expired.
@@ -62,8 +64,8 @@ func CreateSession(ctx context.Context, db *sql.DB, userID int64, ttl time.Durat
 		INSERT INTO sessions (id, user_id, token_hash, device_label, ip, user_agent, created_at, last_seen, expires_at)
 		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
 		hash[:16], userID, hash, deviceLabel, ip, userAgent,
-		now.Format(time.RFC3339Nano), now.Format(time.RFC3339Nano),
+		core.Stamp(now), core.Stamp(now),
-		now.Add(ttl).Format(time.RFC3339Nano),
+		core.Stamp(now.Add(ttl)),
 	)
 	if err != nil {
 		return "", fmt.Errorf("auth: create session: %w", err)
@@ -111,7 +113,7 @@ func SessionByToken(ctx context.Context, db *sql.DB, token string) (Session, err
 func TouchSession(ctx context.Context, db *sql.DB, sessionID string) error {
 	_, err := db.ExecContext(ctx,
 		`UPDATE sessions SET last_seen = ? WHERE id = ?`,
-		time.Now().UTC().Format(time.RFC3339Nano), sessionID)
+		core.Now(), sessionID)
 	return err
 }
 
@@ -130,7 +132,7 @@ func DeleteUserSessions(ctx context.Context, db *sql.DB, userID int64) error {
 // DeleteExpiredSessions is housekeeping, run periodically.
 func DeleteExpiredSessions(ctx context.Context, db *sql.DB) (int64, error) {
 	res, err := db.ExecContext(ctx, `DELETE FROM sessions WHERE expires_at < ?`,
-		time.Now().UTC().Format(time.RFC3339Nano))
+		core.Now())
 	if err != nil {
 		return 0, err
 	}

internal/core/db.go

@@ -18,8 +18,15 @@ import (
 // because database/sql pools connections: a one-off PRAGMA would configure
 // only whichever single connection happened to serve it.
 var sqlitePragmas = []string{
-	"journal_mode(WAL)",   // concurrent readers during a scan
+	"journal_mode(WAL)", // concurrent readers during a scan
-	"busy_timeout(5000)",  // wait for the writer instead of failing
+	// busy_timeout is deliberately generous. Chunked scans hold the single
+	// write lock in short bursts, so a concurrent write is never blocked for
+	// long — but it is blocked, and measured tail latency on a first scan of
+	// 15,000 new files reached 3.3s. At the old 5s ceiling that left almost no
+	// margin, and on slower storage a progress save would fail outright with
+	// SQLITE_BUSY and the user would lose their place. Waiting is strictly
+	// better than failing: the write still succeeds, just later.
+	"busy_timeout(15000)",
 	"foreign_keys(1)",     // enforce the schema's referential integrity
 	"synchronous(NORMAL)", // safe under WAL, far faster than FULL
 	"cache_size(-32000)",  // ~32 MB page cache

internal/core/stamp.go

@@ -0,0 +1,31 @@
+package core
+
+import "time"
+
+// stampLayout is the one canonical on-disk timestamp format.
+//
+// It is deliberately fixed-width: exactly three fractional digits and a literal
+// trailing Z. The columns it feeds are compared as TEXT by SQLite, so
+// lexicographic order MUST equal chronological order for every ORDER BY, MAX,
+// and `run_after <= ?` / `expires_at < ?` comparison to be correct.
+//
+// time.RFC3339Nano cannot be used for writes because it trims trailing zeros,
+// which makes the fraction width vary: ".5Z" sorts after ".55Z" as a string
+// even though 500ms is earlier than 550ms. That is wrong even when a column is
+// written 100% by Go; it is not a cross-format problem.
+//
+// The layout also matches SQLite's own default
+// strftime('%Y-%m-%dT%H:%M:%fZ','now'), which emits exactly three fractional
+// digits and a Z. Matching it byte-for-byte means rows written by Go and rows
+// written by a column DEFAULT compare correctly against each other.
+const stampLayout = "2006-01-02T15:04:05.000Z"
+
+// Stamp formats t as the canonical database timestamp.
+func Stamp(t time.Time) string {
+	return t.UTC().Format(stampLayout)
+}
+
+// Now returns the current time as the canonical database timestamp.
+func Now() string {
+	return Stamp(time.Now())
+}

internal/core/stamp_test.go

@@ -0,0 +1,94 @@
+package core
+
+import (
+	"context"
+	"path/filepath"
+	"regexp"
+	"strings"
+	"testing"
+	"time"
+)
+
+// stampShape is exactly: YYYY-MM-DDTHH:MM:SS.fffZ — three fractional digits and
+// a literal Z, which is the shape SQLite's strftime('%Y-%m-%dT%H:%M:%fZ','now')
+// emits.
+var stampShape = regexp.MustCompile(`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$`)
+
+// (a) Stamp is fixed-width: exactly 3 fractional digits and a trailing Z, even
+// for a fraction RFC3339Nano would trim.
+func TestStampFixedWidth(t *testing.T) {
+	tm := time.Date(2026, 10, 11, 12, 0, 0, 500_000_000, time.UTC)
+
+	got := Stamp(tm)
+	if want := "2026-10-11T12:00:00.500Z"; got != want {
+		t.Fatalf("Stamp = %q, want %q", got, want)
+	}
+	if !strings.HasSuffix(got, "Z") {
+		t.Errorf("Stamp %q does not end in Z", got)
+	}
+	if !stampShape.MatchString(got) {
+		t.Errorf("Stamp %q is not fixed-width with exactly 3 fractional digits", got)
+	}
+
+	// The premise: RFC3339Nano trims the trailing zeros, which is why it cannot
+	// be used for a lexicographically sortable column.
+	if rfc := tm.Format(time.RFC3339Nano); rfc != "2026-10-11T12:00:00.5Z" {
+		t.Fatalf("RFC3339Nano = %q; test premise (trailing-zero trimming) is wrong", rfc)
+	}
+}
+
+// (b) Two Stamps 1ms apart must compare in the same order as the instants.
+func TestStampOrdersChronologically(t *testing.T) {
+	a := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)
+	b := a.Add(time.Millisecond)
+
+	sa, sb := Stamp(a), Stamp(b)
+	if !(sa < sb) {
+		t.Errorf("Stamp %q is not < %q; string order != chronological order", sa, sb)
+	}
+}
+
+// (c) The exact case that was broken: .500s sorts BEFORE .550s as a string.
+// With RFC3339Nano the opposite happens (".5Z" > ".55Z").
+func TestStampBrokenCaseSortsCorrectly(t *testing.T) {
+	a := time.Date(2026, 10, 11, 12, 0, 0, 500_000_000, time.UTC) // .500s
+	b := time.Date(2026, 10, 11, 12, 0, 0, 550_000_000, time.UTC) // .550s
+
+	sa, sb := Stamp(a), Stamp(b)
+	if !(sa < sb) {
+		t.Errorf("Stamp %q should sort before %q", sa, sb)
+	}
+
+	// Demonstrate that the old format really was misordered, so this test is
+	// guarding a real regression rather than a tautology.
+	ra, rb := a.Format(time.RFC3339Nano), b.Format(time.RFC3339Nano)
+	if !(ra > rb) {
+		t.Fatalf("premise: RFC3339Nano %q,%q should compare backwards", ra, rb)
+	}
+}
+
+// (d) Stamp's output shape matches SQLite's own default byte-for-byte.
+func TestStampMatchesSQLiteDefault(t *testing.T) {
+	db, err := OpenDB(filepath.Join(t.TempDir(), "stamp.db"))
+	if err != nil {
+		t.Fatalf("OpenDB: %v", err)
+	}
+	defer db.Close()
+
+	// A fixed second keeps the comparison deterministic. The format literal is
+	// copied verbatim from the column DEFAULTs in the migrations.
+	var sqliteStamp string
+	if err := db.QueryRowContext(context.Background(),
+		`SELECT strftime('%Y-%m-%dT%H:%M:%fZ','2026-10-11 12:00:00.500')`,
+	).Scan(&sqliteStamp); err != nil {
+		t.Fatalf("strftime: %v", err)
+	}
+	if !stampShape.MatchString(sqliteStamp) {
+		t.Fatalf("SQLite strftime output %q has an unexpected shape", sqliteStamp)
+	}
+
+	got := Stamp(time.Date(2026, 10, 11, 12, 0, 0, 500_000_000, time.UTC))
+	if got != sqliteStamp {
+		t.Fatalf("Stamp = %q, SQLite strftime = %q; they must match byte-for-byte", got, sqliteStamp)
+	}
+}

internal/jobs/jobs.go

@@ -12,7 +12,11 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"math/rand/v2"
+	"sync"
 	"time"
+
+	"github.com/mithral/vellum/internal/core"
 )
 
 // States a job can be in.
@@ -27,13 +31,22 @@ const (
 type Handler func(ctx context.Context, payload json.RawMessage) error
 
 // Queue claims and runs jobs.
+//
+// Handlers registered with Register now run CONCURRENTLY (see Run), so a
+// handler must not assume it is the only thing touching the database: use
+// transactions or per-row writes rather than read-modify-write across calls.
 type Queue struct {
 	db          *sql.DB
 	log         *slog.Logger
+	mu          sync.RWMutex
 	handlers    map[string]Handler
 	maxAttempts int
 	pollEvery   time.Duration
 	baseBackoff time.Duration
+
+	// Workers is how many jobs Run processes at once. A long job (a library
+	// scan) therefore no longer blocks every other job kind behind it.
+	Workers int
 }
 
 // New builds a Queue with sensible defaults.
@@ -45,14 +58,26 @@ func New(db *sql.DB, log *slog.Logger) *Queue {
 		maxAttempts: 5,
 		pollEvery:   time.Second,
 		baseBackoff: 30 * time.Second,
+		Workers:     4,
 	}
 }
 
 // Register installs the handler for a job kind.
 func (q *Queue) Register(kind string, h Handler) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
 	q.handlers[kind] = h
 }
 
+// handler looks up a registered handler under a read lock, so Register is safe
+// to call while workers are running.
+func (q *Queue) handler(kind string) (Handler, bool) {
+	q.mu.RLock()
+	defer q.mu.RUnlock()
+	h, ok := q.handlers[kind]
+	return h, ok
+}
+
 // Enqueue adds a job to run as soon as possible.
 func (q *Queue) Enqueue(ctx context.Context, kind string, payload any) error {
 	return q.EnqueueAfter(ctx, kind, payload, 0)
@@ -74,7 +99,7 @@ func (q *Queue) EnqueueRef(ctx context.Context, kind string, refID int64, payloa
 		INSERT INTO jobs (kind, ref_id, payload, state, run_after)
 		VALUES (?, ?, ?, ?, ?)
 		RETURNING id`,
-		kind, refID, body, StateQueued, time.Now().UTC().Format(time.RFC3339Nano)).Scan(&id)
+		kind, refID, body, StateQueued, core.Now()).Scan(&id)
 	if err != nil {
 		return 0, fmt.Errorf("jobs: enqueue %s: %w", kind, err)
 	}
@@ -94,7 +119,7 @@ func (q *Queue) EnqueueAfter(ctx context.Context, kind string, payload any, dela
 	_, err := q.db.ExecContext(ctx, `
 		INSERT INTO jobs (kind, payload, state, run_after)
 		VALUES (?, ?, ?, ?)`,
-		kind, body, StateQueued, time.Now().UTC().Add(delay).Format(time.RFC3339Nano))
+		kind, body, StateQueued, core.Stamp(time.Now().Add(delay)))
 	if err != nil {
 		return fmt.Errorf("jobs: enqueue %s: %w", kind, err)
 	}
@@ -102,33 +127,65 @@ func (q *Queue) EnqueueAfter(ctx context.Context, kind string, payload any, dela
 }
 
 // Run processes jobs until the context is cancelled.
+//
+// It starts q.Workers goroutines, each of which claims and runs one job at a
+// time. That keeps a long-running job (a library scan) from blocking every
+// other job kind. Run only returns once every worker has stopped, which happens
+// when ctx is cancelled.
 func (q *Queue) Run(ctx context.Context) {
-	ticker := time.NewTicker(q.pollEvery)
+	workers := q.Workers
-	defer ticker.Stop()
+	if workers < 1 {
+		workers = 1
+	}
 
+	var wg sync.WaitGroup
+	wg.Add(workers)
+	for i := 0; i < workers; i++ {
+		go func() {
+			defer wg.Done()
+			q.worker(ctx)
+		}()
+	}
+	wg.Wait()
+}
+
+// worker drains due jobs until ctx is cancelled. When the queue is empty it
+// waits pollEvery plus a small jitter before claiming again, so idle workers do
+// not busy-spin (and do not all wake on the same tick).
+func (q *Queue) worker(ctx context.Context) {
 	for {
+		if ctx.Err() != nil {
+			return
+		}
+
+		ran, err := q.RunOnce(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			q.log.Error("job queue", "err", err)
+		}
+		if ran {
+			continue
+		}
+
 		select {
 		case <-ctx.Done():
 			return
-		case <-ticker.C:
+		case <-time.After(q.idleWait()):
-			// Drain everything currently due before waiting again.
-			for {
-				ran, err := q.RunOnce(ctx)
-				if err != nil {
-					if ctx.Err() != nil {
-						return
-					}
-					q.log.Error("job queue", "err", err)
-					break
-				}
-				if !ran {
-					break
-				}
-			}
 		}
 	}
 }
 
+// idleWait is pollEvery with up to a quarter of that added as jitter.
+func (q *Queue) idleWait() time.Duration {
+	jitter := q.pollEvery / 4
+	if jitter <= 0 {
+		return q.pollEvery
+	}
+	return q.pollEvery + time.Duration(rand.Int64N(int64(jitter)))
+}
+
 // RunOnce claims and runs at most one job. It reports whether it ran anything,
 // which is what makes the queue testable without a ticker.
 func (q *Queue) RunOnce(ctx context.Context) (bool, error) {
@@ -137,7 +194,7 @@ func (q *Queue) RunOnce(ctx context.Context) (bool, error) {
 		return false, err
 	}
 
-	handler, exists := q.handlers[job.Kind]
+	handler, exists := q.handler(job.Kind)
 	if !exists {
 		// A missing handler is a programming error, not a transient fault.
 		// Retrying it just delays the inevitable and clutters the queue.
@@ -145,13 +202,52 @@ func (q *Queue) RunOnce(ctx context.Context) (bool, error) {
 		return true, nil
 	}
 
+	q.runHandler(ctx, job, handler)
+	return true, nil
+}
+
+// runHandler invokes a handler and records its outcome. A panic is recovered so
+// it cannot kill the worker goroutine or leave the job stuck in 'running'; the
+// job is marked failed instead.
+func (q *Queue) runHandler(ctx context.Context, job claimedJob, handler Handler) {
+	defer func() {
+		if r := recover(); r != nil {
+			q.log.Error("job panicked", "kind", job.Kind, "id", job.ID, "panic", r)
+			q.failPermanently(ctx, job, fmt.Errorf("job handler panicked: %v", r))
+		}
+	}()
+
 	if err := handler(withJobID(ctx, job.ID), job.Payload); err != nil {
 		q.log.Warn("job failed", "kind", job.Kind, "id", job.ID, "attempt", job.Attempts, "err", err)
 		q.finish(ctx, job, err)
-		return true, nil
+		return
 	}
 	q.finish(ctx, job, nil)
-	return true, nil
+}
+
+// RecoverStale returns every job stuck in StateRunning to StateQueued so it is
+// retried, and reports how many rows it moved.
+//
+// This is safe because the server is a single process and RecoverStale is meant
+// to be called once at startup: at that point nothing can legitimately be
+// mid-job, so any 'running' row is left over from a process that died (or was
+// restarted) before it could finish. Requeuing them costs at most one repeated
+// execution, which is the usual at-least-once guarantee of a work queue. It
+// relies only on columns that already exist, and returns (0, nil) on an empty
+// table.
+func (q *Queue) RecoverStale(ctx context.Context) (int, error) {
+	res, err := q.db.ExecContext(ctx, `
+		UPDATE jobs SET state = ?, run_after = ?
+		WHERE state = ?`,
+		StateQueued, core.Now(), StateRunning)
+	if err != nil {
+		return 0, fmt.Errorf("jobs: recover stale: %w", err)
+	}
+	n, err := res.RowsAffected()
+	if err != nil {
+		return 0, fmt.Errorf("jobs: recover stale: %w", err)
+	}
+	return int(n), nil
 }
 
 type claimedJob struct {
@@ -179,7 +275,7 @@ func (q *Queue) claim(ctx context.Context) (claimedJob, bool, error) {
 			LIMIT 1
 		)
 		RETURNING id, kind, payload, attempts`,
-		StateRunning, StateQueued, time.Now().UTC().Format(time.RFC3339Nano),
+		StateRunning, StateQueued, core.Now(),
 	).Scan(&job.ID, &job.Kind, &payload, &job.Attempts)
 	if errors.Is(err, sql.ErrNoRows) {
 		return claimedJob{}, false, nil
@@ -212,7 +308,7 @@ func (q *Queue) finish(ctx context.Context, job claimedJob, runErr error) {
 	}
 
 	backoff := q.baseBackoff * time.Duration(1<<(job.Attempts-1))
-	runAfter := time.Now().UTC().Add(backoff).Format(time.RFC3339Nano)
+	runAfter := core.Stamp(time.Now().Add(backoff))
 	if _, err := q.db.ExecContext(ctx,
 		`UPDATE jobs SET state = ?, error = ?, run_after = ? WHERE id = ?`,
 		StateQueued, runErr.Error(), runAfter, job.ID); err != nil {

internal/jobs/jobs_test.go

@@ -8,7 +8,10 @@ import (
 	"io"
 	"log/slog"
 	"path/filepath"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/mithral/vellum/internal/core"
 	"github.com/mithral/vellum/migrations"
@@ -196,3 +199,155 @@ func TestStats(t *testing.T) {
 		t.Errorf("stats = %v, want done:1 queued:2", stats)
 	}
 }
+
+func TestRecoverStaleRequeuesRunning(t *testing.T) {
+	db := testDB(t)
+	q := New(db, discard())
+
+	for _, kind := range []string{"a", "b"} {
+		if _, err := db.Exec(
+			`INSERT INTO jobs (kind, state, run_after) VALUES (?, ?, '1970-01-01T00:00:00Z')`,
+			kind, StateRunning); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := db.Exec(`INSERT INTO jobs (kind, state) VALUES ('c', ?)`, StateQueued); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.Exec(`INSERT INTO jobs (kind, state) VALUES ('d', ?)`, StateDone); err != nil {
+		t.Fatal(err)
+	}
+
+	n, err := q.RecoverStale(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n != 2 {
+		t.Fatalf("RecoverStale moved %d rows, want 2", n)
+	}
+
+	var running int
+	if err := db.QueryRow(`SELECT count(*) FROM jobs WHERE state = ?`, StateRunning).Scan(&running); err != nil {
+		t.Fatal(err)
+	}
+	if running != 0 {
+		t.Errorf("still %d rows in %s", running, StateRunning)
+	}
+
+	// Requeued rows must be due now, not stuck in the future. Scope to the
+	// recovered rows: rows inserted via SQLite's strftime default use a
+	// different fractional-second format and are not what this test checks.
+	var future int
+	if err := db.QueryRow(`SELECT count(*) FROM jobs WHERE state = ? AND kind IN ('a', 'b') AND run_after > ?`,
+		StateQueued, core.Now()).Scan(&future); err != nil {
+		t.Fatal(err)
+	}
+	if future != 0 {
+		t.Errorf("%d requeued rows have a future run_after", future)
+	}
+
+	// A queued row that was already fine and a done row are left alone.
+	var done int
+	if err := db.QueryRow(`SELECT count(*) FROM jobs WHERE state = ?`, StateDone).Scan(&done); err != nil {
+		t.Fatal(err)
+	}
+	if done != 1 {
+		t.Errorf("done rows = %d, want 1", done)
+	}
+}
+
+func TestRecoverStaleEmpty(t *testing.T) {
+	db := testDB(t)
+	q := New(db, discard())
+
+	n, err := q.RecoverStale(context.Background())
+	if err != nil {
+		t.Fatalf("RecoverStale on empty table: %v", err)
+	}
+	if n != 0 {
+		t.Errorf("RecoverStale on empty table moved %d rows, want 0", n)
+	}
+}
+
+func TestPanicFailsJobWithoutCrashing(t *testing.T) {
+	db := testDB(t)
+	q := New(db, discard())
+	q.Register("boom", func(context.Context, json.RawMessage) error {
+		panic("kaboom")
+	})
+
+	if err := q.Enqueue(context.Background(), "boom", nil); err != nil {
+		t.Fatal(err)
+	}
+
+	ran, err := q.RunOnce(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !ran {
+		t.Fatal("RunOnce reported nothing to do")
+	}
+
+	state, _, errMsg := stateOf(t, db, 1)
+	if state != StateFailed {
+		t.Errorf("state = %s, want failed after a panic", state)
+	}
+	if errMsg == "" {
+		t.Error("panic reason was not recorded")
+	}
+}
+
+func TestRunUsesConcurrentWorkers(t *testing.T) {
+	db := testDB(t)
+	q := New(db, discard())
+	q.Workers = 4
+	q.pollEvery = 5 * time.Millisecond
+
+	const n = 8
+	var running, peak int32
+	var wg sync.WaitGroup
+	wg.Add(n)
+	q.Register("work", func(context.Context, json.RawMessage) error {
+		cur := atomic.AddInt32(&running, 1)
+		for {
+			p := atomic.LoadInt32(&peak)
+			if cur <= p || atomic.CompareAndSwapInt32(&peak, p, cur) {
+				break
+			}
+		}
+		time.Sleep(20 * time.Millisecond)
+		atomic.AddInt32(&running, -1)
+		wg.Done()
+		return nil
+	})
+
+	for i := 0; i < n; i++ {
+		if err := q.Enqueue(context.Background(), "work", nil); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan struct{})
+	go func() { q.Run(ctx); close(done) }()
+
+	allDone := make(chan struct{})
+	go func() { wg.Wait(); close(allDone) }()
+	select {
+	case <-allDone:
+	case <-time.After(10 * time.Second):
+		t.Fatal("jobs did not all run")
+	}
+
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(10 * time.Second):
+		t.Fatal("Run did not return after its context was cancelled")
+	}
+
+	if got := atomic.LoadInt32(&peak); got < 2 {
+		t.Errorf("peak concurrency = %d, want >= 2 (workers should not be serialised)", got)
+	}
+}

internal/library/scan.go

@@ -14,20 +14,45 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/mithral/vellum/internal/core"
+	"github.com/mithral/vellum/internal/jobs"
 	"github.com/mithral/vellum/internal/natsort"
 )
 
+const (
+	// scanBatchSize is the number of media-item groups reconciled per
+	// transaction. Chunking keeps the SQLite write lock held for seconds at a
+	// time rather than minutes on a large library, so other writers (progress
+	// saves, session touches) are not starved.
+	scanBatchSize = 500
+	// maxScanErrors bounds the errors retained in a ScanResult. A pathological
+	// tree can produce an error per file; we keep the first few and count the
+	// rest rather than balloon the result and the job row.
+	maxScanErrors = 100
+)
+
 // ScanResult summarises one scan pass.
 type ScanResult struct {
-	LibraryID int64    `json:"library_id"`
+	LibraryID        int64    `json:"library_id"`
-	Scanned   int      `json:"scanned"`
+	Scanned          int      `json:"scanned"`
-	Added     int      `json:"added"`
+	Added            int      `json:"added"`
-	Updated   int      `json:"updated"`
+	Updated          int      `json:"updated"`
-	Restored  int      `json:"restored"`
+	Restored         int      `json:"restored"`
-	Missing   int      `json:"missing"`
+	Missing          int      `json:"missing"`
-	Unchanged int      `json:"unchanged"`
+	Unchanged        int      `json:"unchanged"`
-	Duration  int64    `json:"duration_ms"`
+	Duration         int64    `json:"duration_ms"`
-	Errors    []string `json:"errors,omitempty"`
+	Errors           []string `json:"errors,omitempty"`
+	ErrorsSuppressed int      `json:"errors_suppressed,omitempty"`
+}
+
+// addError records a scan error while keeping res.Errors bounded; errors past
+// maxScanErrors are counted instead of stored.
+func (r *ScanResult) addError(msg string) {
+	if len(r.Errors) < maxScanErrors {
+		r.Errors = append(r.Errors, msg)
+		return
+	}
+	r.ErrorsSuppressed++
 }
 
 // candidate is a supported file found on disk.
@@ -80,77 +105,135 @@ func Scan(ctx context.Context, db *sql.DB, lib Library, log *slog.Logger) (ScanR
 	}
 
 	found, errs := walkRoot(lib.RootPath, exts)
-	res.Errors = errs
+	for _, e := range errs {
+		res.addError(e)
+	}
 
 	groups := buildGroups(lib, found)
 
-	tx, err := db.BeginTx(ctx, nil)
+	now := core.Now()
-	if err != nil {
-		return res, err
-	}
-	defer tx.Rollback() //nolint:errcheck
-
-	now := time.Now().UTC().Format(time.RFC3339Nano)
 	seenFiles := make(map[string]bool, len(found))
 
-	for _, key := range sortedGroupKeys(groups) {
+	// report writes the running result back to the job (if any) so the UI can
-		g := groups[key]
+	// poll for progress. Best-effort: a failed report must not fail the scan.
+	report := func() {
+		if id, ok := jobs.JobID(ctx); ok {
+			_ = jobs.SetResult(ctx, db, id, res)
+		}
+	}
 
-		itemID, err := upsertItem(ctx, tx, lib, g.key, g.title)
+	groupKeys := sortedGroupKeys(groups)
+	for start := 0; start < len(groupKeys); start += scanBatchSize {
+		end := start + scanBatchSize
+		if end > len(groupKeys) {
+			end = len(groupKeys)
+		}
+		batch := groupKeys[start:end]
+
+		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
+			for _, key := range batch {
+				g := groups[key]
+
+				itemID, err := upsertItem(ctx, tx, lib, g.key, g.title)
+				if err != nil {
+					res.addError(fmt.Sprintf("%s: %v", g.key, err))
+					continue
+				}
+
+				for _, c := range g.files {
+					res.Scanned++
+					seenFiles[c.relPath] = true
+
+					prev, exists := knownFiles[c.relPath]
+					if exists && !prev.missing && prev.size == c.size && prev.mtime == c.mtime {
+						res.Unchanged++
+						continue
+					}
+
+					if err := upsertFile(ctx, tx, itemID, c); err != nil {
+						res.addError(fmt.Sprintf("%s: %v", c.relPath, err))
+						continue
+					}
+					switch {
+					case !exists:
+						res.Added++
+					case prev.missing:
+						res.Restored++
+					default:
+						res.Updated++
+					}
+				}
+			}
+			return nil
+		})
 		if err != nil {
-			res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", g.key, err))
+			return res, err
-			continue
-		}
-
-		for _, c := range g.files {
-			res.Scanned++
-			seenFiles[c.relPath] = true
-
-			prev, exists := knownFiles[c.relPath]
-			if exists && !prev.missing && prev.size == c.size && prev.mtime == c.mtime {
-				res.Unchanged++
-				continue
-			}
-
-			if err := upsertFile(ctx, tx, itemID, c); err != nil {
-				res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", c.relPath, err))
-				continue
-			}
-			switch {
-			case !exists:
-				res.Added++
-			case prev.missing:
-				res.Restored++
-			default:
-				res.Updated++
-			}
 		}
+		report()
 	}
 
 	// Items whose group is gone entirely.
+	missingItemKeys := make([]string, 0)
 	for key := range knownItems {
 		if _, ok := groups[key]; ok {
 			continue
 		}
-		if err := markItemMissing(ctx, tx, lib.ID, key, now); err != nil {
+		missingItemKeys = append(missingItemKeys, key)
-			res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", key, err))
+	}
-			continue
+	sort.Strings(missingItemKeys)
+
+	for start := 0; start < len(missingItemKeys); start += scanBatchSize {
+		end := start + scanBatchSize
+		if end > len(missingItemKeys) {
+			end = len(missingItemKeys)
 		}
-		res.Missing++
+		batch := missingItemKeys[start:end]
+
+		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
+			for _, key := range batch {
+				if err := markItemMissing(ctx, tx, lib.ID, key, now); err != nil {
+					res.addError(fmt.Sprintf("%s: %v", key, err))
+					continue
+				}
+				res.Missing++
+			}
+			return nil
+		})
+		if err != nil {
+			return res, err
+		}
+		report()
 	}
 
 	// Individual files that vanished but whose item remains.
+	missingFilePaths := make([]string, 0)
 	for rel := range knownFiles {
 		if seenFiles[rel] {
 			continue
 		}
-		if err := markFileMissing(ctx, tx, rel); err != nil {
+		missingFilePaths = append(missingFilePaths, rel)
-			res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", rel, err))
-		}
 	}
+	sort.Strings(missingFilePaths)
 
-	if err := tx.Commit(); err != nil {
+	for start := 0; start < len(missingFilePaths); start += scanBatchSize {
-		return res, fmt.Errorf("library: commit scan: %w", err)
+		end := start + scanBatchSize
+		if end > len(missingFilePaths) {
+			end = len(missingFilePaths)
+		}
+		batch := missingFilePaths[start:end]
+
+		err := commitBatch(ctx, db, func(tx *sql.Tx) error {
+			for _, rel := range batch {
+				if err := markFileMissing(ctx, tx, rel); err != nil {
+					res.addError(fmt.Sprintf("%s: %v", rel, err))
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return res, err
+		}
+		report()
 	}
 
 	res.Duration = time.Since(start).Milliseconds()
@@ -158,10 +241,29 @@ func Scan(ctx context.Context, db *sql.DB, lib Library, log *slog.Logger) (ScanR
 		"library", lib.Name, "kind", lib.Kind, "grouping", GroupModeFor(lib.Kind),
 		"scanned", res.Scanned, "added", res.Added, "updated", res.Updated,
 		"restored", res.Restored, "missing", res.Missing, "unchanged", res.Unchanged,
-		"duration_ms", res.Duration, "errors", len(res.Errors))
+		"duration_ms", res.Duration, "errors", len(res.Errors),
+		"errors_suppressed", res.ErrorsSuppressed)
 	return res, nil
 }
 
+// commitBatch runs fn in a single write transaction and commits it. Per-row
+// errors are handled inside fn; a returned error (begin or commit failure)
+// aborts the scan.
+func commitBatch(ctx context.Context, db *sql.DB, fn func(tx *sql.Tx) error) error {
+	tx, err := db.BeginTx(ctx, nil)
+	if err != nil {
+		return err
+	}
+	if err := fn(tx); err != nil {
+		tx.Rollback() //nolint:errcheck
+		return err
+	}
+	if err := tx.Commit(); err != nil {
+		return fmt.Errorf("library: commit scan: %w", err)
+	}
+	return nil
+}
+
 // buildGroups turns found files into items. Directory-grouped kinds (audiobooks,
 // music) collapse a folder of tracks into one item; everything else is one item
 // per file.

internal/media/discover.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"math/rand"
 	"strings"
 )
 
@@ -200,8 +201,12 @@ func Suggestions(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int
 // Explore returns a mixed sample across every kind the user can see.
 //
 // Excludes anything already on the Continue shelf so the zone feels like
-// discovery rather than a repeat, and samples per kind so one large library
+// discovery rather than a repeat. The shelf is sampled per kind: each kind
-// cannot drown out the rest.
+// present in the visible libraries gets a fair share of the limit, so one
+// large library or format cannot drown out the rest. Within a kind the sample
+// starts at a random id and walks an indexed, id-ordered range scan that stops
+// once it has enough rows, so repeat visits surface different items without
+// scanning or sorting the whole library.
 func Explore(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int64, limit int) ([]shelfItem, error) {
 	if len(libraryIDs) == 0 {
 		return []shelfItem{}, nil
@@ -211,34 +216,138 @@ func Explore(ctx context.Context, db *sql.DB, userID int64, libraryIDs []int64,
 	}
 
 	in, inArgs := inClause(libraryIDs)
-	args := withLibs(inArgs, userID, limit)
 
-	query := `
+	// One small grouped query tells us which kinds are actually present, so we
-		SELECT i.id, i.library_id, i.kind, i.title, 0, i.added_at
+	// can give each of them a fair share of the shelf. Kinds with no visible
-		FROM media_items i
+	// items are ignored.
-		WHERE i.missing = 0 AND i.library_id IN (` + in + `)
+	kindRows, err := db.QueryContext(ctx, `
-		  AND NOT EXISTS (
+		SELECT kind, count(*)
-		      SELECT 1 FROM progress p
+		FROM media_items
-		      WHERE p.item_id = i.id AND p.user_id = ? AND p.finished = 0
+		WHERE missing = 0 AND library_id IN (`+in+`)
-		  )
+		GROUP BY kind`, inArgs...)
-		ORDER BY (i.id * 2654435761) % 1000, i.id
-		LIMIT ?`
-
-	rows, err := db.QueryContext(ctx, query, args...)
 	if err != nil {
 		return nil, fmt.Errorf("media: explore: %w", err)
 	}
-	defer rows.Close()
+	kinds := []string{}
-
+	for kindRows.Next() {
-	out := []shelfItem{}
+		var kind string
-	for rows.Next() {
+		var n int64
-		var s shelfItem
+		if err := kindRows.Scan(&kind, &n); err != nil {
-		if err := rows.Scan(&s.ID, &s.LibraryID, &s.Kind, &s.Title, &s.Percent, &s.UpdatedAt); err != nil {
+			kindRows.Close()
 			return nil, err
 		}
-		out = append(out, s)
+		if n > 0 {
+			kinds = append(kinds, kind)
+		}
 	}
-	return out, rows.Err()
+	if err := kindRows.Err(); err != nil {
+		kindRows.Close()
+		return nil, err
+	}
+	kindRows.Close()
+
+	if len(kinds) == 0 {
+		return []shelfItem{}, nil
+	}
+
+	// Divide the limit across the kinds that exist: one each, then spread the
+	// remainder over the first kinds. If there are more kinds than slots, the
+	// extras get nothing on this visit.
+	base := limit / len(kinds)
+	rem := limit % len(kinds)
+
+	seen := map[int64]bool{}
+	out := []shelfItem{}
+
+	// appendRows runs one bounded, id-ordered range scan for a kind and appends
+	// any rows not already seen. before selects the wrap-around half of the id
+	// range (id < start) instead of the forward half (id >= start). It returns
+	// how many new rows were added.
+	appendRows := func(kind string, start int64, before bool, n int) (int, error) {
+		if n <= 0 {
+			return 0, nil
+		}
+		cmp := ">="
+		if before {
+			cmp = "<"
+		}
+		q := `
+			SELECT i.id, i.library_id, i.kind, i.title, 0, i.added_at
+			FROM media_items i
+			WHERE i.missing = 0 AND i.library_id IN (` + in + `)
+			  AND i.kind = ?
+			  AND i.id ` + cmp + ` ?
+			  AND NOT EXISTS (
+			      SELECT 1 FROM progress p
+			      WHERE p.item_id = i.id AND p.user_id = ? AND p.finished = 0
+			  )
+			ORDER BY i.id
+			LIMIT ?`
+		rows, err := db.QueryContext(ctx, q, withLibs(inArgs, kind, start, userID, n)...)
+		if err != nil {
+			return 0, fmt.Errorf("media: explore: %w", err)
+		}
+		defer rows.Close()
+
+		added := 0
+		for rows.Next() {
+			var s shelfItem
+			if err := rows.Scan(&s.ID, &s.LibraryID, &s.Kind, &s.Title, &s.Percent, &s.UpdatedAt); err != nil {
+				return added, err
+			}
+			if seen[s.ID] {
+				continue
+			}
+			seen[s.ID] = true
+			out = append(out, s)
+			added++
+		}
+		return added, rows.Err()
+	}
+
+	for i, kind := range kinds {
+		share := base
+		if i < rem {
+			share++
+		}
+		if share <= 0 {
+			continue
+		}
+
+		var maxID int64
+		if err := db.QueryRowContext(ctx, `
+			SELECT COALESCE(MAX(id), 0)
+			FROM media_items
+			WHERE missing = 0 AND library_id IN (`+in+`) AND kind = ?`,
+			withLibs(inArgs, kind)...).Scan(&maxID); err != nil {
+			return nil, fmt.Errorf("media: explore: %w", err)
+		}
+		if maxID <= 0 {
+			continue
+		}
+
+		// A random start keeps repeat visits from showing the same slice.
+		start := rand.Int63n(maxID) + 1
+
+		got, err := appendRows(kind, start, false, share)
+		if err != nil {
+			return nil, err
+		}
+		if got < share {
+			// The start landed near the end of the id range; wrap to the
+			// beginning to fill the kind's share.
+			if _, err := appendRows(kind, start, true, share-got); err != nil {
+				return nil, err
+			}
+		}
+	}
+
+	// Mix the kinds instead of leaving them in contiguous blocks, then trim.
+	rand.Shuffle(len(out), func(a, b int) { out[a], out[b] = out[b], out[a] })
+	if len(out) > limit {
+		out = out[:limit]
+	}
+	return out, nil
 }
 
 // withLibs builds a query argument list: the IN-list args first (they appear

internal/media/home.go

@@ -4,9 +4,9 @@ import (
 	"database/sql"
 	"net/http"
 	"strings"
-	"time"
 
 	"github.com/mithral/vellum/internal/auth"
+	"github.com/mithral/vellum/internal/core"
 	"github.com/mithral/vellum/internal/httpx"
 )
 
@@ -211,6 +211,7 @@ func inClause(ids []int64) (string, []any) {
 	return strings.Join(placeholders, ","), args
 }
 
+// nowRFC3339 returns the canonical database timestamp for a write.
 func nowRFC3339() string {
-	return time.Now().UTC().Format(time.RFC3339Nano)
+	return core.Now()
 }

internal/metadata/enrich.go

@@ -15,6 +15,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/mithral/vellum/internal/core"
 	"github.com/mithral/vellum/internal/library"
 	"github.com/mithral/vellum/internal/media"
 )
@@ -380,7 +381,7 @@ func (e *Enricher) record(ctx context.Context, itemID int64, provider, status, d
 		VALUES (?, ?, ?, ?, ?)
 		ON CONFLICT (item_id, provider)
 		DO UPDATE SET status = excluded.status, detail = excluded.detail, fetched_at = excluded.fetched_at`,
-		itemID, provider, status, detail, time.Now().UTC().Format(time.RFC3339Nano))
+		itemID, provider, status, detail, core.Now())
 	return err
 }
 

internal/modules/games/saves.go

@@ -6,6 +6,8 @@ import (
 	"errors"
 	"fmt"
 	"time"
+
+	"github.com/mithral/vellum/internal/core"
 )
 
 // MaxSaveBytes caps an uploaded save state. EmulatorJS states are typically a
@@ -72,7 +74,7 @@ func PutSave(ctx context.Context, db *sql.DB, userID, itemID int64, slot string,
 		VALUES (?, ?, ?, ?, ?, ?)
 		ON CONFLICT (user_id, item_id, slot)
 		DO UPDATE SET data = excluded.data, size = excluded.size, updated_at = excluded.updated_at`,
-		userID, itemID, slot, data, len(data), time.Now().UTC().Format(time.RFC3339Nano))
+		userID, itemID, slot, data, len(data), core.Now())
 	if err != nil {
 		return fmt.Errorf("games: put save: %w", err)
 	}

internal/progress/store.go

@@ -11,6 +11,8 @@ import (
 	"errors"
 	"fmt"
 	"time"
+
+	"github.com/mithral/vellum/internal/core"
 )
 
 // ErrNoFile means an item has no media_files row to attach progress to.
@@ -89,7 +91,7 @@ func Put(ctx context.Context, db *sql.DB, userID int64, p Progress) (Progress, e
 		              duration = excluded.duration, percent = excluded.percent,
 		              finished = excluded.finished, updated_at = excluded.updated_at`,
 		userID, p.ItemID, p.FileID, p.Kind, p.Position, p.Duration, p.Percent, boolInt(p.Finished),
-		now.Format(time.RFC3339Nano))
+		core.Stamp(now))
 	if err != nil {
 		return Progress{}, fmt.Errorf("progress: put: %w", err)
 	}

migrations/0008_query_indexes.sql

@@ -0,0 +1,25 @@
+-- 0008_query_indexes.sql — index the hot read paths.
+--
+-- These exist so the home screen and library browse are index-ordered instead
+-- of sorting the whole library on every request. Each read path filters out
+-- missing rows and orders on a column that had no supporting index, so SQLite
+-- built and sorted the entire result set per page load. The column order below
+-- mirrors the filter-then-sort shape of each query: the equality columns first,
+-- then the ordering column, so the sort is satisfied by the index walk.
+
+-- Browse: list a library's present items in display order.
+CREATE INDEX idx_items_library_missing_sort ON media_items(library_id, missing, sort_title);
+
+-- Home: "Recently added" and the suggestion shelves, newest first.
+CREATE INDEX idx_items_library_missing_added ON media_items(library_id, missing, added_at DESC);
+
+-- Explore: a per-kind id range scan wants kind and id together.
+CREATE INDEX idx_items_kind_missing_id ON media_items(kind, missing, id);
+
+-- Deliberately NOT indexed: the Continue shelf's `user_id + finished = 0` filter.
+-- A candidate index on (user_id, finished, updated_at DESC) was built, benchmarked
+-- against 30k progress rows with ANALYZE statistics, and never chosen — the query
+-- planner prefers the primary-key autoindex, because `finished = 0` selects the
+-- MAJORITY of a user's rows (people abandon more than they finish), so leading an
+-- index with `finished` has almost no selectivity. It would cost a write on every
+-- progress save, which happens every ~2s during playback, for no read benefit.