From aa789481a1eb6dbaa019dad55fe2abba1fb68f50 Mon Sep 17 00:00:00 2001
From: dawn <dawn@tangled.org>
Date: Wed, 21 Oct 2026 18:44:44 +0300
Subject: [PATCH] spindle/db: dequeue jobs by repo instead of scanning the
 queue

Signed-off-by: dawn <dawn@tangled.org>
---
 spindle/db/db.go              |  6 ++++
 spindle/db/jobs.go            | 18 +++++++----
 spindle/db/jobs_depth_test.go | 60 +++++++++++++++++++++++++++++++++++
 3 files changed, 77 insertions(+), 7 deletions(-)
 create mode 100644 spindle/db/jobs_depth_test.go

diff --git a/spindle/db/db.go b/spindle/db/db.go
index 4c2e7f58d..49b91fee6 100644
--- a/spindle/db/db.go
+++ b/spindle/db/db.go
@@ -1252,6 +1252,12 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error
 	}); err != nil {
 		return err
 	}
+	if err := orm.RunMigration(conn, logger, "jobs-repo-index", func(tx *sql.Tx) error {
+		_, err := tx.Exec(`create index if not exists idx_jobs_repo_did_id on jobs(repo_did, id)`)
+		return err
+	}); err != nil {
+		return err
+	}
 	return nil
 }
 
diff --git a/spindle/db/jobs.go b/spindle/db/jobs.go
index a310a5ffc..36b61afc7 100644
--- a/spindle/db/jobs.go
+++ b/spindle/db/jobs.go
@@ -116,15 +116,19 @@ func (d *DB) DequeueJob(ctx context.Context, running map[string]int) (*JobRow, e
 	}
 	defer tx.Rollback()
 
+	// walks the distinct repos with the (repo_did, id) index, so the cost follows the repos
+	// that have work queued and not how many jobs they queued
 	rows, err := tx.QueryContext(ctx, `
-		select owner, min(id) from (
-			select id, coalesce(
-				(select r.owner from repos r where r.repo_did = jobs.repo_did limit 1),
-				repo_did
-			) as owner
-			from jobs
+		with recursive queued(repo_did) as (
+			select min(repo_did) from jobs
+			union all
+			select (select min(repo_did) from jobs where repo_did > queued.repo_did)
+			from queued where queued.repo_did is not null
 		)
-		group by owner
+		select
+			coalesce((select r.owner from repos r where r.repo_did = queued.repo_did limit 1), queued.repo_did),
+			(select min(id) from jobs where repo_did = queued.repo_did)
+		from queued where queued.repo_did is not null
 	`)
 	if err != nil {
 		return nil, err
diff --git a/spindle/db/jobs_depth_test.go b/spindle/db/jobs_depth_test.go
new file mode 100644
index 000000000..f95dbde43
--- /dev/null
+++ b/spindle/db/jobs_depth_test.go
@@ -0,0 +1,60 @@
+package db
+
+import (
+	"context"
+	"fmt"
+	"path/filepath"
+	"slices"
+	"testing"
+	"time"
+)
+
+// raw sql insert, EnqueuePipeline would take minutes at this depth
+func queueOf(t *testing.T, depth, repos int) *DB {
+	t.Helper()
+	database, err := Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { database.Close() })
+	tx, err := database.Begin()
+	if err != nil {
+		t.Fatal(err)
+	}
+	stmt, err := tx.Prepare(`insert into jobs (repo_did, pipeline_id, tpl, created_at_ns) values (?, ?, '{}', 1)`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for i := range depth {
+		if _, err := stmt.Exec(fmt.Sprintf("did:plc:repo%04d", i%repos), fmt.Sprintf("p%d", i)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	return database
+}
+
+func medianDequeue(t *testing.T, d *DB, n int) time.Duration {
+	t.Helper()
+	var took []time.Duration
+	for range n {
+		start := time.Now()
+		if job, err := d.DequeueJob(context.Background(), nil); err != nil || job == nil {
+			t.Fatalf("dequeue = %v, %v", job, err)
+		}
+		took = append(took, time.Since(start))
+	}
+	slices.Sort(took)
+	return took[len(took)/2]
+}
+
+func TestDequeueCostDoesNotFollowTheQueueDepth(t *testing.T) {
+	shallow := medianDequeue(t, queueOf(t, 500, 200), 9)
+	deep := medianDequeue(t, queueOf(t, 100_000, 200), 9)
+	t.Logf("median dequeue %v at depth 500, %v at depth 100000 over 200 repos", shallow, deep)
+	if deep > 8*shallow+5*time.Millisecond {
+		t.Fatalf("median dequeue is %v at depth 100000 against %v at depth 500, want it independent of depth", deep, shallow)
+	}
+}
-- 
2.54.0