1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "encoding/json"
8 "errors"
9 "fmt"
10 "os"
11 "path/filepath"
12 "testing"
13 "time"
14)
15
16type fakeClock struct{ tick int64 }
17
18func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick }
19
20func (c *fakeClock) Sleep(d time.Duration) {
21 c.tick += d.Nanoseconds()
22}
23
24type fakeProcess struct {
25 fail bool
26 measureErr error
27 closeErr error
28 warmups [][]byte
29 measures [][]byte
30 closed bool
31}
32
33func (p *fakeProcess) Warmup(input []byte) error {
34 p.warmups = append(p.warmups, append([]byte(nil), input...))
35 return nil
36}
37
38func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error {
39 p.measures = append(p.measures, append([]byte(nil), input...))
40 if p.measureErr != nil {
41 return p.measureErr
42 }
43 if p.fail {
44 return errors.New("measure failure")
45 }
46 if err := injected(); err != nil {
47 return err
48 }
49 return flush()
50}
51
52func (p *fakeProcess) Close() error { p.closed = true; return p.closeErr }
53
54type fakeLauncher struct {
55 mappings []processMapping
56 args [][]string
57 commands []roleCommand
58 process []*fakeProcess
59 measureErr map[string]error
60 closeErr map[string]error
61 manifestPresent bool
62}
63
64func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) {
65 l.mappings = append(l.mappings, m)
66 l.args = append(l.args, append([]string(nil), command.Args...))
67 l.commands = append(l.commands, command)
68 _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json"))
69 l.manifestPresent = e == nil
70 p := &fakeProcess{measureErr: l.measureErr[m.Role], closeErr: l.closeErr[m.Role]}
71 l.process = append(l.process, p)
72 return p, nil
73}
74
75type lateFlushProcess struct{ clock *fakeClock }
76
77func (*lateFlushProcess) Warmup([]byte) error { return nil }
78
79func (p *lateFlushProcess) Measure(_ []byte, injected, flushed func() error) error {
80 if err := injected(); err != nil {
81 return err
82 }
83 p.clock.tick += minimumDuration.Nanoseconds()
84 return flushed()
85}
86
87func (*lateFlushProcess) Close() error { return nil }
88
89type insufficientEventLauncher struct{ clock *fakeClock }
90
91func (l insufficientEventLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
92 if m.Role == "client" {
93 return &lateFlushProcess{clock: l.clock}, nil
94 }
95 return &fakeProcess{}, nil
96}
97
98type failedFlushProcess struct{ err error }
99
100func (*failedFlushProcess) Warmup([]byte) error { return nil }
101
102func (p *failedFlushProcess) Measure(_ []byte, injected, _ func() error) error {
103 if err := injected(); err != nil {
104 return err
105 }
106 return p.err
107}
108
109func (*failedFlushProcess) Close() error { return nil }
110
111type failedFlushLauncher struct{ err error }
112
113func (l failedFlushLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
114 if m.Role == "client" {
115 return &failedFlushProcess{err: l.err}, nil
116 }
117 return &fakeProcess{}, nil
118}
119
120func closeTestFile(t *testing.T, f *os.File) {
121 t.Helper()
122 t.Cleanup(func() {
123 if err := f.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
124 t.Error(err)
125 }
126 })
127}
128
129func removeTestTree(t *testing.T, path string) {
130 t.Helper()
131 t.Cleanup(func() {
132 if err := os.RemoveAll(path); err != nil {
133 t.Error(err)
134 }
135 })
136}
137
138func readHarnessMarks(t *testing.T, path string) []harnessMark {
139 t.Helper()
140 records, err := readJSONL(path)
141 if err != nil {
142 t.Fatal(err)
143 }
144 marks := make([]harnessMark, len(records))
145 for i, record := range records {
146 encoded, err := json.Marshal(record)
147 if err != nil {
148 t.Fatal(err)
149 }
150 if err := json.Unmarshal(encoded, &marks[i]); err != nil {
151 t.Fatal(err)
152 }
153 }
154 return marks
155}
156
157type fakePTY struct{ writes [][]byte }
158
159func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
160
161func (p *fakePTY) Write(b []byte) (int, error) {
162 p.writes = append(p.writes, append([]byte(nil), b...))
163 return len(b), nil
164}
165
166func (*fakePTY) Close() error { return nil }
167
168type stagedPTY struct {
169 allow <-chan struct{}
170 writes chan []byte
171}
172
173func (*stagedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
174
175func (p *stagedPTY) Write(b []byte) (int, error) {
176 if p.allow != nil {
177 <-p.allow
178 }
179 p.writes <- append([]byte(nil), b...)
180 return len(b), nil
181}
182
183func (*stagedPTY) Close() error { return nil }
184
185type fakeOutput struct {
186 bytes.Buffer
187 syncs int
188}
189
190func (o *fakeOutput) Sync() error { o.syncs++; return nil }
191
192func (*fakeOutput) Close() error { return nil }
193
194type orderedPTY struct {
195 write func([]byte)
196 close func()
197}
198
199func (*orderedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
200
201func (p *orderedPTY) Write(b []byte) (int, error) {
202 if p.write != nil {
203 p.write(b)
204 }
205 return len(b), nil
206}
207
208func (p *orderedPTY) Close() error {
209 p.close()
210 return nil
211}
212
213type orderedOutput struct{ close func() }
214
215func (*orderedOutput) Write(b []byte) (int, error) { return len(b), nil }
216
217func (*orderedOutput) Sync() error { return nil }
218
219func (o *orderedOutput) Close() error {
220 o.close()
221 return nil
222}
223
224type failingLauncher struct{ *fakeLauncher }
225
226func (l *failingLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) {
227 p, err := l.fakeLauncher.Launch(m, command)
228 if err != nil {
229 return nil, err
230 }
231 if m.Role == "client" {
232 fake, ok := p.(*fakeProcess)
233 if !ok {
234 return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
235 }
236 fake.fail = true
237 }
238 return p, nil
239}
240
241func equalStrings(a, b []string) bool {
242 if len(a) != len(b) {
243 return false
244 }
245 for i := range a {
246 if a[i] != b[i] {
247 return false
248 }
249 }
250 return true
251}
252
253type traceLauncher struct{ fakeLauncher }
254
255func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) {
256 p, err := l.fakeLauncher.Launch(m, command)
257 if err != nil {
258 return nil, err
259 }
260 f, err := os.OpenFile(m.TracePath, os.O_WRONLY|os.O_APPEND, 0o600)
261 if err != nil {
262 return nil, err
263 }
264 for i, pair := range spanPairs {
265 for n, kind := range []string{pair.start, pair.end} {
266 b, err := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true})
267 if err != nil {
268 return nil, errors.Join(err, f.Close())
269 }
270 if _, err := f.Write(append(b, '\n')); err != nil {
271 return nil, errors.Join(err, f.Close())
272 }
273 }
274 }
275 if err := f.Close(); err != nil {
276 return nil, err
277 }
278 return p, nil
279}
280
281type closeTraceProcess struct {
282 *fakeProcess
283 traceEnd func() error
284}
285
286func (p *closeTraceProcess) Close() error {
287 p.closed = true
288 return errors.Join(p.closeErr, p.traceEnd())
289}
290
291type closeTraceLauncher struct{ fakeLauncher }
292
293func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) {
294 p, err := l.fakeLauncher.Launch(m, command)
295 if err != nil {
296 return nil, err
297 }
298 start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10, Valid: true}
299 if err := appendTraceRecord(m.TracePath, start); err != nil {
300 return nil, err
301 }
302 fake, ok := p.(*fakeProcess)
303 if !ok {
304 return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
305 }
306 return &closeTraceProcess{fakeProcess: fake, traceEnd: func() error {
307 return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true})
308 }}, nil
309}
310
311func appendTraceRecord(path string, record traceRecord) error {
312 f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o600)
313 if err != nil {
314 return err
315 }
316 b, err := json.Marshal(record)
317 if err != nil {
318 return errors.Join(err, f.Close())
319 }
320 if _, err := f.Write(append(b, '\n')); err != nil {
321 return errors.Join(err, f.Close())
322 }
323 return f.Close()
324}
325
326// boundedLocalConcurrentTrace is an excerpt from the second run of the
327// bounded local public-CLI smoke. Receive sequence 3 is in flight while the
328// same process serializes sequence 4; shared-process records therefore cannot
329// have a global sequence ordering requirement.
330const boundedLocalConcurrentTrace = `{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_start","tick":1783902621495284782,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true}
331{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_start","tick":1783902621495364552,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true}
332{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":4,"request_id":4,"epoch":4,"kind":"adapter_send_end","tick":1783902621495386452,"bytes":96,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true}
333{"schema":1,"process_id":"1x4-idle-local-r002-client-02","component":"ipc","scenario":"1x4-idle-local","run":2,"sequence":3,"request_id":3,"epoch":3,"kind":"adapter_receive_end","tick":1783902621495463332,"bytes":0,"fragments":0,"retransmits":0,"pending":0,"ack_rtt_nanos":0,"valid":true}
334`