From ba999c93222ce907c8637dede9576b85eecbd506 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=80=82?= Date: Wed, 8 Jul 2026 17:02:51 +0800 Subject: [PATCH] =?UTF-8?q?feat(jobs):=20=E5=A2=9E=E5=8A=A0=E5=90=8E?= =?UTF-8?q?=E5=8F=B0=E4=BB=BB=E5=8A=A1=E8=BF=90=E8=A1=8C=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../migration.sql | 32 +++++ apps/server/prisma/schema.prisma | 19 +++ apps/server/src/app.module.ts | 2 + .../jobs/background-job.worker.spec.ts | 48 ++++++++ .../src/modules/jobs/background-job.worker.ts | 58 +++++++++ .../src/modules/jobs/job-lock.service.spec.ts | 51 ++++++++ .../src/modules/jobs/job-lock.service.ts | 57 +++++++++ apps/server/src/modules/jobs/jobs.module.ts | 17 +++ .../src/modules/jobs/jobs.service.spec.ts | 115 ++++++++++++++++++ apps/server/src/modules/jobs/jobs.service.ts | 104 ++++++++++++++++ apps/server/src/modules/jobs/jobs.types.ts | 38 ++++++ docs/architecture.md | 12 ++ docs/decisions.md | 14 +++ docs/roadmap.md | 10 +- 14 files changed, 575 insertions(+), 2 deletions(-) create mode 100644 apps/server/prisma/migrations/20260708040000_v26_background_jobs/migration.sql create mode 100644 apps/server/src/modules/jobs/background-job.worker.spec.ts create mode 100644 apps/server/src/modules/jobs/background-job.worker.ts create mode 100644 apps/server/src/modules/jobs/job-lock.service.spec.ts create mode 100644 apps/server/src/modules/jobs/job-lock.service.ts create mode 100644 apps/server/src/modules/jobs/jobs.module.ts create mode 100644 apps/server/src/modules/jobs/jobs.service.spec.ts create mode 100644 apps/server/src/modules/jobs/jobs.service.ts create mode 100644 apps/server/src/modules/jobs/jobs.types.ts diff --git a/apps/server/prisma/migrations/20260708040000_v26_background_jobs/migration.sql b/apps/server/prisma/migrations/20260708040000_v26_background_jobs/migration.sql new file mode 100644 index 0000000..acae193 --- /dev/null +++ b/apps/server/prisma/migrations/20260708040000_v26_background_jobs/migration.sql @@ -0,0 +1,32 @@ +CREATE TABLE "background_jobs" ( + "id" TEXT NOT NULL, + "type" TEXT NOT NULL, + "status" TEXT NOT NULL DEFAULT 'queued', + "dedupe_key" TEXT, + "payload" JSONB NOT NULL DEFAULT '{}', + "attempts" INTEGER NOT NULL DEFAULT 0, + "max_attempts" INTEGER NOT NULL DEFAULT 3, + "available_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "locked_by" TEXT, + "locked_until" TIMESTAMP(3), + "last_error" TEXT, + "completed_at" TIMESTAMP(3), + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "background_jobs_pkey" PRIMARY KEY ("id") +); + +CREATE INDEX "background_jobs_status_available_at_idx" + ON "background_jobs"("status", "available_at" ASC, "created_at" ASC); + +CREATE INDEX "background_jobs_type_status_available_at_idx" + ON "background_jobs"("type", "status", "available_at" ASC); + +CREATE INDEX "background_jobs_lease_idx" + ON "background_jobs"("status", "locked_until" ASC) + WHERE "locked_until" IS NOT NULL; + +CREATE UNIQUE INDEX "background_jobs_active_dedupe_key_idx" + ON "background_jobs"("type", "dedupe_key") + WHERE "dedupe_key" IS NOT NULL AND "status" IN ('queued', 'running'); diff --git a/apps/server/prisma/schema.prisma b/apps/server/prisma/schema.prisma index 005491c..fc692b5 100644 --- a/apps/server/prisma/schema.prisma +++ b/apps/server/prisma/schema.prisma @@ -423,6 +423,25 @@ model AiLog { @@map("ai_logs") } +model BackgroundJob { + id String @id @default(cuid()) + type String + status String @default("queued") + dedupeKey String? @map("dedupe_key") + payload Json @default("{}") + attempts Int @default(0) + maxAttempts Int @default(3) @map("max_attempts") + availableAt DateTime @default(now()) @map("available_at") + lockedBy String? @map("locked_by") + lockedUntil DateTime? @map("locked_until") + lastError String? @map("last_error") + completedAt DateTime? @map("completed_at") + createdAt DateTime @default(now()) @map("created_at") + updatedAt DateTime @updatedAt @map("updated_at") + + @@map("background_jobs") +} + model XiaobaoRiskSummary { versionId String @id @map("version_id") riskLevel String @map("risk_level") diff --git a/apps/server/src/app.module.ts b/apps/server/src/app.module.ts index bf16a90..c255c88 100644 --- a/apps/server/src/app.module.ts +++ b/apps/server/src/app.module.ts @@ -20,6 +20,7 @@ import { DataModule } from './modules/data/data.module'; import { MigrationModule } from './modules/migration/migration.module'; import { V22QueryModule } from './modules/v22-query/v22-query.module'; import { HealthModule } from './modules/health/health.module'; +import { JobsModule } from './modules/jobs/jobs.module'; @Module({ imports: [ @@ -41,6 +42,7 @@ import { HealthModule } from './modules/health/health.module'; MigrationModule, V22QueryModule, HealthModule, + JobsModule, AiModule, ], controllers: [], diff --git a/apps/server/src/modules/jobs/background-job.worker.spec.ts b/apps/server/src/modules/jobs/background-job.worker.spec.ts new file mode 100644 index 0000000..bbd4818 --- /dev/null +++ b/apps/server/src/modules/jobs/background-job.worker.spec.ts @@ -0,0 +1,48 @@ +import { BackgroundJobWorker } from './background-job.worker'; + +describe('BackgroundJobWorker', () => { + it('runs a registered handler and marks the job succeeded', async () => { + const lock = { + claimNext: jest.fn().mockResolvedValue({ + id: 'job-1', + type: 'demo.job', + payload: { ok: true }, + }), + }; + const jobs = { + markSucceeded: jest.fn(), + markFailed: jest.fn(), + }; + const worker = new BackgroundJobWorker(lock as any, jobs as any); + const handler = jest.fn().mockResolvedValue(undefined); + worker.registerHandler('demo.job', handler); + + await expect(worker.runOnce({ workerId: 'worker-a' })).resolves.toBe(true); + + expect(handler).toHaveBeenCalledWith({ ok: true }, expect.objectContaining({ id: 'job-1' })); + expect(jobs.markSucceeded).toHaveBeenCalledWith('job-1'); + expect(jobs.markFailed).not.toHaveBeenCalled(); + }); + + it('marks a job failed when the handler throws', async () => { + const lock = { + claimNext: jest.fn().mockResolvedValue({ + id: 'job-1', + type: 'demo.job', + payload: {}, + }), + }; + const jobs = { + markSucceeded: jest.fn(), + markFailed: jest.fn(), + }; + const worker = new BackgroundJobWorker(lock as any, jobs as any); + const error = new Error('bad handler'); + worker.registerHandler('demo.job', jest.fn().mockRejectedValue(error)); + + await expect(worker.runOnce({ workerId: 'worker-a' })).resolves.toBe(false); + + expect(jobs.markFailed).toHaveBeenCalledWith('job-1', error, expect.any(Object)); + expect(jobs.markSucceeded).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/server/src/modules/jobs/background-job.worker.ts b/apps/server/src/modules/jobs/background-job.worker.ts new file mode 100644 index 0000000..781ead6 --- /dev/null +++ b/apps/server/src/modules/jobs/background-job.worker.ts @@ -0,0 +1,58 @@ +import { Injectable } from '@nestjs/common'; +import { JobLockService } from './job-lock.service'; +import { JobsService } from './jobs.service'; +import { BackgroundJobRecord } from './jobs.types'; + +export type BackgroundJobHandler = (payload: unknown, job: BackgroundJobRecord) => Promise | void; + +export interface RunJobOnceOptions { + workerId: string; + now?: Date; + leaseMs?: number; + retryDelayMs?: number; +} + +@Injectable() +export class BackgroundJobWorker { + private readonly handlers = new Map(); + + constructor( + private readonly locks: JobLockService, + private readonly jobs: JobsService, + ) {} + + registerHandler(type: string, handler: BackgroundJobHandler) { + this.handlers.set(type, handler); + } + + async runOnce(options: RunJobOnceOptions): Promise { + const job = await this.locks.claimNext({ + workerId: options.workerId, + types: Array.from(this.handlers.keys()), + now: options.now, + leaseMs: options.leaseMs, + }); + if (!job) return false; + + const handler = this.handlers.get(job.type); + if (!handler) { + await this.jobs.markFailed(job.id, `No handler registered for ${job.type}`, { + now: options.now, + retryDelayMs: options.retryDelayMs, + }); + return false; + } + + try { + await handler(job.payload, job); + await this.jobs.markSucceeded(job.id); + return true; + } catch (error) { + await this.jobs.markFailed(job.id, error, { + now: options.now, + retryDelayMs: options.retryDelayMs, + }); + return false; + } + } +} diff --git a/apps/server/src/modules/jobs/job-lock.service.spec.ts b/apps/server/src/modules/jobs/job-lock.service.spec.ts new file mode 100644 index 0000000..fe5baa9 --- /dev/null +++ b/apps/server/src/modules/jobs/job-lock.service.spec.ts @@ -0,0 +1,51 @@ +import { JobLockService } from './job-lock.service'; + +describe('JobLockService', () => { + it('claims queued or expired jobs with a lease and increments attempts', async () => { + const tx = { + $queryRawUnsafe: jest.fn().mockResolvedValue([{ id: 'job-1' }]), + backgroundJob: { + update: jest.fn().mockResolvedValue({ id: 'job-1', status: 'running' }), + }, + }; + const prisma = { + $transaction: jest.fn((callback) => callback(tx)), + }; + const service = new JobLockService(prisma as any); + const now = new Date('2026-07-08T10:00:00.000Z'); + + const result = await service.claimNext({ + workerId: 'worker-a', + types: ['xiaobao.summary.refresh'], + now, + leaseMs: 60_000, + }); + + expect(result).toEqual({ id: 'job-1', status: 'running' }); + expect(tx.$queryRawUnsafe.mock.calls[0][0]).toContain('FOR UPDATE SKIP LOCKED'); + expect(tx.backgroundJob.update).toHaveBeenCalledWith({ + where: { id: 'job-1' }, + data: { + status: 'running', + lockedBy: 'worker-a', + lockedUntil: new Date('2026-07-08T10:01:00.000Z'), + attempts: { increment: 1 }, + lastError: null, + }, + }); + }); + + it('returns null when no claimable jobs exist', async () => { + const tx = { + $queryRawUnsafe: jest.fn().mockResolvedValue([]), + backgroundJob: { update: jest.fn() }, + }; + const prisma = { + $transaction: jest.fn((callback) => callback(tx)), + }; + const service = new JobLockService(prisma as any); + + await expect(service.claimNext({ workerId: 'worker-a' })).resolves.toBeNull(); + expect(tx.backgroundJob.update).not.toHaveBeenCalled(); + }); +}); diff --git a/apps/server/src/modules/jobs/job-lock.service.ts b/apps/server/src/modules/jobs/job-lock.service.ts new file mode 100644 index 0000000..701a143 --- /dev/null +++ b/apps/server/src/modules/jobs/job-lock.service.ts @@ -0,0 +1,57 @@ +import { Inject, Injectable } from '@nestjs/common'; +import { BackgroundJobRecord, ClaimJobInput, JOBS_PRISMA } from './jobs.types'; + +const DEFAULT_LEASE_MS = 60_000; + +@Injectable() +export class JobLockService { + constructor(@Inject(JOBS_PRISMA) private readonly prisma: any) {} + + async claimNext(input: ClaimJobInput): Promise { + const now = input.now ?? new Date(); + const leaseMs = input.leaseMs ?? DEFAULT_LEASE_MS; + const types = input.types?.map((type) => type.trim()).filter(Boolean) ?? []; + + return (this.prisma as any).$transaction(async (tx: any) => { + const params: unknown[] = [now]; + const typeCondition = buildTypeCondition(types, params); + const rows = await tx.$queryRawUnsafe( + ` +SELECT id +FROM background_jobs +WHERE ( + (status = 'queued' AND available_at <= $1) + OR (status = 'running' AND locked_until IS NOT NULL AND locked_until < $1) + ) + ${typeCondition} +ORDER BY available_at ASC, created_at ASC +LIMIT 1 +FOR UPDATE SKIP LOCKED +`, + ...params, + ); + const id = rows[0]?.id; + if (!id) return null; + + return tx.backgroundJob.update({ + where: { id }, + data: { + status: 'running', + lockedBy: input.workerId, + lockedUntil: new Date(now.getTime() + leaseMs), + attempts: { increment: 1 }, + lastError: null, + }, + }); + }); + } +} + +function buildTypeCondition(types: string[], params: unknown[]): string { + if (types.length === 0) return ''; + const placeholders = types.map((type) => { + params.push(type); + return `$${params.length}`; + }); + return `AND type IN (${placeholders.join(', ')})`; +} diff --git a/apps/server/src/modules/jobs/jobs.module.ts b/apps/server/src/modules/jobs/jobs.module.ts new file mode 100644 index 0000000..981974e --- /dev/null +++ b/apps/server/src/modules/jobs/jobs.module.ts @@ -0,0 +1,17 @@ +import { Module } from '@nestjs/common'; +import { PrismaService } from '../../prisma/prisma.service'; +import { BackgroundJobWorker } from './background-job.worker'; +import { JobLockService } from './job-lock.service'; +import { JobsService } from './jobs.service'; +import { JOBS_PRISMA } from './jobs.types'; + +@Module({ + providers: [ + { provide: JOBS_PRISMA, useExisting: PrismaService }, + JobsService, + JobLockService, + BackgroundJobWorker, + ], + exports: [JobsService, JobLockService, BackgroundJobWorker], +}) +export class JobsModule {} diff --git a/apps/server/src/modules/jobs/jobs.service.spec.ts b/apps/server/src/modules/jobs/jobs.service.spec.ts new file mode 100644 index 0000000..401dd2d --- /dev/null +++ b/apps/server/src/modules/jobs/jobs.service.spec.ts @@ -0,0 +1,115 @@ +import { JobsService } from './jobs.service'; + +describe('JobsService', () => { + function makeService() { + const prisma = { + backgroundJob: { + findFirst: jest.fn(), + findUnique: jest.fn(), + create: jest.fn(), + update: jest.fn(), + }, + }; + return { prisma, service: new JobsService(prisma as any) }; + } + + it('dedupes active jobs by dedupeKey before creating a new row', async () => { + const { prisma, service } = makeService(); + prisma.backgroundJob.findFirst.mockResolvedValue({ id: 'job-1', status: 'queued' }); + + const result = await service.enqueue({ + type: 'xiaobao.summary.refresh', + dedupeKey: 'version-1', + payload: { versionId: 'version-1' }, + }); + + expect(result).toEqual({ id: 'job-1', status: 'queued' }); + expect(prisma.backgroundJob.findFirst).toHaveBeenCalledWith({ + where: { + type: 'xiaobao.summary.refresh', + dedupeKey: 'version-1', + status: { in: ['queued', 'running'] }, + }, + orderBy: { createdAt: 'asc' }, + }); + expect(prisma.backgroundJob.create).not.toHaveBeenCalled(); + }); + + it('creates a queued job when no active dedupe match exists', async () => { + const { prisma, service } = makeService(); + prisma.backgroundJob.findFirst.mockResolvedValue(null); + prisma.backgroundJob.create.mockResolvedValue({ id: 'job-2', status: 'queued' }); + + await service.enqueue({ + type: 'xiaobao.summary.refresh', + dedupeKey: 'version-2', + payload: { versionId: 'version-2' }, + maxAttempts: 5, + }); + + expect(prisma.backgroundJob.create).toHaveBeenCalledWith({ + data: { + type: 'xiaobao.summary.refresh', + dedupeKey: 'version-2', + payload: { versionId: 'version-2' }, + maxAttempts: 5, + availableAt: undefined, + }, + }); + }); + + it('recovers from concurrent dedupe unique conflicts by returning the active job', async () => { + const { prisma, service } = makeService(); + prisma.backgroundJob.findFirst + .mockResolvedValueOnce(null) + .mockResolvedValueOnce({ id: 'job-existing', status: 'queued' }); + prisma.backgroundJob.create.mockRejectedValue(Object.assign(new Error('duplicate'), { code: 'P2002' })); + + const result = await service.enqueue({ + type: 'xiaobao.summary.refresh', + dedupeKey: 'version-2', + payload: { versionId: 'version-2' }, + }); + + expect(result).toEqual({ id: 'job-existing', status: 'queued' }); + expect(prisma.backgroundJob.findFirst).toHaveBeenCalledTimes(2); + }); + + it('requeues failed jobs while attempts remain', async () => { + const { prisma, service } = makeService(); + const now = new Date('2026-07-08T10:00:00.000Z'); + prisma.backgroundJob.findUnique.mockResolvedValue({ id: 'job-1', attempts: 1, maxAttempts: 3 }); + + await service.markFailed('job-1', new Error('boom'), { now, retryDelayMs: 30_000 }); + + expect(prisma.backgroundJob.update).toHaveBeenCalledWith({ + where: { id: 'job-1' }, + data: { + status: 'queued', + availableAt: new Date('2026-07-08T10:00:30.000Z'), + lockedBy: null, + lockedUntil: null, + lastError: 'boom', + }, + }); + }); + + it('marks failed jobs terminal when max attempts is reached', async () => { + const { prisma, service } = makeService(); + const now = new Date('2026-07-08T10:00:00.000Z'); + prisma.backgroundJob.findUnique.mockResolvedValue({ id: 'job-1', attempts: 3, maxAttempts: 3 }); + + await service.markFailed('job-1', 'still broken', { now, retryDelayMs: 30_000 }); + + expect(prisma.backgroundJob.update).toHaveBeenCalledWith({ + where: { id: 'job-1' }, + data: { + status: 'failed', + availableAt: now, + lockedBy: null, + lockedUntil: null, + lastError: 'still broken', + }, + }); + }); +}); diff --git a/apps/server/src/modules/jobs/jobs.service.ts b/apps/server/src/modules/jobs/jobs.service.ts new file mode 100644 index 0000000..7c77522 --- /dev/null +++ b/apps/server/src/modules/jobs/jobs.service.ts @@ -0,0 +1,104 @@ +import { Inject, Injectable, NotFoundException } from '@nestjs/common'; +import { ACTIVE_JOB_STATUSES, BackgroundJobRecord, EnqueueJobInput, JOBS_PRISMA, MarkFailedOptions } from './jobs.types'; + +const DEFAULT_MAX_ATTEMPTS = 3; +const DEFAULT_RETRY_DELAY_MS = 60_000; +const MAX_ERROR_LENGTH = 2000; + +@Injectable() +export class JobsService { + constructor(@Inject(JOBS_PRISMA) private readonly prisma: any) {} + + async enqueue(input: EnqueueJobInput): Promise { + const type = input.type.trim(); + if (!type) throw new Error('Job type is required'); + const dedupeKey = input.dedupeKey?.trim() || null; + + if (dedupeKey) { + const existing = await this.findActiveDedupe(type, dedupeKey); + if (existing) return existing; + } + + try { + return await this.delegate.create({ + data: { + type, + dedupeKey, + payload: input.payload ?? {}, + maxAttempts: input.maxAttempts ?? DEFAULT_MAX_ATTEMPTS, + availableAt: input.availableAt, + }, + }); + } catch (error) { + if (dedupeKey && isUniqueConflict(error)) { + const existing = await this.findActiveDedupe(type, dedupeKey); + if (existing) return existing; + } + throw error; + } + } + + markSucceeded(id: string, now = new Date()): Promise { + return this.delegate.update({ + where: { id }, + data: { + status: 'succeeded', + lockedBy: null, + lockedUntil: null, + completedAt: now, + lastError: null, + }, + }); + } + + async markFailed(id: string, error: unknown, options: MarkFailedOptions = {}): Promise { + const job = await this.delegate.findUnique({ where: { id } }); + if (!job) throw new NotFoundException('Background job not found'); + + const now = options.now ?? new Date(); + const attempts = Number(job.attempts ?? 0); + const maxAttempts = Number(job.maxAttempts ?? DEFAULT_MAX_ATTEMPTS); + const hasRetriesLeft = attempts < maxAttempts; + const availableAt = hasRetriesLeft + ? new Date(now.getTime() + (options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS)) + : now; + + return this.delegate.update({ + where: { id }, + data: { + status: hasRetriesLeft ? 'queued' : 'failed', + availableAt, + lockedBy: null, + lockedUntil: null, + lastError: normalizeError(error), + }, + }); + } + + private get delegate(): any { + return (this.prisma as any).backgroundJob; + } + + private findActiveDedupe(type: string, dedupeKey: string): Promise { + return this.delegate.findFirst({ + where: { + type, + dedupeKey, + status: { in: [...ACTIVE_JOB_STATUSES] }, + }, + orderBy: { createdAt: 'asc' }, + }); + } +} + +function normalizeError(error: unknown): string { + const message = error instanceof Error ? error.message : String(error); + return message.slice(0, MAX_ERROR_LENGTH); +} + +function isUniqueConflict(error: unknown): boolean { + return typeof error === 'object' + && error !== null + && 'code' in error + && (error as { code?: unknown }).code === 'P2002'; +} diff --git a/apps/server/src/modules/jobs/jobs.types.ts b/apps/server/src/modules/jobs/jobs.types.ts new file mode 100644 index 0000000..203c70c --- /dev/null +++ b/apps/server/src/modules/jobs/jobs.types.ts @@ -0,0 +1,38 @@ +export const ACTIVE_JOB_STATUSES = ['queued', 'running'] as const; +export const JOBS_PRISMA = 'JOBS_PRISMA'; + +export type BackgroundJobStatus = 'queued' | 'running' | 'succeeded' | 'failed'; + +export interface BackgroundJobRecord { + id: string; + type: string; + status: BackgroundJobStatus; + dedupeKey?: string | null; + payload: unknown; + attempts: number; + maxAttempts: number; + availableAt?: Date; + lockedBy?: string | null; + lockedUntil?: Date | null; + lastError?: string | null; +} + +export interface EnqueueJobInput { + type: string; + dedupeKey?: string | null; + payload?: unknown; + maxAttempts?: number; + availableAt?: Date; +} + +export interface ClaimJobInput { + workerId: string; + types?: string[]; + now?: Date; + leaseMs?: number; +} + +export interface MarkFailedOptions { + now?: Date; + retryDelayMs?: number; +} diff --git a/docs/architecture.md b/docs/architecture.md index b787f6c..7250f48 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -135,6 +135,18 @@ DevTask 没有"已完成"状态，"已提测"就是终态——开发交付完 - V2.3：AppData 保存成功后触发关系表同步，让快读路径保持新鲜；同步失败只记日志，不阻塞用户保存。 - V2.4：领域 CRUD 成为主写入路径，AppData 写桥保留给历史数据和回滚兜底。不要恢复业务 localStorage 缓存，避免线上部署后出现多端数据分叉。 +## Background Job Runtime Layer (V2.6) + +V2.6 introduces a database-backed background job runtime for server-side refresh work that must not depend on a user opening a page. + +- Storage: `background_jobs` stores `type`, `payload`, `dedupe_key`, `status`, `attempts`, `max_attempts`, `available_at`, `locked_by`, `locked_until`, and `last_error`. +- Dedupe: active jobs (`queued` / `running`) are unique by `(type, dedupe_key)` when `dedupe_key` is present. Services still check first and recover from unique conflicts to stay idempotent under concurrent enqueue. +- Lease: `JobLockService.claimNext()` uses `FOR UPDATE SKIP LOCKED` and treats expired `running` rows as claimable, so a crashed worker can be recovered by a later worker. +- Retry: failed handlers are requeued while `attempts < max_attempts`; terminal failures keep `last_error` and move to `failed`. +- Worker boundary: `BackgroundJobWorker` is a small handler registry and single-job runner. Domain modules register typed handlers and only write through their own services. + +This runtime is intentionally DB-backed first. Redis is already available in deployment, but V2.6 jobs need transactional dedupe with domain writes more than high-throughput queue semantics. + ## 生产部署层（2026-07-01） 当前仓库已补齐云服务器生产部署基线： diff --git a/docs/decisions.md b/docs/decisions.md index c5133ab..5b5ed79 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -578,3 +578,17 @@ - 分区键进入每次领域写入，能维持 V2.2 分区表设计的查询边界。 - AppData fallback 让迁移可回滚、可兼容旧数据，但不再制造长期双事实源。 - RBAC/配置表会影响权限模型和管理流程，单独成阶段更安全；V2.4.5 只收口当前高频业务写入，避免为了“全收口”临时设计不稳的权限 schema。 + +## 45. V2.6 后台任务先采用 PostgreSQL Lease 队列 + +**问题**：小宝风险摘要、AI 解读和后续通知都需要在用户不打开页面时后台刷新。直接把这些逻辑放在页面 effect 中会导致无人访问时数据不更新；直接引入 Redis queue 又会增加一套可靠性、幂等和迁移运维面。 + +**决策**： +- 新增 `background_jobs` 表和 `JobsModule`，作为 V2.6 后台任务运行时。 +- Job 行包含 `type`、`payload`、`dedupe_key`、`status`、`attempts`、`max_attempts`、`available_at`、`locked_by`、`locked_until`、`last_error`。 +- 同一 `type + dedupe_key` 在 `queued/running` 状态下唯一；服务层先查 active job，遇到并发唯一冲突再回读，保证 enqueue 幂等。 +- Worker claim 使用数据库事务、`FOR UPDATE SKIP LOCKED` 和 lease 时间；`running` 且 `locked_until` 过期的 job 可以被新 worker 回收。 +- Handler 失败时按 `attempts < max_attempts` 重回 `queued` 并设置下一次 `available_at`；达到上限后进入 `failed`，只记录错误，不修改业务实体。 +- `BackgroundJobWorker` 只负责 handler 注册和单次执行，业务副作用仍放在各领域 service 内，避免队列层知道小宝、通知或审计细节。 + +**理由**：PostgreSQL 队列足够支撑 V2.6 的低频后台刷新，同时能和领域写入共享事务边界、唯一约束和迁移流程。等 V2.7 通知或更高吞吐任务落地后，如确实需要 Redis/专用队列，再通过同一 `JobsService` 接口替换底层实现，而不是现在提前引入第二套事实源。 diff --git a/docs/roadmap.md b/docs/roadmap.md index a03c8e8..6abb9f8 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -1,8 +1,8 @@ # 开发路线图 -## 当前阶段：V2.4 — 领域 CRUD 主写迁移完成 +## 当前阶段：V2.6 — 大数据性能增强 + 小宝后台化（进行中） -V2.4 将高增长和核心业务领域从“AppData 主写 + 关系表同步副本”推进到“领域 CRUD 主写关系表 + AppData 兼容/迁移兜底”。V2.2 快读 API 和 V2.3 AppData 写后同步继续保留，但它们现在是兼容基础设施，不再是已迁移领域的数据新鲜度主链路。 +V2.4 已完成高增长和核心业务领域从“AppData 主写 + 关系表同步副本”到“领域 CRUD 主写关系表 + AppData 兼容/迁移兜底”的迁移。V2.6 当前聚焦大数据 fixture、热查询预算、后台 job runtime，以及把小宝风险摘要和 AI 解读从页面触发迁到服务端后台。 ### 当前状态快照（2026-07-08） @@ -14,10 +14,16 @@ V2.4 将高增长和核心业务领域从“AppData 主写 + 关系表同步副 - 需求池已切到服务端分页、搜索、筛选、排序，不再要求加载全量 AppData 文档。 - `packages/shared` 状态契约已统一为当前业务状态机。 - V2.4.5 保守边界：成员身份写 `users`；部门、角色、密码规则、加班原因暂留 AppData 配置，等待后续 RBAC/配置表阶段。 +- V2.6.1 已新增 deterministic large-data fixture、HTTP performance harness 和性能预算文档。 +- V2.6.2 已新增 hot query explain/index audit 脚本、热查询索引迁移和 `docs/performance-hot-queries.md`。 +- V2.6.3 已新增 PostgreSQL-backed `background_jobs` 运行时、去重/lease/retry 语义和 jobs 单元测试。 ### 已完成（按时间倒序） **2026-07-08** +- V2.6.3 added DB-backed background jobs with active dedupe keys, lease-based claiming, expired lock recovery, retry/terminal-failure handling, and a small handler worker. +- V2.6.2 added `perf:explain`, hot query explain targets, index audit documentation, and V2.6 hot-path indexes for workspace, Xiaobao warning/dirty queues, project/version lists, and evidence scans. +- V2.6.1 added deterministic small/medium/large fixture generation, `perf:check`, and `docs/performance.md` for hot API p50/p95 budgets. - V2.4.0 completed shared domain status contract alignment for Requirement, VersionPlan, DevTask, TestCase, Bug, and Version. - V2.4.1 switched Product / Project / Version root mutations to domain APIs and left `products-overview` as compatibility fallback. - V2.4.2 switched Requirement writes to relation-table CRUD and added server-side pagination, search, filters, sorting, and cursor support for the requirement pool.