Merge pull request 'gitea: the checker's owed results are taken at once and written in the background, so the runner keeps the watch and every mesh/check reaches the forge (issue 1726)' (#468) from fix/1726-owed-watch-not-blocked into main

Reviewed-on: #468

modules/gitea/check.ts

@@ -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]);
   }

modules/gitea/test/check-owed.test.ts

@@ -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);
+});