diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go index 4335d7868..c34b3bcbc 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go @@ -8,10 +8,13 @@ import ( "fmt" "os" "strings" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" + "artifactstoretools/internal/registry" ) @@ -53,10 +56,22 @@ func DoorAt(file string) func() (Door, error) { } // runtimeBus is the holder's bus through the runtime: its reads, its findings and its seat's event. -type runtimeBus struct{} +type runtimeBus struct{ views *views.Reader } -func (runtimeBus) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } -func (runtimeBus) Get(state, key string) ([]byte, bool, error) { +// followed says whether the state is the indexer's or the assigner's, which the holder asks once and follows by events +// (ADR 0615) rather than reads. +func (b runtimeBus) followed(state string) bool { return b.views != nil && b.views.Follows(state) } + +func (b runtimeBus) Keys(state string) ([]string, error) { + if b.followed(state) { + return b.views.Keys(state) + } + return stdio.State(state).Keys() +} +func (b runtimeBus) Get(state, key string) ([]byte, bool, error) { + if b.followed(state) { + return b.views.Get(state, key) + } e, err := stdio.State(state).Get(key) if err != nil || e == nil { return nil, false, err @@ -90,7 +105,7 @@ func (runtimeBus) Publish(subject string, body any) error { // follow watches the builder's done keys and the wants: a build's end sweeps and // observes again; a want changed observes again. Event-driven: nothing reads them on a timer. -func follow(h *Holder) { +func follow(h *Holder, v *views.Reader) { <-stdio.Served() watch := func(state string, handle func(stdio.StateChange) error) { for { @@ -119,11 +134,22 @@ func follow(h *Holder) { } return nil }) - watch(WantsRead, func(c stdio.StateChange) error { - if !c.Current { - go observe() + // The wants are the assigner's: the first view then its events (ADR 0615). What Watch hands over before it returns + // is the first view, not a change. + var current atomic.Bool + current.Store(true) + for { + err := v.Watch(WantsRead, func(string, []byte, bool) { + if !current.Load() { + go observe() + } + }) + if err == nil { + break } - return nil - }) + h.logf("watching %s: %v; asked again in 10s", WantsRead, err) + time.Sleep(10 * time.Second) + } + current.Store(false) observe() } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go index ee0c9242d..ef7422419 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go @@ -29,6 +29,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/views" + "artifactstoretools/internal/registry" ) @@ -900,6 +902,12 @@ func (h *Holder) Observe(ctx context.Context) error { } for _, k := range keys { if _, still := now[k]; !still { + // A finding of a first view unanswered is the reader's (views.Reader), cleared by its answer, not by a sweep. + if raw, found, err := h.Bus.Get(FindingsState, k); err == nil && found { + if f, err := health.DecodeFinding(raw); err == nil && f.Kind == views.UnansweredKind { + continue + } + } if err := h.Bus.Delete(FindingsState, k); err != nil { return err } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go index 8eb92fba7..630b78b78 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go @@ -10,12 +10,16 @@ package main import ( + "context" "fmt" "os" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/handacts" + + "git.novox.be/novox/mesh-sdk/go/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -26,13 +30,23 @@ func main() { Run: ExecRunner, UID: os.Getuid(), } - h := &Holder{Bus: runtimeBus{}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View, + v := &views.Reader{Module: "artifact-store-tools", Logf: logf, Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}} + v.Follow(delivery.AssignerModule, "wants", WantsTool) + v.Follow(delivery.IndexerModule, "manifests", ManifestsTool) + h := &Holder{Bus: runtimeBus{views: v}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View, Now: time.Now, Logf: logf} h.Acts = handacts.Recorder{Ask: stdio.Ask} - go follow(h) + v.Start(context.Background()) + go follow(h, v) // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): artifact-store-tools. - if err := stdio.Serve("", append(Tools(s, h, &Catalogue{Ask: stdio.Ask}), SeatTools(h, s)...)); err != nil { + if err := stdio.Serve("", append(append(Tools(s, h, &Catalogue{Ask: stdio.Ask}), SeatTools(h, s)...), v.Tools("artifact_store")...)); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } } + +// The owners' read tools of the states the holder follows, which the manifest invokes (ADR 0615). +const ( + WantsTool = delivery.AssignerModule + ".assigner_read_wants" + ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests" +) diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go index ea6fdccd8..618543de5 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go @@ -2,12 +2,14 @@ package main import ( "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/views" "context" "encoding/json" "fmt" "os" "reflect" + "slices" "sort" "strconv" "strings" @@ -443,10 +445,11 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { t.Fatal(err) } var m struct { - Tools []string `json:"tools"` - Invokes []string `json:"invokes"` - Reads []string `json:"reads"` - Claims []struct { + Tools []string `json:"tools"` + Invokes []string `json:"invokes"` + Reads []string `json:"reads"` + Consumes []string `json:"consumes"` + Claims []struct { Name string `json:"name"` Serves []string `json:"serves"` } `json:"claims"` @@ -455,7 +458,7 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { t.Fatal(err) } var served []string - for _, tl := range Tools(&Store{}, recordsAsked{&asked{}}, &Catalogue{}) { + for _, tl := range append(Tools(&Store{}, recordsAsked{&asked{}}, &Catalogue{}), (&views.Reader{}).Tools("artifact_store")...) { if !strings.HasPrefix(tl.Name, "artifact_store_") || tl.Description == "" || tl.Run == nil { t.Errorf("tool %q", tl.Name) } @@ -469,10 +472,15 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) { } // ADR 0571 clause 2: it holds the seat mesh-artifact-sweep and serves its verbs, and calls no verb of the controller's; // it reads no controller state, the lease neither (issue 1126). - want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule} + want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule, WantsTool, ManifestsTool} if !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes %v", m.Invokes) } + // ADR 0615: the assigner's wants and the indexer's manifests are asked once and followed by their events, not read. + if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.manifests"}) || + slices.Contains(m.Reads, WantsRead) || slices.Contains(m.Reads, ManifestsRead) { + t.Errorf("consumes %v, reads %v", m.Consumes, m.Reads) + } var verbs []string for _, tl := range SeatTools(&Holder{}, &Store{}) { verbs = append(verbs, strings.TrimPrefix(tl.Name, catalogue.ArtifactSweepSeat+".")) diff --git a/modules/artifact-store-tools/module.json b/modules/artifact-store-tools/module.json index 54cc59169..5fd62c284 100644 --- a/modules/artifact-store-tools/module.json +++ b/modules/artifact-store-tools/module.json @@ -20,15 +20,19 @@ "invokes": [ "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.modules", - "seat:mesh-catalog.module" + "seat:mesh-catalog.module", + "mesh-delivery-assigner.assigner_read_wants", + "mesh-delivery-indexer.indexer_read_manifests" ], "reads": [ - "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-builder.build-log", - "mesh-delivery-indexer.manifests", "node-engine.have" ], + "consumes": [ + "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.manifests" + ], "state": [ "findings" ], @@ -36,7 +40,9 @@ "artifact_store_repositories", "artifact_store_usage", "artifact_store_references", - "artifact_store_collect" + "artifact_store_collect", + "artifact_store_first_views", + "artifact_store_first_view_again" ], "config": { "older-store-names": { diff --git a/modules/gitea/index.ts b/modules/gitea/index.ts index cbf8beddc..b2af463b7 100644 --- a/modules/gitea/index.ts +++ b/modules/gitea/index.ts @@ -45,6 +45,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, unansweredFinding, 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"; @@ -57,7 +60,7 @@ import { keepSiblings, readSiblings, siblingsFile, weighSiblings } from "./sibli import { bindInBackground, bindUntilHeld } from "./bind.js"; import { announcementOf, 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 @@ -392,8 +395,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, (act) => broker().request("seat:mesh-hand-acts.hand-act", act)), ...(gitea ? [commitStatusTool({ setCommitStatus: (owner, repo, sha, status) => (gitea as GiteaClient).setCommitStatus(owner, repo, sha, status), gates, }), mergeTool(gitea, { gates })] : [])]); @@ -511,7 +517,21 @@ 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(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, + take: (change) => w.take(change as ModulesChange), + // Unanswered past its bound is a finding of the module's own, which the condition keeper raises (ADR 0615 §6). + raise: async (asked, error) => { + const f = unansweredFinding("gitea", INDEXER_OWNER, "modules", INDEXER_MODULES_TOOL, asked, error, new Date()); + await state("findings").put(f.key, f.value); + }, + clear: async () => { + await state("findings").delete(unansweredFinding("gitea", INDEXER_OWNER, "modules", INDEXER_MODULES_TOOL, new Date(), "", new Date()).key); + } }); + void bindUntilHeld(INDEXER_MODULES_EVENTS, () => on(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}`); } diff --git a/modules/gitea/module.json b/modules/gitea/module.json index 2fd3ae339..9348fe3bc 100644 --- a/modules/gitea/module.json +++ b/modules/gitea/module.json @@ -562,7 +562,8 @@ ], "state": [ "bound", - "announced" + "announced", + "findings" ], "reads": [ "mesh-provisioner.plans", @@ -572,7 +573,13 @@ "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", + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [ + "mesh-delivery-indexer.views.modules" ] } diff --git a/modules/gitea/test/views.test.ts b/modules/gitea/test/views.test.ts new file mode 100644 index 000000000..3f53160c9 --- /dev/null +++ b/modules/gitea/test/views.test.ts @@ -0,0 +1,62 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; +import { ViewFollower, findingKey, unansweredFinding, 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): ViewAnswer => + ({ v: 1, owner: "o", state: "modules", revision, created: made, values }); +const ev = (revision: number, key: string, value?: unknown, extra: Partial = {}): 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 raised: string[][] = []; + 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); }, raise: (_a, e) => { raised.push(["raised", e]); }, clear: () => { raised.push(["cleared", ""]); }, 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.deepEqual(raised.map((r) => r.slice(0, 2)), [["raised", "no responders"], ["cleared", ""]]); + 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 acts: Record[] = []; + 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" }), async (act) => { acts.push(act); }); + 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, acts.length], [5, 1, 1, 1]); + await assert.rejects(viewTools(() => ({ follower, owner: "o", tool: "o.read", state: "modules" }), async () => { throw new Error("no holder"); })[1].run({}), /not recorded/); + assert.deepEqual(await viewTools(() => undefined, async () => {})[0].run({}), { views: [] }); +}); + +test("the finding of a first view unanswered is the one mesh-sdk writes: key, kind, subject and the words", () => { + const f = unansweredFinding("gitea", "mesh-delivery-indexer", "modules", "mesh-delivery-indexer.indexer_read_modules", new Date("2026-11-04T12:00:00Z"), "no responders", new Date("2026-11-04T12:03:00Z")); + assert.equal(f.key, "module.gitea.mesh-delivery-indexer.modules.first-view-unanswered"); + assert.equal(findingKey("module", "a.b", "k"), "module.a.b.k"); + assert.equal(f.value.severity, "warning"); + assert.match(String(f.value.summary), /gitea has no first view of mesh-delivery-indexer's modules: .* \(no responders\)/); +}); diff --git a/modules/gitea/views.ts b/modules/gitea/views.ts new file mode 100644 index 000000000..dbcc7d6a4 --- /dev/null +++ b/modules/gitea/views.ts @@ -0,0 +1,241 @@ +// 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; + 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; + take: (change: ViewChange) => void | Promise; + say?: (line: string) => void; + sleep?: (ms: number) => Promise; + now?: () => number; + /** How long it asks before it says so; two minutes. */ + boundMs?: number; + /** Told once when the first view stays unanswered past the bound (the module writes its finding), and once when it is answered after. */ + raise?: (asked: Date, error: string) => void | Promise; + clear?: () => void | Promise; +} + +export class ViewFollower { + readonly values = new Map(); + 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 { + await this.askUntil("start"); + } + + /** Asks once more by hand, until answered (a hand act, counted). */ + async again(why: string): Promise { + this.byHand++; + await this.askUntil(`by hand: ${why}`); + } + + private async askUntil(why: string): Promise { + 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((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`); + await this.o.clear?.(); + } + return; + } catch (err) { + if (!said && now() - first >= (this.o.boundMs ?? 120_000)) { + said = true; + await this.o.raise?.(new Date(first), err instanceof Error ? err.message : String(err)); + 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 { + this.asks++; + const values = new Map(); + 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 { + 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; run: (args: Readonly>) => Promise }; + +/** 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 { + 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, + record: (act: Record) => Promise): 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"); + // A hand act, recorded through the seat mesh-hand-acts before it is done; one that cannot be recorded asks nothing. + try { + await record({ what: "asked a first view again by hand (gitea)", why: "the operator asked", cause: "first-view-again", verb: "first-view-again" }); + } catch (err) { + throw new Error(`the hand act was not recorded through mesh-hand-acts, so nothing was asked again: ${err instanceof Error ? err.message : String(err)}`); + } + await v.follower.again("the operator asked"); + return { views: [status(v.follower, v.owner, v.tool, v.state)] }; + }, + }, + ]; +} + +/** The key of a finding about a module: scope, the id's dotted parts, and the kind, each part made safe (mesh-sdk health.Key). */ +export function findingKey(scope: string, id: string, kind: string): string { + return [scope, ...id.split("."), kind].map((p) => p.trim().replace(/[^A-Za-z0-9_=/-]/g, "_") || "_").join("."); +} + +export const UNANSWERED_KIND = "first-view-unanswered"; + +/** A reader's finding of a first view unanswered past its bound, as mesh-sdk views.UnansweredFinding writes it (ADR 0615 §6): the + * condition keeper raises the condition from the module's state `findings`. */ +export function unansweredFinding(reader: string, owner: string, state: string, tool: string, asked: Date, error: string, at: Date): + { key: string; value: Record } { + const id = `${reader}.${owner}.${state}`; + const key = findingKey("module", id, UNANSWERED_KIND); + const summary = `${reader} has no first view of ${owner}'s ${state}: ${tool} has not answered since ${asked.toISOString()} (${error.trim()})`; + return { key, value: { + v: 1, key, kind: UNANSWERED_KIND, subject: { scope: "module", id }, severity: "warning", summary, + headline: `${reader} waits for ${owner}`, + explanation: `${reader} reads ${owner}'s ${state} only through ${tool}, and keeps asking it; it never reads the state itself. The answer clears this. ` + + `Look at ${owner}'s runtime of its own: is it running and connected?`, + resolved: "", evidence: [{ at: at.toISOString(), said: summary }], source: `${reader}.${UNANSWERED_KIND}`, observer: reader, at: at.toISOString(), + } }; +} diff --git a/modules/gitea/writes.ts b/modules/gitea/writes.ts index 552f7d585..029e4d8ba 100644 --- a/modules/gitea/writes.ts +++ b/modules/gitea/writes.ts @@ -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"; diff --git a/modules/mesh-build-go/cmd/mesh-build-go/announce.go b/modules/mesh-build-go/cmd/mesh-build-go/announce.go index 2163bca16..3550c87d1 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/announce.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/announce.go @@ -15,6 +15,8 @@ var Kept = []announce.State{ {Name: "keys", Readable: true}, {Name: "steps", Readable: true}, {Name: "pauses", Readable: true}, + // The condition keeper raises a first view of the indexer's manifests it could not get (ADR 0615 §6) from this. + {Name: "findings", Readable: true}, } // keepStates announces Kept as this module's (stdio.Keeps): main calls it before it serves. diff --git a/modules/mesh-build-go/cmd/mesh-build-go/compose.go b/modules/mesh-build-go/cmd/mesh-build-go/compose.go index 524ada97d..5a5de08a5 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/compose.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/compose.go @@ -22,16 +22,20 @@ import ( "path/filepath" "strings" "sync" + "time" "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" + stdio "git.novox.be/novox/mesh-sdk/go" sdkstep "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/composer" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/views" ) -// ManifestsState is the delivery indexer's manifests by digest, which a compose ask's request names (the module reads -// it, module.json reads). -const ManifestsState = delivery.IndexerModule + ".manifests" +// ManifestsTool is the delivery indexer's read tool of its manifests by digest, which a compose ask's request names (the +// module invokes it, module.json invokes). A manifest by its digest never changes and a compose ask is rare, so the +// module asks for the one key it needs when it needs it, and keeps no picture of the indexer's whole state (ADR 0615). +const ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests" // composerLocks hold one mutex per composer key, so a composer two asks need at once is built once. var composerLocks sync.Map @@ -69,6 +73,41 @@ func (m *Module) compose(ctx context.Context, fetch buildstep.StepFunc, a delive return delivery.ComposeAnswer{Built: built, Key: key, Verdict: verdict}, nil } +// askManifest asks the indexer's read tool for one manifest by its key: nil when it keeps none. A call the indexer does not +// answer is a finding of this module's own, which the condition keeper raises (ADR 0615 §6) and the next answer clears. +func askManifest(key string) (json.RawMessage, error) { + return askManifestVia(stdio.Ask, views.OwnFindings{}, time.Now, key) +} + +func askManifestVia(ask func(string, any) (json.RawMessage, error), findings interface { + Put(string, []byte) error + Delete(string) error +}, now func() time.Time, key string) (json.RawMessage, error) { + st := views.Status{Owner: delivery.IndexerModule, Tool: ManifestsTool, State: "manifests", Asked: now().UTC()} + raw, err := ask(ManifestsTool, map[string]any{"key": key}) + if err != nil { + st.Error = err.Error() + if k, v, ferr := views.UnansweredFinding("mesh-build-go", st, now()); ferr == nil { + _ = findings.Put(k, v) + } + return nil, err + } + if k, _, ferr := views.UnansweredFinding("mesh-build-go", st, now()); ferr == nil { + _ = findings.Delete(k) + } + var a struct { + Found bool `json:"found"` + Value json.RawMessage `json:"value"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("%s answered something that is not a key's value: %w", ManifestsTool, err) + } + if !a.Found { + return nil, nil + } + return a.Value, nil +} + // manifestOf is the manifest the indexer keeps under digest, as its repository holds it. func (m *Module) manifestOf(digest string) ([]byte, error) { if m.ReadManifest == nil { diff --git a/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go b/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go index 17dda5cd8..e38e5a856 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "os" "path/filepath" "slices" @@ -122,13 +123,35 @@ func TestAComposeAskFailsLoudly(t *testing.T) { } } -// 1622: the module reads the delivery indexer's manifests, which a compose ask names by digest. +// 1622, ADR 0615: the module asks the delivery indexer's read tool for the manifests a compose ask names by digest, and +// reads none of the indexer's state. func TestTheModuleReadsTheIndexersManifests(t *testing.T) { - if m := readManifest(t); !slices.Contains(m.Reads, ManifestsState) { - t.Errorf("reads = %v, lacks %s", m.Reads, ManifestsState) + if m := readManifest(t); !slices.Contains(m.Invokes, ManifestsTool) || slices.Contains(m.Reads, delivery.IndexerModule+".manifests") { + t.Errorf("invokes = %v, reads = %v: it asks %s", m.Invokes, m.Reads, ManifestsTool) } } +// ADR 0615 §6: an indexer that does not answer is a finding of the module's own, and its next answer clears it. +func TestAnIndexerThatDoesNotAnswerIsAFinding(t *testing.T) { + held := map[string][]byte{} + f := mapFindings(held) + at := func() time.Time { return time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) } + if _, err := askManifestVia(func(string, any) (json.RawMessage, error) { return nil, errors.New("no responders") }, f, at, "k"); err == nil || len(held) != 1 { + t.Fatalf("err %v, findings %v", err, held) + } + raw, err := askManifestVia(func(string, any) (json.RawMessage, error) { + return json.RawMessage(`{"found":true,"value":{"x":1}}`), nil + }, f, at, "k") + if err != nil || string(raw) != `{"x":1}` || len(held) != 0 { + t.Errorf("raw %s, err %v, findings %v", raw, err, held) + } +} + +type mapFindings map[string][]byte + +func (m mapFindings) Put(k string, v []byte) error { m[k] = v; return nil } +func (m mapFindings) Delete(k string) error { delete(m, k); return nil } + // ADR 0593 §7 (C9): a compose ask naming the node-engine's repository builds its composer command in the Go module at the // repository's root, under a key of that repository, commit and package, and runs the program over the request with its // manifests filled; the module's answer is the program's verdict whole. diff --git a/modules/mesh-build-go/cmd/mesh-build-go/main.go b/modules/mesh-build-go/cmd/mesh-build-go/main.go index 4e96528b5..78b3ec250 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/main.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/main.go @@ -57,7 +57,7 @@ func main() { m.RegistryFile = m.Home + "/registry" m.Settings = m.Home + "/settings.json" m.ReadKept = readOf(stdio.State(KeptState)).Get - m.ReadManifest = readOf(stdio.State(ManifestsState)).Get + m.ReadManifest = askManifest m.Probe = ProbeGo(m.Home, m.Cache) m.Bubblewrap = ProbeBubblewrap // Outside a launch by the runtime — a build, a check — there is no node and no state to write. diff --git a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go b/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go index 8e3790b95..f8c5e1589 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/manifest_test.go @@ -106,7 +106,7 @@ func TestManifestKeepsItsStatesClaimsItsBuildSeatAndDeclaresItsHealthTool(t *tes } } slices.Sort(states) - if want := []string{"keys", "pauses", "steps", "toolchains"}; !slices.Equal(states, want) { + if want := []string{"findings", "keys", "pauses", "steps", "toolchains"}; !slices.Equal(states, want) { t.Errorf("state = %v, want %v", states, want) } if len(m.Claims) != 1 || m.Claims[0].Name != Seat { diff --git a/modules/mesh-build-go/module.json b/modules/mesh-build-go/module.json index 2c6289c6a..cf557092e 100644 --- a/modules/mesh-build-go/module.json +++ b/modules/mesh-build-go/module.json @@ -10,7 +10,7 @@ "state": [ "toolchains", "keys", - {"name": "steps", "ttl-seconds": 172800}, "pauses" + {"name": "steps", "ttl-seconds": 172800}, "pauses", "findings" ], "claims": [ { @@ -20,8 +20,10 @@ } ], "reads": [ - "mesh-delivery-checker.kept", - "mesh-delivery-indexer.manifests" + "mesh-delivery-checker.kept" + ], + "invokes": [ + "mesh-delivery-indexer.indexer_read_manifests" ], "config": { "capacity": { diff --git a/modules/mesh-catalog/heard.ts b/modules/mesh-catalog/heard.ts index 619639ede..c8d58acd8 100644 --- a/modules/mesh-catalog/heard.ts +++ b/modules/mesh-catalog/heard.ts @@ -52,7 +52,6 @@ export interface Built { export const BUILDER = "mesh-delivery-builder"; export const BUILDS_SINCE_TOOL = "builder_builds_since"; export const BUILDS_STATE = "mesh-delivery-builder.builds"; -export const MANIFESTS_STATE = "mesh-delivery-indexer.manifests"; /** One entry of the builder's log, an ended build (mesh-sdk delivery.BuildLogEntry). */ export interface LogEntry { diff --git a/modules/mesh-catalog/index.ts b/modules/mesh-catalog/index.ts index cc0340028..c70d80ec3 100644 --- a/modules/mesh-catalog/index.ts +++ b/modules/mesh-catalog/index.ts @@ -15,7 +15,7 @@ import { invokeTool } from "@novox/mesh-sdk/tools"; import { state } from "@novox/mesh-sdk/state"; import { Graph } from "./store.js"; import { - answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE, MANIFESTS_STATE, + answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE, type Built, type BuildsSince, type LogEntry, } from "./heard.js"; @@ -94,11 +94,19 @@ let caughtUp = false; let asking: Promise | undefined; let askAgain = false; -const manifestOf = async (digest: string | undefined): Promise => { - if (!digest) return {}; - const held = await state<{ manifest?: string }>(MANIFESTS_STATE).get(digest); - if (!held?.value.manifest) return {}; - return JSON.parse(held.value.manifest); +/** + * **The indexer's manifests are no longer read** (novox/hq ADR 0615 §1, ruling U4): the module graph is the indexer's since + * ADR 0604, and this deprecated module's graph feeds nothing (no module consumes what it announces, and the seat + * `mesh-catalog` is the indexer's). Nothing replaces the read: a build is registered with an empty manifest, said once, until + * the module is retired. + */ +let manifestsSaid = false; +const manifestOf = async (_digest: string | undefined): Promise => { + if (!manifestsSaid) { + manifestsSaid = true; + console.error("mesh-catalog: deprecated; builds are registered without their manifests (the indexer's manifests are not read, ADR 0615)"); + } + return {}; }; const takeIn = async (e: LogEntry, replay: boolean): Promise => { diff --git a/modules/mesh-catalog/module.json b/modules/mesh-catalog/module.json index fe30560ed..c49699cbe 100644 --- a/modules/mesh-catalog/module.json +++ b/modules/mesh-catalog/module.json @@ -89,7 +89,6 @@ "mesh-delivery-builder.builder_builds_since" ], "reads": [ - "mesh-delivery-builder.builds", - "mesh-delivery-indexer.manifests" + "mesh-delivery-builder.builds" ] } diff --git a/modules/mesh-catalog/test/heard.test.ts b/modules/mesh-catalog/test/heard.test.ts index 391cb5e9c..b70b02703 100644 --- a/modules/mesh-catalog/test/heard.test.ts +++ b/modules/mesh-catalog/test/heard.test.ts @@ -2,7 +2,7 @@ import { test } from "node:test"; import assert from "node:assert/strict"; import { readFile } from "node:fs/promises"; -import { answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE, MANIFESTS_STATE } from "../heard.js"; +import { answerOf, builtFrom, placeable, BUILDER, BUILDS_SINCE_TOOL, BUILDS_STATE } from "../heard.js"; // A walk's build as the build seat says it (mesh-controller's link.BuildResult), here mailu's of novox/hq issue 712. const walkBuild = { @@ -24,7 +24,8 @@ test("the catalogue takes builds from the builder's log, and hears no controller const manifest = JSON.parse(await readFile(new URL("../module.json", import.meta.url), "utf8")); assert.deepEqual(manifest.consumes, []); assert.ok(manifest.invokes.includes(`${BUILDER}.${BUILDS_SINCE_TOOL}`), "the builder's tool is not invoked"); - for (const read of [BUILDS_STATE, MANIFESTS_STATE]) assert.ok(manifest.reads.includes(read), `${read} is not read`); + // ADR 0615 §1, ruling U4: the indexer's manifests are not read (the module is deprecated and feeds nothing). + assert.deepEqual(manifest.reads, [BUILDS_STATE]); // A log entry is placed as a build: done with its manifest, made and against; failed passed over; queued not taken in. const entry = { event: "build.ended", module: "mailu", fp: "fp1", repository: walkBuild.repository, directory: "modules/mailu", commit: walkBuild.commit, ref: "main", at: "2026-10-29T20:00:00Z", outcome: "done", manifest: "ab", against: ["img@sha256:1"], diff --git a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go index ec3719b3a..b8fa3e7f7 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/keeper.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/keeper.go @@ -40,7 +40,10 @@ const HistoryKeptFor = 90 * 24 * time.Hour // table of 2026-10-29 for the provisioner's and the router's; ADR 0571 clause 2 for the seat artifact-store's holder, // which observes S25, image-missing). var Observers = []string{health.WatchmanModule, health.SelfCheckModule, health.HealersModule, delivery.AssignerModule, - "mesh-provisioner", "messenger", artifactStore, configModule, vaultModule} + "mesh-provisioner", "messenger", artifactStore, configModule, vaultModule, + // The readers of the indexer's and the assigner's state say a first view unanswered past its bound in their own findings + // (ADR 0615 §6). + "mesh-watcher", "minio", "nfs-server", "gitea", "mesh-build-go"} // The other states it reads, as `.`. const ( diff --git a/modules/mesh-conditions/cmd/mesh-conditions/main.go b/modules/mesh-conditions/cmd/mesh-conditions/main.go index a70504ecd..aad52055e 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/main.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/main.go @@ -8,12 +8,17 @@ package main import ( + "context" "fmt" "os" "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -59,10 +64,22 @@ func NewKeeper() *Keeper { // watch keeps a watch of one state, said blind while it cannot be read: a state not on the bus yet (an observer not // delivered) is read again when it is. -func watch(k *Keeper, state, observer string, handle func(stdio.StateChange) error) { +func watch(k *Keeper, v *views.Reader, state, observer string, handle func(stdio.StateChange) error) { go func() { for wait := 5 * time.Second; ; wait = min(wait*2, 5*time.Minute) { - err := stdio.State(state).Watch("", handle) + var err error + if v.Follows(state) { + // The assigner's states are asked once of its read tool and followed by its events (ADR 0615). + err = v.Watch(state, func(key string, value []byte, gone bool) { + op := stdio.OpPut + if gone { + op = stdio.OpDelete + } + _ = handle(stdio.StateChange{Key: key, Value: value, Op: op}) + }) + } else { + err = stdio.State(state).Watch("", handle) + } if err == nil { if observer != "" { k.Blind(observer, "") @@ -79,7 +96,7 @@ func watch(k *Keeper, state, observer string, handle func(stdio.StateChange) err } // Keep starts every watch the keeper reads by (clauses 2, 8, 9, 13). -func Keep(k *Keeper) { +func Keep(k *Keeper, v *views.Reader) { for _, s := range []string{HeldState} { keys, _ := stdio.State(s).Keys() for _, key := range keys { @@ -99,35 +116,62 @@ func Keep(k *Keeper) { if n, err := k.EndCarried(); n > 0 || err != nil { logf("cleared %d condition(s) of kinds that ended with the controller: %v", n, err) } - watch(k, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) }) - watch(k, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) }) - watch(k, AttemptsRead, health.HealersModule+" attempts", func(c stdio.StateChange) error { + watch(k, v, WantsRead, "", func(c stdio.StateChange) error { return said(k.Want(c.Key, c.Value, c.Op != "put")) }) + watch(k, v, HavesRead, "", func(c stdio.StateChange) error { return said(k.Have(c.Key, c.Value, c.Op != "put")) }) + watch(k, v, AttemptsRead, health.HealersModule+" attempts", func(c stdio.StateChange) error { return said(k.Attempt(c.Key, c.Value, c.Op != "put")) }) for _, o := range Observers { - watch(k, o+"."+health.FindingsState, o, func(c stdio.StateChange) error { + watch(k, v, o+"."+health.FindingsState, o, func(c stdio.StateChange) error { return said(k.Finding(o, c.Key, c.Value, c.Op != "put")) }) } - watch(k, BoundsState, "", func(c stdio.StateChange) error { + watch(k, v, BoundsState, "", func(c stdio.StateChange) error { if c.Op == stdio.OpExpire { return said(k.BoundPassed(c.Key)) } return nil }) - watch(k, PlacementRead, "", func(c stdio.StateChange) error { + watch(k, v, PlacementRead, "", func(c stdio.StateChange) error { k.Placement(c.Key, c.Value, c.Op != stdio.OpPut) return nil }) } +// The assigner's read tools of the states the keeper follows, which the manifest invokes (ADR 0615). +const ( + WantsTool = delivery.AssignerModule + ".assigner_read_wants" + PlacementTool = delivery.AssignerModule + ".assigner_read_placement" + FindingsTool = delivery.AssignerModule + ".assigner_read_findings" +) + +// NewViews is the keeper's pictures of the assigner's wants, placement and findings. A first view it cannot get past its +// bound is a finding of its own, taken in as any observer's: the keeper raises the condition itself. +func NewViews(k *Keeper) *views.Reader { + v := &views.Reader{Module: Module, Logf: logf, Findings: ownFindings{k}, Acts: handacts.Recorder{Ask: stdio.Ask}} + v.Follow(delivery.AssignerModule, "wants", WantsTool) + v.Follow(delivery.AssignerModule, "placement", PlacementTool) + v.Follow(delivery.AssignerModule, health.FindingsState, FindingsTool) + return v +} + +// ownFindings takes the keeper's own finding straight into its picture of findings. +type ownFindings struct{ k *Keeper } + +func (o ownFindings) Put(key string, value []byte) error { + return o.k.Finding(Module, key, value, false) +} +func (o ownFindings) Delete(key string) error { return o.k.Finding(Module, key, nil, true) } + func main() { k := NewKeeper() + v := NewViews(k) go func() { <-stdio.Served() - Keep(k) + v.Start(context.Background()) + Keep(k, v) }() - if err := stdio.Serve("", Tools(k)); err != nil { + if err := stdio.Serve("", append(Tools(k), v.Tools("mesh_conditions")...)); err != nil { logf("%v", err) os.Exit(1) } diff --git a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go index 79ccbe5c6..21cae475d 100644 --- a/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go +++ b/modules/mesh-conditions/cmd/mesh-conditions/manifest_test.go @@ -7,6 +7,7 @@ import ( "testing" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" ) @@ -29,9 +30,17 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { !slices.Equal(Verbs, catalogue.VerbNames(seat.Serves)) { t.Errorf("module %s runs as %s and claims %+v; the seat serves %v", m.Module, m.RunsAs, m.Claims, catalogue.VerbNames(seat.Serves)) } - reads := []string{WantsRead, HavesRead, AttemptsRead, PlacementRead} + // The assigner's wants, placement and findings are asked once of its read tools and followed by its events, not read + // (ADR 0615 §1). + reads := []string{HavesRead, AttemptsRead} for _, o := range Observers { - reads = append(reads, o+"."+health.FindingsState) + if o != delivery.AssignerModule { + reads = append(reads, o+"."+health.FindingsState) + } + } + if !slices.Equal(m.Invokes, []string{WantsTool, PlacementTool, FindingsTool, "seat:mesh-hand-acts.hand-act"}) || + !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.findings"}) { + t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } slices.Sort(reads) got := slices.Clone(m.Reads) diff --git a/modules/mesh-conditions/mesh-conditions b/modules/mesh-conditions/mesh-conditions new file mode 100755 index 000000000..36d5a3bc2 Binary files /dev/null and b/modules/mesh-conditions/mesh-conditions differ diff --git a/modules/mesh-conditions/module.json b/modules/mesh-conditions/module.json index 793b7cdbd..41c97b37f 100644 --- a/modules/mesh-conditions/module.json +++ b/modules/mesh-conditions/module.json @@ -17,16 +17,29 @@ "mesh-delivery-watchman.findings", "mesh-self-check.findings", "mesh-healers.findings", - "mesh-delivery-assigner.findings", "messenger.findings", "mesh-provisioner.findings", "artifact-store-tools.findings", "mesh-config.findings", "mesh-vault.findings", + "mesh-watcher.findings", + "minio.findings", + "nfs-server.findings", + "gitea.findings", + "mesh-build-go.findings", "mesh-healers.attempts", - "mesh-delivery-assigner.wants", - "node-engine.have", - "mesh-delivery-assigner.placement" + "node-engine.have" + ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_wants", + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-assigner.assigner_read_findings", + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [ + "mesh-delivery-assigner.views.wants", + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.findings" ], "state": [ "conditions", diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index 7ea6bfe4c..9220412d2 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -12,11 +12,15 @@ import ( "os" "strings" "sync" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/views" ) // Seat is the role this module holds, once for the mesh, and Verb its one verb. @@ -169,9 +173,63 @@ func slowest(took []Took) time.Duration { return time.Duration(took[len(took)-1].MS) * time.Millisecond } +// The owners' read tools of the states the provisioner follows, which the manifest invokes (ADR 0615). +const ( + PlacementTool = delivery.AssignerModule + ".assigner_read_placement" + WantsTool = delivery.AssignerModule + ".assigner_read_wants" + ModulesTool = delivery.IndexerModule + ".indexer_read_modules" + ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests" +) + +// newViews is the provisioner's pictures of the assigner's placement and wants and the indexer's modules and manifests: +// each asked once of its owner's read tool, then followed by the owner's events (ADR 0615). +func newViews() *views.Reader { + v := &views.Reader{Module: Module, Logf: logf, Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}} + v.Follow(delivery.AssignerModule, "placement", PlacementTool) + v.Follow(delivery.AssignerModule, "wants", WantsTool) + v.Follow(delivery.IndexerModule, "modules", ModulesTool) + v.Follow(delivery.IndexerModule, "manifests", ManifestsTool) + return v +} + +// watchedOver is the copy's watch of a state: a followed state is handed from its picture (the first view's values as +// the current ones, then each change), any other from the runtime's watch. +func watchedOver(v *views.Reader) func(string, func(stdio.StateChange) error) error { + return func(state string, handle func(stdio.StateChange) error) error { + if !v.Follows(state) { + return stdio.State(state).Watch("", handle) + } + var landed atomic.Bool + err := v.Watch(state, func(key string, value []byte, gone bool) { + op := stdio.OpPut + if gone { + op = stdio.OpDelete + } + if err := handle(stdio.StateChange{State: state, Key: key, Op: op, Value: value, Current: !landed.Load()}); err != nil { + logf("[%s] %s is not taken: %v", Module, key, err) + } + }) + landed.Store(true) + return err + } +} + func main() { s := runtimeStore{} - ws := stdio.NewWatchedStates() // one copy of what it reads, each state by its watch (ADR 0580 clause 2) + v := newViews() + // One copy of what it reads, each state by its watch (ADR 0580 clause 2); the indexer's and the assigner's by their + // pictures, a manifest by its digest from the picture too. + ws := stdio.WatchedStatesOver(watchedOver(v), func(state, key string) ([]byte, bool, error) { + if v.Follows(state) { + return v.Get(state, key) + } + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil + }) + v.Start(context.Background()) h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8) w := &work{pending: map[string]bool{}, wake: make(chan struct{}, 1)} @@ -234,7 +292,7 @@ func main() { time.Sleep(wait) } }() - if err := stdio.Serve("", append(Tools(s), holderTools(h)...)); err != nil { + if err := stdio.Serve("", append(append(Tools(s), holderTools(h)...), v.Tools("provisioner")...)); err != nil { logf("%v", err) os.Exit(1) } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go index c856a63b6..4bccbcebe 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go @@ -8,6 +8,9 @@ import ( "strings" "sync" "testing" + + "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/views" ) // memStore is the bus as mesh-provisioner reaches it, in memory: the states it keeps. @@ -289,6 +292,8 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { } `json:"claims"` State []string `json:"state"` Reads []string `json:"reads"` + Invokes []string `json:"invokes"` + Consumes []string `json:"consumes"` Tools []string `json:"tools"` OwnSecrets map[string]string `json:"own-secrets"` } @@ -303,13 +308,13 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { t.Errorf("the claim: %+v", m.Claims) } // needs stays declared this release so its first start can empty it (ADR 0570 clause 1). - if want := []string{Registry, Needs, Applied, Collect, ProvisionsState, PlansState, RefusedState}; !slices.Equal(m.State, want) { + if want := []string{Registry, Needs, Applied, Collect, ProvisionsState, PlansState, RefusedState, health.FindingsState}; !slices.Equal(m.State, want) { t.Errorf("the state is %v, want %v", m.State, want) } // Every state and act has its tool (ADR 0570 Consequences): the verb reads the clients' states, the holder's tools // its own and the composition it does on a change. served := map[string]bool{} - for _, tool := range append(Tools(newStore()), holderTools(&Holder{})...) { + for _, tool := range append(append(Tools(newStore()), holderTools(&Holder{})...), (&views.Reader{}).Tools("provisioner")...) { served[tool.Name] = true } for what, tool := range map[string]string{Registry: Seat + "." + Verb, Needs: Seat + "." + Verb, @@ -326,11 +331,25 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { } // The builds read serves only haveOutcome's offer check (ADR 0603 clause 5: which manifest the build a provider runs // was made from) and makes no plan, so ADR 0593 §8 ("it composes nothing") holds. - for _, r := range append(slices.Clone(WatchedReads), ManifestsRead, BuildsRead) { + // The assigner's placement and wants and the indexer's modules and manifests are asked once of their owners' read + // tools and followed by their events, not read (ADR 0615 §1). + followed := []string{PlacementRead, WantsRead, ModulesRead, ManifestsRead} + for _, r := range append(slices.Clone(WatchedReads), BuildsRead) { + if slices.Contains(followed, r) { + if slices.Contains(m.Reads, r) { + t.Errorf("the manifest reads %s, which the holder follows by its owner's events", r) + } + continue + } if !slices.Contains(m.Reads, r) { t.Errorf("the manifest does not read %s, which the holder reads", r) } } + if !slices.Equal(m.Invokes, []string{PlacementTool, WantsTool, ModulesTool, ManifestsTool, "seat:mesh-hand-acts.hand-act"}) || + !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests"}) { + t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) + } // It composes nothing (ADR 0593 §8): no config value is read to make a plan. for _, r := range []string{"mesh-config.resolved"} { if slices.Contains(m.Reads, r) { diff --git a/modules/mesh-provisioner/module.json b/modules/mesh-provisioner/module.json index 1c6f4c746..db0a66b75 100644 --- a/modules/mesh-provisioner/module.json +++ b/modules/mesh-provisioner/module.json @@ -15,10 +15,6 @@ "operator-channel" ], "reads": [ - "mesh-delivery-assigner.placement", - "mesh-delivery-assigner.wants", - "mesh-delivery-indexer.modules", - "mesh-delivery-indexer.manifests", "node-engine.have", "mesh-delivery-builder.builds", "mesh-host.composed", @@ -30,12 +26,27 @@ "node-tools.bound", "mesh-vault.bound" ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-assigner.assigner_read_wants", + "mesh-delivery-indexer.indexer_read_modules", + "mesh-delivery-indexer.indexer_read_manifests", + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [ + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.modules", + "mesh-delivery-indexer.views.manifests" + ], "tools": [ "provisioner_provisions", "provisioner_plans", "provisioner_compose", "provisioner_decide", - "provisioner_health" + "provisioner_health", + "provisioner_first_views", + "provisioner_first_view_again" ], "state": [ "registry", @@ -44,7 +55,8 @@ "collect", "provisions", "plans", - "refused" + "refused", + "findings" ], "core-credentials": true, "own-secrets": { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/main.go b/modules/mesh-self-check/cmd/mesh-self-check/main.go index 71fefc4b1..e8617cf5b 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/main.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/main.go @@ -13,14 +13,23 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } // runtimeStore is the bus, through the runtime that launched this bundle. -type runtimeStore struct{} +type runtimeStore struct{ views *views.Reader } -func (runtimeStore) Get(state, key string) ([]byte, bool, error) { +func (r runtimeStore) followed(state string) bool { return r.views != nil && r.views.Follows(state) } + +func (r runtimeStore) Get(state, key string) ([]byte, bool, error) { + if r.followed(state) { + return r.views.Get(state, key) + } e, err := stdio.State(state).Get(key) if err != nil || e == nil { return nil, false, err @@ -35,9 +44,17 @@ func (runtimeStore) Put(state, key string, value any) error { func (runtimeStore) Delete(state, key string) error { return stdio.State(state).Delete(key) } -func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } +func (r runtimeStore) Keys(state string) ([]string, error) { + if r.followed(state) { + return r.views.Keys(state) + } + return stdio.State(state).Keys() +} -func (runtimeStore) Watch(state string, handle func(key string, value []byte, gone bool)) error { +func (r runtimeStore) Watch(state string, handle func(key string, value []byte, gone bool)) error { + if r.followed(state) { + return r.views.Watch(state, handle) + } return stdio.State(state).Watch("", func(ch stdio.StateChange) error { handle(ch.Key, ch.Value, ch.Op != stdio.OpPut) return nil @@ -46,20 +63,34 @@ func (runtimeStore) Watch(state string, handle func(key string, value []byte, go // NewChecker is the self-check over the runtime. func NewChecker() *Checker { - return &Checker{Store: runtimeStore{}, Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"), + store := runtimeStore{views: newViews()} + return &Checker{Store: store, Registry: Registry(), Now: time.Now, Logf: logf, Host: os.Getenv("MESH_NODE"), Enabled: enabledProbes, Ask: func(key string, body any) (json.RawMessage, error) { return stdio.Ask(key, body) }, Publish: stdio.SeatPublish} } +// newViews is the self-check's pictures of the indexer's and the assigner's states it probes (ADR 0615): each asked once of +// its owner's read tool, then followed by the owner's events. A first view unanswered past its bound is a finding of the +// self-check's own (first-view-unanswered), written into its state findings. +func newViews() *views.Reader { + r := &views.Reader{Module: Module, Logf: logf, Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}} + r.Follow(delivery.AssignerModule, delivery.PlacementState, PlacementTool) + r.Follow(delivery.AssignerModule, delivery.SeatHoldersState, SeatHoldersTool) + r.Follow(delivery.IndexerModule, "modules", ModulesTool) + r.Follow(delivery.IndexerModule, "manifests", ManifestsTool) + return r +} + func main() { c := NewChecker() go func() { <-stdio.Served() + c.Store.(runtimeStore).views.Start(context.Background()) c.WatchKept(context.Background()) c.Keep(context.Background()) }() - if err := stdio.Serve("", Tools(c)); err != nil { + if err := stdio.Serve("", append(Tools(c), c.Store.(runtimeStore).views.Tools("mesh_self_check")...)); err != nil { logf("%v", err) os.Exit(1) } diff --git a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go index e288898c6..5e9ea1fcc 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/manifest_test.go @@ -19,6 +19,8 @@ type manifest struct { Emits []string `json:"emits"` Invokes []string `json:"invokes"` Reads []string `json:"reads"` + Consume []string `json:"consumes"` + Tools []string `json:"tools"` State []json.RawMessage `json:"state"` } @@ -57,6 +59,9 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if !slices.Equal(m.Reads, Reads) { t.Errorf("reads %v, want what the probes read, %v", m.Reads, Reads) } + if !slices.Equal(m.Consume, Consumes) { + t.Errorf("consumes %v, want the views families the probes follow, %v", m.Consume, Consumes) + } states := map[string]int{} for _, s := range m.State { var d struct { diff --git a/modules/mesh-self-check/cmd/mesh-self-check/probes.go b/modules/mesh-self-check/cmd/mesh-self-check/probes.go index 150c42655..71c5d5d5b 100644 --- a/modules/mesh-self-check/cmd/mesh-self-check/probes.go +++ b/modules/mesh-self-check/cmd/mesh-self-check/probes.go @@ -33,8 +33,23 @@ const ( // Invokes are the tools and seat verbs the probes ask, and Reads the states they read: the manifest's `invokes` and // `reads`, which its test holds to these. var ( - Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb} - Reads = []string{PlacementRead, SeatHoldersRead, PresenceRead, HealthRead, ReadingsRead, GoodRead, ModulesRead, ManifestsRead} + Invokes = []string{StreamsTool, ClosedTool, SignalsVerb, BannedVerb, PlacementTool, SeatHoldersTool, ModulesTool, ManifestsTool, HandActVerb} + Reads = []string{PresenceRead, HealthRead, ReadingsRead, GoodRead} + // Consumes are the views families the probes follow (ADR 0615): the indexer's and the assigner's states are asked + // once, by their owners' read tools, and then followed by their events, never read. + Consumes = []string{"mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.seat-holders", + "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests"} +) + +// HandActVerb records a re-read of a first view by hand (ADR 0615 §6). +const HandActVerb = "seat:mesh-hand-acts.hand-act" + +// The owners' read tools of the states the probes follow, which the manifest invokes. +const ( + PlacementTool = "mesh-delivery-assigner.assigner_read_placement" + SeatHoldersTool = "mesh-delivery-assigner.assigner_read_seat_holders" + ModulesTool = "mesh-delivery-indexer.indexer_read_modules" + ManifestsTool = "mesh-delivery-indexer.indexer_read_manifests" ) // Registry is the self-check's registry, in to-be 45's order. **The registry is the design's live form**: a probe diff --git a/modules/mesh-self-check/module.json b/modules/mesh-self-check/module.json index e4037e986..d4ed50772 100644 --- a/modules/mesh-self-check/module.json +++ b/modules/mesh-self-check/module.json @@ -16,17 +16,24 @@ "nats.nats_streams", "nats.nats_closed_connections", "seat:mesh-watchdogs.signals", - "seat:node-intrusion-prevention.banned" + "seat:node-intrusion-prevention.banned", + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-assigner.assigner_read_seat_holders", + "mesh-delivery-indexer.indexer_read_modules", + "mesh-delivery-indexer.indexer_read_manifests", + "seat:mesh-hand-acts.hand-act" ], "reads": [ - "mesh-delivery-assigner.placement", - "mesh-delivery-assigner.seat-holders", "mesh-delivery-watchman.presence", "mesh-host.health", "disk-load.readings", - "restic.good", - "mesh-delivery-indexer.modules", - "mesh-delivery-indexer.manifests" + "restic.good" + ], + "consumes": [ + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.seat-holders", + "mesh-delivery-indexer.views.modules", + "mesh-delivery-indexer.views.manifests" ], "state": [ "findings", diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index 3ff9167f2..cc101b5cd 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -7,6 +7,7 @@ package main import ( + "context" "encoding/json" "fmt" "os" @@ -16,7 +17,10 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-sdk/go/views" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -70,8 +74,10 @@ func main() { node := os.Getenv("MESH_NODE") w := NewWatcher(tg, time.Now, logf, node) l := &listening{now: "not yet: starting"} - go run(w, l, node) - if err := stdio.Serve("", tools(w, l)); err != nil { + placement := &views.Reader{Module: "mesh-watcher", Logf: logf, Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}} + placement.Follow(delivery.AssignerModule, delivery.SeatHoldersState, SeatHoldersTool) + go run(w, l, node, placement) + if err := stdio.Serve("", append(tools(w, l), placement.Tools("watcher")...)); err != nil { logf("%v", err) os.Exit(1) } @@ -79,7 +85,7 @@ func main() { // run keeps time, makes the round trips, reads where the controller is, and listens for heartbeats. // The clock runs whatever the bus does: it is what notices the bus is gone. -func run(w *Watcher, l *listening, node string) { +func run(w *Watcher, l *listening, node string, placement *views.Reader) { if err := w.Telegram.Ready(); err != nil { logf("[mesh-watcher] BLIND until given: %v", err) } @@ -89,17 +95,17 @@ func run(w *Watcher, l *listening, node string) { w.Tick() } }() - go func() { - time.Sleep(2 * time.Second) - for { - if nodes, err := controlNodes(); err == nil { - w.Placed(nodes) - } else { - logf("[mesh-watcher] cannot read where the bus runs (%v); asking again in ten minutes", err) - } - time.Sleep(10 * time.Minute) + // Where the bus runs: the assigner's seat-holders, asked once and then followed by its events (ADR 0615), never read + // from its state and never on a timer. + placed := func() { + if nodes, err := controlNodes(placement); err == nil { + w.Placed(nodes) + } else { + logf("[mesh-watcher] cannot read where the bus runs (%v)", err) } - }() + } + placement.Changed = func(string, string, json.RawMessage, bool) { placed() } + placement.Start(context.Background()) handle := func(e stdio.Envelope) error { if !isHeartbeat(e.Key) { return nil @@ -146,24 +152,17 @@ func roundTrip(node string) error { } } +// SeatHoldersTool is the assigner's read tool of the state of who holds which seat, which the manifest invokes; the +// state is followed as a view (ADR 0615), named as the manifest named it before: SeatHoldersRead. +const SeatHoldersTool = delivery.AssignerModule + ".assigner_read_seat_holders" + // controlNodes reads which machines hold the bus, from the placement seat's state of who holds which seat (novox/hq // ADR 0571 clause 4). The controller is not placed (ADR 0590), so the bus's machines are the control node. -func controlNodes() ([]string, error) { - st := stdio.State(delivery.SeatHoldersRead) - names, err := st.Keys() +func controlNodes(seats *views.Reader) ([]string, error) { + keys, err := seats.Values(delivery.SeatHoldersRead) if err != nil { return nil, err } - keys := map[string][]byte{} - for _, k := range names { - e, err := st.Get(k) - if err != nil { - return nil, err - } - if e != nil { - keys[k] = e.Value - } - } return holderNodes(keys, "mesh-broker") } diff --git a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go index 14107f861..cc004b07a 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/manifest_test.go @@ -9,6 +9,8 @@ import ( "strings" "testing" "time" + + "git.novox.be/novox/mesh-sdk/go/views" ) // The manifest says what the code does: it consumes the heartbeat and nothing else, calls its own @@ -24,6 +26,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { Module string `json:"module"` Consumes []string `json:"consumes"` Invokes []string `json:"invokes"` + Reads []string `json:"reads"` Claims []any `json:"claims"` Own map[string]string `json:"own-secrets"` Tools []string `json:"tools"` @@ -34,11 +37,12 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { if err := json.Unmarshal(raw, &m); err != nil { t.Fatal(err) } - if !reflect.DeepEqual(m.Consumes, []string{"mesh-self-check.checked"}) { + // Where the bus runs is the assigner's seat-holders, asked once and followed by its events (ADR 0615), never read. + if !reflect.DeepEqual(m.Consumes, []string{"mesh-self-check.checked", "mesh-delivery-assigner.views.seat-holders"}) { t.Fatalf("consumes %v", m.Consumes) } - if !reflect.DeepEqual(m.Invokes, []string{m.Module + ".watcher_ping"}) { - t.Fatalf("invokes %v", m.Invokes) + if !reflect.DeepEqual(m.Invokes, []string{SeatHoldersTool, m.Module + ".watcher_ping", "seat:mesh-hand-acts.hand-act"}) || len(m.Reads) != 0 { + t.Fatalf("invokes %v, reads %v", m.Invokes, m.Reads) } if len(m.Claims) != 0 { t.Fatalf("a watcher holds no seat: %v", m.Claims) @@ -48,7 +52,7 @@ func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) { t.Fatalf("token: own %v, env %v", m.Own, env) } var served []string - for _, tool := range tools(NewWatcher(&fakeTelegram{}, time.Now, t.Logf, ""), &listening{}) { + for _, tool := range append(tools(NewWatcher(&fakeTelegram{}, time.Now, t.Logf, ""), &listening{}), (&views.Reader{}).Tools("watcher")...) { served = append(served, tool.Name) if strings.TrimSpace(tool.Description) == "" { t.Errorf("%s has no description", tool.Name) diff --git a/modules/mesh-watcher/module.json b/modules/mesh-watcher/module.json index f4163457b..e4555e822 100644 --- a/modules/mesh-watcher/module.json +++ b/modules/mesh-watcher/module.json @@ -3,13 +3,16 @@ "version": "1", "slug": "watch", "consumes": [ - "mesh-self-check.checked" - ], - "reads": [ - "mesh-delivery-assigner.seat-holders" + "mesh-self-check.checked", + "mesh-delivery-assigner.views.seat-holders" ], "invokes": [ - "mesh-watcher.watcher_ping" + "mesh-delivery-assigner.assigner_read_seat_holders", + "mesh-watcher.watcher_ping", + "seat:mesh-hand-acts.hand-act" + ], + "state": [ + "findings" ], "own-secrets": { "telegram-token": "${dir:state}/telegram-token" @@ -18,7 +21,9 @@ "watcher_status", "watcher_last_heard", "watcher_test", - "watcher_ping" + "watcher_ping", + "watcher_first_views", + "watcher_first_view_again" ], "config": { "telegram-chat-id": { diff --git a/modules/minio/cmd/object-store/main.go b/modules/minio/cmd/object-store/main.go index c789f03cd..9ec5152ec 100644 --- a/modules/minio/cmd/object-store/main.go +++ b/modules/minio/cmd/object-store/main.go @@ -10,10 +10,15 @@ package main import ( + "context" "fmt" "os" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-sdk/go/handacts" + "git.novox.be/novox/mesh-sdk/go/views" ) // Words the manifest gives this bundle (modules/minio/module.json). @@ -24,17 +29,24 @@ const ( ) func main() { - s, err := newServer(os.Getenv(ConfigWord), os.Getenv(LinksWord), os.Getenv(ReceivesWord)) + seats := &views.Reader{Module: "minio", Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[object-store] "+f+"\n", a...) }} + seats.Follow(delivery.AssignerModule, delivery.SeatHoldersState, SeatHoldersTool) + s, err := newServer(os.Getenv(ConfigWord), os.Getenv(LinksWord), os.Getenv(ReceivesWord), + func() (map[string][]byte, error) { return seats.Values(delivery.SeatHoldersRead) }) if err != nil { fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) os.Exit(1) } - if err := stdio.Serve("", tools(s)); err != nil { + seats.Start(context.Background()) + if err := stdio.Serve("", append(tools(s), seats.Tools("object_store")...)); err != nil { fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) os.Exit(1) } } +// SeatHoldersTool is the assigner's read tool of who holds which seat, which the manifest invokes (ADR 0615). +const SeatHoldersTool = delivery.AssignerModule + ".assigner_read_seat_holders" + func tools(s *server) []stdio.Tool { return []stdio.Tool{{ Name: "object-store.link", diff --git a/modules/minio/cmd/object-store/manifest_test.go b/modules/minio/cmd/object-store/manifest_test.go index 58494c7f4..996713a5f 100644 --- a/modules/minio/cmd/object-store/manifest_test.go +++ b/modules/minio/cmd/object-store/manifest_test.go @@ -38,9 +38,10 @@ func TestManifestClaimsTheSeat(t *testing.T) { if c["name"] != "object-store" || c["scope"] != "mesh" || len(serves) != 1 || serves[0] != "link" { t.Fatalf("claim: %v", c) } - reads, _ := m["reads"].([]any) - if _, invokes := m["invokes"]; invokes || len(reads) != 1 || reads[0] != "mesh-delivery-assigner.seat-holders" { - t.Fatalf("reads %v and invokes %v: it reads the placement seat's holders and calls nothing", reads, m["invokes"]) + invokes, _ := m["invokes"].([]any) + consumes, _ := m["consumes"].([]any) + if _, reads := m["reads"]; reads || len(invokes) != 2 || invokes[0] != SeatHoldersTool || invokes[1] != "seat:mesh-hand-acts.hand-act" || len(consumes) != 1 || consumes[0] != "mesh-delivery-assigner.views.seat-holders" { + t.Fatalf("reads %v, invokes %v, consumes %v: it asks the assigner's seat-holders once and follows its events (ADR 0615)", m["reads"], invokes, consumes) } var goBundle map[string]any for _, a := range m["build"].(map[string]any)["artifacts"].([]any) { diff --git a/modules/minio/cmd/object-store/server.go b/modules/minio/cmd/object-store/server.go index 789873b82..14684beb2 100644 --- a/modules/minio/cmd/object-store/server.go +++ b/modules/minio/cmd/object-store/server.go @@ -9,7 +9,6 @@ import ( "os" "path/filepath" "strings" - "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -21,7 +20,6 @@ import ( const ( DefaultSeconds = 300 MaxSeconds = 600 - holdersFor = 60 * time.Second ) // storeConfig is the store as this machine reaches it. The file is rendered by the mesh from files/object-store.json @@ -60,41 +58,18 @@ func linkKeyID(bucket string) string { type server struct { configPath, linksDir, receives string now func() time.Time - // seats reads the placement seat's state of who holds which seat, every key; a test replaces it. + // seats is the picture of the placement seat's state of who holds which seat, every key: asked of the assigner once + // and followed by its events (ADR 0615); a test replaces it. seats func() (map[string][]byte, error) - - mu sync.Mutex - keys map[string][]byte // seat-holders, as read - readAt time.Time - haveSet bool } -func newServer(config, links, receives string) (*server, error) { +func newServer(config, links, receives string, seats func() (map[string][]byte, error)) (*server, error) { for word, v := range map[string]string{ConfigWord: config, LinksWord: links, ReceivesWord: receives} { if strings.TrimSpace(v) == "" { return nil, fmt.Errorf("%s is not given: the manifest names it", word) } } - return &server{configPath: config, linksDir: links, receives: receives, now: time.Now, seats: askSeats}, nil -} - -func askSeats() (map[string][]byte, error) { - state := stdio.State(delivery.SeatHoldersRead) - names, err := state.Keys() - if err != nil { - return nil, err - } - keys := map[string][]byte{} - for _, k := range names { - e, err := state.Get(k) - if err != nil { - return nil, err - } - if e != nil { - keys[k] = e.Value - } - } - return keys, nil + return &server{configPath: config, linksDir: links, receives: receives, now: time.Now, seats: seats}, nil } // link answers one ask. @@ -266,18 +241,13 @@ func has(list []string, v string) bool { } // holds says whether node.module holds the seat, from the placement seat's state of who holds which seat (novox/hq ADR -// 0571 clause 4), read at most once a minute. A read that failed is not cached: the next ask reads again. +// 0571 clause 4), from the picture the assigner's first view and events keep. No first view yet is refused, saying so. func (s *server) holds(seat, node, module string) (bool, error) { - s.mu.Lock() - defer s.mu.Unlock() - if !s.haveSet || s.now().Sub(s.readAt) >= holdersFor { - keys, err := s.seats() - if err != nil { - return false, err - } - s.keys, s.readAt, s.haveSet = keys, s.now(), true + keys, err := s.seats() + if err != nil { + return false, err } - holders, err := delivery.HoldersOf(s.keys, seat, node) + holders, err := delivery.HoldersOf(keys, seat, node) if err != nil { return false, err } diff --git a/modules/minio/cmd/object-store/server_test.go b/modules/minio/cmd/object-store/server_test.go index ba7bc9565..53e72b61a 100644 --- a/modules/minio/cmd/object-store/server_test.go +++ b/modules/minio/cmd/object-store/server_test.go @@ -59,7 +59,7 @@ func newRig(t *testing.T) *rig { write("links/mesh-shanks-brother.secret", "brother-link-secret\n") write("links/mesh-novox-nextcloud.secret", "nextcloud-link-secret\n") grants := write("grants/mesh.json", grantFile) - s, err := newServer(cfg, filepath.Join(dir, "links"), grants) + s, err := newServer(cfg, filepath.Join(dir, "links"), grants, nil) if err != nil { t.Fatal(err) } @@ -165,36 +165,24 @@ func TestRefusals(t *testing.T) { } } -// 1516/AC4: with the controller unreadable a seat share is refused naming why; a good read serves a second ask within -// 60 s from the cache and is read again after. -func TestSeatsUnreadableAndCached(t *testing.T) { +// 1516/AC4 and ADR 0615: with no first view of the seat-holders yet a seat share is refused naming why; once it is +// answered every ask is served from the picture, with no read of the owner's state and none on a timer. +func TestSeatsUnreadableAndFollowed(t *testing.T) { r := newRig(t) - r.seatsErr = errors.New("the controller did not answer") + r.seatsErr = errors.New("no first view of mesh-delivery-assigner.seat-holders yet") _, err := r.s.link("g14.mesh-files", ask("mesh-shanks-brother", "put", nil)) - if err == nil || !strings.Contains(err.Error(), "the controller did not answer") || !strings.Contains(err.Error(), "node-files") { + if err == nil || !strings.Contains(err.Error(), "no first view") || !strings.Contains(err.Error(), "node-files") { t.Fatalf("not refused naming why: %v", err) } r.seatsErr = nil - put := func() { - t.Helper() + for range 3 { if _, err := r.s.link("g14.mesh-files", ask("mesh-shanks-brother", "put", nil)); err != nil { t.Fatal(err) } + r.now = r.now.Add(10 * time.Minute) // time changes nothing: the picture is current by its events } - put() // a failed read is not cached: this reads again - put() // served from the cache - if r.seatsCalls != 2 { - t.Fatalf("the controller was asked %d times, want 2 (one failure, one good read)", r.seatsCalls) - } - r.now = r.now.Add(59 * time.Second) - put() - if r.seatsCalls != 2 { - t.Fatalf("an ask within 60 s read the seats again (%d calls)", r.seatsCalls) - } - r.now = r.now.Add(2 * time.Second) - put() - if r.seatsCalls != 3 { - t.Fatalf("the seats were not read again after 60 s (%d calls)", r.seatsCalls) + if r.seatsCalls != 4 { + t.Fatalf("the picture was asked %d times, want 4 (one refused, three served)", r.seatsCalls) } } diff --git a/modules/minio/module.json b/modules/minio/module.json index 60b40a4b0..0df4e3615 100644 --- a/modules/minio/module.json +++ b/modules/minio/module.json @@ -39,8 +39,15 @@ ] } ], - "reads": [ - "mesh-delivery-assigner.seat-holders" + "invokes": [ + "mesh-delivery-assigner.assigner_read_seat_holders", + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [ + "mesh-delivery-assigner.views.seat-holders" + ], + "state": [ + "findings" ], "contributes": { "route": { diff --git a/modules/minio/object-store b/modules/minio/object-store new file mode 100755 index 000000000..94048c5f2 Binary files /dev/null and b/modules/minio/object-store differ diff --git a/modules/nfs-server/cmd/nfs-server/main.go b/modules/nfs-server/cmd/nfs-server/main.go index 6edda7de3..7447fafd4 100644 --- a/modules/nfs-server/cmd/nfs-server/main.go +++ b/modules/nfs-server/cmd/nfs-server/main.go @@ -22,7 +22,10 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" + + "git.novox.be/novox/mesh-sdk/go/views" ) // ExportsState is the state the exports are kept in on the bus; MountsState the `mounts` modules' state, @@ -47,8 +50,10 @@ func main() { return } m := Here() + seats.Follow(delivery.AssignerModule, delivery.SeatHoldersState, SeatHoldersTool) + seats.Start(context.Background()) go ShareExports(m) - if err := stdio.Serve("", Tools(m)); err != nil { + if err := stdio.Serve("", append(Tools(m), seats.Tools("nfs_server")...)); err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } @@ -117,25 +122,15 @@ func readingsOf(node string) ([]health.Reading, error) { return out, nil } -// seatHolderKeys reads the placement seat's state of who holds which seat, every key. -func seatHolderKeys() (map[string][]byte, error) { - state := stdio.State(delivery.SeatHoldersRead) - names, err := state.Keys() - if err != nil { - return nil, err - } - keys := map[string][]byte{} - for _, k := range names { - e, err := state.Get(k) - if err != nil { - return nil, err - } - if e != nil { - keys[k] = e.Value - } - } - return keys, nil -} +// SeatHoldersTool is the assigner's read tool of who holds which seat, which the manifest invokes (ADR 0615). +const SeatHoldersTool = delivery.AssignerModule + ".assigner_read_seat_holders" + +// seats is the picture of the placement seat's state of who holds which seat: asked of the assigner once and followed by +// its events, never read from its state (ADR 0615). +var seats = &views.Reader{Module: "nfs-server", Findings: views.OwnFindings{}, Acts: handacts.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "nfs-server: "+f+"\n", a...) }} + +// seatHolderKeys is the placement seat's state of who holds which seat, every key. +func seatHolderKeys() (map[string][]byte, error) { return seats.Values(delivery.SeatHoldersRead) } // TakeWants reads every node's `mounts` state from the bus and writes what is asked of this machine to // the wants file, for the root process. When the bus does not answer, the last wants stay. diff --git a/modules/nfs-server/cmd/nfs-server/manifest_test.go b/modules/nfs-server/cmd/nfs-server/manifest_test.go index ae048eb24..d9d745022 100644 --- a/modules/nfs-server/cmd/nfs-server/manifest_test.go +++ b/modules/nfs-server/cmd/nfs-server/manifest_test.go @@ -2,7 +2,6 @@ package main import ( "encoding/json" - "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" "os" "regexp" @@ -132,16 +131,22 @@ func TestTheDefinitionNamesNoFolderNoMachineAndGivesTheSharesNoDefault(t *testin func TestTheManifestCarriesGrantsTheRosterTheWantsAndTheRemovalOrder(t *testing.T) { m, raw := readManifest(t) var full struct { - Reads []string `json:"reads"` - Facts map[string]map[string]string `json:"facts"` - Config map[string]map[string]any `json:"config"` + Reads []string `json:"reads"` + Invokes []string `json:"invokes"` + Consumes []string `json:"consumes"` + Facts map[string]map[string]string `json:"facts"` + Config map[string]map[string]any `json:"config"` } if err := json.Unmarshal([]byte(raw), &full); err != nil { t.Fatal(err) } - if len(full.Reads) != 3 || full.Reads[0] != health.ReadingsRead || full.Reads[1] != delivery.SeatHoldersRead || full.Reads[2] != MountsState { + // The assigner's seat-holders is asked once and followed by its events (ADR 0615), no longer read. + if len(full.Reads) != 2 || full.Reads[0] != health.ReadingsRead || full.Reads[1] != MountsState { t.Errorf("reads: %v", full.Reads) } + if len(full.Invokes) != 2 || full.Invokes[0] != SeatHoldersTool || full.Invokes[1] != "seat:mesh-hand-acts.hand-act" || len(full.Consumes) != 1 || full.Consumes[0] != "mesh-delivery-assigner.views.seat-holders" { + t.Errorf("invokes %v, consumes %v", full.Invokes, full.Consumes) + } roster := full.Facts["machines"] if roster["path"] != RosterFile || !strings.Contains(roster["template"], "{{range .Machines}}{{.Name}} {{.Address}}") { t.Errorf("the roster fact: %v", roster) diff --git a/modules/nfs-server/module.json b/modules/nfs-server/module.json index c7082176f..8beca270d 100644 --- a/modules/nfs-server/module.json +++ b/modules/nfs-server/module.json @@ -45,11 +45,11 @@ "name": "exports", "ttl-seconds": 120, "per-machine": true - } + }, + "findings" ], "reads": [ "disk-load.readings", - "mesh-delivery-assigner.seat-holders", "mounts.shares" ], "config": { @@ -188,5 +188,12 @@ "from": "files" } ] - } + }, + "invokes": [ + "mesh-delivery-assigner.assigner_read_seat_holders", + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [ + "mesh-delivery-assigner.views.seat-holders" + ] }