c10e22a fix(obs): tighten trace and spawn validation · vev

M internal/adapters/ipc/transport_observability_test.go +12 -0


@@ -82,6 +82,7 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
82	82		}
83	83		type mark struct {
84	84		ProcessID string `json:"process_id"`
	85	+	Component string `json:"component"`
85	86		Scenario string `json:"scenario"`
86	87		Run uint64 `json:"run"`
87	88		Sequence uint64 `json:"sequence"`

@@ -95,6 +96,7 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
95	96		run, sequence, requestID, epoch uint64
96	97		}
97	98		starts := make(map[spanKey]mark)
	99	+	startCounts := map[string]int{}
98	100		for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
99	101		var m mark
100	102		if err := json.Unmarshal([]byte(line), &m); err != nil {

@@ -103,10 +105,14 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
103	105		key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
104	106		switch m.Kind {
105	107		case "adapter_receive_start", "adapter_send_start":
	108	+	if m.ProcessID != "ipc-process" \|\| m.Component != "ipc" \|\| m.Scenario != "runtime" \|\| m.Run != 1 \|\| m.Sequence == 0 \|\| m.RequestID != m.Sequence \|\| m.Epoch != m.Sequence \|\| !m.Valid {
	109	+	t.Fatalf("adapter start has invalid identity: %#v", m)
	110	+	}
106	111		if _, exists := starts[key]; exists {
107	112		t.Fatalf("duplicate adapter start for %#v", key)
108	113		}
109	114		starts[key] = m
	115	+	startCounts[m.Kind]++
110	116		case "adapter_receive_end", "adapter_send_end":
111	117		start, ok := starts[key]
112	118		if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {

@@ -121,6 +127,12 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
121	127		if len(starts) != 0 {
122	128		t.Fatalf("unmatched adapter starts: %#v", starts)
123	129		}
	130	+	if got := startCounts["adapter_receive_start"]; got != 2 {
	131	+	t.Errorf("adapter receive starts = %d, want 2", got)
	132	+	}
	133	+	if got := startCounts["adapter_send_start"]; got != 1 {
	134	+	t.Errorf("adapter send starts = %d, want 1", got)
	135	+	}
124	136		}
125	137
126	138		func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {


@@ -82,6 +82,7 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
82	}	82		}
83	type mark struct {	83		type mark struct {
84	ProcessID string `json:"process_id"`	84		ProcessID string `json:"process_id"`
		85	+	Component string `json:"component"`
85	Scenario string `json:"scenario"`	86		Scenario string `json:"scenario"`
86	Run uint64 `json:"run"`	87		Run uint64 `json:"run"`
87	Sequence uint64 `json:"sequence"`	88		Sequence uint64 `json:"sequence"`

@@ -95,6 +96,7 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
95	run, sequence, requestID, epoch uint64	96		run, sequence, requestID, epoch uint64
96	}	97		}
97	starts := make(map[spanKey]mark)	98		starts := make(map[spanKey]mark)
		99	+	startCounts := map[string]int{}
98	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {	100		for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
99	var m mark	101		var m mark
100	if err := json.Unmarshal([]byte(line), &m); err != nil {	102		if err := json.Unmarshal([]byte(line), &m); err != nil {

@@ -103,10 +105,14 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
103	key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}	105		key := spanKey{m.ProcessID, m.Scenario, m.Run, m.Sequence, m.RequestID, m.Epoch}
104	switch m.Kind {	106		switch m.Kind {
105	case "adapter_receive_start", "adapter_send_start":	107		case "adapter_receive_start", "adapter_send_start":
		108	+	if m.ProcessID != "ipc-process" \|\| m.Component != "ipc" \|\| m.Scenario != "runtime" \|\| m.Run != 1 \|\| m.Sequence == 0 \|\| m.RequestID != m.Sequence \|\| m.Epoch != m.Sequence \|\| !m.Valid {
		109	+	t.Fatalf("adapter start has invalid identity: %#v", m)
		110	+	}
106	if _, exists := starts[key]; exists {	111		if _, exists := starts[key]; exists {
107	t.Fatalf("duplicate adapter start for %#v", key)	112		t.Fatalf("duplicate adapter start for %#v", key)
108	}	113		}
109	starts[key] = m	114		starts[key] = m
		115	+	startCounts[m.Kind]++
110	case "adapter_receive_end", "adapter_send_end":	116		case "adapter_receive_end", "adapter_send_end":
111	start, ok := starts[key]	117		start, ok := starts[key]
112	if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {	118		if !ok \|\| strings.TrimSuffix(start.Kind, "_start") != strings.TrimSuffix(m.Kind, "_end") {

@@ -121,6 +127,12 @@ func assertIPCFailedSpanPairs(t *testing.T, path string) {
121	if len(starts) != 0 {	127		if len(starts) != 0 {
122	t.Fatalf("unmatched adapter starts: %#v", starts)	128		t.Fatalf("unmatched adapter starts: %#v", starts)
123	}	129		}
		130	+	if got := startCounts["adapter_receive_start"]; got != 2 {
		131	+	t.Errorf("adapter receive starts = %d, want 2", got)
		132	+	}
		133	+	if got := startCounts["adapter_send_start"]; got != 1 {
		134	+	t.Errorf("adapter send starts = %d, want 1", got)
		135	+	}
124	}	136		}
125		137
126	func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {	138		func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {

M internal/adapters/observability/jsonl_test.go +3 -1


@@ -21,7 +21,9 @@ func TestTransportObservabilityJSONLOwnsTicksAndSchema(t *testing.T) {
21	21		observer.ObserveRuntime(validRuntimeMark(ports.RuntimeComposeStart))
22	22		// A producer's clock domain must never enter a JSONL record. Rejection is
23	23		// observable as no second record, rather than a silently rewritten mark.
24		-	observer.ObserveRuntime(ports.RuntimeMark{Schema: 1, ProcessID: "daemon-process", Component: "daemon", Scenario: "resize", Run: 7, Sequence: 9, Kind: ports.RuntimeComposeEnd, Tick: 99, Valid: true})
	24	+	producerTicked := validRuntimeMark(ports.RuntimeComposeEnd)
	25	+	producerTicked.Tick = 99
	26	+	observer.ObserveRuntime(producerTicked)
25	27		if err := closer.Close(); err != nil {
26	28		t.Fatalf("Close() error = %v", err)
27	29		}

 	observer.ObserveRuntime(validRuntimeMark(ports.RuntimeComposeStart))
 	// A producer's clock domain must never enter a JSONL record. Rejection is
 	// observable as no second record, rather than a silently rewritten mark.
-	observer.ObserveRuntime(ports.RuntimeMark{Schema: 1, ProcessID: "daemon-process", Component: "daemon", Scenario: "resize", Run: 7, Sequence: 9, Kind: ports.RuntimeComposeEnd, Tick: 99, Valid: true})
+	producerTicked := validRuntimeMark(ports.RuntimeComposeEnd)
+	producerTicked.Tick = 99
+	observer.ObserveRuntime(producerTicked)
 	if err := closer.Close(); err != nil {
 		t.Fatalf("Close() error = %v", err)
 	}

M internal/app/spawn.go +3 -0


@@ -76,6 +76,9 @@ func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFun
76	76		// rather than spawning a second daemon.
77	77		defer release()
78	78		slog.Info("spawning daemon", "socket_dir", dir)
	79	+	if err := ctx.Err(); err != nil {
	80	+	return nil, err
	81	+	}
79	82		if err := spawn(); err != nil {
80	83		slog.Error("daemon spawn failed", "err", err)
81	84		return nil, fmt.Errorf("vev: spawning daemon: %w", err)


@@ -76,6 +76,9 @@ func ensureDaemon(ctx context.Context, dir string, dial dialFunc, spawn spawnFun
76	// rather than spawning a second daemon.	76		// rather than spawning a second daemon.
77	defer release()	77		defer release()
78	slog.Info("spawning daemon", "socket_dir", dir)	78		slog.Info("spawning daemon", "socket_dir", dir)
		79	+	if err := ctx.Err(); err != nil {
		80	+	return nil, err
		81	+	}
79	if err := spawn(); err != nil {	82		if err := spawn(); err != nil {
80	slog.Error("daemon spawn failed", "err", err)	83		slog.Error("daemon spawn failed", "err", err)
81	return nil, fmt.Errorf("vev: spawning daemon: %w", err)	84		return nil, fmt.Errorf("vev: spawning daemon: %w", err)

M internal/app/spawn_test.go +22 -0


@@ -153,6 +153,28 @@ func TestEnsureDaemonContextCancelledDoesNotDialOrSpawn(t *testing.T) {
153	153		}
154	154		}
155	155
	156	+	func TestEnsureDaemonDoesNotSpawnWhenContextCancelsAfterFailedDial(t *testing.T) {
	157	+	ctx, cancel := context.WithCancel(context.Background())
	158	+	defer cancel()
	159	+	var spawnCalls atomic.Int32
	160	+	dial := func(context.Context, string) (ports.Transport, error) {
	161	+	cancel()
	162	+	return nil, errDialFailed
	163	+	}
	164	+	spawn := func() error {
	165	+	spawnCalls.Add(1)
	166	+	return nil
	167	+	}
	168	+
	169	+	_, err := ensureDaemon(ctx, filepath.Join(t.TempDir(), "vev"), dial, spawn, fastBackoff)
	170	+	if !errors.Is(err, context.Canceled) {
	171	+	t.Fatalf("ensureDaemon error = %v, want context.Canceled", err)
	172	+	}
	173	+	if got := spawnCalls.Load(); got != 0 {
	174	+	t.Fatalf("spawn called %d times after canceled failed dial, want 0", got)
	175	+	}
	176	+	}
	177	+
156	178		func TestEnsureDaemonSpawnsThenDials(t *testing.T) {
157	179		dir := filepath.Join(t.TempDir(), "vev")
158	180		want := portsmocks.NewMockTransport(t)


@@ -153,6 +153,28 @@ func TestEnsureDaemonContextCancelledDoesNotDialOrSpawn(t *testing.T) {
153	}	153		}
154	}	154		}
155		155
		156	+	func TestEnsureDaemonDoesNotSpawnWhenContextCancelsAfterFailedDial(t *testing.T) {
		157	+	ctx, cancel := context.WithCancel(context.Background())
		158	+	defer cancel()
		159	+	var spawnCalls atomic.Int32
		160	+	dial := func(context.Context, string) (ports.Transport, error) {
		161	+	cancel()
		162	+	return nil, errDialFailed
		163	+	}
		164	+	spawn := func() error {
		165	+	spawnCalls.Add(1)
		166	+	return nil
		167	+	}
		168	+
		169	+	_, err := ensureDaemon(ctx, filepath.Join(t.TempDir(), "vev"), dial, spawn, fastBackoff)
		170	+	if !errors.Is(err, context.Canceled) {
		171	+	t.Fatalf("ensureDaemon error = %v, want context.Canceled", err)
		172	+	}
		173	+	if got := spawnCalls.Load(); got != 0 {
		174	+	t.Fatalf("spawn called %d times after canceled failed dial, want 0", got)
		175	+	}
		176	+	}
		177	+
156	func TestEnsureDaemonSpawnsThenDials(t *testing.T) {	178		func TestEnsureDaemonSpawnsThenDials(t *testing.T) {
157	dir := filepath.Join(t.TempDir(), "vev")	179		dir := filepath.Join(t.TempDir(), "vev")
158	want := portsmocks.NewMockTransport(t)	180		want := portsmocks.NewMockTransport(t)

M internal/app/transport_observability_test.go +4 -6


@@ -10,6 +10,7 @@ import (
10	10		"time"
11	11
12	12		"github.com/bnema/vev/internal/ports"
	13	+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
13	14		)
14	15
15	16		func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {

@@ -19,7 +20,9 @@ func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
19	20		t.Setenv("VEV_PERF_SCENARIO", "1x4-idle-local")
20	21		t.Setenv("VEV_PERF_RUN", "1")
21	22
22		-	observer, closer, err := performanceTrace(&traceTestClock{})
	23	+	clock := portsmocks.NewMockClock(t)
	24	+	clock.EXPECT().Now().Return(time.Unix(0, 1)).Once()
	25	+	observer, closer, err := performanceTrace(clock)
23	26		if err != nil {
24	27		t.Fatal(err)
25	28		}

@@ -51,11 +54,6 @@ func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
51	54		}
52	55		}
53	56
54		-	type traceTestClock struct{}
55		-
56		-	func (*traceTestClock) Now() time.Time { return time.Unix(0, 1) }
57		-	func (*traceTestClock) NewTimer(time.Duration) ports.Timer { return nil }
58		-
59	57		func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
60	58		source, err := os.ReadFile("run.go")
61	59		if err != nil {

 	"time"
 	"github.com/bnema/vev/internal/ports"
+	portsmocks "github.com/bnema/vev/internal/ports/mocks"
 )
 func TestPerformanceTraceUsesHarnessProcessMapping(t *testing.T) {
 	t.Setenv("VEV_PERF_SCENARIO", "1x4-idle-local")
 	t.Setenv("VEV_PERF_RUN", "1")
-	observer, closer, err := performanceTrace(&traceTestClock{})
+	clock := portsmocks.NewMockClock(t)
+	clock.EXPECT().Now().Return(time.Unix(0, 1)).Once()
+	observer, closer, err := performanceTrace(clock)
 	if err != nil {
 		t.Fatal(err)
 	}
 	}
 }
-type traceTestClock struct{}
-func (*traceTestClock) Now() time.Time                     { return time.Unix(0, 1) }
-func (*traceTestClock) NewTimer(time.Duration) ports.Timer { return nil }
 func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
 	source, err := os.ReadFile("run.go")
 	if err != nil {