From 240065f5b4e7a67961bd4e561e0399801b054a78 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 22:01:04 +0100 Subject: [PATCH] mesh-sdk: a bus function called before Serve waits for Serve instead of refusing (issue 1966) A module that starts a goroutine before stdio.Serve and calls the bus there was refused and retried on its own tick, a minute later. Refused now only when the process never serves. Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i --- go/bus_waits_for_serve_test.go | 58 ++++++++++++++++++++++++++++++++++ go/live_before_serve_test.go | 2 ++ go/stdio.go | 35 ++++++++++++++++---- 3 files changed, 88 insertions(+), 7 deletions(-) create mode 100644 go/bus_waits_for_serve_test.go diff --git a/go/bus_waits_for_serve_test.go b/go/bus_waits_for_serve_test.go new file mode 100644 index 0000000..5dd9eef --- /dev/null +++ b/go/bus_waits_for_serve_test.go @@ -0,0 +1,58 @@ +package stdio + +import ( + "io" + "testing" + "time" +) + +// A bus function called before Serve waits for Serve and proceeds (issue 1966): a module that starts a goroutine +// before stdio.Serve and calls the bus there is no longer refused. It is refused only when the process never serves. +func TestBusFunctionsWaitForServe(t *testing.T) { + for _, tc := range []struct { + name string + wait time.Duration + serveAt time.Duration // 0: Serve never runs + serveFirst bool + wantOK bool + }{ + {"a call before Serve completes after Serve", 5 * time.Second, 50 * time.Millisecond, false, true}, + {"a call whose wait ends before Serve is refused", 30 * time.Millisecond, 0, false, false}, + {"a call after Serve is as before", 5 * time.Second, 0, true, true}, + } { + t.Run(tc.name, func(t *testing.T) { + servesAlone(t) + old := serveWait + serveWait = tc.wait + t.Cleanup(func() { serveWait = old }) + inR, inW := io.Pipe() + outR, outW := io.Pipe() + stopped := make(chan struct{}) + started := false + t.Cleanup(func() { + _ = inW.Close() + go func() { _, _ = io.Copy(io.Discard, outR) }() + if started { + <-stopped + } + }) + serve := func() { + started = true + go func() { _ = ServeOn("m", nil, inR, outW); close(stopped) }() + } + if tc.serveFirst { + serve() + <-Served() + } else if tc.serveAt > 0 { + time.AfterFunc(tc.serveAt, serve) + } + c, err := live() + if tc.wantOK != (err == nil && c != nil) { + t.Fatalf("live() = %v, %v; want ok=%v", c, err, tc.wantOK) + } + if !tc.wantOK && err.Error() != "before Serve: a bundle reaches the bus through the runtime that launched it" { + t.Fatalf("refusal words changed: %v", err) + } + }) + } +} diff --git a/go/live_before_serve_test.go b/go/live_before_serve_test.go index 11e3356..0c7e3bf 100644 --- a/go/live_before_serve_test.go +++ b/go/live_before_serve_test.go @@ -4,6 +4,7 @@ import ( "os" "os/exec" "testing" + "time" ) // 1171/AC4: before Serve, a bundle's bus functions still refuse exactly as on main (stdio.go:391-398, @@ -13,6 +14,7 @@ import ( func TestBusFunctionsRefuseBeforeServeInAFreshProcess(t *testing.T) { const marker = "MESH_SDK_TEST_FRESH_PROCESS" if os.Getenv(marker) == "1" { + serveWait = 10 * time.Millisecond if err := Emit("x", map[string]any{}); err == nil || err.Error() != "emit before Serve: a bundle emits through the runtime that launched it" { t.Fatalf("Emit before Serve: %v", err) diff --git a/go/stdio.go b/go/stdio.go index f1e8425..41a07bd 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -441,13 +441,36 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { return scanner.Err() } -func live() (*channel, error) { +// serveWait is how long a bus function called before Serve waits for Serve to have the runtime's channel. A bundle +// that starts a goroutine before stdio.Serve and calls the bus there finds the channel a moment later; only a +// process that never serves is refused, after this long. +var serveWait = 30 * time.Second + +// awaitChannel is the runtime's channel, waiting up to serveWait for Serve to set it. +func awaitChannel() *channel { + currentMu.Lock() + c, ready := current, served + currentMu.Unlock() + if c != nil { + return c + } + t := time.NewTimer(serveWait) + defer t.Stop() + select { + case <-ready: + case <-t.C: + return nil + } currentMu.Lock() defer currentMu.Unlock() - if current == nil { - return nil, errors.New("before Serve: a bundle reaches the bus through the runtime that launched it") + return current +} + +func live() (*channel, error) { + if c := awaitChannel(); c != nil { + return c, nil } - return current, nil + return nil, errors.New("before Serve: a bundle reaches the bus through the runtime that launched it") } // Subscribe asks the runtime for this module's events whose type matches the pattern (`*` one @@ -571,9 +594,7 @@ type EmitOptions struct { // accepted it (ADR 0193). The envelope and headers are the TypeScript SDK's exactly: the source and // node come from MESH_MODULE and MESH_NODE, which the runtime sets. func Emit(eventType string, body any, opts ...EmitOptions) error { - currentMu.Lock() - c := current - currentMu.Unlock() + c := awaitChannel() if c == nil { return errors.New("emit before Serve: a bundle emits through the runtime that launched it") }