From 365aa08e1b4f7d5d0126cd0d09d07cb945471e9e Mon Sep 17 00:00:00 2001 From: dawn Date: Wed, 21 Oct 2026 18:44:44 +0300 Subject: [PATCH] spindle/db: index pipelines by repo and kind Signed-off-by: dawn --- spindle/db/db.go | 6 +++ spindle/db/pipelines.go | 22 +++++++---- spindle/db/pipelines_kind_test.go | 60 ++++++++++++++++++++++++++++++ spindle/db/pipelines_total_test.go | 35 +++++++++++++++++ 4 files changed, 116 insertions(+), 7 deletions(-) create mode 100644 spindle/db/pipelines_kind_test.go create mode 100644 spindle/db/pipelines_total_test.go diff --git a/spindle/db/db.go b/spindle/db/db.go index 49b91fee6..d56be2035 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -1258,6 +1258,12 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error }); err != nil { return err } + if err := orm.RunMigration(conn, logger, "pipelines-repo-kind-index", func(tx *sql.Tx) error { + _, err := tx.Exec(`create index if not exists idx_pipelines_repo_kind_id on pipelines(repo_did, kind, id)`) + return err + }); err != nil { + return err + } return nil } diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go index 2cf58a5b0..900faaf95 100644 --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -33,18 +33,26 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDID string, commits []strin } whereClause := " where " + strings.Join(conditions, " and ") - var total int64 - if err := d.QueryRowContext(ctx, - `select count(*) from pipelines`+whereClause, args...).Scan(&total); err != nil { - return nil, "", 0, err - } + baseWhere, baseArgs := whereClause, args + // the cursor carries the total, so later pages skip the count + var total int64 + counted := false if cursor != "" { - if value, err := strconv.ParseInt(cursor, 10, 64); err == nil { + idText, totalText, _ := strings.Cut(cursor, ":") + if value, err := strconv.ParseInt(idText, 10, 64); err == nil { filter := orm.FilterLt("id", value) conditions = append(conditions, filter.Condition()) args = append(args, filter.Arg()...) whereClause = " where " + strings.Join(conditions, " and ") + if carried, err := strconv.ParseInt(totalText, 10, 64); err == nil { + total, counted = carried, true + } + } + } + if !counted { + if err := d.QueryRowContext(ctx, `select count(*) from pipelines`+baseWhere, baseArgs...).Scan(&total); err != nil { + return nil, "", 0, err } } args = append(args, limit) @@ -82,7 +90,7 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDID string, commits []strin nextCursor := "" if scanned == limit { - nextCursor = strconv.FormatInt(lastID, 10) + nextCursor = strconv.FormatInt(lastID, 10) + ":" + strconv.FormatInt(total, 10) } return records, nextCursor, total, nil } diff --git a/spindle/db/pipelines_kind_test.go b/spindle/db/pipelines_kind_test.go new file mode 100644 index 000000000..6e33bc4c0 --- /dev/null +++ b/spindle/db/pipelines_kind_test.go @@ -0,0 +1,60 @@ +package db + +import ( + "context" + "fmt" + "path/filepath" + "slices" + "testing" + "time" +) + +func historyOf(t *testing.T, pipelines 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 pipelines (pipeline_id, repo_did, commit_sha, kind, payload) values (?, 'did:plc:history', ?, 'push', '{}')`) + if err != nil { + t.Fatal(err) + } + for i := range pipelines { + if _, err := stmt.Exec(fmt.Sprintf("p%d", i), fmt.Sprintf("%040x", i)); err != nil { + t.Fatal(err) + } + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + return database +} + +func medianRareKindQuery(t *testing.T, d *DB) time.Duration { + t.Helper() + var took []time.Duration + for range 9 { + start := time.Now() + records, _, total, err := d.QueryPipelines(context.Background(), "did:plc:history", nil, "", []string{"manual"}, 30) + took = append(took, time.Since(start)) + if err != nil || len(records) != 0 || total != 0 { + t.Fatalf("query = %d records, total %d, %v, want none", len(records), total, err) + } + } + slices.Sort(took) + return took[len(took)/2] +} + +func TestKindFilterCostDoesNotFollowTheHistory(t *testing.T) { + shallow := medianRareKindQuery(t, historyOf(t, 500)) + deep := medianRareKindQuery(t, historyOf(t, 100_000)) + t.Logf("median rare kind query %v over 500 pipelines, %v over 100000", shallow, deep) + if deep > 8*shallow+3*time.Millisecond { + t.Fatalf("a kind filter matching nothing takes %v over 100000 pipelines against %v over 500, want it independent of history", deep, shallow) + } +} diff --git a/spindle/db/pipelines_total_test.go b/spindle/db/pipelines_total_test.go new file mode 100644 index 000000000..21607eb77 --- /dev/null +++ b/spindle/db/pipelines_total_test.go @@ -0,0 +1,35 @@ +package db + +import ( + "context" + "fmt" + "strings" + "testing" +) + +func TestTotalIsCountedOnceForAListing(t *testing.T) { + d := newTestDB(t) + ctx := context.Background() + const repo = "did:plc:totals" + for i := range 5 { + seedPipeline(t, d, fmt.Sprintf("early-%d", i), repo, "push") + } + + _, cursor, total, err := d.QueryPipelines(ctx, repo, nil, "", nil, 2) + if err != nil || total != 5 || cursor == "" { + t.Fatalf("first page: total %d, cursor %q (%v), want 5 and a cursor", total, cursor, err) + } + for i := range 3 { + seedPipeline(t, d, fmt.Sprintf("late-%d", i), repo, "push") + } + + _, _, total, err = d.QueryPipelines(ctx, repo, nil, cursor, nil, 2) + if err != nil || total != 5 { + t.Fatalf("second page total = %d (%v), want the 5 the first page counted", total, err) + } + + plain, _, _ := strings.Cut(cursor, ":") + if _, _, total, err = d.QueryPipelines(ctx, repo, nil, plain, nil, 2); err != nil || total != 8 { + t.Fatalf("plain cursor total = %d (%v), want 8", total, err) + } +} -- 2.54.0