fix: bad_credentials after 150s of inactivity

pkg/connector/sync.go

@@ -1242,6 +1242,14 @@ func (lc *LineClient) pollLoop(ctx context.Context) {
 			if err.Error() != "EOF" {
 				lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("SSE Disconnected")
 
+				if line.IsSSEIdleTimeout(err) {
+					if lc.handleReceiveIdleTimeout(ctx) {
+						return
+					}
+					time.Sleep(sseReconnectDelay)
+					continue
+				}
+
 				if lc.isLoggedOut(err) {
 					lc.markLoggedOutByOtherClient(ctx, err)
 					return
@@ -1258,6 +1266,45 @@ func (lc *LineClient) pollLoop(ctx context.Context) {
 	}
 }
 
+// handleReceiveIdleTimeout handles /operation/receive streams that stop
+// producing ping/event bytes. Probe Talk auth immediately so forced logouts
+// surface before the next send/read receipt.
+func (lc *LineClient) handleReceiveIdleTimeout(ctx context.Context) bool {
+	// This is only a health probe. Keep localRev unchanged so the reconnected
+	// stream replays operations that arrived while the old stream was stalled.
+	_, probeErr := getLastOpRevisionWithClient(lc.newClient())
+	if probeErr == nil {
+		return false
+	}
+
+	if lc.isLoggedOut(probeErr) {
+		lc.markLoggedOutByOtherClient(ctx, probeErr)
+		return true
+	}
+
+	if line.IsUnauthorizedStatus(probeErr) {
+		return lc.handleReceiveAuthError(ctx, probeErr)
+	}
+
+	if lc.shouldAttemptTokenRecovery(ctx, probeErr) {
+		if errRecover := lc.recoverToken(ctx); errRecover != nil {
+			if errors.Is(errRecover, errLineSessionInvalidated) || lc.isLoggedOut(errRecover) {
+				lc.markLoggedOutByOtherClient(ctx, errRecover)
+				return true
+			}
+			if lc.UserLogin != nil && lc.UserLogin.Bridge != nil {
+				lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token after receive idle probe")
+			}
+		}
+		return false
+	}
+
+	if lc.UserLogin != nil && lc.UserLogin.Bridge != nil {
+		lc.UserLogin.Bridge.Log.Warn().Err(probeErr).Msg("Receive idle auth probe failed; reconnecting SSE")
+	}
+	return false
+}
+
 // handleReceiveAuthError handles auth failures from /operation/receive. The
 // receive endpoint may return only a bare 401/403, so probe getProfile to reveal
 // the detailed forced-logout envelope before deciding whether recovery is safe.

pkg/connector/sync_test.go

@@ -114,6 +114,125 @@ func TestPollLoopMarksLoggedOutWhenReceiveAuthFails(t *testing.T) {
 	}
 }
 
+func TestPollLoopMarksLoggedOutWhenReceiveIdleProbeFailsLoggedOut(t *testing.T) {
+	oldGetLastOpRevision := getLastOpRevisionWithClient
+	oldListenSSE := listenSSEWithClient
+	t.Cleanup(func() {
+		getLastOpRevisionWithClient = oldGetLastOpRevision
+		listenSSEWithClient = oldListenSSE
+	})
+
+	var revisionCalls int
+	getLastOpRevisionWithClient = func(client *line.Client) (int64, error) {
+		if client.AccessToken != "stale" {
+			t.Fatalf("revision probe token = %q, want stale", client.AccessToken)
+		}
+		revisionCalls++
+		if revisionCalls == 1 {
+			return 1234, nil
+		}
+		return 0, errLoggedOut
+	}
+
+	listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error {
+		if client.AccessToken != "stale" {
+			t.Fatalf("SSE client token = %q, want stale", client.AccessToken)
+		}
+		if localRev != 1234 {
+			t.Fatalf("localRev = %d, want 1234", localRev)
+		}
+		return line.ErrSSEIdleTimeout
+	}
+
+	lc := &LineClient{
+		AccessToken: "stale",
+		UserLogin: &bridgev2.UserLogin{
+			Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)},
+		},
+	}
+
+	lc.wg.Add(1)
+	lc.pollLoop(context.Background())
+
+	if revisionCalls != 2 {
+		t.Fatalf("revision calls = %d, want 2", revisionCalls)
+	}
+	if lc.hasAccessToken() {
+		t.Fatal("access token was not invalidated after receive idle logout")
+	}
+	if !lc.isSessionInvalidated() {
+		t.Fatal("session was not marked invalidated after receive idle logout")
+	}
+}
+
+func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) {
+	oldGetLastOpRevision := getLastOpRevisionWithClient
+	oldListenSSE := listenSSEWithClient
+	oldReconnectDelay := sseReconnectDelay
+	t.Cleanup(func() {
+		getLastOpRevisionWithClient = oldGetLastOpRevision
+		listenSSEWithClient = oldListenSSE
+		sseReconnectDelay = oldReconnectDelay
+	})
+
+	var revisionCalls int
+	getLastOpRevisionWithClient = func(client *line.Client) (int64, error) {
+		if client.AccessToken != "valid" {
+			t.Fatalf("revision probe token = %q, want valid", client.AccessToken)
+		}
+		revisionCalls++
+		if revisionCalls == 1 {
+			return 1234, nil
+		}
+		// The health probe sees a newer server revision, but reconnecting from it
+		// would skip operations that arrived while the SSE stream was stalled.
+		return 5678, nil
+	}
+	sseReconnectDelay = time.Millisecond
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var listenCalls int
+	listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error {
+		if client.AccessToken != "valid" {
+			t.Fatalf("SSE client token = %q, want valid", client.AccessToken)
+		}
+		if localRev != 1234 {
+			t.Fatalf("localRev = %d, want 1234", localRev)
+		}
+		listenCalls++
+		if listenCalls == 1 {
+			return line.ErrSSEIdleTimeout
+		}
+		cancel()
+		return context.Canceled
+	}
+
+	lc := &LineClient{
+		AccessToken: "valid",
+		UserLogin: &bridgev2.UserLogin{
+			Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)},
+		},
+	}
+
+	lc.wg.Add(1)
+	lc.pollLoop(ctx)
+
+	if revisionCalls != 2 {
+		t.Fatalf("revision calls = %d, want 2", revisionCalls)
+	}
+	if listenCalls != 2 {
+		t.Fatalf("SSE attempts = %d, want 2", listenCalls)
+	}
+	if !lc.hasAccessToken() {
+		t.Fatal("valid access token was invalidated")
+	}
+	if lc.isSessionInvalidated() {
+		t.Fatal("valid session was marked invalidated")
+	}
+}
+
 func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) {
 	oldGetProfile := getProfileWithToken
 	t.Cleanup(func() {

pkg/line/sse.go

@@ -3,20 +3,35 @@ package line
 import (
 	"bufio"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
 	"strconv"
 	"strings"
+	"sync/atomic"
+	"time"
 )
 
 // sseHTTPClient is a dedicated HTTP client for SSE connections with no timeout.
 // Reused across reconnects to avoid allocating a new transport pool each time.
 var sseHTTPClient = &http.Client{}
 
+var ErrSSEIdleTimeout = errors.New("SSE idle timeout")
+
 const maxSSEErrorBodyBytes = 4096
 
+// LINE advertises a 140s polling timeout. The extra 10s lets a normal
+// server-driven close arrive before the bridge declares the stream idle.
+const defaultSSEIdleTimeout = 150 * time.Second
+
+var sseIdleTimeout = defaultSSEIdleTimeout
+
+func IsSSEIdleTimeout(err error) bool {
+	return errors.Is(err, ErrSSEIdleTimeout)
+}
+
 // ListenSSE connects to the Event Stream and blocks
 func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(event, data string)) error {
 	q := url.Values{}
@@ -56,6 +71,20 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev
 	}
 
 	reader := bufio.NewReader(resp.Body)
+	idleTimeout := sseIdleTimeout
+	var idleTimedOut atomic.Bool
+	var idleTimer *time.Timer
+	if idleTimeout > 0 {
+		idleTimer = time.AfterFunc(idleTimeout, func() {
+			idleTimedOut.Store(true)
+			_ = resp.Body.Close()
+		})
+		defer idleTimer.Stop()
+	}
+	stopContextClose := context.AfterFunc(ctx, func() {
+		_ = resp.Body.Close()
+	})
+	defer stopContextClose()
 
 	var currentEvent string
 	var dataLines []string
@@ -79,8 +108,17 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev
 	for {
 		lineBytes, err := reader.ReadBytes('\n')
 		if err != nil {
+			if idleTimedOut.Load() {
+				return fmt.Errorf("%w after %s", ErrSSEIdleTimeout, idleTimeout)
+			}
+			if ctxErr := ctx.Err(); ctxErr != nil {
+				return ctxErr
+			}
 			return err
 		}
+		if idleTimer != nil {
+			idleTimer.Reset(idleTimeout)
+		}
 
 		line := strings.TrimRight(string(lineBytes), "\r\n")
 		if line == "" {

pkg/line/sse_test.go

@@ -2,10 +2,12 @@ package line
 
 import (
 	"context"
+	"errors"
 	"io"
 	"net/http"
 	"strings"
 	"testing"
+	"time"
 )
 
 func TestListenSSENonOKIncludesResponseBody(t *testing.T) {
@@ -37,3 +39,36 @@ func TestListenSSENonOKIncludesResponseBody(t *testing.T) {
 		t.Fatalf("err = %v, want response body detail", err)
 	}
 }
+
+func TestListenSSEIdleTimeout(t *testing.T) {
+	oldClient := sseHTTPClient
+	oldTimeout := sseIdleTimeout
+	t.Cleanup(func() {
+		sseHTTPClient = oldClient
+		sseIdleTimeout = oldTimeout
+	})
+
+	sseIdleTimeout = 10 * time.Millisecond
+	pr, pw := io.Pipe()
+	t.Cleanup(func() {
+		_ = pw.Close()
+	})
+
+	sseHTTPClient = &http.Client{
+		Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusOK,
+				Header:     make(http.Header),
+				Body:       pr,
+			}, nil
+		}),
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	err := NewClient("stale-token").ListenSSE(ctx, 0, func(event, data string) {})
+	if !errors.Is(err, ErrSSEIdleTimeout) {
+		t.Fatalf("err = %v, want ErrSSEIdleTimeout", err)
+	}
+}