Join: merge A8 (wip/registry-a8)
Conflict in mesh-conditions keeper.go Observers and module.json reads: origin/main added mesh-vault's findings, A8 added the first-view readers' findings; both kept.
This commit is contained in:
@@ -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()
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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+"."))
|
||||
|
||||
@@ -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": {
|
||||
|
||||
+23
-3
@@ -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<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),
|
||||
// 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<unknown>("findings").put(f.key, f.value);
|
||||
},
|
||||
clear: async () => {
|
||||
await state<unknown>("findings").delete(unansweredFinding("gitea", INDEXER_OWNER, "modules", INDEXER_MODULES_TOOL, new Date(), "", new Date()).key);
|
||||
} });
|
||||
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}`);
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -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<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 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<string, unknown>[] = [];
|
||||
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\)/);
|
||||
});
|
||||
@@ -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<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;
|
||||
/** 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<void>;
|
||||
clear?: () => void | Promise<void>;
|
||||
}
|
||||
|
||||
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`);
|
||||
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<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,
|
||||
record: (act: Record<string, unknown>) => Promise<unknown>): 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<string, unknown> } {
|
||||
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(),
|
||||
} };
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<void> | undefined;
|
||||
let askAgain = false;
|
||||
|
||||
const manifestOf = async (digest: string | undefined): Promise<unknown> => {
|
||||
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<unknown> => {
|
||||
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<void> => {
|
||||
|
||||
@@ -89,7 +89,6 @@
|
||||
"mesh-delivery-builder.builder_builds_since"
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-indexer.manifests"
|
||||
"mesh-delivery-builder.builds"
|
||||
]
|
||||
}
|
||||
|
||||
@@ -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"],
|
||||
|
||||
@@ -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 `<module>.<name>`.
|
||||
const (
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
Executable
BIN
Binary file not shown.
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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": {
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Executable
BIN
Binary file not shown.
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user