diff --git a/go.mod b/go.mod index ce16253..38b5ffd 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101005736-7b09de1e5502 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index 94a9c74..b03249d 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a h1:DcSvfEmd git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031084708-acbf80ad910a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406 h1:Z4RcH5Ft4OAAG3F538NXIHP9F3pkyk3j0LtP9mqu1qQ= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261031230538-c28c645ca406/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101005736-7b09de1e5502 h1:XndvkzyAFNde/821lEfYObqMTURzmve/w9Ks+QrBBJ8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261101005736-7b09de1e5502/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/selfcompose/data.go b/internal/selfcompose/data.go index b21bcdb..bdcb148 100644 --- a/internal/selfcompose/data.go +++ b/internal/selfcompose/data.go @@ -39,8 +39,9 @@ func (itemsReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, return nil, nil } -// dataItemsOf works out the data items of the scope's node (frombus.DataItems); a test sets what it answers. -var dataItemsOf = frombus.DataItems +// dataItemsOf works out the data items of the scope's node and, by module, why each module held for a manifest the +// indexer refused is left out (frombus.DataItemsHeld, ADR 0578 §3); a test sets what it answers. +var dataItemsOf = frombus.DataItemsHeld // publishData keeps the data items of this node's modules in the engine's state data, when a pass asked it (dataDue). func (c *Composer) publishData(ctx context.Context) { @@ -57,19 +58,29 @@ func (c *Composer) publishData(ctx context.Context) { if scope.Node == "" { scope.Node = c.env.Node } - items, unplaced, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) + items, unplaced, held, err := dataItemsOf(ctx, itemsReader{reader{c: c}}, c.env.Mesh, scope) if err != nil { c.sayData("the data items of this node's modules cannot be worked out, and what the engine's state " + delivery.EngineDataBucket + " holds stands; " + c.oweData() + ": " + err.Error()) return } + // A module held for a refused manifest declares nothing this pass can read: it is named, its other modules' items + // are kept as ever, and its own key stands as it was until the manifest is accepted (ADR 0578 §3). + if len(held) > 0 { + var lines []string + for _, m := range heldModules(held) { + lines = append(lines, m+": "+held[m]) + } + c.env.Say("data items of " + strings.Join(heldModules(held), ", ") + " not worked out, as the manifest is refused " + + "and the module held (its key stands as it was): " + strings.Join(lines, "; ")) + } if len(unplaced) > 0 { c.env.Say("data items left out as not placeable: " + strings.Join(unplaced, "; ")) } c.mu.Lock() digests := c.digestsByModule() c.mu.Unlock() - wrote, removed, err := c.writeData(ctx, items, digests) + wrote, removed, err := c.writeData(ctx, items, digests, held) if err != nil { c.sayData("the data items of this node's modules are not all kept in " + delivery.EngineDataBucket + "; " + c.oweData() + ": " + err.Error()) @@ -130,7 +141,7 @@ func (c *Composer) digestsByModule() map[string]string { // writeData writes each module's key whose value differs from what the key holds, and deletes this node's keys of // modules with no items here now. The keys are read once per run (dataKept), so a restart that finds the same items // writes nothing. -func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string) (wrote, removed int, err error) { +func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.PlacedDataItem, digests map[string]string, held map[string]string) (wrote, removed int, err error) { store := c.env.Data if c.dataKept == nil { pattern, err := delivery.EngineDataNodeKeys(c.env.Node) @@ -198,7 +209,7 @@ func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.P } var gone []string for key := range c.dataKept { - if _, still := want[key]; !still { + if _, still := want[key]; !still && !heldKey(c.env.Node, key, held) { gone = append(gone, key) } } @@ -216,3 +227,13 @@ func (c *Composer) writeData(ctx context.Context, items map[string][]catalogue.P } return wrote, removed, nil } + +// heldKey is whether key is the key of a module held for a refused manifest, which stands as it was. +func heldKey(node, key string, held map[string]string) bool { + for module := range held { + if k, err := delivery.EngineDataKey(node, module); err == nil && k == key { + return true + } + } + return false +} diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index 796fc74..f613bd5 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -77,15 +77,18 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { var items map[string][]catalogue.PlacedDataItem var asked int var itemsErr error - setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, error) { + var heldNow map[string]string + setFor(t, &dataItemsOf, func(context.Context, frombus.Reader, frombus.Mesh, frombus.Scope) (map[string][]catalogue.PlacedDataItem, []string, map[string]string, error) { asked++ - return items, nil, itemsErr + return items, nil, heldNow, itemsErr }) + var said []string + env := Env{Node: "n1", Data: store, Say: func(s string) { said = append(said, s) }} // The engine's grant lets it read its own node's keys, as the listing at a run's first pass does, and no other's. if !grantAllows(t, "n1", delivery.EngineDataBucket, "n1.*") || grantAllows(t, "n1", delivery.EngineDataBucket, "n2.*") { t.Error("the node's grant reads other keys of the engine's state data than its own") } - c := New(Env{Node: "n1", Data: store}) + c := New(env) webKey, _ := delivery.ComposedKey("n1", "web", delivery.SourcePrefix2+strings.Repeat("a", 64)) nodeKey, _ := delivery.ComposedNodeKey("n1") for _, step := range []struct { @@ -94,8 +97,9 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { items map[string][]catalogue.PlacedDataItem digests map[string]string // composed key to digest handed fail error - refused error // what working out the items says - owed time.Duration // the bound the loop waits before trying again; 0 when nothing is owed + refused error // what working out the items says + held map[string]string // modules held for a refused manifest, with why (ADR 0578 §3) + owed time.Duration // the bound the loop waits before trying again; 0 when nothing is owed puts []string deletes []string asked int @@ -126,12 +130,19 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { {name: "a module that left the node is deleted", due: true, items: map[string][]catalogue.PlacedDataItem{"oldie": {item("oldie", "x", "/srv/oldie")}}, digests: map[string]string{nodeKey: "sha256:n1"}, deletes: []string{"n1.web"}, asked: 6}, + // ADR 0578 §3: one refused manifest never ends the read; its module is named, its key stands, the rest is kept. + {name: "a module held for a refused manifest is named, its key stands and every other module's items are written", due: true, + items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}}, + held: map[string]string{"oldie": "the manifest of oldie is refused: a bar item is no icon"}, + digests: map[string]string{webKey: "sha256:w3", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 7, + want: map[string]delivery.EngineData{ + "n1.web": {Module: "web", Digest: "sha256:w3", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}}}}, } { t.Run(step.name, func(t *testing.T) { store.mu.Lock() store.puts, store.deletes, store.fail = nil, nil, step.fail store.mu.Unlock() - items, itemsErr = step.items, step.refused + items, itemsErr, heldNow, said = step.items, step.refused, step.held, nil c.paid() // the pass before composed: nothing owed c.mu.Lock() c.handed = step.digests @@ -153,6 +164,14 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { t.Errorf("%s holds %+v (%v), want %+v", key, got, err, want) } } + for module, why := range step.held { + if !slices.ContainsFunc(said, func(l string) bool { return strings.Contains(l, module) && strings.Contains(l, why) }) { + t.Errorf("the held %s is not named with why in %q", module, said) + } + if store.kv["n1."+module] == nil { + t.Errorf("the key of the held %s was deleted", module) + } + } if store.kv["n2.web"] == nil { t.Error("another node's key was deleted") } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 61fb4f7..cb6f1ea 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -110,6 +110,7 @@ type Composer struct { // at a build of the old form): the node runs it inside its own composition, and composes it there again (issue 1126). inOwn map[string]bool modules map[string]delivery.Module + told map[string]bool // lines said once each (sayOnceEach) values map[string]map[string]any // by module, this node's manifests map[string][]byte // by digest, read once archives map[string][]byte // templates archives by digest, read once @@ -1052,8 +1053,8 @@ func (c *Composer) Policy(ctx context.Context, module string) string { if err != nil { return "" } - parsed, err := catalogue.ParseManifestForInputs(raw) - if err != nil { + parsed, held, err := c.taken(module, m, raw) + if err != nil || held != "" { return "" } policy, _, _ := catalogue.DerivedUpgrade(parsed) @@ -1074,13 +1075,51 @@ func (c *Composer) HoldsData(ctx context.Context, module string) (data, known bo if err != nil { return false, false } - parsed, err := catalogue.ParseManifestForInputs(raw) - if err != nil { + parsed, held, err := c.taken(module, m, raw) + if err != nil || held != "" { return false, false } return len(parsed.DataItems()) > 0, true } +// taken is a module's manifest as this engine reads it (novox/hq ADR 0578 §3): decoded leniently, judging nothing, with +// the indexer's verdict from the module's value (a missing one is accepted). held is why the module is held, "" when it is +// not: a manifest the indexer refused, or one not read past a field this engine does not know. A field set aside is said +// once, and an error is returned only for bytes that are no manifest, which the caller says in its own words. +func (c *Composer) taken(module string, m delivery.Module, raw []byte) (man catalogue.Manifest, held string, err error) { + if v := m.VerdictOf(module); v != nil { + if refused, _ := delivery.TakeManifestVerdict(v); refused { + return catalogue.Manifest{}, v.Why(), nil + } + } + man, err = catalogue.Decode(raw) + if err != nil { + c.sayOnceEach("the manifest of " + module + " does not decode, and the module is held: " + err.Error()) + return man, "", err + } + if aside := man.SetAside(); aside != "" { + c.sayOnceEach("the manifest of " + module + " has a field this engine does not know, set aside: " + aside) + if man.ReadNoFurther() { + return man, aside, nil + } + } + return man, "", nil +} + +// sayOnceEach says a line the first time only, apart from sayOnce's single current line. +func (c *Composer) sayOnceEach(line string) { + c.mu.Lock() + if c.told == nil { + c.told = map[string]bool{} + } + again := c.told[line] + c.told[line] = true + c.mu.Unlock() + if !again { + c.env.Say(line) + } +} + // reader is the composer's view of the bus as frombus reads it. type reader struct{ c *Composer } @@ -1109,7 +1148,14 @@ func (c *Composer) certificateWanted(ctx context.Context, r reader, p delivery.P if err != nil { continue } - if parsed, err := catalogue.ParseManifestForInputs(raw); err == nil && parsed.Certificate != nil { + c.mu.Lock() + mod := c.modules[m.Module] + c.mu.Unlock() + parsed, held, err := c.taken(m.Module, mod, raw) + if err != nil || held != "" { + continue // held or not read: it asks none, and its refusal is named by the composition + } + if parsed.Certificate != nil { return true } } diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index c1413ba..fa5683c 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -450,7 +450,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.kv[delivery.ManifestsBucket+"/"+powerKey] = powerEnc b.mu.Unlock() powerFP := delivery.SourcePrefix2 + strings.Repeat("c", 64) - c4.modules["power"] = delivery.Module{FP2: powerFP, Manifest: powerDig, Repository: "novox/mesh-catalog", Commit: "ca8d6f33"} + c4.modules["power"] = delivery.Module{FP2: powerFP, Manifest: powerDig, Repository: "novox/mesh-catalog", Commit: "ca8d6f33", + // The indexer's verdict on the captured manifest (ADR 0578): the engine judges nothing, it takes this. + Verdict: delivery.VerdictRefused, Problems: []string{`bar item "shows" is none of them`}} c4.wants["power"] = delivery.Want{FP: powerFP, Self: true} p4 := c4.placements["n1"] p4.Modules = append(append([]delivery.PlacedModule{}, p4.Modules...), delivery.PlacedModule{Module: "power"})