Merge pull request #1829 from pikasTech/fix/1760-durable-delivery
fix(cicd): 持久协调并自动恢复 GitHub webhook 投递
This commit is contained in:
Lyon
@@ -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
|
||||
|
||||
@@ -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<string, unknown>, path: string): G
|
||||
function parseWebhookSync(record: Record<string, unknown>, 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<string, unknown>, 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`);
|
||||
}
|
||||
|
||||
@@ -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<Record<string, unknown>> } } }).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<Record<string, unknown>>)[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<string, unknown> }>)[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<string, unknown> {
|
||||
@@ -71,6 +93,16 @@ function fixture(): Record<string, unknown> {
|
||||
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<string, unknown> {
|
||||
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<string, unknown>): Record<stri
|
||||
staleReason: null,
|
||||
deliveryId: delivery.deliveryId,
|
||||
deliveryAccepted: true,
|
||||
selectedPushDelivery: { deliveryId: delivery.deliveryId, hookId: 1, statusCode: 202, status: "OK", deliveredAt: "2026-07-11T00:00:00Z" },
|
||||
latestPushDelivery: { deliveryId: delivery.deliveryId, hookId: 1, statusCode: 202, status: "OK", deliveredAt: "2026-07-11T00:00:00Z" },
|
||||
hookTopology: {
|
||||
ready: true,
|
||||
desiredUrlSha256: "sha256:desired",
|
||||
desiredTopologyUrlSha256: ["sha256:desired"],
|
||||
observedManagedUrlSha256: ["sha256:desired"],
|
||||
driftTypes: [],
|
||||
},
|
||||
durableInboxState: "committed",
|
||||
durableInboxCommitted: true,
|
||||
durableInboxRecord: delivery,
|
||||
|
||||
@@ -27,9 +27,12 @@ export function buildMirrorWebhookStatusDisclosure(
|
||||
const targetId = stringValue(target.id, "<node>");
|
||||
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<string, unknown>): Record<string, unknown> {
|
||||
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<string, unknown>): Record<string, unknown> {
|
||||
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<string, unknown>): Record<string,
|
||||
authorityMatches: repo.authorityMatches === true,
|
||||
stale: repo.bridgeEventStale === true,
|
||||
staleReason: repo.staleReason ?? null,
|
||||
deliveryId: repo.deliveryId,
|
||||
deliveryId: selected.deliveryId ?? null,
|
||||
deliveryAccepted: repo.deliveryAccepted === true,
|
||||
selectedDelivery: {
|
||||
deliveryId: selected.deliveryId ?? null,
|
||||
hookId: selected.hookId ?? null,
|
||||
status: selected.status ?? null,
|
||||
statusCode: selected.statusCode ?? null,
|
||||
deliveredAt: selected.deliveredAt ?? null,
|
||||
accepted: repo.deliveryAccepted === true,
|
||||
},
|
||||
hookTopology: {
|
||||
ready: topology.ready === true,
|
||||
desiredUrlSha256: topology.desiredUrlSha256 ?? null,
|
||||
desiredTopologyUrlSha256: Array.isArray(topology.desiredTopologyUrlSha256) ? topology.desiredTopologyUrlSha256 : [],
|
||||
observedManagedUrlSha256: Array.isArray(topology.observedManagedUrlSha256) ? topology.observedManagedUrlSha256 : [],
|
||||
driftTypes: Array.isArray(topology.driftTypes) ? topology.driftTypes : [],
|
||||
},
|
||||
durableState: repo.durableInboxState,
|
||||
durableCommitted: repo.durableInboxCommitted === true,
|
||||
error: compactErrorValue(repo.error),
|
||||
|
||||
@@ -420,6 +420,7 @@ export function createDurableSyncController(options) {
|
||||
journal: storageReady ? store.journalMetadata() : null,
|
||||
counts,
|
||||
deliveries: statusRecords(records).map(publicInboxRecord),
|
||||
failedDeliveryRecovery: await readFailedDeliveryRecoveryStatus(options.failedDeliveryRecoveryStatePath),
|
||||
};
|
||||
}
|
||||
|
||||
@@ -946,6 +947,7 @@ export function loadRuntimeConfig() {
|
||||
committedRetentionMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_COMMITTED_RETENTION_SECONDS") * 1000,
|
||||
cleanupIntervalMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_CLEANUP_INTERVAL_MS"),
|
||||
},
|
||||
failedDeliveryRecoveryStatePath: process.env.UNIDESK_GITEA_HOOK_RECOVERY_STATE_PATH || null,
|
||||
workerRetry: {
|
||||
maxAttempts: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_MAX_ATTEMPTS"),
|
||||
initialDelayMs: positiveIntegerEnv("UNIDESK_GITEA_WEBHOOK_RETRY_INITIAL_DELAY_MS"),
|
||||
@@ -964,6 +966,51 @@ export function loadRuntimeConfig() {
|
||||
};
|
||||
}
|
||||
|
||||
export async function readFailedDeliveryRecoveryStatus(statePath) {
|
||||
if (!statePath) return { enabled: false, ready: true, state: "disabled", counts: {}, entries: [] };
|
||||
let ledger;
|
||||
try {
|
||||
ledger = JSON.parse(await readFile(statePath, "utf8"));
|
||||
} catch (error) {
|
||||
if (error?.code === "ENOENT") return { enabled: true, ready: false, state: "not-initialized", counts: {}, entries: [] };
|
||||
return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-unreadable", counts: {}, entries: [] };
|
||||
}
|
||||
if (ledger?.version !== 1 || ledger?.kind !== "GiteaGithubHookRecoveryLedger" || !ledger.records || typeof ledger.records !== "object" || Array.isArray(ledger.records)) {
|
||||
return { enabled: true, ready: false, state: "invalid-or-unreadable", errorType: "hook-recovery-ledger-invalid", counts: {}, entries: [] };
|
||||
}
|
||||
const allowedStates = new Set(["pending", "attempted", "delivery-succeeded-awaiting-source", "source-committed", "terminal-exhausted"]);
|
||||
const records = Object.values(ledger.records);
|
||||
const validRecord = (item) => 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 }));
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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");
|
||||
});
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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 },
|
||||
];
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -102,9 +102,11 @@ export function renderMirrorWebhookApply(result: Record<string, unknown>): Rende
|
||||
|
||||
export function renderMirrorWebhookStatus(result: Record<string, unknown>, disclosure: Record<string, unknown>): 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<string, unknown>, 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<string, unknown>, 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<string, unknown>, 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),
|
||||
"",
|
||||
|
||||
@@ -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: {
|
||||
|
||||
@@ -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<string, unknown> {
|
||||
const sync = gitea.sourceAuthority.webhookSync;
|
||||
const base = gitea.app.publicExposure.publicBaseUrl.replace(/\/+$/u, "");
|
||||
const desiredByRepository = new Map<string, Map<string, { targetId: string; url: string; repoKeys: Set<string>; branches: Set<string> }>>();
|
||||
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<string>(),
|
||||
branches: new Set<string>(),
|
||||
};
|
||||
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<string, unknown>): Record<string, unknown> {
|
||||
return {
|
||||
ready: payload.ready === true,
|
||||
@@ -1398,6 +1518,8 @@ function mirrorReadOnlyNextCommands(targetId: string): Record<string, string> {
|
||||
|
||||
function webhookSyncSummary(gitea: GiteaConfig, target: GiteaTarget): Record<string, unknown> {
|
||||
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<str
|
||||
return {
|
||||
enabled: sync.enabled,
|
||||
direction: sync.direction,
|
||||
publicUrl: githubWebhookPublicUrl(gitea, target),
|
||||
publicUrl: targetUrl,
|
||||
hookReconcile: {
|
||||
...sync.hookReconcile,
|
||||
runningOnTarget: sync.hookReconcile.enabled && sync.hookReconcile.ownerTargetId.toLowerCase() === target.id.toLowerCase(),
|
||||
desiredUrlSha256: `sha256:${createHash("sha256").update(targetUrl).digest("hex").slice(0, 16)}`,
|
||||
topologySha256: `sha256:${createHash("sha256").update(JSON.stringify(topology)).digest("hex").slice(0, 16)}`,
|
||||
},
|
||||
events: sync.events,
|
||||
responseBudgetMs: sync.responseBudgetMs,
|
||||
ingressAttemptBudgetMs,
|
||||
|
||||
@@ -0,0 +1,690 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Single-writer GitHub webhook topology reconciler for the Gitea source authority."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
API_BASE = "https://api.github.com"
|
||||
STOP = threading.Event()
|
||||
|
||||
|
||||
class ApiError(RuntimeError):
|
||||
def __init__(self, method, path, status):
|
||||
super().__init__(f"github-api-{method.lower()}-{status}")
|
||||
self.status = status
|
||||
self.path = path
|
||||
|
||||
|
||||
def url_hash(value):
|
||||
return "sha256:" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
|
||||
|
||||
|
||||
def emit(event, **fields):
|
||||
print(json.dumps({"event": event, **fields, "valuesPrinted": False}, ensure_ascii=False, separators=(",", ":")), flush=True)
|
||||
|
||||
|
||||
def load_topology(path):
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
topology = json.load(handle)
|
||||
if topology.get("version") != 1:
|
||||
raise ValueError("hook topology version must be 1")
|
||||
prefix = topology.get("managedUrlPrefix")
|
||||
if not isinstance(prefix, str) or not prefix.startswith("https://"):
|
||||
raise ValueError("managedUrlPrefix must be https")
|
||||
repositories = topology.get("repositories")
|
||||
if not isinstance(repositories, list) or not repositories:
|
||||
raise ValueError("hook topology repositories must not be empty")
|
||||
for item in repositories:
|
||||
if not isinstance(item, dict) or not isinstance(item.get("repository"), str):
|
||||
raise ValueError("hook topology repository is invalid")
|
||||
urls = item.get("desiredUrls")
|
||||
if not isinstance(urls, list) or not urls or any(not isinstance(url, str) or not url.startswith("https://") for url in urls):
|
||||
raise ValueError("hook topology desiredUrls must contain https URLs")
|
||||
if len(urls) != len(set(urls)):
|
||||
raise ValueError("hook topology desiredUrls must be unique per repository")
|
||||
hooks = item.get("desiredHooks")
|
||||
if not isinstance(hooks, list) or not hooks:
|
||||
raise ValueError("hook topology desiredHooks must not be empty")
|
||||
reconcile = topology.get("reconcile") if isinstance(topology.get("reconcile"), dict) else {}
|
||||
recovery = reconcile.get("failedDeliveryRecovery") if isinstance(reconcile.get("failedDeliveryRecovery"), dict) else {}
|
||||
if recovery.get("enabled") is True:
|
||||
if recovery.get("targetId") != topology.get("ownerTargetId"):
|
||||
raise ValueError("failed delivery recovery must be scoped to the owner target")
|
||||
if not isinstance(recovery.get("statePath"), str) or not recovery["statePath"].startswith("/"):
|
||||
raise ValueError("failed delivery recovery statePath must be absolute")
|
||||
return topology
|
||||
|
||||
|
||||
def request(token, timeout_seconds, method, api_path, payload=None, expected=(200, 201, 204), tolerate=()):
|
||||
data = None if payload is None else json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(API_BASE + api_path, data=data, method=method)
|
||||
req.add_header("Authorization", "Bearer " + token)
|
||||
req.add_header("Accept", "application/vnd.github+json")
|
||||
req.add_header("X-GitHub-Api-Version", "2022-11-28")
|
||||
if payload is not None:
|
||||
req.add_header("Content-Type", "application/json")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=timeout_seconds) as response:
|
||||
body = response.read().decode("utf-8", errors="replace")
|
||||
if response.status not in expected:
|
||||
raise ApiError(method, api_path, response.status)
|
||||
return response.status, json.loads(body) if body else None
|
||||
except urllib.error.HTTPError as error:
|
||||
error.read()
|
||||
if error.code in tolerate:
|
||||
return error.code, None
|
||||
raise ApiError(method, api_path, error.code) from error
|
||||
|
||||
|
||||
def list_hooks(token, timeout_seconds, max_pages, repository):
|
||||
hooks = []
|
||||
encoded = urllib.parse.quote(repository, safe="/")
|
||||
for page in range(1, max_pages + 1):
|
||||
_, rows = request(token, timeout_seconds, "GET", f"/repos/{encoded}/hooks?per_page=100&page={page}")
|
||||
if not isinstance(rows, list):
|
||||
raise RuntimeError("github-hooks-response-invalid")
|
||||
hooks.extend(item for item in rows if isinstance(item, dict))
|
||||
if len(rows) < 100:
|
||||
return hooks
|
||||
raise RuntimeError("github-hooks-pagination-bound-exhausted")
|
||||
|
||||
|
||||
def now_iso(epoch_ms=None):
|
||||
value = time.time() if epoch_ms is None else epoch_ms / 1000
|
||||
return datetime.fromtimestamp(value, tz=timezone.utc).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def parse_epoch_ms(value):
|
||||
if not isinstance(value, str) or not value:
|
||||
return 0
|
||||
try:
|
||||
return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
|
||||
except ValueError:
|
||||
return 0
|
||||
|
||||
|
||||
def atomic_write_ledger(path, ledger):
|
||||
directory = os.path.dirname(path)
|
||||
os.makedirs(directory, mode=0o700, exist_ok=True)
|
||||
temporary = os.path.join(directory, f".{os.path.basename(path)}.{os.getpid()}.{time.time_ns()}.tmp")
|
||||
payload = json.dumps(ledger, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n"
|
||||
try:
|
||||
with open(temporary, "x", encoding="utf-8") as handle:
|
||||
os.chmod(temporary, 0o600)
|
||||
handle.write(payload)
|
||||
handle.flush()
|
||||
os.fsync(handle.fileno())
|
||||
os.replace(temporary, path)
|
||||
directory_fd = os.open(directory, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
|
||||
try:
|
||||
os.fsync(directory_fd)
|
||||
finally:
|
||||
os.close(directory_fd)
|
||||
finally:
|
||||
try:
|
||||
os.unlink(temporary)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
|
||||
def load_ledger(path):
|
||||
try:
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
ledger = json.load(handle)
|
||||
except FileNotFoundError:
|
||||
ledger = {"version": 1, "kind": "GiteaGithubHookRecoveryLedger", "createdAt": now_iso(), "records": {}}
|
||||
atomic_write_ledger(path, ledger)
|
||||
return ledger
|
||||
if ledger.get("version") != 1 or ledger.get("kind") != "GiteaGithubHookRecoveryLedger" or not isinstance(ledger.get("records"), dict):
|
||||
raise ValueError("hook recovery ledger is invalid")
|
||||
allowed_states = {"pending", "attempted", "delivery-succeeded-awaiting-source", "source-committed", "terminal-exhausted"}
|
||||
for record in ledger["records"].values():
|
||||
if not isinstance(record, dict) or record.get("state") not in allowed_states:
|
||||
raise ValueError("hook recovery ledger record state is invalid")
|
||||
if not isinstance(record.get("repository"), str) or not record["repository"] or not isinstance(record.get("deliveryId"), str) or not record["deliveryId"]:
|
||||
raise ValueError("hook recovery ledger record identity is invalid")
|
||||
if not isinstance(record.get("attempts"), int) or record["attempts"] < 0 or not isinstance(record.get("cooldownMs"), int) or record["cooldownMs"] < 0:
|
||||
raise ValueError("hook recovery ledger record counters are invalid")
|
||||
return ledger
|
||||
|
||||
|
||||
def cleanup_ledger(ledger, retention_seconds, now_ms):
|
||||
cutoff = now_ms - retention_seconds * 1000
|
||||
terminal = {"source-committed", "terminal-exhausted"}
|
||||
stale = [
|
||||
key for key, record in ledger["records"].items()
|
||||
if isinstance(record, dict) and record.get("state") in terminal and int(record.get("updatedAtEpochMs") or 0) < cutoff
|
||||
]
|
||||
for key in stale:
|
||||
del ledger["records"][key]
|
||||
return len(stale)
|
||||
|
||||
|
||||
def hook_url(hook):
|
||||
config = hook.get("config") if isinstance(hook.get("config"), dict) else {}
|
||||
return config.get("url") if isinstance(config.get("url"), str) else ""
|
||||
|
||||
|
||||
def is_managed_url(value, prefix):
|
||||
return value == prefix or value.startswith(prefix.rstrip("/") + "/")
|
||||
|
||||
|
||||
def hook_config_matches(hook, events):
|
||||
config = hook.get("config") if isinstance(hook.get("config"), dict) else {}
|
||||
return bool(
|
||||
hook.get("name") == "web"
|
||||
and hook.get("active") is True
|
||||
and set(hook.get("events") or []) == set(events)
|
||||
and config.get("content_type") == "json"
|
||||
and str(config.get("insecure_ssl", "0")) == "0"
|
||||
)
|
||||
|
||||
|
||||
def topology_observation(hooks, desired_urls, managed_prefix, events):
|
||||
managed = [item for item in hooks if item.get("name") == "web" and is_managed_url(hook_url(item), managed_prefix)]
|
||||
desired_set = set(desired_urls)
|
||||
drift_types = []
|
||||
for desired_url in desired_urls:
|
||||
matches = [item for item in managed if hook_url(item) == desired_url]
|
||||
if not matches:
|
||||
drift_types.append("desired-url-missing")
|
||||
elif len(matches) > 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.json>")
|
||||
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()
|
||||
@@ -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()
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user