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