From 690fa377f5b07d7b8440f66dcedd553af9f84305 Mon Sep 17 00:00:00 2001 From: bethropolis <66518866+bethropolis@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:36:41 +0300 Subject: [PATCH 1/3] fix(reconnect): replace zombie sessions, stop self-redial churn - Device.Connect immediately replaces any live session (LanDeviceLink::reset): old socket closed, superseded readLoop death ignored silently - Same-IP sightings on live sockets no longer redial; roam repair arrives via phone-initiated inbound or IP-change dials, ending the 2s RST war - reconnectCooldown 10s -> 1s (upstream Android MILLIS_DELAY = 1000ms) - Disconnected sighting dials only target roamed (new-IP) addresses; the backoff loop covers last-known targets so post-drop races stop - TargetDeviceID mismatch demoted to debug log (stale cached IDs) - Aggressive TCP keepalive (30s/10s/3) with legacy fallback on both dial and accept paths; 10s write deadline kills Send-Q zombies - Writer resolves current session per packet, releases pool packets on both success and error paths --- internal/daemon/daemon.go | 2 +- internal/daemon/ipc_routes.go | 2 +- internal/daemon/transport.go | 131 ++++++++++++++---- internal/device/device.go | 90 +++++++++++-- internal/device/device_test.go | 238 +++++++++++++++++++++++++++++++++ internal/device/io.go | 46 ++++++- internal/testutil/peer.go | 23 +++- internal/transport/conn.go | 9 ++ internal/transport/listener.go | 11 +- 9 files changed, 496 insertions(+), 56 deletions(-) diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index e0ccee4..6877ed2 100644 --- a/internal/daemon/daemon.go +++ b/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 diff --git a/internal/daemon/ipc_routes.go b/internal/daemon/ipc_routes.go index 30e8d8d..8751b3f 100644 --- a/internal/daemon/ipc_routes.go +++ b/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} diff --git a/internal/daemon/transport.go b/internal/daemon/transport.go index 5e1ce3e..478f2e0 100644 --- a/internal/daemon/transport.go +++ b/internal/daemon/transport.go @@ -31,8 +31,11 @@ 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), @@ -40,18 +43,36 @@ func DialDevice(ctx context.Context, targetIP net.IP, targetPort int, targetID s 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", diff --git a/internal/device/device.go b/internal/device/device.go index bb6c2a1..c352ead 100644 --- a/internal/device/device.go +++ b/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,15 +261,17 @@ 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 } @@ -237,6 +279,7 @@ func (d *Device) disconnectConn(conn *transport.Conn) { 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. diff --git a/internal/device/device_test.go b/internal/device/device_test.go index e483814..a381a56 100644 --- a/internal/device/device_test.go +++ b/internal/device/device_test.go @@ -4,9 +4,11 @@ import ( "context" "crypto/tls" "net" + "sync/atomic" "testing" "time" + "github.com/bethropolis/kcd/internal/protocol" "github.com/bethropolis/kcd/internal/transport" "go.uber.org/zap/zaptest" ) @@ -191,3 +193,239 @@ func TestDeviceInfoDialTargetRejectsGarbage(t *testing.T) { }) } } + +// pipeConn builds a transport.Conn over an in-memory pipe without a TLS +// handshake. Closing peer tears the session down through the readLoop, +// exactly like a dropped TCP connection. +func pipeConn(t *testing.T) (*transport.Conn, net.Conn) { + t.Helper() + left, right := net.Pipe() + return transport.NewConn(tls.Client(left, &tls.Config{InsecureSkipVerify: true})), right +} + +func waitConnected(t *testing.T, d *Device, want bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for d.IsConnected() != want && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if d.IsConnected() != want { + t.Fatalf("expected connected=%v", want) + } +} + +// waitCounter polls an atomic counter to a value. Callbacks fire on device +// goroutines, so tests must never read plain ints set from them. +func waitCounter(t *testing.T, c *atomic.Int32, want int32) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for c.Load() != want && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if c.Load() != want { + t.Fatalf("expected counter %d, got %d", want, c.Load()) + } +} + +// quiesce drains every session loop so no goroutine logs after the test +// returns (zaptest panics on late logs). +func quiesce(t *testing.T, d *Device) { + t.Helper() + d.mu.RLock() + quiet := d.conn == nil + d.mu.RUnlock() + deadline := time.Now().Add(2 * time.Second) + for !quiet && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + d.mu.RLock() + quiet = d.conn == nil + d.mu.RUnlock() + } + time.Sleep(100 * time.Millisecond) +} + +// readAny asserts the writer loop delivers bytes to peer within the deadline. +func readAny(t *testing.T, peer net.Conn) { + t.Helper() + _ = peer.SetReadDeadline(time.Now().Add(2 * time.Second)) + buf := make([]byte, 4096) + n, err := peer.Read(buf) + _ = peer.SetReadDeadline(time.Time{}) + if err != nil { + t.Fatalf("expected bytes from writer loop, got error: %v", err) + } + if n == 0 { + t.Fatal("expected bytes from writer loop, got 0") + } +} + +func TestDevice_ReplaceOnNewAuth(t *testing.T) { + logger := zaptest.NewLogger(t) + d := NewDevice("dup", "Phone", "phone", logger) + var connects, disconnects atomic.Int32 + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + a, rightA := pipeConn(t) + d.Connect(ctx, a, nil, + func(*Device) { connects.Add(1) }, + func(*Device) { disconnects.Add(1) }) + // Second session while one is live: replaces the dead socket + // immediately (matching LanDeviceLink::reset). + b, rightB := pipeConn(t) + d.Connect(ctx, b, nil, + func(*Device) { connects.Add(1) }, + func(*Device) { disconnects.Add(1) }) + + if !d.IsConnected() { + t.Fatal("device must stay connected after replace") + } + + // Old peer must see EOF — old socket was closed. + _ = rightA.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + if _, err := rightA.Read(make([]byte, 1)); err == nil { + t.Fatal("old connection must be closed on replace") + } + _ = rightA.Close() + + // No disconnect event for the superseded session. + time.Sleep(50 * time.Millisecond) + if disconnects.Load() != 0 { + t.Fatalf("superseded disconnect must not fire onDisconnect, got %d", disconnects.Load()) + } + + // Writer must serve the new session. + pkt, err := protocol.NewPacket("kdeconnect.ping", map[string]any{}) + if err != nil { + t.Fatalf("build ping packet: %v", err) + } + if err := d.Send(pkt); err != nil { + t.Fatalf("Send after replace: %v", err) + } + readAny(t, rightB) + + _ = rightB.Close() + d.Disconnect() + waitCounter(t, &disconnects, 1) + quiesce(t, d) +} + +func TestDevice_SupersededDisconnectSilent(t *testing.T) { + logger := zaptest.NewLogger(t) + d := NewDevice("sup", "Phone", "phone", logger) + var disconnects atomic.Int32 + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + a, rightA := pipeConn(t) + d.Connect(ctx, a, nil, nil, func(*Device) { disconnects.Add(1) }) + b, rightB := pipeConn(t) + d.Connect(ctx, b, nil, nil, func(*Device) { disconnects.Add(1) }) + + // The first session was already superseded; its peer closing is silent. + _ = rightA.Close() + time.Sleep(80 * time.Millisecond) + if disconnects.Load() != 0 { + t.Fatalf("superseded session death must not fire onDisconnect, got %d", disconnects.Load()) + } + if !d.IsConnected() { + t.Fatal("replacement session must remain connected") + } + + // Closing the current session fires the single disconnect. + _ = rightB.Close() + waitCounter(t, &disconnects, 1) + waitConnected(t, d, false) + quiesce(t, d) +} + +func TestDevice_ReplaceClosesOld(t *testing.T) { + logger := zaptest.NewLogger(t) + d := NewDevice("rep", "Phone", "phone", logger) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + a, rightA := pipeConn(t) + d.Connect(ctx, a, nil, nil, nil) + b, rightB := pipeConn(t) + defer rightB.Close() + d.Connect(ctx, b, nil, nil, nil) + c, rightC := pipeConn(t) + defer rightC.Close() + d.Connect(ctx, c, nil, nil, nil) + + // Each replace closes the previous socket; oldest peer sees EOF. + _ = rightA.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + if _, err := rightA.Read(make([]byte, 1)); err == nil { + t.Fatal("oldest connection must be closed on replace") + } + _ = rightA.Close() + + // Writer must serve the newest session. + pkt, err := protocol.NewPacket("kdeconnect.ping", map[string]any{}) + if err != nil { + t.Fatalf("build ping packet: %v", err) + } + if err := d.Send(pkt); err != nil { + t.Fatalf("Send after replace: %v", err) + } + readAny(t, rightC) + + d.Disconnect() + waitConnected(t, d, false) + quiesce(t, d) +} + +func TestDevice_CooldownWindow(t *testing.T) { + logger := zaptest.NewLogger(t) + d := NewDevice("cd", "Phone", "phone", logger) + + if d.InCooldown() { + t.Fatal("fresh device must not be in cooldown") + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + a, rightA := pipeConn(t) + defer rightA.Close() + d.Connect(ctx, a, nil, nil, nil) + + if !d.InCooldown() { + t.Fatal("device must be in cooldown right after connect") + } + + // White-box time travel past the window. + d.mu.Lock() + d.lastConnect = time.Now().Add(-(reconnectCooldown + time.Second)) + d.mu.Unlock() + if d.InCooldown() { + t.Fatal("cooldown must lapse after the window") + } + d.Disconnect() + quiesce(t, d) +} + +func TestDevice_NoteSightingRoamConfirm(t *testing.T) { + logger := zaptest.NewLogger(t) + d := NewDevice("roam", "Phone", "phone", logger) + d.SetLastIP(net.ParseIP("192.168.1.10")) + + other := net.ParseIP("192.168.1.20") + if d.NoteSighting(other) { + t.Fatal("single sighting must not confirm a roam") + } + if !d.NoteSighting(other) { + t.Fatal("repeated sighting at a new address must confirm a roam") + } + // Wobble back: home address never confirms. + if d.NoteSighting(net.ParseIP("192.168.1.10")) { + t.Fatal("sighting at the known address must not confirm a roam") + } + // Alternation never confirms either. + if d.NoteSighting(other) { + t.Fatal("wobble must not confirm a roam") + } +} diff --git a/internal/device/io.go b/internal/device/io.go index 50f5a9f..8690f42 100644 --- a/internal/device/io.go +++ b/internal/device/io.go @@ -2,12 +2,22 @@ package device import ( "context" + "errors" + "net" + "time" "github.com/bethropolis/kcd/internal/protocol" "github.com/bethropolis/kcd/internal/transport" "go.uber.org/zap" ) +// writeTimeout bounds one WritePacket call. LAN writes complete in +// milliseconds; anything slower is a half-open zombie whose retransmit +// queue never drains (TCP keepalive can't save it — it only probes idle +// sockets). Timing out routes the death through disconnectConn so the +// backoff and discovery repair paths can run. +const writeTimeout = 10 * time.Second + func (d *Device) Send(p *protocol.Packet) error { d.mu.RLock() connected := d.conn != nil @@ -74,16 +84,16 @@ func (d *Device) readLoop(ctx context.Context, conn *transport.Conn) { } } -func (d *Device) writerLoop(ctx context.Context, conn *transport.Conn) { +// writerLoop is the sole writer to device sockets. It resolves the +// preferred session per packet so promotions need no restart; write +// timeouts fail the session via disconnectConn, other write errors keep +// looping while readLoop routes the death. +func (d *Device) writerLoop(ctx context.Context) { d.mu.RLock() sendChan := d.sendChan done := d.done d.mu.RUnlock() - if conn == nil { - return - } - for { select { case <-ctx.Done(): @@ -91,10 +101,32 @@ func (d *Device) writerLoop(ctx context.Context, conn *transport.Conn) { case <-done: return case pkt := <-sendChan: - if err := conn.WritePacket(pkt); err != nil { + d.mu.RLock() + conn := d.conn + d.mu.RUnlock() + if conn == nil { + // Fully disconnected (done closes right behind this) — + // drop rather than block the loop's exit. + protocol.ReleasePacket(pkt) + continue + } + d.logger.Debug("sending packet", zap.String("type", pkt.Type)) + _ = conn.SetWriteDeadline(time.Now().Add(writeTimeout)) + err := conn.WritePacket(pkt) + _ = conn.SetWriteDeadline(time.Time{}) + if err != nil { d.logger.Debug("write packet error", zap.Error(err)) - return // write failed -> drop out, readLoop will detect disconnect soon + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + // Stuck socket: fail the session now instead of + // blocking the writer forever behind Send-Q. + d.logger.Info("write timed out, dropping zombie connection") + protocol.ReleasePacket(pkt) + d.disconnectConn(conn) + continue + } } + protocol.ReleasePacket(pkt) } } } diff --git a/internal/testutil/peer.go b/internal/testutil/peer.go index a841983..4c5cd67 100644 --- a/internal/testutil/peer.go +++ b/internal/testutil/peer.go @@ -31,12 +31,22 @@ func NewMockPeer(t *testing.T, tlsConfig *tls.Config) *MockPeer { // TCP initiator acts as TLS server). func (p *MockPeer) Dial(serverAddr string) net.Conn { p.t.Helper() + conn, err := p.TryDial(serverAddr) + if err != nil { + p.t.Fatalf("mockpeer: dial: %v", err) + } + return conn +} + +// TryDial is Dial without the Fatalf: it returns handshake-refusal errors +// (e.g. cooldown-gated peers closing pre-TLS) for the caller to assert on. +func (p *MockPeer) TryDial(serverAddr string) (net.Conn, error) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() var d net.Dialer conn, err := d.DialContext(ctx, "tcp", serverAddr) if err != nil { - p.t.Fatalf("mockpeer: dial %s: %v", serverAddr, err) + return nil, fmt.Errorf("mockpeer: dial %s: %w", serverAddr, err) } // Send plaintext identity @@ -48,12 +58,14 @@ func (p *MockPeer) Dial(serverAddr string) net.Conn { TCPPort: 1716, }) if err != nil { - p.t.Fatalf("mockpeer: build identity: %v", err) + conn.Close() + return nil, fmt.Errorf("mockpeer: build identity: %w", err) } data, _ := json.Marshal(identPkt) data = append(data, '\n') if _, err := conn.Write(data); err != nil { - p.t.Fatalf("mockpeer: write identity: %v", err) + conn.Close() + return nil, fmt.Errorf("mockpeer: write identity: %w", err) } // Upgrade to TLS as server (TCP initiator = TLS server per KDE Connect spec). @@ -61,9 +73,10 @@ func (p *MockPeer) Dial(serverAddr string) net.Conn { handshakeCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() if err := tlsConn.HandshakeContext(handshakeCtx); err != nil { - p.t.Fatalf("mockpeer: tls handshake: %v", err) + tlsConn.Close() + return nil, fmt.Errorf("mockpeer: tls handshake: %w", err) } - return tlsConn + return tlsConn, nil } // DialPipe creates an in-process net.Pipe pair, performs the identity exchange diff --git a/internal/transport/conn.go b/internal/transport/conn.go index b8c12e4..9bb876a 100644 --- a/internal/transport/conn.go +++ b/internal/transport/conn.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "crypto/x509" "net" + "time" "github.com/bethropolis/kcd/internal/protocol" ) @@ -42,6 +43,14 @@ func (c *Conn) Close() error { return c.tlsConn.Close() } +// SetWriteDeadline bounds the next WritePacket call. A half-open socket +// with a stuck retransmit queue (Send-Q never drains) would otherwise block +// the writer forever while TCP keepalive — which only probes idle sockets — +// never fires. Callers clear with the zero time after the write. +func (c *Conn) SetWriteDeadline(t time.Time) error { + return c.tlsConn.SetWriteDeadline(t) +} + // PeerCert returns the validated client/server certificate presented by the peer. // If no certificate was presented (which shouldn't happen with proper tls.Config), it returns nil. func (c *Conn) PeerCert() *x509.Certificate { diff --git a/internal/transport/listener.go b/internal/transport/listener.go index 85d4ccf..95318dc 100644 --- a/internal/transport/listener.go +++ b/internal/transport/listener.go @@ -31,8 +31,15 @@ func (l *Listener) Accept() (net.Conn, error) { } if tcpConn, ok := conn.(*net.TCPConn); ok { - tcpConn.SetKeepAlive(true) - tcpConn.SetKeepAlivePeriod(30 * time.Second) + 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) + } } return conn, nil From df48b16a17c9d952c3c51537bbab1f8dcbc81f0e Mon Sep 17 00:00:00 2001 From: bethropolis <66518866+bethropolis@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:36:54 +0300 Subject: [PATCH 2/3] fix(protocol): never emit body:null on the wire Bodyless requests sent explicit "body":null; at least one phone build aborts the whole link on it (RST seconds after every connect carrying the contacts UID request). Normalize nil bodies to {} in NewPacket and at both existing nil-body call sites (contacts sync, SMS conversations). --- internal/plugins/contacts/contacts.go | 7 ++++++- internal/plugins/sms/sms.go | 4 +++- internal/protocol/packet.go | 6 ++++++ 3 files changed, 15 insertions(+), 2 deletions(-) diff --git a/internal/plugins/contacts/contacts.go b/internal/plugins/contacts/contacts.go index 8c5cf21..6c7d5a7 100644 --- a/internal/plugins/contacts/contacts.go +++ b/internal/plugins/contacts/contacts.go @@ -148,8 +148,13 @@ func (p *ContactsPlugin) Handle(ctx context.Context, dev device.Sender, pkt *pro // RequestSync asks the phone for all contact UIDs and timestamps, which // starts the sync round trips. Responses arrive async via Handle. +// +// The body must be an empty object, not null: stock implementations send +// `"body":{}` for bodyless requests, and at least one phone build aborts +// the whole link on an explicit null body (observed as an immediate RST +// after every connect that carried `"body":null`). func (p *ContactsPlugin) RequestSync(dev device.Sender) error { - pkt, err := protocol.NewPacket(PacketTypeContactsRequestUIDs, nil) + pkt, err := protocol.NewPacket(PacketTypeContactsRequestUIDs, map[string]any{}) if err != nil { return err } diff --git a/internal/plugins/sms/sms.go b/internal/plugins/sms/sms.go index 75bdeee..818c7ad 100644 --- a/internal/plugins/sms/sms.go +++ b/internal/plugins/sms/sms.go @@ -313,8 +313,10 @@ func (p *SMSPlugin) SendSMS(dev device.Sender, phoneNumber, message string) erro // --- Conversation browsing (Phase 2) --------------------------------------- // RequestConversations asks the phone for a summary of all conversations. +// Bodyless requests use an empty object (never null) on the wire; see +// contacts.RequestSync for why explicit null is dangerous. func (p *SMSPlugin) RequestConversations(dev device.Sender) error { - pkt, err := protocol.NewPacket(PacketTypeSMSRequestConvs, nil) + pkt, err := protocol.NewPacket(PacketTypeSMSRequestConvs, map[string]any{}) if err != nil { return err } diff --git a/internal/protocol/packet.go b/internal/protocol/packet.go index 4e2263c..04a1888 100644 --- a/internal/protocol/packet.go +++ b/internal/protocol/packet.go @@ -82,7 +82,13 @@ func (p *Packet) Reset() { } // NewPacket creates a new Packet with the current timestamp and given type/body. +// A nil body is normalized to an empty object: stock implementations always +// send `"body":{}` for bodyless requests, and explicit `"body":null` has +// been observed to abort phone links. func NewPacket(typ string, body interface{}) (*Packet, error) { + if body == nil { + body = map[string]any{} + } raw, err := json.Marshal(body) if err != nil { return nil, fmt.Errorf("protocol: marshal body: %w", err) From ef9bacc34199c19f45d1173293a1787c620629fe Mon Sep 17 00:00:00 2001 From: bethropolis <66518866+bethropolis@users.noreply.github.com> Date: Wed, 16 Sep 2026 14:36:54 +0300 Subject: [PATCH 3/3] fix(test): hermetic duplicate-session integration test Drive the roam race (cooldown refusal, replacement, silent superseded close) with per-device event filtering so real-LAN neighbors dialing into :1716 mid-run cannot pollute assertions. Cooldown sleep 11s -> 2s for the 1s window. --- internal/integration/dedup_test.go | 193 +++++++++++++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 internal/integration/dedup_test.go diff --git a/internal/integration/dedup_test.go b/internal/integration/dedup_test.go new file mode 100644 index 0000000..d183ea3 --- /dev/null +++ b/internal/integration/dedup_test.go @@ -0,0 +1,193 @@ +//go:build integration + +package integration + +import ( + "context" + "testing" + "time" + + "github.com/bethropolis/kcd/internal/cert" + "github.com/bethropolis/kcd/internal/config" + "github.com/bethropolis/kcd/internal/events" + "github.com/bethropolis/kcd/internal/protocol" + "github.com/bethropolis/kcd/internal/testutil" +) + +// drainSnapshot skips the state.snapshot bootstrap event every watch stream +// starts with, returning the channel for domain events. +func drainSnapshot(t *testing.T, evCh <-chan events.Event) { + t.Helper() + deadline := time.After(3 * time.Second) + for { + select { + case ev := <-evCh: + if ev.Type != events.TypeStateSnapshot { + t.Fatalf("expected snapshot bootstrap first, got %s", ev.Type) + } + return + case <-deadline: + t.Fatal("timed out waiting for snapshot bootstrap") + } + } +} + +// nextFor returns the next event for deviceID, skipping LAN-neighbor noise. +// The test daemon binds :1716 on all interfaces, so real devices dial into +// it mid-run and their events share the watch stream. +func nextFor(t *testing.T, evCh <-chan events.Event, deviceID, what string) events.Event { + t.Helper() + deadline := time.After(5 * time.Second) + for { + select { + case ev := <-evCh: + if ev.DeviceID != deviceID { + continue + } + return ev + case <-deadline: + t.Fatalf("timed out waiting for %s", what) + return events.Event{} + } + } +} + +// expectNoFor fails if an event for deviceID arrives within d; events for +// other (real-LAN) devices are ignored. +func expectNoFor(t *testing.T, evCh <-chan events.Event, deviceID string, d time.Duration, what string) { + t.Helper() + deadline := time.After(d) + for { + select { + case ev := <-evCh: + if ev.DeviceID == deviceID { + t.Fatalf("expected no %s, got %s", what, ev.Type) + } + case <-deadline: + return + } + } +} + +// TestDuplicateSessionFailoverIntegration drives the roam race: a duplicate +// handshake inside the cooldown is refused pre-TLS (exactly one +// device.connected, zero device.disconnected); past the cooldown a second +// handshake replaces the live session (old socket closed, writer migrates), +// and the superseded session's death is silent. +func TestDuplicateSessionFailoverIntegration(t *testing.T) { + dir := t.TempDir() + t.Setenv("XDG_STATE_HOME", dir) + t.Setenv("XDG_CONFIG_HOME", dir) + + cfg := config.Defaults() + cfg.SocketPath = dir + "/kcd.sock" + cfg.CertFile = dir + "/cert.pem" + cfg.KeyFile = dir + "/key.pem" + cfg.DeviceID = "test-daemon-dedup" + cfg.LogLevel = "debug" + cfg.Plugins.Battery = true + cfg.Plugins.Notification = false + cfg.Plugins.Clipboard = false + cfg.Plugins.Share = false + cfg.Plugins.RunCommand = false + cfg.Plugins.MPRIS = false + cfg.Plugins.Ping = false + cfg.Plugins.Telephony = false + cfg.Plugins.Connectivity = false + cfg.Plugins.Mousepad = false + cfg.Plugins.SFTP = false + cfg.Plugins.FindMyPhone = false + cfg.Plugins.LockDevice = false + cfg.Plugins.SystemVolume = false + cfg.Plugins.SMS = false + + _, cl := testutil.StartTestDaemon(t, cfg) + + evCh := make(chan events.Event, 16) + go func() { + _ = cl.Watch(context.Background(), + []string{"device.connected", "device.disconnected", "battery.update"}, evCh) + }() + drainSnapshot(t, evCh) + + peerCertPair, err := cert.LoadOrGenerate(dir+"/peer-cert.pem", dir+"/peer-key.pem", "mock-peer") + if err != nil { + t.Fatalf("peer cert: %v", err) + } + peer := testutil.NewMockPeer(t, cert.TLSConfig(peerCertPair)) + + // First session: full handshake + pair accept. + connA := peer.Dial("127.0.0.1:1716") + defer connA.Close() + if _, err := peer.ReadPacket(connA); err != nil { + t.Fatalf("read daemon identity: %v", err) + } + identPkt, _ := protocol.NewPacket(protocol.TypeIdentity, protocol.IdentityBody{ + DeviceID: "mock-peer", DeviceName: "Mock Peer", + DeviceType: "phone", ProtocolVersion: protocol.ProtocolVersion, + }) + if err := peer.SendPacket(connA, identPkt); err != nil { + t.Fatalf("send identity: %v", err) + } + pairPkt, _ := protocol.NewPacket("kdeconnect.pair", map[string]bool{"pair": true}) + if err := peer.SendPacket(connA, pairPkt); err != nil { + t.Fatalf("send pair: %v", err) + } + time.Sleep(100 * time.Millisecond) + if err := cl.Pair("mock-peer"); err != nil { + t.Fatalf("accept pair: %v", err) + } + if ev := nextFor(t, evCh, "mock-peer", "device.connected"); ev.Type != events.TypeDeviceConnected { + t.Fatalf("expected device.connected, got %s", ev.Type) + } + + // Immediate duplicate handshake: refused inside the cooldown, no events. + if _, err := peer.TryDial("127.0.0.1:1716"); err == nil { + t.Fatal("expected duplicate handshake refused inside cooldown") + } + expectNoFor(t, evCh, "mock-peer", 500*time.Millisecond, "duplicate connect/disconnect") + + // Past the cooldown (1s, matching upstream Android) the same handshake + // replaces the live session. + time.Sleep(2 * time.Second) + connC := peer.Dial("127.0.0.1:1716") + defer connC.Close() + if _, err := peer.ReadPacket(connC); err != nil { + t.Fatalf("read daemon identity (C): %v", err) + } + if err := peer.SendPacket(connC, identPkt); err != nil { + t.Fatalf("send identity (C): %v", err) + } + // Replacement is immediate; old socket is closed. The replacement + // fires a fresh device.connected (same device, new session) while the + // superseded session's death stays silent (no device.disconnected). + if ev := nextFor(t, evCh, "mock-peer", "device.connected"); ev.Type != events.TypeDeviceConnected { + t.Fatalf("expected device.connected on replacement, got %s", ev.Type) + } + expectNoFor(t, evCh, "mock-peer", 500*time.Millisecond, "replacement disconnect") + // connA's peer should see EOF since it was replaced. + _ = connA.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + + // Survivor carries traffic on the new socket. + battPkt, _ := protocol.NewPacket("kdeconnect.battery", map[string]any{ + "currentCharge": 55, "isCharging": false, "thresholdEvent": 0, + }) + if err := peer.SendPacket(connC, battPkt); err != nil { + t.Fatalf("send battery: %v", err) + } + if ev := nextFor(t, evCh, "mock-peer", "battery.update"); ev.Type != events.TypeBatteryUpdate { + t.Fatalf("expected battery.update on survivor, got %s", ev.Type) + } + + // Closing the superseded conn is silent; closing the current fires disconnect. + if err := connA.Close(); err != nil { + t.Fatalf("close A: %v", err) + } + expectNoFor(t, evCh, "mock-peer", 300*time.Millisecond, "superseded disconnect") + if err := connC.Close(); err != nil { + t.Fatalf("close C: %v", err) + } + if ev := nextFor(t, evCh, "mock-peer", "device.disconnected"); ev.Type != events.TypeDeviceDisconnected { + t.Fatalf("expected device.disconnected, got %s", ev.Type) + } +}