diff --git a/pkg/module/serial/README.md b/pkg/module/serial/README.md index 86257e5a..03ae746c 100644 --- a/pkg/module/serial/README.md +++ b/pkg/module/serial/README.md @@ -15,9 +15,9 @@ mode: see [Interactive mode](#interactive-mode). ``` ARGUMENTS: - [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] (monitor: stream output) - [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] [--] ... (run a step sequence) - [-t ] -i (interactive console) + [-reconnect] [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] (monitor) + [-reconnect] [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] [--] ... (step sequence) + [-reconnect] [-t ] -i (interactive) step := expect | send | send-raw ``` @@ -51,8 +51,8 @@ is done. While the monitor runs, the device is busy for other users, but your own other commands may run beside it: a power cycle from a second terminal shows up in the monitor's output. Start it once the monitor prints `--- Connected to … ---`, or -the first lines of the boot are lost. If the DUT powers the serial adapter, the -adapter disappears with the power and the monitor ends. +the first lines of the boot are lost. If the serial adapter disappears during a +power cycle, the monitor ends unless `-reconnect` is given. A serial port serves one command at a time, unless dutagent runs as root. Running the same command again is refused, and another command that opens the @@ -94,6 +94,42 @@ Like the monitor, an interactive session keeps the device busy for other users until it ends, and your own other commands, a power cycle for instance, may run beside it. +## Temporary disconnections + +Pass `-reconnect` in any mode to wait for an initially absent device and reopen +it after a disconnect. Configure a stable path such as +`/dev/serial/by-id/usb-...`, so a replacement `/dev/ttyUSBn` number cannot select +another adapter. Reopening uses the same path and baud rate, at 250 ms intervals. +Permission, busy and configuration errors remain fatal. Without the flag, opening +or reading an unavailable device still fails immediately. + +The session and device reservation stay open. Notices appear once per transition: + +```text +--- Serial device disconnected: /dev/serial/by-id/usb-... --- +--- Waiting for serial device to reconnect: /dev/serial/by-id/usb-... --- +--- Serial device reconnected: /dev/serial/by-id/usb-... at 115200 baud --- +``` + +An initially absent device prints the waiting notice. Interactive notices use +stderr with CR LF; stdout continues to contain only DUT bytes. Input received +while disconnected is discarded, with a notice, rather than replayed after +reconnection. An interrupted interactive write is not replayed either. + +Monitor output resumes. A pending `expect` resumes with its match buffer and +partial character-map/escape state cleared, so a match cannot span two boots. +Completed steps are never repeated. A send that has not started may wait for the +device; any failed attempted send terminates the script with uncertain delivery, +including short writes that failed partway through. It is never retried. + +The original `-t` deadline includes all waiting and the final drain, and never +restarts on reconnection. Reaching it still succeeds in monitor/interactive mode; +an unfinished scripted step fails. With `-t 0`, waiting ends on cancellation (or +interactive input EOF). Ctrl-A x remains available while waiting in a console. +Newly opened ports are not flushed, preserving queued boot bytes; output emitted +while the adapter was absent cannot be recovered. Without `-reconnect`, the +initial input buffer is still flushed as before. + ## Flags | Flag | Description | @@ -102,6 +138,7 @@ beside it. | `-eol cr\|lf\|crlf\|none` | Line ending appended by `send`. Default `cr` (`\r`), which is what serial consoles expect on Enter. | | `-keep-escapes` | Keep terminal escape sequences in the output instead of stripping them (e.g. for binary data or exact-byte capture). | | `-i` | Interactive console: bridge the port to your terminal. Takes no steps, and no `-eol` or `-keep-escapes`. | +| `-reconnect` | Wait for an absent device and reopen after disconnection. Default off; see [Temporary disconnections](#temporary-disconnections). | The `--` separator before the steps is optional; it is only needed if a step value would otherwise look like a flag. @@ -139,6 +176,9 @@ serial # Interactive console (end it with Ctrl-A x). serial -i +# Keep the same console through USB serial disconnections during power cycles. +serial -reconnect -i + # Wait for a boot marker, then succeed. serial -- expect 'Welcome to' diff --git a/pkg/module/serial/bridge.go b/pkg/module/serial/bridge.go index 7a03c9c7..d3341850 100644 --- a/pkg/module/serial/bridge.go +++ b/pkg/module/serial/bridge.go @@ -41,6 +41,10 @@ func bridge( ) error { _, _ = fmt.Fprintf(con.Stderr, "--- Connected to %s at %d baud ---\r\n", name, baud) + return bridgeConsole(ctx, serialPort, incoming, con, drain) +} + +func bridgeConsole(ctx context.Context, serialPort port, incoming *inbound, con module.Console, drain time.Duration) error { // The input pump runs on its own goroutine, since a Read on Stdin and a Read // on the port both block. stopped is closed when it returned, and writeErr // is read only after that. @@ -171,7 +175,7 @@ func forwardOutput( } } - if err != nil { + if err != nil && !errors.Is(err, errDisconnected) { return err } } diff --git a/pkg/module/serial/engine.go b/pkg/module/serial/engine.go index a543cab6..de7122c0 100644 --- a/pkg/module/serial/engine.go +++ b/pkg/module/serial/engine.go @@ -6,6 +6,7 @@ package serial import ( "context" + "errors" "io" "regexp" "time" @@ -78,8 +79,8 @@ func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error { } // go.bug.st returns (0, nil) on a read timeout — keep waiting and let - // the ctx check above enforce the overall deadline. Any real error - // (port closed / device gone) ends the step. + // the ctx check above enforce the overall deadline. The reconnecting + // adapter marks device loss as a boundary; other errors end the step. out, err := e.in.read(e.p, readBuf) werr := e.tee(out) @@ -87,6 +88,12 @@ func (e *engine) readUntil(ctx context.Context, pattern *regexp.Regexp) error { return werr } + if errors.Is(err, errDisconnected) { + e.buf = e.buf[:0] + + continue + } + if err != nil { return err } @@ -163,6 +170,10 @@ func (e *engine) pump(ctx context.Context, swallowReadErr bool) error { } } + if errors.Is(err, errDisconnected) { + continue + } + if err != nil { if swallowReadErr { return nil @@ -180,12 +191,11 @@ func (e *engine) monitor(ctx context.Context) error { return e.pump(ctx, false) } -// drain forwards serial output for up to d, so the DUT's reply to a final send -// is visible before the run closes. It is best-effort: a read error (e.g. the -// device rebooted from that send) ends it without failing the run. -func (e *engine) drain(ctx context.Context, d time.Duration) error { +// drain forwards serial output for up to d. Legacy runs tolerate read errors +// after a final send; reconnecting runs handle device loss and expose fatal errors. +func (e *engine) drain(ctx context.Context, d time.Duration, swallowReadErr bool) error { drainCtx, cancel := context.WithTimeout(ctx, d) defer cancel() - return e.pump(drainCtx, true) + return e.pump(drainCtx, swallowReadErr) } diff --git a/pkg/module/serial/engine_test.go b/pkg/module/serial/engine_test.go index 0e5e95af..13644b60 100644 --- a/pkg/module/serial/engine_test.go +++ b/pkg/module/serial/engine_test.go @@ -185,7 +185,7 @@ func TestEngineDrainForwardsBestEffort(t *testing.T) { sink := &bytes.Buffer{} eng := newEngine(fp, newInbound(charMap{}, false), sink) - if err := eng.drain(context.Background(), 50*time.Millisecond); err != nil { + if err := eng.drain(context.Background(), 50*time.Millisecond, true); err != nil { t.Fatalf("drain: %v", err) } diff --git a/pkg/module/serial/inbound.go b/pkg/module/serial/inbound.go index 602238bd..ca1a7230 100644 --- a/pkg/module/serial/inbound.go +++ b/pkg/module/serial/inbound.go @@ -5,6 +5,7 @@ package serial import ( + "errors" "fmt" "io" "strings" @@ -110,7 +111,12 @@ func (in *inbound) read(serialPort port, buf []byte) ([]byte, error) { return in.flush(), nil } - return in.apply(buf[:n]), err + out := in.apply(buf[:n]) + if errors.Is(err, errDisconnected) { + *in = inbound{maps: in.maps, strip: in.strip} + } + + return out, err } // apply returns what of chunk is to be shown. diff --git a/pkg/module/serial/reconnect.go b/pkg/module/serial/reconnect.go new file mode 100644 index 00000000..1ab5d075 --- /dev/null +++ b/pkg/module/serial/reconnect.go @@ -0,0 +1,288 @@ +// Copyright 2025 Blindspot Software +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package serial + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "syscall" + "time" + + "github.com/BlindspotSoftware/dutctl/internal/log" + "github.com/BlindspotSoftware/dutctl/pkg/module" + "go.bug.st/serial" +) + +const reconnectInterval = 250 * time.Millisecond + +// errDisconnected marks a boundary across which filters and expect buffers cannot carry state. +var errDisconnected = errors.New("serial device disconnected") + +type portConfigurationError struct{ error } + +func (e *portConfigurationError) Unwrap() error { return e.error } + +// reconnectingPort has one reader, which owns opening, closing and notifications. +// The input writer shares mu so a disconnected handle is never closed during Write. +type reconnectingPort struct { + ctx context.Context //nolint:containedctx // io.Reader/Writer adapter scoped to one Run, including all reopen waits + mu sync.Mutex + current port + broken error + open portOpener + name string + baud int + interactive bool + connected bool + waiting bool + retryAt time.Time + notify func(string, ...any) error +} + +func (s *Serial) runReconnecting(ctx context.Context, session module.Session, cfg scriptConfig) error { + if cfg.timeout > 0 { + var cancel context.CancelFunc + + ctx, cancel = context.WithTimeout(ctx, cfg.timeout) + defer cancel() + } + + serialPort := &reconnectingPort{ctx: ctx, open: s.open, name: s.Port, baud: s.Baud, interactive: cfg.interactive} + if serialPort.open == nil { + serialPort.open = defaultOpenPort + } + defer serialPort.Close() + + if cfg.interactive { + con := session.OpenConsole(module.ConsoleOptions{Mode: module.ConsoleRaw}) + serialPort.notify = func(format string, args ...any) error { + _, err := fmt.Fprintf(con.Stderr, "\r\n"+format+"\r\n", args...) + + return err + } + + return bridgeConsole(ctx, serialPort, newInbound(s.imap, false), con, s.drainFor()) + } + + clientOut := newClientWriter(session) + serialPort.notify = func(format string, args ...any) error { + clientOut.markerf(format+"\n", args...) + + return nil + } + + // Reconnect waiting and the final drain both obey the original global deadline. + return s.runSequence(ctx, ctx, cfg, serialPort, clientOut) +} + +// poll tries once per interval; returning quiet reads keeps console EOF/drain responsive too. +func (p *reconnectingPort) poll(ctx context.Context) error { + err := ctx.Err() + if err != nil { + return err + } + + err = sleepCtx(ctx, min(time.Until(p.retryAt), readTimeout)) + if err != nil { + return err + } + + if time.Now().Before(p.retryAt) { + return nil + } + + opened, err := p.open(p.name, p.baud) + if err != nil { + if !deviceUnavailable(err) { + return err + } + + p.retryAt = time.Now().Add(reconnectInterval) + + return p.waitNotice() + } + + p.mu.Lock() + p.current = opened + p.mu.Unlock() + p.waiting = false + + label := "Connected to" + if p.connected { + label = "Serial device reconnected:" + } + + p.connected = true + log.FromContext(ctx).Info(fmt.Sprintf("connected to %s at %d baud", p.name, p.baud)) + + // Do not flush: the new handle may already contain the next boot's first bytes. + return p.notify("--- %s %s at %d baud ---", label, p.name, p.baud) +} + +func (p *reconnectingPort) waitNotice() error { + if p.waiting { + return nil + } + + p.waiting = true + + text := "--- Waiting for serial device to reconnect: %s ---" + if p.interactive { + text += "\r\n--- Input while disconnected is discarded ---" + } + + return p.notify(text, p.name) +} + +func (p *reconnectingPort) Read(buf []byte) (int, error) { + p.mu.Lock() + current, broken := p.current, p.broken + p.mu.Unlock() + + if current == nil { + err := p.poll(p.ctx) + if p.ctx.Err() != nil { + return 0, nil //nolint:nilerr // caller's read loop handles cancellation as success or expect failure + } + + return 0, err + } + + var count int + + if broken == nil { + count, broken = current.Read(buf) + } + + p.mu.Lock() + if p.broken != nil { + broken = p.broken + } + p.mu.Unlock() + + if !deviceUnavailable(broken) { + return count, broken + } + + return count, p.disconnect() +} + +func (p *reconnectingPort) disconnect() error { + p.mu.Lock() + _ = p.current.Close() + p.current = nil + p.broken = nil + p.mu.Unlock() + p.retryAt = time.Now().Add(reconnectInterval) + + err := p.notify("--- Serial device disconnected: %s ---", p.name) + if err != nil { + return err + } + + err = p.waitNotice() + if err != nil { + return err + } + + return errDisconnected +} + +// Write drops offline interactive input, including the rest of an interrupted chunk. +// It never retries bytes whose delivery to the old device is uncertain. +func (p *reconnectingPort) Write(data []byte) (int, error) { + if !p.interactive { + err := p.awaitConnection() + if err != nil { + return 0, err + } + } + + p.mu.Lock() + defer p.mu.Unlock() + + if p.current == nil || p.broken != nil { + return len(data), nil + } + + count, err := p.writeChunk(data) + if p.interactive && deviceUnavailable(err) { + p.broken = err + + return len(data), nil + } + + if err != nil && !p.interactive { + return count, fmt.Errorf("serial write interrupted; delivery uncertain, command was not retried: %w", err) + } + + return count, err +} + +func (p *reconnectingPort) awaitConnection() error { + for p.current == nil { + err := p.poll(p.ctx) + if err != nil { + return err + } + } + + return p.ctx.Err() +} + +// A complete input chunk holds mu, including short writes, so no suffix can +// accidentally land on a replacement connection. +func (p *reconnectingPort) writeChunk(data []byte) (int, error) { + total := 0 + + for total < len(data) { + count, err := p.current.Write(data[total:]) + total += count + + if err != nil { + return total, err + } + + if count == 0 { + return total, io.ErrShortWrite + } + } + + return total, nil +} + +func (p *reconnectingPort) ResetInputBuffer() error { return nil } + +func (p *reconnectingPort) Close() error { + p.mu.Lock() + defer p.mu.Unlock() + + if p.current == nil { + return nil + } + + err := p.current.Close() + p.current = nil + + return err +} + +func deviceUnavailable(err error) bool { + var configErr *portConfigurationError + if errors.As(err, &configErr) { + return false + } + + var portErr *serial.PortError + if errors.As(err, &portErr) { + return portErr.Code() == serial.PortNotFound || portErr.Code() == serial.PortClosed + } + + return errors.Is(err, syscall.ENOENT) || errors.Is(err, syscall.ENODEV) || + errors.Is(err, syscall.ENXIO) || errors.Is(err, syscall.EIO) +} diff --git a/pkg/module/serial/reconnect_linux_test.go b/pkg/module/serial/reconnect_linux_test.go new file mode 100644 index 00000000..9d7068cd --- /dev/null +++ b/pkg/module/serial/reconnect_linux_test.go @@ -0,0 +1,86 @@ +// Copyright 2025 Blindspot Software +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package serial + +import ( + "context" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + + "golang.org/x/sys/unix" +) + +func openReconnectPTY(t *testing.T) (*os.File, string) { + t.Helper() + fd, err := unix.Open("/dev/ptmx", unix.O_RDWR|unix.O_NOCTTY|unix.O_CLOEXEC, 0) + if err != nil { + t.Fatal(err) + } + master := os.NewFile(uintptr(fd), "pty-master") + t.Cleanup(func() { _ = master.Close() }) + if err := unix.IoctlSetPointerInt(fd, unix.TIOCSPTLCK, 0); err != nil { + t.Fatal(err) + } + number, err := unix.IoctlGetInt(fd, unix.TIOCGPTN) + if err != nil { + t.Fatal(err) + } + return master, fmt.Sprintf("/dev/pts/%d", number) +} + +func TestReconnectRealPTYReplacement(t *testing.T) { + first, firstPath := openReconnectPTY(t) + link := filepath.Join(t.TempDir(), "stable-serial-id") + if err := os.Symlink(firstPath, link); err != nil { + t.Fatal(err) + } + run := newInteractiveRun() + run.serial.Port = link + run.serial.open = nil // Exercise serial.Open, real hangup errors and the configured read timeout. + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(func() { cancel(); _ = run.input.Close() }) + done := run.start(ctx, "-reconnect", "-t", "5s") + waitReconnect(t, func() bool { return strings.Contains(run.stderr.String(), "Connected to") }) + if _, err := first.Write([]byte("before\r\n")); err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return strings.Contains(run.stdout.String(), "before") }) + if err := first.Close(); err != nil { + t.Fatal(err) + } + if err := os.Remove(link); err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return strings.Contains(run.stderr.String(), "Serial device disconnected:") }) + second, secondPath := openReconnectPTY(t) + if err := os.Symlink(secondPath, link); err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return strings.Contains(run.stderr.String(), "reconnected:") }) + settings, err := unix.IoctlGetTermios(int(second.Fd()), unix.TCGETS) + if err != nil { + t.Fatal(err) + } + if settings.Cflag&unix.CBAUD != unix.B115200 { + t.Errorf("replacement baud=%#x, want B115200", settings.Cflag&unix.CBAUD) + } + if _, err := second.Write([]byte("after\r\n")); err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return strings.Contains(run.stdout.String(), "after") }) + cancel() + if err := awaitReconnectRun(t, done); err != nil { + t.Fatal(err) + } + if got := run.stdout.String(); got != "before\r\nafter\r\n" { + t.Errorf("raw output across real hangup=%q", got) + } + if run.stdin.inRead.Load() != 0 { + t.Error("input goroutine left running after real port teardown") + } +} diff --git a/pkg/module/serial/reconnect_test.go b/pkg/module/serial/reconnect_test.go new file mode 100644 index 00000000..52b43ea4 --- /dev/null +++ b/pkg/module/serial/reconnect_test.go @@ -0,0 +1,518 @@ +// Copyright 2025 Blindspot Software +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +package serial + +import ( + "context" + "errors" + "fmt" + "io" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/BlindspotSoftware/dutctl/internal/test/mock" +) + +type openResult struct { + port port + err error +} + +func serialWithOpens(t *testing.T, results ...openResult) (*Serial, *atomic.Int32) { + t.Helper() + calls := &atomic.Int32{} + s := &Serial{ + Port: "/dev/serial/by-id/test-adapter", Baud: 57600, + drainTimeout: 10 * time.Millisecond, + open: func(name string, baud int) (port, error) { + if name != "/dev/serial/by-id/test-adapter" || baud != 57600 { + t.Errorf("open(%q, %d): settings changed across reconnect", name, baud) + } + idx := int(calls.Add(1)) - 1 + if idx >= len(results) { + return nil, syscall.ENOENT + } + return results[idx].port, results[idx].err + }, + } + return s, calls +} + +func waitReconnect(t *testing.T, ready func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for !ready() { + if time.Now().After(deadline) { + t.Fatal("timed out waiting for serial state") + } + time.Sleep(time.Millisecond) + } +} + +func awaitReconnectRun(t *testing.T, done <-chan error) error { + t.Helper() + select { + case err := <-done: + return err + case <-time.After(3 * time.Second): + t.Fatal("serial Run did not terminate") + return nil + } +} + +func TestParseReconnect(t *testing.T) { + for _, args := range [][]string{ + {"-reconnect"}, {"-reconnect", "-i"}, {"-reconnect", "expect", "ready"}, + } { + cfg, err := parseArgs(args) + if err != nil || !cfg.reconnect { + t.Errorf("parseArgs(%q) = %+v, %v", args, cfg, err) + } + } + cfg, err := parseArgs([]string{"-reconnect=false", "-i"}) + if err != nil || cfg.reconnect { + t.Errorf("explicit false = %+v, %v", cfg, err) + } +} + +func TestReconnectInitialAbsenceAndRepeatedFlaps(t *testing.T) { + first := &fakePort{reads: [][]byte{[]byte("first boot\n")}, readErr: syscall.EIO} + second := &fakePort{reads: [][]byte{[]byte("second boot\n")}, readErr: syscall.ENODEV} + third := &fakePort{reads: [][]byte{[]byte("third boot READY")}} + s, calls := serialWithOpens(t, + openResult{err: syscall.ENOENT}, openResult{port: first}, openResult{err: syscall.ENXIO}, + openResult{port: second}, openResult{port: third}) + sess := &recordingSession{} + err := s.Run(context.Background(), sess, "-reconnect", "-t", "3s", "expect", "READY") + if err != nil { + t.Fatal(err) + } + if calls.Load() != 5 { + t.Errorf("open calls = %d, want 5", calls.Load()) + } + out := sess.out.String() + for _, text := range []string{"first boot", "second boot", "third boot READY", "Script completed"} { + if !strings.Contains(out, text) { + t.Errorf("missing %q in %q", text, out) + } + } + for text, want := range map[string]int{"Waiting for": 3, "Serial device disconnected:": 2, "reconnected:": 2} { + if got := strings.Count(out, text); got != want { + t.Errorf("%q count = %d, want %d: %q", text, got, want, out) + } + } + if !strings.HasPrefix(out, "--- Waiting for") { + t.Errorf("initially absent device claimed to be connected: %q", out) + } + for _, p := range []*fakePort{first, second, third} { + if !p.closed || p.resetCount != 0 { + t.Errorf("port closed=%v, flushes=%d; want closed without flushing boot bytes", p.closed, p.resetCount) + } + } +} + +func TestReconnectDoesNotCarryExpectOrFilterState(t *testing.T) { + tests := []struct { + name, before, after, pattern string + maps charMap + wantError bool + wantOutput string + }{ + {"matcher", "foo", "bar", "foobar", charMap{}, true, "bar"}, + {"escape", "old\x1b]unfinished", "login:", "login:", charMap{}, false, "login:"}, + {"character map", "\r", "\nREADY", "\\r\\nREADY", charMap{crcrlf: true, lfcrlf: true}, false, "\r\nREADY"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + first := &fakePort{reads: [][]byte{[]byte(tt.before)}, readErr: syscall.EIO} + second := &fakePort{reads: [][]byte{[]byte(tt.after)}, readErr: errExhausted} + s, _ := serialWithOpens(t, openResult{port: first}, openResult{port: second}) + s.imap = tt.maps + sess := &recordingSession{} + err := s.Run(context.Background(), sess, "-reconnect", "-t", "2s", "expect", tt.pattern) + if tt.wantError && !errors.Is(err, errExhausted) || !tt.wantError && err != nil { + t.Fatalf("Run = %v, wantError=%v", err, tt.wantError) + } + if !strings.Contains(sess.out.String(), tt.wantOutput) { + t.Errorf("new output lost/filter state carried over: %q", sess.out.String()) + } + }) + } +} + +func TestReconnectDoesNotRepeatCompletedSteps(t *testing.T) { + first := &fakePort{reads: [][]byte{[]byte("prompt")}, readErr: syscall.EIO} + second := &fakePort{reads: [][]byte{[]byte("READY")}} + s, _ := serialWithOpens(t, openResult{port: first}, openResult{port: second}) + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "-t", "2s", + "expect", "prompt", "send", "reboot", "expect", "READY") + if err != nil { + t.Fatal(err) + } + if string(first.written) != "reboot\r" || len(second.written) != 0 { + t.Errorf("command replayed: old=%q, new=%q", first.written, second.written) + } +} + +func TestReconnectInitialSendWaitsWithoutFlushing(t *testing.T) { + p := &fakePort{reads: [][]byte{[]byte("READY")}} + s, _ := serialWithOpens(t, openResult{err: syscall.ENOENT}, openResult{port: p}) + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "-t", "2s", "send", "go", "expect", "READY") + if err != nil || string(p.written) != "go\r" || p.resetCount != 0 { + t.Errorf("Run=%v, written=%q, flushes=%d", err, p.written, p.resetCount) + } +} + +type partialWritePort struct { + fakePort + calls int + zero bool +} + +func (p *partialWritePort) Write(data []byte) (int, error) { + p.calls++ + if p.zero { + return 0, nil + } + if p.calls == 1 { + p.written = append(p.written, data[0]) + return 1, nil + } + p.written = append(p.written, data[0]) + return 1, syscall.EIO +} + +func TestReconnectNeverReplaysInterruptedSend(t *testing.T) { + for _, zero := range []bool{false, true} { + t.Run(fmt.Sprintf("zero=%v", zero), func(t *testing.T) { + p := &partialWritePort{zero: zero} + s, calls := serialWithOpens(t, openResult{port: p}, openResult{port: &fakePort{}}) + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "send", "reboot") + if err == nil || !strings.Contains(err.Error(), "delivery uncertain") || !strings.Contains(err.Error(), "step 1") { + t.Fatalf("Run = %v, want uncertain delivery at step 1", err) + } + if calls.Load() != 1 || !p.closed { + t.Errorf("opens=%d, closed=%v; send must fail without reopening", calls.Load(), p.closed) + } + if !zero && string(p.written) != "re" { + t.Errorf("written=%q, want only interrupted prefix", p.written) + } + }) + } +} + +func TestReconnectGlobalDeadlineIncludesWaiting(t *testing.T) { + for _, args := range [][]string{ + {}, {"expect", "READY"}, {"send", "go"}, + } { + t.Run(strings.Join(args, " "), func(t *testing.T) { + s, calls := serialWithOpens(t) + start := time.Now() + err := s.Run(context.Background(), &mock.Session{}, append([]string{"-reconnect", "-t", "40ms"}, args...)...) + if (err != nil) != (len(args) > 0) { + t.Errorf("Run=%v for %q; monitor deadline succeeds, unfinished steps fail", err, args) + } + if elapsed := time.Since(start); elapsed > 500*time.Millisecond || calls.Load() != 1 { + t.Errorf("deadline ignored: elapsed=%s, opens=%d", elapsed, calls.Load()) + } + }) + } +} + +func TestReconnectDeadlineDoesNotRestartAfterFlap(t *testing.T) { + first := &fakePort{readErr: syscall.EIO} + second := &fakePort{readErr: syscall.EIO} + s, calls := serialWithOpens(t, openResult{port: first}, openResult{port: second}) + start := time.Now() + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "-t", "350ms", "expect", "READY") + if err == nil || !strings.Contains(err.Error(), "timeout of 350ms") { + t.Fatalf("Run=%v, want original global timeout", err) + } + if elapsed := time.Since(start); elapsed > 650*time.Millisecond || calls.Load() != 2 { + t.Errorf("timeout restarted: elapsed=%s, opens=%d", elapsed, calls.Load()) + } +} + +func TestReconnectCancellationWithoutTimeout(t *testing.T) { + s, calls := serialWithOpens(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- s.Run(ctx, &mock.Session{}, "-reconnect", "expect", "READY") }() + waitReconnect(t, func() bool { return calls.Load() > 0 }) + cancel() + err := awaitReconnectRun(t, done) + if err == nil || !strings.Contains(err.Error(), "canceled") { + t.Errorf("Run=%v, want canceled expect", err) + } +} + +func TestReconnectFatalErrorsAreNotRetried(t *testing.T) { + for _, failure := range []error{ + syscall.EACCES, syscall.EPERM, syscall.EBUSY, errors.New("bad baud rate"), + &portConfigurationError{syscall.EIO}, + } { + for _, onRead := range []bool{false, true} { + t.Run(fmt.Sprintf("%v/read=%v", failure, onRead), func(t *testing.T) { + result := openResult{err: fmt.Errorf("wrapped: %w", failure)} + if onRead { + result = openResult{port: &fakePort{readErr: result.err}} + } + s, calls := serialWithOpens(t, result) + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "-t", "1s") + if !errors.Is(err, failure) || calls.Load() != 1 { + t.Errorf("Run=%v, opens=%d; fatal error must be returned immediately", err, calls.Load()) + } + }) + } + } +} + +func TestSerialWithoutReconnectStillFailsImmediately(t *testing.T) { + for _, onRead := range []bool{false, true} { + t.Run(fmt.Sprintf("read=%v", onRead), func(t *testing.T) { + result := openResult{err: syscall.ENOENT} + if onRead { + result = openResult{port: &fakePort{readErr: syscall.EIO}} + } + s, calls := serialWithOpens(t, result) + err := s.Run(context.Background(), &mock.Session{}) + if err == nil || calls.Load() != 1 { + t.Errorf("legacy Run=%v, opens=%d", err, calls.Load()) + } + }) + } +} + +func TestReconnectMonitorResumesOutput(t *testing.T) { + first := &fakePort{reads: [][]byte{[]byte("before\n")}, readErr: syscall.EIO} + second := &fakePort{reads: [][]byte{[]byte("after\n")}} + s, _ := serialWithOpens(t, openResult{port: first}, openResult{port: second}) + sess := &recordingSession{} + err := s.Run(context.Background(), sess, "-reconnect", "-t", "350ms") + if err != nil || !strings.Contains(sess.out.String(), "after\n") || !second.closed { + t.Errorf("Run=%v, output=%q, closed=%v", err, sess.out.String(), second.closed) + } +} + +func TestReconnectInteractiveDiscardsOfflineInput(t *testing.T) { + run := newInteractiveRun([]byte("first\r\n")) + second := newAsyncPort([]byte("second\r\n")) + s, _ := serialWithOpens(t, openResult{port: run.port}, openResult{err: syscall.ENOENT}, openResult{port: second}) + run.serial = s + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := run.start(ctx, "-reconnect") + t.Cleanup(func() { cancel(); _ = run.input.Close() }) + waitReconnect(t, func() bool { return strings.Contains(run.stdout.String(), "first") }) + _, err := run.input.Write([]byte("before\r")) + if err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return len(run.port.writtenBytes()) > 0 }) + run.port.mu.Lock() + run.port.readErr = syscall.EIO + run.port.mu.Unlock() + waitReconnect(t, func() bool { return strings.Contains(run.stderr.String(), "Input while disconnected is discarded") }) + _, err = run.input.Write([]byte("must-not-replay\r")) + if err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return run.stdin.inRead.Load() > 0 }) + waitReconnect(t, func() bool { return strings.Contains(run.stdout.String(), "second") }) + _, err = run.input.Write([]byte("after\r")) + if err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return len(second.writtenBytes()) > 0 }) + cancel() + if err := awaitReconnectRun(t, done); err != nil { + t.Fatal(err) + } + if got := string(run.port.writtenBytes()); got != "before\r" { + t.Errorf("old port input=%q", got) + } + if got := string(second.writtenBytes()); got != "after\r" { + t.Errorf("new port input=%q; offline input must not be replayed", got) + } + if got := run.stdout.String(); got != "first\r\nsecond\r\n" { + t.Errorf("stdout contains notices or lost bytes: %q", got) + } + markers := run.stderr.String() + if strings.Contains(strings.ReplaceAll(markers, "\r\n", ""), "\n") { + t.Errorf("raw terminal notice used bare LF: %q", markers) + } + if strings.Count(markers, "Waiting for") != 1 || strings.Count(markers, "reconnected:") != 1 { + t.Errorf("duplicate/missing notices: %q", markers) + } + if run.stdin.inRead.Load() != 0 || !run.port.isClosed() || !second.isClosed() { + t.Error("interactive teardown left an input reader or open port") + } + if !run.sess.ConsoleOpened || run.sess.PrintCalled { + t.Error("interactive run must keep one console, with no Print output") + } +} + +func TestReconnectInteractiveAbsentDeviceCanExit(t *testing.T) { + for _, end := range []string{"cancel", "eof", "deadline", "client gone", "fatal open"} { + t.Run(end, func(t *testing.T) { + run := newInteractiveRun() + s, _ := serialWithOpens(t) + run.serial = s + if end == "client gone" { + run.sess.Stderr = closedPipeWriter{} + } + if end == "fatal open" { + run.serial, _ = serialWithOpens(t, openResult{err: syscall.EACCES}) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + args := []string{"-reconnect"} + if end == "deadline" { + args = append(args, "-t", "40ms") + } + done := run.start(ctx, args...) + if end == "cancel" || end == "eof" { + waitReconnect(t, func() bool { return strings.Contains(run.stderr.String(), "Waiting for") }) + if end == "cancel" { + cancel() + } else { + _ = run.input.Close() + } + } + err := awaitReconnectRun(t, done) + if (err != nil) != (end == "fatal open") { + t.Errorf("Run=%v, end=%s", err, end) + } + if run.stdin.inRead.Load() != 0 { + t.Error("input goroutine outlived Run") + } + if _, err := run.input.Write([]byte("x")); !errors.Is(err, io.ErrClosedPipe) { + t.Errorf("stdin left open: %v", err) + } + }) + } +} + +type gatedPort struct { + *asyncPort + readStarted, readEnd, readReturned chan struct{} + writeStarted, writeEnd chan struct{} + reading, writing, unsafeClose atomic.Bool +} + +func (p *gatedPort) Read([]byte) (int, error) { + if p.reading.CompareAndSwap(false, true) { + close(p.readStarted) + } + <-p.readEnd + close(p.readReturned) + return 0, syscall.EIO +} + +func (p *gatedPort) Write(data []byte) (int, error) { + p.writing.Store(true) + close(p.writeStarted) + <-p.writeEnd + _, _ = p.asyncPort.Write(data[:1]) + p.writing.Store(false) + return 1, syscall.EIO +} + +func (p *gatedPort) Close() error { + if p.writing.Load() { + p.unsafeClose.Store(true) + } + return p.asyncPort.Close() +} + +func TestReconnectWaitsForInFlightInteractiveWrite(t *testing.T) { + first := &gatedPort{ + asyncPort: newAsyncPort(), + readStarted: make(chan struct{}), readEnd: make(chan struct{}), readReturned: make(chan struct{}), + writeStarted: make(chan struct{}), writeEnd: make(chan struct{}), + } + second := newAsyncPort([]byte("new boot")) + run := newInteractiveRun() + run.serial, _ = serialWithOpens(t, openResult{port: first}, openResult{port: second}) + ctx, cancel := context.WithCancel(context.Background()) + var readOnce, writeOnce sync.Once + releaseRead := func() { readOnce.Do(func() { close(first.readEnd) }) } + releaseWrite := func() { writeOnce.Do(func() { close(first.writeEnd) }) } + t.Cleanup(func() { cancel(); releaseRead(); releaseWrite(); _ = run.input.Close() }) + done := run.start(ctx, "-reconnect") + select { + case <-first.readStarted: + case <-time.After(time.Second): + t.Fatal("read did not start") + } + _, err := run.input.Write([]byte("old-command\r")) + if err != nil { + t.Fatal(err) + } + select { + case <-first.writeStarted: + case <-time.After(time.Second): + t.Fatal("write did not start") + } + releaseRead() + <-first.readReturned + if first.isClosed() { + t.Fatal("port closed under the in-flight write") + } + releaseWrite() + waitReconnect(t, func() bool { return strings.Contains(run.stdout.String(), "new boot") }) + _, err = run.input.Write([]byte("new-command\r")) + if err != nil { + t.Fatal(err) + } + waitReconnect(t, func() bool { return len(second.writtenBytes()) > 0 }) + cancel() + if err := awaitReconnectRun(t, done); err != nil { + t.Fatal(err) + } + if first.unsafeClose.Load() || !first.isClosed() || !second.isClosed() { + t.Error("unsafe close or port left open") + } + if string(first.writtenBytes()) != "o" || string(second.writtenBytes()) != "new-command\r" { + t.Errorf("interrupted input replayed: old=%q, new=%q", first.writtenBytes(), second.writtenBytes()) + } + if run.stdin.inRead.Load() != 0 { + t.Error("input goroutine outlived Run") + } +} + +func TestReconnectFinalDrainHonorsGlobalDeadline(t *testing.T) { + s, _ := serialWithOpens(t, openResult{port: &fakePort{}}) + s.drainTimeout = time.Second + start := time.Now() + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "-t", "40ms", "send", "go") + if err != nil || time.Since(start) > 400*time.Millisecond { + t.Errorf("final drain ignored global deadline: err=%v, elapsed=%s", err, time.Since(start)) + } +} + +func TestReconnectFinalDrainDoesNotHideFatalReopenError(t *testing.T) { + first := &fakePort{readErr: syscall.EIO} + s, _ := serialWithOpens(t, openResult{port: first}, openResult{err: syscall.EACCES}) + s.drainTimeout = time.Second + err := s.Run(context.Background(), &mock.Session{}, "-reconnect", "send", "reboot") + if !errors.Is(err, syscall.EACCES) { + t.Errorf("Run=%v, want fatal reopen error from final drain", err) + } +} + +func TestReconnectReadTimeoutIsNotDisconnection(t *testing.T) { + s, calls := serialWithOpens(t, openResult{port: &fakePort{}}) + sess := &recordingSession{} + err := s.Run(context.Background(), sess, "-reconnect", "-t", "20ms") + if err != nil || calls.Load() != 1 || strings.Contains(sess.out.String(), "disconnected") { + t.Errorf("quiet port treated as unplugged: err=%v, opens=%d, out=%q", err, calls.Load(), sess.out.String()) + } +} diff --git a/pkg/module/serial/serial-example-cfg.yml b/pkg/module/serial/serial-example-cfg.yml index 7391804d..e444ef5f 100644 --- a/pkg/module/serial/serial-example-cfg.yml +++ b/pkg/module/serial/serial-example-cfg.yml @@ -8,10 +8,12 @@ devices: Demo of the Serial module as a passthrough module: the client supplies the send/expect script at runtime, e.g. dutctl server serial -t 60s -- expect 'login:' send root expect '# ' + Keep the console open across USB disconnections with: + dutctl server serial -reconnect -i uses: - module: serial passthrough: true with: - port: /tmp/ttyS0 + port: /dev/serial/by-id/usb-example-console baud: 115200 delay: 50ms diff --git a/pkg/module/serial/serial.go b/pkg/module/serial/serial.go index 93a2dd44..ed8b2305 100644 --- a/pkg/module/serial/serial.go +++ b/pkg/module/serial/serial.go @@ -85,9 +85,9 @@ const abstract = `Scripted or interactive serial connection to the DUT const usage = ` ARGUMENTS: - [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] (monitor: stream output) - [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] [--] ... (run a step sequence) - [-t ] -i (interactive console) + [-reconnect] [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] (monitor) + [-reconnect] [-t ] [-eol cr|lf|crlf|none] [-keep-escapes] [--] ... (step sequence) + [-reconnect] [-t ] -i (interactive) step := expect | send | send-raw @@ -139,6 +139,20 @@ FLAGS (before the steps): queries) instead of stripping them from the output. -i Interactive console. Takes no steps, and no -eol or -keep-escapes: the bytes pass verbatim. + -reconnect Wait for an absent device and reopen after disconnect. + Off by default; permission/busy/configuration errors fail. + +With -reconnect, use a stable /dev/serial/by-id/... path. The same session stays +open, with disconnected/waiting/reconnected notices (stderr with -i). Reopening +reapplies the configured baud rate and preserves queued boot output. Output +emitted while the adapter is absent cannot be recovered. The original -t deadline +includes waiting and never restarts; monitor/interactive deadlines still succeed. + +Interactive input received while offline is discarded, with a notice, and is +never replayed. A pending expect resumes with an empty matcher and filter state; +completed steps are not repeated. An attempted send that fails ends the script +with uncertain delivery instead of retrying it. A send not yet attempted may wait +for the device. Without -reconnect, device errors still end the run immediately. The device's character maps (the imap configuration option, named as picocom names them: lfcrlf for a device that ends its lines in a bare LF, which would @@ -156,6 +170,7 @@ right after a send can match your own input rather than the device's reply. EXAMPLES: monitor the console: (no arguments) interactive console: -i + monitor across power cycles: -reconnect wait for a boot marker: -- expect 'Welcome to' login then run a command: -- expect 'login:' send root expect '# ' send reboot send Ctrl-C then expect shell: -- send-raw '\x03' expect '$ ' @@ -237,7 +252,7 @@ func defaultOpenPort(name string, baud int) (port, error) { if err != nil { _ = serialPort.Close() - return nil, fmt.Errorf("failed to set read timeout on %s: %w", name, err) + return nil, &portConfigurationError{fmt.Errorf("failed to set read timeout on %s: %w", name, err)} } return serialPort, nil @@ -250,8 +265,8 @@ func defaultOpenPort(name string, baud int) (port, error) { // failing on the first expect that times out. With -i it opens a raw console // and passes bytes verbatim until the session ends, -t elapses, or the user's // input ended (all a success). -// -//nolint:cyclop,funlen // monitor/sequence/interactive dispatch with pacing and drain; the branch count is inherent +// With -reconnect the same session waits for absent devices and recovers read +// disconnections within the original deadline; attempted writes are never replayed. func (s *Serial) Run(ctx context.Context, session module.Session, args ...string) error { // The logger carried on ctx is already scoped to this module by the agent // (scope "module", with module/device/command attributes), so the module @@ -265,6 +280,10 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string return err } + if cfg.reconnect { + return s.runReconnecting(ctx, session, cfg) + } + opener := s.open if opener == nil { opener = defaultOpenPort @@ -312,13 +331,20 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string clientOut := newClientWriter(session) clientOut.markerf("--- Connected to %s at %d baud ---\n", s.Port, s.Baud) + return s.runSequence(loopCtx, ctx, cfg, serialPort, clientOut) +} + +//nolint:cyclop // monitor/sequence dispatch with pacing, step progress and final drain +func (s *Serial) runSequence( + ctx, drainCtx context.Context, cfg scriptConfig, serialPort port, clientOut *clientWriter, +) error { eng := newEngine(serialPort, newInbound(s.imap, !cfg.keepEscapes), clientOut) // Monitor mode: no steps — stream the console until cancelled or -t elapses. if len(cfg.steps) == 0 { - l.Debug("monitor mode; streaming until cancelled") + log.FromContext(ctx).Debug("monitor mode; streaming until cancelled") - err = eng.monitor(loopCtx) + err := eng.monitor(ctx) if err != nil { return fmt.Errorf("monitor: %w", err) } @@ -333,7 +359,7 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string for idx, curStep := range cfg.steps { switch curStep.kind { case stepExpect: - err = eng.readUntil(loopCtx, curStep.expect) + err := eng.readUntil(ctx, curStep.expect) if err != nil { return stepError(idx, curStep, cfg.timeout, err) } @@ -341,7 +367,7 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string clientOut.markerf("--- [%d/%d] matched %q ---\n", idx+1, total, curStep.label()) case stepSend, stepSendRaw: // Pace input: pause before each send (interruptible by the deadline). - err = sleepCtx(loopCtx, s.delay) + err := sleepCtx(ctx, s.delay) if err != nil { return fmt.Errorf("step %d (send %q): %w", idx+1, curStep.label(), err) } @@ -356,9 +382,9 @@ func (s *Serial) Run(ctx context.Context, session module.Session, args ...string } // If the sequence ended on a send, drain briefly so the DUT's reply to the - // final input is visible. Uses the original ctx (its own window). + // final input is visible. drainCtx preserves the legacy window without -reconnect. if cfg.steps[total-1].kind != stepExpect { - err = eng.drain(ctx, s.drainFor()) + err := eng.drain(drainCtx, s.drainFor(), !cfg.reconnect) if err != nil { return fmt.Errorf("draining after final send: %w", err) } diff --git a/pkg/module/serial/step.go b/pkg/module/serial/step.go index 74d7f31f..681a28d7 100644 --- a/pkg/module/serial/step.go +++ b/pkg/module/serial/step.go @@ -61,6 +61,7 @@ type scriptConfig struct { timeout time.Duration keepEscapes bool interactive bool + reconnect bool steps []step } @@ -78,12 +79,14 @@ func parseArgs(args []string) (scriptConfig, error) { eolName string keepEscapes bool interactive bool + reconnect bool ) fs.DurationVar(&timeout, "t", 0, "global timeout for the whole run (e.g. 30s, 3m); 0 = no timeout") fs.StringVar(&eolName, "eol", defaultEOL, "line ending appended by 'send': cr|lf|crlf|none") fs.BoolVar(&keepEscapes, "keep-escapes", false, "keep terminal escape sequences in the output instead of stripping them") fs.BoolVar(&interactive, "i", false, "interactive console: bridge the port to the client's terminal") + fs.BoolVar(&reconnect, "reconnect", false, "wait for an absent serial device and reconnect after disconnection") err := fs.Parse(args) if err != nil { @@ -96,7 +99,7 @@ func parseArgs(args []string) (scriptConfig, error) { return scriptConfig{}, err } - return scriptConfig{timeout: timeout, interactive: true}, nil + return scriptConfig{timeout: timeout, interactive: true, reconnect: reconnect}, nil } eol, err := resolveEOL(eolName) @@ -109,7 +112,7 @@ func parseArgs(args []string) (scriptConfig, error) { return scriptConfig{}, err } - return scriptConfig{timeout: timeout, keepEscapes: keepEscapes, steps: steps}, nil + return scriptConfig{timeout: timeout, keepEscapes: keepEscapes, reconnect: reconnect, steps: steps}, nil } // checkInteractive rejects what -i cannot be combined with: steps, and the