CAT-RD2: gitea asks the indexer's modules once and follows its events (TypeScript follower), no reads; mesh-build-go's manifest text kept as it was
This commit is contained in:
+15
-3
@@ -49,6 +49,9 @@ import { statesAnnouncement } from "./kept.js";
|
||||
import { commitStatusTool } from "./delivery.js";
|
||||
import { CHECKER_OWED, CheckStatuses, watchOwed, type OwedChange } from "./check.js";
|
||||
import { state } from "@novox/mesh-sdk/state";
|
||||
import { on } from "@novox/mesh-sdk/events";
|
||||
import { broker } from "@novox/mesh-sdk/messaging";
|
||||
import { ViewFollower, viewTools, type ViewAnswer, type ViewEvent } from "./views.js";
|
||||
import { announce, keepWatermark, readWatermark } from "./announce.js";
|
||||
import { mergeTool } from "./merge.js";
|
||||
import { GiteaAdmin, GiteaClient, movedSince } from "./client.js";
|
||||
@@ -62,7 +65,7 @@ import { keepSiblings, noteComposed, noteVerdict, readSiblings, siblingsFile, si
|
||||
import { bindInBackground, bindUntilHeld } from "./bind.js";
|
||||
import { announcementOf, keepTook, lateVerdictComment, outOfDateWords, sameCommit, tookFile, weighTrunks } from "./trunk.js";
|
||||
import { ReadAll, readAllTools } from "./readall.js";
|
||||
import { DeliveredWrites, INDEXER_MODULES, writeTools, type ModulesChange } from "./writes.js";
|
||||
import { DeliveredWrites, INDEXER_MODULES_EVENTS, INDEXER_MODULES_TOOL, INDEXER_OWNER, writeTools, type ModulesChange } from "./writes.js";
|
||||
import { TOOLS_ACCOUNT, TOOLS_ORG, meshAccounts } from "./tools-account.js";
|
||||
|
||||
// Without a way to a token — configured, or mintable with the admin account (token.ts) — there is
|
||||
@@ -507,8 +510,11 @@ let writesOnce: DeliveredWrites | null = null;
|
||||
const writes = (): DeliveredWrites => (writesOnce ??= new DeliveredWrites(GiteaAdmin.fromEnv(),
|
||||
process.env.MESH_GITEA_TOOLS_USER || TOOLS_ACCOUNT, TOOLS_ORG, (l) => console.log(`[gitea] ${l}`), (type, body) => emit(type, body)));
|
||||
|
||||
// The first view of the indexer's modules, once the module follows it (below); gitea_first_views says where it stands.
|
||||
let modulesView: { follower: ViewFollower; owner: string; tool: string; state: string } | undefined;
|
||||
|
||||
// gitea_events is served whether or not the forge is watched: it says so (novox/hq issue 729).
|
||||
registerModuleTools("gitea", () => [eventsTool(pace), ...readAllTools(readAll), ...writeTools(writes), ...(gitea ? [commitStatusTool({
|
||||
registerModuleTools("gitea", () => [eventsTool(pace), ...readAllTools(readAll), ...writeTools(writes), ...viewTools(() => modulesView), ...(gitea ? [commitStatusTool({
|
||||
setCommitStatus: (owner, repo, sha, status) => (gitea as GiteaClient).setCommitStatus(owner, repo, sha, status),
|
||||
gates,
|
||||
}), mergeTool(gitea, { gates })] : [])]);
|
||||
@@ -626,7 +632,13 @@ if (gitea) {
|
||||
// (writes.ts). A read not granted yet is said and asked again until it holds.
|
||||
try {
|
||||
const w = writes();
|
||||
void bindUntilHeld(INDEXER_MODULES, () => state<unknown>(INDEXER_MODULES).watch((change) => w.take(change as ModulesChange)));
|
||||
// ADR 0615: asked once of the indexer's read tool, then kept by its events; the indexer's state itself is not read.
|
||||
const modules = new ViewFollower({ owner: INDEXER_OWNER, state: "modules", tool: INDEXER_MODULES_TOOL,
|
||||
ask: (after) => broker().request(INDEXER_MODULES_TOOL, after ? { values: true, after } : { values: true }) as Promise<ViewAnswer>,
|
||||
take: (change) => w.take(change as ModulesChange) });
|
||||
void bindUntilHeld(INDEXER_MODULES_EVENTS, () => on<ViewEvent>(INDEXER_MODULES_EVENTS, (event) => modules.hear(event.body)));
|
||||
modulesView = { follower: modules, owner: INDEXER_OWNER, tool: INDEXER_MODULES_TOOL, state: "modules" };
|
||||
void modules.start();
|
||||
} catch (err) {
|
||||
console.error(`[gitea] the merging account is not given write on the delivered repositories: no site administrator's login — ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
|
||||
@@ -572,7 +572,12 @@
|
||||
"mesh-build-node.keys",
|
||||
"mesh-build-image.keys",
|
||||
"mesh-delivery-checker.owed",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-gate.source"
|
||||
],
|
||||
"invokes": [
|
||||
"mesh-delivery-indexer.indexer_read_modules"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-delivery-indexer.views.modules"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
import { ViewFollower, viewTools, type ViewAnswer, type ViewChange, type ViewEvent } from "../views.ts";
|
||||
|
||||
// ADR 0615: the module asks the indexer's read tool once, then keeps its picture of `modules` from the indexer's views
|
||||
// events; a gap asks once more; an owner that is down is asked again and never read instead.
|
||||
|
||||
const made = "2026-11-04T12:00:00Z";
|
||||
const answer = (revision: number, values: Record<string, unknown>): ViewAnswer =>
|
||||
({ v: 1, owner: "o", state: "modules", revision, created: made, values });
|
||||
const ev = (revision: number, key: string, value?: unknown, extra: Partial<ViewEvent> = {}): ViewEvent =>
|
||||
({ v: 1, owner: "o", state: "modules", key, revision, created: made, value, deleted: value === undefined, ...extra });
|
||||
|
||||
test("a follower takes the first view, then the events, and asks again only on a gap", async () => {
|
||||
const changes: ViewChange[] = [];
|
||||
const said: string[] = [];
|
||||
let asks = 0;
|
||||
let down = 2;
|
||||
const f = new ViewFollower({ owner: "o", state: "modules", tool: "o.read",
|
||||
ask: async () => { if (down-- > 0) throw new Error("no responders"); asks++; return answer(9, { a: { repository: "novox/a" }, b: { repository: "novox/b" } }); },
|
||||
take: (c) => { changes.push(c); }, say: (l) => said.push(l), sleep: async () => {}, now: (() => { let t = 0; return () => (t += 200_000); })() });
|
||||
await f.start();
|
||||
assert.equal(f.ready, true);
|
||||
assert.equal(f.values.size, 2);
|
||||
assert.ok(said.some((l) => l.includes("no first view of o's modules") && l.includes("o.read")), said.join("|"));
|
||||
await f.hear(ev(8, "a", { repository: "old" })); // already in the picture
|
||||
await f.hear(ev(10, "c", { repository: "novox/c" }));
|
||||
await f.hear(ev(11, "a")); // deleted
|
||||
assert.deepEqual([f.revision, f.values.has("a"), f.values.has("c"), asks], [11, false, true, 1]);
|
||||
assert.deepEqual(changes.slice(-2).map((c) => [c.key, c.op]), [["c", "put"], ["a", "delete"]]);
|
||||
await f.hear(ev(13, "d", { repository: "x" })); // revision 12 missed: a gap
|
||||
assert.equal(asks, 2);
|
||||
await f.hear(ev(1, "e", { repository: "x" }, { created: "2026-11-04T13:00:00Z" })); // the state made again: a gap
|
||||
assert.equal(asks, 3);
|
||||
await f.hear(ev(10, "f", undefined, { gap: true })); // the next revision, too large to carry: a gap
|
||||
assert.equal(asks, 4);
|
||||
await f.hear({ ...ev(10, "g", { repository: "x" }), state: "other" }); // another state
|
||||
assert.equal(f.values.has("g"), false);
|
||||
});
|
||||
|
||||
test("the reader's two tools say where the first view stands and ask it again by hand", async () => {
|
||||
let asks = 0;
|
||||
const follower = new ViewFollower({ owner: "o", state: "modules", tool: "o.read", ask: async () => { asks++; return answer(5, { a: 1 }); },
|
||||
take: () => {}, sleep: async () => {} });
|
||||
const [first, again] = viewTools(() => ({ follower, owner: "o", tool: "o.read", state: "modules" }));
|
||||
assert.deepEqual(await first.run({}), { views: [{ owner: "o", tool: "o.read", state: "modules", revision: 0, created: "", keys: 0, answered: false, asks: 0, "by-hand": 0 }] });
|
||||
const after = (await again.run({})) as { views: { revision: number; "by-hand": number; asks: number }[] };
|
||||
assert.deepEqual([after.views[0].revision, after.views[0]["by-hand"], asks], [5, 1, 1]);
|
||||
assert.deepEqual(await viewTools(() => undefined)[0].run({}), { views: [] });
|
||||
});
|
||||
@@ -0,0 +1,204 @@
|
||||
// A reader's picture of one state of the indexer or the assigner (novox/hq ADR 0615 §1, §4 to §6; design 25 § 12.2): asked once
|
||||
// of the owner's read tool, then kept by the owner's views events alone. It is the TypeScript form of the Go reader
|
||||
// (mesh-delivery sdk/views Follower) and follows the same rules:
|
||||
//
|
||||
// - at its start it asks the tool until it is answered, waiting longer between asks (up to a minute), and past two minutes
|
||||
// says so naming the owner and the tool; it never reads the owner's state instead;
|
||||
// - an event of another `created` is a gap (the owner's state was made again); one at or below the revision held is already
|
||||
// in the picture; one past the next revision is a gap; one with `gap: true` (a value too large to carry) is a gap;
|
||||
// - a gap asks the tool once more, said in the log;
|
||||
// - events heard while an ask runs are taken again after its answer, so none is lost.
|
||||
|
||||
/** The answer of the owner's read tool with `values`: one page of the state, at the owner's revision. */
|
||||
export interface ViewAnswer {
|
||||
v: number;
|
||||
owner: string;
|
||||
state: string;
|
||||
revision: number;
|
||||
created: string;
|
||||
values: Record<string, unknown>;
|
||||
next?: string;
|
||||
}
|
||||
|
||||
/** One change of one key, said on the owner's views family. */
|
||||
export interface ViewEvent {
|
||||
v: number;
|
||||
owner: string;
|
||||
state: string;
|
||||
key: string;
|
||||
revision: number;
|
||||
created: string;
|
||||
deleted?: boolean;
|
||||
value?: unknown;
|
||||
gap?: boolean;
|
||||
}
|
||||
|
||||
/** One change of the picture, in the form the module's watch handed before. */
|
||||
export interface ViewChange {
|
||||
key: string;
|
||||
op: "put" | "delete";
|
||||
value?: unknown;
|
||||
revision?: number;
|
||||
current?: boolean;
|
||||
}
|
||||
|
||||
export interface ViewFollowerOptions {
|
||||
owner: string;
|
||||
state: string;
|
||||
/** The owner's read tool as the manifest invokes it, for the log. */
|
||||
tool: string;
|
||||
/** Asks the tool for the page after `after` (none: the first). */
|
||||
ask: (after?: string) => Promise<ViewAnswer>;
|
||||
take: (change: ViewChange) => void | Promise<void>;
|
||||
say?: (line: string) => void;
|
||||
sleep?: (ms: number) => Promise<void>;
|
||||
now?: () => number;
|
||||
/** How long it asks before it says so; two minutes. */
|
||||
boundMs?: number;
|
||||
}
|
||||
|
||||
export class ViewFollower {
|
||||
readonly values = new Map<string, unknown>();
|
||||
revision = 0;
|
||||
created = "";
|
||||
ready = false;
|
||||
asks = 0;
|
||||
byHand = 0;
|
||||
private asking = false;
|
||||
private during: ViewEvent[] = [];
|
||||
|
||||
private readonly o: ViewFollowerOptions;
|
||||
|
||||
constructor(o: ViewFollowerOptions) {
|
||||
this.o = o;
|
||||
}
|
||||
|
||||
private say(line: string): void {
|
||||
(this.o.say ?? ((l) => console.error(`[gitea] ${l}`)))(line);
|
||||
}
|
||||
|
||||
/** Asks until answered: each failed ask waits, doubling to a minute, and past the bound says what waits. */
|
||||
async start(): Promise<void> {
|
||||
await this.askUntil("start");
|
||||
}
|
||||
|
||||
/** Asks once more by hand, until answered (a hand act, counted). */
|
||||
async again(why: string): Promise<void> {
|
||||
this.byHand++;
|
||||
await this.askUntil(`by hand: ${why}`);
|
||||
}
|
||||
|
||||
private async askUntil(why: string): Promise<void> {
|
||||
if (this.asking) return; // one ask at a time; the one running brings the picture to its newest
|
||||
this.asking = true;
|
||||
this.during = [];
|
||||
const sleep = this.o.sleep ?? ((ms: number) => new Promise<void>((r) => setTimeout(r, ms)));
|
||||
const now = this.o.now ?? Date.now;
|
||||
const first = now();
|
||||
let said = false;
|
||||
try {
|
||||
for (let wait = 5000; ; wait = Math.min(wait * 2, 60_000)) {
|
||||
try {
|
||||
await this.askOnce();
|
||||
if (said) this.say(`the first view ${this.o.tool} of ${this.o.owner}'s ${this.o.state} is answered`);
|
||||
return;
|
||||
} catch (err) {
|
||||
if (!said && now() - first >= (this.o.boundMs ?? 120_000)) {
|
||||
said = true;
|
||||
this.say(`no first view of ${this.o.owner}'s ${this.o.state}: ${this.o.tool} has not answered since ${new Date(first).toISOString()} ` +
|
||||
`(${err instanceof Error ? err.message : String(err)}); it keeps asking and reads the state of nobody (${why})`);
|
||||
}
|
||||
await sleep(wait);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
this.asking = false;
|
||||
}
|
||||
}
|
||||
|
||||
private async askOnce(): Promise<void> {
|
||||
this.asks++;
|
||||
const values = new Map<string, unknown>();
|
||||
let firstPage: ViewAnswer | undefined;
|
||||
for (let after: string | undefined; ;) {
|
||||
const a = await this.o.ask(after);
|
||||
firstPage ??= a;
|
||||
for (const [k, v] of Object.entries(a.values ?? {})) values.set(k, v);
|
||||
if (!a.next || a.next === after) break;
|
||||
after = a.next;
|
||||
}
|
||||
const old = new Map(this.values);
|
||||
this.values.clear();
|
||||
for (const [k, v] of values) this.values.set(k, v);
|
||||
this.revision = firstPage.revision;
|
||||
this.created = firstPage.created;
|
||||
this.ready = true;
|
||||
const during = this.during.sort((a, b) => a.revision - b.revision);
|
||||
this.during = [];
|
||||
for (const ev of during) {
|
||||
if (ev.created === this.created && ev.revision === this.revision + 1 && !ev.gap) {
|
||||
if (ev.deleted) this.values.delete(ev.key);
|
||||
else this.values.set(ev.key, ev.value);
|
||||
this.revision = ev.revision;
|
||||
}
|
||||
}
|
||||
for (const [k, v] of this.values) {
|
||||
if (!old.has(k) || JSON.stringify(old.get(k)) !== JSON.stringify(v)) await this.o.take({ key: k, op: "put", value: v, current: true });
|
||||
}
|
||||
for (const k of old.keys()) if (!this.values.has(k)) await this.o.take({ key: k, op: "delete", current: true });
|
||||
}
|
||||
|
||||
/** One event of the owner's views family: taken into the picture, or a gap that asks the tool once more. */
|
||||
async hear(ev: ViewEvent): Promise<void> {
|
||||
if (ev.state !== this.o.state) return;
|
||||
if (this.asking) this.during.push(ev);
|
||||
if (!this.ready) return; // the first view holds it
|
||||
let why = "";
|
||||
if (ev.created !== this.created) why = `${this.o.owner}'s ${this.o.state} was made again`;
|
||||
else if (ev.revision <= this.revision) return;
|
||||
else if (ev.revision > this.revision + 1) why = `${this.o.owner}'s ${this.o.state} at revision ${ev.revision}, the picture holds ${this.revision}`;
|
||||
else if (ev.gap) why = `${this.o.owner}'s ${this.o.state}.${ev.key} was too large to carry`;
|
||||
if (why) {
|
||||
this.say(`a gap in ${this.o.state} (${why}): its first view ${this.o.tool} is asked once more`);
|
||||
await this.askUntil(`a gap: ${why}`);
|
||||
return;
|
||||
}
|
||||
if (ev.deleted) this.values.delete(ev.key);
|
||||
else this.values.set(ev.key, ev.value);
|
||||
this.revision = ev.revision;
|
||||
await this.o.take({ key: ev.key, op: ev.deleted ? "delete" : "put", value: ev.value, revision: ev.revision, current: false });
|
||||
}
|
||||
}
|
||||
|
||||
type Tool = { name: string; description: string; input: Record<string, unknown>; run: (args: Readonly<Record<string, unknown>>) => Promise<unknown> };
|
||||
|
||||
/** Where a first view stands (design 25 § 12.2): the owner and its tool, the revision the picture holds, how many asks. (r) */
|
||||
export function status(f: ViewFollower, owner: string, tool: string, state: string): Record<string, unknown> {
|
||||
return { owner, tool, state, revision: f.revision, created: f.created, keys: f.values.size, answered: f.ready, asks: f.asks, "by-hand": f.byHand };
|
||||
}
|
||||
|
||||
/** The reader's two tools: `gitea_first_views` says where its first views stand, `gitea_first_view_again` asks one once more by hand. */
|
||||
export function viewTools(view: () => { follower: ViewFollower; owner: string; tool: string; state: string } | undefined): Tool[] {
|
||||
return [
|
||||
{
|
||||
name: "gitea_first_views",
|
||||
description: "Where each first view this module keeps stands (novox/hq ADR 0615): the owner and its tool, the revision the picture holds, how many asks (by hand among them). (r)",
|
||||
input: {},
|
||||
run: async () => {
|
||||
const v = view();
|
||||
return { views: v ? [status(v.follower, v.owner, v.tool, v.state)] : [] };
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "gitea_first_view_again",
|
||||
description: "Ask the indexer's modules again by hand: a hand act, for a picture the operator doubts. Answers where it stands after. (a)",
|
||||
input: {},
|
||||
run: async () => {
|
||||
const v = view();
|
||||
if (!v) throw new Error("the module follows no first view yet");
|
||||
await v.follower.again("the operator asked");
|
||||
return { views: [status(v.follower, v.owner, v.tool, v.state)] };
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
@@ -9,9 +9,9 @@
|
||||
//
|
||||
// **Which repositories: the indexer's word, read over the bus.** mesh-delivery-indexer keeps every registered module in its
|
||||
// state `modules` (mesh-sdk go/delivery ModulesBucket), each value naming the repository it builds from. This module reads
|
||||
// that state (its manifest's `reads`) with a watch: the watch hands the latest value of every key when it starts, and then
|
||||
// every change, so a repository is granted at the module's start and when a module of it is first registered. Nothing
|
||||
// polls. A repository of the organisation needs nothing: the organisation's team writes it (tools-account.ts). Every other
|
||||
// that state through the indexer's read tool (its manifest's `invokes`) and then its views events (`consumes`; ADR 0615): the
|
||||
// first view hands the latest value of every key when the module starts, and then every change, so a repository is granted
|
||||
// at the module's start and when a module of it is first registered. Nothing polls and the state is not read. A repository of the organisation needs nothing: the organisation's team writes it (tools-account.ts). Every other
|
||||
// repository of a user (the operator's dotfiles, say) is left as it is: read only.
|
||||
//
|
||||
// **Every grant is read back** from the forge (its permission for the account), and said on the bus as
|
||||
@@ -31,8 +31,12 @@ import type { ForgeRepository } from "./client.js";
|
||||
import type { BranchProtection } from "./protection.js";
|
||||
import { CHECK_CONTEXT, CHECK_STATUS, REPO_CHECK_CONTEXT } from "./pulls.js";
|
||||
|
||||
/** The indexer's state this module reads: one key per registered module, its value naming the repository. */
|
||||
export const INDEXER_MODULES = "mesh-delivery-indexer.modules";
|
||||
/** The indexer, whose state `modules` this module follows: one key per registered module, its value naming the repository.
|
||||
* It asks the indexer's read tool once and then hears its views events (ADR 0615), never reading the state. */
|
||||
export const INDEXER_OWNER = "mesh-delivery-indexer";
|
||||
export const INDEXER_MODULES_TOOL = `${INDEXER_OWNER}.indexer_read_modules`;
|
||||
export const INDEXER_MODULES_EVENTS = `${INDEXER_OWNER}.views.modules`;
|
||||
const INDEXER_MODULES = `${INDEXER_OWNER}.modules`;
|
||||
/** The event said once an account writes a repository it could not write before. */
|
||||
export const WRITABLE_EVENT = "repo.writable";
|
||||
|
||||
|
||||
@@ -10,21 +10,13 @@
|
||||
"state": [
|
||||
"toolchains",
|
||||
"keys",
|
||||
{
|
||||
"name": "steps",
|
||||
"ttl-seconds": 172800
|
||||
},
|
||||
"pauses"
|
||||
{"name": "steps", "ttl-seconds": 172800}, "pauses"
|
||||
],
|
||||
"claims": [
|
||||
{
|
||||
"name": "node-build-go",
|
||||
"scope": "node",
|
||||
"serves": [
|
||||
"current",
|
||||
"pause",
|
||||
"resume"
|
||||
]
|
||||
"serves": ["current", "pause", "resume"]
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
@@ -36,7 +28,7 @@
|
||||
"config": {
|
||||
"capacity": {
|
||||
"type": "number",
|
||||
"description": "How many asks of the Go build seat this module runs at once on this node (novox/hq ADR 0525 \u00a75); 0 pulls nothing. It is set per node in the mesh's configuration; a node with no value builds nothing.",
|
||||
"description": "How many asks of the Go build seat this module runs at once on this node (novox/hq ADR 0525 §5); 0 pulls nothing. It is set per node in the mesh's configuration; a node with no value builds nothing.",
|
||||
"minimum": 0,
|
||||
"default": 0
|
||||
},
|
||||
@@ -70,9 +62,7 @@
|
||||
"streams": [
|
||||
{
|
||||
"name": "toolchains-go",
|
||||
"events": [
|
||||
"toolchain.**"
|
||||
]
|
||||
"events": ["toolchain.**"]
|
||||
}
|
||||
],
|
||||
"requires": [
|
||||
@@ -132,11 +122,7 @@
|
||||
"path": "${dir:state}/settings.json",
|
||||
"mode": "0600",
|
||||
"owner": "mesh-build-go",
|
||||
"takes-config": [
|
||||
"capacity",
|
||||
"cores-per-step",
|
||||
"build-hours"
|
||||
],
|
||||
"takes-config": ["capacity", "cores-per-step", "build-hours"],
|
||||
"merge": "json",
|
||||
"template": "files/settings.json"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user