From 1b94f1c591dc02f02663483fbbb82eefc4a63e73 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 4 Aug 2026 14:00:58 +0000 Subject: [PATCH 1/3] Remove legacy job observability storage Co-authored-by: Kent C. Dodds --- docs/contributing/architecture/run-records.md | 55 ++-- .../0142-drop-jobs-observability-columns.sql | 8 + .../src/app/admin-insights-data.node.test.ts | 30 ++- .../worker/src/app/admin-insights-data.ts | 24 +- ...job-run-observability-hydrate.node.test.ts | 38 ++- .../src/jobs/job-run-observability-hydrate.ts | 15 +- packages/worker/src/jobs/process-due-jobs.ts | 7 - packages/worker/src/jobs/repo.ts | 49 +--- packages/worker/src/jobs/repo.workers.test.ts | 204 +++------------ .../run-due-jobs-claim-fence.node.test.ts | 5 - packages/worker/src/jobs/service.node.test.ts | 35 +-- .../src/mcp/run-kody-registry.node.test.ts | 27 +- .../run-records/admin-insights-snapshot.ts | 10 +- .../dedicated-state.workers.test.ts | 240 ++++++++++-------- packages/worker/src/run-records/run-log-do.ts | 69 ++++- .../src/run-records/service.node.test.ts | 3 + packages/worker/src/run-records/service.ts | 5 +- tools/migration-ledger.json | 4 + 18 files changed, 371 insertions(+), 457 deletions(-) create mode 100644 packages/worker/migrations/0142-drop-jobs-observability-columns.sql diff --git a/docs/contributing/architecture/run-records.md b/docs/contributing/architecture/run-records.md index 0351ed15a7..d887b8aa02 100644 --- a/docs/contributing/architecture/run-records.md +++ b/docs/contributing/architecture/run-records.md @@ -367,7 +367,7 @@ read `RunLog`. Run records are **excluded from the `storage_bytes` entitlement**. They are observability history, not user content. -## RunLog authority and deploy sequencing +## RunLog authority The per-user `RunLog` DO is the **sole runtime authority** for: @@ -378,34 +378,16 @@ The per-user `RunLog` DO is the **sole runtime authority** for: `activation_milestones`). Runtime paths read and write those tables only inside `RunLog`. Workflow -lifecycle code does not touch D1 `workflow_runs`. Job finalization and -activation increments do not lazy-seed dedicated RunLog rows from D1 on read or -finish. The completed admin seed/parity MCP capability is absent from the admin -domain. - -**Application deploy (code-only, PR #1205):** ships RunLog-only workflow, -job-observability, and activation behavior; bounded content-free RunLog point -reads for admin workflow/activation insights; and schema-v8 -`job_run_observability.legacy_seeded` kept inert for rollback compatibility. - -**Production parity gate (2026-08-04 02:45–02:46 UTC):** 41/41 non-deleting -users matched across nine keyset pages with zero failed, missing, or undercount -rows before the application deploy. - -**Post-cutover production smoke:** `workflow_run_list` succeeded after the -application deploy. - -**Destructive follow-up deploy (separate PR, migration `0137`):** drops the -retired D1 projection tables `workflow_runs`, `user_package_run_successes`, and -`user_activation_milestones`; removes them from account deletion/export -inventory and retires the hourly D1 `workflow_runs` retention lane. -Pre-migration D1 Time Travel bookmark for database -`8c1014d1-6b41-4695-a0a2-159071f0f919`: +lifecycle code does not use D1 workflow projections. Job finalization and +activation increments do not seed dedicated RunLog rows from D1. D1 `jobs` +retains schedule metadata and retention anchors, but job outcome aggregates come +from `job_run_observability`. + +The pre-drop D1 Time Travel bookmark for database +`8c1014d1-6b41-4695-a0a2-159071f0f919` is: `0000116d-000000d2-000050bd-c7ecd5892a189df7cda145af746bc9c9`. Restore from that bookmark (or a fresher snapshot taken immediately before apply) if the destructive migration must be rolled back — do not reintroduce dual-write paths. -Application cutover evidence: -https://github.com/kentcdodds/kody/actions/runs/30876391371 ## Dedicated RunLog state (outside run-history caps) @@ -429,17 +411,16 @@ cursors remain valid. ### Admin insights RunLog reads -The role-gated `/admin/insights` dashboard loads workflow status totals and -activation funnel/latency from **bounded, content-free per-user RunLog point -reads** (`getAdminInsightsSnapshot`; concurrency capped by -`adminInsightsRunLogConcurrency`). Each snapshot returns aggregate workflow -status counts and activation milestone timestamps/ids only — never workflow -names, errors, logs, or other user-authored content. The page exposes -`runLogCompleteness` (`usersAttempted`, `usersLoaded`, `complete`) so partial -fanout degrades run-derived charts with an explicit warning instead of failing -the whole dashboard. Legacy D1 `workflow_runs`, `user_activation_milestones`, -and `user_package_run_successes` were retired by migration `0137` and are not -queried on that path. +The role-gated `/admin/insights` dashboard loads workflow status totals and job +success/error totals, plus activation funnel/latency, from **bounded, +content-free per-user RunLog point reads** (`getAdminInsightsSnapshot`; +concurrency capped by `adminInsightsRunLogConcurrency`). Each snapshot returns +aggregate workflow statuses, job outcome counts, and activation milestone +timestamps/ids only — never workflow or job names, errors, logs, or other +user-authored content. The page exposes `runLogCompleteness` (`usersAttempted`, +`usersLoaded`, `complete`) so partial fanout degrades run-derived charts with an +explicit warning instead of failing the whole dashboard. D1 supplies only job +schedule totals (`totalJobs` and `enabledJobs`) on that path. ## Related diff --git a/packages/worker/migrations/0142-drop-jobs-observability-columns.sql b/packages/worker/migrations/0142-drop-jobs-observability-columns.sql new file mode 100644 index 0000000000..44ea1cf615 --- /dev/null +++ b/packages/worker/migrations/0142-drop-jobs-observability-columns.sql @@ -0,0 +1,8 @@ +-- RunLog is the sole authority for job error, duration, and outcome counters. +-- Keep the D1 scheduling/retention anchors last_run_at and last_run_status. + +ALTER TABLE jobs DROP COLUMN last_run_error; +ALTER TABLE jobs DROP COLUMN last_duration_ms; +ALTER TABLE jobs DROP COLUMN run_count; +ALTER TABLE jobs DROP COLUMN success_count; +ALTER TABLE jobs DROP COLUMN error_count; diff --git a/packages/worker/src/app/admin-insights-data.node.test.ts b/packages/worker/src/app/admin-insights-data.node.test.ts index 79000f2114..f18071921c 100644 --- a/packages/worker/src/app/admin-insights-data.node.test.ts +++ b/packages/worker/src/app/admin-insights-data.node.test.ts @@ -181,7 +181,11 @@ const defaultInsightsUsers = [ ] function emptySnapshot(): RunLogAdminInsightsSnapshot { - return { workflowStatusCounts: [], activationMilestones: [] } + return { + workflowStatusCounts: [], + activationMilestones: [], + jobRunCounts: { success: 0, error: 0 }, + } } function createInsightsTestDb( @@ -203,8 +207,6 @@ function createInsightsTestDb( return { total: 4, enabled: 3, - success_runs: 20, - error_runs: 2, } as T } if (normalizedQuery.includes('from users where created_at < ?')) { @@ -370,6 +372,7 @@ test('loadAdminInsightsData assembles the dashboard payload from D1 plus RunLog { status: 'complete', count: 5 }, { status: 'errored', count: 1 }, ], + jobRunCounts: { success: 12, error: 2 }, activationMilestones: [ { milestone: 'package_run_succeeded', @@ -385,6 +388,7 @@ test('loadAdminInsightsData assembles the dashboard payload from D1 plus RunLog }, 'user-b': { workflowStatusCounts: [{ status: 'complete', count: 3 }], + jobRunCounts: { success: 8, error: 0 }, activationMilestones: [ { milestone: 'package_run_succeeded', @@ -465,6 +469,12 @@ test('loadAdminInsightsData assembles the dashboard payload from D1 plus RunLog successRuns: 20, errorRuns: 2, }) + const jobStatsQuery = db.seenQueries.find((query) => + normalizeQuery(query).includes('from jobs'), + ) + expect(normalizeQuery(jobStatsQuery ?? '')).toBe( + 'select count(*) as total, sum(enabled) as enabled from jobs', + ) expect(data.activation.steps).toEqual([ { step: 'signed_up', users: 8 }, { step: 'email_verified', users: 5 }, @@ -503,6 +513,7 @@ test('multi-user RunLog aggregation sums workflow statuses and milestone users', { status: 'running', count: 2 }, { status: 'complete', count: 1 }, ], + jobRunCounts: { success: 3, error: 1 }, activationMilestones: [ { milestone: 'package_run_succeeded', @@ -524,6 +535,7 @@ test('multi-user RunLog aggregation sums workflow statuses and milestone users', }, snapshot: { workflowStatusCounts: [{ status: 'complete', count: 4 }], + jobRunCounts: { success: 5, error: 2 }, activationMilestones: [ { milestone: 'package_run_succeeded', @@ -547,6 +559,8 @@ test('multi-user RunLog aggregation sums workflow statuses and milestone users', { status: 'complete', count: 5 }, { status: 'running', count: 2 }, ]) + expect(folded.jobSuccessRuns).toBe(8) + expect(folded.jobErrorRuns).toBe(3) expect(folded.packageRunSucceededUsers).toBe(2) expect(folded.packageActivatedUsers).toBe(1) expect(folded.medianHoursToActivation).toBe(12) @@ -596,6 +610,7 @@ test('activation latency uses package_activated reachedAt against D1 email_verif stubSnapshotsByUser({ 'user-a': { workflowStatusCounts: [], + jobRunCounts: { success: 0, error: 0 }, activationMilestones: [ { milestone: 'package_activated', @@ -606,6 +621,7 @@ test('activation latency uses package_activated reachedAt against D1 email_verif }, 'user-b': { workflowStatusCounts: [], + jobRunCounts: { success: 0, error: 0 }, activationMilestones: [ { milestone: 'package_activated', @@ -617,6 +633,7 @@ test('activation latency uses package_activated reachedAt against D1 email_verif }, 'user-c': { workflowStatusCounts: [], + jobRunCounts: { success: 0, error: 0 }, activationMilestones: [ { milestone: 'package_activated', @@ -694,6 +711,12 @@ test('degraded RunLog snapshots warn and zero run-derived charts without failing expect(data.ok).toBe(true) expect(data.totals.workflowRuns).toBe(0) expect(data.workflowStatuses).toEqual([]) + expect(data.jobHealth).toEqual({ + totalJobs: 4, + enabledJobs: 3, + successRuns: 0, + errorRuns: 0, + }) expect(data.activation.steps).toEqual([ { step: 'signed_up', users: 8 }, { step: 'email_verified', users: 5 }, @@ -733,6 +756,7 @@ test('admin insights uses content-free getAdminInsightsSnapshot point reads only consoleWarn.mockImplementation(() => {}) const snapshot: RunLogAdminInsightsSnapshot = { workflowStatusCounts: [{ status: 'complete', count: 1 }], + jobRunCounts: { success: 2, error: 1 }, activationMilestones: [ { milestone: 'package_run_succeeded', diff --git a/packages/worker/src/app/admin-insights-data.ts b/packages/worker/src/app/admin-insights-data.ts index c26544c27f..a932e81f2a 100644 --- a/packages/worker/src/app/admin-insights-data.ts +++ b/packages/worker/src/app/admin-insights-data.ts @@ -67,8 +67,6 @@ type HeatmapRow = { day: string; hour: string; n: number } type JobStatsRow = { total: number enabled: number | null - success_runs: number | null - error_runs: number | null } type ForkActorRow = { actor: string | null; n: number } type InsightsUserRow = { @@ -79,6 +77,8 @@ type InsightsUserRow = { type AggregatedRunLogInsights = AdminInsightsRunLogCompleteness & { workflowStatuses: Array workflowRuns: number + jobSuccessRuns: number + jobErrorRuns: number packageRunSucceededUsers: number packageActivatedUsers: number medianHoursToActivation: number | null @@ -101,7 +101,7 @@ export async function loadAdminInsightsData( : null if (!cache) return await queryAdminInsights(env, now) return await cachified({ - key: 'admin-insights:v5', + key: 'admin-insights:v6', cache, ttl: insightsCacheTtlMs, getFreshValue: () => queryAdminInsights(env, now), @@ -146,11 +146,7 @@ async function queryAdminInsights( ] = await Promise.all([ queryTotals(db), db - .prepare( - `SELECT COUNT(*) AS total, SUM(enabled) AS enabled, - SUM(success_count) AS success_runs, SUM(error_count) AS error_runs - FROM jobs`, - ) + .prepare(`SELECT COUNT(*) AS total, SUM(enabled) AS enabled FROM jobs`) .first(), db .prepare( @@ -226,8 +222,8 @@ async function queryAdminInsights( const jobHealth: AdminInsightsJobHealth = { totalJobs: Number(jobStats?.total ?? 0), enabledJobs: Number(jobStats?.enabled ?? 0), - successRuns: Number(jobStats?.success_runs ?? 0), - errorRuns: Number(jobStats?.error_runs ?? 0), + successRuns: runLogInsights.jobSuccessRuns, + errorRuns: runLogInsights.jobErrorRuns, } return { @@ -494,6 +490,8 @@ async function aggregateRunLogInsights(input: { ...completeRunLogInsights, workflowStatuses: [], workflowRuns: 0, + jobSuccessRuns: 0, + jobErrorRuns: 0, packageRunSucceededUsers: 0, packageActivatedUsers: 0, medianHoursToActivation: null, @@ -535,6 +533,8 @@ export function foldRunLogSnapshots( let usersLoaded = 0 const statusCounts = new Map() let workflowRuns = 0 + let jobSuccessRuns = 0 + let jobErrorRuns = 0 let packageRunSucceededUsers = 0 let packageActivatedUsers = 0 const activationHours: Array = [] @@ -549,6 +549,8 @@ export function foldRunLogSnapshots( statusCounts.set(status, (statusCounts.get(status) ?? 0) + count) workflowRuns += count } + jobSuccessRuns += Math.max(0, Number(snapshot.jobRunCounts.success) || 0) + jobErrorRuns += Math.max(0, Number(snapshot.jobRunCounts.error) || 0) let hasRunSucceeded = false let activatedReachedAt: string | null = null for (const milestone of snapshot.activationMilestones) { @@ -582,6 +584,8 @@ export function foldRunLogSnapshots( right.count - left.count || left.status.localeCompare(right.status), ), workflowRuns, + jobSuccessRuns, + jobErrorRuns, packageRunSucceededUsers, packageActivatedUsers, medianHoursToActivation: medianOf(activationHours), diff --git a/packages/worker/src/jobs/job-run-observability-hydrate.node.test.ts b/packages/worker/src/jobs/job-run-observability-hydrate.node.test.ts index 53aa42cbd9..790cf88e5b 100644 --- a/packages/worker/src/jobs/job-run-observability-hydrate.node.test.ts +++ b/packages/worker/src/jobs/job-run-observability-hydrate.node.test.ts @@ -67,7 +67,30 @@ test('job view hydrate overlays RunLog outcomes for point and batch reads', asyn successCount: 3, errorCount: 1, }) - expect(applyJobRunObservabilityToJobView(job, null)).toBe(job) + expect( + applyJobRunObservabilityToJobView( + createJobView({ + lastRunAt: '2026-04-20T09:00:00.000Z', + lastRunStatus: 'success', + lastRunError: 'stale D1 error', + lastDurationMs: 999, + runCount: 9, + successCount: 8, + errorCount: 1, + }), + null, + ), + ).toEqual( + expect.objectContaining({ + lastRunAt: '2026-04-20T09:00:00.000Z', + lastRunStatus: 'success', + lastRunError: undefined, + lastDurationMs: undefined, + runCount: 0, + successCount: 0, + errorCount: 0, + }), + ) const env = {} as Env getJobRunObservability.mockResolvedValue({ @@ -116,7 +139,14 @@ test('job view hydrate overlays RunLog outcomes for point and batch reads', asyn userId: 'user-1', jobs: [ createJobView({ id: 'job-a' }), - createJobView({ id: 'job-b', runCount: 0 }), + createJobView({ + id: 'job-b', + lastRunError: 'stale D1 error', + lastDurationMs: 999, + runCount: 9, + successCount: 8, + errorCount: 1, + }), ], }) expect(getJobRunObservabilityBatch).toHaveBeenCalledWith({ @@ -133,5 +163,9 @@ test('job view hydrate overlays RunLog outcomes for point and batch reads', asyn expect(batch[1]).toMatchObject({ id: 'job-b', runCount: 0, + successCount: 0, + errorCount: 0, + lastRunError: undefined, + lastDurationMs: undefined, }) }) diff --git a/packages/worker/src/jobs/job-run-observability-hydrate.ts b/packages/worker/src/jobs/job-run-observability-hydrate.ts index 72e2a7fdf0..b18848c990 100644 --- a/packages/worker/src/jobs/job-run-observability-hydrate.ts +++ b/packages/worker/src/jobs/job-run-observability-hydrate.ts @@ -14,16 +14,15 @@ export function applyJobRunObservabilityToJobView( job: JobView, observability: JobRunObservabilityRecord | null | undefined, ): JobView { - if (!observability) return job return { ...job, - lastRunAt: observability.lastRunAt ?? job.lastRunAt, - lastRunStatus: observability.lastRunStatus ?? job.lastRunStatus, - lastRunError: observability.lastRunError ?? undefined, - lastDurationMs: observability.lastDurationMs ?? undefined, - runCount: observability.runCount, - successCount: observability.successCount, - errorCount: observability.errorCount, + lastRunAt: observability?.lastRunAt ?? job.lastRunAt, + lastRunStatus: observability?.lastRunStatus ?? job.lastRunStatus, + lastRunError: observability?.lastRunError ?? undefined, + lastDurationMs: observability?.lastDurationMs ?? undefined, + runCount: observability?.runCount ?? 0, + successCount: observability?.successCount ?? 0, + errorCount: observability?.errorCount ?? 0, } } diff --git a/packages/worker/src/jobs/process-due-jobs.ts b/packages/worker/src/jobs/process-due-jobs.ts index 09c8c9cb93..2f881533cd 100644 --- a/packages/worker/src/jobs/process-due-jobs.ts +++ b/packages/worker/src/jobs/process-due-jobs.ts @@ -144,15 +144,8 @@ export function applyExecutionOutcome( // Retention ages from execution time and classifies once jobs by status. lastRunAt: outcome.finishedAt, lastRunStatus: status, - // RunLog owns last_run_error, last_duration_ms, and counters. Leave the - // D1 copies untouched so finalization does not copy them back. - lastRunError: job.lastRunError, - lastDurationMs: job.lastDurationMs, nextRunAt: overrides.nextRunAt ?? job.nextRunAt, enabled: overrides.enabled ?? job.enabled, killSwitchEnabled: overrides.killSwitchEnabled ?? job.killSwitchEnabled, - runCount: job.runCount, - successCount: job.successCount, - errorCount: job.errorCount, } } diff --git a/packages/worker/src/jobs/repo.ts b/packages/worker/src/jobs/repo.ts index 8797ad9168..b4f06508ca 100644 --- a/packages/worker/src/jobs/repo.ts +++ b/packages/worker/src/jobs/repo.ts @@ -22,12 +22,7 @@ type JobRowRecord = { updated_at: string last_run_at: string | null last_run_status: JobRecord['lastRunStatus'] | null - last_run_error: string | null - last_duration_ms: number | null next_run_at: string - run_count: number - success_count: number - error_count: number claim_token: string | null running_since: string | null lease_expires_at: string | null @@ -65,12 +60,7 @@ function serializeJob(job: JobRecord) { updated_at: job.updatedAt, last_run_at: job.lastRunAt ?? null, last_run_status: job.lastRunStatus ?? null, - last_run_error: job.lastRunError ?? null, - last_duration_ms: job.lastDurationMs ?? null, next_run_at: job.nextRunAt, - run_count: job.runCount, - success_count: job.successCount, - error_count: job.errorCount, } } @@ -120,16 +110,12 @@ function mapRow(row: Record): JobRow { row['last_run_status'] == null ? undefined : (String(row['last_run_status']) as JobRecord['lastRunStatus']), - lastRunError: - row['last_run_error'] == null ? undefined : String(row['last_run_error']), - lastDurationMs: - row['last_duration_ms'] == null - ? undefined - : Number(row['last_duration_ms']), + lastRunError: undefined, + lastDurationMs: undefined, nextRunAt: String(row['next_run_at']), - runCount: Number(row['run_count']) || 0, - successCount: Number(row['success_count']) || 0, - errorCount: Number(row['error_count']) || 0, + runCount: 0, + successCount: 0, + errorCount: 0, } return { id: record.id, @@ -154,12 +140,7 @@ function mapRow(row: Record): JobRow { updated_at: record.updatedAt, last_run_at: record.lastRunAt ?? null, last_run_status: record.lastRunStatus ?? null, - last_run_error: record.lastRunError ?? null, - last_duration_ms: record.lastDurationMs ?? null, next_run_at: record.nextRunAt, - run_count: record.runCount, - success_count: record.successCount, - error_count: record.errorCount, claim_token: row['claim_token'] == null ? null : String(row['claim_token']), running_since: row['running_since'] == null ? null : String(row['running_since']), @@ -205,9 +186,8 @@ export async function insertJobRow(input: { `INSERT INTO jobs ( id, user_id, name, source_id, published_commit, repo_check_policy_json, storage_id, params_json, schedule_json, timezone, enabled, kill_switch_enabled, preserved, expires_at, caller_context_json, created_at, updated_at, - last_run_at, last_run_status, last_run_error, last_duration_ms, - next_run_at, run_count, success_count, error_count - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + last_run_at, last_run_status, next_run_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .bind( serialized.id, @@ -229,12 +209,7 @@ export async function insertJobRow(input: { serialized.updated_at, serialized.last_run_at, serialized.last_run_status, - serialized.last_run_error, - serialized.last_duration_ms, serialized.next_run_at, - serialized.run_count, - serialized.success_count, - serialized.error_count, ) .run() } @@ -252,8 +227,6 @@ export async function updateJobRow(input: { name = ?, source_id = ?, published_commit = ?, repo_check_policy_json = ?, storage_id = ?, params_json = ?, schedule_json = ?, timezone = ?, enabled = ?, kill_switch_enabled = ?, preserved = ?, expires_at = ?, caller_context_json = ?, updated_at = ?, last_run_at = ?, last_run_status = ?, - last_run_error = CASE WHEN last_run_at IS ? THEN last_run_error ELSE NULL END, - last_duration_ms = CASE WHEN last_run_at IS ? THEN last_duration_ms ELSE NULL END, next_run_at = ?, claim_token = NULL, running_since = NULL, lease_expires_at = NULL, claimed_scheduled_for = NULL, retry_scheduled_for = NULL, @@ -277,9 +250,6 @@ export async function updateJobRow(input: { serialized.updated_at, serialized.last_run_at, serialized.last_run_status, - // NULL-safe: clear legacy fallback only when last_run_at advances. - serialized.last_run_at, - serialized.last_run_at, serialized.next_run_at, serialized.id, input.userId, @@ -584,8 +554,6 @@ export async function finalizeClaimedJobRow(input: { name = ?, source_id = ?, published_commit = ?, repo_check_policy_json = ?, storage_id = ?, params_json = ?, schedule_json = ?, timezone = ?, enabled = ?, kill_switch_enabled = ?, preserved = ?, expires_at = ?, caller_context_json = ?, updated_at = ?, last_run_at = ?, last_run_status = ?, - last_run_error = CASE WHEN last_run_at IS ? THEN last_run_error ELSE NULL END, - last_duration_ms = CASE WHEN last_run_at IS ? THEN last_duration_ms ELSE NULL END, next_run_at = ?, claim_token = NULL, running_since = NULL, lease_expires_at = NULL, claimed_scheduled_for = NULL, retry_scheduled_for = NULL, @@ -609,9 +577,6 @@ export async function finalizeClaimedJobRow(input: { serialized.updated_at, serialized.last_run_at, serialized.last_run_status, - // NULL-safe: clear legacy fallback only when last_run_at advances. - serialized.last_run_at, - serialized.last_run_at, serialized.next_run_at, input.scheduledFor, serialized.id, diff --git a/packages/worker/src/jobs/repo.workers.test.ts b/packages/worker/src/jobs/repo.workers.test.ts index 4bd8a7f30d..053f4f11d7 100644 --- a/packages/worker/src/jobs/repo.workers.test.ts +++ b/packages/worker/src/jobs/repo.workers.test.ts @@ -2,6 +2,7 @@ import { env } from 'cloudflare:workers' import { expect, test } from 'vitest' import { claimJobRow, + disableExpiredJobRowsForUser, finalizeClaimedJobRow, getJobRowById, getNextRunnableJobRow, @@ -34,12 +35,7 @@ async function ensureJobsSchema() { updated_at TEXT NOT NULL, last_run_at TEXT, last_run_status TEXT, - last_run_error TEXT, - last_duration_ms INTEGER, next_run_at TEXT NOT NULL, - run_count INTEGER NOT NULL DEFAULT 0, - success_count INTEGER NOT NULL DEFAULT 0, - error_count INTEGER NOT NULL DEFAULT 0, claim_token TEXT, running_since TEXT, lease_expires_at TEXT, @@ -255,107 +251,15 @@ test('conditional job claims exclude overlap and reclaim only after lease expiry expect(retriedOccurrence?.retry_count).toBe(1) }) -async function seedLegacyRunObservabilityFallback(input: { - userId: string - jobId: string - lastRunAt: string - lastRunStatus: 'success' | 'error' - lastRunError: string - lastDurationMs: number - runCount: number - successCount: number - errorCount: number -}) { - await env.APP_DB.prepare( - `UPDATE jobs SET - last_run_at = ?, last_run_status = ?, last_run_error = ?, last_duration_ms = ?, - run_count = ?, success_count = ?, error_count = ? - WHERE id = ? AND user_id = ?`, - ) - .bind( - input.lastRunAt, - input.lastRunStatus, - input.lastRunError, - input.lastDurationMs, - input.runCount, - input.successCount, - input.errorCount, - input.jobId, - input.userId, - ) - .run() -} - -test('metadata updateJobRow preserves legacy error/duration when last_run_at is unchanged', async () => { - await ensureJobsSchema() - const userId = 'user-metadata-preserve' - const lastRunAt = '2026-04-20T11:00:00.000Z' - await insertJob({ - id: 'metadata-preserve', - userId, - nextRunAt: '2026-04-20T12:00:00.000Z', - }) - await seedLegacyRunObservabilityFallback({ - userId, - jobId: 'metadata-preserve', - lastRunAt, - lastRunStatus: 'error', - lastRunError: 'legacy boom', - lastDurationMs: 321, - runCount: 4, - successCount: 2, - errorCount: 2, - }) - const row = await getJobRowById(env.APP_DB, userId, 'metadata-preserve') - if (!row) throw new Error('Expected job row.') - - expect( - await updateJobRow({ - db: env.APP_DB, - userId, - job: { - ...row.record, - name: 'Renamed without a new run', - updatedAt: '2026-04-20T11:30:00.000Z', - }, - callerContextJson: row.callerContextJson, - }), - ).toBe(true) - - expect( - await getJobRowById(env.APP_DB, userId, 'metadata-preserve'), - ).toMatchObject({ - name: 'Renamed without a new run', - last_run_at: lastRunAt, - last_run_status: 'error', - last_run_error: 'legacy boom', - last_duration_ms: 321, - run_count: 4, - success_count: 2, - error_count: 2, - }) -}) - -test('manual updateJobRow clears stale error/duration when last_run_at advances; counters stay put', async () => { +test('job writes retain D1 run anchors and default RunLog-owned fields', async () => { await ensureJobsSchema() - const userId = 'user-manual-clear' + const userId = 'user-run-anchors' await insertJob({ - id: 'manual-clear', + id: 'run-anchors', userId, nextRunAt: '2026-04-20T12:00:00.000Z', }) - await seedLegacyRunObservabilityFallback({ - userId, - jobId: 'manual-clear', - lastRunAt: '2026-04-20T11:00:00.000Z', - lastRunStatus: 'error', - lastRunError: 'stale legacy error', - lastDurationMs: 888, - runCount: 3, - successCount: 1, - errorCount: 2, - }) - const row = await getJobRowById(env.APP_DB, userId, 'manual-clear') + const row = await getJobRowById(env.APP_DB, userId, 'run-anchors') if (!row) throw new Error('Expected job row.') const finishedAt = '2026-04-20T12:05:00.000Z' @@ -368,8 +272,7 @@ test('manual updateJobRow clears stale error/duration when last_run_at advances; updatedAt: finishedAt, lastRunAt: finishedAt, lastRunStatus: 'success', - // Must not be written authoritatively; also must not survive. - lastRunError: 'poison-should-not-persist', + lastRunError: 'RunLog-only error', lastDurationMs: 999, runCount: 99, successCount: 99, @@ -379,85 +282,61 @@ test('manual updateJobRow clears stale error/duration when last_run_at advances; }), ).toBe(true) - expect(await getJobRowById(env.APP_DB, userId, 'manual-clear')).toMatchObject( - { - last_run_at: finishedAt, - last_run_status: 'success', - last_run_error: null, - last_duration_ms: null, - run_count: 3, - success_count: 1, - error_count: 2, + const updated = await getJobRowById(env.APP_DB, userId, 'run-anchors') + expect(updated).toMatchObject({ + last_run_at: finishedAt, + last_run_status: 'success', + record: { + lastRunAt: finishedAt, + lastRunStatus: 'success', + runCount: 0, + successCount: 0, + errorCount: 0, }, - ) -}) + }) + expect(updated).not.toHaveProperty('last_run_error') + expect(updated).not.toHaveProperty('last_duration_ms') + expect(updated?.record.lastRunError).toBeUndefined() + expect(updated?.record.lastDurationMs).toBeUndefined() -test('finalizeClaimedJobRow clears stale error/duration on new run without writing counters', async () => { - await ensureJobsSchema() - const userId = 'user-finalize-observability' const scheduledFor = '2026-04-20T12:00:00.000Z' - await insertJob({ - id: 'finalize-observability', - userId, - nextRunAt: scheduledFor, - }) - await seedLegacyRunObservabilityFallback({ - userId, - jobId: 'finalize-observability', - lastRunAt: '2026-04-20T11:00:00.000Z', - lastRunStatus: 'error', - lastRunError: 'previous failure', - lastDurationMs: 654, - runCount: 5, - successCount: 3, - errorCount: 2, - }) const claimed = await claimJobRow({ db: env.APP_DB, userId, - jobId: 'finalize-observability', - now: new Date(scheduledFor), - claimToken: 'claim-obs', + jobId: 'run-anchors', + now: new Date('2026-04-20T12:10:00.000Z'), + claimToken: 'claim-run-anchors', }) if (!claimed) throw new Error('Expected job claim.') - - const finishedAt = '2026-04-20T12:00:05.000Z' + const finalizedAt = '2026-04-20T12:10:05.000Z' expect( await finalizeClaimedJobRow({ db: env.APP_DB, userId, job: { ...claimed.record, - updatedAt: finishedAt, - lastRunAt: finishedAt, - lastRunStatus: 'success', - // Poison values: finalization must not copy these into D1. - lastRunError: 'should-not-persist', - lastDurationMs: 999, - runCount: 7, - successCount: 5, - errorCount: 2, - enabled: false, - nextRunAt: scheduledFor, + updatedAt: finalizedAt, + lastRunAt: finalizedAt, + lastRunStatus: 'error', }, callerContextJson: claimed.callerContextJson, - claimToken: 'claim-obs', + claimToken: 'claim-run-anchors', scheduledFor, }), ).toBe(true) - - const row = await getJobRowById(env.APP_DB, userId, 'finalize-observability') - expect(row).toMatchObject({ - last_run_at: finishedAt, - last_run_status: 'success', + expect(await getJobRowById(env.APP_DB, userId, 'run-anchors')).toMatchObject({ + last_run_at: finalizedAt, + last_run_status: 'error', last_completed_scheduled_for: scheduledFor, - claim_token: null, - enabled: 0, - last_run_error: null, - last_duration_ms: null, - run_count: 5, - success_count: 3, - error_count: 2, + record: { + lastRunAt: finalizedAt, + lastRunStatus: 'error', + lastRunError: undefined, + lastDurationMs: undefined, + runCount: 0, + successCount: 0, + errorCount: 0, + }, }) }) @@ -535,7 +414,6 @@ test('ordinary updates cancel claims and completed occurrence guards fence malfo test('expired jobs are skipped by due/claim/next-runnable and disableExpired flips enabled', async () => { await ensureJobsSchema() - const { disableExpiredJobRowsForUser } = await import('./repo.ts') const userId = 'user-expires' const nowIso = '2026-04-20T12:00:00.000Z' await insertJob({ diff --git a/packages/worker/src/jobs/run-due-jobs-claim-fence.node.test.ts b/packages/worker/src/jobs/run-due-jobs-claim-fence.node.test.ts index 8767f7a1db..2e634316ec 100644 --- a/packages/worker/src/jobs/run-due-jobs-claim-fence.node.test.ts +++ b/packages/worker/src/jobs/run-due-jobs-claim-fence.node.test.ts @@ -102,12 +102,7 @@ function claimedRow(record: JobRecord) { updated_at: record.updatedAt, last_run_at: null, last_run_status: null, - last_run_error: null, - last_duration_ms: null, next_run_at: record.nextRunAt, - run_count: 0, - success_count: 0, - error_count: 0, claim_token: 'claim-token', running_since: scheduledFor, lease_expires_at: '2026-07-30T19:10:00.000Z', diff --git a/packages/worker/src/jobs/service.node.test.ts b/packages/worker/src/jobs/service.node.test.ts index 5e8c5c17d9..ac137aecb6 100644 --- a/packages/worker/src/jobs/service.node.test.ts +++ b/packages/worker/src/jobs/service.node.test.ts @@ -807,12 +807,7 @@ function createDatabase( updated_at: params[16], last_run_at: params[17], last_run_status: params[18], - last_run_error: params[19], - last_duration_ms: params[20], - next_run_at: params[21], - run_count: params[22], - success_count: params[23], - error_count: params[24], + next_run_at: params[19], } upsert( 'jobs', @@ -856,22 +851,15 @@ function createDatabase( return { meta: { changes, last_row_id: 0 } } } if (query.startsWith('UPDATE jobs SET')) { - // last_run_at/status + next_run_at; clear error/duration - // only when last_run_at changes (CASE WHEN last_run_at IS ?). const existingJob = selectOne( 'jobs', (existing) => - existing['id'] === params[19] && - existing['user_id'] === params[20], + existing['id'] === params[17] && + existing['user_id'] === params[18], ) - const nextLastRunAt = params[14] - const previousLastRunAt = existingJob?.['last_run_at'] ?? null - const lastRunAtChanged = - previousLastRunAt !== nextLastRunAt && - !(previousLastRunAt == null && nextLastRunAt == null) const row = { - id: params[19], - user_id: params[20], + id: params[17], + user_id: params[18], name: params[0], source_id: params[1], published_commit: params[2], @@ -886,18 +874,9 @@ function createDatabase( expires_at: params[11], caller_context_json: params[12], updated_at: params[13], - last_run_at: nextLastRunAt, + last_run_at: params[14], last_run_status: params[15], - last_run_error: lastRunAtChanged - ? null - : (existingJob?.['last_run_error'] ?? null), - last_duration_ms: lastRunAtChanged - ? null - : (existingJob?.['last_duration_ms'] ?? null), - next_run_at: params[18], - run_count: existingJob?.['run_count'] ?? 0, - success_count: existingJob?.['success_count'] ?? 0, - error_count: existingJob?.['error_count'] ?? 0, + next_run_at: params[16], created_at: existingJob?.['created_at'] ?? params[13], } upsert( diff --git a/packages/worker/src/mcp/run-kody-registry.node.test.ts b/packages/worker/src/mcp/run-kody-registry.node.test.ts index 109f3edf1f..3a8c506f1a 100644 --- a/packages/worker/src/mcp/run-kody-registry.node.test.ts +++ b/packages/worker/src/mcp/run-kody-registry.node.test.ts @@ -701,21 +701,13 @@ function createJobMutationDatabase(input: { return { meta: { changes: 1, last_row_id: 0 } } } if (normalized.startsWith('UPDATE jobs SET')) { - // Matches updateJobRow: last_run_at/status, then CASE - // compares for clearing legacy error/duration, then - // next_run_at, id, user_id. - const id = params[19] - const userId = params[20] + const id = params[17] + const userId = params[18] const existing = selectOne( 'jobs', (row) => row['id'] === id && row['user_id'] === userId, ) if (!existing) return { meta: { changes: 0, last_row_id: 0 } } - const nextLastRunAt = params[14] - const previousLastRunAt = existing['last_run_at'] ?? null - const lastRunAtUnchanged = - previousLastRunAt === nextLastRunAt || - (previousLastRunAt == null && nextLastRunAt == null) const updated = { ...existing, name: params[0], @@ -732,15 +724,9 @@ function createJobMutationDatabase(input: { expires_at: params[11], caller_context_json: params[12], updated_at: params[13], - last_run_at: nextLastRunAt, + last_run_at: params[14], last_run_status: params[15], - last_run_error: lastRunAtUnchanged - ? existing['last_run_error'] - : null, - last_duration_ms: lastRunAtUnchanged - ? existing['last_duration_ms'] - : null, - next_run_at: params[18], + next_run_at: params[16], } const rows = table('jobs') const index = rows.findIndex( @@ -892,12 +878,7 @@ function createJobRow( updated_at: job.updatedAt, last_run_at: job.lastRunAt ?? null, last_run_status: job.lastRunStatus ?? null, - last_run_error: job.lastRunError ?? null, - last_duration_ms: job.lastDurationMs ?? null, next_run_at: job.nextRunAt, - run_count: job.runCount, - success_count: job.successCount, - error_count: job.errorCount, } } diff --git a/packages/worker/src/run-records/admin-insights-snapshot.ts b/packages/worker/src/run-records/admin-insights-snapshot.ts index 42c780a28c..b4318f56b8 100644 --- a/packages/worker/src/run-records/admin-insights-snapshot.ts +++ b/packages/worker/src/run-records/admin-insights-snapshot.ts @@ -1,9 +1,9 @@ /** * Content-free per-user RunLog point-read for post-D1 admin insights. * - * Aggregators may sum workflow status counts and activation funnel/latency - * across users. Never includes workflow/package/job names, errors, logs, or - * other user-authored content. + * Aggregators may sum workflow statuses, job outcome counts, and activation + * funnel/latency across users. Never includes workflow/package/job names, + * errors, logs, or other user-authored content. */ import { type ActivationMilestone } from './package-activation-state.ts' @@ -24,4 +24,8 @@ export type RunLogAdminInsightsActivationMilestone = { export type RunLogAdminInsightsSnapshot = { workflowStatusCounts: Array activationMilestones: Array + jobRunCounts: { + success: number + error: number + } } diff --git a/packages/worker/src/run-records/dedicated-state.workers.test.ts b/packages/worker/src/run-records/dedicated-state.workers.test.ts index dc4db09da8..5f4adb37ed 100644 --- a/packages/worker/src/run-records/dedicated-state.workers.test.ts +++ b/packages/worker/src/run-records/dedicated-state.workers.test.ts @@ -1203,8 +1203,8 @@ function jobRunObservabilityColumnNames(state: DurableObjectState) { .map((row) => String(row.name)) } -test('fresh schema v8 creates job_run_observability with inert legacy_seeded for rollback', async () => { - const userId = uniqueUserId('schema-v8-fresh') +test('fresh schema v9 creates the final job_run_observability contract', async () => { + const userId = uniqueUserId('schema-v9-fresh') const stub = env.RUN_LOG.get(env.RUN_LOG.idFromName(userId)) await runInDurableObject(stub, async (instance: RunLog, state) => { expect(instance).toBeInstanceOf(RunLog) @@ -1213,7 +1213,7 @@ test('fresh schema v8 creates job_run_observability with inert legacy_seeded for `SELECT value FROM run_log_meta WHERE key = 'schema_version' LIMIT 1`, ) .toArray()[0] - expect(Number(version?.value)).toBe(8) + expect(Number(version?.value)).toBe(9) expect(jobRunObservabilityColumnNames(state)).toEqual([ 'job_id', @@ -1225,134 +1225,139 @@ test('fresh schema v8 creates job_run_observability with inert legacy_seeded for 'success_count', 'error_count', 'updated_at', - 'legacy_seeded', ]) - - // Simulated v7-style INSERT including legacy_seeded must succeed on - // fresh v8 storage so a rollback deploy can still write the column. - state.storage.sql.exec( - `INSERT INTO job_run_observability ( - job_id, last_run_at, last_run_status, last_run_error, last_duration_ms, - run_count, success_count, error_count, updated_at, legacy_seeded - ) VALUES ('job-v7-insert', '2026-08-01T00:00:00.000Z', 'success', NULL, 10, - 3, 2, 1, '2026-08-01T00:00:00.000Z', 1)`, - ) - const seeded = state.storage.sql - .exec<{ legacy_seeded: number }>( - `SELECT legacy_seeded FROM job_run_observability - WHERE job_id = 'job-v7-insert' LIMIT 1`, - ) - .toArray()[0] - expect(Number(seeded?.legacy_seeded)).toBe(1) }) const updated = await upsertJobRunObservability({ env, userId, outcome: { - jobId: 'job-v7-insert', + jobId: 'job-v9', status: 'error', ranAt: '2026-08-01T01:00:00.000Z', - error: 'post-v8', + error: 'fresh-v9', }, }) expect(updated).toMatchObject({ - jobId: 'job-v7-insert', - runCount: 4, - successCount: 2, - errorCount: 2, + jobId: 'job-v9', + runCount: 1, + successCount: 0, + errorCount: 1, lastRunStatus: 'error', - lastRunError: 'post-v8', + lastRunError: 'fresh-v9', }) - expect(updated).not.toHaveProperty('legacySeeded') }) -test('warm schema v7 job_run_observability upgrades to v8 without ALTER', async () => { - const userId = uniqueUserId('schema-v7-warm') - const stub = env.RUN_LOG.get(env.RUN_LOG.idFromName(userId)) - await runInDurableObject(stub, async (instance: RunLog, state) => { - expect(instance).toBeInstanceOf(RunLog) - await state.storage.deleteAll() - - state.storage.sql.exec(` - CREATE TABLE run_log_meta ( - key TEXT PRIMARY KEY NOT NULL, - value INTEGER NOT NULL +test('warm schema v7 and v8 objects upgrade to v9 without losing job data', async () => { + const retiredColumn = ['legacy', 'seeded'].join('_') + for (const installedVersion of [7, 8]) { + const userId = uniqueUserId(`schema-v${installedVersion}-warm`) + const stub = env.RUN_LOG.get(env.RUN_LOG.idFromName(userId)) + await runInDurableObject(stub, async (instance: RunLog, state) => { + expect(instance).toBeInstanceOf(RunLog) + await state.storage.deleteAll() + + state.storage.sql.exec(` + CREATE TABLE run_log_meta ( + key TEXT PRIMARY KEY NOT NULL, + value INTEGER NOT NULL + ) + `) + state.storage.sql.exec( + `INSERT INTO run_log_meta (key, value) VALUES ('schema_version', ?)`, + installedVersion, ) - `) - state.storage.sql.exec( - `INSERT INTO run_log_meta (key, value) VALUES ('schema_version', 7)`, - ) - state.storage.sql.exec(` - CREATE TABLE job_run_observability ( - job_id TEXT PRIMARY KEY NOT NULL, - last_run_at TEXT, - last_run_status TEXT, - last_run_error TEXT, - last_duration_ms INTEGER, - run_count INTEGER NOT NULL DEFAULT 0, - success_count INTEGER NOT NULL DEFAULT 0, - error_count INTEGER NOT NULL DEFAULT 0, - updated_at TEXT NOT NULL, - legacy_seeded INTEGER NOT NULL DEFAULT 0 + state.storage.sql.exec(` + CREATE TABLE job_run_observability ( + job_id TEXT PRIMARY KEY NOT NULL, + last_run_at TEXT, + last_run_status TEXT, + last_run_error TEXT, + last_duration_ms INTEGER, + run_count INTEGER NOT NULL DEFAULT 0, + success_count INTEGER NOT NULL DEFAULT 0, + error_count INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL, + ${retiredColumn} INTEGER NOT NULL DEFAULT 0 + ) + `) + state.storage.sql.exec( + `INSERT INTO job_run_observability ( + job_id, last_run_at, last_run_status, last_run_error, + last_duration_ms, run_count, success_count, error_count, + updated_at, ${retiredColumn} + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1)`, + `job-warm-v${installedVersion}`, + '2026-07-01T00:00:00.000Z', + 'error', + 'preserved error', + 57, + 9, + 6, + 3, + '2026-07-01T00:00:01.000Z', ) - `) - state.storage.sql.exec( - `INSERT INTO job_run_observability ( - job_id, last_run_at, last_run_status, last_run_error, last_duration_ms, - run_count, success_count, error_count, updated_at, legacy_seeded - ) VALUES ('job-warm-v7', '2026-07-01T00:00:00.000Z', 'success', NULL, 5, - 1, 1, 0, '2026-07-01T00:00:00.000Z', 1)`, - ) - const alterStatements: Array = [] - const originalExec = state.storage.sql.exec.bind(state.storage.sql) - state.storage.sql.exec = (query: string, ...args: Array) => { - if (/ALTER\s+TABLE/i.test(query)) { - alterStatements.push(query) + const proto = Object.getPrototypeOf(instance) as { + initializeSchema: () => void } - return originalExec(query, ...args) - } - - const proto = Object.getPrototypeOf(instance) as { - initializeSchema: () => void - } - proto.initializeSchema.call(instance) - - expect(alterStatements).toEqual([]) - const version = state.storage.sql - .exec<{ value: number }>( - `SELECT value FROM run_log_meta WHERE key = 'schema_version' LIMIT 1`, - ) - .toArray()[0] - expect(Number(version?.value)).toBe(8) - expect(jobRunObservabilityColumnNames(state)).toContain('legacy_seeded') - - const preserved = state.storage.sql - .exec<{ legacy_seeded: number; run_count: number }>( - `SELECT legacy_seeded, run_count FROM job_run_observability - WHERE job_id = 'job-warm-v7' LIMIT 1`, - ) - .toArray()[0] - expect(Number(preserved?.legacy_seeded)).toBe(1) - expect(Number(preserved?.run_count)).toBe(1) - }) + proto.initializeSchema.call(instance) + + const version = state.storage.sql + .exec<{ value: number }>( + `SELECT value FROM run_log_meta WHERE key = 'schema_version' LIMIT 1`, + ) + .toArray()[0] + expect(Number(version?.value)).toBe(9) + expect(jobRunObservabilityColumnNames(state)).toEqual([ + 'job_id', + 'last_run_at', + 'last_run_status', + 'last_run_error', + 'last_duration_ms', + 'run_count', + 'success_count', + 'error_count', + 'updated_at', + ]) + expect( + state.storage.sql + .exec>( + `SELECT * FROM job_run_observability WHERE job_id = ?`, + `job-warm-v${installedVersion}`, + ) + .one(), + ).toEqual({ + job_id: `job-warm-v${installedVersion}`, + last_run_at: '2026-07-01T00:00:00.000Z', + last_run_status: 'error', + last_run_error: 'preserved error', + last_duration_ms: 57, + run_count: 9, + success_count: 6, + error_count: 3, + updated_at: '2026-07-01T00:00:01.000Z', + }) + }) - const read = await getJobRunObservability({ - env, - userId, - jobId: 'job-warm-v7', - }) - expect(read).toMatchObject({ - jobId: 'job-warm-v7', - runCount: 1, - successCount: 1, - lastRunStatus: 'success', - }) - expect(read).not.toHaveProperty('legacySeeded') + await expect( + getJobRunObservability({ + env, + userId, + jobId: `job-warm-v${installedVersion}`, + }), + ).resolves.toMatchObject({ + jobId: `job-warm-v${installedVersion}`, + runCount: 9, + successCount: 6, + errorCount: 3, + lastRunStatus: 'error', + lastRunError: 'preserved error', + }) + } }) -test('getAdminInsightsSnapshot returns content-free workflow counts and milestones', async () => { +test('getAdminInsightsSnapshot returns content-free workflow, job, and activation aggregates', async () => { silenceExpectedConsoleWarns(['activation-run-record-failed']) const userId = uniqueUserId('admin-insights') const reachedAt = '2026-08-01T12:00:00.000Z' @@ -1432,6 +1437,16 @@ test('getAdminInsightsSnapshot returns content-free workflow counts and mileston }), status: 'success', }) + await upsertJobRunObservability({ + env, + userId, + outcome: { + jobId: 'secret-job-must-not-leak', + status: 'error', + ranAt: reachedAt, + error: 'secret-job-error-must-not-leak', + }, + }) const snapshot = await getAdminInsightsSnapshot({ env, userId }) expect(snapshot.workflowStatusCounts).toEqual([ @@ -1450,16 +1465,23 @@ test('getAdminInsightsSnapshot returns content-free workflow counts and mileston reachedAt: expect.any(String), }), ]) + expect(snapshot.jobRunCounts).toEqual({ success: 1, error: 1 }) const serialized = JSON.stringify(snapshot) expect(serialized).not.toMatch(/secret-name-must-not-leak/) expect(serialized).not.toMatch(/secret-error-must-not-leak/) + expect(serialized).not.toMatch(/secret-job-must-not-leak/) + expect(serialized).not.toMatch(/secret-job-error-must-not-leak/) expect(serialized).not.toMatch(/job-admin/) expect(serialized).not.toMatch(/"name"/) expect(serialized).not.toMatch(/lastError|errorMessage|workflowName|"logs"/) for (const key of Object.keys(snapshot)) { - expect(['workflowStatusCounts', 'activationMilestones']).toContain(key) + expect([ + 'workflowStatusCounts', + 'activationMilestones', + 'jobRunCounts', + ]).toContain(key) } for (const row of snapshot.workflowStatusCounts) { expect(Object.keys(row).sort()).toEqual(['count', 'status']) diff --git a/packages/worker/src/run-records/run-log-do.ts b/packages/worker/src/run-records/run-log-do.ts index f0c5839316..e24a59e9d0 100644 --- a/packages/worker/src/run-records/run-log-do.ts +++ b/packages/worker/src/run-records/run-log-do.ts @@ -60,7 +60,7 @@ const metaRunCountKey = 'run_count' const metaFinishesSinceRetentionKey = 'finishes_since_retention' const metaSchemaVersionKey = 'schema_version' /** Bump when initializeSchema's DDL set changes; warm objects skip DDL. */ -const runLogSchemaVersion = 8 +const runLogSchemaVersion = 9 /** * Cursor namespaces for `exportRuns` phases after the raw run-id phase. @@ -613,10 +613,6 @@ class RunLogBase extends DurableObject { ON workflow_projections(idempotency_key, created_at ASC)`, ) // Dedicated unpruned per-job last-run outcome + counters. Not runs. - // 2026-08-04 rollback compat: fresh CREATE keeps the inert physical - // `legacy_seeded` column so a rollback to schema v7 code can still - // INSERT/UPDATE it. v8 never reads, writes, or exposes the field. - // Remove only via a future table rebuild after the rollback window. this.ctx.storage.sql.exec(` CREATE TABLE IF NOT EXISTS job_run_observability ( job_id TEXT PRIMARY KEY NOT NULL, @@ -627,10 +623,12 @@ class RunLogBase extends DurableObject { run_count INTEGER NOT NULL DEFAULT 0, success_count INTEGER NOT NULL DEFAULT 0, error_count INTEGER NOT NULL DEFAULT 0, - updated_at TEXT NOT NULL, - legacy_seeded INTEGER NOT NULL DEFAULT 0 + updated_at TEXT NOT NULL ) `) + if (installedVersion != null && installedVersion < 9) { + this.rebuildJobRunObservabilityForV9() + } this.ctx.storage.sql.exec( `CREATE INDEX IF NOT EXISTS idx_workflow_projections_binding_idempotency ON workflow_projections(binding_name, idempotency_key, created_at ASC)`, @@ -655,6 +653,38 @@ class RunLogBase extends DurableObject { this.setMeta(metaSchemaVersionKey, runLogSchemaVersion) } + private rebuildJobRunObservabilityForV9() { + this.ctx.storage.transactionSync(() => { + this.ctx.storage.sql.exec( + `CREATE TABLE job_run_observability_v9 ( + job_id TEXT PRIMARY KEY NOT NULL, + last_run_at TEXT, + last_run_status TEXT, + last_run_error TEXT, + last_duration_ms INTEGER, + run_count INTEGER NOT NULL DEFAULT 0, + success_count INTEGER NOT NULL DEFAULT 0, + error_count INTEGER NOT NULL DEFAULT 0, + updated_at TEXT NOT NULL + )`, + ) + this.ctx.storage.sql.exec( + `INSERT INTO job_run_observability_v9 ( + job_id, last_run_at, last_run_status, last_run_error, + last_duration_ms, run_count, success_count, error_count, updated_at + ) + SELECT + job_id, last_run_at, last_run_status, last_run_error, + last_duration_ms, run_count, success_count, error_count, updated_at + FROM job_run_observability`, + ) + this.ctx.storage.sql.exec(`DROP TABLE job_run_observability`) + this.ctx.storage.sql.exec( + `ALTER TABLE job_run_observability_v9 RENAME TO job_run_observability`, + ) + }) + } + private getMeta(key: string): number | null { const row = this.ctx.storage.sql .exec<{ value: number }>( @@ -1783,10 +1813,9 @@ class RunLogBase extends DurableObject { } /** - * Release a claim whose execution never started (transient artifact - * failure, dual-read fallback hit) so retries are not poisoned. Deletes - * the still-`in_progress` ledger row (fenced on `claimUpdatedAt`) and the - * attempt's still-`running` run row. + * Release a claim whose execution never started so retries are not + * poisoned. Deletes the still-`in_progress` ledger row (fenced on + * `claimUpdatedAt`) and the attempt's still-`running` run row. */ async releasePackageInvocation(input: { invocationId: string @@ -2132,9 +2161,9 @@ class RunLogBase extends DurableObject { } /** - * Content-free admin-insights point-read. Workflow status counts across all - * dedicated projections plus activation milestone timestamps/ids needed for - * funnel counts and latency — never names, errors, logs, or user content. + * Content-free admin-insights point-read. Workflow statuses, job outcome + * counts, and activation milestone timestamps/ids needed for aggregate + * reporting — never names, errors, logs, or user content. */ async getAdminInsightsSnapshot(): Promise { const workflowRows = this.ctx.storage.sql @@ -2152,6 +2181,14 @@ class RunLogBase extends DurableObject { ) .toArray() .map(mapActivationMilestoneRow) + const jobRunCounts = this.ctx.storage.sql + .exec<{ error: number; success: number }>( + `SELECT + COALESCE(SUM(success_count), 0) AS success, + COALESCE(SUM(error_count), 0) AS error + FROM job_run_observability`, + ) + .one() return { workflowStatusCounts: workflowRows.map((row) => ({ status: String(row.status), @@ -2162,6 +2199,10 @@ class RunLogBase extends DurableObject { reachedAt: row.reachedAt, packageId: row.packageId, })), + jobRunCounts: { + success: Number(jobRunCounts.success) || 0, + error: Number(jobRunCounts.error) || 0, + }, } } diff --git a/packages/worker/src/run-records/service.node.test.ts b/packages/worker/src/run-records/service.node.test.ts index a388b2ebba..0bccd930bd 100644 --- a/packages/worker/src/run-records/service.node.test.ts +++ b/packages/worker/src/run-records/service.node.test.ts @@ -10,6 +10,7 @@ const mocks = vi.hoisted(() => ({ getAdminInsightsSnapshot: vi.fn(async () => ({ workflowStatusCounts: [], activationMilestones: [], + jobRunCounts: { success: 0, error: 0 }, })), })) @@ -201,6 +202,7 @@ test('getAdminInsightsSnapshot requires RUN_LOG and forwards the RPC', async () mocks.getAdminInsightsSnapshot.mockResolvedValueOnce({ workflowStatusCounts: [{ status: 'running', count: 2 }], + jobRunCounts: { success: 8, error: 3 }, activationMilestones: [ { milestone: 'package_activated', @@ -214,6 +216,7 @@ test('getAdminInsightsSnapshot requires RUN_LOG and forwards the RPC', async () getAdminInsightsSnapshot({ env, userId: 'user-1' }), ).resolves.toEqual({ workflowStatusCounts: [{ status: 'running', count: 2 }], + jobRunCounts: { success: 8, error: 3 }, activationMilestones: [ { milestone: 'package_activated', diff --git a/packages/worker/src/run-records/service.ts b/packages/worker/src/run-records/service.ts index a1c5225c64..69a621ccda 100644 --- a/packages/worker/src/run-records/service.ts +++ b/packages/worker/src/run-records/service.ts @@ -736,9 +736,8 @@ export async function finishPackageInvocationRecord(input: { } /** - * Release a claim whose execution never started (transient artifact failure - * or the dual-read D1 fallback finding a pre-migration owner). Deletes the - * still-`in_progress` ledger row and the attempt's `running` run row. + * Release a claim whose execution never started. Deletes the still-`in_progress` + * ledger row and the attempt's `running` run row. */ export async function releasePackageInvocationRecord(input: { env: Env diff --git a/tools/migration-ledger.json b/tools/migration-ledger.json index 9603e70846..4317a99dbf 100644 --- a/tools/migration-ledger.json +++ b/tools/migration-ledger.json @@ -576,6 +576,10 @@ "filename": "0141-drop-account-write-leases-and-stale-flags.sql", "sha256": "eca135651ff1e7b30c3b611771b69051f8cff46868eb98ce59fc91e0e71a807e" }, + { + "filename": "0142-drop-jobs-observability-columns.sql", + "sha256": "abb01c15071ad39c97a4c9c22ee0d7bcb68be62839593ea0e682c4f153540258" + }, { "filename": "0143-drop-email-postdrop-residue.sql", "sha256": "126f57c6fcab61f8db28113026b2d2b60d9e80671876cad89600df1bba92f508" From e44e883d046445930b85fef173b45fbfba7afa89 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 4 Aug 2026 14:12:03 +0000 Subject: [PATCH 2/3] Remove final jobs observability schema readers Co-authored-by: Kent C. Dodds --- .../worker/src/community/community-flow-test-schema.ts | 7 +------ packages/worker/src/entitlements/service.ts | 1 - 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/packages/worker/src/community/community-flow-test-schema.ts b/packages/worker/src/community/community-flow-test-schema.ts index e04b5a3fab..6682ec0c6e 100644 --- a/packages/worker/src/community/community-flow-test-schema.ts +++ b/packages/worker/src/community/community-flow-test-schema.ts @@ -177,12 +177,7 @@ export async function ensureCommunityFlowSchema(db: D1Database) { updated_at TEXT NOT NULL, last_run_at TEXT, last_run_status TEXT, - last_run_error TEXT, - last_duration_ms INTEGER, - next_run_at TEXT NOT NULL, - run_count INTEGER NOT NULL DEFAULT 0, - success_count INTEGER NOT NULL DEFAULT 0, - error_count INTEGER NOT NULL DEFAULT 0 + next_run_at TEXT NOT NULL )`, `CREATE TABLE IF NOT EXISTS published_bundle_artifacts ( id TEXT PRIMARY KEY, diff --git a/packages/worker/src/entitlements/service.ts b/packages/worker/src/entitlements/service.ts index 76052884a1..e175d1ac99 100644 --- a/packages/worker/src/entitlements/service.ts +++ b/packages/worker/src/entitlements/service.ts @@ -463,7 +463,6 @@ export async function calculateUserD1StorageBytes(input: { 'timezone', 'caller_context_json', 'last_run_status', - 'last_run_error', 'source_id', 'published_commit', 'storage_id', From 7e2c2353e57f1eb29f772d3975d3b6cb565f0fac Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 4 Aug 2026 15:42:24 +0000 Subject: [PATCH 3/3] Renumber jobs observability migration to 0144 Co-authored-by: Kent C. Dodds --- ...lumns.sql => 0144-drop-jobs-observability-columns.sql} | 0 tools/migration-ledger.json | 8 ++++---- 2 files changed, 4 insertions(+), 4 deletions(-) rename packages/worker/migrations/{0142-drop-jobs-observability-columns.sql => 0144-drop-jobs-observability-columns.sql} (100%) diff --git a/packages/worker/migrations/0142-drop-jobs-observability-columns.sql b/packages/worker/migrations/0144-drop-jobs-observability-columns.sql similarity index 100% rename from packages/worker/migrations/0142-drop-jobs-observability-columns.sql rename to packages/worker/migrations/0144-drop-jobs-observability-columns.sql diff --git a/tools/migration-ledger.json b/tools/migration-ledger.json index 4317a99dbf..578c7723bb 100644 --- a/tools/migration-ledger.json +++ b/tools/migration-ledger.json @@ -576,13 +576,13 @@ "filename": "0141-drop-account-write-leases-and-stale-flags.sql", "sha256": "eca135651ff1e7b30c3b611771b69051f8cff46868eb98ce59fc91e0e71a807e" }, - { - "filename": "0142-drop-jobs-observability-columns.sql", - "sha256": "abb01c15071ad39c97a4c9c22ee0d7bcb68be62839593ea0e682c4f153540258" - }, { "filename": "0143-drop-email-postdrop-residue.sql", "sha256": "126f57c6fcab61f8db28113026b2d2b60d9e80671876cad89600df1bba92f508" + }, + { + "filename": "0144-drop-jobs-observability-columns.sql", + "sha256": "abb01c15071ad39c97a4c9c22ee0d7bcb68be62839593ea0e682c4f153540258" } ] }