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