From af7721ebeb5bee65ae475303457de61cd4199651 Mon Sep 17 00:00:00 2001 From: Harlan Wilton Date: Thu, 13 Aug 2026 14:08:39 +1000 Subject: [PATCH] fix(registry): bound D1 reads and retry resets --- .../registry/server/api/clusters/index.get.ts | 28 +++++++------ .../server/utils/cluster-index-rows.ts | 41 +++++++++++++++++++ server/utils/db.ts | 1 + test/unit/cluster-index-d1-bounds.test.ts | 28 +++++++++++++ test/unit/d1-retry.test.ts | 1 + 5 files changed, 87 insertions(+), 12 deletions(-) create mode 100644 layers/registry/server/utils/cluster-index-rows.ts create mode 100644 test/unit/cluster-index-d1-bounds.test.ts diff --git a/layers/registry/server/api/clusters/index.get.ts b/layers/registry/server/api/clusters/index.get.ts index 77bd8be1..636c6adb 100644 --- a/layers/registry/server/api/clusters/index.get.ts +++ b/layers/registry/server/api/clusters/index.get.ts @@ -9,6 +9,7 @@ import type { Cluster } from '../../data/clusters' import { getDB } from '#server/utils/db' import { CLUSTERS } from '../../data/clusters' +import { findClusterIndexRows } from '../../utils/cluster-index-rows' import { curateClusterSkills, parseClusterSkillKeys } from '../../utils/cluster-skill-curation' interface SkillRow { @@ -42,21 +43,24 @@ export default defineCachedEventHandler(async (event) => { if (!allCategories.length) return { items: [] } - const placeholders = allCategories.map(() => '?').join(',') const allPinnedExamples = parseClusterSkillKeys( Array.from(new Set(CLUSTERS.flatMap(c => c.pinnedExamples))), ).map(skill => skill.key) - const pinnedPlaceholders = allPinnedExamples.map(() => '?').join(',') - const sql = ` - SELECT s.owner, s.name, s.repo, s.display_name, r.stars, - s.abstractness_category AS category - FROM skills s - JOIN repos r ON r.owner = s.owner AND r.repo = s.repo - WHERE (s.is_abstract = 1 AND s.abstractness_category IN (${placeholders})) - OR (s.owner || '/' || s.name IN (${pinnedPlaceholders})) - ` - const res = await db.prepare(sql).bind(...allCategories, ...allPinnedExamples).all() - const rows = res.results ?? [] + const rows = await findClusterIndexRows(async (selector, values) => { + const placeholders = values.map(() => '?').join(',') + const predicate = selector === 'category' + ? `s.is_abstract = 1 AND s.abstractness_category IN (${placeholders})` + : `s.owner || '/' || s.name IN (${placeholders})` + const result = await db.prepare(` + SELECT s.owner, s.name, s.repo, s.display_name, r.stars, + s.abstractness_category AS category + FROM skills s + JOIN repos r ON r.owner = s.owner AND r.repo = s.repo + WHERE ${predicate} + `).bind(...values).all() + + return result.results ?? [] + }, allCategories, allPinnedExamples) const items: ClusterCard[] = CLUSTERS.map((c: Cluster) => { const pinned = new Set(c.pinnedExamples) diff --git a/layers/registry/server/utils/cluster-index-rows.ts b/layers/registry/server/utils/cluster-index-rows.ts new file mode 100644 index 00000000..5b93d176 --- /dev/null +++ b/layers/registry/server/utils/cluster-index-rows.ts @@ -0,0 +1,41 @@ +interface ClusterIndexRowIdentity { + owner: string + repo: string + name: string +} + +export type ClusterIndexSelector = 'category' | 'pinned' + +export type ClusterIndexQuery = ( + selector: ClusterIndexSelector, + values: string[], +) => Promise + +// D1 accepts at most 100 bound parameters in one query. +const D1_BOUND_PARAMETER_LIMIT = 100 + +function chunks(values: T[], size: number): T[][] { + return Array.from( + { length: Math.ceil(values.length / size) }, + (_, index) => values.slice(index * size, (index + 1) * size), + ) +} + +export async function findClusterIndexRows( + query: ClusterIndexQuery, + categories: string[], + pinnedSkills: string[], +): Promise { + const groups: [ClusterIndexSelector, string[]][] = [ + ...chunks(categories, D1_BOUND_PARAMETER_LIMIT).map<[ClusterIndexSelector, string[]]>(values => ['category', values]), + ...chunks(pinnedSkills, D1_BOUND_PARAMETER_LIMIT).map<[ClusterIndexSelector, string[]]>(values => ['pinned', values]), + ] + const rows: Row[] = [] + + for (const [selector, values] of groups) + rows.push(...await query(selector, values)) + + return Array.from( + new Map(rows.map(row => [`${row.owner}/${row.repo}/${row.name}`, row])).values(), + ) +} diff --git a/server/utils/db.ts b/server/utils/db.ts index 3cb39ca3..413a471a 100644 --- a/server/utils/db.ts +++ b/server/utils/db.ts @@ -8,6 +8,7 @@ const RETRYABLE_D1_ERROR_MESSAGES = [ 'cannot resolve d1 db due to transient issue on remote node', 'd1 db is overloaded', 'currently processing a long-running export', + 'd1_reset_do', ] export interface RetryD1Options { diff --git a/test/unit/cluster-index-d1-bounds.test.ts b/test/unit/cluster-index-d1-bounds.test.ts new file mode 100644 index 00000000..e8924751 --- /dev/null +++ b/test/unit/cluster-index-d1-bounds.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it, vi } from 'vitest' +import { findClusterIndexRows } from '../../layers/registry/server/utils/cluster-index-rows' + +interface TestRow { + owner: string + repo: string + name: string +} + +describe('findClusterIndexRows', () => { + it('finds categories and pinned skills without exceeding the D1 parameter limit', async () => { + const shared = { owner: 'shared', repo: 'skills', name: 'shared' } + const query = vi.fn(async (selector: 'category' | 'pinned', values: string[]): Promise => { + if (values.length > 100) + throw new Error('D1_ERROR: too many SQL variables') + + return selector === 'category' + ? [shared, { owner: 'category', repo: 'skills', name: 'category' }] + : [shared, ...values.map(name => ({ owner: 'pinned', repo: 'skills', name }))] + }) + const pinned = Array.from({ length: 101 }, (_, index) => `skill-${index}`) + + const rows = await findClusterIndexRows(query, ['framework'], pinned) + + expect(rows).toHaveLength(103) + expect(query.mock.calls.map(([, values]) => values.length)).toEqual([1, 100, 1]) + }) +}) diff --git a/test/unit/d1-retry.test.ts b/test/unit/d1-retry.test.ts index 0afc5c41..b5967520 100644 --- a/test/unit/d1-retry.test.ts +++ b/test/unit/d1-retry.test.ts @@ -44,6 +44,7 @@ describe('d1 read retries', () => { it.each([ 'D1_ERROR: D1 DB is overloaded. Requests queued for too long.', 'D1_ERROR: Currently processing a long-running export.', + 'D1_ERROR: {"D1_RESET_DO":true}', ])('retries the transient platform failure: %s', async (message) => { const operation = vi.fn() .mockRejectedValueOnce(new Error(message))