From 481a83e11c4fe8e9774c5bb8032995f4f94247a4 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 22:24:02 +0100 Subject: [PATCH] Round 2 T2, T3, T4, T5: the tool runner ends G with its merge-base test count; the wait test runs its rows side by side MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tests G added fold into existing ones as subtests (design 47 G §626, A-306: 171 tests down to 154, integration 9 down to 7). TestMeshWaitRefusesLoudlyAndNeverLoops runs its groups of one-second rows as parallel subtests, each on a runtime of its own, 15 s down to 5 s (A-311: the one-second floor of a wait's timeout stands, no clock is injected). The go.mod no longer requires meshconditionsclient, which no package loads (T4). Node names in the lines G adds are role names (T5). Builds: design 47 G §626, design 34 every-event §19, A-306, A-311; review round 2 findings T2, T3, T4, T5 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- node-tools/go.mod | 1 - node-tools/internal/approval/approval_test.go | 83 ++++++++++ node-tools/internal/approval/marks_test.go | 86 ---------- node-tools/internal/bus/chunked_test.go | 81 +++++----- node-tools/internal/bus/sent_test.go | 27 ---- node-tools/internal/bus/state_watch_test.go | 137 ++++++++-------- .../internal/bus/streams_internal_test.go | 25 ++- node-tools/internal/console/refusal_test.go | 48 ++++++ node-tools/internal/console/sent_test.go | 56 ------- node-tools/internal/console/wait_test.go | 147 +++++++++++------- .../internal/firstview/firstview_test.go | 71 +++++---- node-tools/internal/launch/seat_test.go | 60 +++++++ node-tools/internal/launch/sessions_test.go | 65 -------- node-tools/internal/ledger/ledger_test.go | 91 ++++++----- node-tools/internal/runtime/announce_test.go | 99 ++++++------ node-tools/internal/runtime/caller_test.go | 70 ++++++++- node-tools/internal/runtime/chunked_test.go | 30 ---- .../internal/runtime/gate_approvals_test.go | 53 ------- .../runtime/ledger_integration_test.go | 67 ++++++++ .../internal/runtime/own_account_test.go | 37 +++-- node-tools/internal/runtime/sessions_test.go | 118 +++++++------- .../runtime/subscribenew_integration_test.go | 84 ---------- 22 files changed, 757 insertions(+), 779 deletions(-) delete mode 100644 node-tools/internal/bus/sent_test.go delete mode 100644 node-tools/internal/console/sent_test.go delete mode 100644 node-tools/internal/runtime/chunked_test.go delete mode 100644 node-tools/internal/runtime/subscribenew_integration_test.go diff --git a/node-tools/go.mod b/node-tools/go.mod index 7d01aa2..606687c 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -5,7 +5,6 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index d722adb..97c209f 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -3,6 +3,7 @@ package approval import ( "encoding/json" "errors" + "reflect" "strings" "sync" "testing" @@ -506,4 +507,86 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } { t.Run(tc.name, func(t *testing.T) { tc.run(t, newWorld()) }) } + // Design 46 7-approval-mark §5, §3, §4 and design 34 every-event §33: the marks a module's calls are gated by, derived from + // fixture catalog halves and served entries. + t.Run("TheMarks", func(t *testing.T) { + login := map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", + Verbs: []seatdef.Verb{{Name: "execute", NeedsApproval: true}, {Name: "sessions"}}}} + v1 := entry("src2:1", []string{"node-login-shell.execute", "node-login-shell.sessions"}, + map[string]any{"needs-approval": []any{"wipe"}}) + v2 := entry("src2:2", []string{"node-login-shell.execute", "node-login-shell.sessions"}, + map[string]any{"needs-approval": []any{"erase", "node-login-shell.sessions"}}) + mail := catalogseat.CatalogEntry{Module: "mail", Fp: "src2:m", Manifest: map[string]any{ + "needs-approval": []any{"mail.received"}, "emits": []any{"mail.received"}, + "streams": []any{map[string]any{"name": "MAIL", "events": []any{"mail.**"}, "grant": map[string]any{"config": "readers", "events": "mail.{key}.**"}}}}} + for _, tc := range []struct { + name string + cat *fakeCatalog + module string + fp string + tools []string + verbs map[string]string + subjects []string + unread bool + }{ + {name: "a seat's mark stands for every holder, and the holder's own adds up", cat: &fakeCatalog{revision: "7", + byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1}, newest: map[string]catalogseat.CatalogEntry{"shell": v1}, defs: login}, + module: "shell", fp: "src2:1", tools: []string{"wipe"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell"}}, + {name: "a holder's needs-approval cannot unmark its seat's verb, and marks another", cat: &fakeCatalog{revision: "8", + byFp: map[string]catalogseat.CatalogEntry{"src2:2": v2}, newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, + module: "shell", fp: "src2:2", tools: []string{"erase"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", + "node-login-shell.sessions": "entry shell src2:2"}}, + {name: "across a version change the union stands: a dropped mark is asked while the old bundle runs", cat: &fakeCatalog{revision: "9", + byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1, "src2:2": v2}, newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, + module: "shell", fp: "src2:1", tools: []string{"erase", "wipe"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", + "node-login-shell.sessions": "entry shell src2:2"}}, + {name: "a served entry that cannot be read leaves the newest marks, every marked call refused", cat: &fakeCatalog{revision: "9", + newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, + module: "shell", fp: "src2:1", tools: []string{"erase"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", + "node-login-shell.sessions": "entry shell src2:2"}, unread: true}, + {name: "a bundle with no fingerprint has no served entry", cat: &fakeCatalog{revision: "9", defs: login}, + module: "shell", unread: true}, + {name: "a marked event and a stream with a grant give their subjects and each .>", cat: &fakeCatalog{revision: "3", + byFp: map[string]catalogseat.CatalogEntry{"src2:m": mail}, newest: map[string]catalogseat.CatalogEntry{"mail": mail}}, + module: "mail", fp: "src2:m", subjects: []string{"mesh.mod.mail.event.mail.received", "mesh.mod.mail.event.mail.received.>", + "mesh.stream.mail.event.mail.>", "mesh.stream.mail.event.mail.received", "mesh.stream.mail.event.mail.received.>"}}, + } { + t.Run(tc.name, func(t *testing.T) { + m := gate.Derive(tc.cat, tc.module, tc.fp) + if got := keysOf(m.Tools); !reflect.DeepEqual(got, tc.tools) { + t.Errorf("tools %v, want %v", got, tc.tools) + } + if tc.verbs == nil { + tc.verbs = map[string]string{} + } + if !reflect.DeepEqual(m.Verbs, tc.verbs) { + t.Errorf("verbs %v, want %v", m.Verbs, tc.verbs) + } + if got := keysOf(m.Subjects); !reflect.DeepEqual(got, tc.subjects) { + t.Errorf("subjects %v, want %v", got, tc.subjects) + } + if (m.Unread != "") != tc.unread || m.Revision != tc.cat.revision { + t.Errorf("unread %q, revision %q", m.Unread, m.Revision) + } + }) + } + // §4: a bundle whose marks name a tool it does not register is not loaded, its finding mark-unserved raised; a bundle that + // serves it clears it, and a catalog without the mark too. + t.Run("mark-unserved raised and cleared", func(t *testing.T) { + cat := &fakeCatalog{revision: "7", byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1}, newest: map[string]catalogseat.CatalogEntry{"shell": v1}, defs: login} + w := &findings{} + g := gate.New(gate.Deps{Node: "n1", Runner: "node-tools", Catalog: cat, Findings: Findings(w, "n1", "node-tools", nil)}) + if got := g.Unserved("shell", "src2:1", []string{"other"}); !reflect.DeepEqual(got, []string{"wipe"}) || w.raised[keys.EscapeName("n1.shell")] != gate.KindMarkUnserved { + t.Fatalf("unserved %v, findings %v", got, w.raised) + } + if got := g.Unserved("shell", "src2:1", []string{"wipe"}); got != nil || w.raised[keys.EscapeName("n1.shell")] != "" { + t.Fatalf("a bundle serving the mark: unserved %v, findings %v", got, w.raised) + } + _ = g.Unserved("shell", "src2:1", nil) + cat.byFp["src2:1"], cat.newest["shell"] = entry("src2:1", nil, nil), entry("src2:1", nil, nil) + if got := g.Unserved("shell", "src2:1", nil); got != nil || w.raised[keys.EscapeName("n1.shell")] != "" { + t.Fatalf("a catalog without the mark: unserved %v, findings %v", got, w.raised) + } + }) + }) } diff --git a/node-tools/internal/approval/marks_test.go b/node-tools/internal/approval/marks_test.go index a38f217..708536c 100644 --- a/node-tools/internal/approval/marks_test.go +++ b/node-tools/internal/approval/marks_test.go @@ -2,12 +2,9 @@ package approval import ( "context" - "reflect" "sort" - "testing" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" - "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" catalogseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" @@ -47,89 +44,6 @@ func entry(fp string, serves []string, manifest map[string]any) catalogseat.Cata return catalogseat.CatalogEntry{Module: "shell", Fp: fp, Serves: serves, Manifest: manifest} } -// Design 46 7-approval-mark §5, §3, §4 and design 34 every-event §33: the marks a module's calls are gated by, derived from -// fixture catalog halves and served entries. -func TestTheMarks(t *testing.T) { - login := map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", - Verbs: []seatdef.Verb{{Name: "execute", NeedsApproval: true}, {Name: "sessions"}}}} - v1 := entry("src2:1", []string{"node-login-shell.execute", "node-login-shell.sessions"}, - map[string]any{"needs-approval": []any{"wipe"}}) - v2 := entry("src2:2", []string{"node-login-shell.execute", "node-login-shell.sessions"}, - map[string]any{"needs-approval": []any{"erase", "node-login-shell.sessions"}}) - mail := catalogseat.CatalogEntry{Module: "mail", Fp: "src2:m", Manifest: map[string]any{ - "needs-approval": []any{"mail.received"}, "emits": []any{"mail.received"}, - "streams": []any{map[string]any{"name": "MAIL", "events": []any{"mail.**"}, "grant": map[string]any{"config": "readers", "events": "mail.{key}.**"}}}}} - for _, tc := range []struct { - name string - cat *fakeCatalog - module string - fp string - tools []string - verbs map[string]string - subjects []string - unread bool - }{ - {name: "a seat's mark stands for every holder, and the holder's own adds up", cat: &fakeCatalog{revision: "7", - byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1}, newest: map[string]catalogseat.CatalogEntry{"shell": v1}, defs: login}, - module: "shell", fp: "src2:1", tools: []string{"wipe"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell"}}, - {name: "a holder's needs-approval cannot unmark its seat's verb, and marks another", cat: &fakeCatalog{revision: "8", - byFp: map[string]catalogseat.CatalogEntry{"src2:2": v2}, newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, - module: "shell", fp: "src2:2", tools: []string{"erase"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", - "node-login-shell.sessions": "entry shell src2:2"}}, - {name: "across a version change the union stands: a dropped mark is asked while the old bundle runs", cat: &fakeCatalog{revision: "9", - byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1, "src2:2": v2}, newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, - module: "shell", fp: "src2:1", tools: []string{"erase", "wipe"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", - "node-login-shell.sessions": "entry shell src2:2"}}, - {name: "a served entry that cannot be read leaves the newest marks, every marked call refused", cat: &fakeCatalog{revision: "9", - newest: map[string]catalogseat.CatalogEntry{"shell": v2}, defs: login}, - module: "shell", fp: "src2:1", tools: []string{"erase"}, verbs: map[string]string{"node-login-shell.execute": "seat node-login-shell", - "node-login-shell.sessions": "entry shell src2:2"}, unread: true}, - {name: "a bundle with no fingerprint has no served entry", cat: &fakeCatalog{revision: "9", defs: login}, - module: "shell", unread: true}, - {name: "a marked event and a stream with a grant give their subjects and each .>", cat: &fakeCatalog{revision: "3", - byFp: map[string]catalogseat.CatalogEntry{"src2:m": mail}, newest: map[string]catalogseat.CatalogEntry{"mail": mail}}, - module: "mail", fp: "src2:m", subjects: []string{"mesh.mod.mail.event.mail.received", "mesh.mod.mail.event.mail.received.>", - "mesh.stream.mail.event.mail.>", "mesh.stream.mail.event.mail.received", "mesh.stream.mail.event.mail.received.>"}}, - } { - t.Run(tc.name, func(t *testing.T) { - m := gate.Derive(tc.cat, tc.module, tc.fp) - if got := keysOf(m.Tools); !reflect.DeepEqual(got, tc.tools) { - t.Errorf("tools %v, want %v", got, tc.tools) - } - if tc.verbs == nil { - tc.verbs = map[string]string{} - } - if !reflect.DeepEqual(m.Verbs, tc.verbs) { - t.Errorf("verbs %v, want %v", m.Verbs, tc.verbs) - } - if got := keysOf(m.Subjects); !reflect.DeepEqual(got, tc.subjects) { - t.Errorf("subjects %v, want %v", got, tc.subjects) - } - if (m.Unread != "") != tc.unread || m.Revision != tc.cat.revision { - t.Errorf("unread %q, revision %q", m.Unread, m.Revision) - } - }) - } - // §4: a bundle whose marks name a tool it does not register is not loaded, its finding mark-unserved raised; a bundle that - // serves it clears it, and a catalog without the mark too. - t.Run("mark-unserved raised and cleared", func(t *testing.T) { - cat := &fakeCatalog{revision: "7", byFp: map[string]catalogseat.CatalogEntry{"src2:1": v1}, newest: map[string]catalogseat.CatalogEntry{"shell": v1}, defs: login} - w := &findings{} - g := gate.New(gate.Deps{Node: "n1", Runner: "node-tools", Catalog: cat, Findings: Findings(w, "n1", "node-tools", nil)}) - if got := g.Unserved("shell", "src2:1", []string{"other"}); !reflect.DeepEqual(got, []string{"wipe"}) || w.raised[keys.EscapeName("n1.shell")] != gate.KindMarkUnserved { - t.Fatalf("unserved %v, findings %v", got, w.raised) - } - if got := g.Unserved("shell", "src2:1", []string{"wipe"}); got != nil || w.raised[keys.EscapeName("n1.shell")] != "" { - t.Fatalf("a bundle serving the mark: unserved %v, findings %v", got, w.raised) - } - _ = g.Unserved("shell", "src2:1", nil) - cat.byFp["src2:1"], cat.newest["shell"] = entry("src2:1", nil, nil), entry("src2:1", nil, nil) - if got := g.Unserved("shell", "src2:1", nil); got != nil || w.raised[keys.EscapeName("n1.shell")] != "" { - t.Fatalf("a catalog without the mark: unserved %v, findings %v", got, w.raised) - } - }) -} - // findings is the runner's held writer as a fixture: each finding raised now, by id. type findings struct { raised map[string]string diff --git a/node-tools/internal/bus/chunked_test.go b/node-tools/internal/bus/chunked_test.go index c034965..a838f2e 100644 --- a/node-tools/internal/bus/chunked_test.go +++ b/node-tools/internal/bus/chunked_test.go @@ -73,46 +73,45 @@ func TestAChunkedSeatAnswerIsReceivedByTheRunnerAndHandedWhole(t *testing.T) { if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) { t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err) } -} - -// Design 47 G A-257, A-254: a bundle answers a chunked verb with the whole file in the member `file`; the runner that serves the -// verb sends the file in chunks (each acknowledged before the next) and answers the count, so the bundle sends none, and the asker -// of one node's verb receives it whole. -func TestABundlesChunkedAnswerIsSentByTheRunner(t *testing.T) { - url := mt.URL(t) - holder := dial(t, url, "mesh-build-go", "anchor", "") - asker := dial(t, url, "mesh-delivery-indexer", "anchor", "") - file := make([]byte, buscalls.SeatChunkBytes*2+7) - for i := range file { - file[i] = byte(i * 7) - } - stop, err := holder.Handle("seat:node-build-go.output@anchor", func(body json.RawMessage) (any, error) { - answer := map[string]any{"ask": "a1", "file": base64.StdEncoding.EncodeToString(file)} - return holder.AnswerChunked("node-build-go", "output", body, answer) + // Design 47 G A-257, A-254: a bundle answers a chunked verb with the whole file in the member `file`; the runner that serves the + // verb sends the file in chunks (each acknowledged before the next) and answers the count, so the bundle sends none, and the asker + // of one node's verb receives it whole. + t.Run("ABundlesChunkedAnswerIsSentByTheRunner", func(t *testing.T) { + url := mt.URL(t) + holder := dial(t, url, "mesh-build-go", "anchor", "") + asker := dial(t, url, "mesh-delivery-indexer", "anchor", "") + file := make([]byte, buscalls.SeatChunkBytes*2+7) + for i := range file { + file[i] = byte(i * 7) + } + stop, err := holder.Handle("seat:node-build-go.output@anchor", func(body json.RawMessage) (any, error) { + answer := map[string]any{"ask": "a1", "file": base64.StdEncoding.EncodeToString(file)} + return holder.AnswerChunked("node-build-go", "output", body, answer) + }) + if err != nil { + t.Fatal(err) + } + defer stop() + holder.Flush() // the bus holds the verb's subscription before it is asked: the asker is another connection + got, err := asker.AskAsChunked("mesh-delivery-indexer", "seat:node-build-go.output@anchor", map[string]any{"ask": "a1"}) + if err != nil { + t.Fatal(err) + } + var ans struct { + Ask string `json:"ask"` + Chunks int `json:"chunks"` + File string `json:"file"` + } + if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Ask != "a1" || ans.Chunks != 3 { + t.Fatalf("the answer %s: %v", got.Result, err) + } + if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) { + t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err) + } + // an answer with no file is the verb's own: no chunks are sent + plain, err := holder.AnswerChunked("node-build-go", "output", json.RawMessage(`{"call":"c"}`), map[string]any{"ask": "x", "chunks": 0}) + if m, _ := plain.(map[string]any); err != nil || m["chunks"] != 0 { + t.Fatalf("an answer with no file was %v, %v", plain, err) + } }) - if err != nil { - t.Fatal(err) - } - defer stop() - holder.Flush() // the bus holds the verb's subscription before it is asked: the asker is another connection - got, err := asker.AskAsChunked("mesh-delivery-indexer", "seat:node-build-go.output@anchor", map[string]any{"ask": "a1"}) - if err != nil { - t.Fatal(err) - } - var ans struct { - Ask string `json:"ask"` - Chunks int `json:"chunks"` - File string `json:"file"` - } - if err := json.Unmarshal(got.Result, &ans); err != nil || ans.Ask != "a1" || ans.Chunks != 3 { - t.Fatalf("the answer %s: %v", got.Result, err) - } - if body, err := base64.StdEncoding.DecodeString(ans.File); err != nil || string(body) != string(file) { - t.Errorf("the file handed is %d bytes, want the %d sent: %v", len(body), len(file), err) - } - // an answer with no file is the verb's own: no chunks are sent - plain, err := holder.AnswerChunked("node-build-go", "output", json.RawMessage(`{"call":"c"}`), map[string]any{"ask": "x", "chunks": 0}) - if m, _ := plain.(map[string]any); err != nil || m["chunks"] != 0 { - t.Fatalf("an answer with no file was %v, %v", plain, err) - } } diff --git a/node-tools/internal/bus/sent_test.go b/node-tools/internal/bus/sent_test.go deleted file mode 100644 index ca7d1b4..0000000 --- a/node-tools/internal/bus/sent_test.go +++ /dev/null @@ -1,27 +0,0 @@ -package bus - -import ( - "strings" - "testing" - - "git.novox.be/novox/mesh-sdk/go/buscalls" -) - -// Design 47 G §347, §59, §462: a sent object is read by the reference that carried it, from a sender's sent store only; a -// state's objects, and a reference that is none are refused before the bus is asked. -func TestSentGetRefusesWhatIsNoSentObjectsReference(t *testing.T) { - digest := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" - for _, c := range []struct { - name string - req buscalls.ObjectsSentGetRequest - want string - }{ - {"no digest", buscalls.ObjectsSentGetRequest{Store: "m_sent-objects", Object: "o", Size: 3}, "named by the reference"}, - {"a state's objects", buscalls.ObjectsSentGetRequest{Store: "m_asks-objects", Object: "o", Size: 3, Digest: digest}, "no sent store"}, - } { - _, err := (&Conn{}).SentGet(c.req) - if err == nil || !strings.Contains(err.Error(), c.want) { - t.Errorf("%s: got %v, want an error containing %q", c.name, err, c.want) - } - } -} diff --git a/node-tools/internal/bus/state_watch_test.go b/node-tools/internal/bus/state_watch_test.go index 42b16da..15280aa 100644 --- a/node-tools/internal/bus/state_watch_test.go +++ b/node-tools/internal/bus/state_watch_test.go @@ -87,6 +87,74 @@ func TestAWatchHoldsWhileItsCurrentValuesKeepArriving(t *testing.T) { } return nil }) + // **A watch given up, or stopped, hands over nothing more** (issue 774). The runtime's watch stopped its watcher + // when the current values did not arrive within the bound, but the watcher still held values, and its goroutine + // went on handing them over: to a watch id the bundle no longer held (on the provisioner, 51 failed watches were + // followed by 1257 orphan changes). Each row blocks the bundle on its first value, gives the watch up, lets the + // bundle go and counts what it is still handed. + t.Run("AWatchGivenUpHandsOverNothingMore", func(t *testing.T) { + for _, tc := range []struct { + name string + refused bool // the bound gives it up, or the caller stops it + }{ + {"refused at the bound", true}, + {"stopped by its caller", false}, + } { + t.Run(tc.name, func(t *testing.T) { + was := bus.StateTimeout + bus.StateTimeout = 300 * time.Millisecond + t.Cleanup(func() { bus.StateTimeout = was }) + conn, kv := owedWatcher(t) + put := func(i int) { + if _, err := kv.Put(fmt.Sprintf("novox/hq.%d.%040x.status=mesh/check", i+1, i), []byte(`{"v":1}`)); err != nil { + t.Fatal(err) + } + } + if tc.refused { + for i := 0; i < 4; i++ { + put(i) + } + } + var calls atomic.Int32 + first := make(chan struct{}, 1) + release := make(chan struct{}) + deliver := func(ch bus.StateChange) error { + if calls.Add(1) == 1 { + first <- struct{}{} + <-release + } + return nil + } + var stop func() + if tc.refused { + if _, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err == nil { + t.Fatal("the watch whose bundle took nothing was not refused") + } + } else { + var err error + if stop, err = conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err != nil { + t.Fatal(err) + } + for i := 0; i < 4; i++ { + put(i) + } + select { + case <-first: + case <-time.After(5 * time.Second): + t.Fatal("the bundle was handed no change") + } + stop() + } + // Let the watcher's server side deliver what it held, then free the bundle. + time.Sleep(300 * time.Millisecond) + close(release) + time.Sleep(500 * time.Millisecond) + if n := calls.Load(); n != 1 { + t.Fatalf("the bundle was handed %d changes of a watch given up, want 1 (the one in hand)", n) + } + }) + } + }) } // A watch whose bundle takes nothing for longer than the bound is still refused, so a bundle that hangs @@ -187,72 +255,3 @@ func owedWatcher(t *testing.T) (*bus.Conn, interface { }) return conn, kv } - -// **A watch given up, or stopped, hands over nothing more** (issue 774). The runtime's watch stopped its watcher -// when the current values did not arrive within the bound, but the watcher still held values, and its goroutine -// went on handing them over: to a watch id the bundle no longer held (on the provisioner, 51 failed watches were -// followed by 1257 orphan changes). Each row blocks the bundle on its first value, gives the watch up, lets the -// bundle go and counts what it is still handed. -func TestAWatchGivenUpHandsOverNothingMore(t *testing.T) { - for _, tc := range []struct { - name string - refused bool // the bound gives it up, or the caller stops it - }{ - {"refused at the bound", true}, - {"stopped by its caller", false}, - } { - t.Run(tc.name, func(t *testing.T) { - was := bus.StateTimeout - bus.StateTimeout = 300 * time.Millisecond - t.Cleanup(func() { bus.StateTimeout = was }) - conn, kv := owedWatcher(t) - put := func(i int) { - if _, err := kv.Put(fmt.Sprintf("novox/hq.%d.%040x.status=mesh/check", i+1, i), []byte(`{"v":1}`)); err != nil { - t.Fatal(err) - } - } - if tc.refused { - for i := 0; i < 4; i++ { - put(i) - } - } - var calls atomic.Int32 - first := make(chan struct{}, 1) - release := make(chan struct{}) - deliver := func(ch bus.StateChange) error { - if calls.Add(1) == 1 { - first <- struct{}{} - <-release - } - return nil - } - var stop func() - if tc.refused { - if _, err := conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err == nil { - t.Fatal("the watch whose bundle took nothing was not refused") - } - } else { - var err error - if stop, err = conn.StateWatch("gitea", "mesh-delivery-checker.owed", "", deliver); err != nil { - t.Fatal(err) - } - for i := 0; i < 4; i++ { - put(i) - } - select { - case <-first: - case <-time.After(5 * time.Second): - t.Fatal("the bundle was handed no change") - } - stop() - } - // Let the watcher's server side deliver what it held, then free the bundle. - time.Sleep(300 * time.Millisecond) - close(release) - time.Sleep(500 * time.Millisecond) - if n := calls.Load(); n != 1 { - t.Fatalf("the bundle was handed %d changes of a watch given up, want 1 (the one in hand)", n) - } - }) - } -} diff --git a/node-tools/internal/bus/streams_internal_test.go b/node-tools/internal/bus/streams_internal_test.go index be4cefc..5b176c3 100644 --- a/node-tools/internal/bus/streams_internal_test.go +++ b/node-tools/internal/bus/streams_internal_test.go @@ -1,6 +1,11 @@ package bus -import "testing" +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/buscalls" +) // novox/hq ADR 0363 §2, §3: an event goes to the stream whose pattern carries it, as the bus matches subjects. func TestAnEventFallsUnderAStreamsPatternAsTheBusMatchesIt(t *testing.T) { @@ -22,6 +27,24 @@ func TestAnEventFallsUnderAStreamsPatternAsTheBusMatchesIt(t *testing.T) { t.Errorf("%q under %q: %v, want %v", c.event, c.pattern, got, c.want) } } + // Design 47 G §347, §59, §462: a sent object is read by the reference that carried it, from a sender's sent store only; a + // state's objects, and a reference that is none are refused before the bus is asked. + t.Run("SentGetRefusesWhatIsNoSentObjectsReference", func(t *testing.T) { + digest := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + for _, c := range []struct { + name string + req buscalls.ObjectsSentGetRequest + want string + }{ + {"no digest", buscalls.ObjectsSentGetRequest{Store: "m_sent-objects", Object: "o", Size: 3}, "named by the reference"}, + {"a state's objects", buscalls.ObjectsSentGetRequest{Store: "m_asks-objects", Object: "o", Size: 3, Digest: digest}, "no sent store"}, + } { + _, err := (&Conn{}).SentGet(c.req) + if err == nil || !strings.Contains(err.Error(), c.want) { + t.Errorf("%s: got %v, want an error containing %q", c.name, err, c.want) + } + } + }) } // A membership naming no stream publishes every event on EVENTS, exactly as before (issue 725). diff --git a/node-tools/internal/console/refusal_test.go b/node-tools/internal/console/refusal_test.go index 68e350c..5d2111f 100644 --- a/node-tools/internal/console/refusal_test.go +++ b/node-tools/internal/console/refusal_test.go @@ -1,9 +1,18 @@ package console import ( + "context" + "crypto/sha256" + "encoding/hex" "errors" "strings" "testing" + + "git.novox.be/novox/mesh-sdk/go/stateobjects" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-tools/node-tools/internal/bus" + "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) // A refusal is said only for the bus's own refusals; a tool's answer that mentions authorization is the @@ -19,6 +28,45 @@ func TestARefusalIsSaidOnlyForTheBussOwn(t *testing.T) { t.Errorf("%q read as a refusal: %v, want %v", msg, got, refusal) } } + // ADR 0666 §16: an answer a module sent as a reference is read from its sent store for the person asking; one the bus does + // not let the runner read is shown as its reference with its size and why it was not read. + t.Run("AnAnswerSentAsAReferenceIsReadForThePersonAsking", func(t *testing.T) { + nc, err := nats.Connect(meshtest.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + obs, err := js.CreateObjectStore(context.Background(), jetstream.ObjectStoreConfig{Bucket: "m_sent-objects"}) + if err != nil { + t.Fatal(err) + } + conn, err := bus.Connect(bus.Credential{URL: meshtest.URL(t), Module: "console", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + body := []byte(`{"lines":"` + strings.Repeat("x", 9<<20) + `"}`) + sum := sha256.Sum256(body) + ref := stateobjects.Ref{Store: "m_sent-objects", Object: "tool.lines." + hex.EncodeToString(sum[:]), Size: int64(len(body)), + Digest: hex.EncodeToString(sum[:]), Form: stateobjects.FormVersion} + if _, err := obs.PutBytes(context.Background(), ref.Object, body); err != nil { + t.Fatal(err) + } + if got, unread := resolveSent(conn, stateobjects.Wrap(ref)); unread != "" || string(got) != string(body) { + t.Fatalf("read %d bytes of %d, %q", len(got), len(body), unread) + } + missing := ref + missing.Store = "gone_sent-objects" + wrapped := stateobjects.Wrap(missing) + got, unread := resolveSent(conn, wrapped) + if string(got) != string(wrapped) || !strings.Contains(unread, "9437") || !strings.Contains(unread, "gone_sent-objects") { + t.Fatalf("an unread answer showed %q and %q", got, unread) + } + if got, unread := resolveSent(conn, []byte(`{"a":1}`)); unread != "" || string(got) != `{"a":1}` { + t.Fatalf("a plain answer showed %q, %q", got, unread) + } + }) } // **A timeout says it is not a failure** (novox/hq issue 265): a push that outlasted the wait had diff --git a/node-tools/internal/console/sent_test.go b/node-tools/internal/console/sent_test.go deleted file mode 100644 index 80cd477..0000000 --- a/node-tools/internal/console/sent_test.go +++ /dev/null @@ -1,56 +0,0 @@ -package console - -import ( - "context" - "crypto/sha256" - "encoding/hex" - "strings" - "testing" - - "git.novox.be/novox/mesh-sdk/go/stateobjects" - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - - "github.com/novox/mesh-tools/node-tools/internal/bus" - "github.com/novox/mesh-tools/node-tools/internal/meshtest" -) - -// ADR 0666 §16: an answer a module sent as a reference is read from its sent store for the person asking; one the bus does -// not let the runner read is shown as its reference with its size and why it was not read. -func TestAnAnswerSentAsAReferenceIsReadForThePersonAsking(t *testing.T) { - nc, err := nats.Connect(meshtest.URL(t)) - if err != nil { - t.Fatal(err) - } - defer nc.Close() - js, _ := jetstream.New(nc) - obs, err := js.CreateObjectStore(context.Background(), jetstream.ObjectStoreConfig{Bucket: "m_sent-objects"}) - if err != nil { - t.Fatal(err) - } - conn, err := bus.Connect(bus.Credential{URL: meshtest.URL(t), Module: "console", Node: "anchor"}) - if err != nil { - t.Fatal(err) - } - defer conn.Close() - body := []byte(`{"lines":"` + strings.Repeat("x", 9<<20) + `"}`) - sum := sha256.Sum256(body) - ref := stateobjects.Ref{Store: "m_sent-objects", Object: "tool.lines." + hex.EncodeToString(sum[:]), Size: int64(len(body)), - Digest: hex.EncodeToString(sum[:]), Form: stateobjects.FormVersion} - if _, err := obs.PutBytes(context.Background(), ref.Object, body); err != nil { - t.Fatal(err) - } - if got, unread := resolveSent(conn, stateobjects.Wrap(ref)); unread != "" || string(got) != string(body) { - t.Fatalf("read %d bytes of %d, %q", len(got), len(body), unread) - } - missing := ref - missing.Store = "gone_sent-objects" - wrapped := stateobjects.Wrap(missing) - got, unread := resolveSent(conn, wrapped) - if string(got) != string(wrapped) || !strings.Contains(unread, "9437") || !strings.Contains(unread, "gone_sent-objects") { - t.Fatalf("an unread answer showed %q and %q", got, unread) - } - if got, unread := resolveSent(conn, []byte(`{"a":1}`)); unread != "" || string(got) != `{"a":1}` { - t.Fatalf("a plain answer showed %q, %q", got, unread) - } -} diff --git a/node-tools/internal/console/wait_test.go b/node-tools/internal/console/wait_test.go index 984aca2..1fe7562 100644 --- a/node-tools/internal/console/wait_test.go +++ b/node-tools/internal/console/wait_test.go @@ -203,7 +203,7 @@ func TestMeshWaitsTimeoutIsBounded(t *testing.T) { // a wildcard, of one token, more than 16 names, none, or a from of another word is refused at once naming the rule; a // wait whose every subscription the bus refuses answers after its timeout, in words naming no wait grant. func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { - s, mesh := waitingRuntime(t) + s, _ := waitingRuntime(t) var seventeen []any for i := 0; i < 17; i++ { seventeen = append(seventeen, fmt.Sprintf("mesh-knowledge.reviewed%d", i)) @@ -230,63 +230,80 @@ func TestMeshWaitRefusesLoudlyAndNeverLoops(t *testing.T) { } } - // E §14: against the catalog half, a misspelt emitter is refused naming the half and its revision, a known emitter's - // undeclared event is waited on and its timeout says so, and a declared one says nothing of it. - half := NewSurface(s.conn, halfSource{busSource{s.conn}, map[string]meshcatalogclient.Half{"main": { - Revision: catalogseat.CatalogRevision{Revision: 7, Definitions: []catalogseat.CatalogRevisionDefinitionsItem{{Seat: "mesh-catalog"}}}, - Entries: map[string]catalogseat.CatalogEntry{"mesh-delivery-view": {Module: "mesh-delivery-view", - Manifest: map[string]any{"emits": []any{"delivered"}}}}, - }}}, "desk.node-tools") - for _, c := range []struct { - name, says string - refused bool - }{ - {"mesh-delivry-view.delivered", "mesh-delivry-view is neither a seat nor a module of the catalog half (mesh-catalog.revision, revision 7 of main)", true}, - {"mesh-delivery-view.planned", "no declaration names mesh-delivery-view.planned", false}, - {"mesh-catalog.catalog-revision", "", false}, - } { - text, isErr := callWait(half, map[string]any{"subjects": []any{c.name}, "timeout": 1}) - if isErr != c.refused || c.says != "" && !strings.Contains(text, c.says) || - c.says == "" && (strings.Contains(text, "undeclared") || strings.Contains(text, "unchecked")) { - t.Errorf("%s against the catalog half: %s", c.name, text) + t.Run("the catalog half", func(t *testing.T) { + t.Parallel() + // E §14: against the catalog half, a misspelt emitter is refused naming the half and its revision, a known emitter's + // undeclared event is waited on and its timeout says so, and a declared one says nothing of it. + half := NewSurface(s.conn, halfSource{busSource{s.conn}, map[string]meshcatalogclient.Half{"main": { + Revision: catalogseat.CatalogRevision{Revision: 7, Definitions: []catalogseat.CatalogRevisionDefinitionsItem{{Seat: "mesh-catalog"}}}, + Entries: map[string]catalogseat.CatalogEntry{"mesh-delivery-view": {Module: "mesh-delivery-view", + Manifest: map[string]any{"emits": []any{"delivered"}}}}, + }}}, "desk.node-tools") + for _, c := range []struct { + name, says string + refused bool + }{ + {"mesh-delivry-view.delivered", "mesh-delivry-view is neither a seat nor a module of the catalog half (mesh-catalog.revision, revision 7 of main)", true}, + {"mesh-delivery-view.planned", "no declaration names mesh-delivery-view.planned", false}, + {"mesh-catalog.catalog-revision", "", false}, + } { + text, isErr := callWait(half, map[string]any{"subjects": []any{c.name}, "timeout": 1}) + if isErr != c.refused || c.says != "" && !strings.Contains(text, c.says) || + c.says == "" && (strings.Contains(text, "undeclared") || strings.Contains(text, "unchecked")) { + t.Errorf("%s against the catalog half: %s", c.name, text) + } } - } + }) + + t.Run("a bus that refuses every subscription", func(t *testing.T) { + t.Parallel() + // The bus refusing every subscription of the call: answered after the timeout, in §16's words. + b, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true, + Users: []*server.User{{Username: "desk.node-tools", Password: "pw", Permissions: &server.Permissions{ + Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX.desk.node-tools.>", "mesh.assignment.desk.*"}}, + }}}}) + if err != nil { + t.Fatal(err) + } + go b.Start() + if !b.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(b.Shutdown) + conn, err := bus.Connect(bus.Credential{URL: b.ClientURL(), Module: "node-tools", Node: "desk", User: "desk.node-tools", Password: "pw"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + start := time.Now() + text, isErr := callWait(NewSurface(conn, nil, "desk.node-tools"), map[string]any{"subjects": []any{"mesh-delivery-view.delivered"}, "timeout": 1}) + if took := time.Since(start); took < time.Second || took > 3*time.Second { + t.Errorf("a wait the bus refused whole answered after %s, want its timeout of 1s", took) + } + if !isErr || !strings.Contains(text, "the bus refused this account a subscription on mesh.seat.mesh-delivery-view.event.delivered") || + !strings.Contains(text, "novox/hq ADR 0670 §1") || strings.Contains(text, "wait grant") || strings.Contains(text, "0367 §2") { + t.Errorf("a wait the bus refused whole: %s", text) + } + }) - // The bus refusing every subscription of the call: answered after the timeout, in §16's words. - b, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, NoLog: true, NoSigs: true, - Users: []*server.User{{Username: "desk.node-tools", Password: "pw", Permissions: &server.Permissions{ - Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX.desk.node-tools.>", "mesh.assignment.desk.*"}}, - }}}}) - if err != nil { - t.Fatal(err) - } - go b.Start() - if !b.ReadyForConnections(10 * time.Second) { - t.Fatal("the bus did not come up") - } - t.Cleanup(b.Shutdown) - conn, err := bus.Connect(bus.Credential{URL: b.ClientURL(), Module: "node-tools", Node: "desk", User: "desk.node-tools", Password: "pw"}) - if err != nil { - t.Fatal(err) - } - conn.Logf = func(string, ...any) {} - t.Cleanup(conn.Close) - start := time.Now() - text, isErr := callWait(NewSurface(conn, nil, "desk.node-tools"), map[string]any{"subjects": []any{"mesh-delivery-view.delivered"}, "timeout": 1}) - if took := time.Since(start); took < time.Second || took > 3*time.Second { - t.Errorf("a wait the bus refused whole answered after %s, want its timeout of 1s", took) - } - if !isErr || !strings.Contains(text, "the bus refused this account a subscription on mesh.seat.mesh-delivery-view.event.delivered") || - !strings.Contains(text, "novox/hq ADR 0670 §1") || strings.Contains(text, "wait grant") || strings.Contains(text, "0367 §2") { - t.Errorf("a wait the bus refused whole: %s", text) - } // Design 34 every-event §37: a wait whose subjects overlap a marked event's, or a stream's with a `grant`, runs only on the // operator's warrant for that one wait; with none it subscribes nothing, asks once and answers at its timeout or the choice. - markedWaits(t, s, mesh) + // Each group of rows has a runtime of its own, so the waits of the groups run side by side (A-311: the one-second floor of + // a wait's timeout stands, and no clock is injected). + for name, group := range map[string]func(*testing.T, *Surface, *mt.Mesh){ + "no warrant": markedWaitsWithNoWarrant, "a warrant": markedWaitsOnAWarrant, "the catalog half unread": markedWaitsBeforeTheHalfIsRead, + } { + t.Run(name, func(t *testing.T) { + t.Parallel() + s, mesh := waitingRuntime(t) + group(t, s, mesh) + }) + } } -// markedWaits are §37's rows, against a gate on the test bus, the router's records written there as the router writes them. -func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { +// markedWaitsWithNoWarrant are §37's rows of a wait with no warrant, against a gate on the test bus, the router's records written there as the router writes them. +func markedWaitsWithNoWarrant(t *testing.T, s *Surface, mesh *mt.Mesh) { gw := newGatedWaits(t, s, mesh, "src2:runner") g := gw.surface took := func(args map[string]any) (string, bool, time.Duration) { @@ -321,6 +338,17 @@ func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { !strings.Contains(text, `"timed-out": true`) || gw.asks() != 3 { t.Errorf("from a seat: %s, %d asks", text, gw.asks()) } +} + +// markedWaitsOnAWarrant are §37's rows of a wait chosen while it waits and of one on a warrant. +func markedWaitsOnAWarrant(t *testing.T, s *Surface, mesh *mt.Mesh) { + gw := newGatedWaits(t, s, mesh, "src2:runner") + g := gw.surface + took := func(args map[string]any) (string, bool, time.Duration) { + start := time.Now() + text, isErr := callWait(g, args) + return text, isErr, time.Since(start) + } // The most timeout a wait takes, its ask chosen while it waits: answered at the choice; the same call made then // subscribes and hears the event published after it, answered with its approval; a warrant for other names, another // filter or another timeout is refused, asked anew. @@ -328,7 +356,7 @@ func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { done := make(chan string, 1) go func() { text, _, d := took(long); done <- fmt.Sprintf("%s after %s", text, d) }() mt.Until(t, func() error { - if gw.asks() < 4 { + if gw.asks() < 1 { return os.ErrNotExist } return nil @@ -342,7 +370,7 @@ func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { case <-time.After(10 * time.Second): t.Fatal("a long marked wait was not answered at the choice") } - before = g.conn.Subscriptions() + before := g.conn.Subscriptions() go func() { listening(t, g, before, 6) mesh.Emit(t, "mesh.mod.mail.event.message", map[string]any{"n": 2}) @@ -361,6 +389,17 @@ func markedWaits(t *testing.T, s *Surface, mesh *mt.Mesh) { t.Errorf("another wait than the warrant's %v: %s", other, text) } } +} + +// markedWaitsBeforeTheHalfIsRead are §37's rows of a wait before the catalog half is read, and of one whose runner build is unknown. +func markedWaitsBeforeTheHalfIsRead(t *testing.T, s *Surface, mesh *mt.Mesh) { + gw := newGatedWaits(t, s, mesh, "src2:runner") + g := gw.surface + took := func(args map[string]any) (string, bool, time.Duration) { + start := time.Now() + text, isErr := callWait(g, args) + return text, isErr, time.Since(start) + } // Before the catalog half is read nothing is asked: judged when its first view comes within the timeout, refused at // the timeout when it does not. gw.cat.mu.Lock() diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index eacd22f..5221d6e 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -254,43 +254,42 @@ func TestTheFirstViewsAreTheManifestsViews(t *testing.T) { t.Errorf("the manifest still consumes the indexer's %s", ev) } } -} - -// design 47 G §616: the holder of the assigner seat is whichever module the catalog half says serves it; when another one -// claims it the views are of the new holder, and when none does the seat's names answer that the catalog half names no holder. -func TestTheAssignersViewsFollowTheModuleThatClaimsTheSeat(t *testing.T) { - o := newOwners() - deps := Deps{Module: "node-tools", Ask: o.ask, Bound: time.Second, Wait: 10 * time.Millisecond, - Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, Findings: o} - v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) - defer stop() - eventually(t, "not answered", func() bool { _, err := v.State(AssignmentsRead); return err == nil }) - ownerOf := func() string { - for _, s := range v.Status() { - if s.State == "assignments" { - return s.Owner + // design 47 G §616: the holder of the assigner seat is whichever module the catalog half says serves it; when another one + // claims it the views are of the new holder, and when none does the seat's names answer that the catalog half names no holder. + t.Run("TheAssignersViewsFollowTheModuleThatClaimsTheSeat", func(t *testing.T) { + o := newOwners() + deps := Deps{Module: "node-tools", Ask: o.ask, Bound: time.Second, Wait: 10 * time.Millisecond, + Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, Findings: o} + v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) + defer stop() + eventually(t, "not answered", func() bool { _, err := v.State(AssignmentsRead); return err == nil }) + ownerOf := func() string { + for _, s := range v.Status() { + if s.State == "assignments" { + return s.Owner + } + } + return "" + } + if ownerOf() != fixtureHolder { + t.Fatalf("the views are of %q", ownerOf()) + } + revise := func(rev int64, holder string, modules ...string) { + o.mu.Lock() + o.catalog, o.modules, o.holder = rev, modules, holder + o.mu.Unlock() + ev, _ := json.Marshal(catalogseat.CatalogRevisionEvent{V: 1, Branch: "main", Kind: meshcatalogclient.KindTrunk, Revision: rev}) + if err := o.deliver(bus.Envelope{Key: catalogseat.Seat + "." + catalogseat.EventCatalogRevision, Body: ev}); err != nil { + t.Fatal(err) } } - return "" - } - if ownerOf() != fixtureHolder { - t.Fatalf("the views are of %q", ownerOf()) - } - revise := func(rev int64, holder string, modules ...string) { - o.mu.Lock() - o.catalog, o.modules, o.holder = rev, modules, holder - o.mu.Unlock() - ev, _ := json.Marshal(catalogseat.CatalogRevisionEvent{V: 1, Branch: "main", Kind: meshcatalogclient.KindTrunk, Revision: rev}) - if err := o.deliver(bus.Envelope{Key: catalogseat.Seat + "." + catalogseat.EventCatalogRevision, Body: ev}); err != nil { - t.Fatal(err) - } - } - revise(4, "other-holder", "web", "other-holder") - eventually(t, "the views did not move to the new holder", func() bool { return ownerOf() == "other-holder" }) - eventually(t, "not answered", func() bool { _, err := v.State(SeatHoldersRead); return err == nil }) - revise(5, "", "web") - eventually(t, "the views stayed after the claim went", func() bool { - _, err := v.State(AssignmentsRead) - return err != nil && strings.Contains(err.Error(), "names no holder") + revise(4, "other-holder", "web", "other-holder") + eventually(t, "the views did not move to the new holder", func() bool { return ownerOf() == "other-holder" }) + eventually(t, "not answered", func() bool { _, err := v.State(SeatHoldersRead); return err == nil }) + revise(5, "", "web") + eventually(t, "the views stayed after the claim went", func() bool { + _, err := v.State(AssignmentsRead) + return err != nil && strings.Contains(err.Error(), "names no holder") + }) }) } diff --git a/node-tools/internal/launch/seat_test.go b/node-tools/internal/launch/seat_test.go index 1d23ab5..25edae4 100644 --- a/node-tools/internal/launch/seat_test.go +++ b/node-tools/internal/launch/seat_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -68,6 +69,65 @@ func TestABundleTakesASeatsWorkThroughItsOwnChannel(t *testing.T) { if err != nil || string(got) != `{"taken":true}` { t.Fatalf("the work was not handed to the bundle: %s %v", got, err) } + // Design 47 G §576, §518: a bundle reads the runner's capability words with mesh/capabilities, then runs its put session + // through the runner's put-chunk, and the sessions of its channel are dropped when the channel closes. + t.Run("ABundleReadsTheCapabilityWordsRunsAPutSessionAndItsChannelsClosingDropsThem", func(t *testing.T) { + dir := t.TempDir() + entry := filepath.Join(dir, "bundle") + if err := os.WriteFile(entry, []byte(puttingBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &sessionBus{dropped: make(chan struct{})} + l, err := Start("putter", entry, os.Environ(), b, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + select { + case <-b.dropped: + case <-time.After(5 * time.Second): + t.Fatal("the bundle's sessions were not dropped when its channel closed") + } + b.mu.Lock() + defer b.mu.Unlock() + if len(b.chunks) != 1 || !strings.HasPrefix(b.chunks[0], "mesh/objects.put-chunk ") || !strings.Contains(b.chunks[0], `"seq":1`) { + t.Fatalf("the runner took %v", b.chunks) + } + }) + // Design 47 G §559: `mesh/publish` with `ack` is answered with its sequence once the bus acknowledged it, or `{error}`; + // `mesh/subscribe` with `deliver: new` makes a subscription for new messages that ends with the bundle's channel. + t.Run("APublishWithAckIsAnsweredWithItsSequenceAndASubscriptionForNewMessagesEndsWithTheChannel", func(t *testing.T) { + dir := t.TempDir() + entry, out := filepath.Join(dir, "bundle"), filepath.Join(dir, "out") + if err := os.WriteFile(entry, []byte(findingBundle), 0o755); err != nil { + t.Fatal(err) + } + b := &findingBus{stopped: make(chan struct{})} + l, err := Start("writer", entry, append(os.Environ(), "FINDING_OUT="+out), b, func(string, ...any) {}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(l.Stop) + select { + case <-b.stopped: + case <-time.After(5 * time.Second): + t.Fatal("the subscription for new messages was not ended with the bundle's channel") + } + got, _ := os.ReadFile(out) + for _, want := range []string{`"sequence":42`, `"error":"the bus refused it"`} { + if !strings.Contains(string(got), want) { + t.Errorf("the bundle was answered %s, want it to hold %s", got, want) + } + } + b.mu.Lock() + defer b.mu.Unlock() + if b.pattern != "mesh-conditions.findings-asked" { + t.Errorf("the subscription was made for %q", b.pattern) + } + if len(b.published) != 0 { + t.Errorf("a publish with ack went the unacknowledged way: %v", b.published) + } + }) } // A bundle that pulls one ask from its worker when it is listed. diff --git a/node-tools/internal/launch/sessions_test.go b/node-tools/internal/launch/sessions_test.go index f46a5a2..e2d2c93 100644 --- a/node-tools/internal/launch/sessions_test.go +++ b/node-tools/internal/launch/sessions_test.go @@ -3,12 +3,8 @@ package launch import ( "encoding/json" "errors" - "os" - "path/filepath" "strings" "sync" - "testing" - "time" ) // puttingBundle asks the capability words, and once they are answered with put-sessions sends the one chunk of a put @@ -42,32 +38,6 @@ func (b *sessionBus) BusState(method string, params json.RawMessage) (json.RawMe } func (b *sessionBus) DropSessions() { close(b.dropped) } -// Design 47 G §576, §518: a bundle reads the runner's capability words with mesh/capabilities, then runs its put session -// through the runner's put-chunk, and the sessions of its channel are dropped when the channel closes. -func TestABundleReadsTheCapabilityWordsRunsAPutSessionAndItsChannelsClosingDropsThem(t *testing.T) { - dir := t.TempDir() - entry := filepath.Join(dir, "bundle") - if err := os.WriteFile(entry, []byte(puttingBundle), 0o755); err != nil { - t.Fatal(err) - } - b := &sessionBus{dropped: make(chan struct{})} - l, err := Start("putter", entry, os.Environ(), b, func(string, ...any) {}) - if err != nil { - t.Fatal(err) - } - t.Cleanup(l.Stop) - select { - case <-b.dropped: - case <-time.After(5 * time.Second): - t.Fatal("the bundle's sessions were not dropped when its channel closed") - } - b.mu.Lock() - defer b.mu.Unlock() - if len(b.chunks) != 1 || !strings.HasPrefix(b.chunks[0], "mesh/objects.put-chunk ") || !strings.Contains(b.chunks[0], `"seq":1`) { - t.Fatalf("the runner took %v", b.chunks) - } -} - // findingBundle asks, as mesh-sdk's finding writer does, a publish with ack that the bus takes, one it refuses, and a // subscription for new messages; each answer it gets is written to $FINDING_OUT, and it ends after the subscription's. const findingBundle = `#!/bin/sh @@ -106,38 +76,3 @@ func (b *findingBus) SubscribeNew(pattern string, _ func(json.RawMessage) error) b.mu.Unlock() return func() { close(b.stopped) }, nil } - -// Design 47 G §559: `mesh/publish` with `ack` is answered with its sequence once the bus acknowledged it, or `{error}`; -// `mesh/subscribe` with `deliver: new` makes a subscription for new messages that ends with the bundle's channel. -func TestAPublishWithAckIsAnsweredWithItsSequenceAndASubscriptionForNewMessagesEndsWithTheChannel(t *testing.T) { - dir := t.TempDir() - entry, out := filepath.Join(dir, "bundle"), filepath.Join(dir, "out") - if err := os.WriteFile(entry, []byte(findingBundle), 0o755); err != nil { - t.Fatal(err) - } - b := &findingBus{stopped: make(chan struct{})} - l, err := Start("writer", entry, append(os.Environ(), "FINDING_OUT="+out), b, func(string, ...any) {}) - if err != nil { - t.Fatal(err) - } - t.Cleanup(l.Stop) - select { - case <-b.stopped: - case <-time.After(5 * time.Second): - t.Fatal("the subscription for new messages was not ended with the bundle's channel") - } - got, _ := os.ReadFile(out) - for _, want := range []string{`"sequence":42`, `"error":"the bus refused it"`} { - if !strings.Contains(string(got), want) { - t.Errorf("the bundle was answered %s, want it to hold %s", got, want) - } - } - b.mu.Lock() - defer b.mu.Unlock() - if b.pattern != "mesh-conditions.findings-asked" { - t.Errorf("the subscription was made for %q", b.pattern) - } - if len(b.published) != 0 { - t.Errorf("a publish with ack went the unacknowledged way: %v", b.published) - } -} diff --git a/node-tools/internal/ledger/ledger_test.go b/node-tools/internal/ledger/ledger_test.go index cbc315d..04b7c4b 100644 --- a/node-tools/internal/ledger/ledger_test.go +++ b/node-tools/internal/ledger/ledger_test.go @@ -99,51 +99,50 @@ func TestApplyAppendsDerivesAndSnapshots(t *testing.T) { c.run(t, l, f) }) } -} + // design 47 G §587: a ledger's fill is its history's against the ledger's cap, a store that cannot say it is an error naming + // the ledger; a removal takes an instance's operations, snapshot, state and index entries and answers what that was, an + // unknown instance answers none, and nothing of another instance goes. + t.Run("TheLedgerSaysItsFillAndRemovesAnInstanceWhole", func(t *testing.T) { + l, f := newLedger() + f.history.MaxBytes = 1 << 20 + for _, n := range []string{"a", "b"} { + applied(t, l, "decision/0290", "note", `{"says":"`+n+`"}`, owner) + } + applied(t, l, "decision/0291", "note", `{"says":"other"}`, owner) + entry, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0290", Sequence: 2}) + other, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0291", Sequence: 1}) + _ = f.index.Put("t.decision.0290", entry) + _ = f.index.Put("t.decision.0291", other) -// design 47 G §587: a ledger's fill is its history's against the ledger's cap, a store that cannot say it is an error naming -// the ledger; a removal takes an instance's operations, snapshot, state and index entries and answers what that was, an -// unknown instance answers none, and nothing of another instance goes. -func TestTheLedgerSaysItsFillAndRemovesAnInstanceWhole(t *testing.T) { - l, f := newLedger() - f.history.MaxBytes = 1 << 20 - for _, n := range []string{"a", "b"} { - applied(t, l, "decision/0290", "note", `{"says":"`+n+`"}`, owner) - } - applied(t, l, "decision/0291", "note", `{"says":"other"}`, owner) - entry, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0290", Sequence: 2}) - other, _ := json.Marshal(buscalls.IndexEntry{Instance: "decision/0291", Sequence: 1}) - _ = f.index.Put("t.decision.0290", entry) - _ = f.index.Put("t.decision.0291", other) - - fill, err := l.Fill(buscalls.LedgerRequest{Ledger: "records"}) - if err != nil || fill.Count != 3 || fill.Bytes == 0 || fill.MaxBytes != 1<<20 { - t.Fatalf("the fill is %+v (%v), want 3 messages against 1 MiB", fill, err) - } - got, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}) - if err != nil || got.Instance != "decision/0290" || got.Operations != 2 || got.Bytes == 0 { - t.Fatalf("the removal answered %+v (%v), want 2 operations and their bytes", got, err) - } - if kept, _ := f.history.Read("decision.0290", 1, 0); len(kept) != 0 { - t.Errorf("operations are left: %d", len(kept)) - } - if raw, _, _ := f.state.Get("decision.0290"); raw != nil { - t.Errorf("the state is left: %s", raw) - } - if raw, _, _ := f.snapshots.Get("decision.0290"); raw != nil { - t.Errorf("the snapshot is left: %s", raw) - } - if held, _ := f.index.Entries("", 0); len(held) != 1 || held[0].Key != "t.decision.0291" { - t.Errorf("the index holds %v, want the other instance's entry alone", held) - } - if kept, _ := f.history.Read("decision.0291", 1, 0); len(kept) != 1 { - t.Errorf("another instance lost operations: %d left", len(kept)) - } - if again, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}); err != nil || again.Operations != 0 || again.Bytes != 0 { - t.Errorf("a removal said twice answered %+v (%v), want none", again, err) - } - f.history.FillFails = true - if _, err := l.Fill(buscalls.LedgerRequest{}); err == nil || !strings.Contains(err.Error(), "ledger records cannot say the fill") { - t.Errorf("a history that cannot say its fill answered %v", err) - } + fill, err := l.Fill(buscalls.LedgerRequest{Ledger: "records"}) + if err != nil || fill.Count != 3 || fill.Bytes == 0 || fill.MaxBytes != 1<<20 { + t.Fatalf("the fill is %+v (%v), want 3 messages against 1 MiB", fill, err) + } + got, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}) + if err != nil || got.Instance != "decision/0290" || got.Operations != 2 || got.Bytes == 0 { + t.Fatalf("the removal answered %+v (%v), want 2 operations and their bytes", got, err) + } + if kept, _ := f.history.Read("decision.0290", 1, 0); len(kept) != 0 { + t.Errorf("operations are left: %d", len(kept)) + } + if raw, _, _ := f.state.Get("decision.0290"); raw != nil { + t.Errorf("the state is left: %s", raw) + } + if raw, _, _ := f.snapshots.Get("decision.0290"); raw != nil { + t.Errorf("the snapshot is left: %s", raw) + } + if held, _ := f.index.Entries("", 0); len(held) != 1 || held[0].Key != "t.decision.0291" { + t.Errorf("the index holds %v, want the other instance's entry alone", held) + } + if kept, _ := f.history.Read("decision.0291", 1, 0); len(kept) != 1 { + t.Errorf("another instance lost operations: %d left", len(kept)) + } + if again, err := l.Remove(buscalls.LedgerRemoveRequest{Ledger: "records", Instance: "decision/0290"}); err != nil || again.Operations != 0 || again.Bytes != 0 { + t.Errorf("a removal said twice answered %+v (%v), want none", again, err) + } + f.history.FillFails = true + if _, err := l.Fill(buscalls.LedgerRequest{}); err == nil || !strings.Contains(err.Error(), "ledger records cannot say the fill") { + t.Errorf("a history that cannot say its fill answered %v", err) + } + }) } diff --git a/node-tools/internal/runtime/announce_test.go b/node-tools/internal/runtime/announce_test.go index b942524..43556e6 100644 --- a/node-tools/internal/runtime/announce_test.go +++ b/node-tools/internal/runtime/announce_test.go @@ -2,7 +2,6 @@ package runtime import ( "encoding/json" - "git.novox.be/novox/mesh-sdk/go/buscalls" "os" "slices" "sort" @@ -10,6 +9,8 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/micro" @@ -150,57 +151,55 @@ func TestTheRunnerStatesTheCapabilitiesOfItsCallsInItsPing(t *testing.T) { if ping.Metadata[announce.Fingerprint] != "sha256:runner" { t.Errorf("the runner states its fingerprint as %q", ping.Metadata[announce.Fingerprint]) } -} - -// Design 47 G §462, §488, §622: each call is stated under the capability word of its own family, the words mesh-sdk's buscalls -// names: the read of a sent object under sent-objects-calls (no longer under bus-state-calls), the ledger's fill and removal -// under ledger-upkeep-calls, the others where they were. -func TestEachCallIsStatedUnderTheWordOfItsFamily(t *testing.T) { - for method, word := range map[string]string{ - buscalls.ObjectsSentGet: buscalls.SentObjectsCalls, - buscalls.LedgerFill: buscalls.LedgerUpkeepCalls, - buscalls.LedgerRemove: buscalls.LedgerUpkeepCalls, - buscalls.LedgerApply: buscalls.LedgerCalls, - buscalls.ObjectsPut: buscalls.BusStateCalls, - } { - var in []string - for c, methods := range callFamilies { - if slices.Contains(methods, method) { - in = append(in, c) + // Design 47 G §462, §488, §622: each call is stated under the capability word of its own family, the words mesh-sdk's buscalls + // names: the read of a sent object under sent-objects-calls (no longer under bus-state-calls), the ledger's fill and removal + // under ledger-upkeep-calls, the others where they were. + t.Run("EachCallIsStatedUnderTheWordOfItsFamily", func(t *testing.T) { + for method, word := range map[string]string{ + buscalls.ObjectsSentGet: buscalls.SentObjectsCalls, + buscalls.LedgerFill: buscalls.LedgerUpkeepCalls, + buscalls.LedgerRemove: buscalls.LedgerUpkeepCalls, + buscalls.LedgerApply: buscalls.LedgerCalls, + buscalls.ObjectsPut: buscalls.BusStateCalls, + } { + var in []string + for c, methods := range callFamilies { + if slices.Contains(methods, method) { + in = append(in, c) + } + } + if len(in) != 1 || in[0] != word { + t.Errorf("%s is stated under %v, want %s alone", method, in, word) } } - if len(in) != 1 || in[0] != word { - t.Errorf("%s is stated under %v, want %s alone", method, in, word) + }) + // Design 47 G §545: the words the tool runner states are exactly its manifest's artifact `states`, so the delivery plan's edge + // of §605 runs to it for each word a module's artifact `calls`. + t.Run("TheWordsTheRunnerStatesAreItsManifestsStates", func(t *testing.T) { + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) } - } -} - -// Design 47 G §545: the words the tool runner states are exactly its manifest's artifact `states`, so the delivery plan's edge -// of §605 runs to it for each word a module's artifact `calls`. -func TestTheWordsTheRunnerStatesAreItsManifestsStates(t *testing.T) { - raw, err := os.ReadFile("../../module.json") - if err != nil { - t.Fatal(err) - } - var m struct { - Build struct { - Artifacts []struct { - Name string `json:"name"` - States []string `json:"states"` - } `json:"artifacts"` - } `json:"build"` - } - if err := json.Unmarshal(raw, &m); err != nil { - t.Fatal(err) - } - var declared []string - for _, a := range m.Build.Artifacts { - if a.Name == "runtime" { - declared = a.States + var m struct { + Build struct { + Artifacts []struct { + Name string `json:"name"` + States []string `json:"states"` + } `json:"artifacts"` + } `json:"build"` } - } - slices.Sort(declared) - if got := CallCapabilities(); !slices.Equal(got, declared) { - t.Errorf("the runner states %v, its manifest's states are %v", got, declared) - } + if err := json.Unmarshal(raw, &m); err != nil { + t.Fatal(err) + } + var declared []string + for _, a := range m.Build.Artifacts { + if a.Name == "runtime" { + declared = a.States + } + } + slices.Sort(declared) + if got := CallCapabilities(); !slices.Equal(got, declared) { + t.Errorf("the runner states %v, its manifest's states are %v", got, declared) + } + }) } diff --git a/node-tools/internal/runtime/caller_test.go b/node-tools/internal/runtime/caller_test.go index 46c4c95..d44a0d2 100644 --- a/node-tools/internal/runtime/caller_test.go +++ b/node-tools/internal/runtime/caller_test.go @@ -6,6 +6,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/approval" "git.novox.be/novox/mesh-sdk/go/seatdef" "github.com/nats-io/nats.go" @@ -121,9 +122,76 @@ func TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone(t *testing.T) { {map[string]seatdef.Definition{"node-desk": {Seat: "node-desk", Verbs: []seatdef.Verb{{Name: "who", Class: seatdef.ClassAct}}}}, ""}, } { defs := c.defs - err := (&consumers{definitions: func() map[string]seatdef.Definition { return defs }}).admitted("node-desk", "who", "shanks.node-tools") + err := (&consumers{definitions: func() map[string]seatdef.Definition { return defs }}).admitted("node-desk", "who", "workstation.node-tools") if c.says == "" && err != nil || c.says != "" && (err == nil || !strings.Contains(err.Error(), c.says)) { t.Errorf("a proven caller with the definitions %v: %v, want %q", c.defs, err, c.says) } } + // Design 47 G A-249: a bundle's seat call is received by the runner in chunks only for a verb whose definition says `chunks`, + // of any machine: a verb a node holds is asked of the node that ran the ask (A-254). + t.Run("OnlyAChunkedSeatVerbIsAskedInChunks", func(t *testing.T) { + c := &consumers{definitions: func() map[string]seatdef.Definition { + return map[string]seatdef.Definition{"mesh-catalog": {Verbs: []seatdef.Verb{{Name: "input", Chunks: true}, {Name: "entry"}}}} + }} + for key, want := range map[string]bool{ + "seat:mesh-catalog.input": true, + "seat:mesh-catalog.entry": false, + "seat:mesh-catalog.input@home-server": true, + "seat:mesh-catalog.entry@home-server": false, + "seat:mesh-unread.input": false, + "mesh-delivery-indexer.indexer": false, + } { + if got := c.chunked(key); got != want { + t.Errorf("chunked(%s) = %v, want %v", key, got, want) + } + } + if (&consumers{}).chunked("seat:mesh-catalog.input") { + t.Error("a runner that holds no seat definitions asks no chunks") + } + }) + // **The tool runner answers a held seat's marked verb by caller and serves the seat's `approvals`** (design 47 G §677; lead + // answers A-265, A-271), with zsh's node-login-shell `execute` as the case: the marked verb reaches the gate with the caller + // its subject names, and `approvals` is answered from the runner's own record of the calls its gate holds, with the + // arguments for the call's own caller only, though no bundle names the verb. + t.Run("TheRunnerServesApprovalsForTheSeatsItsBundlesHold", func(t *testing.T) { + mesh := mt.New(t) + mesh.Issue(t, mt.MembershipOf("zsh", "anchor", true, map[string][]string{"node-login-shell": {"execute", seatdef.ApprovalsVerb}})) + nodeTools := connect(t, "node-tools", "anchor") + gate := &gateHolding{} + defined := WithSeatDefinitions(func() map[string]seatdef.Definition { + return map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", Verbs: []seatdef.Verb{ + {Name: "execute", NeedsApproval: true}, {Name: seatdef.ApprovalsVerb}}}} + }) + stop, err := Run(nodeTools, []Served{{"zsh", []string{mt.Fixture("login-shell-execute.py")}}}, nil, func(string, ...any) {}, defined, WithGate(gate)) + if err != nil { + t.Fatal(err) + } + defer stop() + asker, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "workstation", User: "workstation.node-tools", Password: "unused"}) + if err != nil { + t.Fatal(err) + } + asker.Logf = func(string, ...any) {} + defer asker.Close() + + mt.Until(t, func() error { + _, err := asker.Ask("seat:node-login-shell.execute@anchor", map[string]any{"command": "ls"}, "") + return err + }) + if len(gate.calls) != 1 || gate.calls[0].Seat != "node-login-shell" || gate.calls[0].Name != "execute" || gate.calls[0].Caller != "workstation.node-tools" { + t.Fatalf("the marked verb reached the gate as %+v, want the seat's execute by workstation.node-tools", gate.calls) + } + var got approval.Approvals + mt.Until(t, func() error { + answered, err := asker.Ask("seat:node-login-shell.approvals@anchor", map[string]any{}, "") + if err == nil { + err = json.Unmarshal(answered.Result, &got) + } + return err + }) + if len(got.Calls) != 1 || got.Calls[0].Caller != "workstation.node-tools" || got.Calls[0].Verb != "node-login-shell.execute" || + string(got.Calls[0].Arguments) != `{"command":"ls"}` { + t.Fatalf("approvals answered %+v, want the one call the gate holds, with its caller's arguments", got) + } + }) } diff --git a/node-tools/internal/runtime/chunked_test.go b/node-tools/internal/runtime/chunked_test.go deleted file mode 100644 index 34da586..0000000 --- a/node-tools/internal/runtime/chunked_test.go +++ /dev/null @@ -1,30 +0,0 @@ -package runtime - -import ( - "testing" - - "git.novox.be/novox/mesh-sdk/go/seatdef" -) - -// Design 47 G A-249: a bundle's seat call is received by the runner in chunks only for a verb whose definition says `chunks`, -// of any machine: a verb a node holds is asked of the node that ran the ask (A-254). -func TestOnlyAChunkedSeatVerbIsAskedInChunks(t *testing.T) { - c := &consumers{definitions: func() map[string]seatdef.Definition { - return map[string]seatdef.Definition{"mesh-catalog": {Verbs: []seatdef.Verb{{Name: "input", Chunks: true}, {Name: "entry"}}}} - }} - for key, want := range map[string]bool{ - "seat:mesh-catalog.input": true, - "seat:mesh-catalog.entry": false, - "seat:mesh-catalog.input@ace": true, - "seat:mesh-catalog.entry@ace": false, - "seat:mesh-unread.input": false, - "mesh-delivery-indexer.indexer": false, - } { - if got := c.chunked(key); got != want { - t.Errorf("chunked(%s) = %v, want %v", key, got, want) - } - } - if (&consumers{}).chunked("seat:mesh-catalog.input") { - t.Error("a runner that holds no seat definitions asks no chunks") - } -} diff --git a/node-tools/internal/runtime/gate_approvals_test.go b/node-tools/internal/runtime/gate_approvals_test.go index 41d1c8b..7f85aac 100644 --- a/node-tools/internal/runtime/gate_approvals_test.go +++ b/node-tools/internal/runtime/gate_approvals_test.go @@ -1,14 +1,7 @@ package runtime import ( - "encoding/json" - "testing" - "git.novox.be/novox/mesh-sdk/go/approval" - "git.novox.be/novox/mesh-sdk/go/seatdef" - - "github.com/novox/mesh-tools/node-tools/internal/bus" - mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" ) // gateHolding is the runner's gate as far as the runtime reaches it: it records the calls it is run for and answers @@ -31,49 +24,3 @@ func (g *gateHolding) Approvals(caller string) (approval.Approvals, error) { } return out, nil } - -// **The tool runner answers a held seat's marked verb by caller and serves the seat's `approvals`** (design 47 G §677; lead -// answers A-265, A-271), with zsh's node-login-shell `execute` as the case: the marked verb reaches the gate with the caller -// its subject names, and `approvals` is answered from the runner's own record of the calls its gate holds, with the -// arguments for the call's own caller only, though no bundle names the verb. -func TestTheRunnerServesApprovalsForTheSeatsItsBundlesHold(t *testing.T) { - mesh := mt.New(t) - mesh.Issue(t, mt.MembershipOf("zsh", "anchor", true, map[string][]string{"node-login-shell": {"execute", seatdef.ApprovalsVerb}})) - nodeTools := connect(t, "node-tools", "anchor") - gate := &gateHolding{} - defined := WithSeatDefinitions(func() map[string]seatdef.Definition { - return map[string]seatdef.Definition{"node-login-shell": {Seat: "node-login-shell", Verbs: []seatdef.Verb{ - {Name: "execute", NeedsApproval: true}, {Name: seatdef.ApprovalsVerb}}}} - }) - stop, err := Run(nodeTools, []Served{{"zsh", []string{mt.Fixture("login-shell-execute.py")}}}, nil, func(string, ...any) {}, defined, WithGate(gate)) - if err != nil { - t.Fatal(err) - } - defer stop() - asker, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "shanks", User: "shanks.node-tools", Password: "unused"}) - if err != nil { - t.Fatal(err) - } - asker.Logf = func(string, ...any) {} - defer asker.Close() - - mt.Until(t, func() error { - _, err := asker.Ask("seat:node-login-shell.execute@anchor", map[string]any{"command": "ls"}, "") - return err - }) - if len(gate.calls) != 1 || gate.calls[0].Seat != "node-login-shell" || gate.calls[0].Name != "execute" || gate.calls[0].Caller != "shanks.node-tools" { - t.Fatalf("the marked verb reached the gate as %+v, want the seat's execute by shanks.node-tools", gate.calls) - } - var got approval.Approvals - mt.Until(t, func() error { - answered, err := asker.Ask("seat:node-login-shell.approvals@anchor", map[string]any{}, "") - if err == nil { - err = json.Unmarshal(answered.Result, &got) - } - return err - }) - if len(got.Calls) != 1 || got.Calls[0].Caller != "shanks.node-tools" || got.Calls[0].Verb != "node-login-shell.execute" || - string(got.Calls[0].Arguments) != `{"command":"ls"}` { - t.Fatalf("approvals answered %+v, want the one call the gate holds, with its caller's arguments", got) - } -} diff --git a/node-tools/internal/runtime/ledger_integration_test.go b/node-tools/internal/runtime/ledger_integration_test.go index 589afce..5c62586 100644 --- a/node-tools/internal/runtime/ledger_integration_test.go +++ b/node-tools/internal/runtime/ledger_integration_test.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "strings" + "sync" "testing" "git.novox.be/novox/mesh-sdk/go/buscalls" @@ -287,4 +288,70 @@ func TestTheLedgerKeepsItsPromisesOnTheBus(t *testing.T) { } }) } + // Design 47 G §559, A-153: against the real bus a publish with ack answers the sequence the bus gave it, and a subscription + // for new messages hears only what is published after it was made, on the subject mesh-sdk's one rule names for the event; + // a wildcard is refused. + t.Run("AnAckedPublishAnswersItsSequenceAndANewSubscriptionHearsOnlyWhatComesAfter", func(t *testing.T) { + mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx := context.Background() + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEATEVENTS", Subjects: []string{"mesh.seat.mesh-conditions.event.>"}}); err != nil { + t.Fatal(err) + } + conn := connect(t, "node-tools", "anchor") + publish := func(body string) { + t.Helper() + if _, err := js.Publish(ctx, "mesh.seat.mesh-conditions.event.findings-asked", []byte(body)); err != nil { + t.Fatal(err) + } + } + publish(`{"before":true}`) + + var mu sync.Mutex + var heard []string + stop, err := conn.SubscribeNew("mesh-conditions.findings-asked", func(raw json.RawMessage) error { + var env bus.Envelope + _ = json.Unmarshal(raw, &env) + mu.Lock() + heard = append(heard, env.Key+" "+string(env.Body)) + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + publish(`{"after":true}`) + publish(`{"after":2}`) + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(heard) < 2 { + return context.DeadlineExceeded + } + return nil + }) + mu.Lock() + if len(heard) != 2 || heard[0] != `mesh-conditions.findings-asked {"after":true}` { + t.Errorf("heard %v, want the messages after the subscription alone, in order", heard) + } + mu.Unlock() + + for name, want := range map[string]string{"mesh-conditions.*": "[A-Za-z0-9_-]"} { + if _, err := conn.SubscribeNew(name, func(json.RawMessage) error { return nil }); err == nil || !strings.Contains(err.Error(), want) { + t.Errorf("a subscription for %s answered %v, want one naming %q", name, err, want) + } + } + + first, err := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + second, err2 := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) + if err != nil || err2 != nil || first == 0 || second != first+1 { + t.Errorf("acknowledged publishes answered %d (%v) and %d (%v), want consecutive sequences", first, err, second, err2) + } + }) } diff --git a/node-tools/internal/runtime/own_account_test.go b/node-tools/internal/runtime/own_account_test.go index a44d681..ff678e6 100644 --- a/node-tools/internal/runtime/own_account_test.go +++ b/node-tools/internal/runtime/own_account_test.go @@ -36,26 +36,25 @@ func TestNoBusWordReachesABundle(t *testing.T) { t.Errorf("%s did not reach the bundle", want) } } -} - -// design 47 G §618, §519: the runtime's own fingerprint reaches no bundle; the fingerprint a module's composed words hold does. -func TestTheRuntimesOwnFingerprintReachesNoBundle(t *testing.T) { - base := []string{"MESH_FINGERPRINT=runner-build", "PATH=/usr/bin"} - for name, c := range map[string]struct { - given map[string]string - want string - }{ - "composed with none sees none": {nil, ""}, - "composed with one sees its own": {map[string]string{"MESH_FINGERPRINT": "module-build"}, "MESH_FINGERPRINT=module-build"}, - } { - env := strings.Join(BundleEnv(base, c.given, "telegram", "anchor"), "\n") - if strings.Contains(env, "runner-build") { - t.Errorf("%s: the runtime's own fingerprint reached the bundle: %s", name, env) + // design 47 G §618, §519: the runtime's own fingerprint reaches no bundle; the fingerprint a module's composed words hold does. + t.Run("TheRuntimesOwnFingerprintReachesNoBundle", func(t *testing.T) { + base := []string{"MESH_FINGERPRINT=runner-build", "PATH=/usr/bin"} + for name, c := range map[string]struct { + given map[string]string + want string + }{ + "composed with none sees none": {nil, ""}, + "composed with one sees its own": {map[string]string{"MESH_FINGERPRINT": "module-build"}, "MESH_FINGERPRINT=module-build"}, + } { + env := strings.Join(BundleEnv(base, c.given, "telegram", "anchor"), "\n") + if strings.Contains(env, "runner-build") { + t.Errorf("%s: the runtime's own fingerprint reached the bundle: %s", name, env) + } + if c.want == "" && strings.Contains(env, "MESH_FINGERPRINT") || !strings.Contains(env, c.want) { + t.Errorf("%s: the environment is %s", name, env) + } } - if c.want == "" && strings.Contains(env, "MESH_FINGERPRINT") || !strings.Contains(env, c.want) { - t.Errorf("%s: the environment is %s", name, env) - } - } + }) } // A tool call reaches only a tool's subject: whatever key a caller names, it is never a seat's event, accept diff --git a/node-tools/internal/runtime/sessions_test.go b/node-tools/internal/runtime/sessions_test.go index 5924c97..4d67f94 100644 --- a/node-tools/internal/runtime/sessions_test.go +++ b/node-tools/internal/runtime/sessions_test.go @@ -64,66 +64,6 @@ func asked(t *testing.T, b *moduleBus, method string, req any, into any) { } } -// Design 47 G §518, A-21, A-25: a put session writes each chunk as it is taken, carries the description of its first chunk, -// and answers the put's own info for the last; a chunk out of order fails the put naming both sequences and leaves none. -func TestAPutSessionSealsOnTheLastChunkAndDropsOnADisorder(t *testing.T) { - f := &fakeObjects{put: map[string][]byte{}} - b := aSession(t, f) - whole := bytes.Repeat([]byte("x"), 2*buscalls.ChunkBytes+5) - var taken buscalls.ObjectsChunkTaken - for seq, at := uint32(1), 0; ; seq++ { - end := min(at+buscalls.ChunkBytes, len(whole)) - req := buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c1", Seq: seq, Last: end == len(whole), Data: whole[at:end]} - if seq == 1 { - req.Description = "what it is" - } - if req.Last { - var info buscalls.ObjectInfo - asked(t, b, buscalls.ObjectsPutChunk, req, &info) - if info.Size != uint64(len(whole)) || info.Description != "what it is" { - t.Fatalf("the put answered %+v", info) - } - break - } - asked(t, b, buscalls.ObjectsPutChunk, req, &taken) - if taken.Seq != seq || taken.Error != "" { - t.Fatalf("chunk %d was answered %+v", seq, taken) - } - at = end - } - if !bytes.Equal(f.put["a"], whole) { - t.Fatalf("the store holds %d bytes, want %d", len(f.put["a"]), len(whole)) - } - - asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 1, Data: []byte("one")}, &taken) - asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 3, Data: []byte("three")}, &taken) - if !strings.Contains(taken.Error, "chunk 3") || !strings.Contains(taken.Error, "chunk 2") { - t.Fatalf("a disorder was answered %+v", taken) - } - if _, ok := f.put["b"]; ok || len(f.deleted) != 1 || f.deleted[0] != "b" { - t.Fatalf("a dropped put left %v, deleted %v", f.put, f.deleted) - } - asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 2, Data: []byte("two")}, &taken) - if taken.Error == "" { - t.Fatal("a chunk of a dropped put was taken") - } -} - -// §518 "The bound": a put not sealed within the bound is dropped, its object deleted. -func TestAPutNotSealedWithinTheBoundIsDropped(t *testing.T) { - was := SessionBound - SessionBound = time.Millisecond - t.Cleanup(func() { SessionBound = was }) - f := &fakeObjects{put: map[string][]byte{}, ctxEnded: make(chan struct{})} - b := aSession(t, f) - var taken buscalls.ObjectsChunkTaken - asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c", Seq: 1, Data: []byte("x")}, &taken) - <-f.ctxEnded - if len(f.deleted) != 1 { - t.Fatalf("the unsealed object was left: %v", f.deleted) - } -} - // §490, §518, §620: a value over one chunk is got as a session (the chunked answer, then each chunk by its call and sequence, // the whole checked against the digest); a sent object over one chunk the same, with `chunked` true. func TestAGetOverOneChunkRunsAsASession(t *testing.T) { @@ -192,4 +132,62 @@ func TestAGetOverOneChunkRunsAsASession(t *testing.T) { if string(small.Data) != "small" { t.Fatalf("a small value was answered %+v", small) } + // Design 47 G §518, A-21, A-25: a put session writes each chunk as it is taken, carries the description of its first chunk, + // and answers the put's own info for the last; a chunk out of order fails the put naming both sequences and leaves none. + t.Run("APutSessionSealsOnTheLastChunkAndDropsOnADisorder", func(t *testing.T) { + f := &fakeObjects{put: map[string][]byte{}} + b := aSession(t, f) + whole := bytes.Repeat([]byte("x"), 2*buscalls.ChunkBytes+5) + var taken buscalls.ObjectsChunkTaken + for seq, at := uint32(1), 0; ; seq++ { + end := min(at+buscalls.ChunkBytes, len(whole)) + req := buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c1", Seq: seq, Last: end == len(whole), Data: whole[at:end]} + if seq == 1 { + req.Description = "what it is" + } + if req.Last { + var info buscalls.ObjectInfo + asked(t, b, buscalls.ObjectsPutChunk, req, &info) + if info.Size != uint64(len(whole)) || info.Description != "what it is" { + t.Fatalf("the put answered %+v", info) + } + break + } + asked(t, b, buscalls.ObjectsPutChunk, req, &taken) + if taken.Seq != seq || taken.Error != "" { + t.Fatalf("chunk %d was answered %+v", seq, taken) + } + at = end + } + if !bytes.Equal(f.put["a"], whole) { + t.Fatalf("the store holds %d bytes, want %d", len(f.put["a"]), len(whole)) + } + + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 1, Data: []byte("one")}, &taken) + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 3, Data: []byte("three")}, &taken) + if !strings.Contains(taken.Error, "chunk 3") || !strings.Contains(taken.Error, "chunk 2") { + t.Fatalf("a disorder was answered %+v", taken) + } + if _, ok := f.put["b"]; ok || len(f.deleted) != 1 || f.deleted[0] != "b" { + t.Fatalf("a dropped put left %v, deleted %v", f.put, f.deleted) + } + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "b", Call: "c2", Seq: 2, Data: []byte("two")}, &taken) + if taken.Error == "" { + t.Fatal("a chunk of a dropped put was taken") + } + }) + // §518 "The bound": a put not sealed within the bound is dropped, its object deleted. + t.Run("APutNotSealedWithinTheBoundIsDropped", func(t *testing.T) { + was := SessionBound + SessionBound = time.Millisecond + t.Cleanup(func() { SessionBound = was }) + f := &fakeObjects{put: map[string][]byte{}, ctxEnded: make(chan struct{})} + b := aSession(t, f) + var taken buscalls.ObjectsChunkTaken + asked(t, b, buscalls.ObjectsPutChunk, buscalls.ObjectsPutChunkRequest{Objects: "files", Name: "a", Call: "c", Seq: 1, Data: []byte("x")}, &taken) + <-f.ctxEnded + if len(f.deleted) != 1 { + t.Fatalf("the unsealed object was left: %v", f.deleted) + } + }) } diff --git a/node-tools/internal/runtime/subscribenew_integration_test.go b/node-tools/internal/runtime/subscribenew_integration_test.go deleted file mode 100644 index 9dc1d14..0000000 --- a/node-tools/internal/runtime/subscribenew_integration_test.go +++ /dev/null @@ -1,84 +0,0 @@ -//go:build integration - -package runtime - -import ( - "context" - "encoding/json" - "strings" - "sync" - "testing" - - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - - "github.com/novox/mesh-tools/node-tools/internal/bus" - mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" -) - -// Design 47 G §559, A-153: against the real bus a publish with ack answers the sequence the bus gave it, and a subscription -// for new messages hears only what is published after it was made, on the subject mesh-sdk's one rule names for the event; -// a wildcard is refused. -func TestAnAckedPublishAnswersItsSequenceAndANewSubscriptionHearsOnlyWhatComesAfter(t *testing.T) { - mt.New(t) - nc, err := nats.Connect(mt.URL(t)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(nc.Close) - js, _ := jetstream.New(nc) - ctx := context.Background() - if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEATEVENTS", Subjects: []string{"mesh.seat.mesh-conditions.event.>"}}); err != nil { - t.Fatal(err) - } - conn := connect(t, "node-tools", "anchor") - publish := func(body string) { - t.Helper() - if _, err := js.Publish(ctx, "mesh.seat.mesh-conditions.event.findings-asked", []byte(body)); err != nil { - t.Fatal(err) - } - } - publish(`{"before":true}`) - - var mu sync.Mutex - var heard []string - stop, err := conn.SubscribeNew("mesh-conditions.findings-asked", func(raw json.RawMessage) error { - var env bus.Envelope - _ = json.Unmarshal(raw, &env) - mu.Lock() - heard = append(heard, env.Key+" "+string(env.Body)) - mu.Unlock() - return nil - }) - if err != nil { - t.Fatal(err) - } - defer stop() - publish(`{"after":true}`) - publish(`{"after":2}`) - mt.Until(t, func() error { - mu.Lock() - defer mu.Unlock() - if len(heard) < 2 { - return context.DeadlineExceeded - } - return nil - }) - mu.Lock() - if len(heard) != 2 || heard[0] != `mesh-conditions.findings-asked {"after":true}` { - t.Errorf("heard %v, want the messages after the subscription alone, in order", heard) - } - mu.Unlock() - - for name, want := range map[string]string{"mesh-conditions.*": "[A-Za-z0-9_-]"} { - if _, err := conn.SubscribeNew(name, func(json.RawMessage) error { return nil }); err == nil || !strings.Contains(err.Error(), want) { - t.Errorf("a subscription for %s answered %v, want one naming %q", name, err, want) - } - } - - first, err := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) - second, err2 := conn.PublishAsAcked("node-tools", bus.Envelope{Key: "finding-written", Body: json.RawMessage(`{}`)}) - if err != nil || err2 != nil || first == 0 || second != first+1 { - t.Errorf("acknowledged publishes answered %d (%v) and %d (%v), want consecutive sequences", first, err, second, err2) - } -}