Review round 1 for the checks, ssh-client and the Go build module: sshdir (1484), socket path (1486), over_bound (1290, 1486), toolchain.reported and streams (1483), re-pin to mesh-sdk 36d6f774; delta part 2 of 4

modules/mesh-build-go/cmd/mesh-build-go/emit_test.go

@@ -0,0 +1,55 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"slices"
+	"testing"
+)
+
+// Issues 1483 and 1488 (for the indexer, issue 1444): the guard emits toolchain.reported after each write of the
+// toolchains key, with the node and the language.
+func TestTheGuardEmitsToolchainReportedAfterEachWrite(t *testing.T) {
+	m := moduleWith(newStates(), "go1.27.1")
+	type event struct {
+		typ  string
+		body map[string]string
+	}
+	var got []event
+	m.Emit = func(typ string, body any) error {
+		raw, _ := json.Marshal(body)
+		var b map[string]string
+		json.Unmarshal(raw, &b)
+		got = append(got, event{typ, b})
+		return nil
+	}
+	if err := m.Check(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 || got[0].typ != "toolchain.reported" || got[0].body["node"] != "ace" || got[0].body["language"] != "go" || len(got[0].body) != 2 {
+		t.Errorf("events = %+v, want one toolchain.reported for ace, go", got)
+	}
+	m.Check(context.Background())
+	if len(got) != 2 {
+		t.Errorf("a second guard pass emitted %d events in all, want 2", len(got))
+	}
+}
+
+// The event is carried by the module's stream, not named in emits (ADR 0363: one place); the validator checks that.
+func TestTheManifestDeclaresTheToolchainsStream(t *testing.T) {
+	raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m struct {
+		Streams []struct {
+			Name   string   `json:"name"`
+			Events []string `json:"events"`
+		} `json:"streams"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil || len(m.Streams) != 1 || m.Streams[0].Name != "toolchains-go" || !slices.Equal(m.Streams[0].Events, []string{"toolchain.**"}) {
+		t.Errorf("streams = %+v (%v)", m.Streams, err)
+	}
+}

modules/mesh-build-go/cmd/mesh-build-go/integration_level_test.go

@@ -155,7 +155,12 @@ func TestTheStoreListensOnASocketUnderTheRunAndNowhereElse(t *testing.T) {
 	if !strings.HasPrefix(run, filepath.Join(m.Work, "ws1", "run-")) {
 		t.Errorf("the run directory is %s, not under the workspace's", run)
 	}
-	for _, want := range []string{"listen_addresses=''", "-k " + filepath.Join(run, "pg", "socket"), "fsync=off", "-w"} {
+	_, afterK, _ := strings.Cut(start, "-k ")
+	socket, _, _ := strings.Cut(afterK, " ")
+	if len(socket)+len(socketName) >= 100 {
+		t.Errorf("the socket %q is too long for postgres", socket)
+	}
+	for _, want := range []string{"listen_addresses=''", "-k " + socket, "fsync=off", "-w"} {
 		if !strings.Contains(start, want) {
 			t.Errorf("the store's start %q lacks %q", start, want)
 		}
@@ -181,7 +186,7 @@ func TestTheStoreListensOnASocketUnderTheRunAndNowhereElse(t *testing.T) {
 		env[k] = v
 	}
 	u, err := url.Parse(env["MESH_TEST_POSTGRES"])
-	if err != nil || u.Scheme != "postgres" || u.Query().Get("host") != filepath.Join(run, "pg", "socket") || u.Host != "" {
+	if err != nil || u.Scheme != "postgres" || u.Query().Get("host") != socket || u.Host != "" {
 		t.Errorf("MESH_TEST_POSTGRES = %q", env["MESH_TEST_POSTGRES"])
 	}
 	if strings.Count(strings.TrimPrefix(env["MESH_TEST_POSTGRES"], "postgres://"), "/") != 1 {
@@ -220,7 +225,7 @@ func TestSetupAndRunningTimeAreJudgedApart(t *testing.T) {
 	f.advance["compile"], f.advance["start"] = 20*time.Second, 15*time.Second
 	m, states := integModule(t, f)
 	res, _ := runLevelTool(t, m, states, "integration")
-	if res.Passed || res.Setup != 35 || !strings.Contains(res.Why, "setup took 35.00 s, over its setup bound of 30 s") || strings.Contains(res.Why, "slowest tests") {
+	if !res.Passed || !res.OverBound || res.OverWhich != "setup" || res.Setup != 35 || !strings.Contains(res.Why, "setup took 35.00 s, over its setup bound of 30 s") || strings.Contains(res.Why, "slowest tests") {
 		t.Errorf("a setup of 35 s: %+v", res)
 	}
 
@@ -228,14 +233,14 @@ func TestSetupAndRunningTimeAreJudgedApart(t *testing.T) {
 	g.advance["test"] = 61 * time.Second
 	m, states = integModule(t, g)
 	res, _ = runLevelTool(t, m, states, "integration")
-	if res.Passed || res.Elapsed != 61 || !strings.Contains(res.Why, "took 61.00 s, over its bound of 60 s; the slowest tests: fixture/slow TestTakesAWhile") || strings.Contains(res.Why, "setup") {
+	if !res.Passed || res.OverWhich != "run" || res.Elapsed != 61 || !strings.Contains(res.Why, "took 61.00 s, over its bound of 60 s; the slowest tests: fixture/slow TestTakesAWhile") || strings.Contains(res.Why, "setup") {
 		t.Errorf("a run of 61 s: %+v", res)
 	}
 
 	h := newIntegFake(t, "slow")
 	h.advance["compile"], h.advance["test"] = 29*time.Second, 59*time.Second
 	m, states = integModule(t, h)
-	if res, _ = runLevelTool(t, m, states, "integration"); !res.Passed || res.Setup != 29 || res.Elapsed != 59 {
+	if res, _ = runLevelTool(t, m, states, "integration"); !res.Passed || res.OverBound || res.Setup != 29 || res.Elapsed != 59 {
 		t.Errorf("29 s of setup and 59 s of running are within both bounds: %+v", res)
 	}
 }
@@ -316,3 +321,46 @@ func TestThePostgresMajorIsReadFromTheVersionLine(t *testing.T) {
 		t.Errorf("accepted a line that is not postgres's")
 	}
 }
+
+// Issue 1486 review: postgres refuses a socket path of 108 bytes and over. A run directory deep in the work directory
+// gets a short directory of its own for the socket, removed with the store; a short one keeps it under the run.
+func TestTheSocketPathStaysUnderTheLimitForALongRunDirectory(t *testing.T) {
+	long := filepath.Join(t.TempDir(), strings.Repeat("deep", 20), "ws1", "run-mvh5uc96142806")
+	if err := os.MkdirAll(long, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	dir, short, err := socketDir(long)
+	if err != nil || !short || len(dir)+len(socketName) >= 100 {
+		t.Fatalf("socketDir(long) = %q, %v, %v", dir, short, err)
+	}
+	if _, err := os.Stat(dir); err != nil {
+		t.Errorf("the short directory was not made: %v", err)
+	}
+	os.RemoveAll(dir)
+
+	near := t.TempDir()
+	if dir, short, err := socketDir(near); err != nil || (len(near) < 50 && (short || !strings.HasPrefix(dir, near))) {
+		t.Errorf("socketDir(short) = %q, %v, %v", dir, short, err)
+	}
+
+	// Through the store: the short directory is removed when the store is stopped.
+	f := newIntegFake(t, "slow")
+	m, _ := integModule(t, f)
+	_, stop, err := m.startStore(context.Background(), long)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var sock string
+	for _, c := range f.calls {
+		if _, a, ok := strings.Cut(c.line, "-k "); ok && strings.HasPrefix(c.line, "pg_ctl") {
+			sock, _, _ = strings.Cut(a, " ")
+		}
+	}
+	if sock == "" || len(sock)+len(socketName) >= 100 {
+		t.Fatalf("the store was started with the socket %q", sock)
+	}
+	stop()
+	if _, err := os.Stat(sock); !os.IsNotExist(err) {
+		t.Errorf("the short socket directory %s outlived the store (%v)", sock, err)
+	}
+}

modules/mesh-build-go/cmd/mesh-build-go/levels.go

@@ -77,8 +77,12 @@ type TestResult struct {
 	Failed     []Failure  `json:"failed,omitempty"`
 	// Store and Bus are what the integration level ran against: the postgres major version and the nats-server
 	// version (ADR 0470 §4).
-	Store string `json:"store,omitempty"`
-	Bus   string `json:"bus,omitempty"`
+	// OverBound says a bound was kept no longer, and OverWhich which: setup, run or both. The level's verdict, Passed, is
+	// the tests' own: the checker judges the bounds (ADR 0453 §9), with the slowest tests this result names.
+	OverBound bool   `json:"over_bound"`
+	OverWhich string `json:"over_which,omitempty"`
+	Store     string `json:"store,omitempty"`
+	Bus       string `json:"bus,omitempty"`
 }
 
 // event is one line of `go test -json` (test2json, and the build events go test adds).
@@ -221,7 +225,7 @@ func (res *TestResult) judge(b Bounds) {
 		res.Slowest = res.Slowest[:5]
 	}
 	res.Bound, res.SetupBound = b.Run.Seconds(), b.Setup.Seconds()
-	var why []string
+	var why, over []string
 	if n := len(res.Failed); n > 0 {
 		names := []string{}
 		for _, f := range res.Failed {
@@ -229,28 +233,27 @@ func (res *TestResult) judge(b Bounds) {
 		}
 		why = append(why, fmt.Sprintf("%d failed: %s", n, strings.Join(names, ", ")))
 	}
+	res.Passed = len(res.Failed) == 0
 	if b.Setup > 0 && res.Setup > b.Setup.Seconds() {
+		over = append(over, "setup")
 		why = append(why, fmt.Sprintf("the %s level's setup took %.2f s, over its setup bound of %s", res.Level, res.Setup, seconds(b.Setup)))
 	}
 	if b.Run > 0 && res.Elapsed > b.Run.Seconds() {
+		over = append(over, "run")
 		slow := []string{}
 		for _, s := range res.Slowest {
 			slow = append(slow, fmt.Sprintf("%s %s (%.2f s)", s.Package, s.Test, s.Elapsed))
 		}
 		why = append(why, fmt.Sprintf("the %s level took %.2f s, over its bound of %s; the slowest tests: %s", res.Level, res.Elapsed, seconds(b.Run), strings.Join(slow, ", ")))
 	}
+	res.OverBound, res.OverWhich = len(over) > 0, strings.Join(over, ",")
 	res.NoTests = res.Tests == 0 && len(res.Failed) == 0
 	switch {
 	case len(why) > 0:
 		res.Why = strings.Join(why, "; ")
 	case res.NoTests:
-		res.Passed, res.Why = true, "the repository has no tests at the "+res.Level+" level"
-		return
-	default:
-		res.Passed = true
-		return
+		res.Why = "the repository has no tests at the " + res.Level + " level"
 	}
-	res.Passed = false
 }
 
 // seconds is a duration as seconds, without trailing zeros: "5 s", "0.01 s".

modules/mesh-build-go/cmd/mesh-build-go/main.go

@@ -35,6 +35,7 @@ func main() {
 		Recipe:     Current(),
 		Probe:      ProbeGo,
 		Toolchains: stdio.State(delivery.ToolchainsState).Put,
+		Emit:       func(t string, b any) error { return stdio.Emit(t, b) },
 		Steps:      stdio.State(delivery.StepsState).Put,
 		Ctx:        ctx,
 		Say:        logf,

modules/mesh-build-go/cmd/mesh-build-go/module.go

@@ -24,6 +24,9 @@ type Module struct {
 	Probe func(ctx context.Context) (string, error)
 	// Toolchains and Steps are the Put of the states of those names.
 	Toolchains, Steps buildstep.Put
+	// Emit tells the bus an event (stdio.Emit); nil emits nothing. The guard emits toolchain.reported after each write of
+	// the toolchains key, so the indexer, which holds no timer, learns that the bucket exists (issue 1444).
+	Emit func(eventType string, body any) error
 	// Ctx is the process's context, which a step runs under (the tool call's is cut at 30 s, the step's is not).
 	Ctx context.Context
 	Say func(format string, a ...any)
@@ -43,7 +46,7 @@ type Module struct {
 }
 
 func (m *Module) guard() buildstep.Guard {
-	return buildstep.Guard{Language: "go", Want: m.Recipe.Go, Recipe: m.Recipe.Version(), Probe: m.Probe, Node: m.Node, Put: m.Toolchains}
+	return buildstep.Guard{Language: "go", Want: m.Recipe.Go, Recipe: m.Recipe.Version(), Probe: m.Probe, Node: m.Node, Put: m.reportingPut()}
 }
 
 // Check reads the installed toolchain, writes <node>.go of the toolchains state and answers nil, or the refusal
@@ -123,3 +126,21 @@ func (m *Module) Status(ctx context.Context) map[string]any {
 	}
 	return out
 }
+
+// ToolchainReported is the event the guard emits after it writes the toolchains key.
+const ToolchainReported = "toolchain.reported"
+
+// reportingPut is the toolchains Put that then emits toolchain.reported with the node and the language. A write that
+// failed emits nothing; an emit that failed is said and does not refuse the step.
+func (m *Module) reportingPut() buildstep.Put {
+	return func(key string, value any) (uint64, error) {
+		rev, err := m.Toolchains(key, value)
+		if err != nil || m.Emit == nil {
+			return rev, err
+		}
+		if eerr := m.Emit(ToolchainReported, map[string]string{"node": m.Node, "language": "go"}); eerr != nil && m.Say != nil {
+			m.Say("the event %s could not be emitted: %v", ToolchainReported, eerr)
+		}
+		return rev, nil
+	}
+}

modules/mesh-build-go/cmd/mesh-build-go/store.go

@@ -49,6 +49,24 @@ func (m *Module) checkPostgres(ctx context.Context) (int, error) {
 	return major, nil
 }
 
+// socketLimit is the longest path of a socket postgres accepts, with the file's name: it refuses 108 bytes and over.
+const socketLimit = 100
+
+// socketName is the file postgres makes in the socket directory.
+const socketName = "/.s.PGSQL.5432"
+
+// socketDir makes the directory of the store's socket: under the run when the socket's path stays under socketLimit,
+// else a short directory of its own under /tmp (answering short, so the caller removes it). A unix socket's path is
+// bounded far below a file's, and a work directory deep in a data root is over it.
+func socketDir(run string) (dir string, short bool, err error) {
+	dir = filepath.Join(run, "pg", "socket")
+	if len(dir)+len(socketName) < socketLimit {
+		return dir, false, os.MkdirAll(dir, 0o700)
+	}
+	dir, err = os.MkdirTemp("/tmp", "pg")
+	return dir, true, err
+}
+
 // storeSettings are the server's own settings: no TCP address, a socket directory, and no durability.
 func storeOptions(socket string) string {
 	return "-k " + socket + " -c listen_addresses='' -c fsync=off -c synchronous_commit=off -c full_page_writes=off"
@@ -59,11 +77,14 @@ func storeOptions(socket string) string {
 // tests cut the URL at its last slash to name their own databases.
 func (m *Module) startStore(ctx context.Context, run string) (string, func(), error) {
 	data := filepath.Join(run, "pg", "data")
-	socket := filepath.Join(run, "pg", "socket")
-	if err := os.MkdirAll(socket, 0o700); err != nil {
+	socket, short, err := socketDir(run)
+	if err != nil {
 		return "", nil, err
 	}
 	stop := func() {
+		if short {
+			defer os.RemoveAll(socket)
+		}
 		sctx, cancel := context.WithTimeout(context.Background(), storeStopBound)
 		defer cancel()
 		m.Run(sctx, nil, "pg_ctl", "-D", data, "-m", "immediate", "-w", "stop")

modules/mesh-build-go/cmd/mesh-build-go/test_real_integration_test.go

@@ -85,7 +85,12 @@ func processesNamed(text string) []string {
 // It starts postgres and a bus, so it runs in the merge check and in a try, never on the way.
 func TestIntegrationLevelRunsAgainstARealStoreAndBusAndLeavesNothing(t *testing.T) {
 	tree := repository(t, "store")
-	work := t.TempDir()
+	// A short work directory: a unix socket's path is bounded at 108 bytes.
+	work, err := os.MkdirTemp("/tmp", "w")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(work) })
 	if err := os.MkdirAll(filepath.Join(work, "ws1"), 0o755); err != nil {
 		t.Fatal(err)
 	}

modules/mesh-build-go/cmd/mesh-build-go/test_test.go

@@ -107,8 +107,8 @@ func TestUnitLevelOverItsBoundFailsNamingTheBoundAndTheSlowestTests(t *testing.T
 	m, states, ws, _ := workspaceFor(t, "slow")
 	m.UnitBound = 10 * time.Millisecond
 	res, _ := runUnit(t, m, states, ws)
-	if res.Passed {
-		t.Fatalf("a level of %.3f s passed under a bound of %v", res.Elapsed, m.UnitBound)
+	if !res.Passed || !res.OverBound || res.OverWhich != "run" {
+		t.Fatalf("a level of %.3f s under a bound of %v: passed %v, over_bound %v (%s); the tests passed and the bound was kept no longer", res.Elapsed, m.UnitBound, res.Passed, res.OverBound, res.OverWhich)
 	}
 	for _, want := range []string{"over its bound of 0.01 s", "fixture/slow TestTakesAWhile"} {
 		if !strings.Contains(res.Why, want) {
@@ -120,7 +120,7 @@ func TestUnitLevelOverItsBoundFailsNamingTheBoundAndTheSlowestTests(t *testing.T
 	}
 
 	m2, states2, ws2, _ := workspaceFor(t, "slow")
-	if res, _ := runUnit(t, m2, states2, ws2); !res.Passed || res.Bound != 5 {
+	if res, _ := runUnit(t, m2, states2, ws2); !res.Passed || res.OverBound || res.Bound != 5 {
 		t.Errorf("under the bound of five seconds: %+v", res)
 	}
 }