diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index ef756cfb..0e8a5a9f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -1414,3 +1414,74 @@ func serveAssignerAssignments(t *testing.T, ctx context.Context, js jetstream.Je } t.Cleanup(func() { _ = sub.Unsubscribe() }) } + +// TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision: the seat's verb source lists a second watched branch, +// beta, with the merge of mesh-catalog it names as its head (design 47 G §13, §14, §291, A-93); the indexer started with the +// branch listed reads source.beta, takes the head's merged message in as a pure computation, and keeps branch.beta.1, which +// the branch follower (index.BranchView, as the branch's builds read it) reads; the trunk's modules are left. +func TestIndexerTakesInAWatchedBranchOtherThanTheFirstIntoItsOwnRevision(t *testing.T) { + manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json") + if err != nil { + t.Fatal(err) + } + const merge = "89abcdef0123456789abcdef0123456789abcdef" + data, err := forgemsg.EncodeForge(forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "beta", Number: 301, + Head: "0123456789abcdef0123456789abcdef01234567", Merge: merge, Root: "r0", + Trees: map[string]string{"modules/avahi": "betatree"}, Manifests: map[string]string{"modules/avahi": string(manifest)}, + Locks: map[string]map[string]string{"modules/avahi": {"go.sum": "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"}}}) + if err != nil { + t.Fatal(err) + } + js, ctx, _, modules := startIndexer(t, func(ctx context.Context, builds buildLog) { + builds.end(t, "novox/mesh-controller", "build-agent", "modules/build-agent") + }, func(ctx context.Context, js jetstream.JetStream) { + seedBuildModules(t, ctx, js, nil) + source := func(m *nats.Msg) { + var ask struct{ Branch string } + _ = json.Unmarshal(m.Data, &ask) + answer := meshdelivery.SourceAnswer{Branch: ask.Branch} + switch ask.Branch { + case "": + answer.Branches = []string{"main", "beta"} + case "beta": + answer.Group, answer.Revision = "g", 9 + answer.Repositories = []meshdelivery.SourceHead{{Repository: "novox/mesh-catalog", Head: merge, Trunk: "beta"}} + } + raw, _ := json.Marshal(map[string]any{"result": answer, "node": "n1"}) + _ = m.Respond(raw) + } + bridge := func(m *nats.Msg) { + raw, _ := json.Marshal(map[string]any{"result": map[string]any{"message": json.RawMessage(data)}, "node": "n1"}) + _ = m.Respond(raw) + } + for subject, h := range map[string]nats.MsgHandler{ + "mesh.seat.mesh-delivery.call.source.>": source, "mesh.seat.mesh-delivery.tool.source": source, + "mesh.mod.mesh-delivery-forge-bridge.call.bridge_merged_message.>": bridge, "mesh.mod.mesh-delivery-forge-bridge.tool.bridge_merged_message": bridge, + } { + if _, err := js.Conn().Subscribe(subject, h); err != nil { + t.Fatal(err) + } + } + }) + objects := newBusObjects(js, testAccount, index.CatalogRevisionsObjects) + var view index.Store + var at int64 + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && at == 0; time.Sleep(50 * time.Millisecond) { + if v, _, n, err := index.BranchView(ctx, objects, "beta"); err == nil { + view, at = v, n + } + } + if at != 1 { + t.Fatalf("the branch follower reads beta at revision %d within 20 s of the start, want 1", at) + } + value, _, found, err := view.Get(ctx, "avahi") + if err != nil || !found { + t.Fatalf("branch.beta.1 holds no avahi (%v)", err) + } + if m, err := meshcatalogclient.DecodeModule(value); err != nil || m.Commit != merge || m.Inputs2 == nil { + t.Errorf("avahi at the branch's head = %+v (%v), want commit %s with its inputs", m, err, merge) + } + if e, err := modules.Get(ctx, "avahi"); err == nil { + t.Errorf("modules.avahi = %s, want the trunk's modules left", e.Value()) + } +} diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 2bddf629..eb3047e3 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -511,7 +511,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The source revision of a watched branch (design 47 G §15, §115, §362): asked of the seat mesh-delivery's verb source as a // kept call, once at the start, before the handler of its event source-moved reads anything; a source-moved heard // meanwhile is read after. - source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, logf: logger.Printf} + source := &sourceFollow{read: sourceVia(calls.CallKept), apply: ix.ApplySource, applyBranch: ix.ApplyBranchSource, logf: logger.Printf} go source.catchUp(ctx) // Only the toolchain reports of nodes the catalogue places the build module on count (issue 1940, ADR 0549 §3): the // assignments is settled first, so the toolchains settle through the gate already knowing it. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go index dd54b4e6..fa510722 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main_test.go @@ -191,6 +191,9 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt( s = &sourceFollow{ read: func(branch string) (meshdelivery.SourceAnswer, error) { order = append(order, "read "+branch) + if branch == "" { // given no branch the seat lists the branches it keeps a source of (A-93) + return meshdelivery.SourceAnswer{Branches: []string{"main", "beta", "feat=2e1"}}, nil + } if revision == 0 { return meshdelivery.SourceAnswer{}, fmt.Errorf("source.main is not readable") } @@ -205,15 +208,21 @@ func TestTheSourceIsReadFromTheSeatAndASourceMovedDuringTheCatchUpIsReadAfterIt( order = append(order, fmt.Sprintf("apply %s %d", a.Branch, a.Revision)) return nil }, + applyBranch: func(_ context.Context, branch string, a meshdelivery.SourceAnswer) error { + order = append(order, fmt.Sprintf("branch %s %d", branch, a.Revision)) + return nil + }, logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }, } s.catchUp(context.Background()) - want := []string{"read main", "apply main 4", "read main", "apply main 4"} // once, after the catch-up; beta is left + // The trunk, then each other watched branch the seat lists (a name unescaped), then what was heard meanwhile once each, + // beta among them: main through the take-in, every other branch into its own revision (A-93). + want := []string{"read main", "apply main 4", "read ", "read beta", "branch beta 4", "read feat=2e1", "branch feat.1 4", "read main", "apply main 4", "read beta", "branch beta 4"} if !reflect.DeepEqual(order, want) { t.Errorf("order = %v, want %v", order, want) } s.moved(context.Background(), "main") // after the catch-up: read at once - if len(order) != 6 { + if len(order) != 13 { t.Errorf("a source-moved after the catch-up was not read at once: %v", order) } order, revision = nil, 0 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go index cc1f6db7..a08fda8c 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named.go @@ -237,7 +237,7 @@ func readVia(call toolCaller) func(ctx context.Context, repository, sha, dir str } // trunkBranch is the branch pull requests are open into. -const trunkBranch = "main" +const trunkBranch = index.TrunkBranch // openVia asks the forge which pull requests into the trunk are open now (the seat git's open-pulls), every repository's. // An answer that lacks the list is refused, never read as none. diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go index 10f92615..16b62e1f 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/source.go @@ -8,6 +8,7 @@ import ( "sync" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/keys" "github.com/nats-io/nats.go/jetstream" ) @@ -40,17 +41,31 @@ func sourceVia(call toolCaller) func(branch string) (meshdelivery.SourceAnswer, type sourceFollow struct { read func(branch string) (meshdelivery.SourceAnswer, error) apply func(context.Context, meshdelivery.SourceAnswer) error - logf func(string, ...any) + // applyBranch takes a watched branch other than the trunk's in (design 47 G §291, A-93); branch is its name, unescaped. + applyBranch func(ctx context.Context, branch string, a meshdelivery.SourceAnswer) error + logf func(string, ...any) mu sync.Mutex caught bool waiting map[string]bool } -// catchUp reads the trunk branch's source revision and applies it, then the branches a source-moved named meanwhile. A -// read that fails is said; the catch-up ends all the same, and the next source-moved reads again. +// catchUp reads the trunk branch's source revision and applies it, then every other branch the seat keeps a source of (the +// answer given no branch lists them, A-93), then the branches a source-moved named meanwhile. A read that fails is said; +// the catch-up ends all the same, and the next source-moved reads again. func (s *sourceFollow) catchUp(ctx context.Context) { s.take(ctx, trunkBranch) + var others []string + if list, err := s.read(""); err != nil { + s.logf("the branches the source names are not read, and each is read at its next source-moved: %v", err) + } else { + others = list.Branches + } + for _, branch := range others { + if branch != trunkBranch { + s.take(ctx, branch) + } + } s.mu.Lock() s.caught = true branches := s.waiting @@ -76,17 +91,28 @@ func (s *sourceFollow) moved(ctx context.Context, branch string) { s.take(ctx, branch) } -// take reads one branch's source revision and applies it. A branch other than the trunk's is left until the indexer keeps -// a catalog graph per watched branch. +// take reads one branch's source revision (branch as the key name source. has it) and applies it: the trunk's +// through the take-in, another watched branch's into its own revision (A-93). func (s *sourceFollow) take(ctx context.Context, branch string) { - if branch != trunkBranch { - return - } a, err := s.read(branch) if err != nil { s.logf("source.%s is not read, and is read again at its next source-moved: %v", branch, err) return } + if branch != trunkBranch { + name, err := keys.UnescapeName(branch) + if err != nil { + s.logf("source.%s names no branch: %v", branch, err) + return + } + if s.applyBranch == nil { + return + } + if err := s.applyBranch(ctx, name, a); err != nil { + s.logf("source.%s at revision %d could not be taken into the branch's revision: %v", branch, a.Revision, err) + } + return + } if err := s.apply(ctx, a); err != nil { s.logf("source.%s at revision %d could not be taken into the modules: %v", branch, a.Revision, err) } diff --git a/modules/mesh-delivery-indexer/internal/index/branch_take_in.go b/modules/mesh-delivery-indexer/internal/index/branch_take_in.go new file mode 100644 index 00000000..82258df1 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/branch_take_in.go @@ -0,0 +1,55 @@ +package index + +import ( + "context" + "errors" + "maps" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// ApplyBranchSource takes a watched branch other than the first in from the source revision the seat answered for it +// (design 47 G §13, §14, §291, ADR 0666 §6, A-85, A-93): a pure computation from the merged messages of the heads the source +// names, over a view in memory seeded from the branch's newest revision, written straight into the branch's next revision +// (KeepBranchEntries): each module's fingerprint, inputs and graph node at the branch's heads, computed as the trunk's are +// (applyMoving). Nothing of the trunk is written: its modules, manifests, taken state, graph and packages stay its own, +// and the branch has no bucket of its own. A head whose merged message is not read keeps no revision, said; the next +// source-moved of the branch takes it in again. +func (ix *Indexer) ApplyBranchSource(ctx context.Context, branch string, a meshdelivery.SourceAnswer) error { + if isTrunkBranch(branch) || ix.Revisions == nil || ix.Message == nil || len(a.Repositories) == 0 { + return nil + } + heads := map[string]string{} + for _, r := range a.Repositories { + heads[r.Repository] = r.Head + } + view, nodes, _, err := BranchView(ctx, ix.Revisions, branch) + if errors.Is(err, process.ErrNotYet) { + ix.logf("catalog graph: the branch %s is not taken in yet: its objects %s are not on the bus", branch, CatalogRevisionsObjects) + return nil + } + if err != nil { + return err + } + d := ix.dry() + // The branch's own view: its modules and graph from its revision, no taken merge, and the packages it publishes in memory + // over the trunk's, never written to them. + d.ix.Modules, d.ix.Graph, d.ix.Taken = view, nodes, newOverlay(nil) + if ix.Packages != nil { + d.ix.Packages = newOverlay(ix.Packages) + } + d.ix.source = sourceNow{rev: uint64(a.Revision), heads: maps.Clone(heads)} + d.ix.Logf = func(format string, args ...any) { + ix.logf("branch "+branch+", computed and not written: "+format, args...) + } + if unread := d.ix.takeInHeads(ctx, heads); len(unread) > 0 { + ix.logf("catalog graph: the branch %s at source revision %d keeps no revision: %d repositories were not taken in (%v); taken in again at its next source-moved", branch, a.Revision, len(unread), unread) + return nil + } + entries, err := d.ix.viewEntries(ctx) + if err != nil { + return err + } + return ix.KeepBranchEntries(ctx, branch, a.Revision, entries) +} diff --git a/modules/mesh-delivery-indexer/internal/index/branch_take_in_test.go b/modules/mesh-delivery-indexer/internal/index/branch_take_in_test.go new file mode 100644 index 00000000..9c9ba9b8 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/branch_take_in_test.go @@ -0,0 +1,75 @@ +package index + +import ( + "context" + "encoding/json" + "fmt" + "maps" + "slices" + "testing" + + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery" +) + +// Design 47 G §13, §14, §291, A-93: a watched branch other than the first is taken in as a pure computation from the merged +// messages of its heads, straight into branch..; the trunk's modules, graph and revision stay the trunk's. +func TestABranchOtherThanTheFirstIsTakenInIntoItsOwnRevisionAndTheTrunkIsLeftAlone(t *testing.T) { + ctx := context.Background() + ix, mods, merged := newFormIndexerUnmerged(t, builderModules(t)...) + ix.Revisions = newMemObjects(0) + ix.Graph = newMem() + ix.Message = func(_ context.Context, repository, head string, _ int) ([]byte, error) { + if repository != catalogueRepo || head != merged.Merge { + return nil, fmt.Errorf("no merge of %s at %s", repository, head) + } + return json.Marshal(merged) + } + var said []meshcatalog.CatalogRevisionEvent + ix.SayCatalogRevision = func(_ context.Context, e meshcatalog.CatalogRevisionEvent) error { + said = append(said, e) + return nil + } + before := map[string]string{} + keys, _ := mods.Keys(ctx) + for _, k := range keys { + v, _, _, _ := mods.Get(ctx, k) + before[k] = string(v) + } + graphBefore, _ := ix.Graph.Keys(ctx) + answer := meshdelivery.SourceAnswer{Branch: "beta", Revision: 9, Repositories: []meshdelivery.SourceHead{{Repository: catalogueRepo, Head: merged.Merge, Trunk: "beta"}}} + if err := ix.ApplyBranchSource(ctx, "beta", answer); err != nil { + t.Fatal(err) + } + if len(said) != 1 || said[0].Revision != 1 || said[0].Branch != "beta" || said[0].Source != 9 || said[0].Entries != int64(len(builderModules(t))) { + t.Fatalf("the branch take-in said %+v, want revision 1 of beta at source 9 with every module", said) + } + rev, entries, err := ReadRevision(ctx, ix.Revisions, BranchObject("beta", 1)) + if err != nil || rev.Branch != "beta" || len(entries) != len(builderModules(t)) { + t.Fatalf("branch.beta.1 = %+v with %d entries (%v)", rev, len(entries), err) + } + for _, e := range entries { + if e.Commit != merged.Merge || e.Inputs == nil { // computed at the branch's head, with the inputs its fingerprint stands on + t.Errorf("the entry of %s is at %q with inputs %v, want the branch's head %s with inputs", e.Module, e.Commit, e.Inputs, merged.Merge) + } + } + if _, any, _ := newestTrunk(ctx, ix.Revisions); any { + t.Error("a branch take-in kept a trunk revision") + } + after := map[string]string{} + keys, _ = mods.Keys(ctx) + for _, k := range keys { + v, _, _, _ := mods.Get(ctx, k) + after[k] = string(v) + } + if !maps.Equal(before, after) { + t.Errorf("a branch take-in wrote the trunk's modules") + } + if graphAfter, _ := ix.Graph.Keys(ctx); !slices.Equal(graphBefore, graphAfter) { + t.Errorf("a branch take-in wrote the trunk's graph: %v, was %v", graphAfter, graphBefore) + } + // Taken in again at the same heads, nothing moved: no second revision. + if err := ix.ApplyBranchSource(ctx, "beta", answer); err != nil || len(said) != 1 { + t.Errorf("the same heads said %d revisions (%v), want one", len(said), err) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go index cea8ecec..cd999c4a 100644 --- a/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go +++ b/modules/mesh-delivery-indexer/internal/index/catalog_revisions.go @@ -246,15 +246,25 @@ func (ix *Indexer) KeepView(ctx context.Context) error { if ix.Revisions == nil || ix.Modules == nil || ix.catchingUp.Load() { return nil } - held, err := ix.heldModules(ctx) + entries, err := ix.viewEntries(ctx) if err != nil { return err } + return ix.keepEntries(ctx, entries) +} + +// viewEntries is the entry of every module the indexer's view (its modules and its module graph) holds: the module's value +// with its node of the graph, as KeepView keeps them. +func (ix *Indexer) viewEntries(ctx context.Context) (map[string]meshcatalog.CatalogEntry, error) { + held, err := ix.heldModules(ctx) + if err != nil { + return nil, err + } nodes := graph.Graph{} if ix.Graph != nil { keys, err := ix.Graph.Keys(ctx) if err != nil && !errors.Is(err, process.ErrNotYet) { - return fmt.Errorf("the module graph: %w", err) + return nil, fmt.Errorf("the module graph: %w", err) } for _, k := range keys { if value, _, found, err := ix.Graph.Get(ctx, k); err == nil && found { @@ -275,7 +285,7 @@ func (ix *Indexer) KeepView(ctx context.Context) error { } entries[name] = ix.withArtifacts(ctx, CatalogEntryOf(name, cur, m, raw, nodes[name]), m) } - return ix.keepEntries(ctx, entries) + return entries, nil } // keepEntries keeps entries as the trunk revision after the newest, says catalog-revision, and keeps none when no entry @@ -411,8 +421,14 @@ func (ix *Indexer) RoomAgain(ctx context.Context) (bool, error) { // graph (NodeOf), and the revision it was read at (0 when the bucket keeps none yet: both start empty and the first start's // catch-up fills them). A lost entry is an error naming the module (ReadRevision), never a module left out of the view. func TrunkView(ctx context.Context, o Objects) (modules, nodes Store, at int64, err error) { + return BranchView(ctx, o, TrunkBranch) +} + +// BranchView is TrunkView of a watched branch: the view read from the newest revision the bucket keeps of that branch +// (design 47 G §291, A-93), empty and at 0 when it keeps none. +func BranchView(ctx context.Context, o Objects, branch string) (modules, nodes Store, at int64, err error) { view, held := newOverlay(nil), newOverlay(nil) - newest, any, err := newestTrunk(ctx, o) + newest, any, err := newestOf(ctx, o, branch) if err != nil || !any { return view, held, 0, err }