package selfcompose import ( "context" "encoding/json" "errors" "fmt" "runtime" "strings" "sync" "sync/atomic" "testing" "time" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/buckets/bucketstest" "github.com/novox/mesh-host/internal/compositions" ) // Design 47 G §421, §683, §698 (the lead's answers A-47, A-50): the node-engine reads the entry its want names by the // fingerprint, through its own calls of the seat mesh-catalog's verb entry, and reads neither the indexer's states nor the // builder's; a want whose entry the catalog half holds none of is held, its have saying entry-missing and naming the // fingerprint; the entries and wants it composes from are kept in its composition state, written before the composition that // uses them and read at its start before anything is asked; a want removed takes its kept want and entry away. func composeFromEntries(t *testing.T) { b := newFakeBus() ctx, cancel := context.WithCancel(context.Background()) defer cancel() fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) ghost := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) var manifest map[string]any if err := json.Unmarshal([]byte(lemurs), &manifest); err != nil { t.Fatal(err) } entry := meshcatalog.CatalogEntry{Module: "lemurs", Fp: fp, Manifest: manifest, Repository: "novox/x", Commit: "c", ManifestDigest: meshcatalogclient.ManifestDigest([]byte(lemurs))} var mu sync.Mutex var asked []string ask := func(_ context.Context, seat, verb string, args any) (json.RawMessage, error) { in := args.(meshcatalog.EntryInput) mu.Lock() asked = append(asked, in.Fingerprint) mu.Unlock() if seat != "mesh-catalog" || verb != "entry" { t.Errorf("asked %s.%s", seat, verb) } if in.Fingerprint != fp { return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept") } return json.Marshal(meshcatalog.EntryAnswer{Entry: entry}) } state := &fakeComposed{kv: map[string]ComposedEntry{}} var held map[string]string env := b.env() env.Entry, env.ComposedState = EntryCall(ask), state refused := make(chan struct{}, 1) // fed at each refusal the composer says, so a wait follows what it says, not a pause env.Refused = func(_ string, h map[string]string) { mu.Lock() held = h mu.Unlock() select { case refused <- struct{}{}: default: } } told := map[string]bool{} env.EntryNotFound = func(fp string, notFound bool) { mu.Lock(); told[fp] = notFound; mu.Unlock() } c := New(env) out, err := c.Watch(ctx, "n1.>") if err != nil { t.Fatal(err) } place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "ghost"}}}) have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true}) b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) b.put(t, buckets.Wants, "want.n1.node", nodeWant) for module, w := range map[string]meshwants.WantsByWantNodeModuleValue{"lemurs": {Fp: fp, Self: true}, "ghost": {Fp: ghost, Self: true}} { raw, _ := bucketstest.Want(w) b.put(t, buckets.Wants, "want.n1."+module, raw) } for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } lemursKey, _ := compositions.Key("n1", "lemurs", fp) handed := map[string]bool{} for !handed[lemursKey] { handed[next(t, out).Key] = true } // Neither the indexer's states nor the builder's are watched or read. b.mu.Lock() for _, r := range b.reads { if r[0] == buckets.Modules || r[0] == buckets.Manifests || r[0] == buckets.Verdicts || strings.HasSuffix(r[0], "_builds") { t.Errorf("the node-engine read %s of %s", r[1], r[0]) } } b.mu.Unlock() // A want whose entry the half holds none of is held, saying entry-missing and the fingerprint; the rest is composed. // The ghost's entry is asked first and answered not found after: wait until the refusal says entry-missing. var why string for deadline := time.After(10 * time.Second); !strings.HasPrefix(why, agent.EntryMissing); { mu.Lock() why = held["ghost"] mu.Unlock() if strings.HasPrefix(why, agent.EntryMissing) { break } select { case <-refused: case <-deadline: t.Fatalf("the ghost's reason is still %q after 10 s, want %s", why, agent.EntryMissing) } } if !strings.HasPrefix(why, agent.EntryMissing) || !strings.Contains(why, ghost) || held["lemurs"] != "" { t.Errorf("held %q, want ghost held with %s naming %s", held, agent.EntryMissing, ghost) } // And the finding entry-not-found is told for the fingerprint the half holds none of, and for no other (design 47 G §614, A-117). mu.Lock() nf, nfOther := told[ghost], told[fp] mu.Unlock() if !nf || nfOther { t.Errorf("entry-not-found told %v for the ghost and %v for the entry held", nf, nfOther) } // An entry asked again by hand is asked again in the shadow's cache too (lead answer A-319): the shadow holds the ghost as not // found, the half then holds it, and the node's own composer asked again takes it in on the shadow as well. var late atomic.Bool shadowEnv := b.env() shadowEnv.Entry = EntryCall(func(ctx context.Context, seat, verb string, args any) (json.RawMessage, error) { if in := args.(meshcatalog.EntryInput); in.Fingerprint == ghost && late.Load() { return json.Marshal(meshcatalog.EntryAnswer{Entry: meshcatalog.CatalogEntry{Module: "ghost", Fp: ghost}}) } return ask(ctx, seat, verb, args) }) shadow := NewShadow(shadowEnv, nil) shadow.c.entries.Need(ctx, ghost) for deadline := time.Now().Add(10 * time.Second); shadow.c.entries.Rows() == nil || shadow.c.entries.Rows()[0].State != meshcatalogclient.EntryNotFound; { if time.Now().After(deadline) { t.Fatal("the shadow did not hold the ghost as not found") } runtime.Gosched() } late.Store(true) c.AlsoEntryAgain = shadow.EntryAgain if err := c.EntryAgain(ctx, ghost); err != nil { t.Fatal(err) } if _, state := shadow.c.entries.Need(ctx, ghost); state != meshcatalogclient.EntryAnswered { t.Errorf("the shadow holds the ghost's entry as %s after entry-again, want answered", state) } // Each entry and want it composes from is kept in the composition state before the composition that uses them. state.mu.Lock() _, hasEntry := state.kv[entryKey(fp)] _, hasWant := state.kv["want.lemurs"] state.mu.Unlock() if !hasEntry || !hasWant { t.Errorf("the composition state holds entry %v and want %v of lemurs: %v", hasEntry, hasWant, state.written()) } // The seat's first-views lists them. _, entries, wants := c.FirstViews() if len(entries) < 2 || len(wants) != 2 { t.Errorf("first-views lists %v and %v", entries, wants) } // A restart reads what is kept before it asks anything: lemurs's entry is not asked again. mu.Lock() asked = nil mu.Unlock() env2 := b.env() env2.Entry, env2.ComposedState = EntryCall(ask), state // The node's own record, stated by the node-engine of itself in memory (A-120). env2.NodeRecord = func() (nodeengine.NodeRecord, bool) { return nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}, true } c2 := New(env2) out2, err := c2.Watch(ctx, "n1.>") if err != nil { t.Fatal(err) } // And composes the module it kept a want and an entry of before any input is read, nothing of the others (design 47 G §683, A-117). if u := next(t, out2); u.Key != lemursKey || u.Deleted { t.Errorf("the early pass handed %q, want the composition %s of lemurs", u.Key, lemursKey) } c2.mu.Lock() kept, keptWant := c2.keptEnt[fp], c2.wants["lemurs"].Fp c2.mu.Unlock() if kept.entry.Module != "lemurs" || keptWant != fp { t.Errorf("a restart read entry %q and want %q from the state", kept.entry.Module, keptWant) } // A want removed takes its kept want and entry away. raw, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Removed: true}) b.put(t, buckets.Wants, "want.n1.lemurs", raw) for range 20 { state.mu.Lock() _, hasWant = state.kv["want.lemurs"] _, hasEntry = state.kv[entryKey(fp)] state.mu.Unlock() if !hasWant && !hasEntry { break } next(t, out) } if hasWant || hasEntry { t.Errorf("a removed want left its kept want %v and entry %v", hasWant, hasEntry) } } // catalogHalf is the catalog half's entries a test composes from, by fingerprint, answered through the seat mesh-catalog's verbs // entry and module (design 47 G §421, A-312): the node-engine reads no indexer. type catalogHalf map[string]meshcatalog.CatalogEntry // add holds the entry of module at fp, made of the manifest body. func (h catalogHalf) add(t *testing.T, module, fp, body string) { t.Helper() var m map[string]any if err := json.Unmarshal([]byte(body), &m); err != nil { t.Fatal(err) } h[fp] = meshcatalog.CatalogEntry{Module: module, Fp: fp, Manifest: m, Repository: "novox/" + module, Commit: "c", ManifestDigest: meshcatalogclient.ManifestDigest([]byte(body))} } func (h catalogHalf) call() meshcatalogclient.Call { return EntryCall(func(_ context.Context, _, verb string, args any) (json.RawMessage, error) { switch in := args.(type) { case meshcatalog.ModuleInput: for _, e := range h { if e.Module == in.Module { return json.Marshal(meshcatalog.ModuleAnswer{Module: meshcatalog.ModuleAnswerModule{Module: e.Module, Repository: e.Repository, Commit: e.Commit, Fp: e.Fp}}) } } return nil, fmt.Errorf("the catalog holds no module %s", in.Module) case meshcatalog.EntryInput: if e, ok := h[in.Fingerprint]; ok { return json.Marshal(meshcatalog.EntryAnswer{Entry: e}) } return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept") } return nil, fmt.Errorf("asked %s of %T", verb, args) }) } // holdEntry makes the composer hold the entry of module at fp, as its call of the verb entry would have answered it (§421), for the // tests that set the rest of its picture directly. func (c *Composer) holdEntry(t *testing.T, module, fp, body string) { t.Helper() half := catalogHalf{} half.add(t, module, fp, body) c.mu.Lock() defer c.mu.Unlock() if c.keptEnt == nil { c.keptEnt = map[string]keptEntry{} } c.keptEnt[fp] = keptEntry{entry: half[fp]} }