Review r1: candidates and the trunk take the source's head for an unpublished package; the own build is read inside ownBuildVia
Builds: design 47 G A-365, A-366 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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).
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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"})
|
||||
|
||||
Reference in New Issue
Block a user