Merge pull request 'gitea announces from the forge's events, not a poll: an edited base and a close are announced at once, one pass at start (issue 1422)' (#539) from fix/1422-gitea-events into main
This commit was merged in pull request #539.
This commit is contained in:
+66
-65
@@ -53,8 +53,9 @@ import { announce, keepWatermark, readWatermark } from "./announce.js";
|
||||
import { mergeTool } from "./merge.js";
|
||||
import { GiteaAdmin, GiteaClient, movedSince } from "./client.js";
|
||||
import { keepMainRules, mainRulesStep } from "./mainrules.js";
|
||||
import { CHECK_CONTEXT, CHECK_STATUS, REPO_CHECK_CONTEXT, commentFor, composedGate, groupGateStatus, headKey, headsToAnnounce, maintained,
|
||||
requiredOf, statusesFor, writesGate, type Announced, type Checked, repoCheckFacts, reposToLook } from "./pulls.js";
|
||||
import { CHECK_CONTEXT, CHECK_STATUS, REPO_CHECK_CONTEXT, commentFor, composedGate, groupGateStatus, headKey, maintained,
|
||||
requiredOf, statusesFor, writesGate, type Announced, type Checked, repoCheckFacts } from "./pulls.js";
|
||||
import { actOn, announceNewHeads, startPass } from "./watch.js";
|
||||
import { GateWriter } from "./gate.js";
|
||||
import { HOOK_PATH, HOOK_PORT, Pace, claimHead, ensureSystemHook, eventsTool, keptSecret, receiver, type ForgeEvent } from "./webhook.js";
|
||||
import { keepSiblings, noteComposed, noteVerdict, readSiblings, siblingsFile, siblingsOf, weighSiblings } from "./siblings.js";
|
||||
@@ -138,12 +139,7 @@ async function pollMerged(client: GiteaClient): Promise<void> {
|
||||
// Closed unmerged (novox/hq ADR 0239): announced once, since the watching began, so its delivery stops.
|
||||
const closedKey = `closed:${repo.full_name}#${pull.number}`;
|
||||
if (!pull.merged && pull.state === "closed" && !announced.has(closedKey)) {
|
||||
if (primedMerges && !!pull.updated_at && !!since && pull.updated_at > since) {
|
||||
await announce("pull.closed", { owner: repo.owner, repo: repo.name, number: pull.number, title: pull.title,
|
||||
head: pull.head, head_sha: pull.head_sha, base: pull.base, html_url: pull.html_url });
|
||||
console.log(`[gitea] announced ${repo.full_name}#${pull.number} closed unmerged`);
|
||||
await siblingMoves(client, repo.full_name, pull.number);
|
||||
}
|
||||
if (primedMerges && !!pull.updated_at && !!since && pull.updated_at > since) await announceClosed(client, repo, pull);
|
||||
announced.add(closedKey);
|
||||
continue;
|
||||
}
|
||||
@@ -168,6 +164,23 @@ async function pollMerged(client: GiteaClient): Promise<void> {
|
||||
type RepoOf = { owner: string; name: string; full_name: string; clone_url?: string };
|
||||
type PullOf = Awaited<ReturnType<GiteaClient["listPullRequests"]>>[number];
|
||||
|
||||
/** A pull request closed without a merge, announced once (ADR 0239): by its event, or by the start's pass. */
|
||||
async function announceClosed(client: GiteaClient, repo: RepoOf, pull: PullOf): Promise<boolean> {
|
||||
const closedKey = `closed:${repo.full_name}#${pull.number}`;
|
||||
if (pull.merged || pull.state !== "closed" || announced.has(closedKey)) return false;
|
||||
announced.add(closedKey);
|
||||
try {
|
||||
await announce("pull.closed", { owner: repo.owner, repo: repo.name, number: pull.number, title: pull.title,
|
||||
head: pull.head, head_sha: pull.head_sha, base: pull.base, html_url: pull.html_url });
|
||||
} catch (err) {
|
||||
announced.delete(closedKey);
|
||||
throw err;
|
||||
}
|
||||
console.log(`[gitea] announced ${repo.full_name}#${pull.number} closed unmerged`);
|
||||
await siblingMoves(client, repo.full_name, pull.number);
|
||||
return true;
|
||||
}
|
||||
|
||||
/** **A merge, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */
|
||||
async function announceMerge(client: GiteaClient, repo: RepoOf, pull: PullOf): Promise<boolean> {
|
||||
if (!pull.merged || !pull.merge_commit_sha || announced.has(pull.merge_commit_sha)) return false;
|
||||
@@ -281,43 +294,18 @@ async function alreadyChecked(client: GiteaClient, repo: RepoOf, pull: OpenOf):
|
||||
}
|
||||
|
||||
let lastPullLook = "";
|
||||
// **A sweep now and then looks at every repository** (novox/hq issue 347), whatever moved: the floor is a
|
||||
// heuristic, and a heuristic that misses is caught by the next sweep rather than never.
|
||||
const SWEEP_MS = 10 * 60_000;
|
||||
let lastSweep = 0;
|
||||
async function pollPulls(client: GiteaClient): Promise<void> {
|
||||
/** The pass over every open pull request: run at the start and when the receiver listens, never on a timer (issue 1422). */
|
||||
async function passPulls(client: GiteaClient): Promise<void> {
|
||||
const began = new Date().toISOString();
|
||||
const floor = lastPullLook ? new Date(Date.parse(lastPullLook) - MARGIN_MS).toISOString() : "";
|
||||
const dayAgo = new Date(Date.now() - 24 * 3600_000).toISOString();
|
||||
// A pull request opened, reopened or retargeted moves its own update time and not its repository's: the
|
||||
// pull requests updated since the floor are found by their own (issue 347). A search that fails is a sweep.
|
||||
let sweep = Date.now() - lastSweep >= SWEEP_MS;
|
||||
let updated: { owner: string; name: string; updated_at?: string }[] = [];
|
||||
if (floor && !sweep) {
|
||||
try {
|
||||
updated = await client.searchOpenPulls(floor);
|
||||
} catch (err) {
|
||||
console.error(`[gitea] the pull requests updated lately could not be searched, so every repository is looked at — ${err instanceof Error ? err.message : err}`);
|
||||
sweep = true;
|
||||
}
|
||||
}
|
||||
if (sweep) lastSweep = Date.now();
|
||||
for (const repo of reposToLook(await client.listAllRepos(), floor, updated, sweep)) {
|
||||
// Every open pull request updated since the floor, paged (novox/hq issue 579): the newest 20 alone left a head
|
||||
// beyond them unannounced. A sweep and the first pass read every page.
|
||||
const open = await client.listOpenPullsSince(repo.owner, repo.name, sweep ? "" : floor);
|
||||
for (const pull of headsToAnnounce(repo.full_name, open, heads)) {
|
||||
const key = headKey(repo.full_name, pull.number, pull.base);
|
||||
const fresh = primedPulls || (!!pull.updated_at && pull.updated_at > dayAgo);
|
||||
// The first look after a start remembers nothing: a head the forge already shows a merge check of was announced.
|
||||
if (fresh && !primedPulls && await alreadyChecked(client, repo, pull)) {
|
||||
heads[key] = String(pull.head_sha);
|
||||
continue;
|
||||
}
|
||||
if (fresh) {
|
||||
if (await announceHead(client, repo, pull)) pace.miss(`${key} at ${String(pull.head_sha).slice(0, 8)}`);
|
||||
} else heads[key] = String(pull.head_sha);
|
||||
}
|
||||
for (const repo of await client.listAllRepos()) {
|
||||
// Every open pull request, paged (novox/hq issue 579).
|
||||
const open = await client.listOpenPullsSince(repo.owner, repo.name, "");
|
||||
await announceNewHeads(repo.full_name, open, heads, primedPulls, dayAgo, {
|
||||
alreadyChecked: (pull) => alreadyChecked(client, repo, pull),
|
||||
announceHead: (pull) => announceHead(client, repo, pull),
|
||||
missed: (what) => pace.miss(what),
|
||||
});
|
||||
if (await trunksMoved(client, repo.owner, repo.name)) changedTrunks = true;
|
||||
}
|
||||
if (changedTrunks) keepTrunks();
|
||||
@@ -497,6 +485,9 @@ const checks = new CheckStatuses({
|
||||
/** How fast the polls look: 30 s until the forge's events are heard, five minutes while they are; and what
|
||||
* gitea_events says of them (webhook.ts). */
|
||||
const pace = new Pace();
|
||||
/** Asks the pass over every repository again, shortly and once (set where the pass is made): an event that could not be
|
||||
* acted on is not lost until the next start. */
|
||||
let catchUp: () => void = () => {};
|
||||
if (!gitea) pace.unheard("the forge is not watched: no way to a token", false);
|
||||
|
||||
// The states this module keeps and which others may read, said on the bus before it serves (novox/hq ADR 0495 clause 2).
|
||||
@@ -522,22 +513,15 @@ registerModuleTools("gitea", () => [eventsTool(pace), ...readAllTools(readAll),
|
||||
gates,
|
||||
}), mergeTool(gitea, { gates })] : [])]);
|
||||
|
||||
/** What the forge's event asks of the announcer at once (issue 729): a new head is announced, and a merge. */
|
||||
async function actOn(client: GiteaClient, event: ForgeEvent): Promise<void> {
|
||||
const b = event.body as { owner?: string; repo?: string; number?: number };
|
||||
if (!b.owner || !b.repo || !b.number) return;
|
||||
if (event.name === "forge.pull-request.closed") {
|
||||
// Closed unmerged is a sibling's move; a merge is weighed with its announcement below (ADR 0382 §5).
|
||||
if (!(event.body as { merged?: boolean }).merged) await siblingMoves(client, `${b.owner}/${b.repo}`, b.number);
|
||||
return;
|
||||
}
|
||||
const head = ["forge.pull-request.opened", "forge.pull-request.synchronized", "forge.pull-request.reopened"].includes(event.name);
|
||||
const merged = event.name === "forge.pull-request.merged";
|
||||
if (!head && !merged) return;
|
||||
const [repo, pull] = await Promise.all([client.getRepo(b.owner, b.repo), client.getPullRequest(b.owner, b.repo, b.number)]);
|
||||
if (head && pull.state === "open") await announceHead(client, repo, pull);
|
||||
if (merged) await announceMerge(client, repo, pull);
|
||||
}
|
||||
/** What the forge's event asks of the announcer at once (issue 729, 1422): see watch.ts. */
|
||||
const actOnEvent = (client: GiteaClient, event: ForgeEvent): Promise<void> => actOn<RepoOf, PullOf>(event, {
|
||||
getRepo: (o, r) => client.getRepo(o, r),
|
||||
getPull: (o, r, n) => client.getPullRequest(o, r, n) as Promise<PullOf & { state?: string; merged?: boolean }>,
|
||||
announceHead: (repo, pull) => announceHead(client, repo, pull as OpenOf),
|
||||
announceMerge: (repo, pull) => announceMerge(client, repo, pull),
|
||||
announceClosed: (repo, pull) => announceClosed(client, repo, pull),
|
||||
siblingMoves: (full, n) => siblingMoves(client, full, n),
|
||||
});
|
||||
|
||||
/**
|
||||
* **The receiver, and the forge's one system webhook pointing at it** (issue 729). The secret is minted and kept beside
|
||||
@@ -563,7 +547,12 @@ async function listenForForge(client: GiteaClient): Promise<boolean> {
|
||||
// A repository made, or one not yet seen readable, is granted read before anything acts on it (readall.ts).
|
||||
await Promise.resolve().then(() => readAll().onEvent(event)).catch((err) =>
|
||||
console.error(`[gitea] whether the mesh's accounts read the repository of ${event.name} could not be settled — ${err instanceof Error ? err.message : err}`));
|
||||
await actOn(client, event);
|
||||
try {
|
||||
await actOnEvent(client, event);
|
||||
} catch (err) {
|
||||
catchUp();
|
||||
throw err;
|
||||
}
|
||||
} });
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
@@ -612,9 +601,9 @@ if (gitea) {
|
||||
// polls slow to every five minutes and announce only what no event said (Pace); otherwise they keep 30 s.
|
||||
// **Not awaited** (novox/hq issue 1726): the receiver's set-up asks the forge's admin API, and the runtime kills a bundle
|
||||
// whose import has not ended in its 10 s handshake and never starts it again; a forge slow to answer while every bundle
|
||||
// of the node starts left gitea_commit_status, gitea_merge_pull_request and gitea_events served nowhere. Until it is
|
||||
// listening, Pace keeps the polls at 30 s, so nothing is missed meanwhile; listenForForge says its own failure.
|
||||
void listenForForge(client);
|
||||
// of the node starts left gitea_commit_status, gitea_merge_pull_request and gitea_events served nowhere. The pass below
|
||||
// runs again once it is listening, so nothing said before is missed; listenForForge says its own failure.
|
||||
const listenOk = listenForForge(client);
|
||||
// The start's one pass over every repository (readall.ts), not awaited for the same reason. A forge not answering yet
|
||||
// is asked again, a minute apart, until one pass has run; after that only the forge's events grant.
|
||||
// An account of the list the forge does not hold yet (the forge-account provider makes nox-builder and nox-agent when
|
||||
@@ -646,8 +635,20 @@ if (gitea) {
|
||||
// Only while the setting keep-main-rules is true (mainrules.ts); false, the default, leaves every repository's rule as set.
|
||||
tick(mainRulesStep(keepMainRules(), () => GiteaAdmin.fromEnv().ensureMainMerges("novox"), (line) => console.log(line)), 600_000);
|
||||
tick(() => pollRepos(client), 60_000);
|
||||
tick(() => pollMerged(client), () => pace.every());
|
||||
tick(() => pollPulls(client), () => pace.every());
|
||||
// **No timer asks the forge for pull requests or merges** (issue 1422): the forge's events announce them. One pass over
|
||||
// every repository runs at the start and again once the receiver listens, so what happened while the module was away or
|
||||
// not yet listening is announced; passes run one at a time, a failed one is asked again a minute later.
|
||||
let passes: Promise<void> = Promise.resolve();
|
||||
const pass = (): Promise<void> => (passes = passes.catch(() => {}).then(() =>
|
||||
startPass(async () => { await pollMerged(client); await passPulls(client); }, { retryMs: 60_000 })));
|
||||
void pass();
|
||||
let catchUpAsked = false;
|
||||
catchUp = () => {
|
||||
if (catchUpAsked) return;
|
||||
catchUpAsked = true;
|
||||
setTimeout(() => { catchUpAsked = false; void pass(); }, 30_000);
|
||||
};
|
||||
void listenOk.then((ok) => (ok ? pass() : undefined));
|
||||
// Bound again until they hold, and not awaited (bind.ts, issue 1793): the runtime kills a bundle that has not answered its
|
||||
// handshake in 10 s, so the import must end while a refused bind is still asked again.
|
||||
bindInBackground([
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { mock, test } from "node:test";
|
||||
import type { GiteaPull } from "../client.ts";
|
||||
|
||||
// **No poll of the forge** (issue 1422, notes 10 and 11): the forge's events announce heads and merges; a pass runs once at
|
||||
// the start and when the receiver listens.
|
||||
|
||||
const edited = (base: string) => ({ name: "forge.pull-request.edited", body: { owner: "novox", repo: "hq", number: 7, base, merged: false } });
|
||||
|
||||
function deps() {
|
||||
const calls: string[] = [];
|
||||
return { calls, d: {
|
||||
getRepo: async () => ({ r: 1 }),
|
||||
getPull: async () => ({ state: "open", base: "trunk" }) as any,
|
||||
announceHead: async () => { calls.push("head"); },
|
||||
announceMerge: async () => { calls.push("merge"); },
|
||||
announceClosed: async () => { calls.push("closed"); },
|
||||
siblingMoves: async () => { calls.push("siblings"); },
|
||||
} };
|
||||
}
|
||||
|
||||
test("an edited event announces the head into the base it now has, without a poll", async () => {
|
||||
const { actOn } = await import("../watch.ts");
|
||||
const { calls, d } = deps();
|
||||
await actOn(edited("trunk"), d);
|
||||
assert.deepEqual(calls, ["head"]);
|
||||
});
|
||||
|
||||
test("a close without a merge is announced by its event, a merged close is left to the merge event", async () => {
|
||||
const { actOn } = await import("../watch.ts");
|
||||
const { calls, d } = deps();
|
||||
await actOn({ name: "forge.pull-request.closed", body: { owner: "novox", repo: "hq", number: 7, merged: false } }, d);
|
||||
assert.deepEqual(calls, ["closed", "siblings"]);
|
||||
calls.length = 0;
|
||||
await actOn({ name: "forge.pull-request.closed", body: { owner: "novox", repo: "hq", number: 7, merged: true } }, d);
|
||||
assert.deepEqual(calls, []);
|
||||
});
|
||||
|
||||
test("no timer polls after the start: one pass ends and nothing is scheduled; a failed one is asked again", async () => {
|
||||
const { startPass } = await import("../watch.ts");
|
||||
mock.timers.enable({ apis: ["setTimeout", "setInterval"] });
|
||||
try {
|
||||
let runs = 0;
|
||||
await startPass(async () => { runs++; }, { retryMs: 60_000, log: () => {} });
|
||||
mock.timers.tick(24 * 3600_000);
|
||||
assert.equal(runs, 1, "a day later the forge was not asked again");
|
||||
let tries = 0;
|
||||
const retried = startPass(async () => { if (++tries < 2) throw new Error("forge down"); }, { retryMs: 60_000, log: () => {} });
|
||||
await Promise.resolve(); await Promise.resolve();
|
||||
mock.timers.tick(60_000);
|
||||
await retried;
|
||||
assert.equal(tries, 2);
|
||||
} finally {
|
||||
mock.timers.reset();
|
||||
}
|
||||
});
|
||||
|
||||
test("a start pass announces a head nobody announced, and only remembers one the forge shows a check of", async () => {
|
||||
const { announceNewHeads } = await import("../watch.ts");
|
||||
const pull = (n: number, sha: string): GiteaPull => ({ number: n, state: "open", head_sha: sha, base: "main",
|
||||
updated_at: new Date().toISOString() } as unknown as GiteaPull);
|
||||
const announced: number[] = [], said: string[] = [];
|
||||
const heads: Record<string, string> = {};
|
||||
await announceNewHeads("novox/hq", [pull(1, "a".repeat(40)), pull(2, "b".repeat(40))], heads, false,
|
||||
new Date(Date.now() - 24 * 3600_000).toISOString(), {
|
||||
alreadyChecked: async (p) => p.number === 2,
|
||||
announceHead: async (p) => { announced.push(p.number); return true; },
|
||||
missed: (w) => said.push(w),
|
||||
});
|
||||
assert.deepEqual(announced, [1]);
|
||||
assert.equal(heads["novox/hq#2@main"], "b".repeat(40), "the checked head is remembered, not announced");
|
||||
assert.equal(said.length, 1);
|
||||
});
|
||||
@@ -251,7 +251,8 @@ test("a forge that never delivers is said loudly once, and unhealthy, when the p
|
||||
assert.equal(r.healthy, true);
|
||||
assert.equal(r.verified.count, 1);
|
||||
assert.equal(r.verified.last.event, "push");
|
||||
assert.equal(r.polls.every_seconds, 300);
|
||||
assert.equal(r.polls.every_seconds, null);
|
||||
assert.equal(r.polls.hearing, true);
|
||||
assert.match(said.join("\n"), /heard again/);
|
||||
});
|
||||
|
||||
@@ -314,24 +315,19 @@ test("without a receiver by configuration the polls alone watch, healthy; a rece
|
||||
assert.match(r.why, /403/);
|
||||
});
|
||||
|
||||
test("the polls look every 30 s until events are heard, every five minutes while they are, and 30 s again on a miss", async () => {
|
||||
test("nothing is paced: the report says no timer asks the forge, and a miss is still counted and said", async () => {
|
||||
const { Pace } = await import("../webhook.ts");
|
||||
let now = 1_000_000;
|
||||
const said: string[] = [];
|
||||
const pace = new Pace(() => now, (l) => said.push(l));
|
||||
assert.equal(pace.every(), 30_000, "no delivery yet");
|
||||
pace.heard();
|
||||
assert.equal(pace.every(), 300_000, "a delivery heard");
|
||||
now += 11 * 60_000;
|
||||
assert.equal(pace.every(), 30_000, "ten minutes without a delivery");
|
||||
assert.equal((pace as any).every, undefined, "no pace to ask");
|
||||
assert.equal((pace.report() as any).polls.every_seconds, null);
|
||||
pace.heard();
|
||||
assert.equal((pace.report() as any).polls.hearing, true);
|
||||
now += 1000;
|
||||
pace.miss("novox/hq#7 at aaaaaaaa");
|
||||
assert.equal(pace.every(), 30_000, "the poll found what no event said");
|
||||
assert.equal((pace.report() as any).polls.hearing, false, "the pass found what no event said");
|
||||
assert.match(said.join("\n"), /MISSED novox\/hq#7/);
|
||||
now += 1000;
|
||||
pace.heard();
|
||||
assert.equal(pace.every(), 300_000, "the next delivery");
|
||||
});
|
||||
|
||||
// Issue 1079: a pull request's edit event carries its description, so mesh-delivery reads a Reviewed-by line written after
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
// What the forge's events ask of the announcer, and the one pass at a start (issue 1422, notes 10 and 11). Nothing polls the
|
||||
// forge: a head or merge is announced when its event is heard, and a pass over every repository runs once when the module
|
||||
// starts and again when the receiver is listening, so what happened while it was down is announced too.
|
||||
//
|
||||
// Apart from index.ts, which has side effects at import, so these are tested with the announcers injected.
|
||||
|
||||
import type { GiteaPull, GiteaRepo } from "./client.js";
|
||||
import type { Announced } from "./pulls.js";
|
||||
import { headKey, headsToAnnounce } from "./pulls.js";
|
||||
import type { ForgeEvent } from "./webhook.js";
|
||||
|
||||
export interface ActDeps<R, P> {
|
||||
getRepo(owner: string, repo: string): Promise<R>;
|
||||
getPull(owner: string, repo: string, number: number): Promise<P & { state?: string; merged?: boolean }>;
|
||||
announceHead(repo: R, pull: P): Promise<unknown>;
|
||||
announceMerge(repo: R, pull: P): Promise<unknown>;
|
||||
announceClosed(repo: R, pull: P): Promise<unknown>;
|
||||
siblingMoves(full: string, number: number): Promise<unknown>;
|
||||
}
|
||||
|
||||
const HEAD_EVENTS = ["forge.pull-request.opened", "forge.pull-request.synchronized", "forge.pull-request.reopened"];
|
||||
|
||||
/**
|
||||
* What the forge's event asks of the announcer at once: a new head is announced, a merge, a close without a merge, and a
|
||||
* pull request retargeted (an edit: the head is announced into the base it now has — the base is in the head's key, so an
|
||||
* edit that changed nothing about the base announces nothing).
|
||||
*/
|
||||
export async function actOn<R, P>(event: ForgeEvent, d: ActDeps<R, P>): Promise<void> {
|
||||
const b = event.body as { owner?: string; repo?: string; number?: number; merged?: boolean };
|
||||
if (!b.owner || !b.repo || !b.number) return;
|
||||
const edited = event.name === "forge.pull-request.edited";
|
||||
const closed = event.name === "forge.pull-request.closed";
|
||||
const head = HEAD_EVENTS.includes(event.name) || edited;
|
||||
const merged = event.name === "forge.pull-request.merged";
|
||||
if (!head && !merged && !closed) return;
|
||||
if (closed && b.merged) return; // a merge is weighed with its own event (ADR 0382 §5)
|
||||
const [repo, pull] = await Promise.all([d.getRepo(b.owner, b.repo), d.getPull(b.owner, b.repo, b.number)]);
|
||||
if (closed) {
|
||||
await d.announceClosed(repo, pull);
|
||||
await d.siblingMoves(`${b.owner}/${b.repo}`, b.number);
|
||||
return;
|
||||
}
|
||||
if (head && pull.state === "open") await d.announceHead(repo, pull);
|
||||
if (merged) await d.announceMerge(repo, pull);
|
||||
}
|
||||
|
||||
/**
|
||||
* The one pass at a start, and its retry: run now; if it fails, say so and run again after `retryMs`, until one pass has
|
||||
* ended. A retry of a failed pass, never a poll: after a pass has ended nothing is scheduled.
|
||||
*/
|
||||
export function startPass(run: () => Promise<void>, opts: { retryMs: number; log?: (l: string) => void }): Promise<void> {
|
||||
const log = opts.log ?? ((l) => console.error(`[gitea] ${l}`));
|
||||
return run().catch((err) => {
|
||||
log(`the pass over every repository failed, asked again in ${opts.retryMs / 1000} s — ${err instanceof Error ? err.message : err}`);
|
||||
return new Promise<void>((resolve) => setTimeout(() => resolve(startPass(run, opts)), opts.retryMs));
|
||||
});
|
||||
}
|
||||
|
||||
export interface HeadDeps {
|
||||
/** Whether the forge shows a merge check of the head: only an announced head has one. */
|
||||
alreadyChecked(pull: GiteaPull): Promise<boolean>;
|
||||
/** Announce it; true when it was this call that announced it (not an event before it). */
|
||||
announceHead(pull: GiteaPull): Promise<boolean>;
|
||||
/** Told of a head this pass announced that no event had. */
|
||||
missed(what: string): void;
|
||||
}
|
||||
|
||||
/**
|
||||
* The heads of one repository's open pull requests a pass announces. A head is announced when it is new and fresh (moved in
|
||||
* the last day); at the first pass after a start (`primed` false) one the forge shows a merge check of was announced
|
||||
* already, and is only remembered.
|
||||
*/
|
||||
export async function announceNewHeads(full: string, open: GiteaPull[], heads: Announced, primed: boolean, dayAgo: string, d: HeadDeps): Promise<void> {
|
||||
for (const pull of headsToAnnounce(full, open, heads)) {
|
||||
const key = headKey(full, pull.number, pull.base);
|
||||
const fresh = primed || (!!pull.updated_at && pull.updated_at > dayAgo);
|
||||
if (fresh && !primed && await d.alreadyChecked(pull)) {
|
||||
heads[key] = String(pull.head_sha);
|
||||
continue;
|
||||
}
|
||||
if (fresh) {
|
||||
if (await d.announceHead(pull)) d.missed(`${key} at ${String(pull.head_sha).slice(0, 8)}`);
|
||||
} else heads[key] = String(pull.head_sha);
|
||||
}
|
||||
}
|
||||
|
||||
export type { GiteaRepo };
|
||||
+19
-22
@@ -3,7 +3,7 @@
|
||||
// with a secret it mints and keeps beside its token; the receiver refuses whatever is unsigned or does not verify, and
|
||||
// publishes each event, trimmed, on the module's events as forge.<kind>.<action> (forge.pull-request.opened, forge.push, …),
|
||||
// on the shared EVENTS path its other events take; its own stream (ADR 0363) is a follow-up once issue 725 ships.
|
||||
// The polls stay as the safety net, slowed; the announced-heads and announced-merges records keep anything from
|
||||
// No poll is the safety net (issue 1422); the announced-heads and announced-merges records keep anything from
|
||||
// being announced twice, whichever path saw it first.
|
||||
//
|
||||
// The secret is never logged: lines say it was minted or reused and where it is kept, never what it is.
|
||||
@@ -144,7 +144,7 @@ export interface ReceiverOptions {
|
||||
publish: (event: ForgeEvent) => Promise<void>;
|
||||
/** Acts on it at once, beside publishing it (the announcer). */
|
||||
act?: (event: ForgeEvent) => Promise<void>;
|
||||
/** Told of every verified delivery — its kind (X-Gitea-Event) and its delivery id — so the polls know events are
|
||||
/** Told of every verified delivery — its kind (X-Gitea-Event) and its delivery id — so the module knows events are
|
||||
* heard, and the module can say when it last heard one. */
|
||||
heard?: (kind: string, delivery: string) => void;
|
||||
/** Told of every refused delivery, with the reason and where it came from. */
|
||||
@@ -242,7 +242,7 @@ export function receiver(opts: ReceiverOptions): Server & { idle(): Promise<void
|
||||
await opts.publish(event);
|
||||
if (opts.act) await opts.act(event);
|
||||
} catch (err) {
|
||||
log(`a forge event ${event.name} could not be published or acted on; the polls catch it — ${err instanceof Error ? err.message : err}`);
|
||||
log(`a forge event ${event.name} could not be published or acted on; the pass over every repository is asked again shortly — ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -276,14 +276,13 @@ export interface HookState {
|
||||
export const SILENT_AFTER = 10 * 60_000;
|
||||
|
||||
/**
|
||||
* **How fast the polls look, and whether the forge's events are heard** (issue 729): every five minutes only while the
|
||||
* forge's events are heard — a verified delivery within the last ten minutes, and none missed since. Until the first
|
||||
* delivery, after ten minutes without one, or once the poll found something no event said, every thirty seconds.
|
||||
* **Whether the forge's events are heard** (issues 729, 1422). Nothing polls the forge, so there is no pace: a delivery
|
||||
* heard within the last ten minutes, and none missed since, is `hearing`.
|
||||
*
|
||||
* It also keeps what `gitea_events` says: where the receiver listens, the hook as ensured, the verified and refused
|
||||
* deliveries, and the polls' finds no event said. **A forge whose events stop is said** (issue 729): once the receiver
|
||||
* deliveries, and the finds of the pass over every repository (at the start and when the receiver listens) that no event said. **A forge whose events stop is said** (issue 729): once the receiver
|
||||
* listens and the hook is ensured, when nothing verified for ten minutes — since the start, or since the last delivery —
|
||||
* and the poll has found at least two heads or merges no event said in that quiet (one could be the poll beating an
|
||||
* and a pass has found at least two heads or merges no event said in that quiet (one could be a pass beating an
|
||||
* event by a second), it is logged loudly once per silent spell, and the module's events are said unhealthy until a
|
||||
* delivery verifies. Deliveries refused since the last verified one are said unhealthy too.
|
||||
*/
|
||||
@@ -318,7 +317,7 @@ export class Pace {
|
||||
this.notHeard = null;
|
||||
}
|
||||
/** No events are listened for, and why: by configuration (`fails` false), or because the receiver could not listen
|
||||
* or the hook could not be ensured (`fails` true, said unhealthy). The polls alone watch. */
|
||||
* or the hook could not be ensured (`fails` true, said unhealthy). Only the pass at the start watches. */
|
||||
unheard(why: string, fails = true): void {
|
||||
this.notHeard = why;
|
||||
this.notHeardFails = fails;
|
||||
@@ -337,10 +336,10 @@ export class Pace {
|
||||
this.refusedSpell++;
|
||||
this.lastRefused = { at: this.now(), reason, from };
|
||||
}
|
||||
/** The poll found a head or merge no forge event announced. */
|
||||
/** The pass over every repository found a head or merge no forge event announced. */
|
||||
miss(what: string): void {
|
||||
const t = this.now();
|
||||
if (this.hearing()) this.log(`THE FORGE'S EVENTS MISSED ${what}: the poll found it, so the polls look every 30 s again until the next event`);
|
||||
if (this.hearing()) this.log(`THE FORGE'S EVENTS MISSED ${what}: the pass over every repository found it`);
|
||||
this.missed = t;
|
||||
this.missCount++;
|
||||
this.lastMiss = { at: t, what };
|
||||
@@ -355,8 +354,8 @@ export class Pace {
|
||||
this.log(`THE FORGE'S EVENTS ARE NOT HEARD: ${this.last ? `no delivery verified since the last at ` +
|
||||
`${new Date(this.last).toISOString()}` : "no delivery ever verified since the start"}, ${minutes} min, from the ` +
|
||||
`system webhook at ${this.hook?.url}${refusedSince ? `; ${refusedSince} refused since, the last because ` +
|
||||
`${this.lastRefused?.reason}` : ""}; the poll found ${what} — the polls alone watch the forge, every 30 s ` +
|
||||
`(novox/hq issue 729)`);
|
||||
`${this.lastRefused?.reason}` : ""}; the pass found ${what}; nothing else watches the forge ` +
|
||||
`(novox/hq issue 729, 1422)`);
|
||||
}
|
||||
}
|
||||
/** Whether a delivery was refused after the last verified one (or with none verified). */
|
||||
@@ -367,9 +366,6 @@ export class Pace {
|
||||
const t = this.now();
|
||||
return this.last > 0 && t - this.last < 10 * 60_000 && this.last > this.missed;
|
||||
}
|
||||
every(): number {
|
||||
return this.hearing() ? 5 * 60_000 : 30_000;
|
||||
}
|
||||
private expecting(): boolean {
|
||||
return this.where !== null && this.hook !== null;
|
||||
}
|
||||
@@ -383,14 +379,14 @@ export class Pace {
|
||||
if (this.silentSince) {
|
||||
state = "unhealthy";
|
||||
why = `${this.last ? `no delivery verified since the last at ${iso(this.last)}` : `the forge has never delivered an ` +
|
||||
`event since the start at ${iso(this.started)}`}, while the poll found ${this.missCount} head(s) or merge(s) no ` +
|
||||
`event since the start at ${iso(this.started)}`}, while the pass found ${this.missCount} head(s) or merge(s) no ` +
|
||||
`event said, the last ${this.lastMiss?.what} at ${iso(this.lastMiss?.at ?? t)}`;
|
||||
} else if (refusedSince) {
|
||||
state = "unhealthy";
|
||||
why = `${refusedSince} deliveries refused ${this.last ? `since the last verified one at ${iso(this.last)}` :
|
||||
"and none verified"}, the last at ${iso(this.lastRefused?.at ?? t)} because ${this.lastRefused?.reason}`;
|
||||
} else if (!this.expecting()) {
|
||||
why = this.notHeard ? `no events are listened for, so the polls alone watch the forge: ${this.notHeard}` :
|
||||
why = this.notHeard ? `no events are listened for, so only the pass at the start watches the forge: ${this.notHeard}` :
|
||||
"the receiver is not listening yet";
|
||||
if (this.notHeard && this.notHeardFails) state = "unhealthy";
|
||||
} else if (this.hearing()) {
|
||||
@@ -398,7 +394,7 @@ export class Pace {
|
||||
} else if (this.last === 0) {
|
||||
why = `no delivery heard yet; the forge is said silent if the poll finds what no event said ${SILENT_AFTER / 60_000} min after the start`;
|
||||
} else {
|
||||
why = `the last delivery was at ${iso(this.last)}; the polls look every 30 s until the next`;
|
||||
why = `the last delivery was at ${iso(this.last)}; nothing polls the forge, so what happens meanwhile is announced at the next start or when the receiver listens again`;
|
||||
}
|
||||
return {
|
||||
healthy: state !== "unhealthy", state, why,
|
||||
@@ -407,7 +403,8 @@ export class Pace {
|
||||
hook: this.hook,
|
||||
verified: { count: this.verifiedCount, last: this.lastVerified && { ...this.lastVerified, at: iso(this.lastVerified.at) } },
|
||||
refused: { count: this.refusedCount, since_verified: refusedSince, last: this.lastRefused && { ...this.lastRefused, at: iso(this.lastRefused.at) } },
|
||||
polls: { every_seconds: this.every() / 1000, hearing: this.hearing(), found_unsaid: this.missCount,
|
||||
polls: { every_seconds: null, note: "no timer asks the forge; one pass at the start and when the receiver listens (issue 1422)",
|
||||
hearing: this.hearing(), found_unsaid: this.missCount,
|
||||
last_found: this.lastMiss && { ...this.lastMiss, at: iso(this.lastMiss.at) } },
|
||||
silent: this.silentSince ? { since: iso(this.silentSince) } : null,
|
||||
};
|
||||
@@ -502,8 +499,8 @@ export function eventsTool(pace: Pace): { name: string; description: string; inp
|
||||
name: "gitea_events",
|
||||
description: "Whether the forge's events reach the mesh (novox/hq issue 729): where the receiver listens, the " +
|
||||
"system webhook as ensured at the start (its id, url, events, kind, and what it replaced), how many deliveries " +
|
||||
"verified and the last one's time and event, how many were refused and the last reason, how often the polls look " +
|
||||
"and whether events are heard, and how many heads or merges the poll found that no event said. Unhealthy when the " +
|
||||
"verified and the last one's time and event, how many were refused and the last reason, whether " +
|
||||
"events are heard (nothing polls; one pass runs at the start), and how many heads or merges that pass found that no event said. Unhealthy when the " +
|
||||
"forge has delivered nothing that verified for ten minutes (since the start or since the last delivery) while the " +
|
||||
"poll found what no event said, when deliveries were refused since the last verified one, or when the receiver " +
|
||||
"could not listen or the webhook could not be made.",
|
||||
|
||||
Reference in New Issue
Block a user