diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 633c848..ef3826f 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -19,6 +19,7 @@ import ( "syscall" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/buscalls" "golang.org/x/sys/unix" @@ -392,6 +393,16 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an if asked, err = mesh.Ask(m.Params); err == nil { result = asked } + case stdio.CapabilitiesCall: + // The capability words this runner states (design 47 G §576); a Bus that states none answers none. + words := []string{} + if w, ok := mesh.(interface{ CapabilityWords() []string }); ok { + words = w.CapabilityWords() + } + var raw []byte + if raw, err = json.Marshal(map[string]any{"words": words}); err == nil { + result = json.RawMessage(raw) + } case "mesh/subscribe": mu.Lock() subscriber = c @@ -577,6 +588,10 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an for _, stop := range watches { stop() } + // A session of the bundle's put or get ends with its channel (design 47 G §518). + if d, ok := mesh.(interface{ DropSessions() }); ok { + d.DropSessions() + } close(c.dead) mu.Lock() // Only a child that was serving is brought back; one that died in its own handshake was diff --git a/node-tools/internal/launch/sessions_test.go b/node-tools/internal/launch/sessions_test.go new file mode 100644 index 0000000..e4a1992 --- /dev/null +++ b/node-tools/internal/launch/sessions_test.go @@ -0,0 +1,68 @@ +package launch + +import ( + "encoding/json" + "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 +// session, as mesh-sdk's bundle does; it then ends, which closes its channel. +const puttingBundle = `#!/bin/sh +while IFS= read -r line; do + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\)[,}].*/\1/p') + case "$line" in + *'"method":"initialize"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;; + *'"method":"tools/list"'*) printf '{"jsonrpc":"2.0","id":%s,"result":{"tools":[]}}\n' "$id" + printf '{"jsonrpc":"2.0","id":"w1","method":"mesh/capabilities","params":{}}\n' ;; + *'"id":"w1"'**put-sessions*) printf '{"jsonrpc":"2.0","id":"p1","method":"mesh/objects.put-chunk","params":{"objects":"o","name":"n","call":"c","seq":1,"last":true,"data":"aGk="}}\n' ;; + *'"id":"p1"'*) exit 0 ;; + esac +done +` + +type sessionBus struct { + subscribing + mu sync.Mutex + chunks []string + dropped chan struct{} +} + +func (b *sessionBus) CapabilityWords() []string { return []string{"put-sessions"} } +func (b *sessionBus) BusState(method string, params json.RawMessage) (json.RawMessage, error) { + b.mu.Lock() + b.chunks = append(b.chunks, method+" "+string(params)) + b.mu.Unlock() + return json.RawMessage(`{"name":"n"}`), nil +} +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) + } +} diff --git a/node-tools/internal/runtime/sessions.go b/node-tools/internal/runtime/sessions.go index 7227f64..817ad6e 100644 --- a/node-tools/internal/runtime/sessions.go +++ b/node-tools/internal/runtime/sessions.go @@ -299,3 +299,9 @@ func newCall() string { _, _ = rand.Read(b[:]) return hex.EncodeToString(b[:]) } + +// CapabilityWords are the capability words this runner states, answered to a bundle's `mesh/capabilities` (§576). +func (b *moduleBus) CapabilityWords() []string { return CallCapabilities() } + +// DropSessions ends the sessions of the module whose bundle's channel closed (§518). +func (b *moduleBus) DropSessions() { b.all.dropModuleSessions(b.module) }