diff --git a/config/platform-infra/gitea.yaml b/config/platform-infra/gitea.yaml index 9f129ce0..3ecbef31 100644 --- a/config/platform-infra/gitea.yaml +++ b/config/platform-infra/gitea.yaml @@ -96,6 +96,29 @@ sourceAuthority: sourceRef: gitea.github-webhook.env sourceKey: GITHUB_WEBHOOK_SECRET createIfMissing: true + hookReconcile: + enabled: true + # 只有 owning target 写 GitHub hook topology；其他 target 仅接收与观察。 + ownerTargetId: NC01 + intervalMs: 60000 + requestTimeoutMs: 15000 + listMaxPages: 10 + failedDeliveryRecovery: + enabled: true + # 恢复只处理 owner target 的 exact hook；其他 target 必须由其本地 authority 收口。 + targetId: NC01 + statePath: /var/lib/gitea-github-sync/hook-recovery/ledger.json + # GitHub REST redelivery 只支持最近 3 天。 + lookbackSeconds: 259200 + historyPerHook: 50 + maxRedeliveriesPerCycle: 4 + cooldownMs: 300000 + maxAttemptsPerDelivery: 3 + retryStatusCodes: + - 502 + - 503 + - 504 + stateRetentionSeconds: 1209600 bridge: deploymentName: gitea-github-sync serviceName: gitea-github-sync diff --git a/scripts/src/platform-infra-gitea-config.ts b/scripts/src/platform-infra-gitea-config.ts index 219436b0..c3e6436a 100644 --- a/scripts/src/platform-infra-gitea-config.ts +++ b/scripts/src/platform-infra-gitea-config.ts @@ -172,6 +172,25 @@ export interface GiteaWebhookSync { sourceKey: string; createIfMissing: boolean; }; + hookReconcile: { + enabled: boolean; + ownerTargetId: string; + intervalMs: number; + requestTimeoutMs: number; + listMaxPages: number; + failedDeliveryRecovery: { + enabled: boolean; + targetId: string; + statePath: string; + lookbackSeconds: number; + historyPerHook: number; + maxRedeliveriesPerCycle: number; + cooldownMs: number; + maxAttemptsPerDelivery: number; + retryStatusCodes: number[]; + stateRetentionSeconds: number; + }; + }; bridge: { deploymentName: string; serviceName: string; @@ -451,6 +470,8 @@ function parseGithubCredential(record: Record, path: string): G function parseWebhookSync(record: Record, path: string): GiteaWebhookSync { const ingressRetry = y.objectField(record, "ingressRetry", path); const secret = y.objectField(record, "secret", path); + const hookReconcile = y.objectField(record, "hookReconcile", path); + const failedDeliveryRecovery = y.objectField(hookReconcile, "failedDeliveryRecovery", `${path}.hookReconcile`); const bridge = y.objectField(record, "bridge", path); const candidateGate = y.objectField(bridge, "candidateGate", `${path}.bridge`); const inbox = y.objectField(bridge, "inbox", `${path}.bridge`); @@ -476,6 +497,25 @@ function parseWebhookSync(record: Record, path: string): GiteaW sourceKey: y.envKeyField(secret, "sourceKey", `${path}.secret`), createIfMissing: y.booleanField(secret, "createIfMissing", `${path}.secret`), }, + hookReconcile: { + enabled: y.booleanField(hookReconcile, "enabled", `${path}.hookReconcile`), + ownerTargetId: y.stringField(hookReconcile, "ownerTargetId", `${path}.hookReconcile`), + intervalMs: positiveInteger(hookReconcile, "intervalMs", `${path}.hookReconcile`), + requestTimeoutMs: positiveInteger(hookReconcile, "requestTimeoutMs", `${path}.hookReconcile`), + listMaxPages: positiveInteger(hookReconcile, "listMaxPages", `${path}.hookReconcile`), + failedDeliveryRecovery: { + enabled: y.booleanField(failedDeliveryRecovery, "enabled", `${path}.hookReconcile.failedDeliveryRecovery`), + targetId: y.stringField(failedDeliveryRecovery, "targetId", `${path}.hookReconcile.failedDeliveryRecovery`), + statePath: y.absolutePathField(failedDeliveryRecovery, "statePath", `${path}.hookReconcile.failedDeliveryRecovery`), + lookbackSeconds: positiveInteger(failedDeliveryRecovery, "lookbackSeconds", `${path}.hookReconcile.failedDeliveryRecovery`), + historyPerHook: positiveInteger(failedDeliveryRecovery, "historyPerHook", `${path}.hookReconcile.failedDeliveryRecovery`), + maxRedeliveriesPerCycle: positiveInteger(failedDeliveryRecovery, "maxRedeliveriesPerCycle", `${path}.hookReconcile.failedDeliveryRecovery`), + cooldownMs: positiveInteger(failedDeliveryRecovery, "cooldownMs", `${path}.hookReconcile.failedDeliveryRecovery`), + maxAttemptsPerDelivery: positiveInteger(failedDeliveryRecovery, "maxAttemptsPerDelivery", `${path}.hookReconcile.failedDeliveryRecovery`), + retryStatusCodes: y.numberArrayField(failedDeliveryRecovery, "retryStatusCodes", `${path}.hookReconcile.failedDeliveryRecovery`), + stateRetentionSeconds: positiveInteger(failedDeliveryRecovery, "stateRetentionSeconds", `${path}.hookReconcile.failedDeliveryRecovery`), + }, + }, bridge: { deploymentName: y.kubernetesNameField(bridge, "deploymentName", `${path}.bridge`), serviceName: y.kubernetesNameField(bridge, "serviceName", `${path}.bridge`), @@ -708,9 +748,34 @@ function validateConfig(gitea: GiteaConfig): void { if (!events.has("push")) throw new Error(`${configLabel}.sourceAuthority.webhookSync.events must include push for GitHub -> Gitea source sync`); if (gitea.sourceAuthority.webhookSync.frpc.remotePort === gitea.app.publicExposure.frpc.remotePort) throw new Error(`${configLabel}.sourceAuthority.webhookSync.frpc.remotePort must differ from app.publicExposure.frpc.remotePort`); const ingressRetry = gitea.sourceAuthority.webhookSync.ingressRetry; + const hookReconcile = gitea.sourceAuthority.webhookSync.hookReconcile; const bridge = gitea.sourceAuthority.webhookSync.bridge; if (bridge.replicas !== 1) throw new Error(`${configLabel}.sourceAuthority.webhookSync.bridge.replicas must be 1 because the durable inbox uses a single-writer Recreate deployment`); if (bridge.retry.terminalRetryDelayMs < bridge.retry.maxDelayMs) throw new Error(`${configLabel}.sourceAuthority.webhookSync.bridge.retry.terminalRetryDelayMs must be >= maxDelayMs`); + if (hookReconcile.enabled) { + const owner = resolveTarget(gitea, hookReconcile.ownerTargetId); + if (!gitea.sourceAuthority.repositories.some((repo) => repo.targetId.toLowerCase() === owner.id.toLowerCase())) { + throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.ownerTargetId must own at least one repository`); + } + if (hookReconcile.intervalMs < 30_000) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.intervalMs must be >= 30000`); + if (hookReconcile.requestTimeoutMs > 30_000) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.requestTimeoutMs must be <= 30000`); + if (hookReconcile.listMaxPages > 20) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.listMaxPages must be <= 20`); + const recovery = hookReconcile.failedDeliveryRecovery; + if (recovery.enabled) { + if (recovery.targetId.toLowerCase() !== owner.id.toLowerCase()) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.targetId must equal the owning target`); + const recoveryRoot = `${bridge.inbox.path.replace(/\/+$/u, "")}/hook-recovery/`; + if (!recovery.statePath.startsWith(recoveryRoot)) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.statePath must be under ${recoveryRoot}`); + if (recovery.historyPerHook > 100) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.historyPerHook must be <= 100`); + if (recovery.lookbackSeconds > 259_200) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.lookbackSeconds must be <= 259200 because GitHub only supports redelivery for three days`); + if (recovery.maxRedeliveriesPerCycle > 20) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.maxRedeliveriesPerCycle must be <= 20`); + if (recovery.maxAttemptsPerDelivery > 10) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.maxAttemptsPerDelivery must be <= 10`); + if (recovery.cooldownMs < hookReconcile.intervalMs) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.cooldownMs must be >= hookReconcile.intervalMs`); + if (recovery.stateRetentionSeconds < recovery.lookbackSeconds) throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.stateRetentionSeconds must be >= lookbackSeconds`); + if (recovery.retryStatusCodes.length < 1 || recovery.retryStatusCodes.some((statusCode) => ![502, 503, 504].includes(statusCode))) { + throw new Error(`${configLabel}.sourceAuthority.webhookSync.hookReconcile.failedDeliveryRecovery.retryStatusCodes may only contain 502, 503 and 504`); + } + } + } if (ingressRetry.retryStatusCodes.length < 1 || ingressRetry.retryStatusCodes.some((statusCode) => ![502, 503, 504].includes(statusCode))) { throw new Error(`${configLabel}.sourceAuthority.webhookSync.ingressRetry.retryStatusCodes may only contain 502, 503 and 504`); } diff --git a/scripts/src/platform-infra-gitea-disclosure.test.ts b/scripts/src/platform-infra-gitea-disclosure.test.ts index 72b55b69..b2da2ab5 100644 --- a/scripts/src/platform-infra-gitea-disclosure.test.ts +++ b/scripts/src/platform-infra-gitea-disclosure.test.ts @@ -1,5 +1,6 @@ import { describe, expect, test } from "bun:test"; import { buildMirrorWebhookStatusDisclosure, rawMirrorWebhookStatus } from "./platform-infra-gitea-disclosure"; +import { renderMirrorWebhookStatus } from "./platform-infra-gitea-render"; describe("Gitea webhook status progressive disclosure", () => { test("keeps --full as a bounded domain aggregate with repository drill-down commands", () => { @@ -24,6 +25,10 @@ describe("Gitea webhook status progressive disclosure", () => { expect((repository.deliveries as unknown[]).length).toBe(1); expect((repository.selection as { effectiveDeliveryId: string }).effectiveDeliveryId).toBe("delivery-current-a"); expect((repository.retry as { totalAttempts: number }).totalAttempts).toBe(1); + const recovery = (repository.bridge as { durableInbox: { failedDeliveryRecovery: { entries: Array> } } }).durableInbox.failedDeliveryRecovery; + expect(recovery.entries).toHaveLength(1); + expect(recovery.entries[0]).toMatchObject({ deliveryId: "delivery-current-a", state: "attempted", attempts: 1, cooldownMs: 300_000, nextAt: "2026-07-11T00:05:00Z" }); + expect(JSON.stringify(recovery)).not.toContain("https://"); const delivery = buildMirrorWebhookStatusDisclosure(fixture(), { repoKey: null, deliveryId: "delivery-5" }); expect(delivery.ok).toBe(true); @@ -47,6 +52,23 @@ describe("Gitea webhook status progressive disclosure", () => { expect((raw.safety as { explicitRawRequired: boolean; valuesPrinted: boolean }).explicitRawRequired).toBe(true); expect((raw.safety as { valuesPrinted: boolean }).valuesPrinted).toBe(false); }); + + test("keeps selected delivery GUID, status and accepted in one observation", () => { + const input = fixture(); + const repo = (input.repositories as Array>)[0]!; + repo.latestPushDelivery = { deliveryId: "newer-unrelated", statusCode: 202, status: "OK" }; + repo.selectedPushDelivery = { deliveryId: "selected-failed", hookId: 22, statusCode: 502, status: "misconfigured", deliveredAt: "2026-07-12T05:54:18Z" }; + repo.deliveryId = "selected-failed"; + repo.deliveryAccepted = false; + + const disclosure = buildMirrorWebhookStatusDisclosure(input, { repoKey: null, deliveryId: null }); + const summary = (disclosure.repositories as Array<{ selectedDelivery: Record }>)[0]!.selectedDelivery; + expect(summary).toMatchObject({ deliveryId: "selected-failed", hookId: 22, statusCode: 502, status: "misconfigured", accepted: false }); + + const rendered = renderMirrorWebhookStatus(input, disclosure).renderedText; + expect(rendered).toContain("selected-failed:502:misconfigured"); + expect(rendered).not.toContain("selected-failed:202:OK"); + }); }); function fixture(): Record { @@ -71,6 +93,16 @@ function fixture(): Record { maxBytes: 1_048_576, errorType: null, journal: { version: 1, kind: "GiteaGithubSyncDurableInbox", createdAt: "2026-07-11T00:00:00Z" }, + failedDeliveryRecovery: { + enabled: true, + ready: true, + state: "ready", + counts: { pending: 0, attempted: 2, "delivery-succeeded-awaiting-source": 0, "source-committed": 0, "terminal-exhausted": 0 }, + entries: [ + { repository: "pikasTech/repo-a", hookId: 1, deliveryId: "delivery-current-a", state: "attempted", attempts: 1, cooldownMs: 300_000, nextAt: "2026-07-11T00:05:00Z", updatedAt: "2026-07-11T00:00:00Z", lastStatusCode: 502 }, + { repository: "pikasTech/repo-b", hookId: 2, deliveryId: "delivery-current-b", state: "attempted", attempts: 2, cooldownMs: 300_000, nextAt: "2026-07-11T00:05:00Z", updatedAt: "2026-07-11T00:00:00Z", lastStatusCode: 504 }, + ], + }, }, }; return { @@ -87,7 +119,8 @@ function fixture(): Record { ingressAttemptBudgetMs: 2_000, ingressResponseHeaderTimeoutMs: 3_000, ingressRetry: { enabled: true, maxAttempts: 3, initialDelayMs: 1_000, maxDelayMs: 30_000 }, - bridge: { deploymentName: "gitea-github-sync", serviceName: "gitea-github-sync" }, + hookReconcile: { enabled: true, ownerTargetId: "NC01", runningOnTarget: true, intervalMs: 60_000, requestTimeoutMs: 15_000, desiredUrlSha256: "sha256:desired", topologySha256: "sha256:topology" }, + bridge: { deploymentName: "gitea-github-sync", serviceName: "gitea-github-sync", retry: { maxAttempts: 3, initialDelayMs: 1_000, maxDelayMs: 30_000 } }, }, bridge, repositories: [repository("repo-a", currentA), repository("repo-b", currentB)], @@ -118,6 +151,15 @@ function repository(key: string, delivery: Record): Record"); const webhook = record(result.webhook); const webhookRetry = record(webhook.ingressRetry); + const webhookHookReconcile = record(webhook.hookReconcile); const webhookBridge = record(webhook.bridge); const bridge = record(result.bridge); const durableInbox = record(bridge.durableInbox); + const failedDeliveryRecovery = record(durableInbox.failedDeliveryRecovery); + const allRecoveryEntries = arrayRecords(failedDeliveryRecovery.entries); const allRepositories = arrayRecords(result.repositories); const inboxDeliveries = arrayRecords(durableInbox.deliveries); const bridgeLogs = record(result.bridgeLogs); @@ -69,6 +72,12 @@ export function buildMirrorWebhookStatusDisclosure( const repositories = selectedRepoKey === null ? allRepositories : allRepositories.filter((item) => stringValue(item.key, "") === selectedRepoKey); + const selectedRepositoryName = selectedRepository === undefined ? null : stringValue(selectedRepository.repository, "") || null; + const recoveryEntries = allRecoveryEntries.filter((item) => { + if (selectedRepositoryName !== null && stringValue(item.repository, "") !== selectedRepositoryName) return false; + if (effectiveDeliveryId !== null && stringValue(item.deliveryId, "") !== effectiveDeliveryId) return false; + return true; + }); const matchingLogEvents = allLogEvents.filter((item) => { if (selectedRepoKey !== null && stringValue(item.repo, "") !== selectedRepoKey) return false; if (effectiveDeliveryId !== null && stringValue(item.deliveryId, "") !== effectiveDeliveryId) return false; @@ -135,7 +144,7 @@ export function buildMirrorWebhookStatusDisclosure( webhook: { enabled: webhook.enabled === true, direction: webhook.direction, - publicUrl: webhook.publicUrl, + publicUrlSha256: webhookHookReconcile.desiredUrlSha256, events: webhook.events, responseBudgetMs: webhook.responseBudgetMs, ingressAttemptBudgetMs: webhook.ingressAttemptBudgetMs, @@ -150,6 +159,14 @@ export function buildMirrorWebhookStatusDisclosure( deploymentName: webhookBridge.deploymentName, serviceName: webhookBridge.serviceName, }, + hookReconcile: { + enabled: webhookHookReconcile.enabled === true, + ownerTargetId: webhookHookReconcile.ownerTargetId, + runningOnTarget: webhookHookReconcile.runningOnTarget === true, + intervalMs: webhookHookReconcile.intervalMs, + requestTimeoutMs: webhookHookReconcile.requestTimeoutMs, + topologySha256: webhookHookReconcile.topologySha256, + }, }, bridge: { deployment: bridge.deployment, @@ -166,6 +183,13 @@ export function buildMirrorWebhookStatusDisclosure( maxBytes: durableInbox.maxBytes, errorType: durableInbox.errorType ?? null, journal: durableInbox.journal, + failedDeliveryRecovery: { + enabled: failedDeliveryRecovery.enabled === true, + ready: failedDeliveryRecovery.ready === true, + state: failedDeliveryRecovery.state, + counts: failedDeliveryRecovery.counts, + entries: recoveryEntries.map(failedDeliveryRecoveryEntrySummary), + }, }, }, repositories: repositories.map(webhookRepositorySummary), @@ -206,7 +230,24 @@ export function buildMirrorWebhookStatusDisclosure( }; } +function failedDeliveryRecoveryEntrySummary(item: Record): Record { + return { + repository: item.repository, + hookId: item.hookId, + deliveryId: item.deliveryId, + state: item.state, + attempts: item.attempts, + cooldownMs: item.cooldownMs, + nextAt: item.nextAt ?? null, + updatedAt: item.updatedAt, + lastStatusCode: item.lastStatusCode ?? null, + completionReason: item.completionReason ?? null, + }; +} + function webhookRepositorySummary(repo: Record): Record { + const selected = record(repo.selectedPushDelivery); + const topology = record(repo.hookTopology); return { key: repo.key, repository: repo.repository, @@ -218,8 +259,23 @@ function webhookRepositorySummary(repo: Record): Record item && typeof item === "object" + && allowedStates.has(item.state) + && typeof item.repository === "string" && item.repository.length > 0 + && (typeof item.hookId === "number" || typeof item.hookId === "string") + && typeof item.deliveryId === "string" && item.deliveryId.length > 0 + && Number.isInteger(item.attempts) && item.attempts >= 0 + && Number.isInteger(item.cooldownMs) && item.cooldownMs >= 0 + && typeof item.updatedAt === "string" && item.updatedAt.length > 0; + if (records.some((item) => !validRecord(item))) { + return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-record-invalid", counts: {}, entries: [] }; + } + const counts = { pending: 0, attempted: 0, "delivery-succeeded-awaiting-source": 0, "source-committed": 0, "terminal-exhausted": 0 }; + for (const item of records) counts[item.state] += 1; + const entries = records + .sort((left, right) => String(right.updatedAt || "").localeCompare(String(left.updatedAt || ""))) + .slice(0, 50) + .map((item) => ({ + repository: item.repository, + hookId: item.hookId, + deliveryId: item.deliveryId, + state: item.state, + attempts: item.attempts, + cooldownMs: item.cooldownMs, + nextAt: item.nextAt ?? null, + updatedAt: item.updatedAt, + lastStatusCode: item.lastStatusCode ?? null, + completionReason: item.completionReason ?? null, + })); + return { enabled: true, ready: true, state: "ready", counts, entries }; +} + export function startServer(config = loadRuntimeConfig()) { const controller = config.controller ?? createDurableSyncController(config); const server = createServer(createGithubSyncHandler({ ...config, controller })); diff --git a/scripts/src/platform-infra-gitea-github-sync-server.test.mjs b/scripts/src/platform-infra-gitea-github-sync-server.test.mjs index 65dca783..c2f0a761 100644 --- a/scripts/src/platform-infra-gitea-github-sync-server.test.mjs +++ b/scripts/src/platform-infra-gitea-github-sync-server.test.mjs @@ -11,6 +11,7 @@ import { createDurableInboxStore, createDurableSyncController, createSyncCoordinator, + readFailedDeliveryRecoveryStatus, run, runResult, syncRepository, @@ -18,6 +19,45 @@ import { const zeroDelay = async () => {}; +test("primary status exposes a bounded redacted failed-delivery recovery ledger", async () => { + const root = mkdtempSync(join(tmpdir(), "gitea-hook-recovery-status-")); + const statePath = join(root, "ledger.json"); + try { + const records = Object.fromEntries(Array.from({ length: 55 }, (_, index) => [`record-${index}`, { + repository: "pikasTech/agentrun", + hookId: 77, + deliveryId: `guid-${index}`, + apiDeliveryId: String(1000 + index), + state: index % 2 === 0 ? "attempted" : "pending", + attempts: index % 3, + cooldownMs: 300_000, + nextAt: "2026-07-12T07:11:40Z", + updatedAt: `2026-07-12T07:${String(index).padStart(2, "0")}:00Z`, + lastStatusCode: 502, + secret: "must-not-leak", + url: "https://must-not-leak.invalid/hook", + }])); + writeFileSync(statePath, JSON.stringify({ version: 1, kind: "GiteaGithubHookRecoveryLedger", records })); + const status = await readFailedDeliveryRecoveryStatus(statePath); + assert.equal(status.ready, true); + assert.equal(status.entries.length, 50); + assert.equal(status.counts.attempted + status.counts.pending, 55); + assert.equal(JSON.stringify(status).includes("must-not-leak"), false); + writeFileSync(statePath, JSON.stringify({ + version: 1, + kind: "GiteaGithubHookRecoveryLedger", + records: { corrupt: { ...records["record-0"], state: "unknown-hidden-state" } }, + })); + const invalid = await readFailedDeliveryRecoveryStatus(statePath); + assert.equal(invalid.ready, false); + assert.equal(invalid.state, "invalid-or-unreadable"); + assert.equal(invalid.errorType, "hook-recovery-ledger-record-invalid"); + assert.deepEqual(invalid.entries, []); + } finally { + rmSync(root, { recursive: true, force: true }); + } +}); + test("explicit entrypoint starts through Kubernetes ConfigMap symlinks", () => { const root = mkdtempSync(join(tmpdir(), "gitea-sync-main-")); try { diff --git a/scripts/src/platform-infra-gitea-gitops-delivery.test.ts b/scripts/src/platform-infra-gitea-gitops-delivery.test.ts index 8cd2e725..0d58a6bc 100644 --- a/scripts/src/platform-infra-gitea-gitops-delivery.test.ts +++ b/scripts/src/platform-infra-gitea-gitops-delivery.test.ts @@ -1,4 +1,5 @@ import { expect, test } from "bun:test"; +import { spawnSync } from "node:child_process"; import { readFileSync } from "node:fs"; import { resolve } from "node:path"; import { readGiteaWebhookGitOpsDelivery, renderGiteaWebhookSyncDesiredManifest } from "./platform-infra-gitea"; @@ -83,4 +84,31 @@ test("owning YAML renders one child Application and the complete durable bridge "ConfigMap", "Deployment", ]); + const candidate = documents.find((item) => item.kind === "ConfigMap" && item.metadata?.name === "gitea-github-sync-candidate"); + const activeConfig = documents.find((item) => item.kind === "ConfigMap" && item.metadata?.name === "gitea-github-sync-config"); + const bridge = documents.find((item) => item.kind === "Deployment" && item.metadata?.name === "gitea-github-sync"); + expect(candidate.data["hook-topology.json"]).toBeUndefined(); + expect(candidate.data["platform_infra_gitea_hook_reconciler.py"]).toBeUndefined(); + expect(activeConfig.data["hook-topology.json"]).toContain('"ownerTargetId": "NC01"'); + expect(activeConfig.data["hook-topology.json"]).toContain('"targetId": "NC01"'); + expect(activeConfig.data["hook-topology.json"]).toContain('"lookbackSeconds": 259200'); + expect(activeConfig.data["hook-topology.json"]).toContain('"desiredHooks"'); + expect(activeConfig.data["platform_infra_gitea_hook_reconciler.py"]).toContain("github-hook-topology-reconciled"); + expect(bridge.spec.template.spec.containers.map((item: any) => item.name)).toEqual([ + "github-to-gitea-sync", + "github-hook-topology-reconciler", + ]); + const candidateContainer = documents.find((item) => item.kind === "Job" && item.metadata?.name === "gitea-github-sync-candidate").spec.template.spec.containers[0]; + expect(candidateContainer.env.map((item: any) => item.name)).not.toContain("UNIDESK_GITEA_HOOK_RECOVERY_STATE_PATH"); + const stableBridge = bridge.spec.template.spec.containers[0]; + expect(stableBridge.env.filter((item: any) => item.name === "UNIDESK_GITEA_HOOK_RECOVERY_STATE_PATH")).toHaveLength(1); + const reconciler = bridge.spec.template.spec.containers[1]; + expect(reconciler.volumeMounts).toContainEqual({ name: "durable-inbox", mountPath: "/var/lib/gitea-github-sync" }); + const sourcePath = resolve(root, "scripts/src/platform_infra_gitea_hook_reconciler.py"); + const syntax = spawnSync("python3", ["-c", "import pathlib,sys; compile(pathlib.Path(sys.argv[1]).read_text(), sys.argv[1], 'exec')", sourcePath], { encoding: "utf8" }); + expect(syntax.status).toBe(0); + expect(syntax.stderr).toBe(""); + const recoveryTests = spawnSync("python3", [resolve(root, "scripts/src/platform_infra_gitea_hook_reconciler.test.py")], { encoding: "utf8" }); + expect(recoveryTests.status).toBe(0); + expect(recoveryTests.stderr).toContain("OK"); }); diff --git a/scripts/src/platform-infra-gitea-payload.test.ts b/scripts/src/platform-infra-gitea-payload.test.ts index 1fc145b4..ebdd93ff 100644 --- a/scripts/src/platform-infra-gitea-payload.test.ts +++ b/scripts/src/platform-infra-gitea-payload.test.ts @@ -17,6 +17,7 @@ test("stdin file envelope materializes a payload larger than the current Gitea m const payloads = { manifest: oversizedManifest, repositories: JSON.stringify([{ key: "sentinel", branch: "master" }]), + hookTopology: JSON.stringify({ version: 1, repositories: [] }), statusEvaluator: "print('ok')\n", frpcToml: "serverAddr = '127.0.0.1'\n", }; @@ -56,6 +57,7 @@ test("Gitea remote runner consumes materialized files instead of base64 environm expect(source).toContain('unidesk_gitea_materialize_payloads "$tmp"'); expect(source).toContain('$tmp/gitea.k8s.yaml'); expect(source).toContain('$tmp/repos.json'); + expect(source).toContain('$tmp/hook-topology.json'); expect(source).not.toContain("UNIDESK_GITEA_MANIFEST_B64"); expect(source).not.toContain("UNIDESK_GITEA_STATUS_EVALUATOR_B64"); expect(source).not.toContain("UNIDESK_GITEA_REPOS_B64"); diff --git a/scripts/src/platform-infra-gitea-payload.ts b/scripts/src/platform-infra-gitea-payload.ts index e1343a3f..a73bab8f 100644 --- a/scripts/src/platform-infra-gitea-payload.ts +++ b/scripts/src/platform-infra-gitea-payload.ts @@ -3,6 +3,7 @@ import { createHash } from "node:crypto"; export interface GiteaRemotePayloads { manifest: string; repositories: string; + hookTopology: string; statusEvaluator: string; frpcToml: string; } @@ -16,6 +17,7 @@ interface GiteaRemotePayloadFile { const payloadFiles = (payloads: GiteaRemotePayloads): GiteaRemotePayloadFile[] => [ { role: "manifest", name: "gitea.k8s.yaml", value: payloads.manifest }, { role: "repositories", name: "repos.json", value: payloads.repositories }, + { role: "hookTopology", name: "hook-topology.json", value: payloads.hookTopology }, { role: "statusEvaluator", name: "platform_infra_gitea_status_evaluator.py", value: payloads.statusEvaluator }, { role: "frpcToml", name: "frpc.toml", value: payloads.frpcToml }, ]; diff --git a/scripts/src/platform-infra-gitea-remote.sh b/scripts/src/platform-infra-gitea-remote.sh index 07b65710..b5ba9dec 100644 --- a/scripts/src/platform-infra-gitea-remote.sh +++ b/scripts/src/platform-infra-gitea-remote.sh @@ -751,8 +751,8 @@ fetch(`http://127.0.0.1:${port}/status`) }); NODE inbox_status_rc=$? - python3 - "$tmp/repos.json" "$bridge_ready" "$service_exists" "$mirror_status_rc" "$bridge_logs_rc" "$inbox_status_rc" "$tmp/mirror-status.json" "$tmp/mirror-status.err" "$tmp/bridge.log" "$tmp/bridge.err" "$tmp/inbox-status.json" "$tmp/inbox-status.err" <<'PY' -import json, os, sys, urllib.error, urllib.parse, urllib.request + python3 - "$tmp/repos.json" "$bridge_ready" "$service_exists" "$mirror_status_rc" "$bridge_logs_rc" "$inbox_status_rc" "$tmp/mirror-status.json" "$tmp/mirror-status.err" "$tmp/bridge.log" "$tmp/bridge.err" "$tmp/inbox-status.json" "$tmp/inbox-status.err" "$tmp/hook-topology.json" <<'PY' +import hashlib, json, os, sys, urllib.error, urllib.parse, urllib.request sys.path.insert(0, os.environ["tmp"]) from platform_infra_gitea_status_evaluator import evaluate_repository, select_committed_head_record, select_target_delivery repos = json.load(open(sys.argv[1], encoding="utf-8")) @@ -767,8 +767,49 @@ bridge_log_path = sys.argv[9] bridge_log_err_path = sys.argv[10] inbox_status_path = sys.argv[11] inbox_status_err_path = sys.argv[12] +hook_topology_path = sys.argv[13] token = os.environ["UNIDESK_GITEA_GITHUB_TOKEN"] url = os.environ["UNIDESK_GITEA_WEBHOOK_PUBLIC_URL"] +try: + hook_topology = json.load(open(hook_topology_path, encoding="utf-8")) +except Exception: + hook_topology = {} +managed_url_prefix = hook_topology.get("managedUrlPrefix") or url +desired_events = hook_topology.get("events") if isinstance(hook_topology.get("events"), list) else ["push"] +topology_repositories = hook_topology.get("repositories") if isinstance(hook_topology.get("repositories"), list) else [] +def url_hash(value): + return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] +def hook_url(item): + config = item.get("config") if isinstance(item.get("config"), dict) else {} + return config.get("url") if isinstance(config.get("url"), str) else "" +def is_managed_url(value): + return value == managed_url_prefix or value.startswith(managed_url_prefix.rstrip("/") + "/") +def hook_config_matches(item): + config = item.get("config") if isinstance(item.get("config"), dict) else {} + return item.get("name") == "web" and item.get("active") is True and set(item.get("events") or []) == set(desired_events) and config.get("content_type") == "json" and str(config.get("insecure_ssl", "0")) == "0" +def topology_observation(repository, hooks): + spec = next((item for item in topology_repositories if isinstance(item, dict) and item.get("repository") == repository), {}) + desired_urls = spec.get("desiredUrls") if isinstance(spec.get("desiredUrls"), list) else [url] + managed = [item for item in hooks if item.get("name") == "web" and is_managed_url(hook_url(item))] + desired_set = set(desired_urls) + drift_types = [] + for desired_url in desired_urls: + exact = [item for item in managed if hook_url(item) == desired_url] + if not exact: + drift_types.append("desired-url-missing") + elif len(exact) > 1: + drift_types.append("desired-url-duplicate") + if exact and not hook_config_matches(exact[0]): + drift_types.append("desired-hook-config-mismatch") + if any(hook_url(item) not in desired_set for item in managed): + drift_types.append("stale-managed-url") + return { + "ready": not drift_types, + "desiredUrlSha256": url_hash(url), + "desiredTopologyUrlSha256": sorted(url_hash(item) for item in desired_urls), + "observedManagedUrlSha256": sorted(url_hash(hook_url(item)) for item in managed), + "driftTypes": sorted(set(drift_types)), + } def text(path, limit=1600): try: return open(path, encoding="utf-8", errors="replace").read()[-limit:] @@ -950,7 +991,8 @@ for repo in repos: hooks = json.loads(result.get("body") or "[]") except Exception: hooks = [] - matches = [item for item in hooks if (item.get("config") or {}).get("url") == url] + matches = [item for item in hooks if hook_url(item) == url] + topology = topology_observation(repository, hooks) head = github_head(repository, branch) head_record = select_committed_head_record(repo["key"], head.get("sha"), inbox_records) head_delivery_id = head_record.get("deliveryId") if isinstance(head_record, dict) else None @@ -995,6 +1037,7 @@ for repo in repos: "matchingHookCount": len(matches), "matchingHookIds": [item.get("id") for item in matches], "hookTopologyReason": topology_reason, + "hookTopology": topology, "latest": latest_delivery, "latestPush": latest_push_delivery, "selectedPush": selected_push_delivery, @@ -1008,8 +1051,9 @@ bridge = { "serviceExists": service_exists, "durableInbox": inbox_status, } +failed_delivery_recovery = inbox_status.get("failedDeliveryRecovery") if isinstance(inbox_status.get("failedDeliveryRecovery"), dict) else {"enabled": False, "ready": True} payload = { - "ok": bridge["ready"] and service_exists and mirror_status_rc == 0 and inbox_status_rc == 0 and inbox_status.get("ready") is True and all(row["hookReady"] and row["authorityMatches"] and not row["bridgeEventStale"] for row in rows), + "ok": bridge["ready"] and service_exists and mirror_status_rc == 0 and inbox_status_rc == 0 and inbox_status.get("ready") is True and (failed_delivery_recovery.get("enabled") is not True or failed_delivery_recovery.get("ready") is True) and all(row["hookReady"] and row.get("hookTopology", {}).get("ready") is True and row["authorityMatches"] and not row["bridgeEventStale"] for row in rows), "bridge": bridge, "repositories": rows, "url": url, diff --git a/scripts/src/platform-infra-gitea-render.ts b/scripts/src/platform-infra-gitea-render.ts index 8caefcce..b7bd9579 100644 --- a/scripts/src/platform-infra-gitea-render.ts +++ b/scripts/src/platform-infra-gitea-render.ts @@ -102,9 +102,11 @@ export function renderMirrorWebhookApply(result: Record): Rende export function renderMirrorWebhookStatus(result: Record, disclosure: Record): RenderedCliResult { const repos = arrayRecords(result.repositories).map((repo) => { - const latestPush = record(repo.latestPushDelivery); - const latest = record(repo.latestDelivery); - const delivery = latestPush.event !== "" ? latestPush : latest; + const delivery = record(repo.selectedPushDelivery); + const topology = record(repo.hookTopology); + const topologyDrift = Array.isArray(topology.driftTypes) && topology.driftTypes.length > 0 + ? topology.driftTypes.join(",") + : topology.ready === true ? "ready" : "unknown"; const inbox = record(repo.durableInboxRecord); const inboxResult = record(inbox.result); const inboxState = stringValue(repo.durableInboxState, "missing"); @@ -121,7 +123,8 @@ export function renderMirrorWebhookStatus(result: Record, discl stringValue(repo.snapshotCommit).slice(0, 12), boolText(repo.authorityMatches), boolText(repo.bridgeEventStale), - `${stringValue(repo.deliveryId)}:${stringValue(delivery.statusCode)}:${stringValue(delivery.status)}`, + topologyDrift, + `${stringValue(delivery.deliveryId)}:${stringValue(delivery.statusCode)}:${stringValue(delivery.status)}`, inboxState, durableProof, compactTail(stringValue(repo.error)), @@ -129,7 +132,12 @@ export function renderMirrorWebhookStatus(result: Record, discl }); const bridge = record(result.bridge); const bridgeLogs = record(result.bridgeLogs); - const retry = record(record(record(result.webhook).bridge).retry); + const recovery = record(record(bridge.durableInbox).failedDeliveryRecovery); + const recoveryCounts = record(recovery.counts); + const recoveryText = `${stringValue(recovery.state)}:${stringValue(recoveryCounts.pending, "0")}/${stringValue(recoveryCounts.attempted, "0")}/${stringValue(recoveryCounts["delivery-succeeded-awaiting-source"], "0")}/${stringValue(recoveryCounts["source-committed"], "0")}/${stringValue(recoveryCounts["terminal-exhausted"], "0")}`; + const webhook = record(result.webhook); + const retry = record(record(webhook.bridge).retry); + const hookReconcile = record(webhook.hookReconcile); const retryText = `${stringValue(retry.maxAttempts)}x/${stringValue(retry.initialDelayMs)}-${stringValue(retry.maxDelayMs)}ms`; const selection = record(disclosure.selection); const selectedRepoKey = stringValue(selection.selectedRepoKey, ""); @@ -176,10 +184,10 @@ export function renderMirrorWebhookStatus(result: Record, discl ]; return rendered(disclosure, "platform-infra gitea mirror webhook status", [ "PLATFORM-INFRA GITEA MIRROR WEBHOOK STATUS", - ...table(["TARGET", "BRIDGE", "READY", "RETRY", "URL", "LOGS"], [[stringValue(record(result.target).id), stringValue(bridge.deployment), boolText(bridge.ready), retryText, stringValue(record(result.webhook).publicUrl), `rc=${stringValue(bridgeLogs.exitCode)}`]]), + ...table(["TARGET", "BRIDGE", "READY", "RETRY", "RECOVERY", "URL_SHA", "LOGS"], [[stringValue(record(result.target).id), stringValue(bridge.deployment), boolText(bridge.ready), retryText, recoveryText, stringValue(hookReconcile.desiredUrlSha256), `rc=${stringValue(bridgeLogs.exitCode)}`]]), "", "GITHUB HOOKS", - ...(repos.length === 0 ? ["-"] : table(["KEY", "HOOK", "GH_HEAD", "GITEA", "SNAPSHOT", "MATCH", "STALE", "DELIVERY", "DURABLE", "PROOF", "ERROR"], repos)), + ...(repos.length === 0 ? ["-"] : table(["KEY", "HOOK", "GH_HEAD", "GITEA", "SNAPSHOT", "MATCH", "STALE", "TOPOLOGY", "DELIVERY", "DURABLE", "PROOF", "ERROR"], repos)), ...drillDown, ...remoteErrorLines(result), "", diff --git a/scripts/src/platform-infra-gitea-status-evaluator.test.ts b/scripts/src/platform-infra-gitea-status-evaluator.test.ts index 7ae08db8..3a650261 100644 --- a/scripts/src/platform-infra-gitea-status-evaluator.test.ts +++ b/scripts/src/platform-infra-gitea-status-evaluator.test.ts @@ -121,6 +121,28 @@ describe("Gitea source authority status evaluator", () => { expect(row.staleReason).toBe("target-branch-delivery-failed"); }); + test("missing selected delivery never inherits latest push status", () => { + const head = "7".repeat(40); + const row = evaluate({ + action: "evaluate-repository", + repo: { key: "fixture", upstream: { repository: "pikasTech/fixture", branch: "main" }, snapshot: { prefix: "refs/snapshots/source" } }, + hook: { + hookId: 1, + hookReady: true, + latestPush: { deliveryId: "unrelated-latest", event: "push", statusCode: 202, repository: "pikasTech/fixture", ref: "refs/heads/main", requestedCommit: head }, + selectedPush: null, + }, + head: { ok: true, sha: head }, + mirror: { branchCommit: head, snapshotCommit: head, snapshotRef: `refs/snapshots/source/${head}` }, + bridgeEvents: [], + inboxRecords: [], + }); + expect(row.latestPushDelivery.deliveryId).toBe("unrelated-latest"); + expect(row.selectedPushDelivery).toBeNull(); + expect(row.deliveryId).toBe(""); + expect(row.deliveryAccepted).toBe(false); + }); + test("selects the exact durable proof across duplicate exact-url hooks without hiding topology drift", () => { const head = "8".repeat(40); const row = evaluate({ @@ -310,6 +332,7 @@ function evaluateRow(input: { status: 200, latest: { deliveryId: input.deliveryId, event: "push", statusCode: input.statusCode, deliveredAt: input.deliveredAt }, latestPush: { deliveryId: input.deliveryId, event: "push", statusCode: input.statusCode, deliveredAt: input.deliveredAt, repository: "pikasTech/fixture", ref: "refs/heads/main", requestedCommit: input.requestedCommit ?? input.head }, + selectedPush: { deliveryId: input.deliveryId, event: "push", statusCode: input.statusCode, deliveredAt: input.deliveredAt, repository: "pikasTech/fixture", ref: "refs/heads/main", requestedCommit: input.requestedCommit ?? input.head }, }, head: { ok: true, sha: input.head }, mirror: { diff --git a/scripts/src/platform-infra-gitea.ts b/scripts/src/platform-infra-gitea.ts index 4ece87e8..ab5468ca 100644 --- a/scripts/src/platform-infra-gitea.ts +++ b/scripts/src/platform-infra-gitea.ts @@ -54,6 +54,7 @@ const configLabel = GITEA_CONFIG_LABEL; const remoteScriptFile = rootPath("scripts", "src", "platform-infra-gitea-remote.sh"); const githubSyncServerFile = rootPath("scripts", "src", "platform-infra-gitea-github-sync-server.mjs"); const githubSyncEntrypointFile = rootPath("scripts", "src", "platform-infra-gitea-github-sync-entrypoint.mjs"); +const githubHookReconcilerFile = rootPath("scripts", "src", "platform_infra_gitea_hook_reconciler.py"); const statusEvaluatorFile = rootPath("scripts", "src", "platform_infra_gitea_status_evaluator.py"); const fieldManager = "unidesk-platform-infra-gitea"; interface CommonOptions { @@ -621,7 +622,64 @@ function renderGithubSyncManifest(gitea: GiteaConfig, target: GiteaTarget): stri const reposJson = JSON.stringify(repositoriesForTarget(gitea, target).map(remoteRepoSpec), null, 2); const serverSource = readFileSync(githubSyncServerFile, "utf8"); const entrypointSource = readFileSync(githubSyncEntrypointFile, "utf8"); - const configHash = createHash("sha256").update(reposJson).update("\0").update(serverSource).update("\0").update(entrypointSource).digest("hex").slice(0, 16); + const ownsHookReconcile = sync.hookReconcile.enabled && sync.hookReconcile.ownerTargetId.toLowerCase() === target.id.toLowerCase(); + const hookTopologyJson = ownsHookReconcile ? JSON.stringify(githubHookTopology(gitea), null, 2) : ""; + const hookReconcilerSource = ownsHookReconcile ? readFileSync(githubHookReconcilerFile, "utf8") : ""; + const hookReconcilerConfig = ownsHookReconcile ? ` hook-topology.json: | +${indentBlock(hookTopologyJson, 4)} + platform_infra_gitea_hook_reconciler.py: | +${indentBlock(hookReconcilerSource, 4)} +` : ""; + const hookRecoveryStateEnv = ownsHookReconcile && sync.hookReconcile.failedDeliveryRecovery.enabled + ? ` - name: UNIDESK_GITEA_HOOK_RECOVERY_STATE_PATH + value: ${yamlQuote(sync.hookReconcile.failedDeliveryRecovery.statePath)} +` + : ""; + const hookReconcilerContainer = ownsHookReconcile ? ` + - name: github-hook-topology-reconciler + image: ${sync.bridge.image} + imagePullPolicy: IfNotPresent + command: + - python3 + - /etc/gitea-github-sync/platform_infra_gitea_hook_reconciler.py + - /etc/gitea-github-sync/hook-topology.json + env: + - name: UNIDESK_GITEA_TARGET_ID + value: ${yamlQuote(target.id)} + - name: GITHUB_TOKEN + valueFrom: + secretKeyRef: + name: ${sync.bridge.secretName} + key: ${gitea.sourceAuthority.credentials.github.gitFetchCredential.secretRef.key} + - name: GITHUB_WEBHOOK_SECRET + valueFrom: + secretKeyRef: + name: ${sync.bridge.secretName} + key: github-webhook-secret + - name: HTTPS_PROXY + value: ${yamlQuote(gitea.sourceAuthority.githubProxy.enabled ? gitea.sourceAuthority.githubProxy.url : "")} + - name: HTTP_PROXY + value: ${yamlQuote(gitea.sourceAuthority.githubProxy.enabled ? gitea.sourceAuthority.githubProxy.url : "")} + - name: NO_PROXY + value: ${yamlQuote(gitea.sourceAuthority.githubProxy.noProxy.join(","))} + volumeMounts: + - name: config + mountPath: /etc/gitea-github-sync + readOnly: true + - name: durable-inbox + mountPath: ${sync.bridge.inbox.path}` : ""; + const configHash = createHash("sha256") + .update(reposJson) + .update("\0") + .update(serverSource) + .update("\0") + .update(entrypointSource) + .update("\0") + .update(hookTopologyJson) + .update("\0") + .update(hookReconcilerSource) + .digest("hex") + .slice(0, 16); const gate = sync.bridge.candidateGate; const candidateManifest = gate.enabled ? `--- apiVersion: v1 @@ -792,6 +850,7 @@ ${indentBlock(reposJson, 4)} ${indentBlock(serverSource, 4)} entrypoint.mjs: | ${indentBlock(entrypointSource, 4)} +${hookReconcilerConfig} --- apiVersion: v1 kind: PersistentVolumeClaim @@ -893,7 +952,7 @@ spec: value: ${yamlQuote(String(sync.ingressRetry.maxBodyBytes))} - name: UNIDESK_GITEA_WEBHOOK_INBOX_PATH value: ${yamlQuote(sync.bridge.inbox.path)} - - name: UNIDESK_GITEA_WEBHOOK_INBOX_MAX_BYTES +${hookRecoveryStateEnv} - name: UNIDESK_GITEA_WEBHOOK_INBOX_MAX_BYTES value: ${yamlQuote(String(sync.bridge.inbox.maxBytes))} - name: UNIDESK_GITEA_WEBHOOK_COMMITTED_RETENTION_SECONDS value: ${yamlQuote(String(sync.bridge.inbox.committedRetentionSeconds))} @@ -940,6 +999,7 @@ spec: readOnly: true - name: durable-inbox mountPath: ${sync.bridge.inbox.path} +${hookReconcilerContainer} volumes: - name: config configMap: @@ -1043,6 +1103,7 @@ function remoteScript(action: "apply" | "status" | "validate" | "mirror-bootstra const payloads = { manifest, repositories: JSON.stringify((params.repos ?? []).map((repo) => remoteRepoSpec(repo))), + hookTopology: JSON.stringify(githubHookTopology(gitea)), statusEvaluator: readFileSync(statusEvaluatorFile, "utf8"), frpcToml: params.frpcSecret?.frpcToml ?? "", }; @@ -1068,6 +1129,17 @@ function policyChecks(gitea: GiteaConfig, target: GiteaTarget, manifest: string) ok: sync.bridge.replicas === 1 && /strategy:\n\s+type: Recreate/u.test(manifest), detail: "The durable inbox has one bridge replica and Deployment strategy Recreate, so journal writes have one owner during rollout.", }, + { + name: "github-hook-topology-single-writer", + ok: !sync.hookReconcile.enabled || ( + sync.hookReconcile.ownerTargetId.toLowerCase() === target.id.toLowerCase() + ? (manifest.match(/name: github-hook-topology-reconciler/gu) ?? []).length === 1 + && manifest.includes("platform_infra_gitea_hook_reconciler.py") + && manifest.includes("hook-topology.json") + : !manifest.includes("name: github-hook-topology-reconciler") + ), + detail: "The owning YAML selects one target-side reconciler for the complete GitHub hook topology; candidate gates and non-owner targets never write hooks.", + }, { name: "bridge-candidate-presync-gate", ok: sync.bridge.candidateGate.enabled @@ -1306,6 +1378,54 @@ function githubWebhookPublicUrl(gitea: GiteaConfig, target: GiteaTarget): string return `${base}${targetWebhookSync(gitea, target).publicPath}`; } +function githubHookTopology(gitea: GiteaConfig): Record { + const sync = gitea.sourceAuthority.webhookSync; + const base = gitea.app.publicExposure.publicBaseUrl.replace(/\/+$/u, ""); + const desiredByRepository = new Map; branches: Set }>>(); + for (const repo of gitea.sourceAuthority.repositories) { + const target = resolveTarget(gitea, repo.targetId); + const desired = desiredByRepository.get(repo.upstream.repository) ?? new Map(); + const key = target.id.toLowerCase(); + const hook = desired.get(key) ?? { + targetId: target.id, + url: githubWebhookPublicUrl(gitea, target), + repoKeys: new Set(), + branches: new Set(), + }; + hook.repoKeys.add(repo.key); + hook.branches.add(repo.upstream.branch); + desired.set(key, hook); + desiredByRepository.set(repo.upstream.repository, desired); + } + return { + version: 1, + kind: "GiteaGithubHookTopology", + ownerTargetId: sync.hookReconcile.ownerTargetId, + managedUrlPrefix: `${base}${sync.publicPath}`, + bridgeStatusUrl: `http://127.0.0.1:${sync.bridge.httpPort}/status`, + events: sync.events, + reconcile: { + intervalMs: sync.hookReconcile.intervalMs, + requestTimeoutMs: sync.hookReconcile.requestTimeoutMs, + listMaxPages: sync.hookReconcile.listMaxPages, + failedDeliveryRecovery: sync.hookReconcile.failedDeliveryRecovery, + }, + repositories: [...desiredByRepository.entries()] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([repository, desired]) => { + const desiredHooks = [...desired.values()] + .sort((left, right) => left.targetId.localeCompare(right.targetId)) + .map((item) => ({ + targetId: item.targetId, + url: item.url, + repoKeys: [...item.repoKeys].sort(), + branches: [...item.branches].sort(), + })); + return { repository, desiredUrls: desiredHooks.map((item) => item.url).sort(), desiredHooks }; + }), + }; +} + function statusSummary(payload: Record): Record { return { ready: payload.ready === true, @@ -1398,6 +1518,8 @@ function mirrorReadOnlyNextCommands(targetId: string): Record { function webhookSyncSummary(gitea: GiteaConfig, target: GiteaTarget): Record { const sync = targetWebhookSync(gitea, target); + const topology = githubHookTopology(gitea); + const targetUrl = githubWebhookPublicUrl(gitea, target); const ingressAttemptBudgetMs = sync.ingressRetry.enabled ? deriveGiteaWebhookAttemptBudgetMs(sync.responseBudgetMs, sync.ingressRetry) : null; @@ -1407,7 +1529,13 @@ function webhookSyncSummary(gitea: GiteaConfig, target: GiteaTarget): Record 1: + drift_types.append("desired-url-duplicate") + if matches and not hook_config_matches(matches[0], events): + drift_types.append("desired-hook-config-mismatch") + if any(hook_url(item) not in desired_set for item in managed): + drift_types.append("stale-managed-url") + return { + "ready": not drift_types, + "desiredUrlHashes": sorted(url_hash(item) for item in desired_urls), + "observedUrlHashes": sorted(url_hash(hook_url(item)) for item in managed), + "driftTypes": sorted(set(drift_types)), + } + + +def list_hook_deliveries(token, timeout_seconds, repository, hook_id, history_per_hook): + encoded = urllib.parse.quote(repository, safe="/") + _, rows = request( + token, + timeout_seconds, + "GET", + f"/repos/{encoded}/hooks/{hook_id}/deliveries?per_page={history_per_hook}", + ) + if not isinstance(rows, list): + raise RuntimeError("github-hook-deliveries-response-invalid") + return [item for item in rows if isinstance(item, dict)] + + +def hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id): + encoded = urllib.parse.quote(repository, safe="/") + _, detail = request(token, timeout_seconds, "GET", f"/repos/{encoded}/hooks/{hook_id}/deliveries/{api_delivery_id}") + if not isinstance(detail, dict): + raise RuntimeError("github-hook-delivery-detail-invalid") + payload = detail.get("request", {}).get("payload", {}) if isinstance(detail.get("request"), dict) else {} + return { + **detail, + "repositoryName": payload.get("repository", {}).get("full_name") if isinstance(payload.get("repository"), dict) else None, + "ref": payload.get("ref"), + "requestedCommit": payload.get("after"), + } + + +def bridge_committed_deliveries(topology, timeout_seconds): + req = urllib.request.Request(topology["bridgeStatusUrl"], method="GET") + try: + with urllib.request.urlopen(req, timeout=timeout_seconds) as response: + body = response.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as error: + body = error.read().decode("utf-8", errors="replace") + payload = json.loads(body or "{}") + return { + (str(item.get("repo") or ""), str(item.get("deliveryId") or "")) + for item in payload.get("deliveries", []) + if isinstance(item, dict) and item.get("state") == "committed" + } + + +def delivery_identity(repository, hook_id, delivery_id): + return f"{repository}|{hook_id}|{delivery_id}" + + +def persist_record(ledger, path, key, record, now_ms): + record["updatedAtEpochMs"] = now_ms + record["updatedAt"] = now_iso(now_ms) + ledger["records"][key] = record + atomic_write_ledger(path, ledger) + + +def hook_body(url, events, secret): + return { + "name": "web", + "active": True, + "events": events, + "config": {"url": url, "content_type": "json", "secret": secret, "insecure_ssl": "0"}, + } + + +def source_committed_record(ledger, state_path, key, record, status_code, now_ms): + record.update({ + "state": "source-committed", + "completionReason": "source-committed", + "lastStatusCode": status_code, + "nextAt": None, + "nextAtEpochMs": 0, + }) + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-recovery-complete", + repository=record["repository"], + hookId=record["hookId"], + deliveryId=record["deliveryId"], + state=record["state"], + completionReason="source-committed", + attempts=record["attempts"], + ok=True, + ) + + +def delivery_succeeded_record(ledger, state_path, key, record, status_code, now_ms): + if record.get("state") == "delivery-succeeded-awaiting-source": + return + record.update({ + "state": "delivery-succeeded-awaiting-source", + "completionReason": "github-delivery-succeeded", + "lastStatusCode": status_code, + "nextAt": None, + "nextAtEpochMs": 0, + }) + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-succeeded-awaiting-source", + repository=record["repository"], + hookId=record["hookId"], + deliveryId=record["deliveryId"], + state=record["state"], + attempts=record["attempts"], + ok=True, + ) + + +def terminal_record(ledger, state_path, key, record, reason, status_code, now_ms): + record.update({ + "state": "terminal-exhausted", + "completionReason": reason, + "lastStatusCode": status_code, + "nextAt": None, + "nextAtEpochMs": 0, + }) + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-recovery-terminal", + repository=record["repository"], + hookId=record["hookId"], + deliveryId=record["deliveryId"], + state=record["state"], + completionReason=reason, + attempts=record["attempts"], + ok=False, + ) + + +def terminalize_no_longer_desired(ledger, recovery, repository, desired_hook, observed_hook_id): + now_ms = int(time.time() * 1000) + desired_url_hash = url_hash(desired_hook["url"]) + for key, record in list(ledger["records"].items()): + if not isinstance(record, dict) or record.get("repository") != repository: + continue + if record.get("targetId") not in (None, recovery["targetId"]): + continue + if record.get("state") not in ("pending", "attempted"): + continue + if int(record.get("hookId") or 0) == observed_hook_id and record.get("hookUrlSha256") == desired_url_hash: + continue + terminal_record( + ledger, + recovery["statePath"], + key, + record, + "hook-no-longer-desired", + int(record.get("lastStatusCode") or 0), + now_ms, + ) + + +def close_committed_ledger_records(ledger, recovery, committed): + now_ms = int(time.time() * 1000) + for key, record in list(ledger["records"].items()): + if not isinstance(record, dict) or record.get("targetId") not in (None, recovery["targetId"]): + continue + if record.get("state") == "source-committed": + continue + delivery_id = str(record.get("deliveryId") or "") + repo_keys = record.get("repoKeys") if isinstance(record.get("repoKeys"), list) else [] + if not any((str(repo_key), delivery_id) in committed for repo_key in repo_keys): + continue + source_committed_record( + ledger, + recovery["statePath"], + key, + record, + int(record.get("lastStatusCode") or 0), + now_ms, + ) + + +def recover_hook_deliveries(topology, repository_spec, desired_hook, observed_hook, token, ledger, committed, budget): + reconcile = topology["reconcile"] + recovery = reconcile["failedDeliveryRecovery"] + timeout_seconds = reconcile["requestTimeoutMs"] / 1000 + repository = repository_spec["repository"] + hook_id = int(observed_hook["id"]) + now_ms = int(time.time() * 1000) + state_path = recovery["statePath"] + desired_refs = {f"refs/heads/{branch}" for branch in desired_hook["branches"]} + retry_codes = set(recovery["retryStatusCodes"]) + rows = list_hook_deliveries(token, timeout_seconds, repository, hook_id, recovery["historyPerHook"]) + rows_by_api_id = {str(item.get("id")): item for item in rows if item.get("id") is not None} + + for record in ledger["records"].values(): + if not isinstance(record, dict) or record.get("repository") != repository or int(record.get("hookId") or 0) != hook_id: + continue + if record.get("state") not in ("pending", "attempted", "delivery-succeeded-awaiting-source"): + continue + api_delivery_id = str(record.get("apiDeliveryId") or "") + if api_delivery_id and api_delivery_id not in rows_by_api_id: + detail = hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id) + rows.append(detail) + rows_by_api_id[api_delivery_id] = detail + + latest_by_delivery = {} + for row in rows: + delivery_id = str(row.get("guid") or row.get("id") or "") + if not delivery_id: + continue + current = latest_by_delivery.get(delivery_id) + score = (parse_epoch_ms(row.get("delivered_at")), int(row.get("id") or 0)) + current_score = (-1, -1) if current is None else (parse_epoch_ms(current.get("delivered_at")), int(current.get("id") or 0)) + if score > current_score: + latest_by_delivery[delivery_id] = row + rows = sorted(latest_by_delivery.values(), key=lambda item: (parse_epoch_ms(item.get("delivered_at")), int(item.get("id") or 0)), reverse=True) + recovery_ok = True + for row in rows: + if row.get("event") != "push" or row.get("id") is None: + continue + api_delivery_id = str(row["id"]) + delivery_id = str(row.get("guid") or row["id"]) + key = delivery_identity(repository, hook_id, delivery_id) + record = ledger["records"].get(key) + if isinstance(record, dict) and record.get("state") in ("source-committed", "terminal-exhausted"): + continue + status_code = int(row.get("status_code") or 0) + eligible_failure = status_code in retry_codes + if record is None and not eligible_failure: + continue + delivered_at_ms = parse_epoch_ms(row.get("delivered_at")) + if record is None and (delivered_at_ms == 0 or delivered_at_ms < now_ms - recovery["lookbackSeconds"] * 1000): + continue + if isinstance(record, dict) and record.get("state") in ("pending", "attempted") and delivered_at_ms > 0 and delivered_at_ms < now_ms - recovery["lookbackSeconds"] * 1000: + terminal_record(ledger, state_path, key, record, "redelivery-window-expired", status_code, now_ms) + continue + + detail = hook_delivery_detail(token, timeout_seconds, repository, hook_id, api_delivery_id) + status_code = int(detail.get("status_code") or status_code or 0) + if detail.get("repositoryName") != repository or detail.get("ref") not in desired_refs: + if record is not None and record.get("state") in ("pending", "attempted"): + terminal_record(ledger, state_path, key, record, "delivery-scope-mismatch", status_code, now_ms) + continue + requested_commit = str(detail.get("requestedCommit") or "") + if record is None: + record = { + "repository": repository, + "targetId": desired_hook["targetId"], + "repoKeys": desired_hook["repoKeys"], + "hookId": hook_id, + "hookUrlSha256": url_hash(desired_hook["url"]), + "deliveryId": delivery_id, + "apiDeliveryId": api_delivery_id, + "ref": detail["ref"], + "requestedCommit": requested_commit, + "state": "pending", + "attempts": 0, + "cooldownMs": recovery["cooldownMs"], + "nextAtEpochMs": now_ms, + "nextAt": now_iso(now_ms), + "createdAt": now_iso(now_ms), + "lastStatusCode": status_code, + "completionReason": None, + } + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-recovery-pending", + repository=repository, + hookId=hook_id, + deliveryId=delivery_id, + state="pending", + attempts=0, + nextAt=record["nextAt"], + ok=True, + ) + + source_committed = any((repo_key, delivery_id) in committed for repo_key in desired_hook["repoKeys"]) + if source_committed: + source_committed_record(ledger, state_path, key, record, status_code, now_ms) + continue + if 200 <= status_code < 300: + delivery_succeeded_record(ledger, state_path, key, record, status_code, now_ms) + continue + if record.get("state") == "delivery-succeeded-awaiting-source": + continue + if status_code == 0: + continue + if status_code not in retry_codes: + terminal_record(ledger, state_path, key, record, "non-retryable-status", status_code, now_ms) + continue + if int(record.get("nextAtEpochMs") or 0) > now_ms: + continue + if int(record.get("attempts") or 0) >= recovery["maxAttemptsPerDelivery"]: + terminal_record(ledger, state_path, key, record, "max-attempts-exhausted", status_code, now_ms) + continue + if budget["remaining"] < 1: + continue + + record["state"] = "attempted" + record["attempts"] = int(record.get("attempts") or 0) + 1 + record["lastStatusCode"] = status_code + record["lastOutcome"] = "redelivery-requested-write-ahead" + record["nextAtEpochMs"] = now_ms + recovery["cooldownMs"] + record["nextAt"] = now_iso(record["nextAtEpochMs"]) + persist_record(ledger, state_path, key, record, now_ms) + budget["remaining"] -= 1 + encoded = urllib.parse.quote(repository, safe="/") + try: + request( + token, + timeout_seconds, + "POST", + f"/repos/{encoded}/hooks/{hook_id}/deliveries/{api_delivery_id}/attempts", + expected=(202,), + ) + record["lastOutcome"] = "redelivery-accepted" + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-redelivery-requested", + repository=repository, + hookId=hook_id, + deliveryId=delivery_id, + state=record["state"], + attempts=record["attempts"], + nextAt=record["nextAt"], + ok=True, + ) + except Exception as error: + recovery_ok = False + record["lastOutcome"] = "redelivery-request-failed" + persist_record(ledger, state_path, key, record, now_ms) + emit( + "github-hook-delivery-redelivery-failed", + repository=repository, + hookId=hook_id, + deliveryId=delivery_id, + state=record["state"], + attempts=record["attempts"], + nextAt=record["nextAt"], + errorType=type(error).__name__, + error=str(error)[:160], + ok=False, + ) + return recovery_ok + + +def reconcile_repository(topology, repository_spec, token, secret, force_patch): + reconcile = topology["reconcile"] + timeout_seconds = reconcile["requestTimeoutMs"] / 1000 + max_pages = reconcile["listMaxPages"] + repository = repository_spec["repository"] + desired_urls = repository_spec["desiredUrls"] + events = topology["events"] + prefix = topology["managedUrlPrefix"] + hooks = list_hooks(token, timeout_seconds, max_pages, repository) + mutations = {"created": 0, "updated": 0, "deleted": 0} + encoded = urllib.parse.quote(repository, safe="/") + + for desired_url in desired_urls: + matches = sorted( + [item for item in hooks if item.get("name") == "web" and hook_url(item) == desired_url], + key=lambda item: int(item.get("id") or 0), + ) + keeper = matches[0] if matches else None + if keeper is None: + _, created = request(token, timeout_seconds, "POST", f"/repos/{encoded}/hooks", hook_body(desired_url, events, secret), expected=(201,)) + if isinstance(created, dict): + hooks.append(created) + mutations["created"] += 1 + elif force_patch or not hook_config_matches(keeper, events): + hook_id = int(keeper["id"]) + request(token, timeout_seconds, "PATCH", f"/repos/{encoded}/hooks/{hook_id}", hook_body(desired_url, events, secret)) + mutations["updated"] += 1 + for duplicate in matches[1:]: + hook_id = int(duplicate["id"]) + request(token, timeout_seconds, "DELETE", f"/repos/{encoded}/hooks/{hook_id}", expected=(204,), tolerate=(404,)) + mutations["deleted"] += 1 + + desired_set = set(desired_urls) + for stale in hooks: + stale_url = hook_url(stale) + if stale.get("name") != "web" or not is_managed_url(stale_url, prefix) or stale_url in desired_set: + continue + hook_id = int(stale["id"]) + request(token, timeout_seconds, "DELETE", f"/repos/{encoded}/hooks/{hook_id}", expected=(204,), tolerate=(404,)) + mutations["deleted"] += 1 + + observed = list_hooks(token, timeout_seconds, max_pages, repository) + observation = topology_observation(observed, desired_urls, prefix, events) + emit( + "github-hook-topology-reconciled", + repository=repository, + ok=observation["ready"], + desiredUrlHashes=observation["desiredUrlHashes"], + observedUrlHashes=observation["observedUrlHashes"], + driftTypes=observation["driftTypes"], + mutations=mutations, + ) + return observation, observed + + +def reconcile_once(topology, token, secret, force_patch): + started = time.monotonic() + topology_ready = True + recovery_ready = True + recovery = topology["reconcile"]["failedDeliveryRecovery"] + ledger = None + committed = set() + budget = {"remaining": recovery["maxRedeliveriesPerCycle"]} + if recovery["enabled"]: + try: + ledger = load_ledger(recovery["statePath"]) + removed = cleanup_ledger(ledger, recovery["stateRetentionSeconds"], int(time.time() * 1000)) + if removed: + atomic_write_ledger(recovery["statePath"], ledger) + committed = bridge_committed_deliveries(topology, topology["reconcile"]["requestTimeoutMs"] / 1000) + close_committed_ledger_records(ledger, recovery, committed) + except Exception as error: + recovery_ready = False + ledger = None + emit("github-hook-delivery-recovery-ledger-unavailable", ok=False, errorType=type(error).__name__, error=str(error)[:160]) + for repository_spec in topology["repositories"]: + try: + observation, observed_hooks = reconcile_repository(topology, repository_spec, token, secret, force_patch) + if not observation["ready"]: + topology_ready = False + continue + if recovery["enabled"] and ledger is not None: + desired_hooks = [item for item in repository_spec["desiredHooks"] if item["targetId"] == recovery["targetId"]] + if len(desired_hooks) != 1: + recovery_ready = False + emit("github-hook-delivery-recovery-scope-invalid", repository=repository_spec["repository"], ok=False, targetId=recovery["targetId"]) + continue + desired_hook = desired_hooks[0] + exact = [item for item in observed_hooks if item.get("name") == "web" and hook_url(item) == desired_hook["url"]] + if len(exact) != 1 or not hook_config_matches(exact[0], topology["events"]): + recovery_ready = False + emit("github-hook-delivery-recovery-skipped-topology-drift", repository=repository_spec["repository"], ok=False, targetId=recovery["targetId"]) + continue + terminalize_no_longer_desired(ledger, recovery, repository_spec["repository"], desired_hook, int(exact[0]["id"])) + if not recover_hook_deliveries(topology, repository_spec, desired_hook, exact[0], token, ledger, committed, budget): + recovery_ready = False + except Exception as error: + topology_ready = False + recovery_ready = False + emit( + "github-hook-topology-reconcile-failed", + repository=repository_spec["repository"], + ok=False, + errorType=type(error).__name__, + error=str(error)[:160], + ) + emit( + "github-hook-topology-cycle-complete", + ok=topology_ready and recovery_ready, + topologyReady=topology_ready, + recoveryReady=recovery_ready, + redeliveryRequests=recovery["maxRedeliveriesPerCycle"] - budget["remaining"], + repositoryCount=len(topology["repositories"]), + elapsedMs=round((time.monotonic() - started) * 1000), + ) + return topology_ready + + +def stop(_signum, _frame): + STOP.set() + + +def main(): + if len(sys.argv) != 2: + raise SystemExit("usage: platform_infra_gitea_hook_reconciler.py ") + topology = load_topology(sys.argv[1]) + if topology.get("ownerTargetId") != os.environ.get("UNIDESK_GITEA_TARGET_ID"): + raise SystemExit("hook reconciler target is not the YAML owner") + token = os.environ["GITHUB_TOKEN"] + secret = os.environ["GITHUB_WEBHOOK_SECRET"] + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + force_patch = True + while not STOP.is_set(): + if reconcile_once(topology, token, secret, force_patch): + force_patch = False + STOP.wait(topology["reconcile"]["intervalMs"] / 1000) + + +if __name__ == "__main__": + main() diff --git a/scripts/src/platform_infra_gitea_hook_reconciler.test.py b/scripts/src/platform_infra_gitea_hook_reconciler.test.py new file mode 100644 index 00000000..096c476b --- /dev/null +++ b/scripts/src/platform_infra_gitea_hook_reconciler.test.py @@ -0,0 +1,240 @@ +#!/usr/bin/env python3 +import importlib.util +import json +import tempfile +import unittest +from pathlib import Path + + +SOURCE = Path(__file__).with_name("platform_infra_gitea_hook_reconciler.py") +SPEC = importlib.util.spec_from_file_location("hook_reconciler", SOURCE) +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + + +class HookRecoveryTest(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.state_path = str(Path(self.temporary.name) / "hook-recovery" / "ledger.json") + self.fixed_seconds = 1_783_840_000 + MODULE.time.time = lambda: self.fixed_seconds + self.delivery = { + "id": 123, + "guid": "delivery-guid", + "event": "push", + "status_code": 502, + "delivered_at": "2026-07-12T06:31:12Z", + } + self.detail = { + **self.delivery, + "repositoryName": "pikasTech/agentrun", + "ref": "refs/heads/v0.2", + "requestedCommit": "f" * 40, + } + self.topology = { + "events": ["push"], + "reconcile": { + "requestTimeoutMs": 15_000, + "failedDeliveryRecovery": { + "enabled": True, + "targetId": "NC01", + "statePath": self.state_path, + "lookbackSeconds": 259_200, + "historyPerHook": 50, + "maxRedeliveriesPerCycle": 4, + "cooldownMs": 300_000, + "maxAttemptsPerDelivery": 3, + "retryStatusCodes": [502, 503, 504], + "stateRetentionSeconds": 1_209_600, + }, + }, + } + self.repository = {"repository": "pikasTech/agentrun"} + self.desired_hook = { + "targetId": "NC01", + "url": "https://example.invalid/_unidesk/github-to-gitea/nc01", + "repoKeys": ["agentrun-nc01-v02"], + "branches": ["v0.2"], + } + self.observed_hook = {"id": 77} + MODULE.list_hook_deliveries = lambda *_args: [dict(self.delivery)] + MODULE.hook_delivery_detail = lambda *_args: dict(self.detail) + + def tearDown(self): + self.temporary.cleanup() + + def test_write_ahead_cooldown_and_source_commit_closeout(self): + ledger = MODULE.load_ledger(self.state_path) + calls = [] + + def request(*args, **_kwargs): + persisted = json.loads(Path(self.state_path).read_text()) + record = next(iter(persisted["records"].values())) + self.assertEqual(record["state"], "attempted") + self.assertEqual(record["attempts"], 1) + calls.append(args[3]) + return 202, None + + MODULE.request = request + budget = {"remaining": 4} + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", ledger, set(), budget, + )) + self.assertEqual(len(calls), 1) + self.assertEqual(budget["remaining"], 3) + + restarted = MODULE.load_ledger(self.state_path) + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", restarted, set(), {"remaining": 4}, + )) + self.assertEqual(len(calls), 1, "restart inside cooldown must not redeliver") + + committed = {("agentrun-nc01-v02", "delivery-guid")} + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", restarted, committed, {"remaining": 4}, + )) + final = next(iter(MODULE.load_ledger(self.state_path)["records"].values())) + self.assertEqual(final["state"], "source-committed") + self.assertEqual(final["completionReason"], "source-committed") + self.assertEqual(len(calls), 1, "committed source must not redeliver") + + def test_restart_preserves_max_attempts_and_scope_mismatch_fails_closed(self): + ledger = MODULE.load_ledger(self.state_path) + key = MODULE.delivery_identity("pikasTech/agentrun", 77, "delivery-guid") + now_ms = int(self.fixed_seconds * 1000) + ledger["records"][key] = { + "repository": "pikasTech/agentrun", + "targetId": "NC01", + "repoKeys": ["agentrun-nc01-v02"], + "hookId": 77, + "deliveryId": "delivery-guid", + "apiDeliveryId": "123", + "state": "attempted", + "attempts": 3, + "cooldownMs": 300_000, + "nextAtEpochMs": now_ms - 1, + "nextAt": MODULE.now_iso(now_ms - 1), + "updatedAtEpochMs": now_ms - 1, + "updatedAt": MODULE.now_iso(now_ms - 1), + } + MODULE.atomic_write_ledger(self.state_path, ledger) + MODULE.request = lambda *_args, **_kwargs: self.fail("maxAttempts must survive restart") + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, + )) + terminal = MODULE.load_ledger(self.state_path)["records"][key] + self.assertEqual(terminal["state"], "terminal-exhausted") + + Path(self.state_path).unlink() + MODULE.hook_delivery_detail = lambda *_args: {**self.detail, "ref": "refs/heads/feature"} + empty = MODULE.load_ledger(self.state_path) + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", empty, set(), {"remaining": 4}, + )) + self.assertEqual(empty["records"], {}) + + def test_same_guid_uses_only_latest_attempt_and_terminal_never_regresses(self): + ledger = MODULE.load_ledger(self.state_path) + key = MODULE.delivery_identity("pikasTech/agentrun", 77, "delivery-guid") + now_ms = int(self.fixed_seconds * 1000) + ledger["records"][key] = { + "repository": "pikasTech/agentrun", + "targetId": "NC01", + "repoKeys": ["agentrun-nc01-v02"], + "hookId": 77, + "hookUrlSha256": MODULE.url_hash(self.desired_hook["url"]), + "deliveryId": "delivery-guid", + "apiDeliveryId": "123", + "state": "attempted", + "attempts": 1, + "cooldownMs": 300_000, + "nextAtEpochMs": now_ms - 1, + "nextAt": MODULE.now_iso(now_ms - 1), + "updatedAtEpochMs": now_ms - 1, + "updatedAt": MODULE.now_iso(now_ms - 1), + } + MODULE.atomic_write_ledger(self.state_path, ledger) + old = dict(self.delivery) + latest = {**self.delivery, "id": 124, "status_code": 200, "delivered_at": "2026-07-12T06:32:12Z"} + MODULE.list_hook_deliveries = lambda *_args: [old, latest] + MODULE.hook_delivery_detail = lambda *_args: { + **(latest if str(_args[-1]) == "124" else old), + "repositoryName": "pikasTech/agentrun", + "ref": "refs/heads/v0.2", + "requestedCommit": "f" * 40, + } + MODULE.request = lambda *_args, **_kwargs: self.fail("latest successful attempt must suppress old failed attempt") + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, + )) + complete = MODULE.load_ledger(self.state_path)["records"][key] + self.assertEqual(complete["state"], "delivery-succeeded-awaiting-source") + self.assertEqual(complete["completionReason"], "github-delivery-succeeded") + + MODULE.list_hook_deliveries = lambda *_args: [old] + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", MODULE.load_ledger(self.state_path), set(), {"remaining": 4}, + )) + self.assertEqual(MODULE.load_ledger(self.state_path)["records"][key]["state"], "delivery-succeeded-awaiting-source") + + committed = {("agentrun-nc01-v02", "delivery-guid")} + self.assertTrue(MODULE.recover_hook_deliveries( + self.topology, self.repository, self.desired_hook, self.observed_hook, + "token", MODULE.load_ledger(self.state_path), committed, {"remaining": 4}, + )) + final = MODULE.load_ledger(self.state_path)["records"][key] + self.assertEqual(final["state"], "source-committed") + self.assertEqual(final["completionReason"], "source-committed") + + def test_removed_hook_pending_record_becomes_terminal_without_post(self): + ledger = MODULE.load_ledger(self.state_path) + key = MODULE.delivery_identity("pikasTech/agentrun", 66, "old-hook-guid") + now_ms = int(self.fixed_seconds * 1000) + ledger["records"][key] = { + "repository": "pikasTech/agentrun", + "targetId": "NC01", + "repoKeys": ["agentrun-nc01-v02"], + "hookId": 66, + "hookUrlSha256": "sha256:old", + "deliveryId": "old-hook-guid", + "apiDeliveryId": "99", + "state": "pending", + "attempts": 0, + "cooldownMs": 300_000, + "nextAtEpochMs": now_ms, + "nextAt": MODULE.now_iso(now_ms), + "updatedAtEpochMs": now_ms, + "updatedAt": MODULE.now_iso(now_ms), + "lastStatusCode": 502, + } + MODULE.atomic_write_ledger(self.state_path, ledger) + MODULE.request = lambda *_args, **_kwargs: self.fail("removed hook must never be redelivered") + MODULE.terminalize_no_longer_desired(ledger, self.topology["reconcile"]["failedDeliveryRecovery"], "pikasTech/agentrun", self.desired_hook, 77) + terminal = MODULE.load_ledger(self.state_path)["records"][key] + self.assertEqual(terminal["state"], "terminal-exhausted") + self.assertEqual(terminal["completionReason"], "hook-no-longer-desired") + + terminal["state"] = "delivery-succeeded-awaiting-source" + terminal["completionReason"] = "github-delivery-succeeded" + reloaded = MODULE.load_ledger(self.state_path) + reloaded["records"][key] = terminal + MODULE.atomic_write_ledger(self.state_path, reloaded) + MODULE.close_committed_ledger_records( + reloaded, + self.topology["reconcile"]["failedDeliveryRecovery"], + {("agentrun-nc01-v02", "old-hook-guid")}, + ) + committed = MODULE.load_ledger(self.state_path)["records"][key] + self.assertEqual(committed["state"], "source-committed") + self.assertEqual(committed["completionReason"], "source-committed") + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/src/platform_infra_gitea_status_evaluator.py b/scripts/src/platform_infra_gitea_status_evaluator.py index 0e3f7f82..0c20a9b2 100644 --- a/scripts/src/platform_infra_gitea_status_evaluator.py +++ b/scripts/src/platform_infra_gitea_status_evaluator.py @@ -69,7 +69,7 @@ def evaluate_repository(repo, hook, head, mirror, bridge_events, inbox_records=N snapshot_ref = mirror.get("snapshotRef") latest_delivery = hook.get("latest") latest_push_delivery = hook.get("latestPush") - selected_push_delivery = hook.get("selectedPush") or latest_push_delivery + selected_push_delivery = hook.get("selectedPush") delivery_id = _delivery_id(selected_push_delivery) delivery_repository = selected_push_delivery.get("repository") if isinstance(selected_push_delivery, dict) else None delivery_ref = selected_push_delivery.get("ref") if isinstance(selected_push_delivery, dict) else None @@ -164,6 +164,7 @@ def evaluate_repository(repo, hook, head, mirror, bridge_events, inbox_records=N "matchingHookCount": hook.get("matchingHookCount"), "matchingHookIds": hook.get("matchingHookIds") or [], "hookTopologyReason": hook.get("hookTopologyReason"), + "hookTopology": hook.get("hookTopology") if isinstance(hook.get("hookTopology"), dict) else {}, "deliveryId": delivery_id, "deliveryRepository": delivery_repository, "deliveryRef": delivery_ref, @@ -204,7 +205,7 @@ def _stale_reason(hook, head, delivery_accepted, delivery_scope_matches, inbox_r if inbox_state != "committed": return f"durable-inbox-{inbox_state or 'unknown'}-not-committed" result = inbox_record.get("result") if isinstance(inbox_record.get("result"), dict) else {} - delivery_requested_commit = (hook.get("selectedPush") or hook.get("latestPush") or {}).get("requestedCommit") + delivery_requested_commit = (hook.get("selectedPush") or {}).get("requestedCommit") inbox_proves_head = ( inbox_record.get("requestedCommit") == delivery_requested_commit and result.get("sourceCommit") == delivery_requested_commit