From b1815fc588727bac3c6b9a7e9c98b63ed6a24ccc Mon Sep 17 00:00:00 2001 From: Christian Walter Date: Fri, 6 Nov 2026 13:10:41 +0100 Subject: [PATCH] feat: reconnect temporarily disconnected serial devices Add opt-in reconnect waiting for monitor, interactive and scripted serial runs. Preserve the global deadline, reset stream state at disconnect boundaries and discard offline console input. Never replay an attempted write. Cover initial absence, repeated flaps, deadlines, fatal errors, partial writes and teardown, including real Linux PTY replacement. Signed-off-by: Christian Walter --- pkg/module/serial/README.md | 50 ++- pkg/module/serial/bridge.go | 6 +- pkg/module/serial/engine.go | 24 +- pkg/module/serial/engine_test.go | 2 +- pkg/module/serial/inbound.go | 8 +- pkg/module/serial/reconnect.go | 288 ++++++++++++ pkg/module/serial/reconnect_linux_test.go | 86 ++++ pkg/module/serial/reconnect_test.go | 518 ++++++++++++++++++++++ pkg/module/serial/serial-example-cfg.yml | 4 +- pkg/module/serial/serial.go | 50 ++- pkg/module/serial/step.go | 7 +- 11 files changed, 1013 insertions(+), 30 deletions(-) create mode 100644 pkg/module/serial/reconnect.go create mode 100644 pkg/module/serial/reconnect_linux_test.go create mode 100644 pkg/module/serial/reconnect_test.go 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