Merge branch 'feat/compose-once' of /home/jochen/projects/novox/mesh-sdk-compose into feat/node-packages
mesh/delivery merged: source.main moved to novox/mesh-sdk@ea8c50136cc7
mesh/delivery-group group feat/node-packages merged: every member is merged
mesh/check pass b2a27ed5cec4 on 621984614dc6 | unit 8060 run, 706 kept, 51 skipped | integration 8198 run, 749 kept, 56 skipped | over bound: unit, in…
mesh/delivery merged: source.main moved to novox/mesh-sdk@ea8c50136cc7
mesh/delivery-group group feat/node-packages merged: every member is merged
mesh/check pass b2a27ed5cec4 on 621984614dc6 | unit 8060 run, 706 kept, 51 skipped | integration 8198 run, 749 kept, 56 skipped | over bound: unit, in…
This commit is contained in:
+110
-39
@@ -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
|
// 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.
|
// 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) {
|
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)
|
placements, err := r.Placements(ctx)
|
||||||
if err != nil {
|
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 {
|
if len(s.Placements) > 0 {
|
||||||
given := make(map[string]delivery.Placement, len(placements)+len(s.Placements))
|
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
|
continue
|
||||||
}
|
}
|
||||||
if p.Node != node {
|
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
|
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)
|
records, err := r.Records(ctx)
|
||||||
if err != nil {
|
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)
|
modules, err := r.Modules(ctx)
|
||||||
if err != nil {
|
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 {
|
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,
|
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)
|
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) {
|
named := func(field *map[string]string, module, why string) {
|
||||||
if *field == nil {
|
if *field == nil {
|
||||||
*field = map[string]string{}
|
*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)
|
raw, err := r.Manifest(ctx, m.Manifest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
// Not read is no refusal of the module: a bus that does not answer is waited out by the caller.
|
// 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
|
// 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
|
// 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)
|
man, why := taken(name, raw, m)
|
||||||
if why != "" {
|
if why != "" {
|
||||||
named(&in.Refused, name, why)
|
named(&in.Refused, name, why)
|
||||||
if refuse != nil && ownPlaced[name] {
|
if ownPlaced[name] {
|
||||||
refuse(name, errors.New(why))
|
b.ownRefused = append(b.ownRefused, [2]string{name, why})
|
||||||
}
|
}
|
||||||
} else if aside := man.SetAside(); aside != "" {
|
} else if aside := man.SetAside(); aside != "" {
|
||||||
named(&in.SetAside, name, 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() {
|
for _, name := range delivery.ProvidedModules() {
|
||||||
raw, err := json.Marshal(overlay.Manifest())
|
raw, err := json.Marshal(overlay.Manifest())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return compose.Input{}, err
|
return b, err
|
||||||
}
|
}
|
||||||
var private catalogue.Manifest
|
var private catalogue.Manifest
|
||||||
if err := json.Unmarshal(raw, &private); err != nil {
|
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
|
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
|
var values map[string]any
|
||||||
if withValues {
|
if withValues {
|
||||||
if values, err = r.Values(ctx, node, m.Module); err != nil {
|
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 {
|
} 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
|
// 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).
|
// 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 {
|
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)
|
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)
|
in.CarriedPeers = carriedPeers(in.Adopted.Tunnel, placements[in.Adopted.Hub].Decided.PeerNames, records)
|
||||||
}
|
}
|
||||||
if len(missing) > 0 {
|
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, ", "))
|
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 !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)
|
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
|
return compose.Input{}, err
|
||||||
}
|
}
|
||||||
// Its own modules' secrets, sealed to its own key by the vault: a composition carries them sealed (ADR 0550 §5).
|
// 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.
|
// 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) {
|
func plansIn(in compose.Input, node string, mode compose.Mode) (map[string]catalogue.Plan, error) {
|
||||||
resolved, err := compose.Resolve(in, node)
|
// One composition for the three, so each node of the mesh is resolved once (issue 1954).
|
||||||
if err != nil {
|
resolved, rendered, gens, err := compose.Composing(in, node, mode)
|
||||||
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)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("frombus: the plans of %s: %w", node, err)
|
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
|
// 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
|
// by the one take-in rule (takein.Manifest), a templates archive read in the artifact store as the node reaches it
|
||||||
// (issue 1126).
|
// (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),
|
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 {
|
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
|
// 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 == "" {
|
if b.Manifest == "" {
|
||||||
return catalogue.Manifest{}, fmt.Errorf("frombus: the build of %s at %s names no manifest", module, fp)
|
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)
|
src, err := read(module, b.Manifest)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return catalogue.Manifest{}, err
|
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 {
|
if resolved, err = parse(raw); err != nil {
|
||||||
return catalogue.Manifest{}, fmt.Errorf("frombus: the built manifest of %s at %s: %w", module, fp, err)
|
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
|
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
|
// 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()) {
|
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
|
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 {
|
if err != nil {
|
||||||
err = fmt.Errorf("frombus: the build of %s at %s: %w", pm.Module, m.FP2, err)
|
err = fmt.Errorf("frombus: the build of %s at %s: %w", pm.Module, m.FP2, err)
|
||||||
if refuse == nil {
|
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) {
|
for _, module := range sortedKeys(keptAt) {
|
||||||
fp := keptAt[module]
|
fp := keptAt[module]
|
||||||
b, err := r.Build(ctx, module, fp)
|
b, err := build(ctx, module, fp)
|
||||||
switch {
|
switch {
|
||||||
case err != nil:
|
case err != nil:
|
||||||
err = fmt.Errorf("frombus: the build of %s at %s: %w", module, fp, err)
|
err = fmt.Errorf("frombus: the build of %s at %s: %w", module, fp, err)
|
||||||
|
|||||||
@@ -2,10 +2,12 @@ package frombustest
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"reflect"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
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/frombus"
|
||||||
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
|
"git.novox.be/novox/mesh-sdk/go/compose/fromruntime"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"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" {
|
restic.Provisions[0].Provider != Node || restic.Provisions[0].ProviderModule != "postgres" {
|
||||||
t.Fatalf("%s: the plans are %+v (held %v)", row.name, plans, held)
|
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 row.bad == "" || row.elsewhere {
|
||||||
if len(held) > 0 {
|
if len(held) > 0 {
|
||||||
t.Errorf("%s: held %v", row.name, held)
|
t.Errorf("%s: held %v", row.name, held)
|
||||||
@@ -167,6 +181,85 @@ func TestTheProvisionedMeshHandsAPlanWithItsProvision(t *testing.T) {
|
|||||||
if fetched != 1 {
|
if fetched != 1 {
|
||||||
t.Errorf("a templates archive was fetched %d times, want once", fetched)
|
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 }
|
func planned[P any](plans map[string]P, module string) bool { _, ok := plans[module]; return ok }
|
||||||
|
|||||||
@@ -43,13 +43,22 @@ type States interface {
|
|||||||
type Fetch func(ctx context.Context, url string) ([]byte, error)
|
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.
|
// 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 {
|
type statesReader struct {
|
||||||
s States
|
s States
|
||||||
fetch Fetch
|
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 {
|
func (r statesReader) each(read string, visit func(key string, value []byte)) error {
|
||||||
keys, err := r.s.Keys(read)
|
keys, err := r.s.Keys(read)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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
|
// 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).
|
// composition finds that node cannot run: what it would have offered is missing here (issue 1126, probe #532).
|
||||||
unhostable []string
|
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{}} }
|
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
|
// 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.
|
// counted, and said.
|
||||||
func (c *composer) onTheNetwork() []Node {
|
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
|
var out []Node
|
||||||
for _, n := range c.in.sorted() {
|
for _, n := range c.in.sorted() {
|
||||||
if n.Place.Address == "" {
|
if n.Place.Address == "" {
|
||||||
|
|||||||
@@ -35,6 +35,26 @@ func Render(in Input, node string, mode Mode) (Rendered, error) {
|
|||||||
return newComposer(in).render(node, mode)
|
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) {
|
func (c *composer) render(nodeName string, mode Mode) (Rendered, error) {
|
||||||
resolved, err := c.planFor(nodeName)
|
resolved, err := c.planFor(nodeName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
+39
-1
@@ -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.
|
// 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) }
|
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) {
|
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)
|
resolved, err := c.resolution(nodeName)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Resolved{}, err
|
return Resolved{}, err
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package compose
|
|||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"slices"
|
||||||
"sort"
|
"sort"
|
||||||
"strings"
|
"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) {
|
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
|
shelf := c.in.Catalogue
|
||||||
holdings := append([]catalogue.Held(nil), c.in.Holdings...)
|
holdings := append([]catalogue.Held(nil), c.in.Holdings...)
|
||||||
nodes := c.in.sorted()
|
nodes := c.in.sorted()
|
||||||
|
|||||||
@@ -283,4 +283,24 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) {
|
|||||||
if gets != 3 {
|
if gets != 3 {
|
||||||
t.Fatalf("a kept value was read by key %d times, want once, and the missing one twice", gets)
|
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)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+52
-3
@@ -33,6 +33,10 @@ type watchedState struct {
|
|||||||
|
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
values map[string][]byte
|
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
|
then []func(StateChange) error
|
||||||
started bool // the current values are in, so a change runs then at once
|
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
|
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
|
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 {
|
if then != nil {
|
||||||
s.then = []func(StateChange) error{then}
|
s.then = []func(StateChange) error{then}
|
||||||
}
|
}
|
||||||
@@ -85,11 +89,17 @@ func (w *WatchedStates) watched(state string, then func(StateChange) error) (*wa
|
|||||||
w.mu.Unlock()
|
w.mu.Unlock()
|
||||||
err := w.watch(state, func(c StateChange) error {
|
err := w.watch(state, func(c StateChange) error {
|
||||||
s.mu.Lock()
|
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...)
|
s.values[c.Key] = append([]byte(nil), c.Value...)
|
||||||
} else {
|
s.revs[c.Key] = c.Revision
|
||||||
|
default:
|
||||||
delete(s.values, c.Key)
|
delete(s.values, c.Key)
|
||||||
|
s.revs[c.Key] = c.Revision
|
||||||
}
|
}
|
||||||
|
s.latest = max(s.latest, c.Revision)
|
||||||
if !s.started {
|
if !s.started {
|
||||||
s.pending = append(s.pending, c)
|
s.pending = append(s.pending, c)
|
||||||
s.mu.Unlock()
|
s.mu.Unlock()
|
||||||
@@ -240,5 +250,44 @@ func (w *WatchedStates) Once(state, key string) ([]byte, bool, error) {
|
|||||||
return v, true, nil
|
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).
|
// 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) }
|
func (w *WatchedStates) Direct(state, key string) ([]byte, bool, error) { return w.get(state, key) }
|
||||||
|
|||||||
Reference in New Issue
Block a user