diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index e03eb3b..2d4f3b6 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) { + 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. diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index cccd9cf..31251f1 100644 --- a/pkg/connector/sync_test.go +++ b/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() { diff --git a/pkg/line/sse.go b/pkg/line/sse.go index d7373e9..faf839d 100644 --- a/pkg/line/sse.go +++ b/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 == "" { 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) + } +}