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
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
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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")
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user