diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go index 3fb08808..5a207e22 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/graph_test.go @@ -345,3 +345,56 @@ func TestARunsCandidateGraphIsKeptApartServedAndDropped(t *testing.T) { t.Error("dropping a candidate dropped trunk") } } + +// Design 47 G §512, §186, A-242: the difference orders its changes itself: each change names the changes it is ordered after +// with the kind of the edge, and Tiers holds the changes in that order, an acting part and a provider before what they act on. +func TestTheDifferenceFillsItsTiersAndEachChangesAfter(t *testing.T) { + ctx := context.Background() + build := func(ne, pg, web string) Computed { + return Compute(Inputs{ + Halves: map[string]meshcatalogclient.Half{"main": half("main", 7, + meshcatalog.CatalogEntry{Module: "node-engine", Fp: ne, Acts: []meshcatalog.ActingRule{{Seat: "node-engine", Step: "compose", On: "every"}}}, + meshcatalog.CatalogEntry{Module: "postgres", Fp: pg, Provides: []string{"database"}}, + meshcatalog.CatalogEntry{Module: "web", Fp: web, Requires: []string{"database"}}, + )}, + Placements: map[string]mma.Assignments{"anchor": {Node: "anchor", Modules: []mma.PlacedModule{{Module: "node-engine"}, {Module: "postgres"}, + {Module: "web", Bindings: map[string]mma.Provider{"database": {Node: "anchor", Module: "postgres"}}}}}}, + FirstBranch: func(string) string { return "main" }, + }) + } + o := memObjects{} + at := time.Date(2026, 11, 12, 1, 0, 0, 0, time.UTC) + if _, _, err := Keep(ctx, o, build("src2:e1", "src2:p1", "src2:w1"), "catalog-revision", at); err != nil { + t.Fatal(err) + } + if _, _, err := Keep(ctx, o, build("src2:e2", "src2:p2", "src2:w2"), "catalog-revision", at); err != nil { + t.Fatal(err) + } + d, err := Difference(ctx, o, meshwants.DifferenceInput{From: 1, To: 2}) + if err != nil || len(d.Changes) != 3 { + t.Fatalf("difference %+v, %v", d, err) + } + after := map[string]string{} + for _, c := range d.Changes { + var a []string + for _, x := range c.After { + a = append(a, x.Kind+":"+x.Node+"/"+x.Module) + } + after[c.Module] = strings.Join(a, ",") + } + if after["node-engine"] != "" || after["postgres"] != "acting:anchor/node-engine" || + after["web"] != "acting:anchor/node-engine,provisioning:anchor/postgres" { + t.Errorf("after = %v", after) + } + var tiers []string + for _, tier := range d.Tiers { + var ms []string + for _, r := range tier { + ms = append(ms, r.Module) + } + tiers = append(tiers, strings.Join(ms, ",")) + } + if strings.Join(tiers, "|") != "node-engine|postgres|web" { + t.Errorf("tiers = %v", tiers) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go index 525fee58..f8e6b63f 100644 --- a/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go +++ b/modules/mesh-delivery-assigner/internal/assigngraph/verbs.go @@ -6,6 +6,7 @@ import ( "fmt" "maps" "slices" + "strings" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" ) @@ -110,6 +111,7 @@ func Compare(ctx context.Context, o Objects, from, to meshwants.AssignmentRevisi return d, err } nodes := map[string]bool{} + edges := map[string][]meshwants.Edge{} for _, k := range slices.Sorted(maps.Keys(union(was, now))) { a, hadA := was[k] b, hasB := now[k] @@ -126,11 +128,95 @@ func Compare(ctx context.Context, o Objects, from, to meshwants.AssignmentRevisi c.Edges = moved(a.Edges, b.Edges) d.Changes = append(d.Changes, c) nodes[c.Node] = true + if hasB { + edges[c.Node+"/"+c.Module] = b.Edges + } else { + edges[c.Node+"/"+c.Module] = a.Edges + } } d.Nodes = slices.Sorted(maps.Keys(nodes)) + order(&d, edges, digests(from, to)) return d, nil } +// digests are the entry digests of the two revisions by node and module, the newer's winning. +func digests(from, to meshwants.AssignmentRevision) map[string]string { + out := map[string]string{} + for _, r := range []meshwants.AssignmentRevision{from, to} { + for _, ref := range r.Entries { + out[ref.Node+"/"+ref.Module] = ref.Digest + } + } + return out +} + +// order fills each change's After and the difference's Tiers (design 47 G §512, §186, A-242): a change is ordered after +// each other change of the difference that an edge of its entry names (the module version it is provisioned by, contributes +// to or is acted on by), with that edge's kind; its tier is one past the highest tier it is after. Changes on a cycle come in +// the last tier. The view orders nothing itself. +func order(d *meshwants.Difference, edges map[string][]meshwants.Edge, digest map[string]string) { + if len(d.Changes) == 0 { + return + } + in := map[string]int{} + for i, c := range d.Changes { + in[c.Node+"/"+c.Module] = i + } + for i, c := range d.Changes { + self := c.Node + "/" + c.Module + seen := map[string]bool{} + for _, e := range edges[self] { + k := e.Node + "/" + e.Module + if _, ok := in[k]; !ok || k == self || seen[e.Kind+k] { + continue + } + seen[e.Kind+k] = true + d.Changes[i].After = append(d.Changes[i].After, meshwants.ChangeAfterItem{Node: e.Node, Module: e.Module, Kind: e.Kind}) + } + slices.SortFunc(d.Changes[i].After, func(a, b meshwants.ChangeAfterItem) int { + return strings.Compare(a.Kind+"\x00"+a.Node+"\x00"+a.Module, b.Kind+"\x00"+b.Node+"\x00"+b.Module) + }) + } + tier := map[int]int{} + var level func(int, map[int]bool) int + level = func(i int, visiting map[int]bool) int { + if t, ok := tier[i]; ok { + return t + } + if visiting[i] { + return -1 + } + visiting[i] = true + t := 0 + for _, a := range d.Changes[i].After { + l := level(in[a.Node+"/"+a.Module], visiting) + if l < 0 { + return -1 + } + t = max(t, l+1) + } + delete(visiting, i) + tier[i] = t + return t + } + top := 0 + for i := range d.Changes { + top = max(top, level(i, map[int]bool{})) + } + for i := range d.Changes { + if _, ok := tier[i]; !ok { + tier[i] = top + 1 + } + } + for i, c := range d.Changes { + for len(d.Tiers) <= tier[i] { + d.Tiers = append(d.Tiers, nil) + } + d.Tiers[tier[i]] = append(d.Tiers[tier[i]], meshwants.NodeEntryRef{Node: c.Node, Module: c.Module, Digest: digest[c.Node+"/"+c.Module]}) + } + d.Tiers = slices.DeleteFunc(d.Tiers, func(t []meshwants.NodeEntryRef) bool { return len(t) == 0 }) +} + func entriesOf(ctx context.Context, o Objects, rev meshwants.AssignmentRevision) (map[string]meshwants.AssignmentEntry, error) { out := make(map[string]meshwants.AssignmentEntry, len(rev.Entries)) for _, ref := range rev.Entries { diff --git a/modules/mesh-delivery-contracts/go/seats/meshwants/contract.go b/modules/mesh-delivery-contracts/go/seats/meshwants/contract.go index 983bf096..f8dd9332 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshwants/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshwants/contract.go @@ -523,8 +523,18 @@ type CatalogOf struct { Source int64 `json:"source,omitempty"` } +// ChangeAfterItem is a type of the contract. +type ChangeAfterItem struct { + // the kind of the edge that orders the change after that one: provisioning, contribution or acting + Kind string `json:"kind"` + Module string `json:"module"` + Node string `json:"node"` +} + // Change is one module on a node whose version differs between two revisions. type Change struct { + // the other changes of the difference it is ordered after, each with its edge's kind (design 47 G section 512) + After []ChangeAfterItem `json:"after,omitempty"` // the edges that moved Edges []Edge `json:"edges,omitempty"` // the fingerprint before; absent when it is new there diff --git a/modules/mesh-delivery-contracts/seats/mesh-wants/seat.json b/modules/mesh-delivery-contracts/seats/mesh-wants/seat.json index 20c99c43..9a0f5535 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-wants/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-wants/seat.json @@ -1028,6 +1028,30 @@ "$ref": "#/defs/Edge" }, "description": "the edges that moved" + }, + "after": { + "type": "array", + "items": { + "type": "object", + "properties": { + "node": { + "type": "string" + }, + "module": { + "type": "string" + }, + "kind": { + "type": "string", + "description": "the kind of the edge that orders the change after that one: provisioning, contribution or acting" + } + }, + "required": [ + "node", + "module", + "kind" + ] + }, + "description": "the other changes of the difference it is ordered after, each with its edge's kind (design 47 G section 512)" } }, "required": [ diff --git a/modules/mesh-delivery-contracts/ts/seats/meshwants/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshwants/index.d.ts index bd830665..1380f032 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshwants/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshwants/index.d.ts @@ -192,8 +192,17 @@ export interface CatalogOf { "source"?: number; } +export interface ChangeAfterItem { + /** the kind of the edge that orders the change after that one: provisioning, contribution or acting */ + "kind": string; + "module": string; + "node": string; +} + /** one module on a node whose version differs between two revisions */ export interface Change { + /** the other changes of the difference it is ordered after, each with its edge's kind (design 47 G section 512) */ + "after"?: ChangeAfterItem[]; /** the edges that moved */ "edges"?: Edge[]; /** the fingerprint before; absent when it is new there */ diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go index c2c8770a..b31a8250 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-cli/main_test.go @@ -22,7 +22,6 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" ) @@ -132,19 +131,18 @@ func mustHash(t *testing.T, h meshdeliveryclient.Head) string { // the gate's, as mesh-sdk go/delivery EncodeAsk writes it under the key MergeAskKey gives (as internal/view asks_test.go). func TestPrintsTheOpenAsksInTheTableAndTheJSON(t *testing.T) { // 1345/AC2 m := view.New() - heads := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}} - key, err := meshdeliveryclient.MergeAskKey("feat/x", heads) + key, err := asks.HoldAskKey("plex", "src2:2222222222222222222222222222222222222222222222222222222222222222") if err != nil { t.Fatal(err) } - ask := oc.Ask{ID: meshdeliveryclient.AskID(key), Headline: "Merge into main: mesh-delivery#12", Explanation: "Merges one pull request.", Who: occ.Operator, - Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "the merge stays held", - Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}} - value, err := meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: "feat/x", Heads: askHeads(heads), Ask: rawAsk(t, ask)}) + ask := oc.Ask{ID: meshdeliveryclient.AskID(key), Headline: "plex is held after a failed health check", Explanation: "ace failed the check on the new build.", Who: occ.Operator, + Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "the fingerprint stays held", + Options: []oc.Option{{ID: "release", Label: "Release", Does: "goes on", Level: occ.Approve}, {ID: "keep", Label: "Keep", Does: "stays", Level: occ.Approve}}} + value, err := asks.EncodeAsk(key, asks.Ask{Ask: rawAsk(t, ask)}) if err != nil { t.Fatal(err) } - m.Apply(view.Entry{Bucket: view.AsksBucket, Key: key, Value: value}) + m.Apply(view.Entry{Bucket: view.AssignerAsksBucket, Key: key, Value: value}) dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "c") if err != nil { t.Fatal(err) @@ -162,7 +160,7 @@ func TestPrintsTheOpenAsksInTheTableAndTheJSON(t *testing.T) { // 1345/AC2 if code := run(nil, socket, &out, &errOut); code != 0 { t.Fatalf("exit %d: %s", code, errOut.String()) } - for _, want := range []string{"ASKS", "Merge into main: mesh-delivery#12", "Merges one pull request.", "answer on the phone"} { + for _, want := range []string{"ASKS", "plex is held after a failed health check", "ace failed the check on the new build.", view.AnswerInTheView} { if !strings.Contains(out.String(), want) { t.Errorf("the table lacks %q:\n%s", want, out.String()) } @@ -179,7 +177,7 @@ func TestPrintsTheOpenAsksInTheTableAndTheJSON(t *testing.T) { // 1345/AC2 } `json:"asks"` } if err := json.Unmarshal(out.Bytes(), &got); err != nil || len(got.Asks) != 1 || got.Asks[0].Key != key || - got.Asks[0].Headline != "Merge into main: mesh-delivery#12" || got.Asks[0].How != "answer on the phone" { + got.Asks[0].Headline != "plex is held after a failed health check" || got.Asks[0].How != view.AnswerInTheView { t.Errorf("json = %s (%v)", out.String(), err) } } @@ -221,16 +219,9 @@ func TestTheAnswerCommandAnswersAnAskNeedingNoProof(t *testing.T) { // 1297/B9b if err != nil { t.Fatal(err) } - merge := oc.Ask{ID: meshdeliveryclient.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: occ.Operator, Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "held", - Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}} - mv, err := meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: "feat/x", Heads: askHeads(heads), Ask: rawAsk(t, merge)}) - if err != nil { - t.Fatal(err) - } m, w := view.New(), &putter{puts: map[string][]byte{}} m.SetWriter(w) m.Apply(view.Entry{Bucket: view.AssignerAsksBucket, Key: key, Value: value}) - m.Apply(view.Entry{Bucket: view.AsksBucket, Key: mkey, Value: mv}) socket := serveWith(t, m) var out, errOut bytes.Buffer @@ -309,12 +300,3 @@ func rawAsk(t *testing.T, a oc.Ask) json.RawMessage { } return b } - -// askHeads is the heads of a merge ask as the seat carries them. -func askHeads(heads []meshdeliveryclient.Head) []meshdelivery.AsksByMergeGroupHeadsValueHeadsItem { - out := make([]meshdelivery.AsksByMergeGroupHeadsValueHeadsItem, len(heads)) - for i, h := range heads { - out[i] = meshdelivery.AsksByMergeGroupHeadsValueHeadsItem{Repository: h.Repository, Number: int64(h.Number), Head: h.Head} - } - return out -} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go index 2ec90638..cdddb592 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/catalogread.go @@ -1,65 +1,55 @@ package main import ( - "context" "encoding/json" "fmt" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "sort" - "sync" - "time" "mesh-delivery-view/internal/view" ) -// catalogWithin is how long a call waits for the seat mesh-catalog's first view before it says there is none. -var catalogWithin = 30 * time.Second - -// catalogRead is the modules the view's tools answer, read from the catalog half through the seat mesh-catalog's verb -// `revision` (design 47, Two graphs): the trunk's newest revision, asked once per call and in pages, each entry in the -// form of the value the tools always answered. With no view the read says so; it never answers no modules. -func catalogRead(ask func(key string, body any) (json.RawMessage, error)) view.StateRead { - var ( - once sync.Once - values map[string]json.RawMessage - rev uint64 - failed error - ) - load := func() { - ctx, cancel := context.WithTimeout(context.Background(), catalogWithin) - defer cancel() - c, err := meshcatalogclient.NewCatalog(ask, "main") - if err != nil { - failed = err - return +// catalogRead is the modules the view's tools answer, read from the catalog half per watched branch through +// meshcatalogclient's branch follower (design 47 G §532, §614): every module a followed branch's view lists, each the +// entry Branches.Entry reads, the first branch of the module's own registration (A-103), in the form of the value the +// tools always answered. Until every watched branch has its first view the read says what it waits for; it never answers +// no modules. +func catalogRead(b *meshcatalogclient.Branches) view.StateRead { + read := func() (map[string]json.RawMessage, error) { + if err := b.Ready(); err != nil { + return nil, fmt.Errorf("%w (%s)", err, meshcatalogclient.RevisionVerb) } - if err := c.Start(ctx); err != nil { - failed = fmt.Errorf("the catalog half has no first view yet (%s): %w", meshcatalogclient.RevisionVerb, err) - return - } - half, ok := c.View() - if !ok { - failed = fmt.Errorf("the catalog half has no first view yet (%s)", meshcatalogclient.RevisionVerb) - return - } - rev, values = uint64(half.Revision.Revision), map[string]json.RawMessage{} - for module, e := range half.Entries { - raw, _ := json.Marshal(e) - v, err := view.ModuleValue(raw) - if err != nil { - failed = fmt.Errorf("the catalog entry of %s: %w", module, err) - return + values := map[string]json.RawMessage{} + for _, branch := range b.Followed() { + c, ok := b.Catalog(branch) + if !ok { + continue + } + half, _ := c.View() + for module := range half.Entries { + if _, done := values[module]; done { + continue + } + e, ok, err := b.Entry(module) + if err != nil { + return nil, err + } + if !ok { + continue + } + raw, _ := json.Marshal(e) + v, err := view.ModuleValue(raw) + if err != nil { + return nil, fmt.Errorf("the catalog entry of %s: %w", module, err) + } + values[module] = v } - values[module] = v } - } - read := func() (map[string]json.RawMessage, uint64, error) { - once.Do(load) - return values, rev, failed + return values, nil } return view.StateRead{ Keys: func() ([]string, error) { - v, _, err := read() + v, err := read() keys := make([]string, 0, len(v)) for k := range v { keys = append(keys, k) @@ -68,16 +58,16 @@ func catalogRead(ask func(key string, body any) (json.RawMessage, error)) view.S return keys, err }, Get: func(key string) (json.RawMessage, error) { - v, _, err := read() - return v[key], err - }, - GetEntry: func(key string) (json.RawMessage, uint64, error) { - v, r, err := read() - return v[key], r, err - }, - All: func() (map[string]json.RawMessage, error) { - v, _, err := read() - return v, err + if err := b.Ready(); err != nil { + return nil, fmt.Errorf("%w (%s)", err, meshcatalogclient.RevisionVerb) + } + e, ok, err := b.Entry(key) + if err != nil || !ok { + return nil, err + } + raw, _ := json.Marshal(e) + return view.ModuleValue(raw) }, + All: read, } } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 9be22969..55d40c03 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -4,6 +4,7 @@ package main import ( + "context" "encoding/json" "fmt" "mesh-delivery-view/internal/view" @@ -17,6 +18,7 @@ import ( "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" @@ -395,11 +397,11 @@ func read(state stdio.KeptState) view.StateRead { // ownersRead is the view's reading of the indexer's and the assigner's states: each through its owner's read tool, asked // by the bundle on each call (ADR 0615 §1, design 25 § 12.2: the view keeps its tools and answers them by asking the // owners), never their buckets. -func ownersRead(ask func(key string, body any) (json.RawMessage, error)) view.Reader { +func ownersRead(ask func(key string, body any) (json.RawMessage, error), catalog *meshcatalogclient.Branches) view.Reader { // Each owner's read tool as the view's invokes names it. indexer := func(tool string) view.StateRead { return view.OwnerRead(ask, view.IndexerModule+"."+tool) } assigner := func(tool string) view.StateRead { return view.OwnerRead(ask, view.AssignerModule+"."+tool) } - return view.Reader{State: catalogRead(ask), CandidatesState: indexer("indexer_read_candidates"), + return view.Reader{State: catalogRead(catalog), CandidatesState: indexer("indexer_read_candidates"), WantsState: assigner("assigner_read_wants"), DifferencesState: assigner("assigner_read_differences"), HeardState: indexer("indexer_read_heard"), PredefinedState: indexer("indexer_read_predefined"), @@ -418,13 +420,26 @@ func holderOf(ask func(key string, body any) (json.RawMessage, error)) func(seat } func main() { - r := ownersRead(stdio.Ask) + logf := func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] "+f+"\n", a...) } + // The catalog half per watched branch, followed by the bundle itself through meshcatalogclient's branch follower on + // its own views reader (design 47 G §532, §614, A-103), as other bundles follow the views they read. + cat := &views.Reader{Module: viewstate.ViewModule, Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + branches, err := meshcatalogclient.FollowBranches(cat, stdio.Ask) + if err != nil { + logf("%v", err) + os.Exit(1) + } + branches.Logf = logf + cat.Start(context.Background()) + r := ownersRead(stdio.Ask, branches) r.Builds, r.HavesState = read(stdio.State("mesh-delivery-builder.builds")), read(stdio.State("node-engine.have")) r.ReachState = read(stdio.State("reach")) + r.WaitsState = read(stdio.State(viewstate.WaitsState)) r.SignersState = read(stdio.State("mesh-vault.signers")) r.ConditionsState, r.AttemptsState = read(stdio.State(view.StatusConditionsState)), read(stdio.State(view.StatusAttemptsState)) r.PresenceState, r.StuckState = read(stdio.State(view.StatusPresenceState)), read(stdio.State(view.StatusStuckState)) - r.SourceState = read(stdio.State(view.GateModule + ".source")) + r.Sources = view.SourcesOf(stdio.Ask, branches.Followed) + r.Assignment, r.Difference = view.AssignmentOf(stdio.Ask), view.DifferenceOf(stdio.Ask) r.ProvisionsState = read(stdio.State(view.ProvisionsRead)) r.BuildLog = stdio.Log(builds.BuildLogRead).Since r.Pending = pendingOf @@ -449,6 +464,7 @@ func main() { // view_first_views and view_first_view_again: where the view's process stands with its first views of the indexer's and // the assigner's states (ADR 0615 §6), and a re-read by hand. tools := append(Tools(r, u), gate.Tools()...) + tools = append(tools, AnswersReadTool(func(name string) views.Source { return views.KeptSource{State: stdio.State(name)} }, time.Now)) tools = append(tools, views.StatusTools(viewstate.ViewModule, "view_first_views", "view_first_view_again", stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err) @@ -456,6 +472,14 @@ func main() { } } +// AnswersReadTool is view_read_answers, the view's ordinary read tool of its own state answers (mesh-sdk views.ReadTool): the +// teller's first view of the answers the view says on views.answers (design 47 G A-175, A-178). +func AnswersReadTool(state func(string) views.Source, now func() time.Time) stdio.Tool { + return views.ReadTool(viewstate.ViewModule, viewstate.ViewAnswersReadTool, viewstate.ViewAnswersState, + "the operator's answers given in the view to the assigner's asks, one key .view each, which the teller fills the ask from.", + state(viewstate.ViewAnswersState), state(views.MarksState), now) +} + // pendingOf asks the indexer's tool indexer_pending (declared in invokes): what it holds back and why (ADR 0633 §5). func pendingOf() (takein.PendingAnswer, error) { raw, err := stdio.Ask(view.IndexerModule+".indexer_pending", map[string]any{}) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go index a4883180..1ace4595 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "fmt" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" @@ -12,6 +13,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/graph" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -88,9 +90,9 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { } asked := map[string]int{} ask := func(key string, body any) (json.RawMessage, error) { - if key == meshcatalogclient.RevisionVerb { + if a, ok := catalogSeats(key, body); ok { asked[key]++ - return json.Marshal(catalogAnswer) + return a() } name, ok := tools[key] if !ok { @@ -116,7 +118,7 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { } return json.Marshal(a) } - owned := ownersRead(ask) + owned := ownersRead(ask, startedBranches(t, context.Background(), ask)) for _, row := range []struct { tool string args map[string]any @@ -146,7 +148,7 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { // the form of the value the tool always answered, the modules without a fingerprint say why, and the module's history // is not a thing the catalog half keeps. a, err := run(t, owned, view.ModuleReadTool, map[string]any{"module": "a"}) - if raw, _ := json.Marshal(a); err != nil || !strings.Contains(string(raw), `"fp2":"`+fp2A+`"`) || !strings.Contains(string(raw), `"commit":"c1"`) || !strings.Contains(string(raw), `"src":7`) { + if raw, _ := json.Marshal(a); err != nil || !strings.Contains(string(raw), `"fp2":"`+fp2A+`"`) || !strings.Contains(string(raw), `"commit":"c1"`) || !strings.Contains(string(raw), `"src":7`) || !strings.Contains(string(raw), `"manifest":"`+manifestDigestA+`"`) { t.Errorf("indexer_module a = %s, %v", raw, err) } if _, err := run(t, owned, view.ModuleReadTool, map[string]any{"module": "gone"}); err == nil { @@ -168,9 +170,11 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { t.Errorf("the want's history through the assigner's read tool = %+v, %v", revs, err) } // With no view of the catalog half the modules are not read: the tools say so, never that no module is held. - catalogWithin = 20 * time.Millisecond - none := ownersRead(func(key string, body any) (json.RawMessage, error) { return nil, fmt.Errorf("the holder is away") }) - if out, err := run(t, none, view.ModulesReadTool, map[string]any{}); err == nil || !strings.Contains(err.Error(), "no first view") { + away := func(key string, body any) (json.RawMessage, error) { return nil, fmt.Errorf("the holder is away") } + gone, cancel := context.WithCancel(context.Background()) + cancel() + none := ownersRead(away, startedBranches(t, gone, away)) + if out, err := run(t, none, view.ModulesReadTool, map[string]any{}); err == nil || !strings.Contains(err.Error(), "no watched branch") { t.Errorf("indexer_modules with no view = %v, %v, want an error naming it", out, err) } // Each owner state the tools read was asked of its owner's read tool. @@ -179,17 +183,96 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { t.Errorf("%s was never asked", tool) } } - if asked[meshcatalogclient.RevisionVerb] == 0 { - t.Errorf("%s was never asked", meshcatalogclient.RevisionVerb) + for _, verb := range []string{meshcatalogclient.RevisionVerb, meshcatalogclient.RegistrationsVerb, meshcatalogclient.SourceVerb} { + if asked[verb] == 0 { + t.Errorf("%s was never asked", verb) + } } } +// catalogBranch is the one watched branch of the catalog half in these tests: the registrations name it, the gate's source +// keeps it, and the seat's revision answers it alone, so a reader that names a branch of its own reads nothing. +const catalogBranch = "release" + +// catalogSeats answers the seats the branch follower asks (design 47 G §614): the registrations, each module on +// catalogBranch, the gate's source, and the newest trunk revision of catalogBranch; ok is false for any other key. +func catalogSeats(key string, body any) (func() (json.RawMessage, error), bool) { + switch key { + case meshcatalogclient.RegistrationsVerb: + return func() (json.RawMessage, error) { + return json.Marshal(meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{ + {Module: "a", Version: 1, Branches: []string{catalogBranch}}, {Module: "b", Version: 1, Branches: []string{catalogBranch}}}}) + }, true + case meshcatalogclient.SourceVerb: + return func() (json.RawMessage, error) { return json.Marshal(map[string][]string{"branches": {catalogBranch}}) }, true + case meshcatalogclient.RevisionVerb: + return func() (json.RawMessage, error) { + if in, _ := body.(meshcatalog.RevisionInput); in.Branch != catalogBranch { + return nil, fmt.Errorf("no watched branch %q", in.Branch) + } + return json.Marshal(catalogAnswer) + }, true + } + return nil, false +} + +// startWiring is the bundle's reader for a test: each view takes its first view as it is adopted, within ctx. +type startWiring struct { + t *testing.T + ctx context.Context +} + +func (w startWiring) Adopt(f *views.Follower, _, _ string, _ func(context.Context, []byte) error) error { + if err := f.Start(w.ctx); err != nil && w.ctx.Err() == nil { + w.t.Errorf("the first view of %s.%s: %v", f.Owner, f.State, err) + } + return nil +} + +func (startWiring) Drop(string, string) error { return nil } + +// startedBranches is the branch follower of meshcatalogclient asking ask, its views started at once. +func startedBranches(t *testing.T, ctx context.Context, ask func(string, any) (json.RawMessage, error)) *meshcatalogclient.Branches { + t.Helper() + b, err := meshcatalogclient.NewBranches(startWiring{t, ctx}, ask) + if err != nil { + t.Fatal(err) + } + return b +} + // catalogAnswer is the seat mesh-catalog's newest trunk revision: a with a fingerprint and the source revision it moved at, // b with none and the reason. var catalogAnswer = meshcatalog.RevisionAnswer{ Revision: &meshcatalog.CatalogRevision{Revision: 4, Kind: meshcatalogclient.KindTrunk, Source: 7}, Entries: []meshcatalog.CatalogEntry{ - {Module: "a", Repository: "novox/r", Commit: "c1", Path: ".", Manifest: map[string]any{"module": "a"}, Fp: fp2A, Inputs: &meshcatalog.EntryInputs{Src: 7}}, + {Module: "a", Repository: "novox/r", Commit: "c1", Path: ".", Manifest: map[string]any{"module": "a"}, ManifestDigest: "sha256:" + manifestDigestA, Fp: fp2A, Inputs: &meshcatalog.EntryInputs{Src: 7}}, {Module: "b", Repository: "novox/r", Commit: "c2", Path: "modules/b", Manifest: map[string]any{"module": "b"}, None: "builder's fingerprint not yet known"}, }, } + +// manifestDigestA is the digest the indexer read of a's manifest bytes, which the module's value carries bare (design 47 G +// §506, §578), never one computed again from the manifest whole. +const manifestDigestA = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + +// view_read_answers answers the view's own answers as every owner's read tool answers its state (design 47 G A-178): one +// key's value, and every key when none is named. +func TestTheViewReadsItsAnswersAsAnOwner(t *testing.T) { + answers := source{"hold.plex.ab.view": `{"asking":"a1","option":"release"}`} + tool := AnswersReadTool(func(name string) views.Source { + if name == viewstate.ViewAnswersState { + return answers + } + return source{} + }, time.Now) + if tool.Name != "view_read_answers" { + t.Fatalf("the tool is %s", tool.Name) + } + for args, want := range map[string]map[string]any{"release": {"key": "hold.plex.ab.view"}, "hold.plex.ab.view": {}} { + out, err := tool.Run(want) + raw, _ := json.Marshal(out) + if err != nil || !strings.Contains(string(raw), args) { + t.Errorf("%v: %s, %v", args, raw, err) + } + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go index 615b516a..da423d87 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go @@ -4,22 +4,16 @@ import ( "encoding/json" "errors" "fmt" - "mesh-delivery-view/internal/view" "reflect" "strings" "testing" "time" - "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" - "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/graph" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" + "mesh-delivery-view/internal/view" ) // encoded is v as its contract's encoder writes it, failing the test when it refuses. @@ -34,78 +28,69 @@ func encoded(t *testing.T, v []byte, err error) string { // short is a fingerprint as the assigner's words shorten it. func short(fp string) string { return fp[:len("src2:")+12] } -// tiersToolSaysEachModulesWait: view_tiers orders a source revision's modules by the module graph and names each one's -// wait (ADR 0603 §6): a consumer after its provider, held by the assigner's provision rule and its provision not provided; -// a module whose build waits for a package; a provider with no wait; a node offline named away; older modules left out. +// tiersToolSaysEachModulesWait: view_tiers reads a source revision's tiers from the assigner's difference of the kept revision +// carrying it against the one before (design 47 G §512, A-242) and names each module's waits as the parts said them in +// wait-held (the view's state waits), in the sayer's own words, beside the indexer's pending revisions. func tiersToolSaysEachModulesWait(t *testing.T) { fp := func(c string) string { return "src2:" + strings.Repeat(c, 64) } - manifest := strings.Repeat("d", 64) - mod := func(f string, src uint64) string { - b, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: f, Manifest: manifest, Src: int64(src), FpSrc: int64(src), Repository: "novox/mesh-catalog", Commit: "c"}) - return encoded(t, b, err) + since := time.Date(2026, 11, 3, 16, 40, 0, 0, time.UTC) + r := fake{}.reader() + // db and lib first, web after db by its provisioning edge; the view orders nothing. + ref := func(m string) meshwants.NodeEntryRef { + return meshwants.NodeEntryRef{Node: "ace", Module: m, Digest: "sha256:" + m} } - node := func(g graph.Node) string { - b, err := graph.Encode(g) - return encoded(t, b, err) + kept := map[int64]meshwants.AssignmentRevision{ + 3: {Revision: 3, Catalogs: []meshwants.CatalogOf{{Branch: "release", Revision: 2, Source: 5}}}, + 4: {Revision: 4, Catalogs: []meshwants.CatalogOf{{Branch: "release", Revision: 3, Source: 8}}}, + 5: {Revision: 5, Catalogs: []meshwants.CatalogOf{{Branch: "release", Revision: 4, Source: 9}}}, + 6: {Revision: 6, Catalogs: []meshwants.CatalogOf{{Branch: "release", Revision: 4, Source: 9}}}, } - build := func(b builds.BuilderBuild) string { - v, err := builds.EncodeBuilderBuild(b) - return encoded(t, v, err) + differences := map[[2]int64]meshwants.Difference{{4, 5}: {From: 4, To: 5, + Changes: []meshwants.Change{{Node: "ace", Module: "db", To: fp("a")}, {Node: "ace", Module: "lib", To: fp("c")}, + {Node: "ace", Module: "web", To: fp("b"), After: []meshwants.ChangeAfterItem{{Node: "ace", Module: "db", Kind: "provisioning"}}}}, + Tiers: [][]meshwants.NodeEntryRef{{ref("db"), ref("lib")}, {ref("web")}}}} + r.Assignment = func(n int64) (meshwants.AssignmentRevision, error) { + if n == 0 { + n = 6 + } + if rev, ok := kept[n]; ok { + return rev, nil + } + return meshwants.AssignmentRevision{}, errors.New("not kept") } - want := func(w meshwantsclient.Want) string { - v, err := meshwantsclient.EncodeWant(w) - return encoded(t, v, err) + r.Difference = func(from, to int64) (meshwants.Difference, error) { + if d, ok := differences[[2]int64{from, to}]; ok { + return d, nil + } + return meshwants.Difference{}, fmt.Errorf("no difference of %d and %d", from, to) } - have := func(h nodeengine.HaveByNodeModuleValue) string { - v, err := nodeengineclient.EncodeHave(h) - return encoded(t, v, err) - } - key := func(k string, err error) string { + // The waits the parts said, as the view's process keeps them: web's delivery wait by the assigner, lib's toolchain wait by + // the builder; db holds none. Their words are the sayers' own. + wait := func(w viewstate.Wait) string { + raw, err := json.Marshal(w) if err != nil { t.Fatal(err) } - return k + return string(raw) } - since := time.Date(2026, 11, 3, 16, 40, 0, 0, time.UTC) - r := fake{"db": mod(fp("a"), 9), "web": mod(fp("b"), 9), "lib": mod(fp("c"), 9), "old": mod(fp("e"), 5)}.reader() - r.GraphState = fake{ - "db": node(graph.Node{FP: fp("a"), Repository: "novox/mesh-catalog", Provides: []string{"database"}}), - "web": node(graph.Node{FP: fp("b"), Repository: "novox/mesh-catalog", Requires: []string{"database"}, Awaits: []string{"database"}}), - "lib": node(graph.Node{FP: fp("c"), Repository: "novox/mesh-catalog"}), - }.reader().State - r.Builds = fake{ - key(builds.BuilderBuildKey("db", fp("a"))): build(builds.BuilderBuild{State: builds.BuildDone, Manifest: manifest}), - key(builds.BuilderBuildKey("web", fp("b"))): build(builds.BuilderBuild{State: builds.BuildDone, Manifest: manifest}), - key(builds.BuilderBuildKey("lib", fp("c"))): build(builds.BuilderBuild{State: builds.BuildQueued, Waits: "@novox/x@1.2.0"}), - }.reader().State - r.WantsState = fake{ - key(meshwantsclient.WantKey("ace", "db")): want(meshwantsclient.Want{Fp: fp("a"), Self: true}), - key(meshwantsclient.WantKey("ace", "web")): want(meshwantsclient.Want{Self: true, Fp: fp("f"), Newest: fp("b"), Rule: meshwantsclient.RuleProvisions, Words: "the provision ace.web.database of web on ace is asked, not provided"}), - key(meshwantsclient.WantKey("g14", "web")): want(meshwantsclient.Want{Self: true, Fp: fp("f"), Newest: fp("b"), Rule: meshwantsclient.RuleCanary, Words: "the canary"}), - }.reader().State - r.WantsState.History = func(string) ([]view.Revision, error) { return []view.Revision{{At: since}}, nil } - r.HavesState = fake{key(nodeengineclient.HaveKey("ace", "db")): have(nodeengine.HaveByNodeModuleValue{Fp: fp("a"), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})}.reader().State - p, err := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "database", Consumer: meshprovisionerclient.End{Module: "web", Node: "ace"}, - Provider: meshprovisionerclient.End{Module: "db", Node: "ace"}, State: meshprovisionerclient.Asked, Why: "the build db runs does not offer it", At: since}) - r.ProvisionsState = fake{key(meshprovisionerclient.Key("ace", "web", "database", "")): encoded(t, p, err)}.reader().State - off, err := meshwatchdogsclient.EncodeNodePresence(meshwatchdogsclient.NodePresence{Online: false}) - r.PresenceState = fake{key(meshwatchdogsclient.PresenceNodeKey("g14")): encoded(t, off, err)}.reader().State + webWait := viewstate.Wait{Owner: "mesh-wants", Step: "delivery", Node: "ace", Module: "web", Part: "db", Machines: []string{"g14"}, Since: since, + Bound: "60m", Words: "waiting for delivery of db to g14"} + libWait := viewstate.Wait{Owner: "mesh-delivery-builder", Step: "building", Module: "lib", On: "toolchain:go@1.27", Since: since, Words: "the toolchain go 1.27 is not agreed yet"} + r.WaitsState = fake{webWait.Key(): wait(webWait), libWait.Key(): wait(libWait)}.reader().State out, err := run(t, r, view.TiersTool, map[string]any{"revision": float64(9)}) got, _ := out.(view.Tiers) - if err != nil || got.Revision != 9 || got.Modules != 3 || len(got.Tiers) != 2 || !reflect.DeepEqual(got.Away, []string{"g14"}) || got.Graph != "" { + if err != nil || got.Revision != 9 || got.Modules != 3 || len(got.Tiers) != 2 || got.Graph != "" { t.Fatalf("view_tiers 9 = %+v, %v", out, err) } - first := fmt.Sprint(got.Tiers[0]) - if len(got.Tiers[0]) != 2 || got.Tiers[0][0].Module != "db" || len(got.Tiers[0][0].Waits) != 0 || got.Tiers[0][1].Module != "lib" || - len(got.Tiers[0][1].Waits) != 1 || !strings.Contains(got.Tiers[0][1].Waits[0].Words, "waiting for the package @novox/x@1.2.0") { - t.Errorf("the first tier = %s, want db with no wait and lib's build waiting for its package", first) + if len(got.Tiers[0]) != 2 || got.Tiers[0][0].Module != "db" || got.Tiers[0][0].FP != fp("a") || len(got.Tiers[0][0].Waits) != 0 || got.Tiers[0][1].Module != "lib" || + !reflect.DeepEqual(got.Tiers[0][1].Waits, []view.TierWait{{Owner: "mesh-delivery-builder", Step: "building", On: "toolchain:go@1.27", Since: since, Words: "the toolchain go 1.27 is not agreed yet"}}) { + t.Errorf("the first tier = %+v, want db with no wait and lib's toolchain wait in its sayer's words", got.Tiers[0]) } web := got.Tiers[1][0] - if web.Module != "web" || !reflect.DeepEqual(web.After, []string{"db"}) || len(web.Waits) != 2 || - web.Waits[0].Edge != "want" || web.Waits[0].Node != "ace" || web.Waits[0].Rule != meshwantsclient.RuleProvisions || !web.Waits[0].Since.Equal(since) || - web.Waits[1].Edge != "provision database" || !strings.Contains(web.Waits[1].Words, "asked: the build db runs does not offer it") { - t.Errorf("web = %+v, want after db, held on ace by the provision rule and its provision asked (g14 is away)", web) + if web.Module != "web" || !reflect.DeepEqual(web.After, []view.TierAfter{{Module: "db", Kind: "provisioning"}}) || + !reflect.DeepEqual(web.Waits, []view.TierWait{{Owner: "mesh-wants", Step: "delivery", Node: "ace", Bound: "60m", Since: since, Words: "waiting for delivery of db to g14"}}) { + t.Errorf("web = %+v, want after db, waiting for the delivery of db in the assigner's words", web) } // ADR 0633 §5 (1585/T24): a module the indexer holds back, read through indexer_pending, waits on the edge take-in from the // wait's since; a held module the revision does not move is named in the first tier with that wait alone, a group's wait @@ -122,63 +107,54 @@ func tiersToolSaysEachModulesWait(t *testing.T) { t.Fatalf("view_tiers with waits = %+v, %v", out, err) } web = held.Tiers[1][0] - if last := web.Waits[len(web.Waits)-1]; last.Edge != view.TakeInEdge || !last.Since.Equal(heldSince) || !strings.Contains(last.Words, "src2:abc") { + if last := web.Waits[len(web.Waits)-1]; last.Step != view.TakeInStep || !last.Since.Equal(heldSince) || !strings.Contains(last.Words, "src2:abc") { t.Errorf("web = %+v, want its last wait on the edge take-in since %s", web, heldSince) } - if lib := held.Tiers[0][1]; lib.Module != "lib" || lib.Waits[len(lib.Waits)-1].Edge != view.TakeInEdge || !strings.Contains(lib.Waits[len(lib.Waits)-1].Words, "group feat") { + if lib := held.Tiers[0][1]; lib.Module != "lib" || lib.Waits[len(lib.Waits)-1].Step != view.TakeInStep || !strings.Contains(lib.Waits[len(lib.Waits)-1].Words, "group feat") { t.Errorf("lib = %+v, want its last wait the group's", lib) } - if nm := held.Tiers[0][len(held.Tiers[0])-1]; nm.Module != "newmod" || len(nm.Waits) != 1 || nm.Waits[0].Edge != view.TakeInEdge { + if nm := held.Tiers[0][len(held.Tiers[0])-1]; nm.Module != "newmod" || len(nm.Waits) != 1 || nm.Waits[0].Step != view.TakeInStep { t.Errorf("the first tier = %+v, want newmod, which the revision does not move, named with its take-in wait alone", held.Tiers[0]) } r.Pending = func() (takein.PendingAnswer, error) { return takein.PendingAnswer{}, errors.New("no answer") } if out, err = run(t, r, view.TiersTool, map[string]any{"revision": float64(9)}); err != nil || !strings.Contains(out.(view.Tiers).Pending, "no answer") || - strings.Contains(fmt.Sprint(out.(view.Tiers).Tiers), view.TakeInEdge) { + strings.Contains(fmt.Sprint(out.(view.Tiers).Tiers), view.TakeInStep) { t.Errorf("view_tiers with no answer of indexer_pending = %+v, %v", out, err) } r.Pending = nil - // No revision named: the gate's newest, its source key's revision. - r.SourceState = view.StateRead{GetEntry: func(string) (json.RawMessage, uint64, error) { return json.RawMessage(`{}`), 9, nil }} - if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || out.(view.Tiers).Revision != 9 || out.(view.Tiers).Modules != 3 { - t.Errorf("view_tiers of the gate's newest = %+v, %v", out, err) + // No revision named: per watched branch, the newest source revision the gate's verb source answers for it (design 47 G + // A-173); a branch with no source revision yet is left out. + asked := map[string]bool{} + call := func(key string, body any) (json.RawMessage, error) { + in, _ := body.(meshdelivery.SourceInput) + asked[key+" "+in.Branch] = true + rev := map[string]int64{"release": 9}[in.Branch] + return json.Marshal(meshdelivery.SourceAnswer{Branch: in.Branch, Revision: rev}) } - // ADR 0614 §2-§5: a want held on a verb, a reader capability or bus state is named by its edge; bus state is no - // provisioner key (none is ever made for it), so a module asking it shows no provision wait of it; and each verb edge - // the graph leaves out on a cycle is named. - r.GraphState = fake{ - "db": node(graph.Node{FP: fp("a"), Repository: "novox/mesh-catalog", Provides: []string{"database"}}), - "web": node(graph.Node{FP: fp("b"), Repository: "novox/mesh-catalog", Requires: []string{"database", graph.BusState}, Awaits: []string{"database", graph.BusState}, - Invokes: []string{"seat:x.v", "db.loop"}, LeftOut: []string{"db.loop"}}), - "lib": node(graph.Node{FP: fp("c"), Repository: "novox/mesh-catalog"}), - }.reader().State - for words, edge := range map[string]string{ - graph.VerbWaitWords("seat:x.v", "web", "ace"): "verb seat:x.v", - graph.ReaderWaitWords("ace", []string{"tool-runner"}, "web", short(fp("b"))): "reader capability tool-runner", - "the provision ace.web.database of web on ace is asked, not provided": "want", - } { - r.WantsState = fake{key(meshwantsclient.WantKey("ace", "web")): want(meshwantsclient.Want{Self: true, Fp: fp("f"), Newest: fp("b"), Rule: meshwantsclient.RuleProvisions, Words: words})}.reader().State - out, err := run(t, r, view.TiersTool, map[string]any{"revision": float64(9)}) - got, _ := out.(view.Tiers) - if err != nil || len(got.Tiers) != 2 { - t.Fatalf("view_tiers 9 = %+v, %v", out, err) - } - web := got.Tiers[1][0] - if len(web.Waits) != 2 || web.Waits[0].Edge != edge || web.Waits[1].Edge != "provision database" || - !reflect.DeepEqual(web.LeftOut, []string{"db.loop"}) { - t.Errorf("web held by %q = %+v, want its want named %s, no wait for a bus-state key and db.loop left out", words, web, edge) - } + r.Sources = view.SourcesOf(call, func() []string { return []string{"release", "beta"} }) + if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || len(out.(view.BranchTiers).Branches) != 1 || + out.(view.BranchTiers).Branches[0].Branch != "release" || out.(view.BranchTiers).Branches[0].Revision != 9 || out.(view.BranchTiers).Branches[0].Modules != 3 { + t.Errorf("view_tiers of each branch's newest = %+v, %v", out, err) } - // The graph not readable: every module in the first tier, said; past 500 modules the answer says it is cut short. - r.GraphState = view.StateRead{Keys: func() ([]string, error) { return nil, errors.New("no bucket") }, Get: r.GraphState.Get} - if out, err := run(t, r, view.TiersTool, map[string]any{"revision": float64(9)}); err != nil || len(out.(view.Tiers).Tiers) != 1 || - !strings.Contains(out.(view.Tiers).Graph, "cannot be read") { - t.Errorf("view_tiers with no graph = %+v, %v", out, err) + if !asked["seat:mesh-delivery.source release"] || !asked["seat:mesh-delivery.source beta"] { + t.Errorf("the gate's verb source was asked %v, want once per watched branch", asked) } - many := fake{} + // No revision kept before the one carrying the source: said, and no module named; past 500 modules the answer says it is cut + // short. + delete(kept, 4) + delete(kept, 3) + if out, err := run(t, r, view.TiersTool, map[string]any{"revision": float64(9)}); err != nil || len(out.(view.Tiers).Tiers) != 0 || + !strings.Contains(out.(view.Tiers).Graph, "no assignment revision is kept before") { + t.Errorf("view_tiers with no earlier revision = %+v, %v", out, err) + } + kept[4] = meshwants.AssignmentRevision{Revision: 4, Catalogs: []meshwants.CatalogOf{{Branch: "release", Source: 8}}} + big := meshwants.Difference{From: 4, To: 5, Tiers: [][]meshwants.NodeEntryRef{{}}} for i := 0; i < view.TiersMost+2; i++ { - many[fmt.Sprintf("m%03d", i)] = mod(fp("a"), 9) + m := fmt.Sprintf("m%03d", i) + big.Changes = append(big.Changes, meshwants.Change{Node: "ace", Module: m, To: fp("a")}) + big.Tiers[0] = append(big.Tiers[0], ref(m)) } - r.State = many.reader().State + differences[[2]int64{4, 5}] = big if out, err := run(t, r, view.TiersTool, map[string]any{"revision": "9"}); err != nil || out.(view.Tiers).Modules != view.TiersMost+2 || !strings.Contains(out.(view.Tiers).Cut, "the first 500") { t.Errorf("view_tiers past its bound = %+v, %v", out.(view.Tiers).Cut, err) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index b17e78d6..b88736d9 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -32,6 +32,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "git.novox.be/novox/mesh-sdk/go/views/viewstest" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" @@ -58,7 +59,9 @@ type catalogHolder struct { // rev is the revision's number; entries are its entries by module. rev int64 entries map[string]meshcatalog.CatalogEntry - ctx context.Context + // src is the gate's source revision of catalogBranch, as its verb source answers it. + src meshdelivery.SourceAnswer + ctx context.Context } func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) *catalogHolder { @@ -85,16 +88,62 @@ func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) _ = m.Respond(raw) } - for _, subject := range []string{"mesh.seat.mesh-catalog.call.revision.>", "mesh.seat.mesh-catalog.call.revision"} { - sub, err := nc.Subscribe(subject, answer) - if err != nil { - t.Fatal(err) + // The branch follower's other two first views (design 47 G §614): the registrations, each module on catalogBranch, and the + // gate's source, which keeps catalogBranch. + registrations := func(m *nats.Msg) { + h.mu.Lock() + a := meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{}} + for module := range h.entries { + a.Registrations = append(a.Registrations, meshcatalog.Registration{Module: module, Version: 1, Branches: []string{catalogBranch}}) + } + h.mu.Unlock() + raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) + _ = m.Respond(raw) + } + // The gate's verb source: given no branch the branches it keeps, given catalogBranch its source revision (A-173). + source := func(m *nats.Msg) { + var in meshdelivery.SourceInput + _ = json.Unmarshal(m.Data, &in) + var a any = map[string]any{"branches": []string{catalogBranch}} + if in.Branch != "" { + h.mu.Lock() + a = meshdelivery.SourceAnswer{Branch: in.Branch, Revision: h.src.Revision, Group: h.src.Group, Repositories: h.src.Repositories} + h.mu.Unlock() + } + raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"}) + _ = m.Respond(raw) + } + for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": answer, + "mesh.seat.mesh-catalog.call.registrations": registrations, "mesh.seat.mesh-delivery.call.source": source} { + for _, s := range []string{subject + ".>", subject} { + sub, err := nc.Subscribe(s, handle) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) } - t.Cleanup(func() { _ = sub.Unsubscribe() }) } return h } +// catalogBranch is the one watched branch of these tests: no branch is named in the view's code (design 47 G §532). +const catalogBranch = "release" + +// moveSource is the gate's next source revision of catalogBranch, said as source-moved (design 47 G §507, A-173). +func (h *catalogHolder) moveSource(rev int64, group string, heads map[string]string) { + h.t.Helper() + h.mu.Lock() + h.src = meshdelivery.SourceAnswer{Revision: rev, Group: group} + for repository, head := range heads { + h.src.Repositories = append(h.src.Repositories, meshdelivery.SourceHead{Repository: repository, Head: head, Trunk: catalogBranch}) + } + h.mu.Unlock() + body, _ := json.Marshal(meshdelivery.SourceMovedEvent{Branch: catalogBranch, Revision: rev, Group: group}) + if _, err := h.js.Publish(h.ctx, meshdelivery.SubjectSourceMoved(catalogBranch), body); err != nil { + h.t.Fatal(err) + } +} + // put is the holder's next trunk revision with the entry of one module changed, said as catalog-revision. func (h *catalogHolder) put(e meshcatalog.CatalogEntry) { h.t.Helper() @@ -103,8 +152,8 @@ func (h *catalogHolder) put(e meshcatalog.CatalogEntry) { h.entries[e.Module] = e rev := h.rev h.mu.Unlock() - body, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: meshcatalogclient.KindTrunk, Branch: "main", Revision: rev, Changed: []string{e.Module}}) - if _, err := h.js.Publish(h.ctx, "mesh.seat.mesh-catalog.event."+meshcatalog.EventCatalogRevision, body); err != nil { + body, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: meshcatalogclient.KindTrunk, Branch: catalogBranch, Revision: rev, Changed: []string{e.Module}}) + if _, err := h.js.Publish(h.ctx, "mesh.seat.mesh-catalog.event."+meshcatalog.EventCatalogRevision+"."+catalogBranch, body); err != nil { h.t.Fatal(err) } } @@ -396,98 +445,6 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124 t.Fatalf("the snapshot never held the indexer's keys; last: %+v", last) } -// TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes starts the program whole before the gate's asks bucket and the -// teller's answers bucket exist, then creates them: an ask appears in the snapshot, and leaves it once the teller has an -// answer under its key (1345/AC1 against a real bus; the retry is the view's own, every 5 s). -func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { // 1345/AC1 - srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true}) - if err != nil { - t.Fatal(err) - } - go srv.Start() - if !srv.ReadyForConnections(10 * time.Second) { - t.Fatal("the bus did not come up") - } - t.Cleanup(srv.Shutdown) - nc, err := nats.Connect(srv.ClientURL()) - if err != nil { - t.Fatal(err) - } - t.Cleanup(nc.Close) - js, err := jetstream.New(nc) - if err != nil { - t.Fatal(err) - } - ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) - defer cancel() - dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { os.RemoveAll(dir) }) - broker := filepath.Join(dir, "broker") - cred, _ := json.Marshal(process.Account{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"}) - if err := os.WriteFile(broker, cred, 0o600); err != nil { - t.Fatal(err) - } - socket := filepath.Join(dir, "run", "cli.sock") - env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} - runCtx, stop := context.WithCancel(ctx) - done := make(chan error, 1) - serveOwners(t, ctx, js, nc) - serveCatalog(t, ctx, js, nc) - go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() - t.Cleanup(func() { stop(); <-done }) - - gate, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: view.AsksBucket, History: view.AsksHistory}) - if err != nil { - t.Fatal(err) - } - teller, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: view.AnswersBucket, History: view.AnswersHistory}) - if err != nil { - t.Fatal(err) - } - heads := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}} - key, err := meshdeliveryclient.MergeAskKey("feat/x", heads) - if err != nil { - t.Fatal(err) - } - ask := oc.Ask{ID: meshdeliveryclient.AskID(key), Headline: "Merge into main: mesh-delivery#12", Explanation: "Merges one pull request.", Who: occ.Operator, - Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "the merge stays held", - Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}} - value, err := meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: "feat/x", Heads: askHeads(heads), Ask: rawAsk(t, ask)}) - if err != nil { - t.Fatal(err) - } - if _, err := gate.Put(ctx, key, value); err != nil { - t.Fatal(err) - } - wait := func(check func(view.Snapshot) bool, what string) { - t.Helper() - var last view.Snapshot - for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) { - if s, err := view.Ask(socket); err == nil { - last = s - if check(s) { - return - } - } - } - t.Fatalf("%s; last: %+v", what, last) - } - wait(func(s view.Snapshot) bool { - return len(s.Asks) == 1 && s.Asks[0].Key == key && s.Asks[0].How == "answer on the phone" - }, "the open ask never showed") - answer, err := meshdeliveryclient.EncodeAnswer(meshdeliveryclient.Answer{Option: "approve", Warrant: ask.ID + "@2026-10-18T12:00:00Z", Words: "the operator, via telegram, chose Approve"}) - if err != nil { - t.Fatal(err) - } - if _, err := teller.Put(ctx, key, answer); err != nil { - t.Fatal(err) - } - wait(func(s view.Snapshot) bool { return len(s.Asks) == 0 }, "the answered ask never left the list") -} - // TestProcessWholeListsEveryDifference starts the program whole before the assigner's differences bucket exists, then // creates it and writes a difference explained by a rule and one unexplained: the socket's snapshot lists both with their // rule, and a difference taken away leaves the list (1297/B8: the view lists every difference with its rule). @@ -608,8 +565,15 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 } askBucket := bucket(view.AssignerAsksBucket, view.AssignerAsksHistory) viewBucket := bucket(viewstate.ViewAnswersBucket, viewstate.ViewAnswersHistory) + bucket(viewstate.ViewModule+"_"+views.MarksState, 1) teller := bucket(view.AnswersBucket, view.AnswersHistory) - gate := bucket(view.AsksBucket, view.AsksHistory) + // The view says its answers on its views family (design 47 G A-178), which the teller follows. + said := make(chan *nats.Msg, 8) + sub, err := nc.ChanSubscribe("mesh.mod."+viewstate.ViewModule+".event."+views.EventPrefix+viewstate.ViewAnswersState, said) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { sub.Unsubscribe() }) dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i") if err != nil { @@ -668,23 +632,20 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 t.Fatalf("answer = %+v, %v", a, err) } wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Chosen == "release" }, "the ask never showed the view's choice") + select { + case msg := <-said: + if !strings.Contains(string(msg.Data), key+".view") { + t.Errorf("the view said %s, not its answer under %s.view", msg.Data, key) + } + case <-time.After(20 * time.Second): + t.Error("the view never said its answer on views.answers") + } - // The merge's Approve is answered on the phone alone. - heads := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}} - mkey, err := meshdeliveryclient.MergeAskKey("feat/x", heads) + // The merge's Approve is no ask of the view's (design 47 G §431): refused by its key, never written. + mkey, err := meshdeliveryclient.MergeAskKey("feat/x", []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "0123456789abcdef0123456789abcdef01234567"}}) if err != nil { t.Fatal(err) } - merge := oc.Ask{ID: meshdeliveryclient.AskID(mkey), Headline: "Merge", Explanation: "Merges.", Who: occ.Operator, Expires: occ.FormatTime(time.Now().Add(time.Hour)), OnExpiry: "held", - Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}} - mv, err := meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: "feat/x", Heads: askHeads(heads), Ask: rawAsk(t, merge)}) - if err != nil { - t.Fatal(err) - } - if _, err := gate.Put(ctx, mkey, mv); err != nil { - t.Fatal(err) - } - wait(func(s view.Snapshot) bool { return len(s.Asks) == 2 }, "the merge ask never showed") if err := view.AnswerAsk(socket, mkey, "approve", ""); err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") { t.Errorf("merge Approve = %v, want the refusal naming the phone", err) } @@ -700,7 +661,7 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 if _, err := teller.Put(ctx, key, answer); err != nil { t.Fatal(err) } - wait(func(s view.Snapshot) bool { return len(s.Asks) == 1 && s.Asks[0].Key == mkey }, "the answered ask never left the list") + wait(func(s view.Snapshot) bool { return len(s.Asks) == 0 }, "the answered ask never left the list") } // TestProcessWholeShowsTheBacklogOfBuildAsks starts the program whole against a bus holding the go seat's work-queue stream @@ -827,7 +788,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { t.Fatal(err) } kvs := map[string]jetstream.KeyValue{} - for _, name := range []string{view.SourceBucket, view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} { + for _, name := range []string{view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} { kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5}) if err != nil { t.Fatal(err) @@ -856,7 +817,6 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { want, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)}) failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed}) good, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) - src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}}) catalog.put(meshcatalog.CatalogEntry{Module: "web", Fp: fp, Manifest: map[string]any{"module": "a"}, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) if _, err := kvs[view.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil { t.Fatal(err) @@ -864,10 +824,9 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", failed); err != nil { t.Fatal(err) } - rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceKey("main"), src) - if err != nil { - t.Fatal(err) - } + // The gate's source of the watched branch moves: the view asks the verb source for it and reads no gate state (A-173). + const rev = 41 + catalog.moveSource(rev, "feat/x", map[string]string{"novox/web": strings.Repeat("1", 40)}) time.Sleep(time.Second) if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event."+DeliveredEventType)); err != nil || len(info.State.Subjects) != 0 { t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State) @@ -883,8 +842,8 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { t.Fatal("the delivery of the source revision was never said") } var e view.DeliveredEvent - if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) { - t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev) + if err := json.Unmarshal(msg.Data, &e); err != nil || e.Branch != catalogBranch || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) { + t.Errorf("the event = %+v, %v; want the branch %s, revision %d, group feat/x and the heads", e, err, catalogBranch, rev) } if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" { t.Errorf("the headers = %v", msg.Header) @@ -1039,12 +998,3 @@ func TestProcessWholeSaysModulesInMotion(t *testing.T) { t.Errorf("events said by type = %v, want moving 2, settled 1, stuck 1", count) } } - -// askHeads is the heads of a merge ask as the seat carries them. -func askHeads(heads []meshdeliveryclient.Head) []meshdelivery.AsksByMergeGroupHeadsValueHeadsItem { - out := make([]meshdelivery.AsksByMergeGroupHeadsValueHeadsItem, len(heads)) - for i, h := range heads { - out[i] = meshdelivery.AsksByMergeGroupHeadsValueHeadsItem{Repository: h.Repository, Number: int64(h.Number), Head: h.Head} - } - return out -} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index bee31ef3..e1b98ecb 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -14,15 +14,19 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "io" "log" + "maps" "mesh-delivery-view/internal/view" "net/http" "os" "os/signal" + "slices" "strings" + "sync" "syscall" "time" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" "git.novox.be/novox/mesh-sdk/go/process" @@ -123,36 +127,62 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - reader := &views.ProcessReader{Module: viewstate.ViewModule, Call: calls.Call, Logf: logger.Printf} + // The module's one pull of its durable consumer: the reader's views events and the waits' events are handlers on it, and it is + // started once, after both (design 47 G §512, A-242). + hub := process.NewEventHub(logger.Printf) + defer hub.Stop() + reader := &views.ProcessReader{Module: viewstate.ViewModule, Call: calls.Call, Logf: logger.Printf, Hub: hub} writer, err := findingWriter(ctx, cred, conn, js, calls.State("findings"), logger.Printf) if err != nil { return fmt.Errorf("the view's finding writer: %w", err) } defer writer.Close() reader.Findings = func() views.WireWriter { return writer } + sayErrs := make(chan error, 1) + sayAnswers(ctx, js, conn, cred, writer, sayErrs, logger.Printf) for _, o := range ownersFollowed { if _, err := reader.Follow(o.owner, o.state, o.tool); err != nil { return err } } open := buckets{js: js, reader: reader} - for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AsksBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.SourceBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket} { + for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket, viewstate.ReachBucket} { go follow(ctx, open, bucket, false, model, logger) } - // The source's history gives each head's source revision (ADR 0556); the modules' history is each revision of the - // catalog half the view takes, in its order (below). - go follow(ctx, open, view.SourceBucket, true, model, logger) + // The view's own state reach, as it stood at the start, is what the derivation lays its history over (design 47 G A-225); + // the modules' history is each revision of the catalog half the view takes, in its order (below), and each watched + // branch's source is the seat mesh-delivery's verb source per branch, followed by its source-moved (A-173). statusBucket := process.AwaitBucket(ctx, js, conn, firstViewsBucket, "view_first_views reads nothing and view_first_view_again is not heard", logger.Printf) - // The modules are the catalog half's, adopted whole by the seat's client: its first view, its status, its hand asks and - // its event catalog-revision (on the module's durable consumer) are the reader's. - catalog, _, err := meshcatalogclient.WireCatalogProcess(reader, calls.Call, "main", catalogChanged(model, logger)) + // The modules are the catalog half's, per watched branch, through meshcatalogclient's branch follower adopted on the + // reader (design 47 G §532, §614): the registrations' branches, before the first registrations the gate's source's; + // each branch's first view, its status, its hand asks and its event catalog-revision are the reader's. A module's entry + // is the one Branches.Entry reads, the first branch of its own registration (A-103). + branches, err := meshcatalogclient.FollowBranchesProcess(reader, calls.Call) if err != nil { return err } + branches.Logf = logger.Printf + sources := &branchSources{ctx: ctx, reader: reader, call: calls.Call, model: model, logger: logger} + modules := &catalogModules{ctx: ctx, branches: branches, sources: sources, model: model, logger: logger, applied: map[string]bool{}} + branches.Changed = modules.changed reader.Start(ctx, js, cred, statusBucket) - catalogHeld(ctx, catalog, model) + if err := hearWaits(ctx, hub, js, logger.Printf); err != nil { + return fmt.Errorf("the waits' events are not handled: %w", err) + } + if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { + logger.Printf("the consumer is not heard yet (tried again until it is): %v", err) + } + go func() { + select { + case <-branches.Registrations.Follower.Ready(): + modules.settle() + case <-ctx.Done(): + } + }() + var said error select { case <-ctx.Done(): + case said = <-sayErrs: case err := <-webUp: if !errors.Is(err, http.ErrServerClosed) { return fmt.Errorf("the web view stopped: %w", err) @@ -161,7 +191,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro shutdown, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() srv.Shutdown(shutdown) - return nil + return said +} + +// sayAnswers says each change of the view's own state answers on its views family (`views.answers`), from the revision its +// marks state view-marks says it said up to, as every owner says its states (design 47 G A-175, A-178): the teller follows it +// by view_read_answers and that family, never by reading the state. A watch that ends is sent on errs, and the program ends +// with it, loudly. +func sayAnswers(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, cred process.Account, writer views.WireWriter, errs chan<- error, logf func(string, ...any)) { + marks := viewstate.ViewModule + "_" + views.MarksState + process.AwaitBucket(ctx, js, nc, marks, "the view's answers are not said on its views family, so the teller waits on its first view", logf).Ready(func(mk jetstream.KeyValue) { + em := views.Emitter{Owner: viewstate.ViewModule, Subject: cred.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, Findings: writer, Logf: logf} + process.AwaitBucket(ctx, js, nc, viewstate.ViewAnswersBucket, "the view's answers are not said on its views family", logf).Ready(func(kv jetstream.KeyValue) { + if err := em.Start(ctx, viewstate.ViewAnswersState, kv, errs); err != nil { + logf("%v: the teller asks view_read_answers at each change until the view starts again", err) + } + }) + }) } // kvWriter is the view's one write: a key of its own bucket answers (manifest state entry answers), which the teller reads. @@ -202,7 +248,7 @@ type ReachedEvent struct { // The view's own state of its first views (views.StatusState). var firstViewsBucket = viewstate.ViewModule + "_" + views.StatusState -// ownersFollowed are the indexer's and the assigner's states the view's process follows by their first views, each by its +// ownersFollowed are the indexer's, the assigner's and the teller's states the view's process follows by their first views, each by its // owner's read tool (as the view's invokes names it) and views family (its events on the shared events stream); the // view's tools ask the other six through the owners' read tools on each call. var ownersFollowed = []struct{ owner, state, tool string }{ @@ -211,6 +257,7 @@ var ownersFollowed = []struct{ owner, state, tool string }{ {view.AssignerModule, "asks", view.AssignerModule + ".assigner_read_asks"}, {view.IndexerModule, "candidates", view.IndexerModule + ".indexer_read_candidates"}, {view.IndexerModule, "taken", view.IndexerModule + ".indexer_read_taken"}, + {view.TellerModule, view.AnswersState, view.TellerModule + ".teller_read_answers"}, } // buckets opens what the view watches: an owner's state as its picture (firstview), any other bucket on the bus. @@ -298,34 +345,164 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account return err } -// catalogChanged hands one module of a revision the view took to the model, in the form of its value; the modules' -// history is these changes in the order of the revisions. -func catalogChanged(model *view.Model, logger *log.Logger) func(string, json.RawMessage, bool, time.Time) { - return func(module string, entry json.RawMessage, gone bool, at time.Time) { - e := view.Entry{Bucket: view.ModulesBucket, Key: module, At: at, Deleted: gone} - if !gone { - v, err := view.ModuleValue(entry) - if err != nil { - logger.Printf("the catalog entry of %s is not taken: %v", module, err) - return - } - e.Value = v - } - model.Apply(e) - e.History = true - model.Apply(e) - } +// catalogModules hands the model the catalog half's modules, one value per module: the entry Branches.Entry reads. Until +// every watched branch has its first view (Branches.Ready) it hands nothing; once they all have, it hands every module +// whole and tells the model the modules are read (Initial), then each module a branch's change names. A branch adopted +// later without its first view stops the handing until it has one, when every module is handed whole again. +type catalogModules struct { + ctx context.Context + branches *meshcatalogclient.Branches + sources *branchSources // the source of each branch followed (A-173) + model *view.Model + logger *log.Logger + + mu sync.Mutex + held bool // every module was handed whole since the last time a branch had no first view + waiting bool // a settle waits on the branches' first views + applied map[string]bool // the modules the model holds } -// catalogHeld makes the model's pictures of the modules initial once the catalog half's first view is held: the process -// starts at once and waits for the seat's holder, saying so. -func catalogHeld(ctx context.Context, cat *meshcatalogclient.Catalog, model *view.Model) { +// changed is the branch follower's Changed: a module of one branch changed. +func (c *catalogModules) changed(_, module string, _ json.RawMessage, _ bool, at time.Time) { + c.follow() + c.mu.Lock() + held := c.held + c.mu.Unlock() + if !held { + c.settle() + return + } + e, ok, err := c.branches.Entry(module) + if err != nil { + c.mu.Lock() + c.held = false + c.mu.Unlock() + c.settle() + return + } + c.apply(module, e, ok, at) +} + +// settle waits, on its own goroutine, for the first view of every followed branch, then hands every module whole. It +// waits on the views' Ready, never on a timer; with no branch followed it ends, and the next change starts it again. +func (c *catalogModules) settle() { + c.mu.Lock() + if c.waiting || c.held { + c.mu.Unlock() + return + } + c.waiting = true + c.mu.Unlock() go func() { - select { - case <-cat.Follower.Ready(): - model.Initial(view.ModulesBucket) - model.Initial(view.ReachHistory(view.ModulesBucket)) - case <-ctx.Done(): + defer func() { + c.mu.Lock() + c.waiting = false + c.mu.Unlock() + }() + for c.ctx.Err() == nil { + followed := c.branches.Followed() + if len(followed) == 0 { + return + } + for _, branch := range followed { + if cat, ok := c.branches.Catalog(branch); ok { + select { + case <-cat.Ready(): + case <-c.ctx.Done(): + return + } + } + } + if c.branches.Ready() == nil { + c.whole() + return + } } }() } + +// whole hands every module the followed branches list, removes from the model each it holds that none lists any more, +// and tells the model the modules are read. +func (c *catalogModules) whole() { + c.follow() + at := time.Now() + listed := map[string]bool{} + for _, branch := range c.branches.Followed() { + if cat, ok := c.branches.Catalog(branch); ok { + half, _ := cat.View() + for module := range half.Entries { + listed[module] = true + } + } + } + c.mu.Lock() + for module := range c.applied { + if !listed[module] { + listed[module] = false + } + } + c.mu.Unlock() + for _, module := range slices.Sorted(maps.Keys(listed)) { + e, ok, err := c.branches.Entry(module) + if err != nil { + c.logger.Printf("the catalog half is handed again at its next change: %v", err) + return + } + c.apply(module, e, ok, at) + } + c.mu.Lock() + c.held = true + c.mu.Unlock() + c.model.Initial(view.ModulesBucket) + c.model.Initial(view.ReachHistory(view.ModulesBucket)) +} + +// follow has the source of every followed branch followed, and no other (A-173). +func (c *catalogModules) follow() { + if c.sources != nil { + c.sources.ensure(c.branches.Followed()) + } +} + +// branchOf is the watched branch a module's entry is read from: the first branch of its registration, before the first +// registrations the first followed branch that lists it (design 47 G A-103, A-173); "" when none. +func (c *catalogModules) branchOf(module string) string { + if reg, ok := c.branches.Registrations.Registration(module); ok && len(reg.Branches) > 0 { + return reg.Branches[0] + } + for _, branch := range c.branches.Followed() { + if cat, ok := c.branches.Catalog(branch); ok { + if _, ok := cat.Entry(module); ok { + return branch + } + } + } + return "" +} + +// apply hands the model one module's entry, or its removal when no branch lists it. +func (c *catalogModules) apply(module string, entry meshcatalog.CatalogEntry, listed bool, at time.Time) { + e := view.Entry{Bucket: view.ModulesBucket, Key: module, At: at, Deleted: !listed} + if listed { + e.Branch = c.branchOf(module) + } + if listed { + raw, _ := json.Marshal(entry) + v, err := view.ModuleValue(raw) + if err != nil { + c.logger.Printf("the catalog entry of %s is not taken: %v", module, err) + return + } + e.Value = v + } + c.mu.Lock() + if listed { + c.applied[module] = true + } else { + delete(c.applied, module) + } + c.mu.Unlock() + c.model.Apply(e) + e.History = true + c.model.Apply(e) +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go new file mode 100644 index 00000000..0721deaa --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go @@ -0,0 +1,153 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log" + "mesh-delivery-view/internal/view" + "sync" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/sourcebranches" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// sourceState is the state of the view of one watched branch's source in the view's reader: `mesh-delivery.source.`. +const sourceState = "source" + +// adopter is how branchSources puts a branch's view on the view's one reader and takes it off (views.ProcessReader). +type adopter interface { + Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) (*views.Picture, error) + Drop(owner, state string) error +} + +// branchSources follows the source of each watched branch on its own (design 47 G §507, §512, A-173): per branch the +// first view of the seat mesh-delivery's verb source given that branch, then each source-moved of that branch, asked +// again until the revision heard is held; each answer goes to the model as the branch's SourceBucket entry. The branches +// are the ones the branch follower follows (ensure); none is named here. +type branchSources struct { + ctx context.Context + reader adopter + call func(key string, body any) (json.RawMessage, error) + model *view.Model + logger *log.Logger + + mu sync.Mutex + branches map[string]*views.Follower // by escaped branch +} + +// ensure follows the source of every branch named and drops the view of each followed branch no longer named. +func (s *branchSources) ensure(followed []string) { + s.mu.Lock() + defer s.mu.Unlock() + if s.branches == nil { + s.branches = map[string]*views.Follower{} + } + named := map[string]bool{} + for _, branch := range followed { + esc := keys.EscapeName(branch) + named[esc] = true + if s.branches[esc] != nil { + continue + } + f := s.follower(branch) + if _, err := s.reader.Adopt(f, meshdelivery.Seat, meshdelivery.EventSourceMoved, s.hear(branch, f)); err != nil { + s.logger.Printf("the source of %s is not followed, and is adopted again at the next change of the catalog half: %v", branch, err) + continue + } + s.branches[esc] = f + } + for esc := range s.branches { + if named[esc] { + continue + } + if err := s.reader.Drop(meshdelivery.Seat, views.BranchState(sourceState, mustUnescape(esc))); err != nil { + s.logger.Printf("the source of %s is not dropped: %v", esc, err) + continue + } + delete(s.branches, esc) + s.model.Apply(view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: true, At: time.Now()}) + } +} + +func mustUnescape(esc string) string { + name, err := keys.UnescapeName(esc) + if err != nil { + return esc + } + return name +} + +// follower is the view of one branch's source: its one value, keyed by the escaped branch, the verb's answer. +func (s *branchSources) follower(branch string) *views.Follower { + esc := keys.EscapeName(branch) + f := &views.Follower{Owner: meshdelivery.Seat, Tool: view.SourceVerb, State: views.BranchState(sourceState, branch)} + f.Ask = func(context.Context, string) (views.Answer, error) { + a, err := view.SourceOf(s.call, branch) + if err != nil { + return views.Answer{}, err + } + raw, err := json.Marshal(a) + if err != nil { + return views.Answer{}, err + } + return views.Answer{V: 1, Owner: meshdelivery.Seat, State: f.State, Revision: uint64(a.Revision), Created: sourcebranches.Created, + Values: map[string]json.RawMessage{esc: raw}, At: time.Now()}, nil + } + f.Changed = func(_ string, value json.RawMessage, deleted bool, at time.Time) { + e := view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: deleted, At: at} + if !deleted { + var a meshdelivery.SourceAnswer + if err := json.Unmarshal(value, &a); err != nil { + s.logger.Printf("the source of %s is not taken: %v", branch, err) + return + } + se, err := view.SourceEntry(branch, a) + if err != nil { + s.logger.Printf("the source of %s is not taken: %v", branch, err) + return + } + se.At = at + e = se + } + s.model.Apply(e) + } + return f +} + +// revisionOf is the source revision f holds of the branch, 0 when none. +func revisionOf(f *views.Follower, esc string) int64 { + raw, ok := f.Get(esc) + if !ok { + return 0 + } + var a meshdelivery.SourceAnswer + if json.Unmarshal(raw, &a) != nil { + return 0 + } + return a.Revision +} + +// hear is the handler of source-moved for one branch's view: another branch's event is not its own; a revision it does not +// hold yet asks the first view again until it does, on its own goroutine, so the consumer is never held. +func (s *branchSources) hear(branch string, f *views.Follower) func(context.Context, []byte) error { + esc := keys.EscapeName(branch) + return func(ctx context.Context, data []byte) error { + var ev meshdelivery.SourceMovedEvent + if err := json.Unmarshal(data, &ev); err != nil || ev.Branch == "" { + return fmt.Errorf("not a %s event: %s", meshdelivery.EventSourceMoved, sourcebranches.ClipBytes(data)) + } + if ev.Branch != branch && ev.Branch != esc { + return nil + } + held := func() bool { return revisionOf(f, esc) >= ev.Revision } + if held() { + return nil + } + go sourcebranches.CatchUp(s.ctx, f, held, fmt.Sprintf("source of %s moved to %d", branch, ev.Revision)) + return nil + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits.go new file mode 100644 index 00000000..0b799309 --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits.go @@ -0,0 +1,94 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" +) + +// The events of a wait, said by every part that holds one (the seats mesh-wants, node-engine, mesh-pipeline-runner and +// mesh-grants on their subjects, the builder and the others on their module's): the view keeps them in its state waits +// (design 47 G §512, §376, A-242). +const ( + eventWaitHeld = "wait-held" + eventWaitEnded = "wait-ended" +) + +// waitPatterns are the subjects the view's hub handles for the waits: a seat's and a module's, each with and without the +// node or key the sayer adds after the event's name. +func waitPatterns() []string { + var out []string + for _, prefix := range []string{"mesh.seat.*.event.", "mesh.mod.*.event."} { + for _, event := range []string{eventWaitHeld, eventWaitEnded} { + out = append(out, prefix+event, prefix+event+".>") + } + } + return out +} + +// applyWait keeps one wait event: wait-held puts the wait under its key, wait-ended deletes it. The owner is the seat or module +// of the subject unless the body names one. +func applyWait(subject string, body []byte, put func(key string, value []byte) error, del func(key string) error) error { + parts := strings.Split(subject, ".") + if len(parts) < 5 || parts[0] != "mesh" || (parts[1] != "seat" && parts[1] != "mod") || parts[3] != "event" { + return fmt.Errorf("%s is no wait's subject", subject) + } + w, err := viewstate.ReadWait(parts[2], body) + if err != nil { + return fmt.Errorf("%s: %w", subject, err) + } + switch parts[4] { + case eventWaitHeld: + raw, err := json.Marshal(w) + if err != nil { + return err + } + return put(w.Key(), raw) + case eventWaitEnded: + return del(w.Key()) + } + return fmt.Errorf("%s is no wait's subject", subject) +} + +// hearWaits registers the waits' handlers on the process's hub; each keeps its event in the state waits and acknowledges it, and +// one that could not be kept is not acknowledged, so the bus gives it again. +func hearWaits(ctx context.Context, hub *process.EventHub, js jetstream.JetStream, logf func(string, ...any)) error { + kv := func() (jetstream.KeyValue, error) { return js.KeyValue(ctx, viewstate.WaitsBucket) } + put := func(key string, value []byte) error { + b, err := kv() + if err != nil { + return err + } + _, err = b.Put(ctx, key, value) + return err + } + del := func(key string) error { + b, err := kv() + if err != nil { + return err + } + if err = b.Delete(ctx, key); errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + return err + } + for _, pattern := range waitPatterns() { + if err := hub.Handle(pattern, func(m jetstream.Msg) { + if err := applyWait(m.Subject(), m.Data(), put, del); err != nil { + logf("the wait event %s is not kept (the bus gives it again): %v", m.Subject(), err) + _ = m.Nak() + return + } + _ = m.Ack() + }); err != nil { + return err + } + } + return nil +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits_test.go new file mode 100644 index 00000000..50e05b0a --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/waits_test.go @@ -0,0 +1,36 @@ +package main + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go/jetstream" +) + +func TestWaitEventsAreKeptUnderTheirKeyAndEnded(t *testing.T) { + kept := map[string]string{} + put := func(k string, v []byte) error { kept[k] = string(v); return nil } + del := func(k string) error { delete(kept, k); return nil } + held := `{"step":"delivery","module":"web","node":"ace","part":"db","machines":["g14"],"since":"2026-11-12T01:00:00Z","words":"waiting for delivery of db to g14"}` + if err := applyWait("mesh.seat.mesh-wants.event.wait-held", []byte(held), put, del); err != nil || len(kept) != 1 || + !strings.Contains(kept["mesh-wants.delivery.ace.web"], "waiting for delivery of db to g14") { + t.Fatalf("kept %v, %v", kept, err) + } + if err := applyWait("mesh.mod.mesh-delivery-builder.event.wait-held.web.fp", []byte(`{"step":"building","module":"web","on":"toolchain:go@1","since":"2026-11-12T01:00:00Z"}`), put, del); err != nil || len(kept) != 2 { + t.Fatalf("a module's wait: %v, %v", kept, err) + } + if err := applyWait("mesh.seat.mesh-wants.event.wait-ended", []byte(`{"step":"delivery","module":"web","node":"ace","how":"ran","since":"2026-11-12T01:00:00Z"}`), put, del); err != nil || len(kept) != 1 { + t.Fatalf("ended: %v, %v", kept, err) + } + if err := applyWait("mesh.seat.mesh-wants.event.wait-held", []byte(`{"module":"web"}`), put, del); err == nil { + t.Error("a wait with no step was kept") + } + // None of the patterns overlaps another: the hub gives a message one taker. + hub := process.NewEventHub(func(string, ...any) {}) + for _, p := range waitPatterns() { + if err := hub.Handle(p, func(jetstream.Msg) {}); err != nil { + t.Fatalf("the pattern %s overlaps another: %v", p, err) + } + } +} diff --git a/modules/mesh-delivery-view/internal/view/answers_test.go b/modules/mesh-delivery-view/internal/view/answers_test.go index 67b2eb85..9ef5ef79 100644 --- a/modules/mesh-delivery-view/internal/view/answers_test.go +++ b/modules/mesh-delivery-view/internal/view/answers_test.go @@ -97,12 +97,15 @@ func TestTheViewWritesItsAnswerUnderTheAsksViewKey(t *testing.T) { // 1297/B9a A func TestTheViewRefusesWhatItMayNotAnswer(t *testing.T) { // 1297/B9a AC e, a := holdAsk(t) - merge := askEntry(t, "feat/x", "Merge into main: mesh-delivery#12", "Merges.") + mergeKey, err := meshdeliveryclient.MergeAskKey("feat/x", []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}) + if err != nil { + t.Fatal(err) + } cases := []struct { name, key, option, want string prep []Entry }{ - {"the merge's Approve", merge.Key, "approve", "answer it on the phone (issue 953)", []Entry{merge}}, + {"the merge's Approve", mergeKey, "approve", "answer it on the phone (issue 953)", nil}, {"an option the ask does not offer", e.Key, "delete", `offers no option "delete"`, []Entry{e}}, {"an ask the view does not hold", e.Key, "release", "no open ask", nil}, {"an ask that is no ask key", "hold.x", "release", "not an ask key", nil}, @@ -145,9 +148,11 @@ func TestTheSocketAnswersAnAsk(t *testing.T) { // 1297/B9b if _, ok := w.puts[e.Key+".view"]; !ok { t.Fatalf("puts = %v", w.puts) } - merge := askEntry(t, "feat/x", "Merge", "Merges.") - fill(m, merge) - err := AnswerAsk(path, merge.Key, "approve", "") + mergeKey, err := meshdeliveryclient.MergeAskKey("feat/x", []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}) + if err != nil { + t.Fatal(err) + } + err = AnswerAsk(path, mergeKey, "approve", "") if err == nil || !strings.Contains(err.Error(), "answer it on the phone (issue 953)") { t.Errorf("merge over the socket: %v", err) } @@ -204,9 +209,11 @@ func TestThePageAnswersAnAskAndRefusesAForeignPost(t *testing.T) { // 1297/B9a t.Fatalf("answer = %d %q, puts %v", rec.Code, rec.Body.String(), w.puts) } // The merge's Approve is refused with its reason, and the rest of the page stays read-only. - merge := askEntry(t, "feat/x", "Merge", "Merges.") - fill(m, merge) - rec = postAnswer(h, url.Values{"key": {merge.Key}, "option": {"approve"}}, nil) + mergeKey, err := meshdeliveryclient.MergeAskKey("feat/x", []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}}) + if err != nil { + t.Fatal(err) + } + rec = postAnswer(h, url.Values{"key": {mergeKey}, "option": {"approve"}}, nil) if rec.Code != http.StatusUnprocessableEntity || !strings.Contains(rec.Body.String(), "answer it on the phone (issue 953)") || len(w.puts) != 1 { t.Errorf("merge = %d %q", rec.Code, rec.Body.String()) } diff --git a/modules/mesh-delivery-view/internal/view/asks_test.go b/modules/mesh-delivery-view/internal/view/asks_test.go index 68f6fb43..54ca5a90 100644 --- a/modules/mesh-delivery-view/internal/view/asks_test.go +++ b/modules/mesh-delivery-view/internal/view/asks_test.go @@ -7,35 +7,12 @@ import ( "net/http/httptest" "strings" "testing" - "time" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" ) -const askHeadA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" - -// askEntry is a key of the gate's asks as the gate writes it (mesh-sdk go/delivery EncodeAsk, MergeAskKey): the router's ask -// whole, under the id its key gives. -func askEntry(t *testing.T, group, headline, words string) Entry { - t.Helper() - heads := []meshdeliveryclient.Head{{Repository: "novox/mesh-delivery", Number: 12, Head: askHeadA}} - key, err := meshdeliveryclient.MergeAskKey(group, heads) - if err != nil { - t.Fatal(err) - } - a := oc.Ask{ID: meshdeliveryclient.AskID(key), Headline: headline, Explanation: words, Who: occ.Operator, Expires: occ.FormatTime(t0.Add(23 * time.Hour)), OnExpiry: "the merge stays held", - Options: []oc.Option{{ID: "approve", Label: "Approve", Does: "merges", Level: occ.Approve, Binds: "sha256:00"}}} - v, err := meshdeliveryclient.EncodeAsk(meshdelivery.AsksByMergeGroupHeadsValue{Group: group, Heads: askHeads(heads), Ask: rawAsk(t, a)}) - if err != nil { - t.Fatal(err) - } - return Entry{Bucket: AsksBucket, Key: key, Value: v, At: t0} -} - // answerEntry is the teller's answer to the first asking of the key; answerToAsking to the asking with this router id. The // answer's warrant is `@