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 15bcb241..f1aa923e 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -462,7 +462,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Predefined: awaitedStore{b: predefined}, ConfigCandidates: awaitedStore{b: configCandidates}, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), ReadOpen: openVia(calls.Call), Packages: awaitedStore{b: packages, js: js, name: index.Bucket(cred.Module, index.PackagesState)}, Files: filesVia(calls.Call), GoSum: goSumVia(calls.Call), GoMod: goModVia(calls.Call), - OwnBuild: ownBuildFirst(process.OwnBuild(), ownBuildVia(calls.Call, cred.Module, cred.Node)), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph, + OwnBuild: ownBuildVia(calls.Call, cred.Module, cred.Node), OwnBuildState: awaitedStore{b: ownBuildState}, Heard: heardStore, Placement: placementVia(calls.CallKept), RunningWhole: runningWholeVia(js), Runs: awaitedStore{b: runs}, Graph: graph, SeatDefinitions: awaitedStore{b: seatDefs}, ReadFile: readFileVia(calls.Call), SaySeatDefinition: saySeats, Revisions: revisions, EntryIndex: awaitedStore{b: process.AwaitBucket(ctx, js, conn, index.Bucket(cred.Module, index.EntryIndexState), "no entry is found by its fingerprint", logger.Printf)}, // Message a kept call: a read, called again through a transport fault (ADR 0668 §1) // Each catalog revision may bring another holder of a build seat: the build modules are found again (buildmodules.go). diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go index 80eed033..c628ec4a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/named_test.go @@ -136,12 +136,15 @@ func TestTheTrunkIsTheDirectoriesTheForgeListsTrees(t *testing.T) { // 1900/AC3 {"started with a fingerprint", process.Build{Fingerprint: "src2:env"}, "src2:env", false}, {"started without one", process.Build{Unknown: process.OwnBuildUnknown}, "src2:abc", true}, } { + started := startedBuild + startedBuild = func() process.Build { return row.started } + defer func() { startedBuild = started }() called := false ask := func(string, any) (json.RawMessage, error) { called = true return json.RawMessage(`{"runs":{"fp":"src2:abc"}}`), nil } - if fp, err := ownBuildFirst(row.started, ownBuildVia(ask, "acme-indexer", "n1"))(context.Background()); err != nil || fp != row.want || called != row.wantAsked { + if fp, err := ownBuildVia(ask, "acme-indexer", "n1")(context.Background()); err != nil || fp != row.want || called != row.wantAsked { t.Errorf("%s: own build = %q %v, engine asked %v, want %q asked %v", row.name, fp, err, called, row.want, row.wantAsked) } } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/ownbuild.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/ownbuild.go index dda2099a..dd7ce7cf 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/ownbuild.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/ownbuild.go @@ -14,20 +14,17 @@ import ( // from the trunk's value of itself or the haves bucket. const compositionVerb = "seat:node-engine.composition" -// ownBuildFirst is the build the process was started with (mesh-sdk process.OwnBuild: MESH_FINGERPRINT, set by the node-engine -// and by a unit started by hand); only when it is unset does it ask the engine through ask (design 47 G A-366). -func ownBuildFirst(started process.Build, ask func(ctx context.Context) (string, error)) func(ctx context.Context) (string, error) { - return func(ctx context.Context) (string, error) { - if started.Fingerprint != "" { - return started.Fingerprint, nil - } - return ask(ctx) - } -} +// startedBuild is the build the process was started with (mesh-sdk process.OwnBuild: MESH_FINGERPRINT, set by the node-engine and +// by a unit started by hand); a variable so a test starts the process with another. +var startedBuild = process.OwnBuild -// ownBuildVia asks the engine of node which fingerprint of the indexer runs there. +// ownBuildVia is the build of the indexer: the one it was started with, else what the engine of node says which fingerprint of the indexer runs there. func ownBuildVia(call toolCaller, module, node string) func(ctx context.Context) (string, error) { return func(context.Context) (string, error) { + // The build the process was started with is the build; the engine is asked only when there is none (design 47 G A-366). + if fp := startedBuild().Fingerprint; fp != "" { + return fp, nil + } raw, err := call(compositionVerb+"@"+node, map[string]any{"module": module}) if err != nil { return "", fmt.Errorf("%s: %w", compositionVerb, err) diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go index 44e4647e..45036e59 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews.go +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -323,7 +323,7 @@ func (ix *Indexer) candidateModule(ctx context.Context, m, dir string, pv meshde fp, none, versions = f.fp, f.none, f.versions } else { inputs.Imports, inputs.ImportForm = nil, "" - versions = candidatePackages(fingerprint.OtherPackages(inputs.Packages, pv.Repository), held, members) + versions = candidatePackages(fingerprint.OtherPackages(inputs.Packages, pv.Repository), held, members, ix.sourceHeads()) for p, v := range pins { if _, named := versions[p]; named || slices.Contains(inputs.Packages, p) { if versions == nil { @@ -352,8 +352,8 @@ func (ix *Indexer) candidateModule(ctx context.Context, m, dir string, pv meshde // candidatePackages is the version each package of a candidate's module is built against (novox/hq ADR 0573 §2, §4): the // head of the candidate's pull request whose repository publishes it (members, by repository), else its latest published -// version, as the trunk's modules hold it. A package with neither is left out, and the fingerprint is then none. -func candidatePackages(packages []string, held map[string]meshcatalog.ModulesByModuleValue, members map[string]string) sdkstep.PackageVersions { +// version, as the trunk's modules hold it, else the commit the gate's source names for its repository (heads; design 47 G A-365). A package with none is left out, and the fingerprint is then none. +func candidatePackages(packages []string, held map[string]meshcatalog.ModulesByModuleValue, members, heads map[string]string) sdkstep.PackageVersions { versions, _ := fingerprint.HeldVersionsOf(packages, held) for _, p := range packages { repository, _, _ := trunk.PackageRepository(p) @@ -362,6 +362,14 @@ func candidatePackages(packages []string, held map[string]meshcatalog.ModulesByM versions = sdkstep.PackageVersions{} } versions[p] = head + } else if _, named := versions[p]; !named { + // A package no module publishes, of a repository outside the group: the commit the gate's source names (A-365). + if head := heads[repository]; len(head) == 40 { + if versions == nil { + versions = sdkstep.PackageVersions{} + } + versions[p] = head + } } } return versions diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 8f1b3a5c..61e7ee40 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -160,12 +160,16 @@ func TestACandidatesModuleGetsTheFingerprintATrunkModuleOfTheSameTreeHas(t *test // mesh-sdk that nothing published gives no fingerprint, and one of its own repository's comes with its tree. const sdk, own = "git.novox.be/novox/mesh-sdk/go", "git.novox.be/novox/mesh-catalog/contract" held := map[string]meshcatalog.ModulesByModuleValue{"mesh-sdk": {Repository: "novox/mesh-sdk", Path: meshcatalogclient.RootPath, Commit: trunkOld}} - if got := candidatePackages([]string{sdk}, held, map[string]string{catalogueRepo: h1}); got[sdk] != trunkOld { + if got := candidatePackages([]string{sdk}, held, map[string]string{catalogueRepo: h1}, nil); got[sdk] != trunkOld { t.Errorf("outside a group: %v, want mesh-sdk at its trunk %s", got, trunkOld) } - if got := candidatePackages([]string{sdk}, held, map[string]string{catalogueRepo: h1, "novox/mesh-sdk": mergeSha}); got[sdk] != mergeSha { + if got := candidatePackages([]string{sdk}, held, map[string]string{catalogueRepo: h1, "novox/mesh-sdk": mergeSha}, nil); got[sdk] != mergeSha { t.Errorf("in a group with mesh-sdk: %v, want its member's head %s", got, mergeSha) } + // Design 47 G A-365: a package no module publishes, of a repository outside the group, takes the commit the source names. + if got := candidatePackages([]string{sdk}, nil, map[string]string{catalogueRepo: h1}, map[string]string{"novox/mesh-sdk": strings.Repeat("9", 40)}); got[sdk] != strings.Repeat("9", 40) { + t.Errorf("an unpublished package outside the group: %v, want the source's head", got) + } m.Packages = []string{own, sdk} pv.Modules["modules/avahi"] = m pv.Merge = "8888888888888888888888888888888888888888" @@ -173,6 +177,13 @@ func TestACandidatesModuleGetsTheFingerprintATrunkModuleOfTheSameTreeHas(t *test if got := fpOf(t, cands); got.Fp2 != "" || got.None2 != fingerprint.NoPackageVersion+sdk { t.Errorf("a package nothing published: %+v", got) } + // Through the indexer's wiring (A-365): once the gate's source names mesh-sdk's head, the candidate is built against it. + if err := applyS(ix, context.Background(), 5, srcT{Group: "g", Heads: map[string]string{catalogueRepo: trunkNow, "novox/mesh-sdk": strings.Repeat("9", 40)}}); err != nil { + t.Fatal(err) + } + if got := fpOf(t, cands); got.Fp2 == "" { + t.Errorf("a package nothing published, the source naming its repository: %+v, want a fingerprint", got) + } } func TestAnUnknownToolchainLeavesTheCandidatesModuleNamingIt(t *testing.T) { // 1445/AC2 diff --git a/modules/mesh-delivery-indexer/internal/index/source.go b/modules/mesh-delivery-indexer/internal/index/source.go index c87c152e..ed801dff 100644 --- a/modules/mesh-delivery-indexer/internal/index/source.go +++ b/modules/mesh-delivery-indexer/internal/index/source.go @@ -12,6 +12,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "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-sdk/trunk" ) @@ -63,6 +64,7 @@ func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) ix.mu.Unlock() return nil } + before := ix.source.heads ix.source = sourceNow{rev: rev, heads: maps.Clone(s.Heads)} ix.mu.Unlock() if ix.Modules == nil { @@ -115,6 +117,13 @@ func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) return fmt.Errorf("modules.%s: %w", key, err) } } + // A module built against the commit the source named as a repository's head (a package no module publishes, A-365) is + // computed again when that head moves: the gate moves the source only after the take-in. + if movedPackages(held, before, s.Heads) { + if err := ix.Recompute(ctx); err != nil { + return err + } + } // The stamps are kept with the view (KeepView), so a restart reads them back. if err := ix.KeepView(ctx); err != nil { return err @@ -123,6 +132,19 @@ func (ix *Indexer) ApplySource(ctx context.Context, a meshdelivery.SourceAnswer) return ix.RecomputeCandidates(ctx) } +// movedPackages is whether a held module's package version is the previous head of a repository whose head the new source moved. +func movedPackages(held map[string]meshcatalog.ModulesByModuleValue, before, now map[string]string) bool { + for _, m := range held { + for p, v := range m.Packages { + repo, _, ok := trunk.PackageRepository(p) + if ok && before[repo] != "" && before[repo] == v && now[repo] != before[repo] { + return true + } + } + } + return false +} + // kickCatchUp takes the trunk in at a first start, when the objects keep no trunk revision yet (the lead's ruling A on // milestone 47/G: the catch-up path): the head the source names for every repository, by the merged message the forge // bridge gives for it, each applied as its merge, and revision 1 kept from them all at its end. Until then the revision diff --git a/modules/mesh-delivery-indexer/internal/index/source_test.go b/modules/mesh-delivery-indexer/internal/index/source_test.go index 96fc8758..6aa46d1a 100644 --- a/modules/mesh-delivery-indexer/internal/index/source_test.go +++ b/modules/mesh-delivery-indexer/internal/index/source_test.go @@ -5,6 +5,7 @@ import ( "reflect" "testing" + "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" ) @@ -116,3 +117,21 @@ func TestAMergeThatNoSourceRevisionNamesIsWrittenAsToday(t *testing.T) { // 1344 t.Errorf("module a = %+v, want the new merge with Src 0", m) } } + +func TestAModuleBuiltAgainstASourceHeadIsComputedAgainWhenThatHeadMoves(t *testing.T) { // design 47 G A-365 + held := map[string]meshcatalog.ModulesByModuleValue{"a": {Packages: map[string]string{"git.novox.be/novox/mesh-contracts/go/seats/findings": otherMerge}}} + for _, row := range []struct { + name string + before, now map[string]string + want bool + }{ + {"the head its version came from moved", map[string]string{"novox/mesh-contracts": otherMerge}, map[string]string{"novox/mesh-contracts": merge}, true}, + {"the head stands", map[string]string{"novox/mesh-contracts": otherMerge}, map[string]string{"novox/mesh-contracts": otherMerge}, false}, + {"another repository's head moved", map[string]string{"novox/mesh-contracts": otherMerge}, map[string]string{"novox/mesh-contracts": otherMerge, "novox/mesh-sdk": merge}, false}, + {"no source seen before", nil, map[string]string{"novox/mesh-contracts": merge}, false}, + } { + if got := movedPackages(held, row.before, row.now); got != row.want { + t.Errorf("%s: movedPackages = %v, want %v", row.name, got, row.want) + } + } +} diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index a1c1e3ca..49f926c5 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -5271,3 +5271,55 @@ func TestPassBeforeTheCatalogHalfComposesFromTheTrunkToo(t *testing.T) { }) } } + +// Design 47 G A-371: a seat stream the part reads whole at its start, that it creates itself and the bus does not hold yet, has an +// empty first view and is no wait: with the catalog half unread and the stream's read waiting, a list that changes is sealed and not +// held for a failed read; with the stream held on the bus it still waits. +func TestAStreamThePartCreatesAndTheBusLacksIsAnEmptyFirstView(t *testing.T) { + const stream = "SEAT_node-engine_running" + days, mib := 7, 64 + gate := keptGate{seatdef.Definition{Seat: "node-engine", Events: []seatdef.Message{ + {Name: "running", Keep: &seatdef.Keep{Object: true, MaxAgeDays: &days, MaxMiB: &mib}}}}} + for _, row := range []struct { + name string + held bool // the bus holds the stream, so the read is a wait + }{{"the bus lacks the stream", false}, {"the bus holds the stream", true}} { + t.Run(row.name, func(t *testing.T) { + b := fixture() + busKey, err := seal.GenerateKeyPair() + if err != nil { + t.Fatal(err) + } + b.put(grants.ReadHave, "a.nats", nodeengine.HaveByNodeModuleValue{V: keys.Version, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK, + UserList: &nodeengine.UserListWritten{Source: grantsstate.GrantsUserListSource, Revision: 8, Wanted: "sha256:aa", Digest: "sha256:aa", Users: 5, + Key: busKey.PublicKey, Follows: grantsstate.FollowGrants}}) + for _, u := range shadowOf(t, b) { + b.put(grants.ReadAccounts, u.User, map[string]any{"user": u.User, "hash": "$2a$10$x", "generation": 1, "fingerprint": "sha256:1"}) + } + p := partOn(b) + p.Bus = grants.BusAddress{URL: "nats://a.mesh:4222", Fingerprint: "ab"} + p.Objects = &fakeObjects{} + p.Compose() + if g, _, _ := p.Sealed(); g.Digest == "" { + t.Fatal("no list in force sealed") + } + // Now the catalog half is unread, the read of the stream waits, and the list changes (mod-x runs on b too). + b.mu.Lock() + b.catalogWaited = []string{"the catalog half of the branch main"} + if row.held { + b.streams = append(b.streams, map[string]any{"name": stream, "subjects": []string{}, "retention": "limits"}) + } + b.mu.Unlock() + p.SeatList = gate + p.ReadingRunning(stream) + b.assign("b", "mod-x") + res := p.Compose() + if res.Sealed == nil { + t.Fatal("the pass says nothing of the sealed list") + } + if held := strings.Contains(res.Sealed.Why, "a read of this pass failed"); held != row.held { + t.Errorf("held for a failed read = %v, want %v: %q", held, row.held, res.Sealed.Why) + } + }) + } +} diff --git a/modules/mesh-grants/internal/grants/running.go b/modules/mesh-grants/internal/grants/running.go index 5d21e384..4d219838 100644 --- a/modules/mesh-grants/internal/grants/running.go +++ b/modules/mesh-grants/internal/grants/running.go @@ -98,10 +98,25 @@ func (p *Part) StreamReads() []StreamRead { return []StreamRead{r} } +// createsAndLacks says a stream the part reads whole at its start is one of the kept event streams it creates itself and the bus +// does not hold yet: its first view is empty, no wait (design 47 G, A-371). The pass composes, writes the part's own create right, +// and the part creates the stream and hears it from its start. A bus whose streams cannot be read leaves the read a wait. +func (p *Part) createsAndLacks(in Inputs, stream string) bool { + if !slices.ContainsFunc(in.Kept, func(k buscompose.KeptEvent) bool { return k.Stream == stream }) || p.Call == nil { + return false + } + held, err := HeldStreams(p.Call) + if err != nil { + return false + } + _, has := held[stream] + return !has +} + // readRan reads each key of StateRunning and the manifest its digest names (the indexer's, by digest), into in.Ran. A key // whose manifest cannot be read is a gap, and that node's module is read as it was before (the trunk's). func (p *Part) readRan(in *Inputs) { - if reads := p.StreamReads(); len(reads) > 0 && reads[0].State == "waiting" { + if reads := p.StreamReads(); len(reads) > 0 && reads[0].State == "waiting" && !p.createsAndLacks(*in, reads[0].Stream) { // The whole read of `running` has not ended: the part acts on none of it, and keeps the list it has (design 47 G §556). in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Failed: true, Why: "the stream " + reads[0].Stream + " is waiting: its whole read at the part's start has not ended"})