jschoubben
af66f72931
Issue 1585, S17: the engine decodes manifests leniently; a module whose manifest the indexer refused is held and named, its data key stands, and every other module's items are still published. Policy, HoldsData and the certificate check no longer parse with the judge. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
215 lines
10 KiB
Go
215 lines
10 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
)
|
|
|
|
// fakeData is the engine's state data as a map, counting the writes and deletes made.
|
|
type fakeData struct {
|
|
mu sync.Mutex
|
|
kv map[string][]byte
|
|
puts []string
|
|
deletes []string
|
|
fail error
|
|
}
|
|
|
|
func (f *fakeData) Get(_ context.Context, key string) ([]byte, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
return f.kv[key], nil
|
|
}
|
|
|
|
func (f *fakeData) Put(_ context.Context, key string, value []byte) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
if f.fail != nil {
|
|
return f.fail
|
|
}
|
|
f.kv[key] = value
|
|
f.puts = append(f.puts, key)
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeData) Delete(_ context.Context, key string) error {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
delete(f.kv, key)
|
|
f.deletes = append(f.deletes, key)
|
|
return nil
|
|
}
|
|
|
|
func (f *fakeData) Keys(_ context.Context, pattern string) ([]string, error) {
|
|
f.mu.Lock()
|
|
defer f.mu.Unlock()
|
|
var out []string
|
|
for k := range f.kv {
|
|
if matches(pattern, k) {
|
|
out = append(out, k)
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// ADR 0577 clause 1 (and its check in clause 5): after a pass that composed something new, the engine keeps each placed
|
|
// module's data items in its state data, key `<node>.<module>`, with its composition's digest: written when the module's
|
|
// items change, not written when they do not, and deleted when the module leaves the node; a key of another node is never
|
|
// touched, a pass that handed nothing new reads nothing, and items that could not be worked out or written are tried
|
|
// again with no new composition, after the composer's growing bound (the loop's owed pass).
|
|
func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) {
|
|
item := func(module, id, path string) catalogue.PlacedDataItem {
|
|
return catalogue.PlacedDataItem{Module: module, ID: id, Class: "valuable", Path: path, Measure: catalogue.MeasureWalk}
|
|
}
|
|
store := &fakeData{kv: map[string][]byte{
|
|
// Left by a module that went while the engine was down, and another node's key.
|
|
"n1.gone": []byte(`{"v":1}`), "n2.web": []byte(`{"v":1}`)}}
|
|
var items map[string][]catalogue.PlacedDataItem
|
|
var asked int
|
|
var itemsErr 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, 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)
|
|
webKey, _ := delivery.ComposedKey("n1", "web", delivery.SourcePrefix2+strings.Repeat("a", 64))
|
|
nodeKey, _ := delivery.ComposedNodeKey("n1")
|
|
for _, step := range []struct {
|
|
name string
|
|
due bool
|
|
items map[string][]catalogue.PlacedDataItem
|
|
digests map[string]string // composed key to digest handed
|
|
fail error
|
|
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
|
|
// want is each key's items and digest after the step.
|
|
want map[string]delivery.EngineData
|
|
}{
|
|
{name: "the run's first pass writes each module's items and deletes what left", due: true,
|
|
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}, "bare": nil},
|
|
digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"},
|
|
puts: []string{"n1.oldie", "n1.web"}, deletes: []string{"n1.gone"}, asked: 1,
|
|
want: map[string]delivery.EngineData{
|
|
"n1.web": {Module: "web", Digest: "sha256:w1", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db")}},
|
|
"n1.oldie": {Module: "oldie", Digest: "sha256:n1", Items: []catalogue.PlacedDataItem{item("oldie", "x", "/srv/oldie")}}}},
|
|
{name: "a pass that handed nothing new reads nothing", due: false, asked: 1},
|
|
{name: "a pass that changes no module's items writes nothing", due: true,
|
|
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
|
|
digests: map[string]string{webKey: "sha256:w1", nodeKey: "sha256:n1"}, asked: 2},
|
|
{name: "items that cannot be worked out are tried again after the bound", due: true, refused: errors.New("no answer"),
|
|
digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 3, owed: LaterWait(0)},
|
|
{name: "a write that fails is tried again with no new composition, after a longer bound", due: false, fail: errors.New("the bus said no"),
|
|
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
|
|
digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, asked: 4, owed: LaterWait(1)},
|
|
{name: "a module whose composition changed its items is written again", due: false,
|
|
items: map[string][]catalogue.PlacedDataItem{"web": {item("web", "db", "/srv/web/db2")}, "oldie": {item("oldie", "x", "/srv/oldie")}},
|
|
digests: map[string]string{webKey: "sha256:w2", nodeKey: "sha256:n1"}, puts: []string{"n1.web"}, asked: 5,
|
|
want: map[string]delivery.EngineData{
|
|
"n1.web": {Module: "web", Digest: "sha256:w2", Items: []catalogue.PlacedDataItem{item("web", "db", "/srv/web/db2")}}}},
|
|
{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, heldNow, said = step.items, step.refused, step.held, nil
|
|
c.paid() // the pass before composed: nothing owed
|
|
c.mu.Lock()
|
|
c.handed = step.digests
|
|
c.dataDue = c.dataDue || step.due
|
|
c.mu.Unlock()
|
|
c.publishData(context.Background())
|
|
slices.Sort(store.puts)
|
|
if !slices.Equal(store.puts, step.puts) || !slices.Equal(store.deletes, step.deletes) || asked != step.asked {
|
|
t.Fatalf("wrote %v and deleted %v, items asked %d times; want %v, %v, %d", store.puts, store.deletes, asked,
|
|
step.puts, step.deletes, step.asked)
|
|
}
|
|
if wait, _ := c.waitNow(); c.owed.Load() != (step.owed > 0) || wait != step.owed {
|
|
t.Errorf("owed a pass: %v after %s; want after %s", c.owed.Load(), wait, step.owed)
|
|
}
|
|
for key, want := range step.want {
|
|
got, err := delivery.DecodeEngineData(store.kv[key])
|
|
if err != nil || got.Node != "n1" || got.Module != want.Module || got.Digest != want.Digest ||
|
|
!slices.Equal(got.Items, want.Items) {
|
|
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")
|
|
}
|
|
})
|
|
}
|
|
// The items are worked out from what the pass read: a done build is read over the bus once, not again for the data
|
|
// items after the pass (a module in a running build is read again, since it may land).
|
|
b := newFakeBus()
|
|
done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte(lemurs))})
|
|
running, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: "building"})
|
|
fp := delivery.SourcePrefix2 + strings.Repeat("b", 64)
|
|
doneKey, _ := delivery.BuilderBuildKey("web", fp)
|
|
runningKey, _ := delivery.BuilderBuildKey("pg", fp)
|
|
b.kv[delivery.BuilderBuildsBucket+"/"+doneKey], b.kv[delivery.BuilderBuildsBucket+"/"+runningKey] = done, running
|
|
r := reader{c: New(b.env())}
|
|
for range 3 {
|
|
if got, err := r.Build(context.Background(), "web", fp); err != nil || got.State != delivery.BuildDone {
|
|
t.Fatalf("the done build: %+v, %v", got, err)
|
|
}
|
|
if got, err := r.Build(context.Background(), "pg", fp); err != nil || got.State != "building" {
|
|
t.Fatalf("the running build: %+v, %v", got, err)
|
|
}
|
|
}
|
|
reads := map[string]int{}
|
|
for _, read := range b.reads {
|
|
reads[read[1]]++
|
|
}
|
|
if reads[doneKey] != 1 || reads[runningKey] != 3 {
|
|
t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads)
|
|
}
|
|
// A done key the builder claims again (a gone step, issue 1937) and lands done again is read again, not served kept.
|
|
again, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: delivery.ManifestDigest([]byte("{}"))})
|
|
b.mu.Lock()
|
|
b.kv[delivery.BuilderBuildsBucket+"/"+doneKey] = again
|
|
b.mu.Unlock()
|
|
r.c.takeBuildDone(agent.Update{Key: doneKey, Value: running})
|
|
if got, err := r.Build(context.Background(), "web", fp); err != nil || got.Manifest != delivery.ManifestDigest([]byte("{}")) {
|
|
t.Errorf("a done build written again is served as %+v (%v), not read again", got, err)
|
|
}
|
|
}
|