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