diff --git a/contracts/go/seats/meshwants/contract.go b/contracts/go/seats/meshwants/contract.go index e86953fe..7e958b08 100644 --- a/contracts/go/seats/meshwants/contract.go +++ b/contracts/go/seats/meshwants/contract.go @@ -67,8 +67,16 @@ const SubjectAssignmentRevision = "mesh.seat.mesh-wants.event.assignment-revisio // SubjectDifference is where difference is published. const SubjectDifference = "mesh.seat.mesh-wants.event.difference" -// SubjectWanted is where wanted is published. -const SubjectWanted = "mesh.seat.mesh-wants.event.wanted" +// StreamWanted is the stream wanted is kept in. +const StreamWanted = "SEAT_mesh-wants_wanted" + +// SubjectWanted is where wanted is published, on mesh.seat.mesh-wants.event.wanted... +func SubjectWanted(node, module string) string { + return "mesh.seat.mesh-wants.event.wanted." + node + "." + module +} + +// SubjectWantedAll hears every wanted. +const SubjectWantedAll = "mesh.seat.mesh-wants.event.wanted.*.*" // SubjectWaitHeld is where wait-held is published. const SubjectWaitHeld = "mesh.seat.mesh-wants.event.wait-held" @@ -675,7 +683,7 @@ type Wait struct { // the acting part waited for Part string `json:"part,omitempty"` Since time.Time `json:"since"` - // which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498) + // which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706) Step string `json:"step"` // its new version's fingerprint Version string `json:"version,omitempty"` @@ -697,6 +705,7 @@ const ( WaitStepDelivery = "delivery" WaitStepKeptBuild = "kept-build" WaitStepBackup = "backup" + WaitStepOutside = "outside" ) // UpgradeInput is what upgrade takes. diff --git a/contracts/go/seats/meshwants/contract_test.go b/contracts/go/seats/meshwants/contract_test.go index d18fcac6..b21c64c9 100644 --- a/contracts/go/seats/meshwants/contract_test.go +++ b/contracts/go/seats/meshwants/contract_test.go @@ -25,10 +25,10 @@ func TestFixtures(t *testing.T) { }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.assignment-revision$")}, {name: "a difference worked out", subject: "mesh.seat.mesh-wants.event.difference", given: "{\"changes\":1,\"from\":40,\"nodes\":[\"n1\"],\"source\":3589,\"to\":41,\"v\":1}", wire: "{\"changes\":1,\"from\":40,\"nodes\":[\"n1\"],\"source\":3589,\"to\":41,\"v\":1}", decode: func(b []byte) (any, error) { var v DifferenceEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.difference$")}, - {name: "an entry wanted from a trunk revision", subject: "mesh.seat.mesh-wants.event.wanted", given: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", - decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted$")}, - {name: "a want removed, its node left or its module unplaced", subject: "mesh.seat.mesh-wants.event.wanted", given: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", - decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted$")}, + {name: "an entry wanted from a trunk revision", subject: "mesh.seat.mesh-wants.event.wanted.n1.xclip", given: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:00:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"revision\":40,\"source\":3589,\"v\":1}", + decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted\\.[^.*> ]+\\.[^.*> ]+$")}, + {name: "a want removed, its node left or its module unplaced", subject: "mesh.seat.mesh-wants.event.wanted.n1.xclip", given: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", wire: "{\"at\":\"2026-11-09T19:05:00Z\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"removed\":true,\"revision\":41,\"v\":1}", + decode: func(b []byte) (any, error) { var v WantedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wanted\\.[^.*> ]+\\.[^.*> ]+$")}, {name: "a composing wait held with its bound", subject: "mesh.seat.mesh-wants.event.wait-held", given: "{\"bound\":\"30m\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", wire: "{\"bound\":\"30m\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, {name: "a wait ended, released by hand", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"bound\":\"30m\",\"by\":\"lead@shanks: the node is away\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"how\":\"released\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", wire: "{\"bound\":\"30m\",\"by\":\"lead@shanks: the node is away\",\"fp\":\"src2:b427bdde1f100e2f52490cfa9df0c5112dd3700226b4892349f0c14a15a4af56\",\"how\":\"released\",\"module\":\"xclip\",\"node\":\"n1\",\"part\":\"node-engine\",\"since\":\"2026-11-09T19:00:00Z\",\"step\":\"compose\",\"version\":\"src2:7dd0db8e98224f46121a5ae05c0b9c5a430eff531a5120aa40b1a59237853a3f\"}", @@ -55,6 +55,10 @@ func TestFixtures(t *testing.T) { decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, {name: "the builder's toolchain wait ended by a newer fingerprint", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"superseded\",\"module\":\"mesh-sdk\",\"on\":\"toolchain\",\"owner\":\"mesh-delivery-builder\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"build\"}", wire: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"superseded\",\"module\":\"mesh-sdk\",\"on\":\"toolchain\",\"owner\":\"mesh-delivery-builder\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"build\"}", decode: func(b []byte) (any, error) { var v WaitEndedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-ended$")}, + {name: "an outside wait held on a hub's silence, counted from its first start", subject: "mesh.seat.mesh-wants.event.wait-held", given: "{\"bound\":\"30m\",\"clock-from\":\"2026-11-11T10:00:00Z\",\"fingerprint\":\"src2:aaaa\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", wire: "{\"bound\":\"30m\",\"clock-from\":\"2026-11-11T10:00:00Z\",\"fingerprint\":\"src2:aaaa\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", + decode: func(b []byte) (any, error) { var v WaitHeldEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-held$")}, + {name: "an outside wait ended by the operator's Go on", subject: "mesh.seat.mesh-wants.event.wait-ended", given: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"go-on\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", wire: "{\"fingerprint\":\"src2:aaaa\",\"how\":\"go-on\",\"module\":\"nats\",\"node\":\"ace\",\"on\":\"silent:g14\",\"owner\":\"mesh-delivery-assigner\",\"since\":\"2026-11-11T10:00:00Z\",\"step\":\"outside\"}", + decode: func(b []byte) (any, error) { var v WaitEndedEvent; err := json.Unmarshal(b, &v); return v, err }, subjectIs: regexp.MustCompile("^mesh\\.seat\\.mesh-wants\\.event\\.wait-ended$")}, {name: "a want", subject: "", given: "{\"fp\":\"ab12\",\"rule\":3,\"v\":1}", wire: "{\"fp\":\"ab12\",\"rule\":3,\"v\":1}", decode: func(b []byte) (any, error) { var v WantsByWantNodeModuleValue diff --git a/contracts/seats/mesh-wants/fixtures/events.json b/contracts/seats/mesh-wants/fixtures/events.json index 85b2e909..d30e8279 100644 --- a/contracts/seats/mesh-wants/fixtures/events.json +++ b/contracts/seats/mesh-wants/fixtures/events.json @@ -36,7 +36,7 @@ { "name": "an entry wanted from a trunk revision", "of": "event:wanted", - "subject": "mesh.seat.mesh-wants.event.wanted", + "subject": "mesh.seat.mesh-wants.event.wanted.n1.xclip", "given": { "v": 1, "node": "n1", @@ -50,7 +50,7 @@ { "name": "a want removed, its node left or its module unplaced", "of": "event:wanted", - "subject": "mesh.seat.mesh-wants.event.wanted", + "subject": "mesh.seat.mesh-wants.event.wanted.n1.xclip", "given": { "v": 1, "node": "n1", @@ -253,5 +253,36 @@ "since": "2026-11-11T10:00:00Z", "how": "superseded" } + }, + { + "name": "an outside wait held on a hub's silence, counted from its first start", + "of": "event:wait-held", + "subject": "mesh.seat.mesh-wants.event.wait-held", + "given": { + "step": "outside", + "owner": "mesh-delivery-assigner", + "node": "ace", + "module": "nats", + "fingerprint": "src2:aaaa", + "on": "silent:g14", + "since": "2026-11-11T10:00:00Z", + "clock-from": "2026-11-11T10:00:00Z", + "bound": "30m" + } + }, + { + "name": "an outside wait ended by the operator's Go on", + "of": "event:wait-ended", + "subject": "mesh.seat.mesh-wants.event.wait-ended", + "given": { + "step": "outside", + "owner": "mesh-delivery-assigner", + "node": "ace", + "module": "nats", + "fingerprint": "src2:aaaa", + "on": "silent:g14", + "since": "2026-11-11T10:00:00Z", + "how": "go-on" + } } ] diff --git a/contracts/seats/mesh-wants/seat.json b/contracts/seats/mesh-wants/seat.json index 05966ed3..ca64e17c 100644 --- a/contracts/seats/mesh-wants/seat.json +++ b/contracts/seats/mesh-wants/seat.json @@ -303,6 +303,10 @@ }, { "name": "wanted", + "subject": "wanted..", + "keep": { + "last-per-subject": true + }, "body": { "type": "object", "description": "an entry written as a want, from a trunk revision only", @@ -1038,9 +1042,10 @@ "put", "delivery", "kept-build", - "backup" + "backup", + "outside" ], - "description": "which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498)" + "description": "which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706)" }, "part": { "type": "string", diff --git a/contracts/ts/seats/meshwants/index.d.ts b/contracts/ts/seats/meshwants/index.d.ts index 399bdeaa..134842cd 100644 --- a/contracts/ts/seats/meshwants/index.d.ts +++ b/contracts/ts/seats/meshwants/index.d.ts @@ -40,7 +40,11 @@ export declare const assignmentRevisionSubject: "mesh.seat.mesh-wants.event.assi export declare const differenceSubject: "mesh.seat.mesh-wants.event.difference"; -export declare const wantedSubject: "mesh.seat.mesh-wants.event.wanted"; +export declare const wantedStream: "SEAT_mesh-wants_wanted"; + +export declare function wantedSubject(node: string, module: string): string; + +export declare const wantedSubjectAll: "mesh.seat.mesh-wants.event.wanted.*.*"; export declare const waitHeldSubject: "mesh.seat.mesh-wants.event.wait-held"; @@ -328,8 +332,8 @@ export interface Wait { /** the acting part waited for */ "part"?: string; "since": string; - /** which wait: an acting step of section 203 or put, delivery, kept-build, backup (design 47 G section 498) */ - "step": "take-in" | "validate" | "check" | "build" | "compose" | "grant" | "place" | "pipeline" | "put" | "delivery" | "kept-build" | "backup"; + /** which wait: an acting step of section 203 or put, delivery, kept-build, backup, or outside, a moved module's failing check held on an outside cause (design 47 G section 498, section 706) */ + "step": "take-in" | "validate" | "check" | "build" | "compose" | "grant" | "place" | "pipeline" | "put" | "delivery" | "kept-build" | "backup" | "outside"; /** its new version's fingerprint */ "version"?: string; /** the wait in plain words (design 47 G section 512, section 629) */ diff --git a/contracts/ts/seats/meshwants/index.js b/contracts/ts/seats/meshwants/index.js index df1d0f62..15556f82 100644 --- a/contracts/ts/seats/meshwants/index.js +++ b/contracts/ts/seats/meshwants/index.js @@ -40,7 +40,13 @@ export const assignmentRevisionSubject = "mesh.seat.mesh-wants.event.assignment- export const differenceSubject = "mesh.seat.mesh-wants.event.difference"; -export const wantedSubject = "mesh.seat.mesh-wants.event.wanted"; +export const wantedStream = "SEAT_mesh-wants_wanted"; + +export function wantedSubject(node, module) { + return "mesh.seat.mesh-wants.event.wanted." + node + "." + module; +} + +export const wantedSubjectAll = "mesh.seat.mesh-wants.event.wanted.*.*"; export const waitHeldSubject = "mesh.seat.mesh-wants.event.wait-held"; diff --git a/modules/mesh-delivery-assigner/asks/asks.go b/modules/mesh-delivery-assigner/asks/asks.go index 41e01ef1..0ddf4dba 100644 --- a/modules/mesh-delivery-assigner/asks/asks.go +++ b/modules/mesh-delivery-assigner/asks/asks.go @@ -38,6 +38,9 @@ const ( // `assignments.`. The kind needs no proof, as the bus's put does not: its yes is an Approve, which only a // channel proving who answered takes and the view never does (ViewAnswerFits). AskKindAssignments = "assignments" + // AskKindOutside is the kind of the assigner's ask about a moved module whose failing health check has waited past its bound + // on an outside cause (design 47 G §706): go on, or count it as failed. `outside...`. + AskKindOutside = "outside" ) // AskKey is an ask's key read whole: its Kind, and the parts its kind has: a merge's Group and heads Hash, a hold's Module @@ -60,11 +63,11 @@ func AskNeedsProof(kind string) (bool, error) { switch kind { case meshdeliveryclient.AskKindMerge: return true, nil - case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments: + case AskKindHold, AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside: return false, nil } - return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold, - AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments) + return false, fmt.Errorf("asks: %q is not a kind of ask, %s, %s, %s, %s, %s, %s or %s", kind, meshdeliveryclient.AskKindMerge, AskKindHold, + AskKindRestore, AskKindBackup, AskKindRecord, AskKindAssignments, AskKindOutside) } // HoldAskKey is the key of the ask about a module's held fingerprint: `hold..`. @@ -89,6 +92,19 @@ func RestoreAskKey(node, module, fingerprint string) (string, error) { return AskKindRestore + "." + node + "." + key, nil } +// OutsideAskKey is the key of the ask about a module's failing check on a node that waited past its bound on an outside cause: +// `outside...`. +func OutsideAskKey(node, module, fingerprint string) (string, error) { + key, err := RestoreAskKey(node, module, fingerprint) + if err != nil { + return "", err + } + return AskKindOutside + strings.TrimPrefix(key, AskKindRestore), nil +} + +// OutsideAbout is what the outside ask is about: `..` (design 47 G §706). +func OutsideAbout(key string) string { return strings.TrimPrefix(key, AskKindOutside+".") } + // BackupAskKey is the key of the ask about a module's failed backup on a node before it applies a fingerprint: // `backup...`. func BackupAskKey(node, module, fingerprint string) (string, error) { @@ -157,6 +173,13 @@ func ParseAnyAskKey(key string) (AskKey, error) { return refused(err) } return AskKey{Kind: kind, Node: node, Module: module, FP: fp}, nil + case AskKindOutside: + k, err := ParseAnyAskKey(AskKindRestore + "." + rest) + if err != nil { + return refused(fmt.Errorf("an outside ask is outside...")) + } + k.Kind = kind + return k, nil case AskKindBackup: node, module, part, isNode, err := nodeengineclient.ParseComposedKey(rest) if err != nil || isNode { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go new file mode 100644 index 00000000..941d3bc2 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "slices" + "strings" + "time" + + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-sdk/go/keys" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// graphKeeper is the pass's assignment graph (design 47 G §292, A-67): per catalog revision taken in it computes the revision +// from the catalog halves it follows, the placements, the first branch of each registration and each node's capability reading, +// keeps it in the objects assignment-revisions and says the seat mesh-wants' event assignment-revision; the wants of the pass +// that follows are written from the revision it holds. +type graphKeeper struct { + objects assigngraph.Objects + // halves is every followed branch's catalog graph revision held now. + halves func() map[string]mcc.Half + // last is the revision last computed, so a pass whose inputs did not change keeps and says nothing. + last meshwants.AssignmentRevision + has bool + // kept is the revision the wants follow, and kept's source the trunk's source revision it carries; ok is false until one is kept. + kept meshwants.AssignmentRevision + source int64 + ok bool + said string +} + +// keepGraph computes the revision of this pass and keeps it when the pass's inputs differ from the last computed. +func (p *passer) keepGraph(ctx context.Context) { + g := p.graph + if g == nil || g.objects == nil || g.halves == nil { + return + } + in := assigngraph.Inputs{Halves: g.halves(), Placements: p.w.assignments(), Capabilities: p.capabilityDigests()} + // With no pipeline ends kept, the wants are the trunk-only form (pass.go): a placement naming no branch runs the trunk's. + in.FirstBranch = func(string) string { return trunkBranch } + if p.ends != nil { + in.FirstBranch = func(module string) string { return p.ends.branchOf(module, "") } + } + c := assigngraph.Compute(in) + for _, u := range c.Uncomputed { + if g.said != u { + g.said = u + p.logf("the assignment graph leaves out %s", u) + } + } + if g.has && g.ok && assigngraph.Same(g.last, c.Revision) { + return + } + rev, fresh, err := assigngraph.Keep(ctx, g.objects, c, meshwants.EventAssignmentRevision, p.clock()) + if err != nil { + p.logf("the assignment graph revision is not kept, so no want is said as wanted until it is: %v", err) + return + } + g.last, g.has, g.kept, g.ok = c.Revision, true, rev, true + g.source = 0 + for _, cat := range rev.Catalogs { + if cat.Branch == trunkBranch { + g.source = cat.Source + } + } + if !fresh { + return + } + id := fmt.Sprintf("%s-%s-%d", meshwants.EventAssignmentRevision, rev.Kind, rev.Revision) + if err := p.wantsSay(meshwants.EventAssignmentRevision, assigngraph.Event(rev), id); err != nil { + p.logf("the assignment revision %d is kept but not said: %v", rev.Revision, err) + } +} + +// capabilityDigests is the digest of each node's capability reading, the statement its engine made, by node. +func (p *passer) capabilityDigests() []meshwants.AssignmentRevisionCapabilitiesItem { + if p.reads == nil { + return nil + } + known := p.w.knownNodes() + slices.Sort(known) + var out []meshwants.AssignmentRevisionCapabilitiesItem + for _, node := range known { + if !p.reads.Known(node) { + continue + } + reads := slices.Clone(p.reads.Reads([]string{node})[node]) + slices.Sort(reads) + raw, _ := json.Marshal(reads) + sum := sha256.Sum256(raw) + out = append(out, meshwants.AssignmentRevisionCapabilitiesItem{Node: node, Digest: "sha256:" + hex.EncodeToString(sum[:])}) + } + return out +} + +// sayWanted says `wanted` for a want written or removed now, from the revision the pass keeps (A-61, A-67); with none kept, it is +// said in the log that it was not said. fp is the want's, and for a removal the fingerprint last wanted there. +func (p *passer) sayWanted(node, module, fp string, removed bool) { + if p.publishWants == nil { + return + } + g := p.graph + if g == nil || !g.ok { + p.logf("the want of %s on %s is written, but no assignment graph revision is kept, so wanted is not said", module, node) + return + } + at := p.clock().UTC() + body := meshwants.WantedEvent{V: keys.Version, Node: node, Module: module, Fp: fp, Removed: removed, Revision: g.kept.Revision, Source: g.source, At: &at} + event := strings.TrimPrefix(meshwants.SubjectWanted(keys.EscapeName(node), keys.EscapeName(module)), "mesh.seat."+meshwants.Seat+".event.") + id := fmt.Sprintf("wanted.%s.%s.%s.%t.%d.%s", keys.EscapeName(node), keys.EscapeName(module), fp, removed, g.kept.Revision, at.Format(time.RFC3339Nano)) + if err := p.wantsSay(event, body, id); err != nil { + p.logf("wanted of %s on %s is not said: %v", module, node, err) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go new file mode 100644 index 00000000..a40924b9 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go @@ -0,0 +1,80 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/buckets" +) + +// said is one event the passer said on the seat mesh-wants. +type said struct { + event, id string + body json.RawMessage +} + +// Design 47 G A-67, §292: the pass keeps an assignment graph revision per catalog revision taken in, before it writes any want, and +// says assignment-revision; A-61, §693: a want written or removed is said `wanted` on the subject of its node and module in the same +// write, from the kept revision and its catalog's source, a put that wrote nothing says nothing, and with no revision kept none is said. +func TestTheWantsAreSaidWantedFromTheKeptAssignmentGraphRevision(t *testing.T) { + ctx := context.Background() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + kv := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: kv}, at) + p.logf = t.Logf + var events []said + p.publishWants = func(event string, body any, id string) error { + raw, _ := json.Marshal(body) + events = append(events, said{event, id, raw}) + return nil + } + placeOn(t, p.w, "ace", "web") + half := mcc.Half{Revision: meshcatalog.CatalogRevision{Branch: trunkBranch, Kind: "trunk", Revision: 4, Source: 9, + Entries: []meshcatalog.EntryRef{{Module: "web", Digest: "sha256:" + strings.Repeat("a", 64)}}}, + Entries: map[string]meshcatalog.CatalogEntry{"web": {Module: "web", Fp: fpTest}}} + objects := memObjects{} + p.graph = &graphKeeper{objects: objects, halves: func() map[string]mcc.Half { return map[string]mcc.Half{trunkBranch: half} }} + want := mwc.Want{Fp: fpTest, Self: true} + + p.putWant(ctx, "ace", "web", want) // no revision kept yet: written, and not said + if len(events) != 0 || kv.puts != 1 { + t.Fatalf("with no revision kept: %d events, %d puts", len(events), kv.puts) + } + key, _ := mwc.WantKey("ace", "web") + delete(p.w.read(buckets.Wants), key) // the world forgets the want, as before the first pass + kv.puts = 0 + p.w.put(buckets.Wants, key, nil) + + p.keepGraph(ctx) + if len(events) != 1 || events[0].event != meshwants.EventAssignmentRevision || len(objects) == 0 { + t.Fatalf("the pass kept %d objects and said %+v", len(objects), events) + } + p.keepGraph(ctx) // the same inputs: nothing kept, nothing said + if len(events) != 1 { + t.Fatalf("a pass of the same inputs said %d events", len(events)) + } + p.putWant(ctx, "ace", "web", want) + p.putWant(ctx, "ace", "web", want) // unchanged: not written, not said + if len(events) != 2 || kv.puts != 1 { + t.Fatalf("a want put twice: %d events, %d puts", len(events), kv.puts) + } + if wantEvent := "wanted.ace.web"; events[1].event != wantEvent { + t.Errorf("wanted said on %q, want %q", events[1].event, wantEvent) + } + var w meshwants.WantedEvent + if err := json.Unmarshal(events[1].body, &w); err != nil || w.Node != "ace" || w.Module != "web" || w.Fp != fpTest || w.Removed || w.Revision != 1 || w.Source != 9 { + t.Errorf("wanted = %+v (%v)", w, err) + } + p.putWant(ctx, "ace", "web", mwc.Want{Removed: true}) + if err := json.Unmarshal(events[len(events)-1].body, &w); err != nil || len(events) != 3 || !w.Removed || w.Fp != fpTest { + t.Errorf("a removing want said %+v (%v), %d events", w, err, len(events)) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go index b57d1575..40fd4ee2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go @@ -110,3 +110,16 @@ func fromEntry(e meshcatalog.CatalogEntry) (meshcatalog.ModulesByModuleValue, gr n.Serves = graph.ServesOf(man) return v, n, &parsed{module: e.Module, manifest: man} } + +// views is the catalog graph revision held now of each followed branch whose first view is answered. +func (h *halves) views() map[string]mcc.Half { + h.mu.Lock() + defer h.mu.Unlock() + out := make(map[string]mcc.Half, len(h.cats)) + for branch, c := range h.cats { + if half, ok := c.View(); ok { + out[branch] = half + } + } + return out +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go new file mode 100644 index 00000000..0bbb0bf7 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/inputs.go @@ -0,0 +1,117 @@ +package main + +import ( + "crypto/sha256" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + + meshwants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// A step's input by its digest (design 47 G §354): the assigner keeps the input of the composition step it asked a build seat to +// run in its state `inputs`, one key `