diff --git a/go/busstate.go b/go/busstate.go new file mode 100644 index 0000000..a77133f --- /dev/null +++ b/go/busstate.go @@ -0,0 +1,31 @@ +package stdio + +import ( + "encoding/json" + "fmt" + "strings" + "time" +) + +// busStateCalls are the method prefixes of the calls a bundle makes on its own bus state through the runtime (novox/hq +// ADR 0595 §9, design 25 § 12.5): their shapes and names are package buscalls's, and package busstate the typed way +// to make them. +var busStateCalls = []string{"mesh/objects.", "mesh/stream.", "mesh/workqueue.", "mesh/ledger."} + +// BusState asks the runtime one call on this bundle's own bus state, an entry its module asks for in `requires`, named +// by its local name in params; the runtime serves it from the module's own bindings only. A method that is none of +// those calls is refused here, before the runtime is asked. +func BusState(method string, params any) (json.RawMessage, error) { + known := false + for _, p := range busStateCalls { + known = known || strings.HasPrefix(method, p) + } + if !known { + return nil, fmt.Errorf("%s is no call on a bundle's own bus state (%s)", method, strings.Join(busStateCalls, ", ")) + } + c, err := live() + if err != nil { + return nil, err + } + return c.ask(method, params, 60*time.Second) +} diff --git a/go/busstate/busstate.go b/go/busstate/busstate.go new file mode 100644 index 0000000..d105dc1 --- /dev/null +++ b/go/busstate/busstate.go @@ -0,0 +1,59 @@ +// Package busstate is how a bundle uses its own bus state through the runtime (novox/hq ADR 0595 §9, design 25 § +// 12.5): the object store's put, get, list, delete and fill, each naming an entry by the local name its module asks for +// it by in `requires`. The runtime serves the calls from the module's own bindings, its own entries only. The shapes +// are package buscalls's; this package makes the calls. +package busstate + +import ( + "encoding/json" + "fmt" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// Objects is one object store the module asks for in `requires` (the template objects, ADR 0595 §2), by its local name. +type Objects struct{ local string } + +// ObjectsOf is the object store of that local name. +func ObjectsOf(local string) Objects { return Objects{local: local} } + +func call[T any](method string, params any) (T, error) { + var out T + raw, err := stdio.BusState(method, params) + if err != nil { + return out, err + } + if err := json.Unmarshal(raw, &out); err != nil { + return out, fmt.Errorf("the runtime's answer to %s: %w", method, err) + } + return out, nil +} + +// Put stores one object whole under its name, replacing one of that name, and answers it as the store holds it. +func (o Objects) Put(name string, data []byte, description string) (buscalls.ObjectInfo, error) { + return call[buscalls.ObjectInfo](buscalls.ObjectsPut, buscalls.ObjectsPutRequest{Objects: o.local, Name: name, + Data: data, Description: description}) +} + +// Get is one object, its info and its bytes. +func (o Objects) Get(name string) (buscalls.ObjectData, error) { + return call[buscalls.ObjectData](buscalls.ObjectsGet, buscalls.ObjectsNameRequest{Objects: o.local, Name: name}) +} + +// List is every object the store holds. +func (o Objects) List() ([]buscalls.ObjectInfo, error) { + l, err := call[buscalls.ObjectsListed](buscalls.ObjectsList, buscalls.ObjectsRequest{Objects: o.local}) + return l.Objects, err +} + +// Delete removes one object. +func (o Objects) Delete(name string) error { + _, err := call[buscalls.Done](buscalls.ObjectsDelete, buscalls.ObjectsNameRequest{Objects: o.local, Name: name}) + return err +} + +// Fill is how much the store holds against its cap. +func (o Objects) Fill() (buscalls.Fill, error) { + return call[buscalls.Fill](buscalls.ObjectsInfo, buscalls.ObjectsRequest{Objects: o.local}) +} diff --git a/go/busstate/busstate_test.go b/go/busstate/busstate_test.go new file mode 100644 index 0000000..3d1c365 --- /dev/null +++ b/go/busstate/busstate_test.go @@ -0,0 +1,102 @@ +package busstate + +import ( + "bufio" + "encoding/json" + "io" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" +) + +// A bundle keeps objects in its own object store through the runtime (ADR 0595 §2, §9; A3's request for distribution's +// incoming): each call is the runtime's method of buscalls, naming the store by its local name, and answers as the +// runtime answered; a method that is no bus-state call is refused before the runtime is asked. +func TestABundleKeepsObjectsThroughTheRuntime(t *testing.T) { + t.Setenv(stdio.ServedAs, "distribution") + inR, inW := io.Pipe() + outR, outW := io.Pipe() + type result struct { + what string + err error + got any + } + results := make(chan result, 8) + go func() { + _ = stdio.ServeOn("", []stdio.Tool{{Name: "keep", Run: func(map[string]any) (any, error) { + go func() { + in := ObjectsOf("incoming") + info, err := in.Put("t1.part0", []byte("hello"), "a part") + results <- result{"put", err, info.Digest} + data, err := in.Get("t1.part0") + results <- result{"get", err, string(data.Data)} + list, err := in.List() + results <- result{"list", err, len(list)} + results <- result{"delete", in.Delete("t1.part0"), nil} + fill, err := in.Fill() + results <- result{"fill", err, fill.MaxBytes} + _, err = stdio.BusState("mesh/state.get", nil) + results <- result{"refused", err, nil} + }() + return "started", nil + }}}, inR, outW) + }() + lines := bufio.NewScanner(outR) + next := func() map[string]any { + done := make(chan map[string]any, 1) + go func() { + if lines.Scan() { + var m map[string]any + _ = json.Unmarshal(lines.Bytes(), &m) + done <- m + } + }() + select { + case m := <-done: + return m + case <-time.After(3 * time.Second): + t.Fatal("the bundle said nothing") + return nil + } + } + send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) } + send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "keep"}}) + answers := map[string]any{ + buscalls.ObjectsPut: buscalls.ObjectInfo{Name: "t1.part0", Size: 5, Digest: "SHA-256=x"}, + buscalls.ObjectsGet: buscalls.ObjectData{Info: buscalls.ObjectInfo{Name: "t1.part0"}, Data: []byte("hello")}, + buscalls.ObjectsList: buscalls.ObjectsListed{Objects: []buscalls.ObjectInfo{{Name: "t1.part0"}}}, + buscalls.ObjectsDelete: buscalls.Done{}, + buscalls.ObjectsInfo: buscalls.Fill{Count: 0, MaxBytes: 20 << 30}, + } + asked := map[string]map[string]any{} + for len(asked) < len(answers) { + m := next() + method, _ := m["method"].(string) + if method == "" { + continue // the tool's own answer + } + p, _ := m["params"].(map[string]any) + asked[method] = p + send(map[string]any{"jsonrpc": "2.0", "id": m["id"], "result": answers[method]}) + } + want := map[string]any{"put": "SHA-256=x", "get": "hello", "list": 1, "delete": nil, "fill": uint64(20 << 30)} + for range 5 { + r := <-results + if r.err != nil || r.got != want[r.what] { + t.Errorf("%s answered %v, %v; want %v", r.what, r.got, r.err, want[r.what]) + } + } + if r := <-results; r.err == nil { + t.Error("a call that is not on the bundle's bus state was asked of the runtime") + } + for method, p := range asked { + if p["objects"] != "incoming" { + t.Errorf("%s named the store %v, not its local name", method, p["objects"]) + } + } + if p := asked[buscalls.ObjectsPut]; p["name"] != "t1.part0" || p["data"] != "aGVsbG8=" || p["description"] != "a part" { + t.Errorf("put asked %v", p) + } +} diff --git a/go/delivery/builder_reference_test.go b/go/delivery/builder_reference_test.go index 5e38964..86d6bba 100644 --- a/go/delivery/builder_reference_test.go +++ b/go/delivery/builder_reference_test.go @@ -104,6 +104,11 @@ func TestOnlyAPushFailureIsBuiltAgainBeforeTheFingerprintChanges(t *testing.T) { if NotStored != "not stored: " || NotPinned != "not pinned: " || VersionTaken != "version taken: " || RefusedKind != "refused kind: " { t.Fatal("a prefix changed its words") } + // A Node fetch that finds the lock naming a package version the registry does not hold fails with WaitsFor; the + // builder then writes the key queued with Waits, and the error is not one that waits for the fingerprint to change. + if WaitsFor != "waits for: " || FailedUntilChanged(WaitsFor+"@novox/mesh-sdk@0.1.15") { + t.Fatalf("WaitsFor = %q", WaitsFor) + } } // A key whose module's lock names a version of the group's package not yet published waits, queued, saying which @@ -144,10 +149,3 @@ func TestADoneKeyWithNoArtifactIsReferenced(t *testing.T) { } } -// A Node fetch that finds the lock naming a package version the registry does not hold fails with WaitsFor; the -// builder then writes the key queued with Waits, and the error is not one that waits for the fingerprint to change. -func TestWaitsForIsAPrefixThatIsBuiltAgain(t *testing.T) { - if WaitsFor != "waits for: " || FailedUntilChanged(WaitsFor+"@novox/mesh-sdk@0.1.15") { - t.Fatalf("WaitsFor = %q", WaitsFor) - } -} diff --git a/go/packages_list_test.go b/go/packages_list_test.go index 78328ed..d55f12a 100644 --- a/go/packages_list_test.go +++ b/go/packages_list_test.go @@ -21,7 +21,7 @@ import ( // Logic only some modules need is its owner module's package. A package added here is a decision to say in the pull // request: does every module share it, or is it a shape more than one part reads? var meshSDKPackages = []string{ - ".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busretry", "calladdress", "catalogue", + ".", "announce", "announcetest", "asker", "asks", "buildstep", "buscalls", "buscompose", "busstate", "busretry", "calladdress", "catalogue", "composer", "declaration", "delivery", "forgekey", "handacts", "health", "internal/beside", "kvlimit", "membership", "objectlink", "outward", "overlay", "pernode", "process", "provider", "provider/harness", "provisions", "sdkversion", "seal", "seatdef", "testbus",