diff --git a/docs/architecture.md b/docs/architecture.md index cc454b9..1f0583a 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -52,8 +52,8 @@ flowchart TD ``` 1. HTTP 接收原始字节，`Pipeline` 验签后解析并归一化事件，检查允许的源仓与 delivery 去重。只有有效且匹配的事件进入业务监听；未知、重复、范围外或验签失败的投递仍保存各自请求和应用响应。存储失败返回 503，不确认已保存。 -2. Worker 的看板监听器仅暂存刷新意图。R2 正文／详情写入成功后，SQLite 事务一起提交索引、去重和刷新待办；随后由持久 Alarm 触发 `collect.yml`，HTTP 请求不会等待 Python 采集结束。 -3. `collect.yml` 从目标 main 读取上轮进度与历史，校验唯一源仓映射。GitHub 为采集 job 签发 OIDC 身份；对应 Worker 的 `/actions/token` 校验该身份后，用本环境 App 私钥分别申请源仓只读、看板仓 Contents 写安装令牌。采集器用这些令牌读取 GitHub API，再将新进度与变化数据一起提交到看板仓；结束时尝试撤销令牌。Bot 的 dispatch 参数不携带令牌，看板 Actions 不保存 App 私钥。 +2. Worker 的看板监听器暂存刷新意图，或合并 `workflow_run.completed` 的重跑 ID。R2 正文／详情写入成功后，SQLite 事务一起提交索引、去重和待办；随后仍由原有持久 Alarm 触发 `collect.yml`，重跑通知不提前调度，HTTP 请求不会等待 Python 采集结束。 +3. `collect.yml` 从目标 main 读取上轮进度与历史，校验唯一源仓映射。GitHub 为采集 job 签发 OIDC 身份；对应 Worker 的 `/actions/token` 校验该身份后，用本环境 App 私钥分别申请源仓只读、看板仓 Contents 写安装令牌。采集器用这些令牌读取 GitHub API，再将新进度与变化数据一起提交到看板仓；结束时尝试撤销令牌。它还以固定采集身份每轮领取最多 20 条重跑 ID，按 ID 补采而不为此扫描全部历史，提交成功后才按 revision 确认；失败项留在 Worker，进度与指标仍只由看板仓维护。Bot 的 dispatch 参数不携带令牌，看板 Actions 不保存 App 私钥。 4. `pages.yml` 只在 `site/` 变化或手动运行时上传整个 `site/`。仅进度变化不触发部署；页面无变化时不能把“没有新 Pages 运行”判成采集失败。 Webhook 已接受、Actions 已触发、采集提交成功、Pages 部署成功是四个独立状态。Bot 的 `last_dispatch` 只代表触发成功，最终结果分别查看采集与发布工作流。详情见[看板更新](features/board-publication.md)。 diff --git a/docs/features/actions-oidc.md b/docs/features/actions-oidc.md index fe7b02d..6790cde 100644 --- a/docs/features/actions-oidc.md +++ b/docs/features/actions-oidc.md @@ -66,6 +66,8 @@ Worker 保留 `SDBOT_GITHUB_APP_ID` 和 Secret `SDBOT_GITHUB_APP_PRIVATE_KEY`， `POST /actions/gitcode-sync` 使用同一身份校验（不需要请求体，也不受每次运行一次的限制），只返回本看板源仓的 GitCode 同步记录，不签发令牌、不进入 Webhook 档案；同步未启用时返回 `enabled: false`。见 [GitHub PR 同步到 GitCode](gitcode-sync.md)。 +`POST /actions/reruns` 与 `POST /actions/reruns/ack` 也不签发令牌、不进入投递档案，只允许既有看板配置中的固定采集身份：仓库与所有者 ID、`main`、`.github/workflows/collect.yml`、`schedule`／`workflow_dispatch`、固定 audience 和管理域名；其他访问规则不能借此查询待办，新增接口不扩大 OIDC 信任范围。读取请求为 `{}`，返回 `{"ok":true,"source":"owner/source","runs":[{"id":123,"attempt":2,"revision":"32位小写十六进制"}]}`，最多 20 条。确认请求为 `{"commit":"已提交的40位小写SHA","runs":[原样成功项]}`，正文上限 4096 字节；仅删除版本仍匹配的项。客户端必须在公开数据和进度提交成功后确认，并重新申请短期 OIDC 身份，避免长采集结束时原 JWT 已过期。接口不受令牌兑换次数限制，失败不清待办；响应均为 `no-store`。见 [CI 重跑通知](board-publication.md#ci-重跑通知)。 + 成功返回 `token`、`expires_at`、`repository`、`purpose`（请求没带时为 null）、`permissions`，响应 `Cache-Control: no-store`。认证失败为 401，身份超出信任范围为 403，重复兑换为 409，配置不完整为 503，上游签发失败为 502。发布新版本后的几分钟内，Worker 的后台对象可能还在运行旧代码，这时返回 503 且带 `Retry-After: 60` 和 `retryable: true`，稍后重跑即可。错误只返回短说明，不透传上游响应。该端点及 `/actions/` 保留路径不进入 Webhook 档案；它也不能查询管理信息。 ## 采集器与失败恢复 diff --git a/docs/features/board-publication.md b/docs/features/board-publication.md index 829ea49..2565544 100644 --- a/docs/features/board-publication.md +++ b/docs/features/board-publication.md @@ -38,6 +38,14 @@ App 安装到源仓及目标仓。Bot 触发时只申请目标仓 Metadata read Workers 使用持久 Alarm 和事务 outbox 调用同一个 Actions 入口，详见 [Workers 指南](workers.md)。不要同时运行两个 Bot 调度同站点。切换 Node 到 Actions 不会自动部署 Cloudflare Worker。 +## CI 重跑通知 + +Worker 的 `board.on_workflow_run` 监听已验签、范围内且未重复的 GitHub `workflow_run.completed`。仅当 `workflow_run.status=completed`、`run_attempt>1` 时登记重跑；第一次执行仍由现有增量列表发现，不为每次普通完成增加一条待办。不依赖重跑时不会触发的 `requested`。 + +待办只保存源仓、run ID、最新 attempt 和随机 revision，同一 run 合并，旧 attempt 不覆盖新值；与投递索引在归档成功后的同一 SQLite 事务内提交，重启不丢失。它不增加刷新代数、不重置合并窗口、不提前 Alarm，也不为每个完成事件 dispatch。通知随下一次原有 Issue／PR 刷新或空闲刷新一起处理，频率不变。Node 本地模式没有这条持久通知链路。 + +看板采集用现有 OIDC 身份调用 `POST /actions/reruns`，每轮最多读 20 条，轮转读取，失败项不会一直挡住后续 ID；只按 ID 查询最新 attempt、状态和必要报告，不为重跑翻全部 Actions 历史。看板将进度与公开数据原子提交后，才向 `POST /actions/reruns/ack` 确认成功项。删除必须同时匹配源仓、ID、attempt、revision，旧采集不能误删后来到达的新重跑。查询失败、预算耗尽、报告读取未完成、提交冲突或确认丢失均保留待办，下次沿用原频率再处理；不保存额外正文或产物。接口权限见 [Actions OIDC](actions-oidc.md)。 + ## 看板仓负责的工作 每次从 main 读取 `.sync/` 和 `site/data/history/`，在请求／时间预算内增量刷新、补历史、重试报告。进度与数据在同一提交内写入；使用准确 checkout 父提交，冲突则失败并从新 main 重试，禁止强推。看板仓自己的定时每天一次，只在 Bot 停止触发时兜底；无站点变化只更新进度，不触发 Pages。 @@ -49,3 +57,5 @@ Issue / PR 和 run / attempt 指标分片长期保留；不复制测试日志、 ## 验证 `npm run check`、`npm test` 覆盖 Webhook 完整存档、范围／去重、队列恢复、触发失败重试、App 最小权限及管理状态。`tests-ts/board-app.test.ts` 明确验证 dispatch 不伪造提交或发布成功。实际部署验收应核对 Actions 完成、进度与数据同一提交、Pages 完成及站点内容。 + +`tests-ts/worker-run-changes.test.mjs` 在真实 workerd 中验证完成通知合并、事务回滚、重启、轮转批次、OIDC 范围和带版本确认，并断言重跑通知不改变 outbox 与 Alarm。看板仓的 `test/journey-rerun-collection.py` 用隔离 HTTP 源和真实采集 CLI 验证旧 run 变绿、失败保留与发布冲突不确认。 diff --git a/docs/features/event-bus.md b/docs/features/event-bus.md index 50722a7..c416002 100644 --- a/docs/features/event-bus.md +++ b/docs/features/event-bus.md @@ -59,6 +59,7 @@ router.bus.subscribe({ - 内容分析：Issue opened／edited／reopened，Issue 评论，PR opened／synchronize／reopened／edited／ready_for_review、评审和合并；当前仅记录调用，不调用 LLM 或发评论。 - 看板更新：只有 Issue 新建、重新打开、关闭，PR 新建、重新打开、关闭（含合并，合并即关闭 PR）立即触发采集；重新打开按新建处理。转为可评审、评审、`workflow_run`、发布、job／check、push、分支与标签、评论和编辑都不触发，由每 8 小时一次的空闲刷新（`SDBOT_BOARD_REFRESH`）补上；CI、测试与覆盖率因此最多延迟约 8 小时，除非期间有 Issue／PR 新建、重新打开或关闭带动一次完整采集。 - 未配置发布时，看板监听处于占位状态；启用发布后显示 active，并列出实际来源范围。详见[看板更新](board-publication.md)。 +- Worker 另注册实际监听 `board.on_workflow_run`，只把范围内的 `workflow_run.completed` 重跑（attempt 大于 1）合并成待更新 run ID；不立即触发采集、不改变上述刷新频率。待原有刷新按 ID 补查并成功发布后确认，失败保留。Node 本地模式不注册此监听。 - GitCode 同步：PR opened／synchronize／reopened／closed／merged。监听器只把每个 PR 的最新期望状态写入持久队列，推送、MR 操作和 CodeCheck 读取由队列在 Webhook 之外完成。另一监听点只收同步目标仓的 GitCode MR 评论与标签变化，把对应 PR 的结论读取提前到现在；不轮询。第三个监听点收 GitHub 上对 `CodeCheck (GitCode)` 的重新运行（`check_run.rerequested`／`check_suite.rerequested`）：已同步的 head 立即重新读取结论，未同步成功的重新同步。没有 `GITCODE_TOKEN`、`SDBOT_GITCODE_SYNC_TARGET=off`、缺少 GitHub App 凭据或缺少 GitHub Webhook secret 时，以 `enabled: false` 注册，监听点页显示「已停用」并列出全部原因，不改变 PR 投递的 hooks，也不让 Worker 停止服务。详见 [GitHub PR 同步到 GitCode](gitcode-sync.md)。 ## 可观测性与验证 diff --git a/docs/testing.md b/docs/testing.md index cf50401..7967c40 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -51,6 +51,8 @@ npm run test:e2e:cloud-admin Actions 触发模式还需验证队列重启后恢复、失败不确认代数，以及 `last_dispatch` 不被误报成 `last_success` 或 Pages 提交。看板仓负责历史续跑与指标缓存的 Python、浏览器及真实 Actions 验证。 +`node --import tsx --test tests-ts/worker-run-changes.test.mjs` 是重跑通知的 API 旅程：真实 workerd 接收签名 Webhook，使用现场签发的 OIDC 身份读写待办，验证重复、重启、事务回滚、轮转及旧确认不删除新 attempt；不请求真实 Actions，也断言不提前刷新。看板仓另有 `PYTHONDONTWRITEBYTECODE=1 python3 test/journey-rerun-collection.py`，通过本地 HTTP 替身运行实际发布 CLI，验证窗口外重跑变绿、源端失败和发布冲突不确认。 + ## 提交前 云端没有测试实例：合入 `main` 即由 Workers Builds 发布正式 Worker `sciencediscovery-bot`，所以上面的本地检查、workerd 测试和浏览器旅程就是合入前的全部验证，不再有“先部署测试 Worker”这一步。 diff --git a/src/core/bus.ts b/src/core/bus.ts index 4fdd060..bbaf3eb 100644 --- a/src/core/bus.ts +++ b/src/core/bus.ts @@ -109,6 +109,9 @@ export function registerBuiltin(bus: EventBus, board: Board, sync: PullRequestSy ]; for (const [method, routes, exclude] of boards) bus.subscribe({ id: `board.${method}`, business: '看板更新', description: board.mode === 'active' ? '按源仓排队更新对应静态看板。' : '未启用静态发布，当前仅记录调用。', routes, exclude, mode: board.mode, providers: board.mode === 'active' ? ['github'] : [], repositories: board.repositories, handler: event => board.handle(method, event) }); + if (board.observeRun) bus.subscribe({ id: 'board.on_workflow_run', business: '看板更新', routes: ['workflow_run.completed'], + description: '仅合并记录重跑完成的 run ID，等待原有刷新节奏批量补采，不立即触发采集。', + providers: ['github'], repositories: board.repositories, mode: board.mode, enabled: board.mode === 'active', handler: event => board.observeRun!(event) }); // Without a token, with target `off`, or with missing GitHub credentials, the listener stays registered // but disabled, so PR routes and hooks are unchanged and the Worker keeps serving everything else. const active = sync.mode === 'active'; diff --git a/src/core/types.ts b/src/core/types.ts index eb9dc86..120253d 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -56,6 +56,8 @@ export interface Board { readonly mode: 'active' | 'noop'; readonly repositories: readonly string[]; handle(method: string, event: BotEvent): Promise; + /** Workers may retain changed run IDs without requesting an immediate refresh. */ + observeRun?(event: BotEvent): Promise; status(): Doc; } /** Stable public codes for why GitCode sync is not running; missing credentials can appear together. */ diff --git a/src/worker/actions-auth.ts b/src/worker/actions-auth.ts index f88d9fa..3fa765d 100644 --- a/src/worker/actions-auth.ts +++ b/src/worker/actions-auth.ts @@ -5,6 +5,7 @@ import { GitHubApp } from '../core/github-app.js'; import { object } from '../core/types.js'; import { PURPOSE, matchActionsRule, type ActionsClaims, type ActionsRule } from '../core/actions-rules.js'; import { OBJECT_NAME, workerConfig, type WorkerEnv } from './env.js'; +import { RUN_CHANGE_BATCH, validRunChange } from './run-changes.js'; import { describePermissions } from './token-grants.js'; const issuer = 'https://token.actions.githubusercontent.com'; @@ -120,3 +121,26 @@ export async function readGitCodeSync(request: Request, env: WorkerEnv): Promise try { return jsonResponse(await env.BOT.getByName(OBJECT_NAME).gitcodeSync()); } catch { return fail('sync records unavailable', 503); } } + +/** Collector-only rerun hints. Acknowledgement follows the dashboard's atomic publication. */ +export async function runChanges(request: Request, env: WorkerEnv, acknowledge = false): Promise { + const verified = await verifyCollector(request, env, false); + if (verified instanceof Response) return verified; + const captured = await captureBody(request, 4096); + if (captured.error) return fail(captured.error.message, captured.error.status); + let input: Record; + try { + const raw = JSON.parse(new TextDecoder('utf-8', { fatal: true, ignoreBOM: false }).decode(captured.body) || '{}'); + if (raw === null || typeof raw !== 'object' || Array.isArray(raw)) return fail('bad request', 400); + input = object(raw); + } catch { return fail('bad request', 400); } + const bot = env.BOT.getByName(OBJECT_NAME); + if (!acknowledge) { + if (Object.keys(input).length) return fail('bad request', 400); + return jsonResponse({ ok: true, source: verified.source, runs: await bot.runChanges(verified.source) }); + } + if (Object.keys(input).sort().join(',') !== 'commit,runs' || typeof input.commit !== 'string' || !/^[0-9a-f]{40}$/.test(input.commit) + || !Array.isArray(input.runs) || input.runs.length > RUN_CHANGE_BATCH || !input.runs.every(validRunChange)) return fail('bad request', 400); + await bot.ackRunChanges(verified.source, input.runs); + return jsonResponse({ ok: true }); +} diff --git a/src/worker/board.ts b/src/worker/board.ts index fb52382..bbb9455 100644 --- a/src/worker/board.ts +++ b/src/worker/board.ts @@ -1,4 +1,5 @@ import { boardBlocked, targets, type Config } from '../core/config.js'; +import { WorkerRunChanges } from './run-changes.js'; import { category, type Board, type BotEvent, type Doc } from '../core/types.js'; interface DispatchState { @@ -12,7 +13,9 @@ export class WorkerBoard implements Board { readonly repositories: string[]; readonly blocked: '' | 'no_github_app'; private readonly pending = new Set(); + readonly changes: WorkerRunChanges; constructor(readonly storage: DurableObjectStorage, readonly cfg: Config, readonly clock = () => Date.now()) { + this.changes = new WorkerRunChanges(storage); // Without App credentials nothing can be dispatched: report it instead of scheduling doomed work. this.blocked = boardBlocked(cfg, true); this.repositories = this.blocked ? [] : Object.keys(targets(cfg)).map(source => source.toLowerCase()); @@ -36,9 +39,14 @@ export class WorkerBoard implements Board { if (event.provider !== 'github' || !this.repositories.includes(event.repo.toLowerCase())) return { hook: 'board', method, status: this.mode === 'noop' ? 'noop' : 'ignored' }; this.pending.add(event.repo.toLowerCase()); return { hook: 'board', method, status: 'queued' }; } - clearPending(): void { this.pending.clear(); } + async observeRun(event: BotEvent): Promise { + if (event.provider !== 'github' || !this.repositories.includes(event.repo.toLowerCase())) return { status: 'ignored' }; + return this.changes.stage(event); + } + clearPending(): void { this.pending.clear(); this.changes.clearPending(); } commitPending(): void { for (const source of this.pending) this.request(this.get(source)!); + this.changes.commitPending(); this.pending.clear(); } private request(state: DispatchState): void { diff --git a/src/worker/index.ts b/src/worker/index.ts index a77e695..0f38593 100644 --- a/src/worker/index.ts +++ b/src/worker/index.ts @@ -8,10 +8,11 @@ import type { ActionsRule } from '../core/actions-rules.js'; import { dispatchCollection } from '../core/actions.js'; import { Mutex, object, type Doc } from '../core/types.js'; import { WorkerArchive, type PruneResult } from './archive.js'; +import type { RunChange } from './run-changes.js'; import { WorkerBoard } from './board.js'; import { OBJECT_NAME, workerConfig, type WorkerEnv } from './env.js'; import { authorizeAdmin } from './access.js'; -import { exchangeActionsToken, readGitCodeSync } from './actions-auth.js'; +import { exchangeActionsToken, readGitCodeSync, runChanges } from './actions-auth.js'; import { WorkerGitCodeSync } from './gitcode-sync.js'; import { AnalyticsCache } from './usage.js'; import { WorkerForwards, type AdminReply } from './forward.js'; @@ -133,6 +134,8 @@ export class BotObject extends DurableObject { tokenGrants(): Doc { return { ok: true, retention_days: GRANT_RETENTION_DAYS, limit: GRANT_LIMIT, shown: GRANT_PAGE, grants: this.grants.list() }; } // Read-only RPC for the OIDC-authenticated dashboard collector. gitcodeSync(): Promise> { return this.sync.snapshot(); } + runChanges(source: string): RunChange[] { return this.board.changes.list(source); } + ackRunChanges(source: string, rows: RunChange[]): void { this.board.changes.ack(source, rows); } async wake(): Promise { await this.mutex.run(() => this.schedule()); await this.forwards.schedule(); } // Cron-driven retention step; bounded per call and independent of the ingestion mutex. pruneArchive(now?: number): Promise { return this.archive.prune(now); } @@ -199,6 +202,7 @@ export default { const url = new URL(request.url); if (url.pathname === '/actions/token') return await exchangeActionsToken(request, env); if (url.pathname === '/actions/gitcode-sync') return await readGitCodeSync(request, env); + if (url.pathname === '/actions/reruns' || url.pathname === '/actions/reruns/ack') return await runChanges(request, env, url.pathname.endsWith('/ack')); if (url.pathname === '/actions' || url.pathname.startsWith('/actions/')) return jsonResponse({ ok: false, error: 'not found' }, 404); // Certificate callers authenticate with their own JWT; this path is never behind Access and never a webhook. if (url.pathname === '/caller' || url.pathname.startsWith('/caller/')) return await handleCaller(request, env); diff --git a/src/worker/run-changes.ts b/src/worker/run-changes.ts new file mode 100644 index 0000000..d34b5cb --- /dev/null +++ b/src/worker/run-changes.ts @@ -0,0 +1,66 @@ +/// +import { object, type BotEvent, type Doc } from '../core/types.js'; + +export const RUN_CHANGE_BATCH = 20; +export interface RunChange { id: number; attempt: number; revision: string } +const positive = (value: unknown): value is number => Number.isSafeInteger(value) && (value as number) > 0; +export const validRunChange = (value: unknown): value is RunChange => { + const row = object(value); + return Object.keys(row).length === 3 && positive(row.id) && positive(row.attempt) && row.attempt > 1 + && typeof row.revision === 'string' && /^[0-9a-f]{32}$/.test(row.revision); +}; + +/** Only unacknowledged reruns live here; normal completion never schedules another collection. */ +export class WorkerRunChanges { + private readonly staged = new Map(); + private ready = false; + constructor(readonly storage: DurableObjectStorage) {} + private sql(): SqlStorage { + if (!this.ready) { + this.storage.sql.exec(`CREATE TABLE IF NOT EXISTS board_run_changes ( + source TEXT NOT NULL, run_id INTEGER NOT NULL, attempt INTEGER NOT NULL, revision TEXT NOT NULL, + PRIMARY KEY(source, run_id)); + CREATE TABLE IF NOT EXISTS board_run_cursor (source TEXT PRIMARY KEY, last_id INTEGER NOT NULL);`); + this.ready = true; + } + return this.storage.sql; + } + stage(event: BotEvent): Doc { + const run = object(event.payload.workflow_run); + if (event.kind !== 'workflow_run' || event.action !== 'completed' || run.status !== 'completed' + || !positive(run.id) || !positive(run.run_attempt) || run.run_attempt <= 1) return { status: 'ignored' }; + this.sql(); // Create the schema before the archive's transaction can roll back. + const source = event.repo.toLowerCase(), key = `${source}:${run.id}`; + if (run.run_attempt > (this.staged.get(key)?.attempt || 0)) this.staged.set(key, { source, id: run.id, attempt: run.run_attempt }); + return { status: 'deferred' }; + } + clearPending(): void { this.staged.clear(); } + /** Called inside the archive's SQLite transaction, after its R2 objects were saved. */ + commitPending(): void { + for (const run of this.staged.values()) this.sql().exec(`INSERT INTO board_run_changes VALUES (?,?,?,?) + ON CONFLICT(source, run_id) DO UPDATE SET attempt=excluded.attempt, revision=excluded.revision + WHERE excluded.attempt > board_run_changes.attempt`, run.source, run.id, run.attempt, crypto.randomUUID().replaceAll('-', '')); + this.staged.clear(); + } + list(source: string): RunChange[] { + source = source.toLowerCase(); + const sql = this.sql(); + return this.storage.transactionSync(() => { + const cursor = sql.exec<{ last_id: number }>('SELECT last_id FROM board_run_cursor WHERE source=?', source).toArray()[0]?.last_id || 0; + const select = (op: '>' | '<=', limit: number) => sql.exec<{ id: number; attempt: number; revision: string }>( + `SELECT run_id AS id, attempt, revision FROM board_run_changes WHERE source=? AND run_id ${op} ? ORDER BY run_id LIMIT ?`, source, cursor, limit).toArray(); + const rows = select('>', RUN_CHANGE_BATCH); + if (rows.length < RUN_CHANGE_BATCH) rows.push(...select('<=', RUN_CHANGE_BATCH - rows.length)); + // Rotate between small batches: a permanently failing run must not starve newer changes. + if (rows.length) sql.exec('INSERT INTO board_run_cursor VALUES (?,?) ON CONFLICT(source) DO UPDATE SET last_id=excluded.last_id', source, rows.at(-1)!.id); + return rows; + }); + } + ack(source: string, rows: RunChange[]): void { + const sql = this.sql(); + this.storage.transactionSync(() => { + for (const row of rows) sql.exec('DELETE FROM board_run_changes WHERE source=? AND run_id=? AND attempt=? AND revision=?', + source.toLowerCase(), row.id, row.attempt, row.revision); + }); + } +} diff --git a/test/journey-cloud-admin.spec.cjs b/test/journey-cloud-admin.spec.cjs index c82d853..17d9ae3 100644 --- a/test/journey-cloud-admin.spec.cjs +++ b/test/journey-cloud-admin.spec.cjs @@ -28,7 +28,8 @@ test('authenticated cloud management shows exchanges and listeners without inges await page.screenshot({ path: testInfo.outputPath('cloud-details-mobile.png') }); await dialog.getByRole('button', { name: '关闭', exact: true }).click(); await page.getByRole('link', { name: '监听点', exact: true }).click(); - await expect(page.locator('#subscription-count')).toContainText('11 / 11'); + await expect(page.locator('#subscription-count')).toContainText('12 / 12'); + await expect(page.locator('#subscription-list')).toContainText('board.on_workflow_run'); await page.locator('#subscription-search').fill('does-not-match'); await expect(page.locator('#subscription-list')).toContainText('没有匹配的监听点'); await page.locator('#subscription-search').clear(); diff --git a/tests-ts/cloud-admin.test.mjs b/tests-ts/cloud-admin.test.mjs index ccc0d50..49fafaa 100644 --- a/tests-ts/cloud-admin.test.mjs +++ b/tests-ts/cloud-admin.test.mjs @@ -51,7 +51,7 @@ test('cloud management validates Access and is read-only on the public Worker', assert.equal(detail.request.body, body); assert.equal(detail.request.headers.cookie, '[REDACTED]'); assert.equal(detail.request.headers['cf-access-jwt-assertion'], '[REDACTED]'); assert.deepEqual(JSON.parse(detail.response.body), await response.json()); - assert.equal((await get('/admin/api/listeners')).listeners.length, 11); + assert.equal((await get('/admin/api/listeners')).listeners.length, 12); const before = await get('/admin/api/status'); assert.equal(before.environment, 'test'); assert.equal(before.runtime, 'cloudflare'); assert.equal(before.config.data_dir, undefined); assert.equal(before.config.webhook, undefined); diff --git a/tests-ts/support/worker-run-changes-control.mjs b/tests-ts/support/worker-run-changes-control.mjs new file mode 100644 index 0000000..6944bf1 --- /dev/null +++ b/tests-ts/support/worker-run-changes-control.mjs @@ -0,0 +1,28 @@ +// Test-only visibility into the existing outbox; never bundled for deployment. +import worker, { BotObject as Original } from '../../src/worker/index.ts'; +import { WorkerRunChanges } from '../../src/worker/run-changes.ts'; +import { OBJECT_NAME } from '../../src/worker/env.ts'; +export class BotObject extends Original { + async rerunState() { + return { outbox: this.ctx.storage.sql.exec('SELECT state FROM board_outbox ORDER BY source').toArray().map(row => JSON.parse(row.state)), + alarm: await this.ctx.storage.getAlarm() }; + } + rollbackRerun() { + const changes = new WorkerRunChanges(this.ctx.storage); + const event = { repo: 'example/source', kind: 'workflow_run', action: 'completed', + payload: { workflow_run: { id: 999001, run_attempt: 2, status: 'completed' } } }; + changes.stage(event); + const before = changes.list(event.repo); + try { this.ctx.storage.transactionSync(() => { changes.commitPending(); throw new Error('archive rollback'); }); } catch { /* expected */ } + changes.clearPending(); changes.commitPending(); + return { before, after: changes.list(event.repo) }; + } +} +export default { + async fetch(request, env, ctx) { + const stub = env.BOT.getByName(OBJECT_NAME), path = new URL(request.url).pathname; + if (path === '/_test/rerun-state') return Response.json(await stub.rerunState()); + if (path === '/_test/rerun-rollback') return Response.json(await stub.rollbackRerun()); + return worker.fetch(request, env, ctx); + }, scheduled: worker.scheduled, +}; diff --git a/tests-ts/worker-adapter.test.mjs b/tests-ts/worker-adapter.test.mjs index d5dc987..b465c81 100644 --- a/tests-ts/worker-adapter.test.mjs +++ b/tests-ts/worker-adapter.test.mjs @@ -78,7 +78,7 @@ test('production Worker: durable history, isolated admin, recovery and Actions o assert.equal((await admin.fetch('http://localhost/api/status', { headers: { ...auth, 'cf-ray': 'test' } })).status, 403); assert.equal((await admin.fetch('http://public.example/api/status', { headers: auth })).status, 403); assert.match(await (await admin.fetch('http://localhost/')).text(), /Webhook/); - assert.equal((await get('/api/listeners')).listeners.length, 11); + assert.equal((await get('/api/listeners')).listeners.length, 12); }); await t.test('store exact exchanges; concurrent duplicates request only one refresh', async () => { const options = signed(issue(sources[0]), delivery); diff --git a/tests-ts/worker-run-changes.test.mjs b/tests-ts/worker-run-changes.test.mjs new file mode 100644 index 0000000..514fbb7 --- /dev/null +++ b/tests-ts/worker-run-changes.test.mjs @@ -0,0 +1,99 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { createHmac, randomUUID } from 'node:crypto'; +import { mkdir, mkdtemp, rm } from 'node:fs/promises'; +import { resolve } from 'node:path'; +import { generateKeyPair, exportJWK, SignJWT } from 'jose'; +import { workerRuntime } from '../tools/worker-runtime.mjs'; + +const issuer = 'https://token.actions.githubusercontent.com', source = 'example/source', destination = 'example/board'; +const secret = 'isolated-rerun-webhook'; + +test('completed reruns coalesce durably without accelerating collection; acknowledgements are revision-bound', { timeout: 60000 }, async t => { + await mkdir('.tmp/tests-ts', { recursive: true }); + const directory = await mkdtemp(resolve('.tmp/tests-ts/run-changes-')); + const oidc = await generateKeyPair('RS256'); + const key = { ...await exportJWK(oidc.publicKey), kid: 'rerun-key', alg: 'RS256', use: 'sig' }; + const bindings = { SDBOT_GITHUB_WEBHOOK_SECRET: secret, SDBOT_GITHUB_APP_ID: '123', SDBOT_GITHUB_APP_PRIVATE_KEY: 'isolated-placeholder', + SDBOT_REPOS: source, SDBOT_BOARD_TARGETS: JSON.stringify({ [source]: destination }), SDBOT_ADMIN_HOSTNAME: 'localhost', + SDBOT_BOARD_DEBOUNCE: '600', SDBOT_ACTIONS_REPOSITORY_ID: '200', SDBOT_ACTIONS_OWNER_ID: '100', SDBOT_ACTIONS_AUDIENCE: 'reruns:test' }; + const start = () => workerRuntime({ directory, entry: 'tests-ts/support/worker-run-changes-control.mjs', bindings, + outboundService: request => { + assert.equal(request.url, issuer + '/.well-known/jwks', 'reruns must not dispatch new collections or mint tokens'); + return Response.json({ keys: [key] }); + } }); + let mf = await start(); + t.after(async () => { await mf.dispose(); await rm(directory, { recursive: true, force: true }); }); + const token = (extra = {}) => { + const now = Math.floor(Date.now() / 1000); + return new SignJWT({ iss: issuer, aud: 'reruns:test', sub: `repo:${destination}:ref:refs/heads/main`, iat: now, nbf: now, exp: now + 300, + jti: randomUUID(), repository: destination, repository_id: '200', repository_owner_id: '100', ref: 'refs/heads/main', ref_type: 'branch', + workflow_ref: `${destination}/.github/workflows/collect.yml@refs/heads/main`, event_name: 'workflow_dispatch', run_id: '400', run_attempt: '1', ...extra }) + .setProtectedHeader({ alg: 'RS256', kid: key.kid }).sign(oidc.privateKey); + }; + const jwt = await token(); + const api = (path = '', body = {}, authorization = jwt, method = 'POST') => mf.dispatchFetch('http://localhost/actions/reruns' + path, + { method, headers: { authorization: 'Bearer ' + authorization, 'content-type': 'application/json' }, ...(method === 'GET' ? {} : { body: JSON.stringify(body) }) }); + const list = async () => { const r = await api(); assert.equal(r.status, 200); const value = await r.json(); assert.equal(value.source, source); return value.runs; }; + const state = async () => (await mf.dispatchFetch('http://localhost/_test/rerun-state')).json(); + const send = async (id, attempt, extra = {}, delivery = randomUUID(), signing = secret, repository = source) => { + const body = JSON.stringify({ action: 'completed', repository: { full_name: repository }, workflow_run: { id, run_attempt: attempt, status: 'completed', ...extra } }); + return mf.dispatchFetch('http://localhost/webhook/github', { method: 'POST', body, headers: { 'x-github-event': 'workflow_run', + 'x-github-delivery': delivery, 'x-hub-signature-256': 'sha256=' + createHmac('sha256', signing).update(body).digest('hex') } }); + }; + await t.test('only the fixed collector can read or acknowledge this source', async () => { + assert.equal((await api('', {}, 'invented')).status, 401); + assert.equal((await api('', {}, jwt, 'GET')).status, 405); + for (const claims of [{ repository: 'other/board' }, { repository_id: '201' }, { repository_owner_id: '101' }, { ref: 'refs/heads/develop' }, + { event_name: 'push' }, { workflow_ref: `${destination}/.github/workflows/evil.yml@refs/heads/main` }, + { job_workflow_ref: `other/reusable/.github/workflows/a.yml@refs/heads/main` }]) + assert.equal((await api('', {}, await token(claims))).status, 403); + assert.equal((await api('', { source: 'other/source' })).status, 400); + assert.equal((await api('/ack', { runs: [], commit: 'bad' })).status, 400); + assert.deepEqual(await list(), []); + }); + await t.test('first attempts, invalid input and untrusted deliveries create no hints', async () => { + await send(1, 1); await send(2, 2, { status: 'in_progress' }); await send('3', 2); await send(3, true); + await send(4, 2, {}, randomUUID(), 'wrong-secret'); + await send(5, 2, {}, randomUUID(), secret, 'other/source'); + assert.deepEqual(await list(), []); + }); + await t.test('reruns change neither the eight-hour schedule nor a pending Issue refresh', async () => { + const before = await state(), delivery = randomUUID(); + assert.equal((await send(42, 2, {}, delivery)).status, 200); + await send(42, 2, {}, delivery); await send(42, 2); await send(42, 1); + const rows = await list(); assert.equal(rows.length, 1); assert.equal(rows[0].attempt, 2); + assert.deepEqual(await state(), before); + const body = JSON.stringify({ action: 'opened', repository: { full_name: source }, issue: { number: 9 } }); + await mf.dispatchFetch('http://localhost/webhook/github', { method: 'POST', body, headers: { 'x-github-event': 'issues', 'x-github-delivery': randomUUID(), + 'x-hub-signature-256': 'sha256=' + createHmac('sha256', secret).update(body).digest('hex') } }); + const issueState = await state(); assert.equal(issueState.outbox[0].requested, 1); + await send(42, 3); assert.deepEqual(await state(), issueState, 'completion does not restart debounce or add a generation'); + const current = (await list())[0]; + assert.notEqual(current.revision, rows[0].revision); + assert.equal((await api('/ack', { commit: 'a'.repeat(40), runs: rows })).status, 200); + assert.deepEqual(await list(), [current], 'a late acknowledgement cannot remove a newer attempt'); + await api('/ack', { commit: 'a'.repeat(40), runs: [{ ...current, revision: '0'.repeat(32) }] }); + assert.deepEqual(await list(), [current]); + }); + await t.test('pending hints survive restart; only a matching acknowledgement removes them', async () => { + const rows = await list(); + await mf.dispose(); mf = await start(); + assert.deepEqual(await list(), rows); + assert.equal((await api('/ack', { commit: 'b'.repeat(40), runs: rows })).status, 200); + assert.deepEqual(await list(), []); + }); + await t.test('small batches rotate fairly even when failed items remain pending', async () => { + for (let id = 100; id < 133; id++) await send(id, 2); + const first = await list(), second = await list(); + assert.equal(first.length, 20); assert.equal(second.length, 20); + assert.equal(new Set([...first, ...second].map(row => row.id)).size, 33); + assert.equal((await api('/ack', { commit: 'a'.repeat(40), runs: Array(21).fill(first[0]) })).status, 400); + assert.equal((await api('/ack', { commit: 'a'.repeat(40), runs: [{ ...first[0], id: '../evil' }] })).status, 400); + }); + await t.test('staged changes are not visible until the archive transaction commits', async () => { + const result = await (await mf.dispatchFetch('http://localhost/_test/rerun-rollback')).json(); + assert.ok(!result.before.some(row => row.id === 999001)); + assert.ok(!result.after.some(row => row.id === 999001)); + }); +});