1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "context"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "io"
12 "os"
13 "os/exec"
14 "path/filepath"
15 "strconv"
16 "strings"
17 "sync"
18 "syscall"
19 "time"
20
21 "github.com/bnema/vev/pkg/rawterm"
22 "github.com/bnema/vev/pkg/safedir"
23)
24
25type launchedProcess interface {
26 Warmup([]byte) error
27 Measure([]byte, func() error, func() error) error
28 Close() error
29}
30
31// roleCommand contains only argv accepted by app.parseArgs. Transport routing
32// metadata configures the harness's public-command seam; it is never passed as
33// an unrecognised vev flag.
34type roleCommand struct {
35 Args []string
36 Transport transport
37}
38
39type launcher interface {
40 Launch(processMapping, roleCommand) (launchedProcess, error)
41}
42
43type launchedRole struct {
44 mapping processMapping
45 process launchedProcess
46}
47
48// closeLaunchedRoles closes every role in reverse launch order. errors.Join
49// retains every cleanup failure in that stable order.
50func closeLaunchedRoles(processes []launchedRole) error {
51 var failures []error
52 for i := len(processes) - 1; i >= 0; i-- {
53 if err := processes[i].process.Close(); err != nil {
54 failures = append(failures, fmt.Errorf("close %s role: %w", processes[i].mapping.Role, err))
55 }
56 }
57 return errors.Join(failures...)
58}
59
60// cliLauncher is deliberately limited to executing the supplied public vev
61// binary. Scenario orchestration never reaches into an in-process daemon.
62type cliLauncher struct {
63 bin string
64 mu sync.Mutex
65 peers map[string]peerRoute
66 runtimes map[string]string
67}
68
69type peerRoute struct {
70 mapping processMapping
71 command roleCommand
72 runtimeDir string
73 stateDir string
74}
75
76type preparedPeer struct {
77 route peerRoute
78 cleanupRuntime func() error
79}
80
81func (p preparedPeer) Warmup([]byte) error { return nil }
82
83func (p preparedPeer) Measure([]byte, func() error, func() error) error { return nil }
84
85func (p preparedPeer) Close() error {
86 if p.cleanupRuntime != nil {
87 return p.cleanupRuntime()
88 }
89 return nil
90}
91
92type terminalOutput interface {
93 io.Writer
94 Close() error
95}
96
97type capturedTerminalOutput struct {
98 file *os.File
99 mu sync.Mutex
100 data bytes.Buffer
101}
102
103type synchronizedBuffer struct {
104 mu sync.Mutex
105 data bytes.Buffer
106}
107
108func (b *synchronizedBuffer) Write(p []byte) (int, error) {
109 b.mu.Lock()
110 defer b.mu.Unlock()
111 return b.data.Write(p)
112}
113
114func (b *synchronizedBuffer) WriteString(value string) {
115 _, _ = b.Write([]byte(value))
116}
117
118func (b *synchronizedBuffer) String() string {
119 b.mu.Lock()
120 defer b.mu.Unlock()
121 return b.data.String()
122}
123
124func (o *capturedTerminalOutput) Write(p []byte) (int, error) {
125 o.mu.Lock()
126 _, _ = o.data.Write(p)
127 o.mu.Unlock()
128 return o.file.Write(p)
129}
130func (o *capturedTerminalOutput) Close() error { return o.file.Close() }
131func (o *capturedTerminalOutput) String() string {
132 o.mu.Lock()
133 defer o.mu.Unlock()
134 return o.data.String()
135}
136
137var setPTYWinsize = rawterm.SetWinsize
138
139type cliProcess struct {
140 cmd *exec.Cmd
141 pty io.ReadWriteCloser
142 output terminalOutput
143 chunks chan []byte
144 done chan struct{}
145 closed sync.Once
146 waitErr chan error
147 waitTimeout func() <-chan time.Time
148 gracefulShutdown func()
149 forceCleanup func()
150 forceKill func()
151 shutdown func() error
152 cleanupRuntime func() error
153 terminalReady bool
154 readyCommand *exec.Cmd
155 startCommand *exec.Cmd
156 configureCommand *exec.Cmd
157 serverOutput *synchronizedBuffer
158 peerCommand *exec.Cmd
159 peerDone chan error
160 closeResult error
161}
162
163func (p *cliProcess) Resize(cols, rows uint16) error {
164 fd, ok := p.pty.(interface{ Fd() uintptr })
165 if !ok {
166 return errors.New("client PTY does not expose a file descriptor")
167 }
168 return setPTYWinsize(int(fd.Fd()), cols, rows)
169}
170
171func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) {
172 if len(role.Args) == 0 {
173 return nil, errors.New("public CLI role has no arguments")
174 }
175 bin, err := filepath.Abs(l.bin)
176 if err != nil {
177 return nil, err
178 }
179 if info, err := os.Stat(bin); err != nil {
180 return nil, err
181 } else if info.IsDir() {
182 return nil, fmt.Errorf("public CLI binary %q is a directory", bin)
183 }
184 if m.Role == "ssh_stdio_peer" {
185 return l.preparePeer(m, role)
186 }
187 cmd := exec.Command(bin, role.Args...)
188 runDir := filepath.Dir(m.TracePath)
189 runtimeDir, err := l.runtimeDir(runDir)
190 if err != nil {
191 return nil, err
192 }
193 // All role inputs are absolute paths resolved at harness startup. Do not set
194 // a role working directory: daemon-owned subprocess cleanup may otherwise
195 // treat the evidence directory as its working tree and remove a preallocated
196 // trace while a later repetition is being merged.
197 cmd.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "VEV_PERF_BIN", "XDG_RUNTIME_DIR", "XDG_STATE_HOME", "XDG_CONFIG_HOME"), traceEnvironment(m)...)
198 cmd.Env = append(cmd.Env, "XDG_RUNTIME_DIR="+runtimeDir,
199 "XDG_STATE_HOME="+filepath.Join(runDir, "state"), "XDG_CONFIG_HOME="+filepath.Join(runDir, "config"), "TERM=xterm-256color", "SHELL=/bin/sh", "VEV_PERF_BIN="+bin)
200 l.mu.Lock()
201 peer, routed := l.peers[runDir]
202 l.mu.Unlock()
203 if routed && (m.Role == "client" || m.Role == "daemon") {
204 if _, err := hostTransport(peer.command.Transport); err != nil {
205 return nil, err
206 }
207 cmd.Env = append(withoutEnv(cmd.Env, "PATH"), "PATH="+runDir+":"+os.Getenv("PATH"))
208 }
209 if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
210 return nil, err
211 }
212 if m.Role == "daemon" {
213 configDir := filepath.Join(runDir, "config", "vev")
214 if err := safedir.EnsurePrivate(configDir); err != nil {
215 return nil, err
216 }
217 if err := os.WriteFile(filepath.Join(configDir, "broker.json"), []byte(`{"marker":"vev.broker.offline/v1","registrations":[]}`), 0o600); err != nil {
218 return nil, err
219 }
220 }
221 p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), serverOutput: &synchronizedBuffer{}}
222 if m.Role == "daemon" {
223 // ls is read-only and cannot start an absent daemon. Start the daemon
224 // explicitly before requiring its observed local catalogue.
225 p.startCommand = exec.Command(bin, "--daemon-launcher")
226 p.startCommand.Env = cmd.Env
227 p.readyCommand = exec.Command(bin, "_broker-ready", "--require", "local-catalogue", "--timeout", "10s")
228 p.readyCommand.Env = cmd.Env
229 l.mu.Lock()
230 peer, routed := l.peers[runDir]
231 l.mu.Unlock()
232 if routed {
233 transportName, transportErr := hostTransport(peer.command.Transport)
234 if transportErr != nil {
235 return nil, transportErr
236 }
237 p.configureCommand = exec.Command(bin, "host", "add", "--transport", transportName, "harness@127.0.0.1")
238 p.configureCommand.Env = append(withoutEnv(cmd.Env, "PATH"), "PATH="+runDir+":"+os.Getenv("PATH"))
239 peerStart := exec.Command(bin, "--daemon")
240 peerStart.Env = append(withoutEnv(os.Environ(), "VEV", "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN", "XDG_RUNTIME_DIR", "XDG_STATE_HOME", "PATH"),
241 "PATH="+runDir+":"+os.Getenv("PATH"),
242 "XDG_RUNTIME_DIR="+peer.runtimeDir,
243 "XDG_STATE_HOME="+peer.stateDir,
244 "TERM=xterm-256color")
245 peerStartOutput := &bytes.Buffer{}
246 peerStart.Stdin, peerStart.Stdout, peerStart.Stderr = nil, peerStartOutput, peerStartOutput
247 if err := peerStart.Start(); err != nil {
248 return nil, fmt.Errorf("start remote peer daemon: %w", err)
249 }
250 peerSocket := filepath.Join(peer.runtimeDir, "vev", "daemonmux.sock")
251 deadline := time.Now().Add(10 * time.Second)
252 for {
253 if info, err := os.Stat(peerSocket); err == nil && info.Mode()&os.ModeSocket != 0 {
254 break
255 }
256 if time.Now().After(deadline) {
257 return nil, fmt.Errorf("remote peer daemon did not publish %s (output=%q)", peerSocket, peerStartOutput.String())
258 }
259 time.Sleep(10 * time.Millisecond)
260 }
261 p.peerCommand = peerStart
262 p.peerDone = make(chan error, 1)
263 go func() {
264 err := peerStart.Wait()
265 if err != nil {
266 p.serverOutput.WriteString(" remote-peer: " + err.Error() + " " + peerStartOutput.String())
267 }
268 p.peerDone <- err
269 }()
270 }
271 }
272 if m.Role == "daemon" {
273 p.cleanupRuntime = func() error { return l.releaseRuntime(runDir) }
274 }
275 if m.Role == "client" {
276 master, slave, err := openPTY()
277 if err != nil {
278 return nil, err
279 }
280 if err := setPTYWinsize(int(master.Fd()), 120, 40); err != nil {
281 _ = master.Close()
282 _ = slave.Close()
283 return nil, fmt.Errorf("set canonical client PTY geometry: %w", err)
284 }
285 output, err := os.OpenFile(filepath.Join(filepath.Dir(m.TracePath), "terminal-output"), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
286 if err != nil {
287 _ = master.Close()
288 _ = slave.Close()
289 return nil, err
290 }
291 cmd.Stdin, cmd.Stdout, cmd.Stderr = slave, slave, slave
292 // Ctty is the child descriptor number after Start remaps slave to stdin.
293 cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true, Setctty: true, Ctty: 0}
294 if err := cmd.Start(); err != nil {
295 _ = output.Close()
296 _ = master.Close()
297 _ = slave.Close()
298 if p.cleanupRuntime != nil {
299 _ = p.cleanupRuntime()
300 }
301 return nil, err
302 }
303 _ = slave.Close()
304 p.pty, p.output, p.chunks, p.done = master, &capturedTerminalOutput{file: output}, make(chan []byte, 32), make(chan struct{})
305 // Stop workload generation through the attached shell before any signal.
306 // A normal shell exit makes the daemon detach the client; the client then
307 // closes and waits for its ssh transport, allowing a traced broker mux stdio
308 // descendant to flush end marks and exit. Process-group signals are reserved
309 // for timeout escalation because they would kill the descendant concurrently.
310 p.gracefulShutdown = func() { _, _ = master.Write([]byte("exit\n")) }
311 go p.copyTerminal()
312 } else {
313 // Every launched role owns a dedicated process group, so the bounded
314 // cleanup path never signals the harness/test runner's group.
315 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
316 devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
317 if err != nil {
318 return nil, err
319 }
320 cmd.Stdin, cmd.Stdout, cmd.Stderr = devNull, p.serverOutput, p.serverOutput
321 if err := cmd.Start(); err != nil {
322 _ = devNull.Close()
323 if p.cleanupRuntime != nil {
324 _ = p.cleanupRuntime()
325 }
326 return nil, err
327 }
328 _ = devNull.Close()
329 }
330 if m.Role == "daemon" {
331 // Ask the daemon owned by the foreground broker to stop before the
332 // broker's bounded process-group fallback. This lets blocked carriage
333 // operations serialize their failed end marks before either process exits.
334 p.shutdown = func() error {
335 ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
336 defer cancel()
337 shutdown := exec.CommandContext(ctx, bin, "_daemon-stop")
338 shutdown.Env = withoutEnv(cmd.Env, "VEV_PERF_TRACE", "VEV_PERF_PROCESS_ID", "VEV_PERF_SCENARIO", "VEV_PERF_RUN")
339 return shutdown.Run()
340 }
341 }
342 go func() { p.waitErr <- cmd.Wait() }()
343 return p, nil
344}
345
346// runtimeDir returns one short, private runtime directory shared by the roles
347// of one run. Keeping the Unix socket outside deeply nested evidence output
348// avoids the platform socket-path limit without weakening run isolation.
349func (l *cliLauncher) runtimeDir(runDir string) (string, error) {
350 l.mu.Lock()
351 defer l.mu.Unlock()
352 if path := l.runtimes[runDir]; path != "" {
353 return path, nil
354 }
355 path, err := os.MkdirTemp("", "vev-harness-runtime-")
356 if err != nil {
357 return "", err
358 }
359 if err := safedir.EnsurePrivate(path); err != nil {
360 _ = os.RemoveAll(path)
361 return "", err
362 }
363 if l.runtimes == nil {
364 l.runtimes = make(map[string]string)
365 }
366 l.runtimes[runDir] = path
367 return path, nil
368}
369
370func (l *cliLauncher) releaseRuntime(runDir string) error {
371 l.mu.Lock()
372 path := l.runtimes[runDir]
373 delete(l.runtimes, runDir)
374 l.mu.Unlock()
375 if path == "" {
376 return nil
377 }
378 return os.RemoveAll(path)
379}
380
381// preparePeer installs a per-run ssh command seam. The ordinary public client
382// still executes its documented remote attach command; its ssh child invokes
383// only the broker mux stdio helper. This makes the peer
384// carrying client traffic the declared, exclusively traced role rather than a
385// separately launched but unused process.
386func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedProcess, error) {
387 runDir := filepath.Dir(m.TracePath)
388 l.mu.Lock()
389 runtimeDir := l.runtimes[runDir]
390 l.mu.Unlock()
391 peerOwnsRuntime := runtimeDir == ""
392 fail := func(err error) error {
393 if peerOwnsRuntime {
394 return errors.Join(err, l.releaseRuntime(runDir))
395 }
396 return err
397 }
398 if peerOwnsRuntime {
399 var err error
400 runtimeDir, err = l.runtimeDir(runDir)
401 if err != nil {
402 return nil, fail(err)
403 }
404 }
405 if err := safedir.EnsurePrivate(runDir); err != nil {
406 return nil, fail(err)
407 }
408 if err := safedir.EnsurePrivate(runtimeDir); err != nil {
409 return nil, fail(err)
410 }
411 if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil {
412 return nil, fail(err)
413 }
414 if len(role.Args) != 2 || role.Args[0] != "_broker-mux-stdio" || role.Args[1] != "--production" {
415 return nil, fail(fmt.Errorf("unsupported broker mux peer command %q", role.Args))
416 }
417 route := peerRoute{mapping: m, command: role, runtimeDir: filepath.Join(runDir, "peer-runtime"), stateDir: filepath.Join(runDir, "peer-state")}
418 if err := safedir.EnsurePrivate(route.runtimeDir); err != nil {
419 return nil, fail(err)
420 }
421 if err := safedir.EnsurePrivate(route.stateDir); err != nil {
422 return nil, fail(err)
423 }
424 shim := filepath.Join(runDir, "ssh")
425 // The SSH command seam forwards the broker mux stdio helper with the
426 // peer's unique trace identity, so the peer carrying client traffic is the
427 // declared, exclusively traced role.
428 body := fmt.Sprintf(`#!/bin/sh
429set -eu
430case "$*" in
431 *"_broker-mux-stdio"*)
432 exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[6]q VEV_PERF_RUN=%[7]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _broker-mux-stdio --production --daemon-start existing-only
433 ;;
434 *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;;
435esac
436`, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, m.Scenario, m.Run)
437 if err := os.WriteFile(shim, []byte(body), 0o700); err != nil {
438 return nil, fail(err)
439 }
440 l.mu.Lock()
441 if l.peers == nil {
442 l.peers = make(map[string]peerRoute)
443 }
444 if _, exists := l.peers[runDir]; exists {
445 l.mu.Unlock()
446 return nil, fail(fmt.Errorf("duplicate transport peer for %s", runDir))
447 }
448 l.peers[runDir] = route
449 l.mu.Unlock()
450 peer := preparedPeer{route: route}
451 if peerOwnsRuntime {
452 peer.cleanupRuntime = func() error { return l.releaseRuntime(runDir) }
453 }
454 return peer, nil
455}
456
457// hostTransport maps a harness transport to the `vev host add --transport` value.
458func hostTransport(t transport) (string, error) {
459 switch t.Kind {
460 case "ssh_stdio":
461 return "ssh", nil
462 default:
463 return "", fmt.Errorf("transport %q cannot route through a remote peer", t.ID)
464 }
465}
466
467// traceEnvironment is the complete process-local identity supplied by the
468// persisted harness processMapping. App turns Scenario/Run into production
469// mark correlation; sequence/request/epoch remain process-local operation IDs.
470func traceEnvironment(m processMapping) []string {
471 return []string{
472 "VEV_PERF_TRACE=" + m.TracePath,
473 "VEV_PERF_PROCESS_ID=" + m.ProcessID,
474 "VEV_PERF_SCENARIO=" + m.Scenario,
475 "VEV_PERF_RUN=" + strconv.Itoa(m.Run),
476 }
477}
478
479func withoutEnv(env []string, names ...string) []string {
480 remove := make(map[string]bool, len(names))
481 for _, name := range names {
482 remove[name] = true
483 }
484 out := make([]string, 0, len(env))
485 for _, entry := range env {
486 name, _, _ := strings.Cut(entry, "=")
487 if !remove[name] {
488 out = append(out, entry)
489 }
490 }
491 return out
492}
493
494func (p *cliProcess) copyTerminal() {
495 defer close(p.chunks)
496 buf := make([]byte, 4096)
497 for {
498 n, err := p.pty.Read(buf)
499 if n > 0 {
500 chunk := append([]byte(nil), buf[:n]...)
501 if _, writeErr := p.output.Write(chunk); writeErr != nil {
502 return
503 }
504 select {
505 case p.chunks <- chunk:
506 case <-p.done:
507 return
508 }
509 }
510 if err != nil {
511 return
512 }
513 }
514}
515
516// WaitReady executes the production dial-only readiness contract under the
517// exact XDG environment shared by the servers and clients. Its bounded JSON
518// stdout, stderr and owner-process death are retained in any failure.
519func (p *cliProcess) WaitReady() error {
520 if p.startCommand != nil {
521 // Starting the daemon is separate from readiness: only the semantic
522 // catalogue probe below can establish that it is serving.
523 if output, err := p.startCommand.CombinedOutput(); err != nil {
524 return fmt.Errorf("start daemon for broker readiness: %w (output=%q)", err, output)
525 }
526 }
527 if p.readyCommand == nil {
528 return errors.New("broker readiness command is empty")
529 }
530 var stdout, stderr bytes.Buffer
531 p.readyCommand.Stdout, p.readyCommand.Stderr = &stdout, &stderr
532 probeDone := make(chan error, 1)
533 if err := p.readyCommand.Start(); err != nil {
534 return fmt.Errorf("start broker readiness probe: %w", err)
535 }
536 go func() { probeDone <- p.readyCommand.Wait() }()
537 select {
538 case err := <-probeDone:
539 if err != nil {
540 return fmt.Errorf("broker readiness probe: %w (stdout=%q stderr=%q)", err, stdout.String(), stderr.String())
541 }
542 var report struct {
543 Status string `json:"status"`
544 }
545 if err := json.Unmarshal(bytes.TrimSpace(stdout.Bytes()), &report); err != nil || report.Status != "ready" {
546 return fmt.Errorf("broker readiness invalid JSON/status: %v (stdout=%q stderr=%q)", err, stdout.String(), stderr.String())
547 }
548 if p.configureCommand != nil {
549 if output, err := p.configureCommand.CombinedOutput(); err != nil {
550 return fmt.Errorf("configure remote broker membership: %w (output=%q)", err, output)
551 }
552 }
553 return nil
554 case err := <-p.waitErr:
555 p.waitErr <- err
556 _ = p.readyCommand.Process.Kill()
557 probeErr := <-probeDone
558 return fmt.Errorf("production server group died before readiness: %v (probe=%v stdout=%q stderr=%q servers=%q)", err, probeErr, stdout.String(), stderr.String(), p.serverOutput.String())
559 }
560}
561
562func (p *cliProcess) Warmup(input []byte) error {
563 if p.pty == nil || len(input) == 0 {
564 return nil
565 }
566 // A client cannot write application output until it has received Welcome and
567 // entered raw mode. Require that observable output before submitting any
568 // warmup input, so the PTY's line discipline cannot echo it locally.
569 if !p.terminalReady {
570 if err := p.waitForTerminalReady(); err != nil {
571 return err
572 }
573 p.terminalReady = true
574 }
575 if _, err := p.pty.Write(input); err != nil {
576 return err
577 }
578 // Do not begin measured input until the public client has rendered a real
579 // shell response to warmup input. This is intentionally not a measurement.
580 return p.waitForTerminalMarker(inputMarker(input), input)
581}
582
583// Measure records a boundary after copyTerminal has read the marker from the
584// real client PTY and the harness-owned output writer has accepted that chunk.
585// This models successful terminal-byte delivery without adding filesystem
586// durability latency. The marker is never fabricated by the harness, and PTY
587// local echo is discarded so it cannot satisfy the boundary.
588func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error {
589 if p.pty == nil {
590 return errors.New("measured process has no terminal PTY")
591 }
592 if _, err := p.pty.Write(input); err != nil {
593 return err
594 }
595 // The injected callback runs only after the owned PTY accepts the bytes.
596 if err := injected(); err != nil {
597 return err
598 }
599 if err := p.waitForTerminalMarker(inputMarker(input), input); err != nil {
600 return err
601 }
602 return flushed()
603}
604
605// waitForTerminalReady waits for public client output produced before any
606// harness input. The client enters raw mode before its output loop can render
607// this initial prompt/state, so this is an observable readiness boundary.
608func (p *cliProcess) waitForTerminalReady() error {
609 timer := time.NewTimer(10 * time.Second)
610 defer timer.Stop()
611 var observed []byte
612 for {
613 select {
614 case chunk, ok := <-p.chunks:
615 if !ok {
616 return p.terminalClosedError("readiness output")
617 }
618 observed = append(observed, chunk...)
619 if bytes.Contains(observed, []byte("$")) || bytes.Contains(observed, []byte("#")) {
620 return nil
621 }
622 if len(observed) > 64*1024 {
623 observed = append([]byte(nil), observed[len(observed)-32*1024:]...)
624 }
625 case err := <-p.waitErr:
626 // Keep the exit result available to Close, which owns reaping.
627 p.waitErr <- err
628 if err == nil {
629 return errors.New("client exited before readiness output")
630 }
631 return fmt.Errorf("client exited before readiness output: %w", err)
632 case <-timer.C:
633 if captured, ok := p.output.(*capturedTerminalOutput); ok {
634 return fmt.Errorf("timed out waiting for client readiness output (terminal=%q)", tailString(captured.String(), 4096))
635 }
636 return errors.New("timed out waiting for client readiness output")
637 }
638 }
639}
640
641func tailString(value string, limit int) string {
642 if len(value) <= limit {
643 return value
644 }
645 return value[len(value)-limit:]
646}
647
648func (p *cliProcess) terminalClosedError(boundary string) error {
649 if p.waitErr != nil {
650 select {
651 case err := <-p.waitErr:
652 // Close owns reaping and may need the result after a failed boundary.
653 p.waitErr <- err
654 if err != nil {
655 return fmt.Errorf("client terminal closed before %s: %w", boundary, err)
656 }
657 case <-time.After(10 * time.Millisecond):
658 }
659 }
660 return fmt.Errorf("client terminal closed before %s", boundary)
661}
662
663func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error {
664 var received []byte
665 echo := newPTYLocalEcho(input)
666 timer := time.NewTimer(10 * time.Second)
667 defer timer.Stop()
668 for !bytes.Contains(received, marker) {
669 select {
670 case chunk, ok := <-p.chunks:
671 if !ok {
672 return p.terminalClosedError("application output")
673 }
674 received = append(received, echo.applicationOutput(chunk)...)
675 case err := <-p.waitErr:
676 // Keep the exit result available to Close, which owns reaping.
677 p.waitErr <- err
678 if err == nil {
679 return errors.New("client exited before application output")
680 }
681 return fmt.Errorf("client exited before application output: %w", err)
682 case <-timer.C:
683 if captured, ok := p.output.(*capturedTerminalOutput); ok {
684 return fmt.Errorf("timed out waiting for application terminal output (terminal=%q)", captured.String())
685 }
686 return errors.New("timed out waiting for application terminal output")
687 }
688 }
689 return nil
690}
691
692// ptyLocalEcho removes the one ordered input echo that a not-yet-raw slave PTY
693// can produce. It normalizes NL to the conventional ONLCR echo spelling; any
694// divergence is retained as application output. The line discipline emits the
695// echo before the child can read and respond to the same write.
696type ptyLocalEcho struct {
697 expected []byte
698 matched int
699 done bool
700}
701
702func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte {
703 if e.done || len(e.expected) == 0 {
704 return chunk
705 }
706 for i, b := range chunk {
707 if b != e.expected[e.matched] {
708 e.done = true
709 return append([]byte(nil), chunk[i:]...)
710 }
711 e.matched++
712 if e.matched == len(e.expected) {
713 e.done = true
714 return append([]byte(nil), chunk[i+1:]...)
715 }
716 }
717 return nil
718}
719
720// Close tears down a launched role along a bounded escalation ladder. Each rung
721// performs one action, then waits a single closeDeadline for the process to be
722// reaped before the next rung escalates. The graceful rungs — the public daemon
723// stop and a client shell exit — let a role close its blocked transport
724// operations and drain descendants' end marks. Process-group SIGTERM is reserved
725// for timeout escalation; only a SIGKILL rung fails teardown. Every path closes
726// the PTY exactly once and finalizes output and runtime state.
727func (p *cliProcess) Close() error {
728 p.closed.Do(func() {
729 // 1. Drain terminal output for the whole teardown. Without an active
730 // consumer a full chunks channel stalls copyTerminal, which then stops
731 // reading the PTY master and blocks the client mid-restore before it can
732 // close its observed receive operation. The drain stops with p.done.
733 drainDone := p.drainTerminalOutput()
734
735 exited := false
736 // waitForExit blocks up to one bounded deadline for the reap.
737 waitForExit := func() {
738 if p.waitErr == nil || exited {
739 return
740 }
741 select {
742 case <-p.waitErr:
743 exited = true
744 case <-p.closeDeadline():
745 }
746 }
747 // reaped reports whether the process is already gone; a role with no
748 // process to reap needs no escalation at all.
749 reaped := func() bool {
750 if p.waitErr == nil || exited {
751 return true
752 }
753 select {
754 case <-p.waitErr:
755 exited = true
756 return true
757 default:
758 return false
759 }
760 }
761 ptyClosed := false
762 closePTY := func() {
763 if p.pty != nil && !ptyClosed {
764 _ = p.pty.Close()
765 ptyClosed = true
766 }
767 }
768
769 // 2. Daemon role: ask the public daemon to stop before any signal so its
770 // blocked carriage can close and serialize failed end marks.
771 if p.shutdown != nil && !reaped() {
772 _ = p.shutdown()
773 waitForExit()
774 }
775 // 3. Client role: stop the shell normally. Session teardown detaches the
776 // client, which closes and waits for ssh/broker mux stdio before it exits.
777 if p.gracefulShutdown != nil && !reaped() {
778 p.gracefulShutdown()
779 waitForExit()
780 }
781 // 4. Escalate a role that ignored graceful shutdown to process-group
782 // SIGTERM. This remains bounded and cleans up hung descendants.
783 if !reaped() {
784 p.forceProcessGroupCleanup()
785 waitForExit()
786 }
787 // 5. SIGKILL the role's process group. This severs the process; an
788 // in-flight observed operation cannot be closed, so its trace may truncate.
789 killed := false
790 if !reaped() {
791 killed = true
792 p.forceProcessGroupKill()
793 waitForExit()
794 }
795 // 6. A SIGKILL escalation cannot guarantee a complete trace. Fail teardown
796 // with a clear cause rather than letting the truncation surface later as a
797 // downstream merge mystery. The graceful rungs never error on their own.
798 if killed {
799 if exited {
800 p.closeResult = errors.New("public CLI process did not exit after graceful shutdown and SIGTERM; killed")
801 } else {
802 p.closeResult = errors.New("timed out reaping public CLI process")
803 }
804 }
805
806 // Close the PTY exactly once on every path: an early graceful exit skips
807 // the hangup rung, so close it here before the drain stops.
808 closePTY()
809 if p.done != nil {
810 close(p.done)
811 }
812 if drainDone != nil {
813 <-drainDone
814 }
815 if p.peerCommand != nil && p.peerCommand.Process != nil {
816 _ = p.peerCommand.Process.Signal(syscall.SIGTERM)
817 if p.peerDone != nil {
818 select {
819 case <-p.peerDone:
820 case <-time.After(2 * time.Second):
821 _ = p.peerCommand.Process.Kill()
822 <-p.peerDone
823 }
824 }
825 }
826 if p.output != nil {
827 if err := p.output.Close(); p.closeResult == nil {
828 p.closeResult = err
829 }
830 }
831 if p.cleanupRuntime != nil {
832 if err := p.cleanupRuntime(); p.closeResult == nil {
833 p.closeResult = err
834 }
835 }
836 })
837 return p.closeResult
838}
839
840// drainTerminalOutput consumes copyTerminal's chunks for the duration of a
841// teardown so a full channel can never wedge copyTerminal (and thus the client's
842// PTY). It returns nil when there is no terminal to drain, and otherwise a
843// channel closed once the consumer has stopped on p.done.
844func (p *cliProcess) drainTerminalOutput() <-chan struct{} {
845 if p.chunks == nil || p.done == nil {
846 return nil
847 }
848 done := make(chan struct{})
849 chunks := p.chunks
850 go func() {
851 defer close(done)
852 for {
853 select {
854 case _, ok := <-chunks:
855 if !ok {
856 // copyTerminal closed the channel; a closed channel is always
857 // ready, so stop selecting it and wait only for the stop signal.
858 chunks = nil
859 }
860 case <-p.done:
861 return
862 }
863 }
864 }()
865 return done
866}
867
868func (p *cliProcess) closeDeadline() <-chan time.Time {
869 if p.waitTimeout != nil {
870 return p.waitTimeout()
871 }
872 if p.shutdown != nil {
873 // A public kill asks the daemon to finish bounded detached notices and
874 // close blocked carriage before it exits; do not turn that graceful
875 // protocol into a SIGKILL at the two-second client deadline.
876 return time.After(5 * time.Second)
877 }
878 return time.After(2 * time.Second)
879}
880
881func (p *cliProcess) forceProcessGroupCleanup() {
882 if p.forceCleanup != nil {
883 p.forceCleanup()
884 return
885 }
886 if p.cmd != nil && p.cmd.Process != nil {
887 // The client is a session/process-group leader; forced cleanup must
888 // include its ssh seam/broker mux stdio descendants.
889 _ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGTERM)
890 }
891}
892
893func (p *cliProcess) forceProcessGroupKill() {
894 if p.forceKill != nil {
895 p.forceKill()
896 return
897 }
898 if p.cmd != nil && p.cmd.Process != nil {
899 _ = syscall.Kill(-p.cmd.Process.Pid, syscall.SIGKILL)
900 }
901}
902
903// openPTY is the small Linux PTY boundary needed to drive the public terminal
904// client. No daemon or adapter implementation is imported by this command.
905func openPTY() (*os.File, *os.File, error) {
906 // O_CLOEXEC keeps these harness-owned descriptors out of the launched client:
907 // exec dup2s the slave onto the child's stdio, but an inherited master (or a
908 // second slave) would otherwise leak in and hold the terminal open, so closing
909 // the harness master could never hang up the client. cmd.Start still remaps the
910 // slave onto fds 0-2, which are exempt from close-on-exec.
911 masterFD, err := syscall.Open("/dev/ptmx", syscall.O_RDWR|syscall.O_NOCTTY|syscall.O_CLOEXEC, 0)
912 if err != nil {
913 return nil, nil, err
914 }
915 closeMaster := func(err error) (*os.File, *os.File, error) { _ = syscall.Close(masterFD); return nil, nil, err }
916 number, err := rawterm.PtsNumber(masterFD)
917 if err != nil {
918 return closeMaster(err)
919 }
920 if err := rawterm.UnlockPt(masterFD); err != nil {
921 return closeMaster(err)
922 }
923 slaveFD, err := syscall.Open(fmt.Sprintf("/dev/pts/%d", number), syscall.O_RDWR|syscall.O_NOCTTY|syscall.O_CLOEXEC, 0)
924 if err != nil {
925 return closeMaster(err)
926 }
927 return os.NewFile(uintptr(masterFD), "vev-client-master"), os.NewFile(uintptr(slaveFD), "vev-client-slave"), nil
928}
929
930func inputMarker(input []byte) []byte {
931 start := bytes.Index(input, []byte("__VEV_HARNESS_"))
932 if start < 0 {
933 return input
934 }
935 end := bytes.Index(input[start:], []byte(`\n`))
936 if end < 0 {
937 return input[start:]
938 }
939 return input[start : start+end]
940}