From d6d0dc5f598de04fcdae4afde5d9e26f191a6001 Mon Sep 17 00:00:00 2001 From: Malash Date: Sun, 11 Oct 2026 19:20:29 +0800 Subject: [PATCH] fix(modem): prevent AT command response desynchronization --- internal/device/cell_lock.go | 9 +- internal/device/cell_lock_test.go | 10 +- internal/device/cellular_ims.go | 8 +- internal/device/cellular_ims_test.go | 42 +- internal/device/manager.go | 16 +- internal/device/mbn.go | 6 +- internal/device/sms.go | 9 - internal/device/sms_test.go | 7 +- internal/device/snapshot.go | 23 +- internal/device/snapshot_guard_test.go | 78 ++- internal/modem/serial_linux.go | 41 +- internal/modem/serial_linux_test.go | 77 +++ internal/modem/session.go | 674 +++++++++--------- internal/modem/session_test.go | 920 +++++++++++++++---------- internal/modem/wwan_at_linux.go | 70 +- internal/modem/wwan_at_linux_test.go | 75 +- 16 files changed, 1228 insertions(+), 837 deletions(-) diff --git a/internal/device/cell_lock.go b/internal/device/cell_lock.go index 6c8c5a50..1ea9d7ef 100644 --- a/internal/device/cell_lock.go +++ b/internal/device/cell_lock.go @@ -226,18 +226,13 @@ func (manager *Manager) prepareCellLock(ctx context.Context, id string) (*manage return state, queries, nil } -// Called with opMu held; discard the client on missing finals to isolate late replies. func (manager *Manager) cellLockCommand(ctx context.Context, state *managedDevice, command string) (modem.Response, error) { if err := ctx.Err(); err != nil { return modem.Response{}, err } response, err := manager.command(ctx, state.client, command) - if response.Final == "" { - _ = state.client.Close() - state.client = nil - if err == nil { - err = fmt.Errorf("%s: modem command completed without a final result", command) - } + if response.Final == "" && err == nil { + err = fmt.Errorf("%s: modem command completed without a final result", command) } return response, err } diff --git a/internal/device/cell_lock_test.go b/internal/device/cell_lock_test.go index b09f39a1..dc7e75d0 100644 --- a/internal/device/cell_lock_test.go +++ b/internal/device/cell_lock_test.go @@ -84,10 +84,9 @@ func TestSetCellLockFailures(t *testing.T) { for _, test := range []struct { name string setter clientStep - closes int }{ {name: "setter rejected", setter: rejectedCommand(command)}, - {name: "missing final", setter: clientStep{command: command, err: context.DeadlineExceeded}, closes: 1}, + {name: "missing final", setter: clientStep{command: command, err: context.DeadlineExceeded}}, {name: "readback mismatch", setter: clientStep{command: command, response: okResponse()}}, } { t.Run(test.name, func(t *testing.T) { @@ -108,8 +107,11 @@ func TestSetCellLockFailures(t *testing.T) { if test.setter.err != nil && !errors.Is(err, test.setter.err) { t.Fatalf("lost modem error: %v", err) } - if client.closeCount != test.closes { - t.Fatalf("client closed %d times, want %d", client.closeCount, test.closes) + if client.closeCount != 0 { + t.Fatalf("client closed %d times, want 0", client.closeCount) + } + if manager.devices[id].client != client { + t.Fatal("client not retained after unconfirmed write") } client.assertDone(t) }) diff --git a/internal/device/cellular_ims.go b/internal/device/cellular_ims.go index 814257cc..0f5e7de9 100644 --- a/internal/device/cellular_ims.go +++ b/internal/device/cellular_ims.go @@ -186,10 +186,12 @@ func (manager *Manager) SetCellularIMS(ctx context.Context, id string, mode Cell rebootCtx, cancel := manager.withTimeout(ctx, manager.longTimeout) _, err = client.Execute(rebootCtx, "AT+CFUN=1,1") cancel() - if closeErr := client.Close(); err == nil { - err = closeErr + if !isCommandInterrupted(err) { + if closeErr := client.Close(); err == nil { + err = closeErr + } + state.client = nil } - state.client = nil state.preFlightMode = nil manager.clearSnapshot(id, state) manager.setResult(id, state, nil, err) diff --git a/internal/device/cellular_ims_test.go b/internal/device/cellular_ims_test.go index f5a27ef3..c5fe57b7 100644 --- a/internal/device/cellular_ims_test.go +++ b/internal/device/cellular_ims_test.go @@ -2,6 +2,7 @@ package device import ( "context" + "errors" "testing" "vocat/internal/modem" @@ -49,22 +50,33 @@ func TestParseCellularCSRegistration(t *testing.T) { } func TestSetCellularIMSEnablesAndRebootsOnlyOnce(t *testing.T) { - client := &transcriptClient{steps: []clientStep{ - {command: `AT+QCFG="ims"`, response: modem.Response{Lines: []string{`+QCFG: "ims",0,0`}, Final: "OK"}}, - {command: "AT+CFUN?", response: okResponse("+CFUN: 1")}, - {command: `AT+QCFG="ims",1`, response: modem.Response{Final: "OK"}}, - {command: `AT+QCFG="ims"`, response: modem.Response{Lines: []string{`+QCFG: "ims",1,0`}, Final: "OK"}}, - {command: "AT+CFUN=1,1", response: modem.Response{Final: "OK"}}, - }} - manager, id := newStartedTestManager(t, client) - status, err := manager.SetCellularIMS(context.Background(), id, CellularIMSModeForceEnabled) - if err != nil || !status.Configured || !status.Changed || !status.Rebooting { - t.Fatalf("SetCellularIMS = %+v, %v", status, err) + for name, rebootErr := range map[string]error{"success": nil, "timeout": modem.ErrCommandTimeout, "canceled": context.Canceled} { + t.Run(name, func(t *testing.T) { + client := &transcriptClient{steps: []clientStep{ + {command: `AT+QCFG="ims"`, response: modem.Response{Lines: []string{`+QCFG: "ims",0,0`}, Final: "OK"}}, + {command: "AT+CFUN?", response: okResponse("+CFUN: 1")}, + {command: `AT+QCFG="ims",1`, response: modem.Response{Final: "OK"}}, + {command: `AT+QCFG="ims"`, response: modem.Response{Lines: []string{`+QCFG: "ims",1,0`}, Final: "OK"}}, + {command: "AT+CFUN=1,1", response: modem.Response{Final: "OK"}, err: rebootErr}, + }} + manager, id := newStartedTestManager(t, client) + status, err := manager.SetCellularIMS(context.Background(), id, CellularIMSModeForceEnabled) + if !errors.Is(err, rebootErr) || !status.Configured || !status.Changed || !status.Rebooting { + t.Fatalf("SetCellularIMS = %+v, %v", status, err) + } + wantCloses := 1 + if rebootErr != nil { + wantCloses = 0 + if manager.devices[id].client != client { + t.Fatal("IMS restart discarded the session before its final") + } + } + if client.closeCount != wantCloses { + t.Fatalf("close count = %d, want %d", client.closeCount, wantCloses) + } + client.assertDone(t) + }) } - if client.closeCount != 1 { - t.Fatalf("close count = %d, want 1", client.closeCount) - } - client.assertDone(t) } func TestSetCellularIMSNoopDoesNotReboot(t *testing.T) { diff --git a/internal/device/manager.go b/internal/device/manager.go index dbea068e..e0badcc4 100644 --- a/internal/device/manager.go +++ b/internal/device/manager.go @@ -880,16 +880,24 @@ func (manager *Manager) Reboot(ctx context.Context, id string) error { commandCtx, cancel := manager.withTimeout(ctx, manager.longTimeout) defer cancel() _, err = client.Execute(commandCtx, "AT+CFUN=1,1") - if closeErr := client.Close(); err == nil { - err = closeErr + if !isCommandInterrupted(err) { + if closeErr := client.Close(); err == nil { + err = closeErr + } + state.client = nil } - state.client = nil state.preFlightMode = nil manager.clearSnapshot(id, state) manager.setResult(id, state, nil, err) return err } +// Interrupted commands still own their late replies; reopening the port would +// let those replies complete an unrelated command in the new session. +func isCommandInterrupted(err error) bool { + return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, modem.ErrCommandTimeout) +} + // softResetForProfileSwitch resets the baseband SIM stack using a soft CFUN sequence // (AT+CFUN=0 -> AT+CFUN=4) instead of rebooting the entire hardware module (AT+CFUN=1,1). // This causes the baseband to reload the new eSIM profile files within ~1-2 seconds @@ -927,7 +935,7 @@ func (manager *Manager) softResetSIMLocked(ctx context.Context, id string, state } if _, err := client.Execute(commandCtx, "AT+CFUN=0"); err != nil { // 超时或取消时不能确认 CFUN=0 是否已生效，仍尝试恢复到 RF 关闭的工作模式。 - if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, modem.ErrCommandTimeout) { + if isCommandInterrupted(err) { reloadCtx, cancelReload := context.WithTimeout(context.WithoutCancel(ctx), manager.commandTimeout) _, reloadErr := client.Execute(reloadCtx, "AT+CFUN=4") cancelReload() diff --git a/internal/device/mbn.go b/internal/device/mbn.go index c9b46605..c56b730e 100644 --- a/internal/device/mbn.go +++ b/internal/device/mbn.go @@ -429,8 +429,10 @@ func (manager *Manager) ReconcileEC20MBNAfterProfileSwitch(ctx context.Context, rebootContext, cancelReboot := context.WithTimeout(ctx, manager.longTimeout) _, rebootErr := client.Execute(rebootContext, "AT+CFUN=1,1") cancelReboot() - _ = client.Close() - state.client = nil + if !isCommandInterrupted(rebootErr) { + _ = client.Close() + state.client = nil + } state.preFlightMode = nil manager.clearSnapshot(id, state) state.opMu.Unlock() diff --git a/internal/device/sms.go b/internal/device/sms.go index 8a3f4cc8..fd4a54b5 100644 --- a/internal/device/sms.go +++ b/internal/device/sms.go @@ -131,15 +131,6 @@ func (manager *Manager) SendSMS( default: result.SubmissionStatus = "unknown" } - if errors.Is(submitErr, modem.ErrCommandTimeout) || - errors.Is(submitErr, context.Canceled) || - errors.Is(submitErr, context.DeadlineExceeded) { - // A timeout after Ctrl-Z has an inherently uncertain outcome. Close - // this session so a late +CMGS/OK cannot corrupt the next command; - // callers must decide whether it is safe to retry. - _ = client.Close() - state.client = nil - } partErr := fmt.Errorf( "submit SMS part %d/%d: %w", part.part, diff --git a/internal/device/sms_test.go b/internal/device/sms_test.go index 9c835e27..04ef8180 100644 --- a/internal/device/sms_test.go +++ b/internal/device/sms_test.go @@ -260,8 +260,11 @@ func TestManagerSendSMSTimeoutNeverClaimsAcceptanceOrDelivery(t *testing.T) { client.mu.Lock() closeCount := client.closeCount client.mu.Unlock() - if closeCount != 1 { - t.Fatalf("close count = %d, want 1 after uncertain timeout", closeCount) + if closeCount != 0 { + t.Fatalf("close count = %d, want 0 after uncertain timeout", closeCount) + } + if manager.devices[id].client != client { + t.Fatal("client not retained after uncertain timeout") } client.assertDone(t) } diff --git a/internal/device/snapshot.go b/internal/device/snapshot.go index 95676030..1462286e 100644 --- a/internal/device/snapshot.go +++ b/internal/device/snapshot.go @@ -50,8 +50,8 @@ func (manager *Manager) readSnapshot( // Native MHI/QMI devices expose their immutable modem identity through DMS. // Read it before any SIM-dependent AT probes: a missing/bad card can make // those commands slow or fail, but must never prevent IMEI from appearing. - if strings.EqualFold(strings.TrimSpace(backend), "qmi") && isNativeQMICandidate(candidate) { - qmiContext, cancelQMI := manager.withTimeout(ctx, manager.commandTimeout*5) + if isNativeQMICandidate(candidate) { + qmiContext, cancelQMI := context.WithTimeout(ctx, manager.commandTimeout) qmiIMEI, qmiErr := manager.readNativeQMIIMEI(qmiContext, candidate) cancelQMI() if qmiErr == nil { @@ -252,12 +252,9 @@ func (manager *Manager) readSnapshot( snapshot.RegistrationStatus = 1 snapshot.RegistrationSource = "COPS" } - if snapshot.IMEI == "" { - // AT+CGSN on some MHI modems (the UFI dongle behind the OpenStick 410) - // returns the IMEI line but never a final OK, so it would block until the - // caller's deadline (30s during a periodic refresh) and starve every other - // device operation behind the lock. Give it an independent short timeout - // and let the WWAN transport's drain discard the trailing stale bytes. + // Some native modems never finish CGSN. Do not leave the AT session + // waiting for its final; native QMI identity is read through DMS instead. + if snapshot.IMEI == "" && !isNativeQMICandidate(candidate) { for _, command := range []string{"AT+CGSN", "AT+CGSN=1"} { cgsnCtx, cancelCGSN := context.WithTimeout(ctx, manager.commandTimeout) cgsnResponse, cgsnErr := manager.command(cgsnCtx, client, command) @@ -270,16 +267,6 @@ func (manager *Manager) readSnapshot( } } } - if snapshot.IMEI == "" && strings.EqualFold(strings.TrimSpace(backend), "qmi") && isNativeQMICandidate(candidate) { - qmiContext, cancelQMI := manager.withTimeout(ctx, manager.commandTimeout*5) - qmiIMEI, qmiErr := manager.readNativeQMIIMEI(qmiContext, candidate) - cancelQMI() - if qmiErr == nil { - snapshot.IMEI = qmiIMEI - } else { - snapshot.Warnings = append(snapshot.Warnings, "read IMEI via QMI DMS: "+qmiErr.Error()) - } - } if snapshot.IMEI == "" && previousSnapshot != nil { // IMEI is hardware identity and does not change with the inserted card. // Preserve a prior successful read across a transient QMI/AT failure. diff --git a/internal/device/snapshot_guard_test.go b/internal/device/snapshot_guard_test.go index f7623c9d..fe15d563 100644 --- a/internal/device/snapshot_guard_test.go +++ b/internal/device/snapshot_guard_test.go @@ -57,10 +57,66 @@ func (c *lenientATClient) saw(command string) bool { return false } -// AT+CGSN on some MHI modems returns the IMEI line but never a final OK, so it -// would block until the caller's deadline and hold the device lock for the -// whole periodic refresh. The snapshot must bound CGSN with its own short -// timeout instead of inheriting the refresh deadline. +func TestSnapshotSkipsCGSNOnlyForNativeQMI(t *testing.T) { + const commandTimeout = 20 * time.Millisecond + for _, test := range []struct { + name, id, backend string + qmiErr error + }{ + {name: "native AT", id: "mhi-wwan0", backend: "at"}, + {name: "native QMI", id: "mhi-wwan0", backend: "qmi"}, + {name: "native AT failure", id: "mhi-wwan0", backend: "at", qmiErr: errors.New("DMS unavailable")}, + {name: "native QMI failure", id: "mhi-wwan0", backend: "qmi", qmiErr: errors.New("DMS unavailable")}, + {name: "native AT timeout", id: "mhi-wwan0", backend: "at", qmiErr: context.DeadlineExceeded}, + {name: "native QMI timeout", id: "mhi-wwan0", backend: "qmi", qmiErr: context.DeadlineExceeded}, + {name: "USB AT", id: "usb-ec20", backend: "at"}, + } { + t.Run(test.name, func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + candidate := modem.Candidate{ID: test.id, QMIControl: "/dev/wwan0qmi0"} + client := &lenientATClient{} + qmiCalls := 0 + manager := &Manager{commandTimeout: commandTimeout, qmiRadioOpener: func(ctx context.Context, _ string) (qmiRadioSession, error) { + qmiCalls++ + if deadline, ok := ctx.Deadline(); !ok || time.Until(deadline) > commandTimeout { + t.Fatal("DMS probe deadline exceeds CommandTimeout") + } + if test.qmiErr == context.DeadlineExceeded { + <-ctx.Done() + return nil, ctx.Err() + } + return &fakeQMIRadioSession{imei: "866241014372802", imeiErr: test.qmiErr}, nil + }} + previous := &Snapshot{IMEI: "866241014372801"} + snapshot, err := manager.readSnapshot(ctx, test.id, candidate, test.backend, "", previous, client) + if err != nil { + t.Fatal(err) + } + if !client.saw("AT+CPIN?") || ctx.Err() != nil { + t.Fatalf("CPIN did not continue with a live parent: commands = %v, error = %v", client.commands, ctx.Err()) + } + if test.qmiErr != nil && !strings.Contains(strings.Join(snapshot.Warnings, "\n"), test.qmiErr.Error()) { + t.Fatalf("warnings = %v, want DMS error %v", snapshot.Warnings, test.qmiErr) + } + wantCGSN := test.id == "usb-ec20" + if client.saw("AT+CGSN") != wantCGSN || client.saw("AT+CGSN=1") != wantCGSN { + t.Fatalf("commands = %v, want CGSN probes = %t", client.commands, wantCGSN) + } + wantCalls, wantIMEI := 0, previous.IMEI + if !wantCGSN { + wantCalls = 1 + if test.qmiErr == nil { + wantIMEI = "866241014372802" + } + } + if qmiCalls != wantCalls || snapshot.IMEI != wantIMEI { + t.Fatalf("DMS calls = %d, IMEI = %q; want %d, %q", qmiCalls, snapshot.IMEI, wantCalls, wantIMEI) + } + }) + } +} + func TestManagerRefreshBoundsCGSNTimeout(t *testing.T) { client := &lenientATClient{cgsnDelay: 5 * time.Second} manager, id := newStartedTestManager(t, client) @@ -92,9 +148,6 @@ func TestManagerRefreshBoundsCGSNTimeout(t *testing.T) { // card that call blocks until its long timeout and starves the AT terminal // behind the device lock. func TestManagerRefreshSkipsQMIICCIDWithoutReadySIM(t *testing.T) { - // CGSN succeeds so the snapshot does not fall back to the QMI DMS IMEI - // read either; the test focuses on the UIM ICCID fallback being skipped - // without a READY card. client := &lenientATClient{cgsnIMEI: "866241014372802"} manager, err := NewManager(Options{ Discoverer: staticDiscoverer{candidates: []modem.Candidate{{ @@ -127,12 +180,13 @@ func TestManagerRefreshSkipsQMIICCIDWithoutReadySIM(t *testing.T) { if err != nil { t.Fatalf("Refresh: %v", err) } - // Exactly one QMI open is expected: the immutable DMS IMEI read runs - // unconditionally for native QMI candidates (IMEI is hardware identity, - // independent of the card). The UIM ICCID fallback, which would block - // without a READY SIM, must be skipped. + // Only the SIM-independent DMS read is attempted; the UIM ICCID fallback + // must still be skipped without a READY SIM. if qmiCalls != 1 { - t.Fatalf("qmiRadioOpener called %d times, want 1 (DMS IMEI only, UIM ICCID must be skipped without a READY SIM)", qmiCalls) + t.Fatalf("qmiRadioOpener called %d times, want 1 DMS attempt and no UIM ICCID read", qmiCalls) + } + if client.saw("AT+CGSN") || client.saw("AT+CGSN=1") || snapshot.IMEI != "" { + t.Fatalf("native IMEI fell back to AT: %q, commands = %v", snapshot.IMEI, client.commands) } for _, warning := range snapshot.Warnings { if strings.Contains(warning, "QMI UIM") { diff --git a/internal/modem/serial_linux.go b/internal/modem/serial_linux.go index fb3d870d..ac0e391b 100644 --- a/internal/modem/serial_linux.go +++ b/internal/modem/serial_linux.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "os" + "sync" "sync/atomic" "syscall" "time" @@ -21,6 +22,7 @@ import ( // read bounded and lets Session observe its context deadline without closing // a transport that is in use. type linuxSerialTransport struct { + mu sync.RWMutex fd atomic.Int64 readTimeout atomic.Int64 } @@ -109,6 +111,8 @@ func (transport *linuxSerialTransport) Read(buffer []byte) (int, error) { if len(buffer) == 0 { return 0, nil } + transport.mu.RLock() + defer transport.mu.RUnlock() fd, err := transport.currentFD() if err != nil { return 0, err @@ -119,13 +123,16 @@ func (transport *linuxSerialTransport) Read(buffer []byte) (int, error) { deadline = time.Now().Add(timeout) } for { - wait := -1 + if transport.fd.Load() < 0 { + return 0, os.ErrClosed + } + wait := 100 if !deadline.IsZero() { remaining := time.Until(deadline) if remaining <= 0 { return 0, nil } - wait = int((remaining + time.Millisecond - 1) / time.Millisecond) + wait = int((min(remaining, 100*time.Millisecond) + time.Millisecond - 1) / time.Millisecond) } fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN | unix.POLLERR | unix.POLLHUP}} _, err = unix.Poll(fds, wait) @@ -135,8 +142,11 @@ func (transport *linuxSerialTransport) Read(buffer []byte) (int, error) { if err != nil { return 0, err } + if transport.fd.Load() < 0 { + return 0, os.ErrClosed + } if fds[0].Revents == 0 { - return 0, nil + continue } if fds[0].Revents&unix.POLLNVAL != 0 { return 0, os.ErrClosed @@ -154,11 +164,16 @@ func (transport *linuxSerialTransport) Read(buffer []byte) (int, error) { } func (transport *linuxSerialTransport) Write(buffer []byte) (int, error) { + transport.mu.RLock() + defer transport.mu.RUnlock() fd, err := transport.currentFD() if err != nil { return 0, err } for { + if transport.fd.Load() < 0 { + return 0, os.ErrClosed + } count, writeErr := unix.Write(fd, buffer) if errors.Is(writeErr, unix.EINTR) { continue @@ -170,19 +185,32 @@ func (transport *linuxSerialTransport) Write(buffer []byte) (int, error) { } continue } - return count, writeErr + return max(count, 0), writeErr } } func (transport *linuxSerialTransport) Drain() error { + transport.mu.RLock() + defer transport.mu.RUnlock() fd, err := transport.currentFD() if err != nil { return err } - return unix.IoctlSetInt(fd, unix.TCSBRK, 1) + for { + if transport.fd.Load() < 0 { + return os.ErrClosed + } + pending, err := unix.IoctlGetInt(fd, unix.TIOCOUTQ) + if err != nil || pending == 0 { + return err + } + time.Sleep(time.Millisecond) + } } func (transport *linuxSerialTransport) ResetInputBuffer() error { + transport.mu.RLock() + defer transport.mu.RUnlock() fd, err := transport.currentFD() if err != nil { return err @@ -199,7 +227,10 @@ func (transport *linuxSerialTransport) SetReadTimeout(timeout time.Duration) err } func (transport *linuxSerialTransport) Close() error { + // Signal pending I/O before waiting; do not release its descriptor early. fd := int(transport.fd.Swap(-1)) + transport.mu.Lock() + defer transport.mu.Unlock() if fd < 0 { return nil } diff --git a/internal/modem/serial_linux_test.go b/internal/modem/serial_linux_test.go index 071c0959..dac04afa 100644 --- a/internal/modem/serial_linux_test.go +++ b/internal/modem/serial_linux_test.go @@ -10,6 +10,7 @@ import ( "io" "os" "os/exec" + "sync" "testing" "time" @@ -43,6 +44,82 @@ func TestLinuxSerialExecHelper(t *testing.T) { os.Exit(0) } +func TestLinuxNativeTransportCloseInterruptsIO(t *testing.T) { + for _, kind := range []string{"serial", "wwan"} { + for _, operation := range []string{"read", "write", "drain"} { + if kind == "wwan" && operation == "drain" { + continue // WWAN Drain never waits for output. + } + t.Run(kind+"/"+operation, func(t *testing.T) { + fd, peer := socketpair(t) + serial := &linuxSerialTransport{} + serial.fd.Store(int64(fd)) + serial.readTimeout.Store(int64(time.Hour)) + var transport Transport = serial + mu, wantErr := &serial.mu, os.ErrClosed + if kind == "wwan" { + wwan := &nativeWWANATTransport{fd: fd, readTimeout: -1} + transport, mu, wantErr = wwan, &wwan.mu, io.ErrClosedPipe + } + defer transport.Close() + defer unix.Close(peer) + if operation != "read" { + for { + _, err := unix.Write(fd, make([]byte, 4096)) + if errors.Is(err, unix.EAGAIN) { + break + } + if err != nil { + t.Fatal(err) + } + } + } + ioDone := make(chan error, 1) + go func() { + var err error + switch operation { + case "read": + _, err = transport.Read(make([]byte, 1)) + case "write": + _, err = transport.Write([]byte("AT\r")) + case "drain": + err = transport.Drain() + } + ioDone <- err + }() + waitForTransportReadLock(t, mu) + closeDone := make(chan error, 1) + go func() { closeDone <- transport.Close() }() + for _, result := range []struct { + done chan error + want error + }{{closeDone, nil}, {ioDone, wantErr}} { + select { + case err := <-result.done: + if !errors.Is(err, result.want) { + t.Fatalf("Close/IO error = %v, want %v", err, result.want) + } + case <-time.After(time.Second): + t.Fatal("Close did not interrupt pending I/O") + } + } + }) + } + } +} + +func waitForTransportReadLock(t *testing.T, mu *sync.RWMutex) { + t.Helper() + deadline := time.Now().Add(time.Second) + for mu.TryLock() { + mu.Unlock() + if time.Now().After(deadline) { + t.Fatal("I/O did not acquire transport lock") + } + time.Sleep(time.Millisecond) + } +} + func TestLinuxSerialCloseReleasesTTYWhileExecChildRemainsAlive(t *testing.T) { for cycle := 0; cycle < 5; cycle++ { t.Run(fmt.Sprintf("reset-%d", cycle), func(t *testing.T) { diff --git a/internal/modem/session.go b/internal/modem/session.go index 4c87d7f9..064c8b43 100644 --- a/internal/modem/session.go +++ b/internal/modem/session.go @@ -8,6 +8,7 @@ import ( "io" "strings" "sync" + "sync/atomic" "syscall" "time" ) @@ -38,26 +39,46 @@ func (options SessionOptions) withDefaults() SessionOptions { return options } -// Session serializes commands for one physical AT port. Reading is intentionally -// performed under the same mutex as writing: this prevents two callers from -// consuming each other's responses while still allowing interleaved URCs to be -// separated and queued. +// Session serializes all access to one physical AT port. A timed-out command +// keeps ownership of its response until its final arrives or recovery expires. type Session struct { - mu sync.Mutex + gate chan struct{} transport Transport options SessionOptions readBuf []byte + readErr error + skipSpace bool urcs []string - closed bool - poisoned bool + pending *atTransaction + poisoned atomic.Bool + + recoveryDeadline time.Time + ioDeadline time.Time + + lifetime context.Context + cancel context.CancelFunc + received chan atRead + readDone chan struct{} + ioDone chan error + closeOnce sync.Once + closeErr error } -// PoisonedClient is implemented by Session. A poisoned session has hit a -// transport-fatal error (a failed write/drain/read or a closed serial line); -// the underlying fd is wedged and every subsequent command reuses the corpse. -// AT-level failures (CommandError) do not poison. A command deadline that has -// to close a blocked transport does poison, because the fd was wedged and must -// be reopened before the next operation. +type atRead struct { + data []byte + err error +} + +type atTransaction struct { + command string + responsePrefix string + commandEcho bool + remainingEcho []string + awaitPrompt bool +} + +// PoisonedClient reports failures that require reopening the port. +// Caller cancellation alone does not poison a session or release its pending reply. type PoisonedClient interface { Poisoned() bool } @@ -70,20 +91,35 @@ func NewSession(transport Transport, options SessionOptions) (*Session, error) { if err := transport.SetReadTimeout(options.ReadTimeout); err != nil { return nil, fmt.Errorf("set serial read timeout: %w", err) } - return &Session{ - transport: transport, - options: options, - }, nil + lifetime, cancel := context.WithCancel(context.Background()) + session := &Session{ + transport: transport, options: options, + gate: make(chan struct{}, 1), lifetime: lifetime, cancel: cancel, + received: make(chan atRead, 8), readDone: make(chan struct{}), + } + go session.readLoop() + return session, nil } -// Poisoned reports whether this session has hit a transport-fatal error and -// should be discarded rather than reused. It is safe to call concurrently. +// Poisoned reports whether this session must be discarded after a transport +// failure or expired response recovery. It is safe to call concurrently. func (session *Session) Poisoned() bool { - session.mu.Lock() - defer session.mu.Unlock() - return session.poisoned || session.closed + return session.poisoned.Load() || session.lifetime.Err() != nil } +func (session *Session) lock(ctx context.Context) error { + select { + case session.gate <- struct{}{}: + return nil + case <-ctx.Done(): + return commandContextError(ctx.Err()) + case <-session.lifetime.Done(): + return ErrSessionClosed + } +} + +func (session *Session) unlock() { <-session.gate } + func (session *Session) Execute(ctx context.Context, command string) (Response, error) { command, err := normalizeATCommand(command) if err != nil { @@ -92,22 +128,14 @@ func (session *Session) Execute(ctx context.Context, command string) (Response, ctx, cancel := session.commandContext(ctx) defer cancel() - session.mu.Lock() - defer session.mu.Unlock() - if session.closed { - return Response{}, ErrSessionClosed + if err := session.lock(ctx); err != nil { + return Response{Command: command}, err } - return session.executeLocked(ctx, command) + defer session.unlock() + return session.executeLocked(ctx, command, nil, false) } -// ExecutePrompt executes the controlled two-phase AT+CMGS transaction. It does -// not release the session mutex between the command, the '>' prompt, the -// payload terminator, and the final result. -func (session *Session) ExecutePrompt( - ctx context.Context, - command string, - payload []byte, -) (Response, error) { +func (session *Session) ExecutePrompt(ctx context.Context, command string, payload []byte) (Response, error) { command, err := normalizeATCommand(command) if err != nil { return Response{}, err @@ -124,310 +152,341 @@ func (session *Session) ExecutePrompt( ctx, cancel := session.commandContext(ctx) defer cancel() - session.mu.Lock() - defer session.mu.Unlock() - if session.closed { - return Response{}, ErrSessionClosed + if err := session.lock(ctx); err != nil { + return Response{Command: command}, err } - return session.executePromptLocked(ctx, command, payload) + defer session.unlock() + return session.executeLocked(ctx, command, append([]byte(nil), payload...), true) } func (session *Session) commandContext(ctx context.Context) (context.Context, context.CancelFunc) { if ctx == nil { ctx = context.Background() } - if _, ok := ctx.Deadline(); ok || session.options.CommandTimeout <= 0 { + if _, ok := ctx.Deadline(); ok { return context.WithCancel(ctx) } return context.WithTimeout(ctx, session.options.CommandTimeout) } -func (session *Session) executeLocked(ctx context.Context, command string) (Response, error) { - started := time.Now() - response := Response{Command: command} - if err := ctx.Err(); err != nil { - return response, err - } - // Drain the transport before writing the command. Serial transports wait - // for any pending output here (a no-op after a synchronous command), while - // WWAN transports discard bytes left over from a previous command that - // timed out; without this, a late reply (e.g. a slow CGSN response) would - // be mis-parsed as this command's output. - if err := drainTransport(ctx, session.transport); err != nil { - session.poisonLocked() - return response, fmt.Errorf("drain %s: %w", command, err) - } - if err := writeAll(session.transport, []byte(command+"\r")); err != nil { - session.poisonLocked() - return response, fmt.Errorf("write %s: %w", command, err) +func commandContextError(err error) error { + if errors.Is(err, context.DeadlineExceeded) { + return errors.Join(ErrCommandTimeout, err) } - return session.readFinalLocked(ctx, started, command, "", response) -} - -// poisonLocked marks the session unusable after a transport-fatal error. Held -// under session.mu by the caller; idempotent. -func (session *Session) poisonLocked() { - session.poisoned = true + return err } -func (session *Session) executePromptLocked( - ctx context.Context, - command string, - payload []byte, -) (Response, error) { - started := time.Now() - response := Response{Command: command} +func (session *Session) executeLocked(ctx context.Context, command string, payload []byte, interactive bool) (response Response, err error) { + response.Command = command + if session.Poisoned() { + return response, ErrSessionClosed + } if err := ctx.Err(); err != nil { - return response, err + return response, commandContextError(err) } - if err := writeAll(session.transport, []byte(command+"\r")); err != nil { - session.poisonLocked() - return response, fmt.Errorf("write %s: %w", command, err) + if err := session.waitIOLocked(ctx); err != nil { + return response, err } - if err := drainTransport(ctx, session.transport); err != nil { - session.poisonLocked() + if err := session.runIOLocked(ctx, func() error { return drainTransport(session.lifetime, session.transport) }); err != nil { return response, fmt.Errorf("drain %s: %w", command, err) } - if err := session.waitPromptLocked(ctx, command, &response); err != nil { - response.Duration = time.Since(started) - return response, session.normalizeReadError(command, err) + // Never assign an old reply (including idle input and partial trailing + // lines) to a new command. Flushing cannot isolate bytes still in flight. + for { + session.receivePendingLocked() + if session.pending == nil { + session.readBuf = bytes.TrimLeft(session.readBuf, "\r\n\x00 \t") + if len(session.readBuf) == 0 && session.readErr == nil { + session.recoveryDeadline = time.Time{} + break + } + } + if err := session.readNextLocked(ctx, nil, nil); err != nil { + return response, commandContextError(err) + } } if err := ctx.Err(); err != nil { - session.abortPromptLocked() - response.Duration = time.Since(started) - return response, err + return response, commandContextError(err) + } + started := time.Now() + defer func() { response.Duration = time.Since(started) }() + transaction := &atTransaction{ + command: command, responsePrefix: expectedResponsePrefix(command), awaitPrompt: interactive, + } + session.pending = transaction + // Cancellation does not stop the modem. Preserve the caller's deadline and + // allow known long commands to finish even when it was shorter than usual. + deadline, _ := ctx.Deadline() + minimum := max(session.options.CommandTimeout, deadline.Sub(started)) + if interactive { + minimum = max(minimum, 125*time.Second) + } else if strings.EqualFold(command, "AT+COPS=?") { + minimum = max(minimum, 150*time.Second) + } + session.recoveryDeadline = started.Add(minimum + 5*session.options.CommandTimeout) + // Cancellation releases the caller, not a partially written command. + if err := session.runIOLocked(ctx, func() error { + if err := writeAll(session.transport, []byte(command+"\r")); err != nil { + return err + } + if interactive { + return drainTransport(session.lifetime, session.transport) + } + return nil + }); err != nil { + return response, fmt.Errorf("write %s: %w", command, err) } - if err := writeAll(session.transport, payload); err != nil { - session.poisonLocked() - session.abortPromptLocked() - response.Duration = time.Since(started) - return response, fmt.Errorf("write %s payload: %w", command, err) + for session.pending != nil { + if err := session.readNextLocked(ctx, &response, payload); err != nil { + return response, commandContextError(err) + } } - if err := writeAll(session.transport, []byte{0x1a}); err != nil { - session.poisonLocked() - session.abortPromptLocked() - response.Duration = time.Since(started) - return response, fmt.Errorf("terminate %s payload: %w", command, err) + if !response.OK() { + return response, &CommandError{ + Command: command, + Final: response.Final, + Lines: append([]string(nil), response.Lines...), + } } - if err := drainTransport(ctx, session.transport); err != nil { - session.poisonLocked() - response.Duration = time.Since(started) - return response, fmt.Errorf("drain %s payload: %w", command, err) + if transaction.awaitPrompt { + return response, fmt.Errorf("%w: %s", ErrPromptNotReceived, command) } - return session.readFinalLocked(ctx, started, command, string(payload), response) + return response, nil } -// drainTransport retries tcdrain/TCSBRK when the kernel interrupts it with a -// signal. go.bug.st/serial already retries EINTR for Read, but its Linux -// Drain implementation currently returns the transient error directly. -func drainTransport(ctx context.Context, transport Transport) error { - for { - err := transport.Drain() - if !errors.Is(err, syscall.EINTR) { - return err +func (session *Session) readLoop() { + defer close(session.readDone) + buffer := make([]byte, 1024) + for session.lifetime.Err() == nil { + count, err := session.transport.Read(buffer) + if count == 0 && (err == nil || errors.Is(err, syscall.EINTR)) { + continue } - if err := ctx.Err(); err != nil { - return err + batch := atRead{data: append([]byte(nil), buffer[:max(count, 0)]...), err: err} + if err != nil { + session.poisoned.Store(true) + } + select { + case session.received <- batch: + case <-session.lifetime.Done(): + return + } + if err != nil { + return } } } -func (session *Session) readFinalLocked( - ctx context.Context, - started time.Time, - command string, - payloadEcho string, - response Response, -) (Response, error) { - expectedPrefix := expectedResponsePrefix(command) +func (session *Session) receivePendingLocked() { for { - line, err := session.readLineLocked(ctx) - if err != nil { - response.Duration = time.Since(started) - return response, session.normalizeReadError(command, err) + select { + case batch := <-session.received: + session.readBuf = append(session.readBuf, batch.data...) + session.readErr = batch.err + default: + return } - line = strings.TrimSpace(strings.Trim(line, "\x00")) - if line == "" || strings.EqualFold(line, command) || - (payloadEcho != "" && line == payloadEcho) { - continue - } - if isFinalResult(line) { - response.Final = line - response.Duration = time.Since(started) - if response.OK() { - return response, nil - } - return response, &CommandError{ - Command: command, - Final: line, - Lines: append([]string(nil), response.Lines...), - } + } +} + +// Only one I/O task may run; even after cancellation, later operations wait +// for it before accessing the transaction or issuing another write. +func (session *Session) runIOLocked(ctx context.Context, operation func() error) error { + done := make(chan error, 1) + session.ioDone = done + session.ioDeadline = time.Now().Add(5 * session.options.CommandTimeout) + if session.recoveryDeadline.After(session.ioDeadline) { + session.ioDeadline = session.recoveryDeadline + } + go func() { + err := operation() + if err != nil { + session.poisoned.Store(true) } - if isURC(line) && !strings.HasPrefix(strings.ToUpper(line), expectedPrefix) { - response.URCs = append(response.URCs, line) - session.enqueueURCLocked(line) - continue + done <- err + }() + return session.waitIOLocked(ctx) +} + +func (session *Session) waitIOLocked(ctx context.Context) error { + if session.ioDone == nil { + return nil + } + timer := time.NewTimer(time.Until(session.ioDeadline)) + defer timer.Stop() + select { + case err := <-session.ioDone: + session.ioDone = nil + return err + case <-ctx.Done(): + return commandContextError(ctx.Err()) + case <-timer.C: + select { + case err := <-session.ioDone: + session.ioDone = nil + return err + default: + session.poisoned.Store(true) + return fmt.Errorf("%w: transport I/O never completed", ErrSessionClosed) } - response.Lines = append(response.Lines, line) + case <-session.lifetime.Done(): + return ErrSessionClosed } } -func (session *Session) waitPromptLocked( - ctx context.Context, - command string, - response *Response, -) error { - expectedPrefix := expectedResponsePrefix(command) +// A nil response means the caller of the pending transaction has already left. +// Its remaining lines are discarded, but URCs and the final still get processed. +func (session *Session) readNextLocked(ctx context.Context, response *Response, payload []byte) error { + if err := session.waitIOLocked(ctx); err != nil { + return err + } + var recovery <-chan time.Time for { - if index := promptIndex(session.readBuf); index >= 0 { - prefix := string(session.readBuf[:index]) - session.readBuf = session.readBuf[index+1:] - for len(session.readBuf) > 0 && - (session.readBuf[0] == ' ' || session.readBuf[0] == '\t') { + if recovery != nil && !time.Now().Before(session.recoveryDeadline) { + session.poisoned.Store(true) + return fmt.Errorf("%w: stale response never completed", ErrSessionClosed) + } + if err := ctx.Err(); err != nil { + return err + } + if session.skipSpace && len(session.readBuf) > 0 { + session.skipSpace = false + if session.readBuf[0] == ' ' { session.readBuf = session.readBuf[1:] + continue } - for _, line := range strings.FieldsFunc(prefix, func(character rune) bool { - return character == '\r' || character == '\n' - }) { - if err := session.consumePromptLineLocked( - command, - expectedPrefix, - line, - response, - ); err != nil { - return err - } - } - return nil + } + if session.pending != nil && session.pending.awaitPrompt && len(session.readBuf) > 0 && session.readBuf[0] == '>' { + session.readBuf = session.readBuf[1:] + session.skipSpace = true + return session.sendPayloadLocked(ctx, response != nil, payload) } if line, ok := popLine(&session.readBuf); ok { - if err := session.consumePromptLineLocked( - command, - expectedPrefix, - line, - response, - ); err != nil { - return err + if session.pending == nil { + session.recoveryDeadline = time.Time{} } - continue - } - if err := ctx.Err(); err != nil { - return err + session.consumeLineLocked(line, response) + return nil } - buffer := make([]byte, 1024) - count, err := readTransportContext(ctx, session.transport, buffer) - if count > 0 { - session.readBuf = append(session.readBuf, buffer[:count]...) - continue + if session.readErr != nil { + session.poisoned.Store(true) + return fmt.Errorf("read serial response: %w", session.readErr) } - if err != nil { - if errors.Is(err, io.EOF) && session.closed { - return ErrSessionClosed + if response == nil && recovery == nil && (session.pending != nil || len(bytes.Trim(session.readBuf, "\r\n\x00 \t")) != 0) { + if session.recoveryDeadline.IsZero() { + session.recoveryDeadline = time.Now().Add(5 * session.options.CommandTimeout) } - session.poisonLocked() - return fmt.Errorf("read serial prompt: %w", err) + timer := time.NewTimer(time.Until(session.recoveryDeadline)) + defer timer.Stop() + recovery = timer.C + } + select { + case batch := <-session.received: + session.readBuf = append(session.readBuf, batch.data...) + session.readErr = batch.err + case <-ctx.Done(): + continue + case <-recovery: + continue + case <-session.lifetime.Done(): + return ErrSessionClosed } } } -func promptIndex(buffer []byte) int { - for index, character := range buffer { - if character != '>' { - continue +func (session *Session) sendPayloadLocked(ctx context.Context, active bool, payload []byte) error { + if !active || ctx.Err() != nil { + session.recoveryDeadline = time.Now().Add(5 * session.options.CommandTimeout) + } + return session.runIOLocked(ctx, func() error { + terminator := byte(0x1b) // ESC aborts a prompt whose caller has left. + if active && ctx.Err() == nil { + if session.pending.commandEcho { + session.pending.remainingEcho = strings.FieldsFunc(string(payload), func(r rune) bool { return r == '\r' || r == '\n' }) + } + if err := writeAll(session.transport, payload); err != nil { + return fmt.Errorf("write SMS payload: %w", err) + } + if ctx.Err() == nil { + terminator = 0x1a + } } - if index == 0 || buffer[index-1] == '\r' || buffer[index-1] == '\n' { - return index + if err := writeAll(session.transport, []byte{terminator}); err != nil { + return fmt.Errorf("terminate SMS payload: %w", err) } - } - return -1 + session.pending.awaitPrompt = false + return drainTransport(session.lifetime, session.transport) + }) } -func (session *Session) consumePromptLineLocked( - command string, - expectedPrefix string, - line string, - response *Response, -) error { +func (session *Session) consumeLineLocked(line string, response *Response) { + pending := session.pending + if pending != nil && len(pending.remainingEcho) > 0 && line == pending.remainingEcho[0] { + pending.remainingEcho = pending.remainingEcho[1:] + return + } line = strings.TrimSpace(strings.Trim(line, "\x00")) - if line == "" || strings.EqualFold(line, command) { - return nil + if line == "" { + return } - if isFinalResult(line) { - response.Final = line - if response.OK() { - return fmt.Errorf("%w: %s", ErrPromptNotReceived, command) - } - return &CommandError{ - Command: command, - Final: line, - Lines: append([]string(nil), response.Lines...), + if pending == nil { + if !isFinalResult(line) { + session.enqueueURCLocked(line) } + return } - if isURC(line) && !strings.HasPrefix(strings.ToUpper(line), expectedPrefix) { - response.URCs = append(response.URCs, line) + if strings.EqualFold(line, pending.command) { + pending.commandEcho = true + return + } + callResult := isCallResult(line) + if callResult || (isURC(line) && !strings.HasPrefix(strings.ToUpper(line), pending.responsePrefix)) { session.enqueueURCLocked(line) - return nil + if response != nil { + response.URCs = append(response.URCs, line) + } + if !callResult || !isCallCommand(pending.command) { + return + } } - response.Lines = append(response.Lines, line) - return nil -} - -func (session *Session) normalizeReadError(command string, err error) error { - if errors.Is(err, context.DeadlineExceeded) || - errors.Is(err, context.Canceled) { - _ = session.transport.ResetInputBuffer() - session.readBuf = nil - if errors.Is(err, context.DeadlineExceeded) { - return fmt.Errorf("%w: %s", ErrCommandTimeout, command) + pending.remainingEcho = nil + if isFinalResult(line) || callResult { + session.pending = nil + session.recoveryDeadline = time.Time{} + if response != nil { + response.Final = line } + } else if response != nil { + response.Lines = append(response.Lines, line) } - return err -} - -func (session *Session) abortPromptLocked() { - _ = writeAll(session.transport, []byte{0x1b}) - _ = session.transport.Drain() - _ = session.transport.ResetInputBuffer() - session.readBuf = nil } // WaitURC waits for an unsolicited result matching predicate. Non-matching URCs -// remain queued for another consumer. -func (session *Session) WaitURC( - ctx context.Context, - predicate func(string) bool, -) (string, error) { +// remain queued for another consumer. It shares Execute's transaction lock. +func (session *Session) WaitURC(ctx context.Context, predicate func(string) bool) (string, error) { if predicate == nil { return "", errors.New("modem: URC predicate is required") } if ctx == nil { ctx = context.Background() } - session.mu.Lock() - defer session.mu.Unlock() - if session.closed { - return "", ErrSessionClosed + if err := session.lock(ctx); err != nil { + return "", err } - - for index, line := range session.urcs { - if predicate(line) { - session.urcs = append(session.urcs[:index], session.urcs[index+1:]...) - return line, nil - } + defer session.unlock() + if session.Poisoned() { + return "", ErrSessionClosed } for { - line, err := session.readLineLocked(ctx) - if err != nil { - return "", err - } - line = strings.TrimSpace(strings.Trim(line, "\x00")) - if line == "" { - continue + for index, line := range session.urcs { + if predicate(line) { + session.urcs = append(session.urcs[:index], session.urcs[index+1:]...) + return line, nil + } } - if predicate(line) { - return line, nil + if err := session.readNextLocked(ctx, nil, nil); err != nil { + return "", err } - session.enqueueURCLocked(line) } } @@ -440,51 +499,19 @@ func (session *Session) enqueueURCLocked(line string) { session.urcs = append(session.urcs, line) } -func (session *Session) readLineLocked(ctx context.Context) (string, error) { +// drainTransport retries tcdrain/TCSBRK when interrupted by a signal. +func drainTransport(ctx context.Context, transport Transport) error { for { - if line, ok := popLine(&session.readBuf); ok { - return line, nil + err := transport.Drain() + if !errors.Is(err, syscall.EINTR) { + return err } if err := ctx.Err(); err != nil { - return "", err - } - buffer := make([]byte, 1024) - count, err := readTransportContext(ctx, session.transport, buffer) - if count > 0 { - session.readBuf = append(session.readBuf, buffer[:count]...) - continue - } - if err != nil { - if errors.Is(err, io.EOF) && session.closed { - return "", ErrSessionClosed - } - session.poisonLocked() - return "", fmt.Errorf("read serial response: %w", err) + return err } } } -// readTransportContext checks cancellation around one bounded transport read. -// It must not close a transport from another goroutine: some serial libraries -// wait in Close for the active reader while that reader is itself stuck in a -// blocking read(2), deadlocking the session and every caller queued behind it. -// Linux serial transports therefore keep their fd non-blocking and enforce the -// configured read timeout with poll(2). -func readTransportContext( - ctx context.Context, - transport Transport, - buffer []byte, -) (int, error) { - if err := ctx.Err(); err != nil { - return 0, err - } - count, err := transport.Read(buffer) - if contextErr := ctx.Err(); contextErr != nil { - return 0, contextErr - } - return count, err -} - func popLine(buffer *[]byte) (string, bool) { data := *buffer for index, character := range data { @@ -543,13 +570,24 @@ func expectedResponsePrefix(command string) string { return name + ":" } +func isCallCommand(command string) bool { + command = strings.ToUpper(strings.TrimSpace(command)) + return command == "ATA" || command == "ATA0" || strings.HasPrefix(command, "ATD") +} + +func isCallResult(line string) bool { + switch strings.ToUpper(strings.TrimSpace(line)) { + case "NO CARRIER", "BUSY", "NO ANSWER": + return true + default: + return false + } +} + func isFinalResult(line string) bool { upper := strings.ToUpper(strings.TrimSpace(line)) return upper == "OK" || upper == "ERROR" || - upper == "NO CARRIER" || - upper == "BUSY" || - upper == "NO ANSWER" || strings.HasPrefix(upper, "+CME ERROR:") || strings.HasPrefix(upper, "+CMS ERROR:") } @@ -573,6 +611,8 @@ func isURC(line string) bool { "+CUSD:", "+CLIP:", "+CRING:", + "CIEV:", + "+CIEV:", "+QIND:", "+QIURC:", "+QSIMSTAT:", @@ -589,26 +629,34 @@ func isURC(line string) bool { func writeAll(writer io.Writer, payload []byte) error { for len(payload) > 0 { count, err := writer.Write(payload) + if count < 0 || count > len(payload) { + return io.ErrShortWrite + } + payload = payload[count:] + if errors.Is(err, syscall.EINTR) { + continue + } if err != nil { return err } - if count <= 0 { + if count == 0 { return io.ErrShortWrite } - if count > len(payload) { - return io.ErrShortWrite - } - payload = payload[count:] } return nil } func (session *Session) Close() error { - session.mu.Lock() - defer session.mu.Unlock() - if session.closed { - return nil - } - session.closed = true - return session.transport.Close() + session.closeOnce.Do(func() { + session.cancel() + session.closeErr = session.transport.Close() + }) + <-session.readDone + session.gate <- struct{}{} + defer session.unlock() + if session.ioDone != nil { + <-session.ioDone + session.ioDone = nil + } + return session.closeErr } diff --git a/internal/modem/session_test.go b/internal/modem/session_test.go index 04b91949..751a356c 100644 --- a/internal/modem/session_test.go +++ b/internal/modem/session_test.go @@ -5,55 +5,43 @@ import ( "errors" "fmt" "io" + "strings" "sync" "syscall" "testing" + "testing/synctest" "time" ) type transportStep struct { write string chunks []string + result func() (int, error) } type transcriptTransport struct { - mu sync.Mutex - steps []transportStep - chunks [][]byte - pendingWrite string - pendingChunks []string - readTimeout time.Duration - resetCount int - closed bool - unexpected error - writePartial bool - writeEvents chan string - drainErrors []error - drainCount int + mu sync.Mutex + steps []transportStep + chunks [][]byte + readReady chan struct{} + closed bool + unexpected error + writeEvents chan string + drainErrors []error + ops []string + onRead func() + onClose func() + readErr error } func (transport *transcriptTransport) Write(payload []byte) (int, error) { transport.mu.Lock() defer transport.mu.Unlock() + defer transport.wakeReaderLocked() if transport.closed { return 0, io.ErrClosedPipe } - if transport.pendingWrite != "" { - if string(payload) != transport.pendingWrite { - transport.unexpected = fmt.Errorf( - "partial write %q, want %q", - payload, - transport.pendingWrite, - ) - return 0, transport.unexpected - } - for _, chunk := range transport.pendingChunks { - transport.chunks = append(transport.chunks, []byte(chunk)) - } - transport.pendingWrite = "" - transport.pendingChunks = nil - return len(payload), nil - } + transport.ops = append(transport.ops, "write:"+string(payload)) if len(transport.steps) == 0 { transport.unexpected = fmt.Errorf("unexpected write %q", payload) return 0, transport.unexpected @@ -65,62 +53,70 @@ func (transport *transcriptTransport) Write(payload []byte) (int, error) { return 0, transport.unexpected } if transport.writeEvents != nil { - select { - case transport.writeEvents <- string(payload): - default: - } - } - if transport.writePartial && len(payload) > 1 { - transport.writePartial = false - count := len(payload) / 2 - transport.pendingWrite = step.write[count:] - transport.pendingChunks = append([]string(nil), step.chunks...) - return count, nil + transport.writeEvents <- string(payload) } for _, chunk := range step.chunks { transport.chunks = append(transport.chunks, []byte(chunk)) } + if step.result != nil { + transport.mu.Unlock() + count, err := step.result() + transport.mu.Lock() + return count, err + } return len(payload), nil } func (transport *transcriptTransport) enqueue(chunks ...string) { transport.mu.Lock() + defer transport.mu.Unlock() + defer transport.wakeReaderLocked() for _, chunk := range chunks { transport.chunks = append(transport.chunks, []byte(chunk)) } - transport.mu.Unlock() } func (transport *transcriptTransport) Read(buffer []byte) (int, error) { transport.mu.Lock() + defer transport.mu.Unlock() if transport.closed { - transport.mu.Unlock() return 0, io.EOF } if len(transport.chunks) > 0 { - chunk := transport.chunks[0] - count := copy(buffer, chunk) - if count == len(chunk) { + if transport.onRead != nil { + transport.onRead() + } + count := copy(buffer, transport.chunks[0]) + transport.chunks[0] = transport.chunks[0][count:] + if len(transport.chunks[0]) == 0 { transport.chunks = transport.chunks[1:] - } else { - transport.chunks[0] = chunk[count:] } - transport.mu.Unlock() return count, nil } - timeout := transport.readTimeout - transport.mu.Unlock() - if timeout <= 0 || timeout > 2*time.Millisecond { - timeout = time.Millisecond + if transport.readErr != nil { + return 0, transport.readErr } - time.Sleep(timeout) + if transport.readReady == nil { + transport.readReady = make(chan struct{}) + } + ready := transport.readReady + transport.mu.Unlock() + <-ready + transport.mu.Lock() return 0, nil } +func (transport *transcriptTransport) wakeReaderLocked() { + if transport.readReady != nil { + close(transport.readReady) + transport.readReady = nil + } +} + func (transport *transcriptTransport) Drain() error { transport.mu.Lock() defer transport.mu.Unlock() - transport.drainCount++ + transport.ops = append(transport.ops, "drain") if len(transport.drainErrors) == 0 { return nil } @@ -131,392 +127,572 @@ func (transport *transcriptTransport) Drain() error { func (transport *transcriptTransport) ResetInputBuffer() error { transport.mu.Lock() - transport.chunks = nil - transport.resetCount++ - transport.mu.Unlock() - return nil + defer transport.mu.Unlock() + transport.unexpected = errors.New("a running session must not discard input") + return transport.unexpected } -func (transport *transcriptTransport) SetReadTimeout(timeout time.Duration) error { - transport.mu.Lock() - transport.readTimeout = timeout - transport.mu.Unlock() - return nil -} +func (transport *transcriptTransport) SetReadTimeout(time.Duration) error { return nil } func (transport *transcriptTransport) Close() error { transport.mu.Lock() + defer transport.mu.Unlock() + defer transport.wakeReaderLocked() transport.closed = true - transport.mu.Unlock() - return nil -} - -func TestSessionRetriesInterruptedDrain(t *testing.T) { - transport := &transcriptTransport{ - steps: []transportStep{{ - write: "AT+CSQ\r", - chunks: []string{"\r\nAT+CSQ\r\n+CSQ: 24,99\r\nOK\r\n"}, - }}, - drainErrors: []error{syscall.EINTR}, - } - session, err := NewSession(transport, SessionOptions{}) - if err != nil { - t.Fatalf("NewSession() error = %v", err) - } - - response, err := session.Execute(context.Background(), "AT+CSQ") - if err != nil { - t.Fatalf("Execute() error = %v", err) - } - if response.Final != "OK" { - t.Fatalf("response final = %q", response.Final) - } - if transport.drainCount != 2 { - t.Fatalf("Drain() calls = %d, want 2", transport.drainCount) - } -} - -func TestSessionSeparatesInterleavedURCs(t *testing.T) { - transport := &transcriptTransport{steps: []transportStep{{ - write: "AT+CSQ\r", - chunks: []string{ - "\r\nAT+CSQ\r\n+CMTI: \"SM\",7\r\n", - "+CSQ: 24,99\r\nOK\r\n", - }, - }}} - session := newTestSession(t, transport) - response, err := session.Execute(context.Background(), "AT+CSQ") - if err != nil { - t.Fatalf("Execute: %v", err) - } - if got := response.Text(); got != "+CSQ: 24,99" { - t.Fatalf("response = %q", got) - } - if len(response.URCs) != 1 || response.URCs[0] != `+CMTI: "SM",7` { - t.Fatalf("URCs = %#v", response.URCs) - } - urc, err := session.WaitURC(context.Background(), func(line string) bool { - return line == `+CMTI: "SM",7` - }) - if err != nil || urc == "" { - t.Fatalf("WaitURC = %q, %v", urc, err) - } -} - -func TestSessionKeepsExpectedRegistrationLineInResponse(t *testing.T) { - transport := &transcriptTransport{steps: []transportStep{{ - write: "AT+CEREG?\r", - chunks: []string{"\r\n+CEREG: 0,5\r\nOK\r\n"}, - }}} - session := newTestSession(t, transport) - response, err := session.Execute(context.Background(), "AT+CEREG?") - if err != nil { - t.Fatalf("Execute: %v", err) - } - if response.Text() != "+CEREG: 0,5" || len(response.URCs) != 0 { - t.Fatalf("response = %#v", response) + if transport.onClose != nil { + transport.onClose() } + return nil } -func TestSessionQueuesCUSDThatArrivesBeforeOK(t *testing.T) { - transport := &transcriptTransport{steps: []transportStep{{ - write: "AT+CUSD=1,\"*100#\",15\r", - chunks: []string{"\r\n+CUSD: 0,\"004F004B\",72\r\nOK\r\n"}, - }}} - session := newTestSession(t, transport) - response, err := session.Execute(context.Background(), `AT+CUSD=1,"*100#",15`) - if err != nil { - t.Fatalf("Execute: %v", err) - } - if len(response.URCs) != 1 { - t.Fatalf("URCs = %#v", response.URCs) - } - urc, err := session.WaitURC(context.Background(), func(line string) bool { - return len(line) >= 6 && line[:6] == "+CUSD:" - }) - if err != nil || urc != `+CUSD: 0,"004F004B",72` { - t.Fatalf("WaitURC = %q, %v", urc, err) +func TestSessionWritesCommand(t *testing.T) { + final := []string{"\r\n+CSQ: 24,99\r\nOK\r\n"} + for _, test := range []struct { + name string + steps []transportStep + drainErrors []error + wantOps string + }{ + {name: "drain before write", steps: []transportStep{{write: "AT+CSQ\r", chunks: final}}, + wantOps: "drain|write:AT+CSQ\r"}, + {name: "interrupted drain", steps: []transportStep{{write: "AT+CSQ\r", chunks: final}}, drainErrors: []error{syscall.EINTR}, + wantOps: "drain|drain|write:AT+CSQ\r"}, + {name: "partial and interrupted writes", steps: []transportStep{ + {write: "AT+CSQ\r", result: func() (int, error) { return 2, nil }}, + {write: "+CSQ\r", result: func() (int, error) { return 1, syscall.EINTR }}, + {write: "CSQ\r", chunks: final}, + }, wantOps: "drain|write:AT+CSQ\r|write:+CSQ\r|write:CSQ\r"}, + } { + t.Run(test.name, func(t *testing.T) { + transport := &transcriptTransport{steps: test.steps, drainErrors: test.drainErrors} + session := newTestSession(t, transport) + response, err := session.Execute(context.Background(), "AT+CSQ") + if err != nil || !response.OK() || response.Text() != "+CSQ: 24,99" { + t.Fatalf("Execute = %#v, %v", response, err) + } + if got := strings.Join(transport.ops, "|"); got != test.wantOps { + t.Fatalf("transport operations = %q, want %q", got, test.wantOps) + } + }) } } -func TestSessionReturnsTypedCommandError(t *testing.T) { - transport := &transcriptTransport{steps: []transportStep{{ - write: "AT+CPIN?\r", - chunks: []string{"\r\n+CME ERROR: 10\r\n"}, - }}} - session := newTestSession(t, transport) - _, err := session.Execute(context.Background(), "AT+CPIN?") - var commandErr *CommandError - if !errors.As(err, &commandErr) || commandErr.Final != "+CME ERROR: 10" { - t.Fatalf("error = %#v", err) - } -} - -func TestSessionTimeoutResetsInputAndRejectsCommandInjection(t *testing.T) { - transport := &transcriptTransport{steps: []transportStep{{write: "AT\r"}}} - session, err := NewSession(transport, SessionOptions{ - ReadTimeout: time.Millisecond, - CommandTimeout: 15 * time.Millisecond, - }) - if err != nil { - t.Fatal(err) - } - _, err = session.Execute(context.Background(), "AT") - if !errors.Is(err, ErrCommandTimeout) { - t.Fatalf("error = %v", err) - } - if transport.resetCount != 1 { - t.Fatalf("reset count = %d", transport.resetCount) - } - if _, err := session.Execute(context.Background(), "AT\rAT+CFUN=0"); err == nil { - t.Fatal("expected command delimiter rejection") - } -} - -func TestSessionHandlesPartialWrites(t *testing.T) { - transport := &transcriptTransport{ - writePartial: true, - steps: []transportStep{{ - write: "AT+CSQ\r", - chunks: []string{"\r\n+CSQ: 1,99\r\nOK\r\n"}, - }}, - } - session := newTestSession(t, transport) - response, err := session.Execute(context.Background(), "AT+CSQ") - if err != nil { - t.Fatalf("Execute: %v", err) - } - if response.Text() != "+CSQ: 1,99" { - t.Fatalf("response = %#v", response) - } - transport.mu.Lock() - defer transport.mu.Unlock() - if transport.pendingWrite != "" || len(transport.steps) != 0 || - transport.unexpected != nil { - t.Fatalf( - "unfinished transcript: pending=%q steps=%d err=%v", - transport.pendingWrite, - len(transport.steps), - transport.unexpected, - ) +func TestSessionParsesResponses(t *testing.T) { + for _, test := range []struct { + name, command, text, final, urc string + chunks []string + }{ + {name: "interleaved URC and split line", command: "AT+CSQ", text: "+CSQ: 24,99", final: "OK", urc: `+CMTI: "SM",7`, + chunks: []string{"\r\nAT+CSQ\r", "\n+CMTI: \"SM\",7\r\n+CS", "Q: 24,99\r\nOK\r", "\n"}}, + {name: "expected registration", command: "AT+CEREG?", text: "+CEREG: 0,5", final: "OK", chunks: []string{"\r\n+CEREG: 0,5\r\nOK\r\n"}}, + {name: "asynchronous CUSD", command: `AT+CUSD=1,"*100#",15`, final: "OK", urc: `+CUSD: 0,"004F004B",72`, chunks: []string{"\r\n+CUSD: 0,\"004F004B\",72\r\nOK\r\n"}}, + {name: "command error", command: "AT+CPIN?", final: "+CME ERROR: 10", chunks: []string{"\r\n+CME ERROR: 10\r\n"}}, + {name: "dial final", command: "ATD12345;", final: "NO CARRIER", urc: "NO CARRIER", chunks: []string{"NO CARRIER\r\n"}}, + {name: "answer final", command: "ATA", final: "BUSY", urc: "BUSY", chunks: []string{"BUSY\r\n"}}, + {name: "query during call", command: "AT+CFUN?", text: "+CFUN: 1", final: "OK", urc: "NO CARRIER", chunks: []string{"NO CARRIER\r\n+CFUN: 1\r\nOK\r\n"}}, + {name: "hangup waits for OK", command: "ATH", final: "OK", urc: "NO ANSWER", chunks: []string{"NO ANSWER\r\nOK\r\n"}}, + {name: "ML307 call indication", command: "ATI", text: "ML307A", final: "OK", urc: "CIEV: CALL 1", chunks: []string{"CIEV: CALL 1\r\nML307A\r\nOK\r\n"}}, + } { + t.Run(test.name, func(t *testing.T) { + transport := &transcriptTransport{steps: []transportStep{{write: test.command + "\r", chunks: test.chunks}}} + session := newTestSession(t, transport) + response, err := session.Execute(context.Background(), test.command) + if response.Text() != test.text || response.Final != test.final || session.Poisoned() { + t.Fatalf("Execute = %#v, %v", response, err) + } + var commandErr *CommandError + if test.final == "OK" { + if err != nil { + t.Fatal(err) + } + } else if !errors.As(err, &commandErr) || commandErr.Command != test.command || commandErr.Final != test.final { + t.Fatalf("command error = %v", err) + } + if test.urc == "" { + if len(response.URCs) != 0 { + t.Fatalf("unexpected URCs: %q", response.URCs) + } + } else { + if len(response.URCs) != 1 || response.URCs[0] != test.urc { + t.Fatalf("URCs = %q, want %q", response.URCs, test.urc) + } + urc, err := session.WaitURC(context.Background(), func(line string) bool { return line == test.urc }) + if err != nil || urc != test.urc { + t.Fatalf("WaitURC = %q, %v", urc, err) + } + } + }) } } func TestSessionExecutePromptQueuesURCsAndReturnsCMGS(t *testing.T) { const pdu = "00010005912143F50008044F60597D" transport := &transcriptTransport{steps: []transportStep{ - { - write: "AT+CMGS=14\r", - chunks: []string{ - "\r\nAT+CMGS=14\r\n+CMTI: \"SM\",7\r\n> ", - }, - }, + {write: "AT+CMGS=14\r", chunks: []string{"\r", "\nAT+CMGS=14\r\n+CMTI: \"SM\",7\r", "\n>", " "}}, {write: pdu}, - { - write: string([]byte{0x1a}), - chunks: []string{ - "\r\n" + pdu + "\r\n+CMGS: 42\r\n", - "+CMTI: \"SM\",8\r\nOK\r\n", - }, - }, + {write: "\x1a", chunks: []string{"\r\n" + pdu + "\r\n+CMGS: 42\r\n", "+CMTI: \"SM\",8\r\nOK\r\n"}}, }} session := newTestSession(t, transport) - response, err := session.ExecutePrompt( - context.Background(), - "AT+CMGS=14", - []byte(pdu), - ) - if err != nil { - t.Fatalf("ExecutePrompt: %v", err) - } - if response.Text() != "+CMGS: 42" || !response.OK() { - t.Fatalf("response = %#v", response) - } - if len(response.URCs) != 2 { - t.Fatalf("URCs = %#v", response.URCs) + response, err := session.ExecutePrompt(context.Background(), "AT+CMGS=14", []byte(pdu)) + if err != nil || response.Text() != "+CMGS: 42" || !response.OK() || len(response.URCs) != 2 { + t.Fatalf("ExecutePrompt = %#v, %v", response, err) } for _, wanted := range []string{`+CMTI: "SM",7`, `+CMTI: "SM",8`} { - line, waitErr := session.WaitURC( - context.Background(), - func(line string) bool { return line == wanted }, - ) - if waitErr != nil || line != wanted { - t.Fatalf("WaitURC(%q) = %q, %v", wanted, line, waitErr) + line, err := session.WaitURC(context.Background(), func(line string) bool { return line == wanted }) + if err != nil || line != wanted { + t.Fatalf("WaitURC(%q) = %q, %v", wanted, line, err) } } } -func TestSessionExecutePromptTimeoutDoesNotWritePayload(t *testing.T) { - transport := &transcriptTransport{ - steps: []transportStep{{write: "AT+CMGS=5\r"}}, - } - session, err := NewSession(transport, SessionOptions{ - ReadTimeout: time.Millisecond, - CommandTimeout: 15 * time.Millisecond, - }) - if err != nil { - t.Fatal(err) - } - _, err = session.ExecutePrompt( - context.Background(), - "AT+CMGS=5", - []byte("001122"), - ) - if !errors.Is(err, ErrCommandTimeout) { - t.Fatalf("error = %v", err) - } - transport.mu.Lock() - defer transport.mu.Unlock() - if transport.resetCount != 1 || len(transport.steps) != 0 || - transport.unexpected != nil { - t.Fatalf( - "transport = reset %d, steps %d, error %v", - transport.resetCount, - len(transport.steps), - transport.unexpected, - ) - } -} - func TestSessionExecutePromptSerializesConcurrentCommand(t *testing.T) { events := make(chan string, 4) - transport := &transcriptTransport{ - writeEvents: events, - steps: []transportStep{ - {write: "AT+CMGS=\"12345\"\r"}, - {write: "HELLO"}, - { - write: string([]byte{0x1a}), - chunks: []string{"\r\n+CMGS: 9\r\nOK\r\n"}, - }, - { - write: "AT+CSQ\r", - chunks: []string{"\r\n+CSQ: 20,99\r\nOK\r\n"}, - }, - }, - } + transport := &transcriptTransport{writeEvents: events, steps: []transportStep{ + {write: "AT+CMGS=5\r"}, + {write: "HELLO"}, + {write: "\x1a", chunks: []string{"\r\n+CMGS: 9\r\nOK\r\n"}}, + {write: "AT+CSQ\r", chunks: []string{"\r\n+CSQ: 20,99\r\nOK\r\n"}}, + }} session := newTestSession(t, transport) - promptResult := make(chan error, 1) + promptResult, normalResult := make(chan error, 1), make(chan error, 1) go func() { - _, err := session.ExecutePrompt( - context.Background(), - `AT+CMGS="12345"`, - []byte("HELLO"), - ) + _, err := session.ExecutePrompt(context.Background(), "AT+CMGS=5", []byte("HELLO")) promptResult <- err }() - if first := <-events; first != "AT+CMGS=\"12345\"\r" { + if first := <-events; first != "AT+CMGS=5\r" { t.Fatalf("first write = %q", first) } - - normalStarted := make(chan struct{}) - normalResult := make(chan error, 1) go func() { - close(normalStarted) _, err := session.Execute(context.Background(), "AT+CSQ") normalResult <- err }() - <-normalStarted transport.enqueue("\r\n> ") - - if err := <-promptResult; err != nil { - t.Fatalf("ExecutePrompt: %v", err) - } - if err := <-normalResult; err != nil { - t.Fatalf("concurrent Execute: %v", err) + for _, err := range []error{<-promptResult, <-normalResult} { + if err != nil { + t.Fatal(err) + } } - writes := []string{<-events, <-events, <-events} - want := []string{"HELLO", string([]byte{0x1a}), "AT+CSQ\r"} - for index := range want { - if writes[index] != want[index] { - t.Fatalf("write[%d] = %q, want %q", index, writes[index], want[index]) + for _, want := range []string{"HELLO", "\x1a", "AT+CSQ\r"} { + if got := <-events; got != want { + t.Fatalf("write = %q, want %q", got, want) } } } -func TestSessionExecutePromptRejectsUnsafeInput(t *testing.T) { - transport := &transcriptTransport{} - session := newTestSession(t, transport) - if _, err := session.ExecutePrompt( - context.Background(), - "AT+CSQ", - []byte("payload"), - ); err == nil { - t.Fatal("expected non-CMGS prompt command rejection") - } - if _, err := session.ExecutePrompt( - context.Background(), - "AT+CMGS=1", - []byte{'A', 0x1a}, - ); err == nil { - t.Fatal("expected Ctrl-Z payload rejection") +func TestSessionPayloadEchoDoesNotReplaceFinal(t *testing.T) { + for _, test := range []struct { + name, payload string + echo, fail bool + }{ + {name: "OK without echo", payload: "OK"}, + {name: "OK with echo", payload: "OK", echo: true}, + {name: "multiline echo", payload: "ERROR\nHELLO", echo: true}, + {name: "echo with whitespace", payload: " OK", echo: true}, + {name: "error after echo", payload: "OK", echo: true, fail: true}, + {name: "error without echo", payload: "ERROR", fail: true}, + } { + t.Run(test.name, func(t *testing.T) { + command, prompt := `AT+CMGS="12345"`, "\r\n> " + final, text := "OK", "+CMGS: 9" + if test.fail { + final, text = "ERROR", "" + } + reply := "\r\n" + text + "\r\n" + final + "\r\n" + if test.echo { + prompt = command + "\r\n> " + reply = strings.ReplaceAll(test.payload, "\n", "\r\n") + reply + } + transport := &transcriptTransport{steps: []transportStep{ + {write: command + "\r", chunks: []string{prompt}}, + {write: test.payload}, + {write: "\x1a", chunks: []string{reply}}, + }} + response, err := newTestSession(t, transport).ExecutePrompt(context.Background(), command, []byte(test.payload)) + if (err != nil) != test.fail || response.Final != final || response.Text() != text { + t.Fatalf("prompt response = %#v, %v", response, err) + } + }) } } -// drainOrderTransport forwards to an inner Transport while recording -// transport-level events, so a test can assert the exact order of Drain and -// Write calls. -type drainOrderTransport struct { - inner Transport - events chan string +func TestSessionIdleInputStaysUnsolicited(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + transport := &transcriptTransport{steps: []transportStep{{write: "AT+CEREG?\r", chunks: []string{"+CEREG: 0,1\r\nOK\r\n"}}}} + session := newTestSession(t, transport) + transport.enqueue("+CEREG: 5\r\n") + synctest.Wait() + response, err := session.Execute(context.Background(), "AT+CEREG?") + if err != nil || response.Text() != "+CEREG: 0,1" { + t.Fatalf("idle URC leaked into query: %#v, %v", response, err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + line, err := session.WaitURC(ctx, func(line string) bool { return line == "+CEREG: 5" }) + if err != nil || line != "+CEREG: 5" { + t.Fatalf("idle URC was lost: %q, %v", line, err) + } + if _, err := session.WaitURC(ctx, func(string) bool { return false }); !errors.Is(err, context.DeadlineExceeded) || session.Poisoned() { + t.Fatalf("idle wait = %v, poisoned = %t", err, session.Poisoned()) + } + }) } -func (transport *drainOrderTransport) Write(payload []byte) (int, error) { - transport.events <- "write:" + string(payload) - return transport.inner.Write(payload) +func TestSessionCancelsAndClosesBlockedWrite(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + started, release := make(chan struct{}), make(chan struct{}) + releaseWrite := sync.OnceFunc(func() { close(release) }) + transport := &transcriptTransport{onClose: releaseWrite, steps: []transportStep{ + {write: "AT+CSQ\r", result: func() (int, error) { return 2, nil }}, + {write: "+CSQ\r", result: func() (int, error) { close(started); <-release; return 0, io.ErrClosedPipe }}, + }} + session := newTestSession(t, transport) + t.Cleanup(releaseWrite) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { _, err := session.Execute(ctx, "AT+CSQ"); done <- err }() + <-started + cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("canceled write = %v", err) + } + case <-time.After(time.Second): + t.Fatal("cancellation waited for the blocked write") + } + waitStarted := time.Now() + time.Sleep(2 * time.Second) + retryCtx, cancelRetry := context.WithTimeout(context.Background(), 10*time.Second) + _, err := session.Execute(retryCtx, "AT") + cancelRetry() + if !errors.Is(err, ErrSessionClosed) || !session.Poisoned() || + time.Since(waitStarted) != 6*time.Second || session.ioDone == nil || session.pending == nil { + t.Fatalf("blocked I/O recovery = %v after %s, poisoned = %t", err, time.Since(waitStarted), session.Poisoned()) + } + go func() { done <- session.Close() }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(time.Second): + t.Fatal("Close did not interrupt the blocked write") + } + }) } -func (transport *drainOrderTransport) Read(buffer []byte) (int, error) { - return transport.inner.Read(buffer) +func TestSessionCompletedIOWinsExpiredDeadline(t *testing.T) { + session := newTestSession(t, &transcriptTransport{}) + session.ioDeadline = time.Now().Add(-time.Second) + for range 32 { + session.ioDone = make(chan error, 1) + session.ioDone <- nil + if err := session.waitIOLocked(context.Background()); err != nil || session.Poisoned() || session.ioDone != nil { + t.Fatalf("completed I/O = %v, poisoned = %t", err, session.Poisoned()) + } + } } -func (transport *drainOrderTransport) Drain() error { - transport.events <- "drain" - return transport.inner.Drain() +func TestSessionRejectsUnsafeInput(t *testing.T) { + session := newTestSession(t, &transcriptTransport{}) + if _, err := session.Execute(context.Background(), "AT\rAT+CFUN=0"); err == nil { + t.Fatal("expected command delimiter rejection") + } + if _, err := session.ExecutePrompt(context.Background(), "AT+CSQ", []byte("payload")); err == nil { + t.Fatal("expected non-CMGS prompt command rejection") + } + if _, err := session.ExecutePrompt(context.Background(), "AT+CMGS=1", []byte{'A', 0x1a}); err == nil { + t.Fatal("expected Ctrl-Z payload rejection") + } } -func (transport *drainOrderTransport) ResetInputBuffer() error { - return transport.inner.ResetInputBuffer() +func TestSessionExpiredTransactionKeepsWireOwnership(t *testing.T) { + for _, test := range []struct { + name string + prompt, submitted, payloadCancel bool + timeout, waitURC, echo bool + idle time.Duration + }{ + {name: "query canceled"}, + {name: "query timeout via WaitURC", timeout: true, waitURC: true}, + {name: "late prompt canceled", prompt: true}, + {name: "late prompt after long idle", prompt: true, idle: 3 * time.Minute}, + {name: "late prompt timeout via WaitURC", prompt: true, timeout: true, waitURC: true}, + {name: "payload canceled", prompt: true, payloadCancel: true}, + {name: "payload echo canceled", prompt: true, payloadCancel: true, echo: true}, + {name: "submitted SMS canceled", prompt: true, submitted: true}, + } { + t.Run(test.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + command := "AT+CFUN?" + if test.prompt { + command = "AT+CMGS=5" + } + first := transportStep{write: command + "\r"} + if test.submitted || test.payloadCancel { + first.chunks = []string{"\r\n> "} + } else if !test.timeout { + first.result = func() (int, error) { cancel(); return len(first.write), nil } + } + body := "HELLO" + if test.echo { + body = "ERROR\nHELLO" + first.chunks = []string{command + "\r\n> "} + } + transport := &transcriptTransport{} + steps := []transportStep{first} + if !test.prompt && !test.timeout { + steps[0].result = func() (int, error) { cancel(); return 2, nil } + steps = append(steps, transportStep{write: first.write[2:]}) + } + lateReply := []string{"\r\nlate reply\r\nOK\r\n+CFUN: STA", "LE\r\nOK\r\nRING\r\n"} + if test.submitted || test.payloadCancel { + payload := transportStep{write: body} + if test.echo { + payload.chunks = []string{"\r\n" + strings.ReplaceAll(body, "\n", "\r\n") + "\r\n"} + } + if test.payloadCancel { + payload.result = func() (int, error) { cancel(); return len(body), nil } + } + steps = append(steps, payload) + } + if test.submitted { + steps = append(steps, transportStep{write: "\x1a", result: func() (int, error) { cancel(); return 1, nil }}) + } else if test.prompt { + escape := transportStep{write: "\x1b"} + if !test.payloadCancel { + escape.result = func() (int, error) { + time.AfterFunc(time.Millisecond, func() { transport.enqueue(lateReply...) }) + return 1, nil + } + } + steps = append(steps, escape) + } + steps = append(steps, transportStep{write: "AT+CPIN?\r", chunks: []string{"\r\n+CPIN: READY\r\nOK\r\n"}}) + transport.steps = steps + session := newTestSession(t, transport) + var response Response + var err error + if test.prompt { + response, err = session.ExecutePrompt(ctx, command, []byte(body)) + } else { + response, err = session.Execute(ctx, command) + } + wantErr := context.Canceled + if test.timeout { + wantErr = ErrCommandTimeout + } + if !errors.Is(err, wantErr) || response.Final != "" || session.Poisoned() { + t.Fatalf("expired transaction = %#v, %v; poisoned = %t", response, err, session.Poisoned()) + } + blockedCtx, cancelBlocked := context.WithTimeout(context.Background(), time.Millisecond) + _, err = session.Execute(blockedCtx, "AT") + cancelBlocked() + if !errors.Is(err, ErrCommandTimeout) { + t.Fatalf("command before old final = %v", err) + } + if test.prompt && !test.submitted && !test.payloadCancel { + transport.enqueue("\r\n> ") + } else { + transport.enqueue(lateReply...) + } + time.Sleep(test.idle) + if test.waitURC { + urcCtx, cancelURC := context.WithTimeout(context.Background(), time.Second) + line, err := session.WaitURC(urcCtx, func(line string) bool { return line == "RING" }) + cancelURC() + if err != nil || line != "RING" { + t.Fatalf("WaitURC while draining old response = %q, %v", line, err) + } + } + response, err = session.Execute(context.Background(), "AT+CPIN?") + if err != nil || !response.OK() || response.Text() != "+CPIN: READY" { + t.Fatalf("query after late final = %#v, %v", response, err) + } + }) + }) + } } -func (transport *drainOrderTransport) SetReadTimeout(timeout time.Duration) error { - return transport.inner.SetReadTimeout(timeout) +func TestSessionBoundsStaleResponseRecovery(t *testing.T) { + for _, test := range []struct { + name, command, reply string + timeout, recovery time.Duration + waitURC bool + }{ + {"original deadline", "AT", "", 20 * time.Second, 25 * time.Second, false}, + {"scan with short caller", "AT+COPS=?", "", time.Second, 155 * time.Second, true}, + {"prompt with short caller", "AT+CMGS=5", "", time.Second, 130 * time.Second, false}, + {"unterminated tail", "AT", "OK\r\npartial", time.Second, 5 * time.Second, true}, + } { + t.Run(test.name, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + started := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), test.timeout) + defer cancel() + step := transportStep{write: test.command + "\r", chunks: []string{test.reply}} + if test.reply == "" { + step.result = func() (int, error) { cancel(); return len(step.write), nil } + } + session := newTestSession(t, &transcriptTransport{steps: []transportStep{step}}) + var err error + if strings.HasPrefix(test.command, "AT+CMGS=") { + _, err = session.ExecutePrompt(ctx, test.command, []byte("HELLO")) + } else { + _, err = session.Execute(ctx, test.command) + } + if test.reply == "" && !errors.Is(err, context.Canceled) || test.reply != "" && err != nil { + t.Fatalf("initial command = %v", err) + } + for range 2 { + retryCtx, cancelRetry := context.WithTimeout(context.Background(), time.Second) + _, err = session.Execute(retryCtx, "AT") + cancelRetry() + if !errors.Is(err, ErrCommandTimeout) || session.Poisoned() { + t.Fatalf("short retry = %v, poisoned = %t", err, session.Poisoned()) + } + } + recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), test.recovery) + defer cancelRecovery() + if test.waitURC { + _, err = session.WaitURC(recoveryCtx, func(string) bool { return false }) + } else { + _, err = session.Execute(recoveryCtx, "AT") + } + if !errors.Is(err, ErrSessionClosed) || !session.Poisoned() || time.Since(started) != test.recovery { + t.Fatalf("recovery = %v after %s, poisoned = %t; want %s", err, time.Since(started), session.Poisoned(), test.recovery) + } + }) + }) + } } -func (transport *drainOrderTransport) Close() error { - return transport.inner.Close() +func TestSessionKeepsBytesReadDuringCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + reads := 0 + transport := &transcriptTransport{onRead: func() { + reads++ + if reads == 2 { + cancel() + } + }, steps: []transportStep{ + {write: "AT+CFUN?\r", chunks: []string{"\r\n+CFUN: 1\r\n", "OK\r\n"}}, + {write: "AT+CPIN?\r", chunks: []string{"\r\n+CPIN: READY\r\nOK\r\n"}}, + }} + session := newTestSession(t, transport) + first, err := session.Execute(ctx, "AT+CFUN?") + // The reader may prefetch both chunks before the caller parses the first. + firstText := first.Text() + if !errors.Is(err, context.Canceled) || firstText != "" && firstText != "+CFUN: 1" || first.Final != "" { + t.Fatalf("canceled response = %#v, %v", first, err) + } + next, err := session.Execute(context.Background(), "AT+CPIN?") + if err != nil || !next.OK() || next.Text() != "+CPIN: READY" || first.Text() != firstText { + t.Fatalf("next response = %#v, %v; previous response = %#v", next, err, first) + } } -// WWAN transports discard stale bytes left over from a timed-out command -// inside Drain, so the session must call it before writing the next command; -// otherwise a late reply (e.g. a slow CGSN response) would be mis-parsed as -// the new command's output. -func TestSessionDrainsBeforeWritingCommand(t *testing.T) { - inner := &transcriptTransport{steps: []transportStep{{ - write: "AT+CSQ\r", - chunks: []string{"\r\n+CSQ: 24,99\r\nOK\r\n"}, - }}} - events := make(chan string, 8) - session := newTestSession(t, &drainOrderTransport{inner: inner, events: events}) - if _, err := session.Execute(context.Background(), "AT+CSQ"); err != nil { - t.Fatalf("Execute: %v", err) - } - if first := <-events; first != "drain" { - t.Fatalf("first transport event = %q, want drain before the command write", first) - } - if second := <-events; second != "write:AT+CSQ\r" { - t.Fatalf("second transport event = %q, want the command write", second) +func TestSessionConsumesBufferedTailBeforeNextCommand(t *testing.T) { + for _, tail := range []string{"\x00 \t", "+CFUN: STA"} { + t.Run(tail, func(t *testing.T) { + chunks := []string{"+CFUN: 1\r\nOK\r\n" + tail} + if tail == "+CFUN: STA" { + chunks = append(chunks, "LE\r\nOK\r\n") + } + transport := &transcriptTransport{steps: []transportStep{ + {write: "AT+CFUN?\r", chunks: chunks}, + {write: "AT+CPIN?\r", chunks: []string{"+CPIN: READY\r\nOK\r\n"}}, + }} + session := newTestSession(t, transport) + for _, query := range []struct{ command, text string }{ + {"AT+CFUN?", "+CFUN: 1"}, {"AT+CPIN?", "+CPIN: READY"}, + } { + response, err := session.Execute(context.Background(), query.command) + if err != nil || !response.OK() || response.Text() != query.text { + t.Fatalf("response with old tail = %#v, %v; want %q", response, err, query.text) + } + } + }) } } -func newTestSession(t *testing.T, transport Transport) *Session { - t.Helper() - session, err := NewSession(transport, SessionOptions{ - ReadTimeout: time.Millisecond, - CommandTimeout: time.Second, +func TestSessionPoisonsTransportFailures(t *testing.T) { + for _, operation := range []string{"read", "write", "drain", "payload"} { + t.Run(operation, func(t *testing.T) { + transport := &transcriptTransport{} + command := transportStep{write: "AT\r"} + switch operation { + case "drain": + transport.drainErrors = []error{syscall.EIO} + case "read": + command.result = func() (int, error) { + transport.mu.Lock() + transport.readErr = syscall.EIO + transport.mu.Unlock() + return len(command.write), nil + } + transport.steps = []transportStep{command} + case "write": + command.result = func() (int, error) { return 1, syscall.EIO } + transport.steps = []transportStep{command} + case "payload": + transport.steps = []transportStep{ + {write: "AT+CMGS=5\r", chunks: []string{"\r\n> "}}, + {write: "HELLO", result: func() (int, error) { return 2, syscall.EIO }}, + } + } + session := newTestSession(t, transport) + var err error + if operation == "payload" { + _, err = session.ExecutePrompt(context.Background(), "AT+CMGS=5", []byte("HELLO")) + } else { + _, err = session.Execute(context.Background(), "AT") + } + if !errors.Is(err, syscall.EIO) || !session.Poisoned() { + t.Fatalf("transport failure = %v; poisoned = %t", err, session.Poisoned()) + } + if _, err := session.Execute(context.Background(), "AT"); !errors.Is(err, ErrSessionClosed) { + t.Fatalf("command after transport failure = %v", err) + } + }) + } +} + +func TestSessionKeepsFinalBeforeEOF(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + transport := &transcriptTransport{} + transport.steps = []transportStep{{write: "AT\r", chunks: []string{"OK\r\n"}, result: func() (int, error) { + transport.mu.Lock() + transport.readErr = io.EOF + transport.mu.Unlock() + return 3, nil + }}} + session := newTestSession(t, transport) + response, err := session.Execute(context.Background(), "AT") + synctest.Wait() + if err != nil || !response.OK() || !session.Poisoned() { + t.Fatalf("final before EOF = %#v, %v; poisoned = %t", response, err, session.Poisoned()) + } }) +} + +func newTestSession(t *testing.T, transport *transcriptTransport) *Session { + t.Helper() + session, err := NewSession(transport, SessionOptions{ReadTimeout: time.Millisecond, CommandTimeout: time.Second}) if err != nil { t.Fatal(err) } + t.Cleanup(func() { + _ = session.Close() + transport.mu.Lock() + defer transport.mu.Unlock() + if transport.unexpected != nil || len(transport.steps) != 0 { + t.Errorf("unfinished transcript: steps=%d err=%v", len(transport.steps), transport.unexpected) + } + }) return session } diff --git a/internal/modem/wwan_at_linux.go b/internal/modem/wwan_at_linux.go index 3251b4fc..33abf8d8 100644 --- a/internal/modem/wwan_at_linux.go +++ b/internal/modem/wwan_at_linux.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "sync" + "sync/atomic" "time" "golang.org/x/sys/unix" @@ -19,7 +20,7 @@ type nativeWWANATTransport struct { mu sync.RWMutex fd int readTimeout time.Duration - closed bool + closed atomic.Bool } func openNativeWWANATTransport(path string) (Transport, error) { @@ -33,7 +34,7 @@ func openNativeWWANATTransport(path string) (Transport, error) { func (transport *nativeWWANATTransport) Read(buffer []byte) (int, error) { transport.mu.RLock() defer transport.mu.RUnlock() - if transport.closed { + if transport.closed.Load() { return 0, io.ErrClosedPipe } @@ -42,13 +43,16 @@ func (transport *nativeWWANATTransport) Read(buffer []byte) (int, error) { deadline = time.Now().Add(transport.readTimeout) } for { - timeout := -1 + if transport.closed.Load() { + return 0, io.ErrClosedPipe + } + timeout := 100 if !deadline.IsZero() { remaining := time.Until(deadline) if remaining <= 0 { return 0, nil } - timeout = int((remaining + time.Millisecond - 1) / time.Millisecond) + timeout = int((min(remaining, 100*time.Millisecond) + time.Millisecond - 1) / time.Millisecond) } fds := []unix.PollFd{{Fd: int32(transport.fd), Events: unix.POLLIN}} ready, err := unix.Poll(fds, timeout) @@ -58,8 +62,11 @@ func (transport *nativeWWANATTransport) Read(buffer []byte) (int, error) { if err != nil { return 0, err } + if transport.closed.Load() { + return 0, io.ErrClosedPipe + } if ready == 0 { - return 0, nil + continue } if fds[0].Revents&(unix.POLLERR|unix.POLLHUP|unix.POLLNVAL) != 0 && fds[0].Revents&unix.POLLIN == 0 { @@ -72,6 +79,9 @@ func (transport *nativeWWANATTransport) Read(buffer []byte) (int, error) { if count < 0 { count = 0 } + if len(buffer) > 0 && count == 0 && err == nil && fds[0].Revents&(unix.POLLHUP|unix.POLLERR) != 0 { + return 0, io.EOF + } return count, err } } @@ -79,17 +89,17 @@ func (transport *nativeWWANATTransport) Read(buffer []byte) (int, error) { func (transport *nativeWWANATTransport) Write(buffer []byte) (int, error) { transport.mu.RLock() defer transport.mu.RUnlock() - if transport.closed { - return 0, io.ErrClosedPipe - } for { + if transport.closed.Load() { + return 0, io.ErrClosedPipe + } count, err := unix.Write(transport.fd, buffer) if errors.Is(err, unix.EINTR) { continue } if errors.Is(err, unix.EAGAIN) { fds := []unix.PollFd{{Fd: int32(transport.fd), Events: unix.POLLOUT}} - if _, pollErr := unix.Poll(fds, 1000); pollErr != nil { + if _, pollErr := unix.Poll(fds, 100); pollErr != nil && !errors.Is(pollErr, unix.EINTR) { return 0, pollErr } continue @@ -104,41 +114,25 @@ func (transport *nativeWWANATTransport) Write(buffer []byte) (int, error) { func (transport *nativeWWANATTransport) Drain() error { transport.mu.RLock() defer transport.mu.RUnlock() - if transport.closed { + if transport.closed.Load() { return io.ErrClosedPipe } - // WWAN character-device writes are handed to the modem synchronously and - // have no termios output queue to drain. A previous command that timed out - // can leave late bytes in the input buffer (e.g. a slow CGSN reply that - // arrives after the command deadline); discard them here so the next - // command starts from a clean stream instead of mis-parsing stale output. - buffer := make([]byte, 4096) - for { - fds := []unix.PollFd{{Fd: int32(transport.fd), Events: unix.POLLIN}} - ready, err := unix.Poll(fds, 0) - if err != nil { - return err - } - if ready == 0 || fds[0].Revents&unix.POLLIN == 0 { - return nil - } - if _, err := unix.Read(transport.fd, buffer); err != nil { - if errors.Is(err, unix.EINTR) || errors.Is(err, unix.EAGAIN) { - continue - } - return err - } - } + // WWAN writes have no termios output queue. Input belongs to Session; + // discarding it here could lose a pending final or an unsolicited event. + return nil } func (transport *nativeWWANATTransport) ResetInputBuffer() error { transport.mu.RLock() defer transport.mu.RUnlock() - if transport.closed { + if transport.closed.Load() { return io.ErrClosedPipe } buffer := make([]byte, 4096) for { + if transport.closed.Load() { + return io.ErrClosedPipe + } fds := []unix.PollFd{{Fd: int32(transport.fd), Events: unix.POLLIN}} ready, err := unix.Poll(fds, 0) if err != nil { @@ -162,7 +156,7 @@ func (transport *nativeWWANATTransport) SetReadTimeout(timeout time.Duration) er } transport.mu.Lock() defer transport.mu.Unlock() - if transport.closed { + if transport.closed.Load() { return io.ErrClosedPipe } transport.readTimeout = timeout @@ -170,11 +164,13 @@ func (transport *nativeWWANATTransport) SetReadTimeout(timeout time.Duration) er } func (transport *nativeWWANATTransport) Close() error { + transport.closed.Store(true) transport.mu.Lock() defer transport.mu.Unlock() - if transport.closed { + if transport.fd < 0 { return nil } - transport.closed = true - return unix.Close(transport.fd) + err := unix.Close(transport.fd) + transport.fd = -1 + return err } diff --git a/internal/modem/wwan_at_linux_test.go b/internal/modem/wwan_at_linux_test.go index 19481d14..4c7e73e5 100644 --- a/internal/modem/wwan_at_linux_test.go +++ b/internal/modem/wwan_at_linux_test.go @@ -3,47 +3,66 @@ package modem import ( + "bytes" "errors" "io" "testing" + "time" "golang.org/x/sys/unix" ) -// TestNativeWWANATTransportDrainDiscardsPendingBytes verifies Drain discards -// every byte already buffered on the transport. A command that timed out (e.g. -// AT+CGSN on an MHI modem that never answers OK) can leave its late reply in -// the input buffer; the next command's Drain must clear it, however much data -// is pending, before the session writes the new command. -func TestNativeWWANATTransportDrainDiscardsPendingBytes(t *testing.T) { +func TestNativeWWANATTransportReadEOF(t *testing.T) { + fd, peer := socketpair(t) + transport := &nativeWWANATTransport{fd: fd, readTimeout: 20 * time.Millisecond} + defer transport.Close() + defer unix.Close(peer) + buffer := make([]byte, 64) + if n, err := transport.Read(buffer); n != 0 || err != nil { + t.Fatalf("Read timeout = %d, %v; want 0, nil", n, err) + } + payload := []byte("\r\nRING\r\n") + if _, err := unix.Write(peer, payload); err != nil { + t.Fatal(err) + } + if err := unix.Shutdown(peer, unix.SHUT_RDWR); err != nil { + t.Fatal(err) + } + for _, step := range []struct { + buffer []byte + want []byte + err error + }{{buffer, payload, nil}, {nil, nil, nil}, {buffer, nil, io.EOF}} { + n, err := transport.Read(step.buffer) + if !errors.Is(err, step.err) || !bytes.Equal(step.buffer[:n], step.want) { + t.Fatalf("Read = %q, %v; want %q, %v", step.buffer[:n], err, step.want, step.err) + } + } +} + +func TestNativeWWANATTransportDrainPreservesPendingBytes(t *testing.T) { readFD, writeFD := socketpair(t) defer unix.Close(writeFD) + transport := &nativeWWANATTransport{fd: readFD, readTimeout: -1} + defer transport.Close() - // More than one 4096-byte Drain read: a slow CGSN reply (echo + IMEI + - // trailing CRLF) can exceed a single buffer. - payload := make([]byte, 12000) - for index := range payload { - payload[index] = byte('A' + index%26) - } - payload = append(payload, []byte("\r\n+CGSN: 357091089453326\r\n")...) + payload := []byte("\r\nRING\r\n+CFUN: 1\r\nOK\r\n") if _, err := unix.Write(writeFD, payload); err != nil { - t.Fatalf("seed stale bytes: %v", err) + t.Fatal(err) } - - transport := &nativeWWANATTransport{fd: readFD, readTimeout: -1} if err := transport.Drain(); err != nil { t.Fatalf("Drain: %v", err) } - assertNoPendingBytes(t, readFD, "after Drain") - - // Draining a clean transport is a fast no-op that must not block or error. + buffer := make([]byte, len(payload)) + count, err := unix.Read(readFD, buffer) + if err != nil || count != len(payload) || !bytes.Equal(buffer, payload) { + t.Fatalf("input after Drain = %q, %d, %v; want %q", buffer, count, err, payload) + } if err := transport.Drain(); err != nil { t.Fatalf("second Drain: %v", err) } } -// TestNativeWWANATTransportDrainRejectsClosedTransport covers the guard that -// keeps a poisoned session from draining a wedged, already-closed fd. func TestNativeWWANATTransportDrainRejectsClosedTransport(t *testing.T) { readFD, writeFD := socketpair(t) defer unix.Close(writeFD) @@ -58,21 +77,9 @@ func TestNativeWWANATTransportDrainRejectsClosedTransport(t *testing.T) { func socketpair(t *testing.T) (int, int) { t.Helper() - fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM, 0) + fds, err := unix.Socketpair(unix.AF_UNIX, unix.SOCK_STREAM|unix.SOCK_NONBLOCK|unix.SOCK_CLOEXEC, 0) if err != nil { t.Fatal(err) } return fds[0], fds[1] } - -func assertNoPendingBytes(t *testing.T, fd int, context string) { - t.Helper() - fds := []unix.PollFd{{Fd: int32(fd), Events: unix.POLLIN}} - ready, err := unix.Poll(fds, 0) - if err != nil { - t.Fatalf("poll %s: %v", context, err) - } - if ready != 0 { - t.Fatalf("%s: fd still readable", context) - } -}