fix: reconnect flap and null-body RST (v1.18.1)

internal/daemon/daemon.go

@@ -214,7 +214,7 @@ func Run(ctx context.Context, cfg *config.Config) error {
 			port = 1716
 		}
 		go func() {
-			DialDevice(ctx, ip, port, deviceID, protocol.ProtocolVersion, identity, tlsCfg, devices, plugins, cfg.DeviceID, logger)
+			DialDevice(ctx, ip, port, deviceID, protocol.ProtocolVersion, identity, tlsCfg, devices, plugins, cfg.DeviceID, logger, true)
 			if !dev.IsConnected() {
 				logger.Warn("on-demand pair dial failed", zap.String("device_id", deviceID))
 				return

internal/daemon/ipc_routes.go

@@ -72,7 +72,7 @@ func registerIPCRoutes(handler *ipc.Handler, cfg *config.Config, devices *device
 			if err != nil {
 				return
 			}
-			DialDevice(ctx, addr, 1716, "manual", protocol.ProtocolVersion, identityPkt, tlsCfg, devices, plugins, cfg.DeviceID, logger)
+			DialDevice(ctx, addr, 1716, "manual", protocol.ProtocolVersion, identityPkt, tlsCfg, devices, plugins, cfg.DeviceID, logger, true)
 		}()
 
 		return ipc.Response{OK: true}

internal/daemon/transport.go

@@ -31,27 +31,48 @@ func validDialPort(port int) bool {
 	return port > 0 && port <= 65535
 }
 
-// DialDevice manually connects to a device at the given IP and port.
-func DialDevice(ctx context.Context, targetIP net.IP, targetPort int, targetID string, targetProto int, identity *protocol.Packet, cfg *tls.Config, devices *device.Registry, plugins *plugin.Registry, localDeviceID string, logger *zap.Logger) {
+// DialDevice connects to a device at IP:port. Unless force is set, dials
+// inside reconnectCooldown are skipped so post-roam bursts can't complete
+// near-simultaneously and churn the peer's duplicate resolution.
+// Explicit user actions (pair intent, manual connect) pass force=true.
+func DialDevice(ctx context.Context, targetIP net.IP, targetPort int, targetID string, targetProto int, identity *protocol.Packet, cfg *tls.Config, devices *device.Registry, plugins *plugin.Registry, localDeviceID string, logger *zap.Logger, force bool) {
 	if targetIP == nil || !validDialPort(targetPort) {
 		logger.Debug("refusing to dial invalid target",
 			zap.String("device_id", targetID),
 			zap.String("ip", targetIP.String()),
 			zap.Int("port", targetPort))
 		return
 	}
+	if !force {
+		if dev, ok := devices.Get(targetID); ok && dev.InCooldown() {
+			logger.Debug("skipping dial inside reconnect cooldown",
+				zap.String("device_id", targetID),
+				zap.String("ip", targetIP.String()))
+			return
+		}
+	}
 	addr := fmt.Sprintf("%s:%d", targetIP, targetPort)
 	logger.Debug("dialing discovered device", zap.String("device_id", targetID), zap.String("addr", addr))
 
 	dialer := &net.Dialer{
-		Timeout:   5 * time.Second,
-		KeepAlive: 30 * time.Second, // Crucial for detecting dead connections
+		Timeout: 5 * time.Second,
 	}
 	conn, err := dialer.DialContext(ctx, "tcp", addr)
 	if err != nil {
 		logger.Debug("failed to dial peer", zap.Error(err))
 		return
 	}
+	if tcpConn, ok := conn.(*net.TCPConn); ok {
+		if err := tcpConn.SetKeepAliveConfig(net.KeepAliveConfig{
+			Enable:   true,
+			Idle:     30 * time.Second,
+			Interval: 10 * time.Second,
+			Count:    3,
+		}); err != nil {
+			_ = tcpConn.SetKeepAlive(true)
+			_ = tcpConn.SetKeepAlivePeriod(30 * time.Second)
+		}
+	}
 
 	var myID protocol.IdentityBody
 	json.Unmarshal(identity.Body, &myID)
@@ -196,32 +217,58 @@ func runTransport(ctx context.Context, cfg *tls.Config, bc *discovery.Broadcaste
 				logger.Debug("closing ephemeral discovery connection",
 					zap.String("device_id", body.DeviceID))
 				dev.Disconnect()
+				return
+			}
+			if dev.State() != device.StatePaired {
+				return
 			}
-			return
 		}
 
 		if known && dev.State() == device.StatePaired {
-			// A sighting proves the peer is alive at the sighted address,
-			// so dial it: whoever answers must present the paired
-			// certificate (CN + pinned fingerprint are verified in
-			// handleNewConnection) or setup fails. A spoofed sighting can
-			// only cost a throttled dial, never a session. The persisted
-			// LastIP remains the fallback for a silent (non-broadcasting)
-			// peer via the backoff loop.
+			// Paired sighting proves the peer is alive at the sighted
+			// address. Redial (rate-limited) when disconnected or when
+			// the sighted IP differs (true roam — the live socket is a
+			// half-open zombie). A same-IP sighting on a live socket is
+			// ignored like upstream: phone traffic is event-driven with
+			// long quiet gaps, so read-idle can't tell a zombie from a
+			// healthy session, and redialling churns duplicates the peer
+			// RSTs (split-second flap). Same-IP roam repair arrives via
+			// phone-initiated inbound, which replaces via Connect().
+			// Whoever answers must present the paired certificate (CN +
+			// pinned fingerprint are verified in handleNewConnection) or
+			// setup fails.
 			dev.SetLastSeen(time.Now())
-			if !dev.IsConnected() {
-				// New information beats old backoff: a sighting from an
-				// address other than the failing target proves the peer
-				// roamed, so later cycles restart escalation at the floor.
-				// (The throttled dial below is what hurries this cycle;
-				// an in-flight backoff keeps its own counter.) Same-address
-				// sightings leave the counter alone, preserving flap
-				// protection for a dying peer that keeps announcing.
-				if lastIP := dev.LastIP(); lastIP == nil || !lastIP.Equal(ip) {
-					dev.ResetReconnectAttempt()
+			if dev.NoteSighting(ip) {
+				dev.ResetReconnectAttempt()
+			}
+			// Instant dial only when the peer is somewhere new: the
+			// backoff loop already covers the last-known address, and the
+			// phone redials inbound on its own within a second of a drop.
+			// Racing either with a second outbound breeds crossing
+			// duplicates the peer RSTs (flap war that latches its UI to
+			// "not reachable"). A sighting at a new address is a genuine
+			// roam the backoff can't reach — dial it now.
+			if lastIP := dev.LastIP(); lastIP == nil || !lastIP.Equal(ip) {
+				if dev.ShouldDiscoveryDial(2 * time.Second) {
+					// Prefer the peer's last authenticated listening port
+					// over the unauthenticated sighted one.
+					port := tcpPort
+					if p := dev.LastPort(); validDialPort(p) {
+						port = p
+					}
+					go DialDevice(ctx, ip, port, body.DeviceID, body.ProtocolVersion, identity, cfg, devices, plugins, localDeviceID, logger, false)
 				}
-				if dev.ShouldDiscoveryDial(10 * time.Second) {
-					go DialDevice(ctx, ip, tcpPort, body.DeviceID, body.ProtocolVersion, identity, cfg, devices, plugins, localDeviceID, logger)
+			} else if dev.IsConnected() {
+				if currIP := dev.RemoteIP(); currIP == nil || !currIP.Equal(ip) {
+					// Live socket is bound elsewhere: the peer roamed but
+					// kept its address assertion — replace the zombie.
+					if dev.ShouldDiscoveryDial(2 * time.Second) {
+						port := tcpPort
+						if p := dev.LastPort(); validDialPort(p) {
+							port = p
+						}
+						go DialDevice(ctx, ip, port, body.DeviceID, body.ProtocolVersion, identity, cfg, devices, plugins, localDeviceID, logger, false)
+					}
 				}
 			}
 			return
@@ -240,7 +287,7 @@ func runTransport(ctx context.Context, cfg *tls.Config, bc *discovery.Broadcaste
 		if pairingMode || dev.ConsumePairDial() {
 			// Spawn goroutine to prevent blocking the discovery listener
 			go func(targetIP net.IP, targetPort int, targetID string, targetProto int) {
-				DialDevice(ctx, targetIP, targetPort, targetID, targetProto, identity, cfg, devices, plugins, localDeviceID, logger)
+				DialDevice(ctx, targetIP, targetPort, targetID, targetProto, identity, cfg, devices, plugins, localDeviceID, logger, true)
 			}(ip, tcpPort, body.DeviceID, body.ProtocolVersion)
 			return
 		}
@@ -252,7 +299,7 @@ func runTransport(ctx context.Context, cfg *tls.Config, bc *discovery.Broadcaste
 			}
 			// Spawn goroutine to prevent blocking the discovery listener
 			go func(targetIP net.IP, targetPort int, targetID string, targetProto int) {
-				DialDevice(ctx, targetIP, targetPort, targetID, targetProto, identity, cfg, devices, plugins, localDeviceID, logger)
+				DialDevice(ctx, targetIP, targetPort, targetID, targetProto, identity, cfg, devices, plugins, localDeviceID, logger, false)
 			}(ip, tcpPort, body.DeviceID, body.ProtocolVersion)
 		}
 		// Otherwise the device already had its ephemeral dial for this
@@ -286,6 +333,29 @@ func runTransport(ctx context.Context, cfg *tls.Config, bc *discovery.Broadcaste
 				if err != nil {
 					return
 				}
+				// Refuse duplicate bursts pre-TLS: a device that completed a
+				// handshake inside the cooldown already has a live session;
+				// letting this one through would churn both ends' duplicate
+				// resolution. Pairing traffic always passes (strangers have
+				// no cooldown entry anyway).
+				var preBody protocol.IdentityBody
+				if err := json.Unmarshal(preTlsPkt.Body, &preBody); err == nil {
+					if preBody.TargetDeviceID != "" && preBody.TargetDeviceID != localDeviceID {
+						// Log-only: a stale cached ID on the peer (e.g. after
+						// our reinstall minted a new device ID) must not kill
+						// an otherwise legitimate inbound.
+						logger.Debug("inbound addressed to another device",
+							zap.String("device_id", preBody.DeviceID),
+							zap.String("target_device_id", preBody.TargetDeviceID))
+					}
+					if dev, ok := devices.Get(preBody.DeviceID); ok && dev.InCooldown() &&
+						(bc == nil || !bc.IsRunning()) {
+						logger.Debug("refusing inbound inside reconnect cooldown",
+							zap.String("device_id", preBody.DeviceID))
+						protocol.ReleasePacket(preTlsPkt)
+						return
+					}
+				}
 				protocol.ReleasePacket(preTlsPkt)
 
 				tlsConn := tls.Client(newConn, cfg)
@@ -488,7 +558,14 @@ func reconnectWithBackoff(
 			zap.Int("attempt", attempt+1),
 		)
 
-		DialDevice(ctx, ip, 1716, dev.ID(), protocol.ProtocolVersion, identity, cfg, devices, plugins, localDeviceID, logger)
+		// Prefer the peer's last advertised listening port over the
+		// default: the identity may carry a non-standard port (or none
+		// at all, in which case LastPort is 0 and we fall back).
+		port := 1716
+		if p := dev.LastPort(); validDialPort(p) {
+			port = p
+		}
+		DialDevice(ctx, ip, port, dev.ID(), protocol.ProtocolVersion, identity, cfg, devices, plugins, localDeviceID, logger, false)
 
 		if dev.IsConnected() {
 			logger.Info("auto-reconnect: succeeded",

internal/device/device.go

@@ -73,6 +73,12 @@ type Device struct {
 	done      chan struct{}
 	closeOnce sync.Once
 
+	// lastConnect marks the last completed handshake; new handshakes
+	// inside reconnectCooldown are refused to starve duplicate bursts.
+	lastConnect time.Time
+	// lastSightedIP remembers the previous discovery sighting so only
+	// confirmed roams reset the reconnect backoff (see NoteSighting).
+	lastSightedIP net.IP
 	BatteryCharge int
 	IsCharging    bool
 
@@ -132,24 +138,56 @@ func (d *Device) SetBus(bus *events.Bus) {
 	d.bus = bus
 }
 
+// reconnectCooldown refuses new handshakes this long after a completed
+// one, so post-roam bursts can't complete near-simultaneously and churn
+// the peer's duplicate resolution. Reference stacks rate-limit the same
+// way (desktop 500ms, Android MILLIS_DELAY_BETWEEN_CONNECTIONS_TO_SAME_DEVICE
+// 1000ms); 1s matches upstream Android.
+const reconnectCooldown = 1 * time.Second
+
 // Connect establishes a connection for the device and starts the reader and writer loops.
+// A new authenticated connection immediately replaces any existing one
+// (matching LanDeviceLink::reset / LanLink.reset). The old socket is
+// closed; its readLoop will exit and disconnectConn will ignore it
+// because d.conn no longer points at it.
 func (d *Device) Connect(ctx context.Context, conn *transport.Conn, dispatch func(context.Context, *Device, *protocol.Packet) bool, onConnect func(*Device), onDisconnect func(*Device)) {
 	d.mu.Lock()
-	if d.conn != nil {
-		_ = d.conn.Close()
-	}
-	d.conn = conn
 	d.pluginDispatch = dispatch
 	d.onConnect = onConnect
 	d.onDisconnect = onDisconnect
 
-	// Renew the send channel on connect in case it was closed during disconnect.
+	var oldConn *transport.Conn
+	var oldAddr, newAddr string
+	if d.conn != nil {
+		oldConn = d.conn
+		oldDone := d.done
+		oldAddr = oldConn.RemoteAddr().String()
+		newAddr = conn.RemoteAddr().String()
+		// Close the old done so its writerLoop exits; the new
+		// writerLoop will own the fresh channel.
+		d.closeOnce.Do(func() {
+			if oldDone != nil {
+				close(oldDone)
+			}
+		})
+	}
+
+	d.conn = conn
+
+	// Renew the send channel on (re)connect in case it was closed during disconnect.
 	d.sendChan = make(chan *protocol.Packet, 32)
 	d.done = make(chan struct{})
 	d.closeOnce = sync.Once{}
 	bus := d.bus
 	d.mu.Unlock()
 
+	if oldConn != nil {
+		_ = oldConn.Close()
+		d.logger.Debug("replacing superseded connection",
+			zap.String("old_addr", oldAddr),
+			zap.String("new_addr", newAddr))
+	}
+
 	d.logger.Info("device connected", zap.String("remote_addr", conn.RemoteAddr().String()))
 	if bus != nil {
 		bus.Publish(events.TypeDeviceConnected, d.id, map[string]interface{}{
@@ -170,17 +208,18 @@ func (d *Device) Connect(ctx context.Context, conn *transport.Conn, dispatch fun
 		d.mu.Unlock()
 	}
 
-	// Record when the connection was established so a quick drop can be
-	// distinguished from a genuinely stable connection.
+	// connectStarted distinguishes quick drops from stable connections,
+	// and doubles as the duplicate-cooldown clock (see InCooldown).
 	d.mu.Lock()
 	d.connectStarted = time.Now()
+	d.lastConnect = d.connectStarted
 	d.mu.Unlock()
 
 	go d.readLoop(ctx, conn)
-	go d.writerLoop(ctx, conn)
+	go d.writerLoop(ctx)
 }
 
-// Disconnect terminates the connection and stops the loops.
+// Disconnect terminates the session and stops the loops.
 func (d *Device) Disconnect() {
 	d.mu.Lock()
 	d.lastSeen = time.Now()
@@ -193,6 +232,7 @@ func (d *Device) Disconnect() {
 	d.logger.Info("device disconnected")
 	_ = d.conn.Close()
 	d.conn = nil
+	d.lastConnect = time.Time{}
 
 	// Capture these to call outside the lock to prevent deadlocks!
 	onDisc := d.onDisconnect
@@ -221,22 +261,25 @@ func (d *Device) IsConnected() bool {
 	return d.conn != nil
 }
 
-// disconnectConn disconnects only if the provided connection matches the current one.
-// This prevents old readLoops from terminating new connections.
+// disconnectConn handles a session's death. If the conn that died is no
+// longer the current d.conn (it was superseded by a newer authenticated
+// connection via Connect), the event is ignored silently — matching
+// LanDeviceLink::reset's `if (m_socket == socket)` guard. Only the
+// current preferred session's death triggers the full disconnect.
 func (d *Device) disconnectConn(conn *transport.Conn) {
 	d.mu.Lock()
 
-	// Only disconnect if this conn is still the active one
 	if d.conn != conn {
 		d.mu.Unlock()
-		d.logger.Debug("ignoring disconnect from old connection")
+		d.logger.Debug("ignoring disconnect from superseded connection")
 		return
 	}
 
 	d.lastSeen = time.Now()
 	d.logger.Info("device disconnected")
 	_ = d.conn.Close()
 	d.conn = nil
+	d.lastConnect = time.Time{}
 
 	// Capture callbacks to execute outside the lock
 	onDisc := d.onDisconnect
@@ -504,6 +547,27 @@ func (d *Device) ClearEphemeral() {
 	d.ephemeralDialed = false
 }
 
+// InCooldown reports whether a handshake completed too recently to start
+// another one for this device.
+func (d *Device) InCooldown() bool {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	return !d.lastConnect.IsZero() && time.Since(d.lastConnect) < reconnectCooldown
+}
+
+// NoteSighting records a discovery sighting while disconnected, reporting
+// whether it confirms a genuine roam: same new address twice running.
+// Single sightings prove nothing (IPv4/IPv6 alternation, AP flicker).
+func (d *Device) NoteSighting(sighted net.IP) (roamed bool) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	defer func() { d.lastSightedIP = sighted }()
+	if lastIP := d.lastIP; lastIP == nil || !lastIP.Equal(sighted) {
+		return sighted.Equal(d.lastSightedIP)
+	}
+	return false
+}
+
 // TryReconnect attempts to mark the device as reconnecting.
 // Returns true if this goroutine should proceed; false if another
 // reconnect goroutine is already running.