diff --git a/go/bus_waits_for_serve_test.go b/go/bus_waits_for_serve_test.go index 5dd9eef..90502be 100644 --- a/go/bus_waits_for_serve_test.go +++ b/go/bus_waits_for_serve_test.go @@ -8,50 +8,66 @@ import ( // 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. +// The test drives the wait itself: serveWait is the test's own timer channel, no row sleeps on a clock. func TestBusFunctionsWaitForServe(t *testing.T) { + const refusal = "before Serve: a bundle reaches the bus through the runtime that launched it" for _, tc := range []struct { - name string - wait time.Duration - serveAt time.Duration // 0: Serve never runs - serveFirst bool - wantOK bool + name string + serve string // "first": before the call; "during": while the call waits; "never" + 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}, + {"a call before Serve completes after Serve", "during", true}, + {"a call whose wait ends before Serve is refused", "never", false}, + {"a call after Serve is as before", "first", true}, } { t.Run(tc.name, func(t *testing.T) { servesAlone(t) - old := serveWait - serveWait = tc.wait - t.Cleanup(func() { serveWait = old }) + expire := make(chan time.Time) + oldAfter := serveTimer + serveTimer = func() (<-chan time.Time, func()) { return expire, func() {} } + t.Cleanup(func() { serveTimer = oldAfter }) + 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 { + t.Cleanup(func() { _ = inW.Close(); go func() { _, _ = io.Copy(io.Discard, outR) }() }) + + type result struct { + c *channel + err error + } + done := make(chan result, 1) + if tc.serve == "first" { serve() <-Served() - } else if tc.serveAt > 0 { - time.AfterFunc(tc.serveAt, serve) + c, err := live() + done <- result{c, err} + } else { + go func() { c, err := live(); done <- result{c, err} }() + select { + case r := <-done: + t.Fatalf("live() returned before Serve and before its wait ended: %v", r.err) + default: + } + if tc.serve == "during" { + serve() + } else { + expire <- time.Time{} + } } - c, err := live() - if tc.wantOK != (err == nil && c != nil) { - t.Fatalf("live() = %v, %v; want ok=%v", c, err, tc.wantOK) + r := <-done + if tc.wantOK != (r.err == nil && r.c != nil) { + t.Fatalf("live() = %v, %v; want ok=%v", r.c, r.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) + if !tc.wantOK && r.err.Error() != refusal { + t.Fatalf("refusal words changed: %v", r.err) + } + if tc.serve != "never" { + _ = inW.Close() + <-stopped } }) } diff --git a/go/live_before_serve_test.go b/go/live_before_serve_test.go index 0c7e3bf..58cb90f 100644 --- a/go/live_before_serve_test.go +++ b/go/live_before_serve_test.go @@ -14,7 +14,7 @@ import ( func TestBusFunctionsRefuseBeforeServeInAFreshProcess(t *testing.T) { const marker = "MESH_SDK_TEST_FRESH_PROCESS" if os.Getenv(marker) == "1" { - serveWait = 10 * time.Millisecond + serveTimer = func() (<-chan time.Time, func()) { return time.After(10 * time.Millisecond), func() {} } 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/seatdef/seatdef_test.go b/go/seatdef/seatdef_test.go index e9a9cad..d679037 100644 --- a/go/seatdef/seatdef_test.go +++ b/go/seatdef/seatdef_test.go @@ -29,6 +29,7 @@ func TestADefinitionIsReadStrictlyNamingWhatIsWrong(t *testing.T) { {"no decision", `{"format":1,"seat":"a-b","scope":"mesh","grain":"seat-list"}`, "decision: missing"}, {"no domain", `{"format":1,"seat":"a-b","scope":"mesh","grain":"seat-list","decision":"x"}`, "domain: missing"}, {"a domain the glossary lacks", `{"format":1,"seat":"a-b","scope":"mesh","grain":"seat-list","domain":"Weather","decision":"x"}`, "domain"}, + {"the Placement domain is gone (ADR 0641 section 5)", `{"format":1,"seat":"a-b","scope":"mesh","grain":"seat-list","domain":"Placement","decision":"x"}`, "domain"}, {"an unknown field", `{"format":1,"seat":"a-b","scope":"mesh","grain":"seat-list","domain":"Core","decision":"x","sevres":[]}`, "sevres: a field the format does not know"}, {"a whole verb with no answer", strings.Replace(whole, `,"answer":{"type":"object"}`, "", 1), diff --git a/go/stdio.go b/go/stdio.go index 41a07bd..742e831 100644 --- a/go/stdio.go +++ b/go/stdio.go @@ -446,6 +446,12 @@ func ServeOn(name string, tools []Tool, in io.Reader, out io.Writer) error { // process that never serves is refused, after this long. var serveWait = 30 * time.Second +// serveTimer starts that wait: its channel fires when the wait ends, and stop releases it. A test replaces it. +var serveTimer = func() (<-chan time.Time, func()) { + t := time.NewTimer(serveWait) + return t.C, func() { t.Stop() } +} + // awaitChannel is the runtime's channel, waiting up to serveWait for Serve to set it. func awaitChannel() *channel { currentMu.Lock() @@ -454,11 +460,11 @@ func awaitChannel() *channel { if c != nil { return c } - t := time.NewTimer(serveWait) - defer t.Stop() + expired, stop := serveTimer() + defer stop() select { case <-ready: - case <-t.C: + case <-expired: return nil } currentMu.Lock()