From c9c2f2295895026faf67c1ccbf83d8a8defc3131 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 9 Jul 2026 14:58:46 -0500 Subject: [PATCH 1/2] fix: bad_credentials after 150s of inactivity --- pkg/connector/sync.go | 46 +++++++++++++++ pkg/connector/sync_test.go | 114 +++++++++++++++++++++++++++++++++++++ pkg/line/sse.go | 48 +++++++++++++++- pkg/line/sse_test.go | 35 ++++++++++++ 4 files changed, 241 insertions(+), 2 deletions(-) diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index e03eb3b..a30a315 100644 --- a/pkg/connector/sync.go +++ b/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, err) { + return + } + time.Sleep(sseReconnectDelay) + continue + } + if lc.isLoggedOut(err) { lc.markLoggedOutByOtherClient(ctx, err) return @@ -1258,6 +1266,44 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } } +// handleReceiveIdleTimeout handles /operation/receive streams that stop +// producing ping/event bytes. Chrome advertises a 140s polling timeout, so an +// idle timeout here means the receive path is no longer healthy. Probe Talk +// auth immediately so forced logouts surface before the next send/read receipt. +func (lc *LineClient) handleReceiveIdleTimeout(ctx context.Context, _ error) bool { + _, 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. diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index cccd9cf..9fe5481 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -114,6 +114,120 @@ 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++ + return 1234, 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() { diff --git a/pkg/line/sse.go b/pkg/line/sse.go index d7373e9..c98ce16 100644 --- a/pkg/line/sse.go +++ b/pkg/line/sse.go @@ -3,19 +3,32 @@ package line import ( "bufio" "context" + "errors" "fmt" "io" "net/http" "net/url" "strconv" "strings" + "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{} -const maxSSEErrorBodyBytes = 4096 +var ErrSSEIdleTimeout = errors.New("SSE idle timeout") + +const ( + maxSSEErrorBodyBytes = 4096 + 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 { @@ -77,7 +90,7 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev } for { - lineBytes, err := reader.ReadBytes('\n') + lineBytes, err := readSSELine(ctx, resp.Body, reader) if err != nil { return err } @@ -106,3 +119,34 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev } } } + +type sseReadResult struct { + line []byte + err error +} + +func readSSELine(ctx context.Context, body io.Closer, reader *bufio.Reader) ([]byte, error) { + if sseIdleTimeout <= 0 { + return reader.ReadBytes('\n') + } + + resultCh := make(chan sseReadResult, 1) + go func() { + lineBytes, err := reader.ReadBytes('\n') + resultCh <- sseReadResult{line: lineBytes, err: err} + }() + + timer := time.NewTimer(sseIdleTimeout) + defer timer.Stop() + + select { + case result := <-resultCh: + return result.line, result.err + case <-timer.C: + _ = body.Close() + return nil, fmt.Errorf("%w after %s", ErrSSEIdleTimeout, sseIdleTimeout) + case <-ctx.Done(): + _ = body.Close() + return nil, ctx.Err() + } +} diff --git a/pkg/line/sse_test.go b/pkg/line/sse_test.go index 20d556d..7827f53 100644 --- a/pkg/line/sse_test.go +++ b/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) + } +} From 010bbe9789f13093663579beeaa2c50dd55b2548 Mon Sep 17 00:00:00 2001 From: highesttt Date: Thu, 9 Jul 2026 15:24:33 -0500 Subject: [PATCH 2/2] fix: address receive idle review --- pkg/connector/sync.go | 11 ++++--- pkg/connector/sync_test.go | 7 +++- pkg/line/sse.go | 66 +++++++++++++++++--------------------- 3 files changed, 42 insertions(+), 42 deletions(-) diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index a30a315..2d4f3b6 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1243,7 +1243,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("SSE Disconnected") if line.IsSSEIdleTimeout(err) { - if lc.handleReceiveIdleTimeout(ctx, err) { + if lc.handleReceiveIdleTimeout(ctx) { return } time.Sleep(sseReconnectDelay) @@ -1267,10 +1267,11 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } // handleReceiveIdleTimeout handles /operation/receive streams that stop -// producing ping/event bytes. Chrome advertises a 140s polling timeout, so an -// idle timeout here means the receive path is no longer healthy. Probe Talk -// auth immediately so forced logouts surface before the next send/read receipt. -func (lc *LineClient) handleReceiveIdleTimeout(ctx context.Context, _ error) bool { +// 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 diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 9fe5481..31251f1 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -181,7 +181,12 @@ func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) { t.Fatalf("revision probe token = %q, want valid", client.AccessToken) } revisionCalls++ - return 1234, nil + 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 diff --git a/pkg/line/sse.go b/pkg/line/sse.go index c98ce16..faf839d 100644 --- a/pkg/line/sse.go +++ b/pkg/line/sse.go @@ -10,6 +10,7 @@ import ( "net/url" "strconv" "strings" + "sync/atomic" "time" ) @@ -19,10 +20,11 @@ var sseHTTPClient = &http.Client{} var ErrSSEIdleTimeout = errors.New("SSE idle timeout") -const ( - maxSSEErrorBodyBytes = 4096 - defaultSSEIdleTimeout = 150 * time.Second -) +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 @@ -69,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 @@ -90,10 +106,19 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev } for { - lineBytes, err := readSSELine(ctx, resp.Body, reader) + 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 == "" { @@ -119,34 +144,3 @@ func (c *Client) ListenSSE(ctx context.Context, localRev int64, callback func(ev } } } - -type sseReadResult struct { - line []byte - err error -} - -func readSSELine(ctx context.Context, body io.Closer, reader *bufio.Reader) ([]byte, error) { - if sseIdleTimeout <= 0 { - return reader.ReadBytes('\n') - } - - resultCh := make(chan sseReadResult, 1) - go func() { - lineBytes, err := reader.ReadBytes('\n') - resultCh <- sseReadResult{line: lineBytes, err: err} - }() - - timer := time.NewTimer(sseIdleTimeout) - defer timer.Stop() - - select { - case result := <-resultCh: - return result.line, result.err - case <-timer.C: - _ = body.Close() - return nil, fmt.Errorf("%w after %s", ErrSSEIdleTimeout, sseIdleTimeout) - case <-ctx.Done(): - _ = body.Close() - return nil, ctx.Err() - } -}