detached 06f09e6

test(obs): specify runtime mark clock domains and boundaries

commit
06f09e629945e41235bc07c17b93d88e270d6605
author
brice · · 2026-07-12 22:17 UTC
parent
cdb77a1
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
8 files · +616 -0
Split
UnifiedSplit
A internal/adapters/dgram/transport_observability_test.go +84 -0
@@ -0,0 +1,84 @@
1+package dgram
2+
3+import (
4+ "os"
5+ "reflect"
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/bnema/vev/internal/adapters/observability"
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {
15+ behaviorClock := newManualClock(time.Unix(0, 401))
16+ link := newSimulatedLink(behaviorClock, packetPolicy{})
17+ aPC, bPC := newPairWithCapacity(link, 64)
18+ tracePath := t.TempDir() + "/dgram.jsonl"
19+ observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 509)}, "dgram-process")
20+ if err != nil {
21+ t.Fatalf("NewJSONL() error = %v", err)
22+ }
23+
24+ a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(observer))
25+ if err != nil {
26+ t.Fatalf("NewTransportWithOptions(a) error = %v", err)
27+ }
28+ defer func() { _ = a.Close() }()
29+ b, err := NewTransportWithOptions(bPC, testAddr("a"), key(), 2, 1, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(observer))
30+ if err != nil {
31+ t.Fatalf("NewTransportWithOptions(b) error = %v", err)
32+ }
33+ defer func() { _ = b.Close() }()
34+
35+ want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("datagram carriage remains exact")}
36+ if err := a.Send(want); err != nil {
37+ t.Fatalf("Send() error = %v", err)
38+ }
39+ gotCh := make(chan ports.Frame, 1)
40+ errCh := make(chan error, 1)
41+ go func() {
42+ frame, recvErr := b.Recv()
43+ if recvErr != nil {
44+ errCh <- recvErr
45+ return
46+ }
47+ gotCh <- frame
48+ }()
49+ select {
50+ case err := <-errCh:
51+ t.Fatalf("Recv() error = %v", err)
52+ case got := <-gotCh:
53+ if !reflect.DeepEqual(got, want) {
54+ t.Fatalf("carried frame = %#v, want %#v", got, want)
55+ }
56+ case <-time.After(time.Second):
57+ t.Fatal("Recv() did not receive deterministic simulated datagram")
58+ }
59+ if err := closer.Close(); err != nil {
60+ t.Fatalf("trace Close() error = %v", err)
61+ }
62+ trace, err := os.ReadFile(tracePath)
63+ if err != nil {
64+ t.Fatalf("ReadFile(trace): %v", err)
65+ }
66+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
67+ if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
68+ t.Errorf("trace missing %s", kind)
69+ }
70+ }
71+}
72+
73+type dgramObserverClock struct{ now time.Time }
74+
75+func (c dgramObserverClock) Now() time.Time { return c.now }
76+func (c dgramObserverClock) NewTimer(d time.Duration) ports.Timer {
77+ return dgramObserverTimer{timer: time.NewTimer(d)}
78+}
79+
80+type dgramObserverTimer struct{ timer *time.Timer }
81+
82+func (t dgramObserverTimer) C() <-chan time.Time { return t.timer.C }
83+func (t dgramObserverTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
84+func (t dgramObserverTimer) Stop() bool { return t.timer.Stop() }
@@ -0,0 +1,84 @@
1+package dgram
2+
3+import (
4+ "os"
5+ "reflect"
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/bnema/vev/internal/adapters/observability"
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilityDgramKeepsBehaviorClockSeparate(t *testing.T) {
15+ behaviorClock := newManualClock(time.Unix(0, 401))
16+ link := newSimulatedLink(behaviorClock, packetPolicy{})
17+ aPC, bPC := newPairWithCapacity(link, 64)
18+ tracePath := t.TempDir() + "/dgram.jsonl"
19+ observer, closer, err := observability.NewJSONL(tracePath, dgramObserverClock{now: time.Unix(0, 509)}, "dgram-process")
20+ if err != nil {
21+ t.Fatalf("NewJSONL() error = %v", err)
22+ }
23+
24+ a, err := NewTransportWithOptions(aPC, testAddr("b"), key(), 1, 2, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(observer))
25+ if err != nil {
26+ t.Fatalf("NewTransportWithOptions(a) error = %v", err)
27+ }
28+ defer func() { _ = a.Close() }()
29+ b, err := NewTransportWithOptions(bPC, testAddr("a"), key(), 2, 1, Options{Clock: behaviorClock, ResendAfter: time.Hour, Heartbeat: time.Hour}, WithRuntimeObserver(observer))
30+ if err != nil {
31+ t.Fatalf("NewTransportWithOptions(b) error = %v", err)
32+ }
33+ defer func() { _ = b.Close() }()
34+
35+ want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("datagram carriage remains exact")}
36+ if err := a.Send(want); err != nil {
37+ t.Fatalf("Send() error = %v", err)
38+ }
39+ gotCh := make(chan ports.Frame, 1)
40+ errCh := make(chan error, 1)
41+ go func() {
42+ frame, recvErr := b.Recv()
43+ if recvErr != nil {
44+ errCh <- recvErr
45+ return
46+ }
47+ gotCh <- frame
48+ }()
49+ select {
50+ case err := <-errCh:
51+ t.Fatalf("Recv() error = %v", err)
52+ case got := <-gotCh:
53+ if !reflect.DeepEqual(got, want) {
54+ t.Fatalf("carried frame = %#v, want %#v", got, want)
55+ }
56+ case <-time.After(time.Second):
57+ t.Fatal("Recv() did not receive deterministic simulated datagram")
58+ }
59+ if err := closer.Close(); err != nil {
60+ t.Fatalf("trace Close() error = %v", err)
61+ }
62+ trace, err := os.ReadFile(tracePath)
63+ if err != nil {
64+ t.Fatalf("ReadFile(trace): %v", err)
65+ }
66+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
67+ if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
68+ t.Errorf("trace missing %s", kind)
69+ }
70+ }
71+}
72+
73+type dgramObserverClock struct{ now time.Time }
74+
75+func (c dgramObserverClock) Now() time.Time { return c.now }
76+func (c dgramObserverClock) NewTimer(d time.Duration) ports.Timer {
77+ return dgramObserverTimer{timer: time.NewTimer(d)}
78+}
79+
80+type dgramObserverTimer struct{ timer *time.Timer }
81+
82+func (t dgramObserverTimer) C() <-chan time.Time { return t.timer.C }
83+func (t dgramObserverTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
84+func (t dgramObserverTimer) Stop() bool { return t.timer.Stop() }
A internal/adapters/ipc/transport_observability_test.go +69 -0
@@ -0,0 +1,69 @@
1+package ipc
2+
3+import (
4+ "net"
5+ "os"
6+ "reflect"
7+ "strings"
8+ "sync"
9+ "testing"
10+ "time"
11+
12+ "github.com/bnema/vev/internal/adapters/observability"
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {
17+ trace := t.TempDir() + "/ipc.jsonl"
18+ observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")
19+ if err != nil {
20+ t.Fatalf("NewJSONL() error = %v", err)
21+ }
22+
23+ left, right := net.Pipe()
24+ defer func() { _ = left.Close() }()
25+ defer func() { _ = right.Close() }()
26+ client := NewTransport(left, WithRuntimeObserver(observer))
27+ server := NewTransport(right, WithRuntimeObserver(observer))
28+ want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("wire bytes stay exact")}
29+
30+ var wg sync.WaitGroup
31+ wg.Go(func() {
32+ if err := client.Send(want); err != nil {
33+ t.Errorf("Send() error = %v", err)
34+ }
35+ })
36+ got, err := server.Recv()
37+ wg.Wait()
38+ if err != nil {
39+ t.Fatalf("Recv() error = %v", err)
40+ }
41+ if !reflect.DeepEqual(got, want) {
42+ t.Fatalf("carried frame = %#v, want %#v", got, want)
43+ }
44+ if err := closer.Close(); err != nil {
45+ t.Fatalf("trace Close() error = %v", err)
46+ }
47+ traceBytes, err := os.ReadFile(trace)
48+ if err != nil {
49+ t.Fatalf("ReadFile(trace): %v", err)
50+ }
51+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
52+ if !strings.Contains(string(traceBytes), `"kind":"`+kind+`"`) {
53+ t.Errorf("trace missing %s", kind)
54+ }
55+ }
56+}
57+
58+type ipcRuntimeClock struct{ now time.Time }
59+
60+func (c ipcRuntimeClock) Now() time.Time { return c.now }
61+func (c ipcRuntimeClock) NewTimer(d time.Duration) ports.Timer {
62+ return ipcRuntimeTimer{timer: time.NewTimer(d)}
63+}
64+
65+type ipcRuntimeTimer struct{ timer *time.Timer }
66+
67+func (t ipcRuntimeTimer) C() <-chan time.Time { return t.timer.C }
68+func (t ipcRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
69+func (t ipcRuntimeTimer) Stop() bool { return t.timer.Stop() }
@@ -0,0 +1,69 @@
1+package ipc
2+
3+import (
4+ "net"
5+ "os"
6+ "reflect"
7+ "strings"
8+ "sync"
9+ "testing"
10+ "time"
11+
12+ "github.com/bnema/vev/internal/adapters/observability"
13+ "github.com/bnema/vev/internal/ports"
14+)
15+
16+func TestTransportObservabilityIPCMarksCarriageWithoutChangingBytes(t *testing.T) {
17+ trace := t.TempDir() + "/ipc.jsonl"
18+ observer, closer, err := observability.NewJSONL(trace, ipcRuntimeClock{now: time.Unix(0, 211)}, "ipc-process")
19+ if err != nil {
20+ t.Fatalf("NewJSONL() error = %v", err)
21+ }
22+
23+ left, right := net.Pipe()
24+ defer func() { _ = left.Close() }()
25+ defer func() { _ = right.Close() }()
26+ client := NewTransport(left, WithRuntimeObserver(observer))
27+ server := NewTransport(right, WithRuntimeObserver(observer))
28+ want := ports.Frame{Type: ports.MsgOutput, Payload: []byte("wire bytes stay exact")}
29+
30+ var wg sync.WaitGroup
31+ wg.Go(func() {
32+ if err := client.Send(want); err != nil {
33+ t.Errorf("Send() error = %v", err)
34+ }
35+ })
36+ got, err := server.Recv()
37+ wg.Wait()
38+ if err != nil {
39+ t.Fatalf("Recv() error = %v", err)
40+ }
41+ if !reflect.DeepEqual(got, want) {
42+ t.Fatalf("carried frame = %#v, want %#v", got, want)
43+ }
44+ if err := closer.Close(); err != nil {
45+ t.Fatalf("trace Close() error = %v", err)
46+ }
47+ traceBytes, err := os.ReadFile(trace)
48+ if err != nil {
49+ t.Fatalf("ReadFile(trace): %v", err)
50+ }
51+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
52+ if !strings.Contains(string(traceBytes), `"kind":"`+kind+`"`) {
53+ t.Errorf("trace missing %s", kind)
54+ }
55+ }
56+}
57+
58+type ipcRuntimeClock struct{ now time.Time }
59+
60+func (c ipcRuntimeClock) Now() time.Time { return c.now }
61+func (c ipcRuntimeClock) NewTimer(d time.Duration) ports.Timer {
62+ return ipcRuntimeTimer{timer: time.NewTimer(d)}
63+}
64+
65+type ipcRuntimeTimer struct{ timer *time.Timer }
66+
67+func (t ipcRuntimeTimer) C() <-chan time.Time { return t.timer.C }
68+func (t ipcRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
69+func (t ipcRuntimeTimer) Stop() bool { return t.timer.Stop() }
A internal/adapters/observability/jsonl_test.go +178 -0
@@ -0,0 +1,178 @@
1+package observability
2+
3+import (
4+ "bufio"
5+ "encoding/json"
6+ "os"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilityJSONLOwnsTicksAndSchema(t *testing.T) {
15+ path := t.TempDir() + "/daemon.jsonl"
16+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 41)}, "daemon-process")
17+ if err != nil {
18+ t.Fatalf("NewJSONL() error = %v", err)
19+ }
20+
21+ observer.ObserveRuntime(validRuntimeMark(ports.RuntimeComposeStart))
22+ // A producer's clock domain must never enter a JSONL record. Rejection is
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})
25+ if err := closer.Close(); err != nil {
26+ t.Fatalf("Close() error = %v", err)
27+ }
28+
29+ records := readJSONLRecords(t, path)
30+ if len(records) != 1 {
31+ t.Fatalf("JSONL records = %d, want 1 (caller Tick must be rejected)", len(records))
32+ }
33+ wantKeys := map[string]bool{
34+ "schema": true, "process_id": true, "component": true, "scenario": true,
35+ "run": true, "sequence": true, "request_id": true, "epoch": true,
36+ "kind": true, "tick": true, "bytes": true, "fragments": true,
37+ "retransmits": true, "pending": true, "ack_rtt_nanos": true, "valid": true,
38+ }
39+ if len(records[0]) != len(wantKeys) {
40+ t.Fatalf("JSONL fields = %#v, want exactly canonical schema", records[0])
41+ }
42+ for key := range wantKeys {
43+ if _, ok := records[0][key]; !ok {
44+ t.Errorf("JSONL missing %q", key)
45+ }
46+ }
47+ if got := records[0]["tick"]; got != float64(41) {
48+ t.Errorf("JSONL tick = %v, want observer clock tick 41", got)
49+ }
50+ if got := records[0]["process_id"]; got != "daemon-process" {
51+ t.Errorf("JSONL process_id = %v, want observer process", got)
52+ }
53+}
54+
55+func TestTransportObservabilityJSONLRejectsInvalidRequiredFields(t *testing.T) {
56+ for _, mark := range []ports.RuntimeMark{
57+ {},
58+ {Schema: 1, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart},
59+ {Schema: 2, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart, Valid: true},
60+ } {
61+ path := t.TempDir() + "/invalid.jsonl"
62+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 5)}, "p")
63+ if err != nil {
64+ t.Fatalf("NewJSONL() error = %v", err)
65+ }
66+ observer.ObserveRuntime(mark)
67+ if err := closer.Close(); err != nil {
68+ t.Fatalf("Close() error = %v", err)
69+ }
70+ if records := readJSONLRecords(t, path); len(records) != 0 {
71+ t.Errorf("invalid mark %#v wrote %#v", mark, records)
72+ }
73+ }
74+}
75+
76+func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {
77+ for _, tc := range []struct {
78+ name string
79+ start ports.RuntimeMarkKind
80+ end ports.RuntimeMarkKind
81+ field string
82+ }{
83+ {"diff", ports.RuntimeDiffStart, ports.RuntimeDiffEnd, "diff_duration"},
84+ {"queue", ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued, "queue_wait"},
85+ {"ack", ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd, "ack_blocked_interval"},
86+ {"send", ports.RuntimeAdapterSendStart, ports.RuntimeAdapterSendEnd, "adapter_send_duration"},
87+ {"receive", ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, "adapter_receive_duration"},
88+ } {
89+ t.Run(tc.name, func(t *testing.T) {
90+ start := validRuntimeMark(tc.start)
91+ start.Tick = 11
92+ end := start
93+ end.Kind, end.Tick = tc.end, 19
94+ if got, err := sameProcessDuration(start, end); err != nil || got != 8 {
95+ t.Fatalf("%s = %d, %v; want 8, nil", tc.field, got, err)
96+ }
97+
98+ end.ProcessID = "other-process"
99+ if _, err := sameProcessDuration(start, end); err == nil {
100+ t.Fatalf("%s accepted cross-process ticks", tc.field)
101+ }
102+ })
103+ }
104+}
105+
106+func TestTransportObservabilityJSONLSerializesConcurrentMarks(t *testing.T) {
107+ path := t.TempDir() + "/concurrent.jsonl"
108+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 73)}, "ipc-process")
109+ if err != nil {
110+ t.Fatalf("NewJSONL() error = %v", err)
111+ }
112+ var wg sync.WaitGroup
113+ for i := uint64(0); i < 32; i++ {
114+ wg.Add(1)
115+ go func(sequence uint64) {
116+ defer wg.Done()
117+ mark := validRuntimeMark(ports.RuntimeAdapterSendEnd)
118+ mark.Sequence = sequence + 1
119+ observer.ObserveRuntime(mark)
120+ }(i)
121+ }
122+ wg.Wait()
123+ if err := closer.Close(); err != nil {
124+ t.Fatalf("Close() error = %v", err)
125+ }
126+ if records := readJSONLRecords(t, path); len(records) != 32 {
127+ t.Fatalf("concurrently serialized records = %d, want 32", len(records))
128+ }
129+}
130+
131+func sameProcessDuration(start, end ports.RuntimeMark) (int64, error) {
132+ if start.ProcessID == "" || start.ProcessID != end.ProcessID {
133+ return 0, os.ErrInvalid
134+ }
135+ if start.Scenario != end.Scenario || start.Run != end.Run || start.Sequence != end.Sequence || start.RequestID != end.RequestID || start.Epoch != end.Epoch || end.Tick < start.Tick {
136+ return 0, os.ErrInvalid
137+ }
138+ return end.Tick - start.Tick, nil
139+}
140+
141+func validRuntimeMark(kind ports.RuntimeMarkKind) ports.RuntimeMark {
142+ return ports.RuntimeMark{Schema: 1, ProcessID: "daemon-process", Component: "daemon", Scenario: "resize", Run: 7, Sequence: 9, RequestID: 11, Epoch: 13, Kind: kind, Valid: true}
143+}
144+
145+func readJSONLRecords(t *testing.T, path string) []map[string]any {
146+ t.Helper()
147+ f, err := os.Open(path)
148+ if err != nil {
149+ t.Fatalf("Open(%q): %v", path, err)
150+ }
151+ defer func() { _ = f.Close() }()
152+ var records []map[string]any
153+ scanner := bufio.NewScanner(f)
154+ for scanner.Scan() {
155+ var record map[string]any
156+ if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
157+ t.Fatalf("invalid JSONL record %q: %v", scanner.Text(), err)
158+ }
159+ records = append(records, record)
160+ }
161+ if err := scanner.Err(); err != nil {
162+ t.Fatalf("reading JSONL: %v", err)
163+ }
164+ return records
165+}
166+
167+type fixedRuntimeClock struct{ now time.Time }
168+
169+func (c fixedRuntimeClock) Now() time.Time { return c.now }
170+func (c fixedRuntimeClock) NewTimer(d time.Duration) ports.Timer {
171+ return fixedRuntimeTimer{timer: time.NewTimer(d)}
172+}
173+
174+type fixedRuntimeTimer struct{ timer *time.Timer }
175+
176+func (t fixedRuntimeTimer) C() <-chan time.Time { return t.timer.C }
177+func (t fixedRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
178+func (t fixedRuntimeTimer) Stop() bool { return t.timer.Stop() }
@@ -0,0 +1,178 @@
1+package observability
2+
3+import (
4+ "bufio"
5+ "encoding/json"
6+ "os"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilityJSONLOwnsTicksAndSchema(t *testing.T) {
15+ path := t.TempDir() + "/daemon.jsonl"
16+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 41)}, "daemon-process")
17+ if err != nil {
18+ t.Fatalf("NewJSONL() error = %v", err)
19+ }
20+
21+ observer.ObserveRuntime(validRuntimeMark(ports.RuntimeComposeStart))
22+ // A producer's clock domain must never enter a JSONL record. Rejection is
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})
25+ if err := closer.Close(); err != nil {
26+ t.Fatalf("Close() error = %v", err)
27+ }
28+
29+ records := readJSONLRecords(t, path)
30+ if len(records) != 1 {
31+ t.Fatalf("JSONL records = %d, want 1 (caller Tick must be rejected)", len(records))
32+ }
33+ wantKeys := map[string]bool{
34+ "schema": true, "process_id": true, "component": true, "scenario": true,
35+ "run": true, "sequence": true, "request_id": true, "epoch": true,
36+ "kind": true, "tick": true, "bytes": true, "fragments": true,
37+ "retransmits": true, "pending": true, "ack_rtt_nanos": true, "valid": true,
38+ }
39+ if len(records[0]) != len(wantKeys) {
40+ t.Fatalf("JSONL fields = %#v, want exactly canonical schema", records[0])
41+ }
42+ for key := range wantKeys {
43+ if _, ok := records[0][key]; !ok {
44+ t.Errorf("JSONL missing %q", key)
45+ }
46+ }
47+ if got := records[0]["tick"]; got != float64(41) {
48+ t.Errorf("JSONL tick = %v, want observer clock tick 41", got)
49+ }
50+ if got := records[0]["process_id"]; got != "daemon-process" {
51+ t.Errorf("JSONL process_id = %v, want observer process", got)
52+ }
53+}
54+
55+func TestTransportObservabilityJSONLRejectsInvalidRequiredFields(t *testing.T) {
56+ for _, mark := range []ports.RuntimeMark{
57+ {},
58+ {Schema: 1, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart},
59+ {Schema: 2, ProcessID: "p", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, Kind: ports.RuntimeEmitStart, Valid: true},
60+ } {
61+ path := t.TempDir() + "/invalid.jsonl"
62+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 5)}, "p")
63+ if err != nil {
64+ t.Fatalf("NewJSONL() error = %v", err)
65+ }
66+ observer.ObserveRuntime(mark)
67+ if err := closer.Close(); err != nil {
68+ t.Fatalf("Close() error = %v", err)
69+ }
70+ if records := readJSONLRecords(t, path); len(records) != 0 {
71+ t.Errorf("invalid mark %#v wrote %#v", mark, records)
72+ }
73+ }
74+}
75+
76+func TestTransportObservabilitySameProcessSpanClockDomains(t *testing.T) {
77+ for _, tc := range []struct {
78+ name string
79+ start ports.RuntimeMarkKind
80+ end ports.RuntimeMarkKind
81+ field string
82+ }{
83+ {"diff", ports.RuntimeDiffStart, ports.RuntimeDiffEnd, "diff_duration"},
84+ {"queue", ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued, "queue_wait"},
85+ {"ack", ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd, "ack_blocked_interval"},
86+ {"send", ports.RuntimeAdapterSendStart, ports.RuntimeAdapterSendEnd, "adapter_send_duration"},
87+ {"receive", ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, "adapter_receive_duration"},
88+ } {
89+ t.Run(tc.name, func(t *testing.T) {
90+ start := validRuntimeMark(tc.start)
91+ start.Tick = 11
92+ end := start
93+ end.Kind, end.Tick = tc.end, 19
94+ if got, err := sameProcessDuration(start, end); err != nil || got != 8 {
95+ t.Fatalf("%s = %d, %v; want 8, nil", tc.field, got, err)
96+ }
97+
98+ end.ProcessID = "other-process"
99+ if _, err := sameProcessDuration(start, end); err == nil {
100+ t.Fatalf("%s accepted cross-process ticks", tc.field)
101+ }
102+ })
103+ }
104+}
105+
106+func TestTransportObservabilityJSONLSerializesConcurrentMarks(t *testing.T) {
107+ path := t.TempDir() + "/concurrent.jsonl"
108+ observer, closer, err := NewJSONL(path, fixedRuntimeClock{now: time.Unix(0, 73)}, "ipc-process")
109+ if err != nil {
110+ t.Fatalf("NewJSONL() error = %v", err)
111+ }
112+ var wg sync.WaitGroup
113+ for i := uint64(0); i < 32; i++ {
114+ wg.Add(1)
115+ go func(sequence uint64) {
116+ defer wg.Done()
117+ mark := validRuntimeMark(ports.RuntimeAdapterSendEnd)
118+ mark.Sequence = sequence + 1
119+ observer.ObserveRuntime(mark)
120+ }(i)
121+ }
122+ wg.Wait()
123+ if err := closer.Close(); err != nil {
124+ t.Fatalf("Close() error = %v", err)
125+ }
126+ if records := readJSONLRecords(t, path); len(records) != 32 {
127+ t.Fatalf("concurrently serialized records = %d, want 32", len(records))
128+ }
129+}
130+
131+func sameProcessDuration(start, end ports.RuntimeMark) (int64, error) {
132+ if start.ProcessID == "" || start.ProcessID != end.ProcessID {
133+ return 0, os.ErrInvalid
134+ }
135+ if start.Scenario != end.Scenario || start.Run != end.Run || start.Sequence != end.Sequence || start.RequestID != end.RequestID || start.Epoch != end.Epoch || end.Tick < start.Tick {
136+ return 0, os.ErrInvalid
137+ }
138+ return end.Tick - start.Tick, nil
139+}
140+
141+func validRuntimeMark(kind ports.RuntimeMarkKind) ports.RuntimeMark {
142+ return ports.RuntimeMark{Schema: 1, ProcessID: "daemon-process", Component: "daemon", Scenario: "resize", Run: 7, Sequence: 9, RequestID: 11, Epoch: 13, Kind: kind, Valid: true}
143+}
144+
145+func readJSONLRecords(t *testing.T, path string) []map[string]any {
146+ t.Helper()
147+ f, err := os.Open(path)
148+ if err != nil {
149+ t.Fatalf("Open(%q): %v", path, err)
150+ }
151+ defer func() { _ = f.Close() }()
152+ var records []map[string]any
153+ scanner := bufio.NewScanner(f)
154+ for scanner.Scan() {
155+ var record map[string]any
156+ if err := json.Unmarshal(scanner.Bytes(), &record); err != nil {
157+ t.Fatalf("invalid JSONL record %q: %v", scanner.Text(), err)
158+ }
159+ records = append(records, record)
160+ }
161+ if err := scanner.Err(); err != nil {
162+ t.Fatalf("reading JSONL: %v", err)
163+ }
164+ return records
165+}
166+
167+type fixedRuntimeClock struct{ now time.Time }
168+
169+func (c fixedRuntimeClock) Now() time.Time { return c.now }
170+func (c fixedRuntimeClock) NewTimer(d time.Duration) ports.Timer {
171+ return fixedRuntimeTimer{timer: time.NewTimer(d)}
172+}
173+
174+type fixedRuntimeTimer struct{ timer *time.Timer }
175+
176+func (t fixedRuntimeTimer) C() <-chan time.Time { return t.timer.C }
177+func (t fixedRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
178+func (t fixedRuntimeTimer) Stop() bool { return t.timer.Stop() }
A internal/adapters/sshstdio/transport_observability_test.go +64 -0
@@ -0,0 +1,64 @@
1+package sshstdio
2+
3+import (
4+ "bytes"
5+ "os"
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/bnema/vev/internal/adapters/observability"
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
15+ tracePath := t.TempDir() + "/ssh.jsonl"
16+ observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")
17+ if err != nil {
18+ t.Fatalf("NewJSONL() error = %v", err)
19+ }
20+
21+ frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte("stdio bytes stay exact")}
22+ var observed, baseline bytes.Buffer
23+ if err := NewTransport(bytes.NewReader(nil), &observed, nil, WithRuntimeObserver(observer)).Send(frame); err != nil {
24+ t.Fatalf("observed Send() error = %v", err)
25+ }
26+ if err := NewTransport(bytes.NewReader(nil), &baseline, nil).Send(frame); err != nil {
27+ t.Fatalf("baseline Send() error = %v", err)
28+ }
29+ if !bytes.Equal(observed.Bytes(), baseline.Bytes()) {
30+ t.Fatalf("observer changed SSH stdio wire bytes: got %x, want %x", observed.Bytes(), baseline.Bytes())
31+ }
32+ received, err := NewTransport(bytes.NewReader(baseline.Bytes()), &bytes.Buffer{}, nil, WithRuntimeObserver(observer)).Recv()
33+ if err != nil {
34+ t.Fatalf("observed Recv() error = %v", err)
35+ }
36+ if received.Type != frame.Type || !bytes.Equal(received.Payload, frame.Payload) {
37+ t.Fatalf("observed Recv() = %#v, want %#v", received, frame)
38+ }
39+ if err := closer.Close(); err != nil {
40+ t.Fatalf("trace Close() error = %v", err)
41+ }
42+ trace, err := os.ReadFile(tracePath)
43+ if err != nil {
44+ t.Fatalf("ReadFile(trace): %v", err)
45+ }
46+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
47+ if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
48+ t.Errorf("trace missing %s", kind)
49+ }
50+ }
51+}
52+
53+type sshRuntimeClock struct{ now time.Time }
54+
55+func (c sshRuntimeClock) Now() time.Time { return c.now }
56+func (c sshRuntimeClock) NewTimer(d time.Duration) ports.Timer {
57+ return sshRuntimeTimer{timer: time.NewTimer(d)}
58+}
59+
60+type sshRuntimeTimer struct{ timer *time.Timer }
61+
62+func (t sshRuntimeTimer) C() <-chan time.Time { return t.timer.C }
63+func (t sshRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
64+func (t sshRuntimeTimer) Stop() bool { return t.timer.Stop() }
@@ -0,0 +1,64 @@
1+package sshstdio
2+
3+import (
4+ "bytes"
5+ "os"
6+ "strings"
7+ "testing"
8+ "time"
9+
10+ "github.com/bnema/vev/internal/adapters/observability"
11+ "github.com/bnema/vev/internal/ports"
12+)
13+
14+func TestTransportObservabilitySSHStdioPreservesCarriage(t *testing.T) {
15+ tracePath := t.TempDir() + "/ssh.jsonl"
16+ observer, closer, err := observability.NewJSONL(tracePath, sshRuntimeClock{now: time.Unix(0, 307)}, "ssh-process")
17+ if err != nil {
18+ t.Fatalf("NewJSONL() error = %v", err)
19+ }
20+
21+ frame := ports.Frame{Type: ports.MsgOutput, Payload: []byte("stdio bytes stay exact")}
22+ var observed, baseline bytes.Buffer
23+ if err := NewTransport(bytes.NewReader(nil), &observed, nil, WithRuntimeObserver(observer)).Send(frame); err != nil {
24+ t.Fatalf("observed Send() error = %v", err)
25+ }
26+ if err := NewTransport(bytes.NewReader(nil), &baseline, nil).Send(frame); err != nil {
27+ t.Fatalf("baseline Send() error = %v", err)
28+ }
29+ if !bytes.Equal(observed.Bytes(), baseline.Bytes()) {
30+ t.Fatalf("observer changed SSH stdio wire bytes: got %x, want %x", observed.Bytes(), baseline.Bytes())
31+ }
32+ received, err := NewTransport(bytes.NewReader(baseline.Bytes()), &bytes.Buffer{}, nil, WithRuntimeObserver(observer)).Recv()
33+ if err != nil {
34+ t.Fatalf("observed Recv() error = %v", err)
35+ }
36+ if received.Type != frame.Type || !bytes.Equal(received.Payload, frame.Payload) {
37+ t.Fatalf("observed Recv() = %#v, want %#v", received, frame)
38+ }
39+ if err := closer.Close(); err != nil {
40+ t.Fatalf("trace Close() error = %v", err)
41+ }
42+ trace, err := os.ReadFile(tracePath)
43+ if err != nil {
44+ t.Fatalf("ReadFile(trace): %v", err)
45+ }
46+ for _, kind := range []string{"adapter_send_start", "adapter_send_end", "adapter_receive_start", "adapter_receive_end"} {
47+ if !strings.Contains(string(trace), `"kind":"`+kind+`"`) {
48+ t.Errorf("trace missing %s", kind)
49+ }
50+ }
51+}
52+
53+type sshRuntimeClock struct{ now time.Time }
54+
55+func (c sshRuntimeClock) Now() time.Time { return c.now }
56+func (c sshRuntimeClock) NewTimer(d time.Duration) ports.Timer {
57+ return sshRuntimeTimer{timer: time.NewTimer(d)}
58+}
59+
60+type sshRuntimeTimer struct{ timer *time.Timer }
61+
62+func (t sshRuntimeTimer) C() <-chan time.Time { return t.timer.C }
63+func (t sshRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
64+func (t sshRuntimeTimer) Stop() bool { return t.timer.Stop() }
A internal/app/transport_observability_test.go +28 -0
@@ -0,0 +1,28 @@
1+package app
2+
3+import (
4+ "os"
5+ "strings"
6+ "testing"
7+)
8+
9+func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
10+ source, err := os.ReadFile("run.go")
11+ if err != nil {
12+ t.Fatalf("ReadFile(run.go): %v", err)
13+ }
14+ text := string(source)
15+ for _, required := range []string{
16+ "VEV_PERF_TRACE",
17+ "VEV_PERF_PROCESS_ID",
18+ "observability.NewJSONL",
19+ "WithRuntimeObserver",
20+ } {
21+ if !strings.Contains(text, required) {
22+ t.Errorf("runtime trace wiring missing %q", required)
23+ }
24+ }
25+ if strings.Contains(text, "WithRuntimeObserver(clock") || strings.Contains(text, "WithRuntimeObserver(clk") {
26+ t.Error("consumer observer configuration must not accept a clock")
27+ }
28+}
@@ -0,0 +1,28 @@
1+package app
2+
3+import (
4+ "os"
5+ "strings"
6+ "testing"
7+)
8+
9+func TestPerformanceTraceAppWiresOneObserverPerProcess(t *testing.T) {
10+ source, err := os.ReadFile("run.go")
11+ if err != nil {
12+ t.Fatalf("ReadFile(run.go): %v", err)
13+ }
14+ text := string(source)
15+ for _, required := range []string{
16+ "VEV_PERF_TRACE",
17+ "VEV_PERF_PROCESS_ID",
18+ "observability.NewJSONL",
19+ "WithRuntimeObserver",
20+ } {
21+ if !strings.Contains(text, required) {
22+ t.Errorf("runtime trace wiring missing %q", required)
23+ }
24+ }
25+ if strings.Contains(text, "WithRuntimeObserver(clock") || strings.Contains(text, "WithRuntimeObserver(clk") {
26+ t.Error("consumer observer configuration must not accept a clock")
27+ }
28+}
A internal/ports/runtime_observer_test.go +74 -0
@@ -0,0 +1,74 @@
1+package ports
2+
3+import (
4+ "reflect"
5+ "testing"
6+)
7+
8+func TestRuntimeObserverContract(t *testing.T) {
9+ wantFields := []struct {
10+ name string
11+ typ reflect.Type
12+ }{
13+ {"Schema", reflect.TypeFor[uint16]()},
14+ {"ProcessID", reflect.TypeFor[string]()},
15+ {"Component", reflect.TypeFor[string]()},
16+ {"Scenario", reflect.TypeFor[string]()},
17+ {"Run", reflect.TypeFor[uint64]()},
18+ {"Sequence", reflect.TypeFor[uint64]()},
19+ {"RequestID", reflect.TypeFor[uint64]()},
20+ {"Epoch", reflect.TypeFor[uint64]()},
21+ {"Kind", reflect.TypeFor[RuntimeMarkKind]()},
22+ {"Tick", reflect.TypeFor[int64]()},
23+ {"Bytes", reflect.TypeFor[uint64]()},
24+ {"Fragments", reflect.TypeFor[uint64]()},
25+ {"Retransmits", reflect.TypeFor[uint64]()},
26+ {"Pending", reflect.TypeFor[uint64]()},
27+ {"AckRTTNanos", reflect.TypeFor[int64]()},
28+ {"Valid", reflect.TypeFor[bool]()},
29+ }
30+
31+ typ := reflect.TypeFor[RuntimeMark]()
32+ if typ.NumField() != len(wantFields) {
33+ t.Fatalf("RuntimeMark has %d fields, want exactly %d", typ.NumField(), len(wantFields))
34+ }
35+ for i, want := range wantFields {
36+ field := typ.Field(i)
37+ if field.Name != want.name || field.Type != want.typ {
38+ t.Fatalf("RuntimeMark field %d = %s %v, want %s %v", i, field.Name, field.Type, want.name, want.typ)
39+ }
40+ }
41+
42+ var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})
43+}
44+
45+func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
46+ pairs := []struct {
47+ start RuntimeMarkKind
48+ end RuntimeMarkKind
49+ }{
50+ {RuntimeDiffStart, RuntimeDiffEnd},
51+ {RuntimeQueueEnqueued, RuntimeQueueDequeued},
52+ {RuntimeACKBlockedStart, RuntimeACKBlockedEnd},
53+ {RuntimeAdapterSendStart, RuntimeAdapterSendEnd},
54+ {RuntimeAdapterReceiveStart, RuntimeAdapterReceiveEnd},
55+ }
56+ for _, pair := range pairs {
57+ if pair.start == "" || pair.end == "" || pair.start == pair.end {
58+ t.Fatalf("invalid required span mapping %#v", pair)
59+ }
60+ }
61+}
62+
63+func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
64+ typ := reflect.TypeFor[RuntimeMark]()
65+ for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {
66+ if _, ok := typ.FieldByName(forbidden); ok {
67+ t.Fatalf("RuntimeMark must not carry %s", forbidden)
68+ }
69+ }
70+}
71+
72+type runtimeObserverFunc func(RuntimeMark)
73+
74+func (f runtimeObserverFunc) ObserveRuntime(mark RuntimeMark) { f(mark) }
@@ -0,0 +1,74 @@
1+package ports
2+
3+import (
4+ "reflect"
5+ "testing"
6+)
7+
8+func TestRuntimeObserverContract(t *testing.T) {
9+ wantFields := []struct {
10+ name string
11+ typ reflect.Type
12+ }{
13+ {"Schema", reflect.TypeFor[uint16]()},
14+ {"ProcessID", reflect.TypeFor[string]()},
15+ {"Component", reflect.TypeFor[string]()},
16+ {"Scenario", reflect.TypeFor[string]()},
17+ {"Run", reflect.TypeFor[uint64]()},
18+ {"Sequence", reflect.TypeFor[uint64]()},
19+ {"RequestID", reflect.TypeFor[uint64]()},
20+ {"Epoch", reflect.TypeFor[uint64]()},
21+ {"Kind", reflect.TypeFor[RuntimeMarkKind]()},
22+ {"Tick", reflect.TypeFor[int64]()},
23+ {"Bytes", reflect.TypeFor[uint64]()},
24+ {"Fragments", reflect.TypeFor[uint64]()},
25+ {"Retransmits", reflect.TypeFor[uint64]()},
26+ {"Pending", reflect.TypeFor[uint64]()},
27+ {"AckRTTNanos", reflect.TypeFor[int64]()},
28+ {"Valid", reflect.TypeFor[bool]()},
29+ }
30+
31+ typ := reflect.TypeFor[RuntimeMark]()
32+ if typ.NumField() != len(wantFields) {
33+ t.Fatalf("RuntimeMark has %d fields, want exactly %d", typ.NumField(), len(wantFields))
34+ }
35+ for i, want := range wantFields {
36+ field := typ.Field(i)
37+ if field.Name != want.name || field.Type != want.typ {
38+ t.Fatalf("RuntimeMark field %d = %s %v, want %s %v", i, field.Name, field.Type, want.name, want.typ)
39+ }
40+ }
41+
42+ var _ RuntimeObserver = runtimeObserverFunc(func(RuntimeMark) {})
43+}
44+
45+func TestRuntimeObserverRequiredSpanKinds(t *testing.T) {
46+ pairs := []struct {
47+ start RuntimeMarkKind
48+ end RuntimeMarkKind
49+ }{
50+ {RuntimeDiffStart, RuntimeDiffEnd},
51+ {RuntimeQueueEnqueued, RuntimeQueueDequeued},
52+ {RuntimeACKBlockedStart, RuntimeACKBlockedEnd},
53+ {RuntimeAdapterSendStart, RuntimeAdapterSendEnd},
54+ {RuntimeAdapterReceiveStart, RuntimeAdapterReceiveEnd},
55+ }
56+ for _, pair := range pairs {
57+ if pair.start == "" || pair.end == "" || pair.start == pair.end {
58+ t.Fatalf("invalid required span mapping %#v", pair)
59+ }
60+ }
61+}
62+
63+func TestRuntimeMarkHasNoRenderingPayloadOrPolicyFields(t *testing.T) {
64+ typ := reflect.TypeFor[RuntimeMark]()
65+ for _, forbidden := range []string{"Cell", "Cells", "Payload", "Frame", "Renderer", "Policy", "Transport"} {
66+ if _, ok := typ.FieldByName(forbidden); ok {
67+ t.Fatalf("RuntimeMark must not carry %s", forbidden)
68+ }
69+ }
70+}
71+
72+type runtimeObserverFunc func(RuntimeMark)
73+
74+func (f runtimeObserverFunc) ObserveRuntime(mark RuntimeMark) { f(mark) }
A internal/usecase/client/transport_observability_test.go +68 -0
@@ -0,0 +1,68 @@
1+package client_test
2+
3+import (
4+ "bytes"
5+ "context"
6+ "sync"
7+ "sync/atomic"
8+ "testing"
9+
10+ "github.com/stretchr/testify/require"
11+
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/ports"
14+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
15+ "github.com/bnema/vev/internal/usecase/client"
16+)
17+
18+func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
19+ var out bytes.Buffer
20+ var restores atomic.Int32
21+ resizeCh := make(chan domain.Size)
22+ term, in := newHappyTerminal(t, &out, &restores, resizeCh)
23+ defer in.unblock()
24+
25+ observer := &clientRuntimeObserver{}
26+ transport := portsmocks.NewMockTransport(t)
27+ transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
28+ unblock := scriptRecv(transport,
29+ recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
30+ recvItem{f: frameOf(ports.MsgOutput, ports.MarshalOutput(ports.Output{Data: []byte("unchanged-by-observer"), NewStateNum: 3}))},
31+ recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
32+ )
33+ defer unblock()
34+ transport.EXPECT().Send(isType(ports.MsgAck)).Return(nil).Maybe()
35+ transport.EXPECT().Close().Return(nil).Once()
36+
37+ err := client.Attach(context.Background(), transport, term, realClock{}, ports.IntentEphemeral, "", client.WithRuntimeObserver(observer))
38+ require.NoError(t, err)
39+ require.Equal(t, "unchanged-by-observer", out.String(), "observer must not alter terminal bytes")
40+ require.Equal(t, int32(1), restores.Load())
41+ observer.requireOrdered(t, ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, ports.RuntimeTerminalFlushed)
42+}
43+
44+type clientRuntimeObserver struct {
45+ mu sync.Mutex
46+ marks []ports.RuntimeMark
47+}
48+
49+func (o *clientRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
50+ o.mu.Lock()
51+ defer o.mu.Unlock()
52+ o.marks = append(o.marks, mark)
53+}
54+
55+func (o *clientRuntimeObserver) requireOrdered(t *testing.T, kinds ...ports.RuntimeMarkKind) {
56+ t.Helper()
57+ o.mu.Lock()
58+ defer o.mu.Unlock()
59+ next := 0
60+ for _, mark := range o.marks {
61+ if next < len(kinds) && mark.Kind == kinds[next] {
62+ next++
63+ }
64+ }
65+ if next != len(kinds) {
66+ t.Fatalf("runtime marks %#v do not contain ordered %#v", o.marks, kinds)
67+ }
68+}
@@ -0,0 +1,68 @@
1+package client_test
2+
3+import (
4+ "bytes"
5+ "context"
6+ "sync"
7+ "sync/atomic"
8+ "testing"
9+
10+ "github.com/stretchr/testify/require"
11+
12+ "github.com/bnema/vev/internal/domain"
13+ "github.com/bnema/vev/internal/ports"
14+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
15+ "github.com/bnema/vev/internal/usecase/client"
16+)
17+
18+func TestTerminalFlushBoundaryTransportObservability(t *testing.T) {
19+ var out bytes.Buffer
20+ var restores atomic.Int32
21+ resizeCh := make(chan domain.Size)
22+ term, in := newHappyTerminal(t, &out, &restores, resizeCh)
23+ defer in.unblock()
24+
25+ observer := &clientRuntimeObserver{}
26+ transport := portsmocks.NewMockTransport(t)
27+ transport.EXPECT().Send(isType(ports.MsgHello)).Return(nil).Once()
28+ unblock := scriptRecv(transport,
29+ recvItem{f: frameOf(ports.MsgWelcome, ports.MarshalWelcome(ports.Welcome{SessionID: "s1"}))},
30+ recvItem{f: frameOf(ports.MsgOutput, ports.MarshalOutput(ports.Output{Data: []byte("unchanged-by-observer"), NewStateNum: 3}))},
31+ recvItem{f: frameOf(ports.MsgDetached, ports.MarshalDetached(ports.Detached{Reason: ports.ReasonDetach}))},
32+ )
33+ defer unblock()
34+ transport.EXPECT().Send(isType(ports.MsgAck)).Return(nil).Maybe()
35+ transport.EXPECT().Close().Return(nil).Once()
36+
37+ err := client.Attach(context.Background(), transport, term, realClock{}, ports.IntentEphemeral, "", client.WithRuntimeObserver(observer))
38+ require.NoError(t, err)
39+ require.Equal(t, "unchanged-by-observer", out.String(), "observer must not alter terminal bytes")
40+ require.Equal(t, int32(1), restores.Load())
41+ observer.requireOrdered(t, ports.RuntimeAdapterReceiveStart, ports.RuntimeAdapterReceiveEnd, ports.RuntimeTerminalFlushed)
42+}
43+
44+type clientRuntimeObserver struct {
45+ mu sync.Mutex
46+ marks []ports.RuntimeMark
47+}
48+
49+func (o *clientRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
50+ o.mu.Lock()
51+ defer o.mu.Unlock()
52+ o.marks = append(o.marks, mark)
53+}
54+
55+func (o *clientRuntimeObserver) requireOrdered(t *testing.T, kinds ...ports.RuntimeMarkKind) {
56+ t.Helper()
57+ o.mu.Lock()
58+ defer o.mu.Unlock()
59+ next := 0
60+ for _, mark := range o.marks {
61+ if next < len(kinds) && mark.Kind == kinds[next] {
62+ next++
63+ }
64+ }
65+ if next != len(kinds) {
66+ t.Fatalf("runtime marks %#v do not contain ordered %#v", o.marks, kinds)
67+ }
68+}
A internal/usecase/daemon/transport_observability_test.go +51 -0
@@ -0,0 +1,51 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+// This acceptance boundary deliberately configures the daemon with an
11+// observer only. The clock remains the daemon's existing behavior clock;
12+// timestamp ownership belongs to the concrete process observer.
13+func TestTransportObservabilityDaemonBoundaries(t *testing.T) {
14+ observer := &daemonRuntimeObserver{}
15+ d := New(nil, daemonRuntimeClock{now: time.Unix(0, 101)}, nil, WithRuntimeObserver(observer))
16+ if d == nil {
17+ t.Fatal("New() returned nil")
18+ }
19+
20+ want := []ports.RuntimeMarkKind{
21+ ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
22+ ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
23+ ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
24+ ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
25+ ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
26+ ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd,
27+ ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
28+ }
29+ if len(want) != 14 {
30+ t.Fatalf("daemon boundary mark inventory changed: %d", len(want))
31+ }
32+}
33+
34+type daemonRuntimeObserver struct{ marks []ports.RuntimeMark }
35+
36+func (o *daemonRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
37+ o.marks = append(o.marks, mark)
38+}
39+
40+type daemonRuntimeClock struct{ now time.Time }
41+
42+func (c daemonRuntimeClock) Now() time.Time { return c.now }
43+func (c daemonRuntimeClock) NewTimer(d time.Duration) ports.Timer {
44+ return daemonRuntimeTimer{timer: time.NewTimer(d)}
45+}
46+
47+type daemonRuntimeTimer struct{ timer *time.Timer }
48+
49+func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C }
50+func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
51+func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() }
@@ -0,0 +1,51 @@
1+package daemon
2+
3+import (
4+ "testing"
5+ "time"
6+
7+ "github.com/bnema/vev/internal/ports"
8+)
9+
10+// This acceptance boundary deliberately configures the daemon with an
11+// observer only. The clock remains the daemon's existing behavior clock;
12+// timestamp ownership belongs to the concrete process observer.
13+func TestTransportObservabilityDaemonBoundaries(t *testing.T) {
14+ observer := &daemonRuntimeObserver{}
15+ d := New(nil, daemonRuntimeClock{now: time.Unix(0, 101)}, nil, WithRuntimeObserver(observer))
16+ if d == nil {
17+ t.Fatal("New() returned nil")
18+ }
19+
20+ want := []ports.RuntimeMarkKind{
21+ ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
22+ ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
23+ ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
24+ ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
25+ ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
26+ ports.RuntimeACKBlockedStart, ports.RuntimeACKBlockedEnd,
27+ ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
28+ }
29+ if len(want) != 14 {
30+ t.Fatalf("daemon boundary mark inventory changed: %d", len(want))
31+ }
32+}
33+
34+type daemonRuntimeObserver struct{ marks []ports.RuntimeMark }
35+
36+func (o *daemonRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
37+ o.marks = append(o.marks, mark)
38+}
39+
40+type daemonRuntimeClock struct{ now time.Time }
41+
42+func (c daemonRuntimeClock) Now() time.Time { return c.now }
43+func (c daemonRuntimeClock) NewTimer(d time.Duration) ports.Timer {
44+ return daemonRuntimeTimer{timer: time.NewTimer(d)}
45+}
46+
47+type daemonRuntimeTimer struct{ timer *time.Timer }
48+
49+func (t daemonRuntimeTimer) C() <-chan time.Time { return t.timer.C }
50+func (t daemonRuntimeTimer) Reset(d time.Duration) bool { return t.timer.Reset(d) }
51+func (t daemonRuntimeTimer) Stop() bool { return t.timer.Stop() }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 13

↑↓ moveenter openesc close