From 5d26c658f5fd31b72fd08a55c7841b11859b0a91 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 1 Nov 2026 16:18:44 +0100 Subject: [PATCH] Compose a node's plans once per read, resolving each machine once (issue 1954) A node's plans took 2.6-13.6 s of CPU on the provisioner: one composition resolved every machine of the mesh some two hundred times (who is on the private network, and the rest of the mesh, worked out again on every look), three compositions did so again each, and the holder composed twice. A composer now works each out once over its unchanging Input; Composing gives Resolve, Render and Generators one composer; PlansHeldTwice composes the wanted and the newest builds over one read of the bus, says where the time went by phase, and the states reader keeps a build's take-in by its content. WatchedStates takes a process's own writes into the copy at once, so a holder reads its own states from it. Measured on a captured novox mesh: 11.7 s to 0.49 s per composition of an unchanged node, plans and provisions identical on all four nodes. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- go/compose/frombus/frombus.go | 149 +++++++++++++----- .../frombus/frombustest/frombustest_test.go | 93 +++++++++++ go/compose/frombus/states.go | 11 +- go/compose/frombus/twice.go | 147 +++++++++++++++++ go/compose/network.go | 25 +++ go/compose/render.go | 20 +++ go/compose/resolve.go | 40 ++++- go/compose/rest.go | 17 ++ go/state_test.go | 20 +++ go/watched.go | 59 ++++++- 10 files changed, 535 insertions(+), 46 deletions(-) create mode 100644 go/compose/frombus/twice.go diff --git a/go/compose/frombus/frombus.go b/go/compose/frombus/frombus.go index a74764f..2960221 100644 --- a/go/compose/frombus/frombus.go +++ b/go/compose/frombus/frombus.go @@ -223,9 +223,56 @@ func taken(name string, raw []byte, m delivery.Module) (man catalogue.Manifest, // manifest is refused is read for its offer alone (unusable). A manifest the bus does not answer for still ends the // read: that is the bus's, not the module's. func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module string, err error), wanted map[string]string) (compose.Input, error) { + b, err := readBase(ctx, r, mesh, s) + b.refuseOwn(refuse) + if err != nil { + return compose.Input{}, err + } + return b.finish(ctx, r, r.Build, mesh, s, b.in, refuse, wanted) +} + +// base is the part of a node's input that the fingerprints it composes its modules at leave alone (issue 1954): read once +// for its plans at the wanted builds and at the newest, whose builds alone differ (PlansHeldTwice). +type base struct { + in compose.Input + placements map[string]delivery.Placement + records map[string]delivery.NodeRecord + modules map[string]delivery.Module + // ownRefused is each module placed on the composing node whose manifest the indexer refused, with why, in the order + // read: every scope holds it. + ownRefused [][2]string +} + +// refuseOwn hands refuse each module placed on the composing node whose manifest is refused. +func (b base) refuseOwn(refuse func(module string, err error)) { + if refuse == nil { + return + } + for _, r := range b.ownRefused { + refuse(r[0], errors.New(r[1])) + } +} + +// cloned is the input with its own catalogue, nodes and kept manifests, what finishing it sets: two scopes finished over +// one base never see each other's builds. +func (b base) cloned() compose.Input { + in := b.in + in.Catalogue = maps.Clone(b.in.Catalogue) + in.Nodes = slices.Clone(b.in.Nodes) + for i := range in.Nodes { + in.Nodes[i].Kept = maps.Clone(in.Nodes[i].Kept) + } + in.KeptCoreBoxes = slices.Clone(b.in.KeptCoreBoxes) + in.OwnSecrets = slices.Clone(b.in.OwnSecrets) + return in +} + +// readBase is input up to the builds: placement, records, the indexer's modules and manifests, each node's values. +func readBase(ctx context.Context, r Reader, mesh Mesh, s Scope) (base, error) { + var b base placements, err := r.Placements(ctx) if err != nil { - return compose.Input{}, fmt.Errorf("frombus: placement: %w", err) + return b, fmt.Errorf("frombus: placement: %w", err) } if len(s.Placements) > 0 { given := make(map[string]delivery.Placement, len(placements)+len(s.Placements)) @@ -238,7 +285,7 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module continue } if p.Node != node { - return compose.Input{}, fmt.Errorf("frombus: the placement given for %s names the node %q", node, p.Node) + return b, fmt.Errorf("frombus: the placement given for %s names the node %q", node, p.Node) } given[node] = *p } @@ -246,14 +293,14 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module } records, err := r.Records(ctx) if err != nil { - return compose.Input{}, fmt.Errorf("frombus: node records: %w", err) + return b, fmt.Errorf("frombus: node records: %w", err) } modules, err := r.Modules(ctx) if err != nil { - return compose.Input{}, fmt.Errorf("frombus: the indexer's modules: %w", err) + return b, fmt.Errorf("frombus: the indexer's modules: %w", err) } if _, ok := placements[s.Node]; !ok && !s.Every { - return compose.Input{}, fmt.Errorf("frombus: %s has no placement in the catalogue", s.Node) + return b, fmt.Errorf("frombus: %s has no placement in the catalogue", s.Node) } in := compose.Input{OverlayCIDR: mesh.OverlayCIDR, InternalSuffix: mesh.InternalSuffix, Authority: mesh.Authority, @@ -279,18 +326,6 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module in.Authority = delivery.AuthorityBundle(in.Authority, vault) } } - parse := parser(mesh) - read := func(module, digest string) (catalogue.Manifest, error) { - raw, err := r.Manifest(ctx, digest) - if err != nil { - return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", module, digest, err) - } - man, err := parse(raw) - if err != nil { - return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s) does not decode: %w", module, digest, err) - } - return man, nil - } named := func(field *map[string]string, module, why string) { if *field == nil { *field = map[string]string{} @@ -309,7 +344,7 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module raw, err := r.Manifest(ctx, m.Manifest) if err != nil { // Not read is no refusal of the module: a bus that does not answer is waited out by the caller. - return compose.Input{}, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", name, m.Manifest, err) + return b, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", name, m.Manifest, err) } // One module's manifest the indexer refused refuses that module alone, by the rule a build that cannot be read // follows (issue 774: power's bar item refused every node's composition, 2026-10-30; ADR 0578 §3: one refused @@ -318,8 +353,8 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module man, why := taken(name, raw, m) if why != "" { named(&in.Refused, name, why) - if refuse != nil && ownPlaced[name] { - refuse(name, errors.New(why)) + if ownPlaced[name] { + b.ownRefused = append(b.ownRefused, [2]string{name, why}) } } else if aside := man.SetAside(); aside != "" { named(&in.SetAside, name, aside) @@ -332,11 +367,11 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module for _, name := range delivery.ProvidedModules() { raw, err := json.Marshal(overlay.Manifest()) if err != nil { - return compose.Input{}, err + return b, err } var private catalogue.Manifest if err := json.Unmarshal(raw, &private); err != nil { - return compose.Input{}, fmt.Errorf("frombus: the manifest mesh-sdk provides for %s: %w", name, err) + return b, fmt.Errorf("frombus: the manifest mesh-sdk provides for %s: %w", name, err) } in.Catalogue[name] = private } @@ -357,13 +392,13 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module var values map[string]any if withValues { if values, err = r.Values(ctx, node, m.Module); err != nil { - return compose.Input{}, fmt.Errorf("frombus: %s's config values on %s: %w", m.Module, node, err) + return b, fmt.Errorf("frombus: %s's config values on %s: %w", m.Module, node, err) } } else if keys := catalogue.ServedSettings(in.Catalogue[m.Module]); len(keys) > 0 { // What a provider elsewhere serves from its settings is told to every consumer: the composing node reads // those keys of it, and no other (issue 1126: mailu's smtp domain, a consumer's on shanks). if values, err = r.Served(ctx, node, m.Module, keys); err != nil { - return compose.Input{}, fmt.Errorf("frombus: what %s serves on %s (%s): %w", m.Module, node, + return b, fmt.Errorf("frombus: what %s serves on %s (%s): %w", m.Module, node, strings.Join(keys, ", "), err) } } @@ -436,14 +471,35 @@ func input(ctx context.Context, r Reader, mesh Mesh, s Scope, refuse func(module in.CarriedPeers = carriedPeers(in.Adopted.Tunnel, placements[in.Adopted.Hub].Decided.PeerNames, records) } if len(missing) > 0 { - return compose.Input{}, fmt.Errorf("frombus: modules are placed on %s, whose agent has written no record yet", + return b, fmt.Errorf("frombus: modules are placed on %s, whose agent has written no record yet", strings.Join(missing, ", ")) } + b.in, b.placements, b.records, b.modules = in, placements, records, modules + return b, nil +} + +// finish is input from its base on: the composing node's modules from their builds (build reads one), its own secrets. +// in is the base's input, or a copy of it (base.cloned) where one base is finished more than once. +func (b base) finish(ctx context.Context, r Reader, build func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error), + mesh Mesh, s Scope, in compose.Input, refuse func(module string, err error), wanted map[string]string) (compose.Input, error) { + var err error + parse := parser(mesh) + read := func(module, digest string) (catalogue.Manifest, error) { + raw, err := r.Manifest(ctx, digest) + if err != nil { + return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s): %w", module, digest, err) + } + man, err := parse(raw) + if err != nil { + return catalogue.Manifest{}, fmt.Errorf("frombus: the manifest of %s (digest %s) does not decode: %w", module, digest, err) + } + return man, nil + } if !s.Every { - if _, ok := records[s.Node]; !ok { + if _, ok := b.records[s.Node]; !ok { return compose.Input{}, fmt.Errorf("frombus: %s has written no record of itself yet", s.Node) } - if err := fromBuilds(ctx, r, &in, s, placements[s.Node], modules, read, parse, refuse, wanted); err != nil { + if err := fromBuilds(ctx, r, build, &in, s, b.placements[s.Node], b.modules, read, parse, refuse, wanted); err != nil { return compose.Input{}, err } // Its own modules' secrets, sealed to its own key by the vault: a composition carries them sealed (ADR 0550 §5). @@ -603,15 +659,8 @@ func PlansOf(in compose.Input, node string) (map[string]catalogue.Plan, error) { // plansIn is the controller's handedPlans in a mode: Foreseeing is the controller's own. func plansIn(in compose.Input, node string, mode compose.Mode) (map[string]catalogue.Plan, error) { - resolved, err := compose.Resolve(in, node) - if err != nil { - return nil, fmt.Errorf("frombus: the plans of %s: %w", node, err) - } - rendered, err := compose.Render(in, node, mode) - if err != nil { - return nil, fmt.Errorf("frombus: the plans of %s: %w", node, err) - } - gens, err := compose.Generators(in) + // One composition for the three, so each node of the mesh is resolved once (issue 1954). + resolved, rendered, gens, err := compose.Composing(in, node, mode) if err != nil { return nil, fmt.Errorf("frombus: the plans of %s: %w", node, err) } @@ -673,7 +722,8 @@ func ranOf(ctx context.Context, r Reader, mesh Mesh, running []delivery.Composed // module it keeps at another build (Scope.KeptAt, into its node's Kept), read from the builder's done key and resolved // by the one take-in rule (takein.Manifest), a templates archive read in the artifact store as the node reaches it // (issue 1126). -func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p delivery.Placement, modules map[string]delivery.Module, +func fromBuilds(ctx context.Context, r Reader, build func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error), + in *compose.Input, s Scope, p delivery.Placement, modules map[string]delivery.Module, read func(module, digest string) (catalogue.Manifest, error), parse func([]byte) (catalogue.Manifest, error), refuse func(module string, err error), wanted map[string]string) error { // stub keeps a refused module in the composition as a stub for this node alone (compose.Stub): what it provides @@ -716,6 +766,24 @@ func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p del if b.Manifest == "" { return catalogue.Manifest{}, fmt.Errorf("frombus: the build of %s at %s names no manifest", module, fp) } + // A take-in kept by the reader is read back as below, never shared: each composition holds its own manifest. + keeper, keeps := r.(takenInKeeper) + if keeps && !addressed { + address, addressErr = compose.ArtifactStoreAddress(*in, s.Node) + addressed = true + } + // Kept by the store's address too: a node that reaches no store takes in as before, and fails where it did. + key, keyed := takenInKey(module, fp, address, *b) + keyed = keyed && addressErr == nil && address != "" + if keeps && keyed { + if raw, ok := keeper.takenIn(key); ok { + resolved, err := parse(raw) + if err != nil { + return catalogue.Manifest{}, fmt.Errorf("frombus: the built manifest of %s at %s: %w", module, fp, err) + } + return resolved, nil + } + } src, err := read(module, b.Manifest) if err != nil { return catalogue.Manifest{}, err @@ -733,6 +801,9 @@ func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p del if resolved, err = parse(raw); err != nil { return catalogue.Manifest{}, fmt.Errorf("frombus: the built manifest of %s at %s: %w", module, fp, err) } + if keeps && keyed { + keeper.keepTakenIn(key, raw) + } return resolved, nil } // A module wanted at another build than the trunk's is composed from the build its want names (issue 774), as one @@ -766,7 +837,7 @@ func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p del if man, ok := in.Catalogue[pm.Module]; !ok || (man.Build == nil && !man.Templated()) { continue // nothing is built of it: its manifest is what it composes from } - b, err := r.Build(ctx, pm.Module, m.FP2) + b, err := build(ctx, pm.Module, m.FP2) if err != nil { err = fmt.Errorf("frombus: the build of %s at %s: %w", pm.Module, m.FP2, err) if refuse == nil { @@ -801,7 +872,7 @@ func fromBuilds(ctx context.Context, r Reader, in *compose.Input, s Scope, p del } for _, module := range sortedKeys(keptAt) { fp := keptAt[module] - b, err := r.Build(ctx, module, fp) + b, err := build(ctx, module, fp) switch { case err != nil: err = fmt.Errorf("frombus: the build of %s at %s: %w", module, fp, err) diff --git a/go/compose/frombus/frombustest/frombustest_test.go b/go/compose/frombus/frombustest/frombustest_test.go index 84ec290..97c71b4 100644 --- a/go/compose/frombus/frombustest/frombustest_test.go +++ b/go/compose/frombus/frombustest/frombustest_test.go @@ -2,10 +2,12 @@ package frombustest import ( "context" + "reflect" "strings" "testing" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/compose/fromruntime" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -117,6 +119,18 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) { restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" { t.Fatalf("%s: the plans are %+v (held %v)", row.name, plans, held) } + // Issue 1954: the plans at the wanted builds and at the newest, over one read, are what PlansHeld answers for each, + // and where the time went is said by phase. + keptAt := FPs() + at, newest, took := frombus.PlansHeldTwice(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node, KeptAt: keptAt}) + wantAt, heldAt, errAt := frombus.PlansHeld(context.Background(), b, frombus.Mesh{}, frombus.Scope{Node: Node, KeptAt: keptAt}) + if !reflect.DeepEqual(at, frombus.HeldPlans{Plans: wantAt, Held: heldAt, Err: errAt}) || + !reflect.DeepEqual(newest, frombus.HeldPlans{Plans: plans, Held: held}) { + t.Errorf("%s: twice composed %+v and %+v, want %v %v %v and %v %v", row.name, at, newest, wantAt, heldAt, errAt, plans, held) + } + if said := frombus.PhasesSaid(took); !strings.HasPrefix(said, "read ") || !strings.Contains(said, "plans at newest ") { + t.Errorf("%s: the phases are said as %q", row.name, said) + } if row.bad == "" || row.elsewhere { if len(held) > 0 { t.Errorf("%s: held %v", row.name, held) @@ -167,6 +181,85 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) { if fetched != 1 { t.Errorf("a templates archive was fetched %d times, want once", fetched) } + + // Issue 1954: the states reader takes a build in once per its content: composed again, its archive is not read again + // and the plans are the same. + b, err := Provisioned() + if err != nil { + t.Fatal(err) + } + reader, archives := frombus.StatesReader(statesOf(t, b), func(_ context.Context, url string) ([]byte, error) { + fetched++ + _, digest, _ := strings.Cut(url, "/blobs/") + return b.Archives[digest], nil + }), fetched + var first map[string]catalogue.Plan + for i := range 2 { + plans, held, err := frombus.PlansHeld(context.Background(), reader, frombus.Mesh{}, frombus.Scope{Node: Node}) + if err != nil || len(held) > 0 || (i == 1 && !reflect.DeepEqual(plans, first)) || !planned(plans, "restic") { + t.Fatalf("composed %v (held %v): %v", plans, held, err) + } + first = plans + } + if fetched-archives != 1 { + t.Errorf("restic's archive was read %d times over two compositions, want once", fetched-archives) + } +} + +// statesOf is the bus b as a module's read states hold it. +func statesOf(t *testing.T, b *Bus) frombus.States { + t.Helper() + s := states{} + put := func(state, key string, raw []byte, err error) { + if err != nil { + t.Fatal(err) + } + if s[state] == nil { + s[state] = map[string][]byte{} + } + s[state][key] = raw + } + for node, p := range b.Placement { + raw, err := delivery.EncodePlacement(p) + put(frombus.PlacementRead, "placement."+node, raw, err) + } + for node, rec := range b.Record { + raw, err := delivery.EncodeHaveNode(delivery.HaveNode{Record: &rec}) + put(frombus.HaveRead, node+".node", raw, err) + } + for module, m := range b.Module { + key, _ := delivery.ModulesKey(module) + raw, err := delivery.EncodeModule(m) + put(frombus.ModulesRead, key, raw, err) + } + for digest, man := range b.Manifests { + key, _ := delivery.ManifestsKey(digest) + raw, err := delivery.EncodeManifest(delivery.Manifest{Module: "m", JSON: string(man)}) + put(frombus.ManifestsRead, key, raw, err) + } + for at, build := range b.Builds { + module, fp, _ := strings.Cut(at, " ") + key, _ := delivery.BuilderBuildKey(module, fp) + raw, err := delivery.EncodeBuilderBuild(*build) + put(frombus.BuildsRead, key, raw, err) + } + return s +} + +// states are read states in memory. +type states map[string]map[string][]byte + +func (s states) Get(state, key string) ([]byte, bool, error) { + v, ok := s[state][key] + return v, ok, nil +} + +func (s states) Keys(state string) ([]string, error) { + var keys []string + for k := range s[state] { + keys = append(keys, k) + } + return keys, nil } func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok } diff --git a/go/compose/frombus/states.go b/go/compose/frombus/states.go index 3ea4618..b26c314 100644 --- a/go/compose/frombus/states.go +++ b/go/compose/frombus/states.go @@ -43,13 +43,22 @@ type States interface { type Fetch func(ctx context.Context, url string) ([]byte, error) // StatesReader is the Reader over s; fetch reads a templates archive, nil for a holder that reads none. -func StatesReader(s States, fetch Fetch) Reader { return statesReader{s: s, fetch: fetch} } +// It keeps what each build's take-in made, by the build's content (issue 1954), so a holder composing again and again +// takes in only a build it has not seen. +func StatesReader(s States, fetch Fetch) Reader { + return statesReader{s: s, fetch: fetch, kept: &keptTakeIns{}} +} type statesReader struct { s States fetch Fetch + kept *keptTakeIns } +func (r statesReader) takenIn(key string) ([]byte, bool) { return r.kept.takenIn(key) } + +func (r statesReader) keepTakenIn(key string, raw []byte) { r.kept.keepTakenIn(key, raw) } + func (r statesReader) each(read string, visit func(key string, value []byte)) error { keys, err := r.s.Keys(read) if err != nil { diff --git a/go/compose/frombus/twice.go b/go/compose/frombus/twice.go new file mode 100644 index 0000000..cb133df --- /dev/null +++ b/go/compose/frombus/twice.go @@ -0,0 +1,147 @@ +package frombus + +// **A node's plans at two sets of builds, over one read** (issue 1954; novox/hq ADR 0570 clause 2, ADR 0580 clause 2): +// the seat mesh-provisioner's holder composes each node at the builds its wants name and at the newest build of each +// placed module. Both are composed from one read of the bus (placement, records, the indexer's modules and manifests, +// values: base); only the builds the node's modules are composed from differ, each build read once for both, and a +// build taken in once per content by a reader that keeps it (StatesReader). Where the time went is said by phase. + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// HeldPlans is one scope's plans, as PlansHeld answers them: err a refusal that is no single module's. +type HeldPlans struct { + Plans map[string]catalogue.Plan + Held map[string]string + Err error +} + +// Phase is one step of a composition and how long it took. +type Phase struct { + Name string `json:"phase"` + Took time.Duration `json:"took"` +} + +// PlansHeldTwice is PlansHeld at s and at s with no kept builds (each placed module at its newest build), over one +// read of the bus: what each answers is what PlansHeld answers for that scope. A read that fails is both scopes' err. +// took names each phase in order: the read, then each scope's builds and its plans. +func PlansHeldTwice(ctx context.Context, r Reader, mesh Mesh, s Scope) (at, newest HeldPlans, took []Phase) { + start := time.Now() + mark := func(name string) { + now := time.Now() + took = append(took, Phase{Name: name, Took: now.Sub(start)}) + start = now + } + b, err := readBase(ctx, r, mesh, s) + mark("read") + builds := onceBuilds(r) + newestScope := s + newestScope.KeptAt = nil + for _, sc := range []struct { + scope Scope + into *HeldPlans + name string + }{{s, &at, "wanted"}, {newestScope, &newest, "newest"}} { + held := map[string]string{} + sc.into.Held = held + refuse := func(module string, err error) { held[module] = err.Error() } + b.refuseOwn(refuse) + if err != nil { + sc.into.Err = err + continue + } + in, err := b.finish(ctx, r, builds, mesh, sc.scope, b.cloned(), refuse, nil) + mark("builds at " + sc.name) + if err != nil { + sc.into.Err = err + continue + } + sc.into.Plans, sc.into.Err = plansHeldOf(in, s.Node, held) + mark("plans at " + sc.name) + } + return at, newest, took +} + +// onceBuilds is r's Build, each module's build at a fingerprint read once: both scopes of one composition read the +// builds of the same moment. +func onceBuilds(r Reader) func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) { + type answer struct { + b *delivery.BuilderBuild + err error + } + read := map[string]answer{} + return func(ctx context.Context, module, fp string) (*delivery.BuilderBuild, error) { + key := module + " " + fp + if a, ok := read[key]; ok { + return a.b, a.err + } + b, err := r.Build(ctx, module, fp) + read[key] = answer{b, err} + return b, err + } +} + +// takenInKeeper is a Reader that keeps what a build's take-in made (StatesReader does): the resolved manifest's bytes, +// by the content it was made from, so a composition made again on every change takes in only a build it has not seen. +type takenInKeeper interface { + takenIn(key string) ([]byte, bool) + keepTakenIn(key string, raw []byte) +} + +// takenInKey is what a take-in is made from: the artifact store's address and the build's record (its manifest's digest, its artifacts by digest), read +// whole; an archive is checked against its digest on every read, so the same record takes in to the same bytes. +func takenInKey(module, fp, address string, b delivery.BuilderBuild) (string, bool) { + raw, err := json.Marshal(b) + if err != nil { + return "", false + } + sum := sha256.Sum256(append([]byte(module+" "+fp+" "+address+" "), raw...)) + return hex.EncodeToString(sum[:]), true +} + +// takenInAtMost bounds the take-ins a reader keeps; past it they are emptied, which costs only a take-in again. +const takenInAtMost = 512 + +// keptTakeIns is a reader's kept take-ins. +type keptTakeIns struct { + mu sync.Mutex + kept map[string][]byte +} + +func (k *keptTakeIns) takenIn(key string) ([]byte, bool) { + k.mu.Lock() + defer k.mu.Unlock() + raw, ok := k.kept[key] + return raw, ok +} + +func (k *keptTakeIns) keepTakenIn(key string, raw []byte) { + k.mu.Lock() + defer k.mu.Unlock() + if k.kept == nil || len(k.kept) >= takenInAtMost { + k.kept = map[string][]byte{} + } + k.kept[key] = raw +} + +// PhasesSaid says the phases in one line, each in milliseconds. +func PhasesSaid(took []Phase) string { + out := "" + for i, p := range took { + if i > 0 { + out += ", " + } + out += fmt.Sprintf("%s %d ms", p.Name, p.Took.Milliseconds()) + } + return out +} diff --git a/go/compose/network.go b/go/compose/network.go index f51d6d7..6998bb8 100644 --- a/go/compose/network.go +++ b/go/compose/network.go @@ -27,6 +27,23 @@ type composer struct { // unhostable is, from the last look at the rest of the mesh, each module assigned to another node that this // composition finds that node cannot run: what it would have offered is missing here (issue 1126, probe #532). unhostable []string + + // What one composition looks at again and again, each worked out once over its Input, which it never changes (issue + // 1954: a node's plans resolved every machine of the mesh some two hundred times, 11.7 s of CPU for novox): who is + // on the private network, and the rest of the mesh as each node sees it. + onNetwork []Node + onNetworkSet bool + rest map[string]restOf + plans map[string]planOf +} + +// restOf is theRestOfTheMesh's answer for one excluded node, kept: what it found unhostable beside it. +type restOf struct { + world catalogue.World + passes Passes + left []LeftOut + unhostable []string + err error } func newComposer(in Input) *composer { return &composer{in: in, seen: map[string]bool{}} } @@ -249,6 +266,14 @@ func (c *composer) whereEveryoneIs() (map[string]string, error) { // same rule as network(), so its name and its address stay where the tunnel to it does (issue 1833); one without is not // counted, and said. func (c *composer) onTheNetwork() []Node { + if !c.onNetworkSet { + c.onNetwork, c.onNetworkSet = c.onTheNetworkNow(), true + } + return slices.Clone(c.onNetwork) +} + +// onTheNetworkNow is onTheNetwork worked out: every placed machine's set resolved. +func (c *composer) onTheNetworkNow() []Node { var out []Node for _, n := range c.in.sorted() { if n.Place.Address == "" { diff --git a/go/compose/render.go b/go/compose/render.go index 5eb1ea2..3354fad 100644 --- a/go/compose/render.go +++ b/go/compose/render.go @@ -35,6 +35,26 @@ func Render(in Input, node string, mode Mode) (Rendered, error) { return newComposer(in).render(node, mode) } +// Composing is Resolve, Render and Generators of one node over one composition, which resolves each node of the mesh +// once for all three (issue 1954: three compositions resolved the mesh again each, 11.7 s for one node's plans). Each +// answers as it does alone; the first refusal ends it, in that order, with what follows it empty. +func Composing(in Input, node string, mode Mode) (Resolved, Rendered, map[string]catalogue.Generator, error) { + c := newComposer(in) + resolved, err := c.planFor(node) + if err != nil { + return Resolved{}, Rendered{}, nil, err + } + rendered, err := c.render(node, mode) + if err != nil { + return resolved, Rendered{}, nil, err + } + gens, err := c.generators() + if err != nil { + return resolved, rendered, nil, err + } + return resolved, rendered, gens, nil +} + func (c *composer) render(nodeName string, mode Mode) (Rendered, error) { resolved, err := c.planFor(nodeName) if err != nil { diff --git a/go/compose/resolve.go b/go/compose/resolve.go index 356c837..4fd5de1 100644 --- a/go/compose/resolve.go +++ b/go/compose/resolve.go @@ -36,8 +36,46 @@ func Resolve(in Input, node string) (Resolved, error) { // IsUnresolvable reports whether err is a node's own set failing to compose, rather than the mesh being unable to answer. func IsUnresolvable(err error) bool { return unresolvable(err) } -// planFor is the controller's planFor. +// planFor is the controller's planFor, worked out once per node in one composition (issue 1954): each caller has its +// own copy of what a caller may set (a need's fields, as Resolution.Compose sets its Serves), the rest only read. func (c *composer) planFor(nodeName string) (Resolved, error) { + p, ok := c.plans[nodeName] + if !ok { + p.resolved, p.err = c.planForNow(nodeName) + if c.plans == nil { + c.plans = map[string]planOf{} + } + c.plans[nodeName] = p + } + if p.err != nil { + return Resolved{}, p.err + } + return p.resolved.copied(), nil +} + +// planOf is planFor's answer for one node, kept. +type planOf struct { + resolved Resolved + err error +} + +// copied is r with its own needs, words and settings lists: what a caller may change of it. +func (r Resolved) copied() Resolved { + out := r + out.Plan.Needs = slices.Clone(r.Plan.Needs) + out.Foreseen = slices.Clone(r.Foreseen) + out.NotGiven = slices.Clone(r.NotGiven) + if r.Settings != nil { + out.Settings = make(catalogue.SettingsBy, len(r.Settings)) + for m, layers := range r.Settings { + out.Settings[m] = slices.Clone(layers) + } + } + return out +} + +// planForNow is planFor worked out. +func (c *composer) planForNow(nodeName string) (Resolved, error) { resolved, err := c.resolution(nodeName) if err != nil { return Resolved{}, err diff --git a/go/compose/rest.go b/go/compose/rest.go index a5caf70..7ddb928 100644 --- a/go/compose/rest.go +++ b/go/compose/rest.go @@ -3,6 +3,7 @@ package compose import ( "errors" "fmt" + "slices" "sort" "strings" @@ -59,6 +60,22 @@ func TheRestOfTheMesh(in Input, exclude string) (catalogue.World, Passes, []Left } func (c *composer) theRestOfTheMesh(exclude string) (catalogue.World, Passes, []LeftOut, error) { + r, ok := c.rest[exclude] + if !ok { + r.world, r.passes, r.left, r.err = c.theRestOfTheMeshNow(exclude) + r.unhostable = c.unhostable + if c.rest == nil { + c.rest = map[string]restOf{} + } + c.rest[exclude] = r + } + // Each caller has its own copy of what it may set or add to; the maps and the slices within are only read. + c.unhostable = slices.Clone(r.unhostable) + return r.world, r.passes, slices.Clone(r.left), r.err +} + +// theRestOfTheMeshNow is theRestOfTheMesh worked out: every other machine's set resolved, twice. +func (c *composer) theRestOfTheMeshNow(exclude string) (catalogue.World, Passes, []LeftOut, error) { shelf := c.in.Catalogue holdings := append([]catalogue.Held(nil), c.in.Holdings...) nodes := c.in.sorted() diff --git a/go/state_test.go b/go/state_test.go index e8bdc2a..29a3b16 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -283,4 +283,24 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if gets != 3 { t.Fatalf("a kept value was read by key %d times, want once, and the missing one twice", gets) } + // Issue 1954: a state the process alone writes reads its own put and delete back from the copy at once, and the + // watch's changes of that key up to the write are not taken over it; a later change is. + _ = handles["placement"](StateChange{Key: "own", Op: OpPut, Value: []byte("old"), Revision: 5}) + if err := w.Wrote("placement", "own", []byte("mine"), 7); err != nil { + t.Fatal(err) + } + _ = w.Wrote("placement", "gone", []byte("x"), 8) + _ = w.Removed("placement", "gone") + for _, c := range []StateChange{{Key: "own", Op: OpPut, Value: []byte("old"), Revision: 6}, + {Key: "own", Op: OpPut, Value: []byte("mine"), Revision: 7}, {Key: "gone", Op: OpPut, Value: []byte("x"), Revision: 8}} { + _ = handles["placement"](c) + } + own, _, _ := w.Get("placement", "own") + if _, found, _ := w.Get("placement", "gone"); found || string(own) != "mine" { + t.Fatalf("after its own writes and the watch's older changes the copy holds own=%s, gone found %v", own, found) + } + _ = handles["placement"](StateChange{Key: "gone", Op: OpPut, Value: []byte("again"), Revision: 9}) + if v, _, _ := w.Get("placement", "gone"); string(v) != "again" { + t.Fatalf("a change after its own delete is not taken: %s", v) + } } diff --git a/go/watched.go b/go/watched.go index 1f7953e..ed8b666 100644 --- a/go/watched.go +++ b/go/watched.go @@ -31,8 +31,12 @@ type watchedState struct { ready chan struct{} // closed once the watch's current values are all in the copy, or it was refused err error - mu sync.Mutex - values map[string][]byte + mu sync.Mutex + values map[string][]byte + // revs is, by key, the revision its value in the copy is at, and latest the highest revision the copy has seen: a + // change the watch hands at a revision not newer than its key's is one the copy already took (Wrote, Removed). + revs map[string]uint64 + latest uint64 then []func(StateChange) error started bool // the current values are in, so a change runs then at once pending []StateChange // the current values, run through then once all are in @@ -77,7 +81,7 @@ func (w *WatchedStates) watched(state string, then func(StateChange) error) (*wa } return s, nil } - s = &watchedState{ready: make(chan struct{}), values: map[string][]byte{}} + s = &watchedState{ready: make(chan struct{}), values: map[string][]byte{}, revs: map[string]uint64{}} if then != nil { s.then = []func(StateChange) error{then} } @@ -85,11 +89,17 @@ func (w *WatchedStates) watched(state string, then func(StateChange) error) (*wa w.mu.Unlock() err := w.watch(state, func(c StateChange) error { s.mu.Lock() - if c.Op == OpPut { + switch { + case c.Revision != 0 && c.Revision <= s.revs[c.Key]: + // Already in the copy, or older than what is: this process wrote the key since (Wrote, Removed). + case c.Op == OpPut: s.values[c.Key] = append([]byte(nil), c.Value...) - } else { + s.revs[c.Key] = c.Revision + default: delete(s.values, c.Key) + s.revs[c.Key] = c.Revision } + s.latest = max(s.latest, c.Revision) if !s.started { s.pending = append(s.pending, c) s.mu.Unlock() @@ -240,5 +250,44 @@ func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) { return v, true, nil } +// Wrote takes a put this process made into the copy at once, at the revision the bus answered (0 when not known): a +// state the process alone writes (a holder's own) is read back from the copy right after its own write, before the +// watch hands the change back, and the watch's older changes of the key are not taken over it (issue 1954). The state +// is watched first when it is not yet. +func (w *WatchedStates) Wrote(state, key string, value []byte, revision uint64) error { + return w.took(state, key, value, revision, true) +} + +// Removed takes a delete this process made into the copy at once, as Wrote takes a put. +func (w *WatchedStates) Removed(state, key string) error { return w.took(state, key, nil, 0, false) } + +func (w *WatchedStates) took(state, key string, value []byte, revision uint64, put bool) error { + s, err := w.watched(state, nil) + if err != nil { + return err + } + s.mu.Lock() + defer s.mu.Unlock() + switch { + case put && revision != 0: + if revision <= s.revs[key] { + return nil // the watch handed a newer change of the key already + } + s.values[key] = append([]byte(nil), value...) + s.revs[key] = revision + s.latest = max(s.latest, revision) + default: + // A delete's revision is not answered, nor a put's at times: it is newer than every revision the copy has seen, + // so the watch's changes of the key up to that one are not taken over it. + if put { + s.values[key] = append([]byte(nil), value...) + } else { + delete(s.values, key) + } + s.revs[key] = s.latest + } + return nil +} + // Direct is a key read by key now, not from the copy: a value that changes and is read one at a time (a build). func (w *WatchedStates) Direct(state, key string) ([]byte, bool, error) { return w.get(state, key) }