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.
This commit is contained in:
mithral
parent
f680e1d9f6
commit
c50b3cfb3b
14 changed files with 750 additions and 114 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
now.Add(ttl).Format(time.RFC3339Nano),
|
||||
core.Stamp(now), core.Stamp(now),
|
||||
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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
"busy_timeout(5000)", // wait for the writer instead of failing
|
||||
"journal_mode(WAL)", // concurrent readers during a scan
|
||||
// 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
|
||||
|
|
|
|||
31
internal/core/stamp.go
Normal file
31
internal/core/stamp.go
Normal file
|
|
@ -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())
|
||||
}
|
||||
94
internal/core/stamp_test.go
Normal file
94
internal/core/stamp_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
defer ticker.Stop()
|
||||
workers := q.Workers
|
||||
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:
|
||||
// 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
|
||||
}
|
||||
}
|
||||
case <-time.After(q.idleWait()):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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 {
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
Scanned int `json:"scanned"`
|
||||
Added int `json:"added"`
|
||||
Updated int `json:"updated"`
|
||||
Restored int `json:"restored"`
|
||||
Missing int `json:"missing"`
|
||||
Unchanged int `json:"unchanged"`
|
||||
Duration int64 `json:"duration_ms"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
LibraryID int64 `json:"library_id"`
|
||||
Scanned int `json:"scanned"`
|
||||
Added int `json:"added"`
|
||||
Updated int `json:"updated"`
|
||||
Restored int `json:"restored"`
|
||||
Missing int `json:"missing"`
|
||||
Unchanged int `json:"unchanged"`
|
||||
Duration int64 `json:"duration_ms"`
|
||||
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)
|
||||
if err != nil {
|
||||
return res, err
|
||||
}
|
||||
defer tx.Rollback() //nolint:errcheck
|
||||
|
||||
now := time.Now().UTC().Format(time.RFC3339Nano)
|
||||
now := core.Now()
|
||||
seenFiles := make(map[string]bool, len(found))
|
||||
|
||||
for _, key := range sortedGroupKeys(groups) {
|
||||
g := groups[key]
|
||||
// report writes the running result back to the job (if any) so the UI can
|
||||
// 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))
|
||||
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++
|
||||
}
|
||||
return res, err
|
||||
}
|
||||
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 {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", key, err))
|
||||
continue
|
||||
missingItemKeys = append(missingItemKeys, key)
|
||||
}
|
||||
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 {
|
||||
res.Errors = append(res.Errors, fmt.Sprintf("%s: %v", rel, err))
|
||||
}
|
||||
missingFilePaths = append(missingFilePaths, rel)
|
||||
}
|
||||
sort.Strings(missingFilePaths)
|
||||
|
||||
if err := tx.Commit(); err != nil {
|
||||
return res, fmt.Errorf("library: commit scan: %w", err)
|
||||
for start := 0; start < len(missingFilePaths); start += scanBatchSize {
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
// cannot drown out the rest.
|
||||
// discovery rather than a repeat. The shelf is sampled per kind: each kind
|
||||
// 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 := `
|
||||
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 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 * 2654435761) % 1000, i.id
|
||||
LIMIT ?`
|
||||
|
||||
rows, err := db.QueryContext(ctx, query, args...)
|
||||
// One small grouped query tells us which kinds are actually present, so we
|
||||
// can give each of them a fair share of the shelf. Kinds with no visible
|
||||
// items are ignored.
|
||||
kindRows, err := db.QueryContext(ctx, `
|
||||
SELECT kind, count(*)
|
||||
FROM media_items
|
||||
WHERE missing = 0 AND library_id IN (`+in+`)
|
||||
GROUP BY kind`, inArgs...)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("media: explore: %w", err)
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := []shelfItem{}
|
||||
for rows.Next() {
|
||||
var s shelfItem
|
||||
if err := rows.Scan(&s.ID, &s.LibraryID, &s.Kind, &s.Title, &s.Percent, &s.UpdatedAt); err != nil {
|
||||
kinds := []string{}
|
||||
for kindRows.Next() {
|
||||
var kind string
|
||||
var n int64
|
||||
if err := kindRows.Scan(&kind, &n); 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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
25
migrations/0008_query_indexes.sql
Normal file
25
migrations/0008_query_indexes.sql
Normal file
|
|
@ -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.
|
||||
Loading…
Add table
Reference in a new issue