diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index d37e5243..165d89f5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -386,6 +386,15 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { } b.compose(t, "shanks", 'a') b.start(t) + // DLV-F5r (ADR 0621 §1-§3, design 25 § 12 on a fresh mesh): the assigner asks nothing of the bus module (its manifest + // has no bus-state entry, the bed runs no bus module and holds no binding of one), and while it still waits on the + // indexer its runtime of its own answers each first-view tool to the readers its manifest's callers lists name, at + // its mark, and refuses every other account, the assigner's own included. + firstViewsBeforeAnyNewFormState(t, b, map[string][]string{ + "placement": {"n1.mesh-grants", "n1.node-tools", "n1.mesh-delivery-indexer"}, + "seat-holders": {"n1.node-tools"}, + "wants": {"n1.mesh-grants"}, + }) close(up) wkey, _ := delivery.WantKey("ace", "web") @@ -1332,3 +1341,127 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { t.Cleanup(func() { _ = sub.Unsubscribe() }) } } + +// firstViewsBeforeAnyNewFormState serves the assigner's first-view tools as its runtime of its own does (the tools bundle's +// viewTools over the bed's buckets, the env its manifest's bundle writes, the caller the runtime proves from the subject's +// last token), before any new-form state exists, and asks each from every reader granted and from the others. +func firstViewsBeforeAnyNewFormState(t *testing.T, b *bed, granted map[string][]string) { + t.Helper() + raw, err := os.ReadFile(filepath.Join("..", "..", "module.json")) + if err != nil { + t.Fatal(err) + } + m, err := catalogue.Decode(raw) + if err != nil { + t.Fatal(err) + } + if len(m.BusState) != 0 { + t.Fatalf("the assigner asks bus state of the bus module: %+v", m.BusState) + } + var manifest struct { + Build struct { + Artifacts []struct { + Kind string `json:"kind"` + Env map[string]string `json:"env"` + } `json:"artifacts"` + } `json:"build"` + } + if err := json.Unmarshal(raw, &manifest); err != nil { + t.Fatal(err) + } + env := map[string]string{} + for _, a := range manifest.Build.Artifacts { + if a.Kind == "bundle" { + env = a.Env + } + } + state := func(name string) views.Source { + if name == views.MarksState { + return bedSource{b.ctx, b.buckets[marksBucket]} + } + return bedSource{b.ctx, b.buckets[viewedStates[name]]} + } + // The runtime: one bundle on a pair of pipes, a call on mesh.mod..call.. handed to it with + // the caller in _meta, its answer replied as the tool runner replies ({result, error, node}). + inR, inW := io.Pipe() + outR, outW := io.Pipe() + t.Cleanup(func() { inW.Close(); outW.Close() }) + go func() { + _ = stdio.ServeOn(delivery.AssignerModule, viewTools(func(k string) string { return env[k] }, state, time.Now), inR, outW) + }() + answers := make(chan map[string]any, 16) + go func() { + dec := json.NewDecoder(outR) + for { + var line map[string]any + if dec.Decode(&line) != nil { + return + } + if _, asked := line["method"]; !asked { + answers <- line + } + } + }() + one := make(chan struct{}, 1) // one call at a time on the pipe, so each answer is its call's + nc, err := nats.Connect(b.srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + sub, err := nc.Subscribe("mesh.mod."+delivery.AssignerModule+".call.*.*", func(msg *nats.Msg) { + one <- struct{}{} + defer func() { <-one }() + tokens := strings.Split(msg.Subject, ".") + var args map[string]any + _ = json.Unmarshal(msg.Data, &args) + call, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": "c", "method": "tools/call", "params": map[string]any{"name": tokens[4], + "arguments": args, "_meta": map[string]any{stdio.MetaCaller: strings.ReplaceAll(tokens[5], "~", ".")}}}) + if _, err := inW.Write(append(call, '\n')); err != nil { + return + } + a := <-answers + res, _ := a["result"].(map[string]any) + content, _ := res["content"].([]any) + text := "" + if len(content) == 1 { + text, _ = content[0].(map[string]any)["text"].(string) + } + out := map[string]any{"result": json.RawMessage(text), "node": "n1"} + if isErr, _ := res["isError"].(bool); isErr || len(content) != 1 { + out = map[string]any{"error": text} + } + reply, _ := json.Marshal(out) + _ = msg.Respond(reply) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + ask := func(user, tool string) (views.Answer, error) { + return views.AskVia(func(key string, body any) (json.RawMessage, error) { + data, _ := json.Marshal(body) + msg, err := nc.Request("mesh.mod."+delivery.AssignerModule+".call."+tool+"."+strings.ReplaceAll(user, ".", "~"), data, 5*time.Second) + if err != nil { + return nil, err + } + return msg.Data, nil + }, tool)(b.ctx, "") + } + readers := []string{"n1.mesh-grants", "n1.node-tools", "n1.mesh-delivery-indexer", "n1.mesh-delivery-assigner", "n1.mesh-delivery-builder"} + for tool, allowed := range granted { + for _, user := range readers { + a, err := ask(user, tool) + switch { + case slices.Contains(allowed, user) && err != nil: + t.Errorf("%s's first view of %s before any new-form state: %v", user, tool, err) + case slices.Contains(allowed, user): + keys, _ := b.buckets[viewedStates[tool]].Keys(b.ctx) + if a.Owner != delivery.AssignerModule || a.State != tool || len(a.Values) != len(keys) { + t.Errorf("%s's first view of %s = %+v, the state holds %v", user, tool, a, keys) + } + case err == nil || !strings.Contains(err.Error(), "refused"): + t.Errorf("%s, granted no first view of %s, was answered %+v, %v", user, tool, a, err) + } + } + } +}