Take the checker's owed without waiting on the forge, so a large owed never times its watch out (issue 1726)
mesh/delivery delivered
mesh/check checking 5a7076a6a25e on f836c021e14f
mesh/delivery delivered
mesh/check checking 5a7076a6a25e on f836c021e14f
The runtime gives a watch's current values 10 s and waits for each to be taken; a forge write per key made ~200 keys outlast it at every attempt, so after node-tools restarted on novox at 23:25:58 the watch never held and no verdict reached mesh/check. Writes now run in the background, a few at a time, each key's in revision order.
This commit is contained in:
+28
-9
@@ -83,6 +83,9 @@ export function checkStatusOf(change: OwedChange): { owner: string; repo: string
|
||||
return { owner, repo, sha, revision, status };
|
||||
}
|
||||
|
||||
/** How many mesh/check writes this module has at the forge at once. */
|
||||
export const WRITES_AT_ONCE = 4;
|
||||
|
||||
/** The one writer of mesh/check: each owed revision of the checker's, in revision order per key. */
|
||||
export class CheckStatuses {
|
||||
/** The newest revision written of each key, and the newest seen; the oldest keys go past what any forge holds open. */
|
||||
@@ -91,6 +94,9 @@ export class CheckStatuses {
|
||||
/** The turn each key's writes wait on, and the asks again still running. */
|
||||
private turns = new Map<string, Promise<unknown>>();
|
||||
private again = new Set<Promise<void>>();
|
||||
/** Writes at the forge at once, at most WRITES_AT_ONCE: a watch's start hands over every key together (issue 1726). */
|
||||
private writing = 0;
|
||||
private queued: (() => void)[] = [];
|
||||
private forge: CheckForge;
|
||||
private say: (line: string) => void;
|
||||
private sleep: (ms: number) => Promise<void>;
|
||||
@@ -123,7 +129,7 @@ export class CheckStatuses {
|
||||
this.say(`[gitea] ${asked.owner}/${asked.repo}@${asked.sha.slice(0, 12)}: mesh/check at revision ${asked.revision} is older than ${newest}, already written, and is dropped`);
|
||||
return "dropped";
|
||||
}
|
||||
await this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status);
|
||||
await this.atTheForge(() => this.forge.setCommitStatus(asked.owner, asked.repo, asked.sha, asked.status));
|
||||
this.remember(this.written, key, asked.revision);
|
||||
return "written";
|
||||
});
|
||||
@@ -133,19 +139,32 @@ export class CheckStatuses {
|
||||
return turn;
|
||||
}
|
||||
|
||||
/** What the watch hands over: written, or when the forge refuses, asked again on a bound in the background until it is
|
||||
* written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses the change. */
|
||||
/** What the watch hands over: taken at once and written in the background, or when the forge refuses, asked again on a
|
||||
* bound until it is written or a newer revision of the key was seen. Never throws: a watch's handler that throws loses
|
||||
* the change. It never waits on the forge (novox/hq issue 1726): the runtime hands a watch's current values one at a
|
||||
* time, waits for each to be taken and gives the whole set 10 s; a forge write per key made a large owed outlast that
|
||||
* bound at every attempt, so the watch never held and no verdict reached mesh/check. Each key's writes still take
|
||||
* turns in revision order (apply). */
|
||||
async take(change: OwedChange): Promise<void> {
|
||||
const run = this.apply(change).then(() => {}, (err) => this.askAgain(change, err));
|
||||
this.again.add(run);
|
||||
void run.finally(() => this.again.delete(run));
|
||||
}
|
||||
|
||||
/** Runs one forge write when fewer than WRITES_AT_ONCE are running, else after one ends. */
|
||||
private async atTheForge<T>(write: () => Promise<T>): Promise<T> {
|
||||
if (this.writing >= WRITES_AT_ONCE) await new Promise<void>((r) => this.queued.push(r));
|
||||
else this.writing++;
|
||||
try {
|
||||
await this.apply(change);
|
||||
} catch (err) {
|
||||
const run = this.askAgain(change, err);
|
||||
this.again.add(run);
|
||||
void run.finally(() => this.again.delete(run));
|
||||
return await write();
|
||||
} finally {
|
||||
const next = this.queued.shift();
|
||||
if (next) next();
|
||||
else this.writing--;
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolves when no ask again is running. */
|
||||
/** Resolves when no write or ask again is running. */
|
||||
async settled(): Promise<void> {
|
||||
while (this.again.size) await Promise.all([...this.again]);
|
||||
}
|
||||
|
||||
@@ -133,3 +133,45 @@ test("no tool writes mesh/check: gitea_check_status is gone and gitea_commit_sta
|
||||
const index = readFileSync(new URL("../index.ts", import.meta.url), "utf8");
|
||||
assert.ok(!/gitea_check_status|checkStatusTool/.test(index), "index.ts serves no gitea_check_status");
|
||||
});
|
||||
|
||||
test("the watch's current values are taken without waiting on the forge, so a slow forge never times the watch out (issue 1726)", async () => {
|
||||
// The tool runtime hands a watch's current values one at a time and waits for each to be taken; it gives the whole set
|
||||
// 10 s (mesh-tools StateTimeout) and otherwise ends the watch. On 2026-10-23 the owed held ~200 keys, each written to the
|
||||
// forge before the next was handed over: the set took longer than 10 s, the watch ended at every attempt, and no
|
||||
// verdict reached the forge's mesh/check again.
|
||||
const { CheckStatuses, watchOwed } = await import("../check.ts");
|
||||
const sent: string[] = [];
|
||||
let release: () => void = () => {};
|
||||
const slow = new Promise<void>((r) => { release = r; });
|
||||
const f = { setCommitStatus: async (_o: string, _r: string, s: string, st: any) => { await slow; sent.push(`${s.slice(0, 4)} ${st.state}`); } };
|
||||
const checks = new CheckStatuses(f, () => {}, noSleep);
|
||||
const keys = Array.from({ length: 200 }, (_, i) => key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1));
|
||||
let handedOver = 0;
|
||||
let attempts = 0;
|
||||
const watch = async (handler: (c: any) => Promise<void>) => {
|
||||
if (++attempts > 1) return; // a second attempt is not what this test waits for: the first must hold
|
||||
// The runtime's bound on the current values: the forge has answered none of them yet.
|
||||
const timedOut = new Promise<"timeout">((r) => setTimeout(() => r("timeout"), 200));
|
||||
const delivered = (async () => { for (const [i, k] of keys.entries()) { await handler(put(k, owed(), i + 1, true)); handedOver++; } return "delivered" as const; })();
|
||||
if (await Promise.race([delivered, timedOut]) === "timeout") throw new Error("the current values of mesh-delivery-checker.owed did not arrive in 10s");
|
||||
};
|
||||
const said: string[] = [];
|
||||
await watchOwed(watch, checks, noSleep, (l) => said.push(l));
|
||||
assert.equal(handedOver, 200, "every current value was taken before the forge answered");
|
||||
assert.ok(!said.some((l) => /could not be bound/.test(l)), said.join(" | "));
|
||||
assert.equal(sent.length, 0, "nothing is written before the forge answers");
|
||||
release();
|
||||
await checks.settled();
|
||||
assert.equal(sent.length, 200, "every owed status is written once the forge answers");
|
||||
});
|
||||
|
||||
test("the writes a watch's start hands over go to the forge a few at a time, never all at once (issue 1726)", async () => {
|
||||
const { CheckStatuses, WRITES_AT_ONCE } = await import("../check.ts");
|
||||
let running = 0, most = 0, done = 0;
|
||||
const f = { setCommitStatus: async () => { running++; most = Math.max(most, running); await new Promise((r) => setTimeout(r, 1)); running--; done++; } };
|
||||
const checks = new CheckStatuses(f, () => {}, noSleep);
|
||||
for (let i = 0; i < 50; i++) await checks.take(put(key(i.toString(16).padStart(4, "0") + sha.slice(4), "novox/mesh-sdk", i + 1), owed(), i + 1, true));
|
||||
await checks.settled();
|
||||
assert.equal(done, 50);
|
||||
assert.equal(most, WRITES_AT_ONCE);
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user