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) }