diff --git a/checks/_helpers/collectors.mjs b/checks/_helpers/collectors.mjs index 38a8daa9..566b0919 100644 --- a/checks/_helpers/collectors.mjs +++ b/checks/_helpers/collectors.mjs @@ -107,6 +107,8 @@ export function collectD1(context) { const root = context.rootDir const now = context.now const sinceSec = Math.floor(context.since.getTime() / 1000) + const nowSec = Math.floor(now.getTime() / 1000) + const dayAgoSec = nowSec - 24 * 60 * 60 const utcDay = now.toISOString().slice(0, 10) async function d1Query(sql) { const output = await commandJson(context, join(root, 'node_modules/.bin/wrangler'), ['d1', 'execute', 'DB', '--remote', '--json', '--config', 'wrangler.jsonc', '--command', sql]) @@ -226,11 +228,11 @@ export function collectD1(context) { // Hot posts are refresh's remaining exposure: each can cost at most one // more read, when its window crosses midnight UTC. has('x_posts') ? `(SELECT COUNT(*) FROM x_posts WHERE refresh_tier = 'hot') AS x_hot_posts` : 'NULL AS x_hot_posts', - has('x_posts') ? `(SELECT COUNT(*) FROM x_posts WHERE first_seen_at >= ${sinceSec}) AS x_posts_24h` : 'NULL AS x_posts_24h', + has('x_posts') ? `(SELECT COUNT(*) FROM x_posts WHERE first_seen_at BETWEEN ${dayAgoSec} AND ${nowSec}) AS x_posts_24h` : 'NULL AS x_posts_24h', // How stale the freshest ingested post is. Climbs when discovery lags. has('x_posts') ? `(SELECT CAST((${Math.floor(now.getTime() / 1000)} - MAX(posted_at)) / 3600 AS INTEGER) FROM x_posts) AS x_newest_post_age_hours` : 'NULL AS x_newest_post_age_hours', has('x_post_skills') ? `(SELECT COUNT(*) FROM x_post_skills) AS x_verified_skills` : 'NULL AS x_verified_skills', - has('x_post_skills') ? `(SELECT COUNT(*) FROM x_post_skills WHERE verified_at >= ${sinceSec}) AS x_verified_skills_24h` : 'NULL AS x_verified_skills_24h', + has('x_post_skills') ? `(SELECT COUNT(*) FROM x_post_skills WHERE verified_at BETWEEN ${dayAgoSec} AND ${nowSec}) AS x_verified_skills_24h` : 'NULL AS x_verified_skills_24h', (await hasColumn('x_posts', 'skills_scanned_at')) ? `(SELECT COUNT(*) FROM x_posts WHERE skills_scanned_at IS NULL) AS x_posts_unscanned` : 'NULL AS x_posts_unscanned', (await hasColumn('discovery_ledger', 'held_reason')) ? `(SELECT COUNT(*) FROM discovery_ledger WHERE status = 'pending' AND held_reason IS NULL) AS x_ledger_pending` : 'NULL AS x_ledger_pending', (await hasColumn('discovery_ledger', 'held_reason')) ? `(SELECT COUNT(*) FROM discovery_ledger WHERE held_reason IS NOT NULL) AS x_ledger_held` : 'NULL AS x_ledger_held', diff --git a/server/jobs/registry/repo-maintenance.ts b/server/jobs/registry/repo-maintenance.ts index 92fee814..632e0dd2 100644 --- a/server/jobs/registry/repo-maintenance.ts +++ b/server/jobs/registry/repo-maintenance.ts @@ -114,7 +114,12 @@ async function claimRepoProgress( next_offset = 0, total_skills = NULL, updated_at = excluded.updated_at - WHERE repo_sync_progress.updated_at <= ?`, + WHERE repo_sync_progress.updated_at <= ? + OR NOT EXISTS ( + SELECT 1 FROM jobs + WHERE id = repo_sync_progress.job_id + AND completed_at IS NULL AND failed_at IS NULL + )`, ).bind( input.owner, input.repo, diff --git a/test/unit/external-checkin-collectors.test.ts b/test/unit/external-checkin-collectors.test.ts index 0d68aebb..78fe0501 100644 --- a/test/unit/external-checkin-collectors.test.ts +++ b/test/unit/external-checkin-collectors.test.ts @@ -109,6 +109,43 @@ it('counts only X Posts observed within the UTC day at the check time', async () expect(report.results[0]?.result).toMatchObject({ evidence: { cost: { x_observed_posts_today: 1 } } }) }) +it.each(['2026-10-05T09:00:00Z', '2026-10-03T10:00:00Z'])('keeps rolling X day totals independent of baseline %s', async (since) => { + const now = new Date('2026-10-05T10:00:00Z') + const root = await mkdtemp(join(tmpdir(), 'skilld-checkin-x-day-')) + roots.push(root) + await mkdir(join(root, 'shared/server'), { recursive: true }) + await mkdir(join(root, 'migrations')) + await writeFile(join(root, 'shared/server/x-ingest.ts'), 'export const DAILY_DISCOVERY_READ_BUDGET = 400\nconst MIN_SEARCH_PAGE_SIZE = 10') + const db = new Database(':memory:') + db.exec(` + CREATE TABLE x_posts (post_id TEXT, platform TEXT, refresh_tier TEXT, first_seen_at INTEGER, posted_at INTEGER); + CREATE TABLE x_post_skills (verified_at INTEGER); + INSERT INTO x_posts VALUES + ('older', 'x', 'frozen', unixepoch('2026-10-04T09:59:59Z'), 0), + ('boundary', 'x', 'frozen', unixepoch('2026-10-04T10:00:00Z'), 0), + ('middle', 'x', 'frozen', unixepoch('2026-10-04T12:00:00Z'), 0), + ('recent', 'x', 'frozen', unixepoch('2026-10-05T09:30:00Z'), 0), + ('future', 'x', 'frozen', unixepoch('2026-10-05T10:00:01Z'), 0); + INSERT INTO x_post_skills VALUES + (unixepoch('2026-10-04T09:59:59Z')), + (unixepoch('2026-10-04T10:00:00Z')), + (unixepoch('2026-10-04T12:00:00Z')), + (unixepoch('2026-10-05T09:30:00Z')), + (unixepoch('2026-10-05T10:00:01Z')); + `) + boundary.command.mockImplementation(async (_context, command: string, args: string[]) => { + if (command === 'git') + return { _tag: 'Ok', stdout: args[0] === 'rev-parse' ? 'abc' : '', stderr: '' } + const sql = args[args.indexOf('--command') + 1]! + return { _tag: 'Ok', stdout: JSON.stringify([{ success: true, results: db.prepare(sql).all() }]), stderr: '' } + }) + + const { report } = await runExternalChecks([databaseCheck], { required: [databaseCheck.id] }, { rootDir: root, env: {}, clock: () => now, since: new Date(since) }) + + db.close() + expect(report.results[0]?.result).toMatchObject({ evidence: { cost: { x_posts_24h: 3, x_verified_skills_24h: 3 } } }) +}) + it('rejects oversized Worker analytics evidence', async () => { vi.stubGlobal('fetch', vi.fn().mockResolvedValue(Response.json({ data: { viewer: { accounts: [{ workersInvocationsAdaptive: [] }] } }, padding: 'x'.repeat(2_097_153) }))) const { report } = await runExternalChecks([workersCheck], { required: [workersCheck.id] }, { env: { CLOUDFLARE_USAGE_TOKEN: 'test-token' } }) diff --git a/test/unit/repo-maintenance-continuation.test.ts b/test/unit/repo-maintenance-continuation.test.ts index 289e707b..ca696797 100644 --- a/test/unit/repo-maintenance-continuation.test.ts +++ b/test/unit/repo-maintenance-continuation.test.ts @@ -123,6 +123,7 @@ describe('repo sync continuation', () => { next_offset INTEGER, total_skills INTEGER, updated_at INTEGER, PRIMARY KEY (owner, repo) ); + CREATE TABLE jobs (id TEXT PRIMARY KEY, completed_at INTEGER, failed_at INTEGER); `) db = wrapSqlite(sqlite) }) @@ -135,6 +136,35 @@ describe('repo sync continuation', () => { claimDiscovery: false, } as const + it.each(['missing', 'failed', 'completed'])('reclaims progress immediately when its previous job is %s', async (state) => { + const now = Math.floor(Date.now() / 1000) + sqlite.prepare('INSERT INTO repo_sync_progress VALUES (?, ?, ?, ?, ?, ?, ?, ?)') + .run('acme', 'skills', 'old-job', 'old-tree', now, 250, 500, now) + if (state !== 'missing') { + sqlite.prepare('INSERT INTO jobs VALUES (?, ?, ?)') + .run('old-job', state === 'completed' ? now : null, state === 'failed' ? now : null) + } + syncRepo.mockResolvedValue(failedWith('tree_truncated')) + const { ctx, control } = jobContext(db) + + await handleRegistryRepoJob(payload, ctx as never) + + expect(control).toMatchObject({ action: 'failed', error: 'tree_truncated' }) + }) + + it('waits while another active job owns fresh progress', async () => { + const now = Math.floor(Date.now() / 1000) + sqlite.prepare('INSERT INTO repo_sync_progress VALUES (?, ?, ?, ?, ?, ?, ?, ?)') + .run('acme', 'skills', 'active-job', 'tree-1', now, 250, 500, now) + sqlite.prepare('INSERT INTO jobs VALUES (?, NULL, NULL)').run('active-job') + const { ctx, control } = jobContext(db) + + await handleRegistryRepoJob(payload, ctx as never) + + expect(control).toMatchObject({ action: 'released', delaySeconds: 60 }) + expect(syncRepo).not.toHaveBeenCalled() + }) + it('holds a new submission before syncing and releases its progress claim', async () => { purposeAdmission.mockResolvedValue({ _tag: 'held', reason: 'repository_purpose_review_required' }) const { ctx, control } = jobContext(db) @@ -191,6 +221,7 @@ describe('permanent repository failures on the sync path', () => { next_offset INTEGER, total_skills INTEGER, updated_at INTEGER, PRIMARY KEY (owner, repo) ); + CREATE TABLE jobs (id TEXT PRIMARY KEY, completed_at INTEGER, failed_at INTEGER); `) db = wrapSqlite(sqlite) }) diff --git a/test/unit/skill-detail-d1-cost.test.ts b/test/unit/skill-detail-d1-cost.test.ts index b2c1da5a..e6d1d676 100644 --- a/test/unit/skill-detail-d1-cost.test.ts +++ b/test/unit/skill-detail-d1-cost.test.ts @@ -33,14 +33,18 @@ vi.stubGlobal('fetch', rawFetch) let harness: SqliteD1 let background: Promise[] +let handler: (event: H3Event) => Promise> -beforeEach(() => { +beforeEach(async () => { vi.resetModules() upstreamFetch.mockClear() rawFetch.mockReset() rawFetch.mockImplementation(async () => new Response(UPSTREAM_RAW)) background = [] slug = `${OWNER}/${REPO}/alpha` + // Compile the fresh handler during setup. The request deadline should cover + // handler behaviour, rather than Vite's first import of its dependency tree. + handler = (await import('../../layers/registry/server/api/skills/[...slug].get')).default as typeof handler }) function seed(options: { maximumQueries?: number, renderedAt?: number | null, renderedStatus?: string | null, sourceResolved?: number } = {}) { @@ -79,7 +83,6 @@ function seed(options: { maximumQueries?: number, renderedAt?: number | null, re } async function render(): Promise> { - const handler = (await import('../../layers/registry/server/api/skills/[...slug].get')).default as (event: H3Event) => Promise> const body = await handler({ context: { platform: { db: harness.db },