From 8c50668a2ec9df354dea7e669b012e9f61089d7f Mon Sep 17 00:00:00 2001 From: Nick Mills-Barrett Date: Wed, 28 Oct 2026 16:13:32 +0000 Subject: [PATCH 1/4] connector: retry transient token recovery failures instead of logging out A failed token recovery used to send BAD_CREDENTIALS "logged out by another client" for any error, keep the dead token so IsLoggedIn stayed true, and stop the poll loop for good, while every later send re-ran a password login. Network, 5xx and rate-limit failures now report TRANSIENT_DISCONNECT and the poll loop keeps retrying with capped, jittered backoff, and recovery itself backs off so sends no longer trigger a password login each. Only a rejected refresh and re-login invalidates the session, clearing the token and sending line-token-expired, which is persisted so a restart reports the same state. Idle SSE timeouts and a startup token check that fails transiently now report TRANSIENT_DISCONNECT, and CONNECTED is sent once the stream opens. Co-Authored-By: Claude Opus 5.5 --- pkg/connector/auth_recovery.go | 27 +++++ pkg/connector/auth_recovery_test.go | 126 +++++++++++++++++++++- pkg/connector/client.go | 91 ++++++++++------ pkg/connector/connector.go | 1 + pkg/connector/forced_logout_test.go | 19 +++- pkg/connector/phone_notifications_test.go | 3 +- pkg/connector/sync.go | 57 +++++----- pkg/connector/sync_test.go | 52 +++++++-- 8 files changed, 296 insertions(+), 80 deletions(-) diff --git a/pkg/connector/auth_recovery.go b/pkg/connector/auth_recovery.go index dc08999c..26e39839 100644 --- a/pkg/connector/auth_recovery.go +++ b/pkg/connector/auth_recovery.go @@ -2,7 +2,10 @@ package connector import ( "context" + "errors" "fmt" + "net" + "net/http" "github.com/beeper/line/pkg/line" ) @@ -65,6 +68,27 @@ func callLineWithRecovery[T any](ctx context.Context, client *line.Client, deps return client, res, err } +func isTransientLineError(err error) bool { + // An unanswered PIN prompt times out like a network request, but retrying + // would only send the user another prompt. + if errors.Is(err, errLinePINNotVerified) { + return false + } + var netErr net.Error + if errors.As(err, &netErr) { + return true + } + statusCode := line.HTTPStatusCode(err) + return statusCode >= http.StatusInternalServerError || statusCode == http.StatusTooManyRequests +} + +func isRecoveryRejected(err error) bool { + return !isTransientLineError(err) && + !errors.Is(err, context.Canceled) && + !errors.Is(err, errLineSessionInvalidated) && + !errors.Is(err, errLineClientSuperseded) +} + func (lc *LineClient) isTokenError(err error) bool { if line.IsNoUsableE2EEGroupKey(err) || line.IsNoUsableE2EEPublicKey(err) { return false @@ -139,6 +163,9 @@ func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedCli // case another serialized recovery rotated that token first. return lc.recoverClientAfterAuthError(ctx, newLineAPIClient(recoveryToken), errRecover) } + if ctx.Err() == nil && isRecoveryRejected(errRecover) { + lc.markRecoveryRejected(ctx, recoveryToken, errRecover) + } return nil, errRecover } if ctx.Err() != nil { diff --git a/pkg/connector/auth_recovery_test.go b/pkg/connector/auth_recovery_test.go index d74affc0..b42c93cf 100644 --- a/pkg/connector/auth_recovery_test.go +++ b/pkg/connector/auth_recovery_test.go @@ -3,6 +3,9 @@ package connector import ( "context" "errors" + "fmt" + "net/url" + "os" "strings" "sync" "sync/atomic" @@ -13,13 +16,126 @@ import ( ) var ( - errAuthRequired = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":119,"reason":"Access token refresh required"}}`) - errLoggedOut = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":8,"reason":"V3_TOKEN_CLIENT_LOGGED_OUT"}}`) - errSenderKey = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":83,"reason":"invalid sender key"}}`) - errNotMember = errors.New(`API error 400: {"code":10051,"data":{"name":"TalkException","code":10,"reason":"not a member"}}`) - errNetwork = errors.New("request failed: dial tcp: i/o timeout") + errAuthRequired = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":119,"reason":"Access token refresh required"}}`) + errLoggedOut = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":8,"reason":"V3_TOKEN_CLIENT_LOGGED_OUT"}}`) + errSenderKey = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":83,"reason":"invalid sender key"}}`) + errNotMember = errors.New(`API error 400: {"code":10051,"data":{"name":"TalkException","code":10,"reason":"not a member"}}`) + errNetwork = errors.New("request failed: dial tcp: i/o timeout") + errRecoveryTimeout = fmt.Errorf("request failed: %w", &url.Error{Op: "Post", URL: "https://line-chrome-gw.line-apps.com/api/auth/tokenRefresh", Err: os.ErrDeadlineExceeded}) + errLoginRejected = errors.New("login failed: loginV2 failed: invalid password (code 1)") ) +func TestIsTransientLineError(t *testing.T) { + tests := []struct { + name string + err error + want bool + }{ + {name: "network timeout", err: errRecoveryTimeout, want: true}, + {name: "server error", err: &line.HTTPError{StatusCode: 503}, want: true}, + {name: "rate limited", err: &line.HTTPError{StatusCode: 429}, want: true}, + {name: "backed off network failure", err: fmt.Errorf("waiting to retry LINE token recovery: %w", errRecoveryTimeout), want: true}, + {name: "unauthorized", err: &line.HTTPError{StatusCode: 401}, want: false}, + {name: "login rejected", err: errLoginRejected, want: false}, + {name: "unanswered PIN prompt", err: fmt.Errorf("%w: %w", errLinePINNotVerified, errRecoveryTimeout), want: false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := isTransientLineError(tt.err); got != tt.want { + t.Fatalf("isTransientLineError(%v) = %v, want %v", tt.err, got, tt.want) + } + }) + } +} + +func TestRecoveryFailureInvalidatesOnlyRejectedSessions(t *testing.T) { + oldRecover := recoverLineToken + t.Cleanup(func() { + recoverLineToken = oldRecover + }) + + tests := []struct { + name string + recoverErr error + wantInvalidated bool + }{ + {name: "transient failure keeps the session", recoverErr: errRecoveryTimeout}, + {name: "rejected credentials invalidate the session", recoverErr: errLoginRejected, wantInvalidated: true}, + {name: "unanswered PIN prompt invalidates the session", recoverErr: fmt.Errorf("%w: %w", errLinePINNotVerified, errRecoveryTimeout), wantInvalidated: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + recoverLineToken = func(*LineClient, context.Context) error { + return tt.recoverErr + } + lc := &LineClient{AccessToken: "current-token"} + retryClient, err := lc.recoverClientAfterAuthError(context.Background(), line.NewClient("current-token"), errAuthRequired) + if !errors.Is(err, tt.recoverErr) { + t.Fatalf("recoverClientAfterAuthError error = %v, want %v", err, tt.recoverErr) + } + if retryClient != nil { + t.Fatalf("retry client = %#v, want nil", retryClient) + } + if lc.isSessionInvalidated() != tt.wantInvalidated || lc.IsLoggedIn() == tt.wantInvalidated { + t.Fatalf("invalidated = %v, logged in = %v, want invalidated = %v", lc.isSessionInvalidated(), lc.IsLoggedIn(), tt.wantInvalidated) + } + }) + } +} + +func TestRunTokenRecoveryBacksOffAfterFailure(t *testing.T) { + lc := &LineClient{} + var calls int + recover := func(context.Context) error { + calls++ + if calls == 1 { + return errRecoveryTimeout + } + return nil + } + + if err := lc.runTokenRecovery(context.Background(), recover); !errors.Is(err, errRecoveryTimeout) { + t.Fatalf("first recovery error = %v, want %v", err, errRecoveryTimeout) + } + lc.recoverRetryAt = time.Now().Add(time.Minute) + if err := lc.runTokenRecovery(context.Background(), recover); !errors.Is(err, errRecoveryTimeout) { + t.Fatalf("backed off recovery error = %v, want %v", err, errRecoveryTimeout) + } + if calls != 1 { + t.Fatalf("recovery calls during backoff = %d, want 1", calls) + } + + lc.recoverRetryAt = time.Now() + if err := lc.runTokenRecovery(context.Background(), recover); err != nil { + t.Fatalf("recovery after backoff returned error: %v", err) + } + if calls != 2 { + t.Fatalf("recovery calls after backoff = %d, want 2", calls) + } + if lc.recoverBackoff.delay != 0 || !lc.recoverRetryAt.IsZero() { + t.Fatal("successful recovery did not reset the backoff") + } +} + +func TestRecoverTokenDoesNotReloginAfterTransientRefreshFailure(t *testing.T) { + lc := &LineClient{} + var reloginCalls int + err := lc.recoverTokenWith( + context.Background(), + func(context.Context) error { return errRecoveryTimeout }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + if !errors.Is(err, errRecoveryTimeout) { + t.Fatalf("recoverTokenWith error = %v, want %v", err, errRecoveryTimeout) + } + if reloginCalls != 0 { + t.Fatalf("relogin calls = %d, want 0", reloginCalls) + } +} + func TestCallLineWithRecovery(t *testing.T) { tests := []struct { name string diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 4369df38..58596512 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -24,6 +24,7 @@ var ( errLineSessionInvalidated = errors.New("LINE session invalidated by another client") errLineClientSuperseded = errors.New("LINE client was superseded") errLineQRLoginRequired = errors.New("LINE requires a new QR scan. Reconnect in Beeper to continue") + errLinePINNotVerified = errors.New("PIN verification failed") ) func init() { @@ -81,6 +82,9 @@ type LineClient struct { tokenMu sync.RWMutex recoverMu sync.Mutex recoverTime time.Time + recoverErr error + recoverRetryAt time.Time + recoverBackoff reconnectBackoff recoveryStopped bool missingE2EEKeyMu sync.Mutex missingE2EEKeyMarked bool @@ -356,10 +360,24 @@ func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) lc.markLoggedOutByOtherClientLocked(ctx, err) } -// markLoggedOutByOtherClientLocked applies the persistent forced-logout -// transition. The caller must hold recoverMu so token rotation and invalidation -// cannot interleave. func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err error) { + lc.invalidateSessionLocked(ctx, err, false) +} + +// markRecoveryRejected invalidates the session unless another recovery already +// replaced the token the rejected recovery started from. +func (lc *LineClient) markRecoveryRejected(ctx context.Context, recoveryToken string, err error) { + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + if lc.isSessionInvalidated() || lc.getAccessToken() != recoveryToken { + return + } + lc.invalidateSessionLocked(ctx, err, true) +} + +// invalidateSessionLocked applies the persistent forced-logout transition. The +// caller must hold recoverMu so token rotation and invalidation cannot interleave. +func (lc *LineClient) invalidateSessionLocked(ctx context.Context, err error, expired bool) { if lc.UserLogin == nil { lc.invalidateAccessToken() lc.obsTokenCache.Invalidate() @@ -380,6 +398,9 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err lc.obsTokenCache.Invalidate() stateError := status.BridgeStateErrorCode("line-logged-out") stateMessage := "LINE logged this Chrome Extension session out because another LINE client connected. Click Reconnect in Beeper to reconnect LINE." + if expired { + stateError, stateMessage = "line-token-expired", "" + } invalidSenderKey := line.IsInvalidSenderKey(err) needsFullE2EELogin := invalidSenderKey lc.missingE2EEKeyMu.Lock() @@ -397,7 +418,7 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err stateError = "line-e2ee-key-missing" stateMessage = lineMissingE2EEKeyMessage } - lc.saveSessionInvalidated(ctx) + lc.saveSessionInvalidated(ctx, expired) if sendState && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("LINE session invalidated; marking login bad credentials") } @@ -407,6 +428,7 @@ func (lc *LineClient) markLoggedOutByOtherClientLocked(ctx context.Context, err Error: stateError, Message: stateMessage, UserAction: status.UserActionRelogin, + Info: map[string]any{"go_error": err.Error()}, }) } lc.cancelActiveRun() @@ -459,7 +481,7 @@ func (lc *LineClient) sendSSEDisconnectedIfCurrent(ctx context.Context, err erro return true } -func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { +func (lc *LineClient) saveSessionInvalidated(ctx context.Context, expired bool) { if lc.UserLogin == nil || lc.UserLogin.UserLogin == nil { return } @@ -469,6 +491,7 @@ func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { } meta.AccessToken = "" meta.SessionInvalidated = true + meta.SessionExpired = expired if err := lc.UserLogin.Save(ctx); err != nil && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to save LINE session invalidation") } @@ -518,7 +541,6 @@ func (lc *LineClient) applyRefreshedLoginE2EEKeys(meta *UserLoginMetadata, res * } // recoverToken attempts to restore a valid session by refreshing, then re-logging in. -// Returns nil on success. On failure the caller should send StateBadCredentials. func (lc *LineClient) recoverToken(ctx context.Context) error { return lc.recoverTokenWith(ctx, lc.refreshAndSave, lc.tryLogin) } @@ -537,7 +559,7 @@ func (lc *LineClient) recoverTokenWith( return nil } else if ctx.Err() != nil { return ctx.Err() - } else if lc.isLoggedOut(err) || errors.Is(err, errLineSessionInvalidated) || errors.Is(err, errLineClientSuperseded) || errors.Is(err, context.Canceled) { + } else if lc.isLoggedOut(err) || isTransientLineError(err) || errors.Is(err, errLineSessionInvalidated) || errors.Is(err, errLineClientSuperseded) || errors.Is(err, context.Canceled) { return err } lc.UserLogin.Bridge.Log.Info().Msg("Refresh failed, attempting re-login with stored credentials...") @@ -566,8 +588,15 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context if !lc.recoverTime.IsZero() && time.Since(lc.recoverTime) < recentTokenRecoveryWindow { return nil } + if time.Now().Before(lc.recoverRetryAt) { + return fmt.Errorf("waiting to retry LINE token recovery: %w", lc.recoverErr) + } if err := recover(ctx); err != nil { + if ctx.Err() == nil { + lc.recoverErr = err + lc.recoverRetryAt = time.Now().Add(lc.recoverBackoff.next()) + } return err } if ctx.Err() != nil { @@ -579,6 +608,9 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context if lc.isSessionInvalidated() { return errLineSessionInvalidated } + lc.recoverBackoff.reset() + lc.recoverErr = nil + lc.recoverRetryAt = time.Time{} lc.recoverTime = time.Now() return nil } @@ -629,7 +661,10 @@ func (lc *LineClient) Connect(ctx context.Context) { } if !lc.hasAccessToken() { if lc.isSessionInvalidated() { - lc.markLoggedOutByOtherClient(ctx, errLineSessionInvalidated) + meta, _ := lc.UserLogin.Metadata.(*UserLoginMetadata) + lc.recoverMu.Lock() + lc.invalidateSessionLocked(ctx, errLineSessionInvalidated, meta != nil && meta.SessionExpired) + lc.recoverMu.Unlock() return } if err := lc.runTokenRecovery(ctx, lc.tryLogin); err != nil { @@ -651,21 +686,15 @@ func (lc *LineClient) Connect(ctx context.Context) { } // Verify the token is still valid before proceeding - if err := lc.ensureValidToken(ctx); err != nil { - if ctx.Err() != nil { - return + tokenErr := lc.ensureValidToken(ctx) + if tokenErr != nil { + if ctx.Err() == nil && lc.isLoggedOut(tokenErr) { + lc.markLoggedOutByOtherClient(ctx, tokenErr) } - if lc.isLoggedOut(err) { - lc.markLoggedOutByOtherClient(ctx, err) + if ctx.Err() != nil || lc.isSessionInvalidated() { return } - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-token-expired", - Info: map[string]any{"go_error": err.Error()}, - UserAction: status.UserActionRelogin, - }) - return + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, continuing so the poll loop can retry") } // LINE Chrome disables mobile notifications while its session is active by // default. Explicitly clear that setting once per bridge connection so the @@ -674,7 +703,11 @@ func (lc *LineClient) Connect(ctx context.Context) { if ctx.Err() != nil { return } - if !lc.sendConnectedStateIfCurrent(ctx) { + if tokenErr != nil { + if !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { + return + } + } else if !lc.sendConnectedStateIfCurrent(ctx) { return } @@ -738,7 +771,7 @@ func (lc *LineClient) Connect(ctx context.Context) { lc.wg.Add(3) go lc.syncDMChats(ctx) go lc.prefetchMessages(ctx) - go lc.pollLoop(ctx) + go lc.pollLoop(ctx, tokenErr != nil) workersStarted = true } @@ -801,7 +834,7 @@ func (lc *LineClient) tryLogin(ctx context.Context) error { waitClient.LoginAttempt = res.Attempt waitRes, err := waitClient.WaitForLogin(res.Verifier, res.NoE2EE) if err != nil { - return fmt.Errorf("PIN verification failed: %w", err) + return fmt.Errorf("%w: %w", errLinePINNotVerified, err) } if waitRes.AuthToken == "" { return fmt.Errorf("PIN verification completed but no auth token received") @@ -976,17 +1009,7 @@ func (lc *LineClient) ensureValidToken(ctx context.Context) error { _, _, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { return getProfileWithToken(ctx, client.AccessToken) }) - if err == nil { - return nil - } - if ctx.Err() != nil { - return ctx.Err() - } - if line.IsAuthError(err) { - return err - } - lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("GetProfile failed with non-auth error, continuing anyway") - return nil + return err } func (lc *LineClient) Disconnect() { diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index ccf3e06f..037f908b 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -130,6 +130,7 @@ type UserLoginMetadata struct { AccessToken string `json:"access_token"` RefreshToken string `json:"refresh_token,omitempty"` SessionInvalidated bool `json:"session_invalidated,omitempty"` + SessionExpired bool `json:"session_expired,omitempty"` Email string `json:"email,omitempty"` Password string `json:"password,omitempty"` Certificate string `json:"certificate,omitempty"` diff --git a/pkg/connector/forced_logout_test.go b/pkg/connector/forced_logout_test.go index efe123a3..24463ea0 100644 --- a/pkg/connector/forced_logout_test.go +++ b/pkg/connector/forced_logout_test.go @@ -2,6 +2,7 @@ package connector import ( "context" + "errors" "io" "path/filepath" "testing" @@ -20,7 +21,7 @@ import ( ) func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { - for _, mode := range []string{"ordinary", "invalid-key", "late-invalid-key", "late-stale-key"} { + for _, mode := range []string{"ordinary", "invalid-key", "late-invalid-key", "late-stale-key", "rejected-recovery"} { t.Run(mode, func(t *testing.T) { invalidKey := mode == "invalid-key" || mode == "late-invalid-key" ctx := context.Background() @@ -57,6 +58,13 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { if mode == "invalid-key" { failure = errSenderKey } + var wantRecoverErr error + if mode == "rejected-recovery" { + failure, wantRecoverErr = errAuthRequired, errLoginRejected + oldRecover := recoverLineToken + t.Cleanup(func() { recoverLineToken = oldRecover }) + recoverLineToken = func(*LineClient, context.Context) error { return errLoginRejected } + } recovered := make(chan error, 1) if mode == "invalid-key" { lc.missingE2EEKeyMu.Lock() @@ -74,8 +82,8 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { } lc.missingE2EEKeyMu.Unlock() } - if err := <-recovered; err != nil { - t.Fatal(err) + if err := <-recovered; !errors.Is(err, wantRecoverErr) { + t.Fatalf("recovery error = %v, want %v", err, wantRecoverErr) } if mode == "late-invalid-key" || mode == "late-stale-key" { if !lc.isTokenError(errSenderKey) { @@ -94,6 +102,8 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { wantStateError := status.BridgeStateErrorCode("line-logged-out") if mode == "invalid-key" { wantStateError = "line-e2ee-key-missing" + } else if mode == "rejected-recovery" { + wantStateError = "line-token-expired" } select { case state := <-matrix.states: @@ -134,6 +144,9 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { if invalidKey { wantRestartError = "line-e2ee-key-missing" wantRestartMessage = lineMissingE2EEKeyMessage + } else if mode == "rejected-recovery" { + wantRestartError = "line-token-expired" + wantRestartMessage = status.BridgeStateHumanErrors["line-token-expired"] } if state.StateEvent != status.StateBadCredentials || state.Error != wantRestartError || state.Message != wantRestartMessage || state.UserAction != status.UserActionRelogin { t.Fatalf("unexpected state after bridge restart: %v", state) diff --git a/pkg/connector/phone_notifications_test.go b/pkg/connector/phone_notifications_test.go index 728befd7..8d91505c 100644 --- a/pkg/connector/phone_notifications_test.go +++ b/pkg/connector/phone_notifications_test.go @@ -136,11 +136,10 @@ func TestConfigurePhoneNotificationsTreatsAuthRecoveryFailureAsBestEffort(t *tes ) error { return errAuthRequired } - recoveryErr := errors.New("stored credentials rejected") var recoveries int recoverLineToken = func(*LineClient, context.Context) error { recoveries++ - return recoveryErr + return errRecoveryTimeout } ctx := context.Background() diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index c74ad6c8..3df605f3 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "io" + "math/rand/v2" "slices" "sort" "strconv" @@ -21,7 +22,6 @@ import ( "maunium.net/go/mautrix/bridgev2/database" "maunium.net/go/mautrix/bridgev2/networkid" "maunium.net/go/mautrix/bridgev2/simplevent" - "maunium.net/go/mautrix/bridgev2/status" "maunium.net/go/mautrix/event" "github.com/beeper/line/pkg/line" @@ -38,6 +38,7 @@ const ( groupPortalCreateWait = 30 * time.Second beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline" defaultReceiveAuthProbeInterval = 150 * time.Second + maxReconnectDelay = 5 * time.Minute ) var ( @@ -1590,11 +1591,25 @@ func isReceiveAuthProbeDue(receiveCtx context.Context, nextProbeAt time.Time) bo return errors.Is(context.Cause(receiveCtx), errReceiveAuthProbeDue) || !receiveAuthProbeNow().Before(nextProbeAt) } -func waitForSSEReconnect(ctx context.Context) bool { - if sseReconnectDelay <= 0 { +type reconnectBackoff struct { + delay time.Duration +} + +func (b *reconnectBackoff) next() time.Duration { + b.delay = min(max(b.delay*2, sseReconnectDelay), maxReconnectDelay) + half := b.delay / 2 + return half + rand.N(half+1) +} + +func (b *reconnectBackoff) reset() { + b.delay = 0 +} + +func waitForSSEReconnect(ctx context.Context, delay time.Duration) bool { + if delay <= 0 { return ctx.Err() == nil } - timer := time.NewTimer(sseReconnectDelay) + timer := time.NewTimer(delay) defer timer.Stop() select { case <-ctx.Done(): @@ -1604,7 +1619,7 @@ func waitForSSEReconnect(ctx context.Context) bool { } } -func (lc *LineClient) pollLoop(ctx context.Context) { +func (lc *LineClient) pollLoop(ctx context.Context, sseDisconnected bool) { defer lc.wg.Done() var localRev int64 = 0 @@ -1638,8 +1653,9 @@ func (lc *LineClient) pollLoop(ctx context.Context) { return false } - sseDisconnected := false + var backoff reconnectBackoff handler := func(eventType, data string) { + backoff.reset() if sseDisconnected && lc.sendSSEConnectedIfCurrent(ctx) { sseDisconnected = false } @@ -1728,21 +1744,14 @@ func (lc *LineClient) pollLoop(ctx context.Context) { if probeAndReschedule() { return } - if !waitForSSEReconnect(receiveCtx) { - continue - } - continue + } else if line.IsAuthError(err) && lc.handleReceiveAuthError(ctx, client, err) { + return } - - if line.IsAuthError(err) { - if lc.handleReceiveAuthError(ctx, client, err) { - return - } - } else if !sseDisconnected && lc.sendSSEDisconnectedIfCurrent(ctx, err) { + if !sseDisconnected && lc.sendSSEDisconnectedIfCurrent(ctx, err) { sseDisconnected = true } } - if !waitForSSEReconnect(receiveCtx) { + if !waitForSSEReconnect(receiveCtx, backoff.next()) { continue } } @@ -1838,24 +1847,16 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return true } - if ctx.Err() != nil { + if ctx.Err() != nil || lc.isSessionInvalidated() { return true } if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) { return true } if lc.UserLogin != nil && lc.UserLogin.Bridge != nil { - lc.UserLogin.Bridge.Log.Error().Err(errRecover).Msg("Failed to recover session, stopping poll loop") - } - if lc.UserLogin != nil && lc.UserLogin.BridgeState != nil { - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateBadCredentials, - Error: "line-logged-out", - Message: "LINE session was invalidated (logged out by another client). Please re-authenticate the bridge.", - UserAction: status.UserActionRelogin, - }) + lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover session, retrying") } - return true + return false } func (lc *LineClient) handleOperation(ctx context.Context, op line.Operation) { diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 7b3f9243..8bd28635 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -601,7 +601,7 @@ func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(ctx) + lc.pollLoop(ctx, false) if len(tokens) != 2 { t.Fatalf("SSE attempts = %d, want 2", len(tokens)) @@ -649,7 +649,7 @@ func TestPollLoopMarksLoggedOutWhenReceiveAuthFails(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(context.Background()) + lc.pollLoop(context.Background(), false) if profileCalls != 1 { t.Fatalf("profile calls = %d, want 1", profileCalls) @@ -700,7 +700,7 @@ func TestPollLoopMarksLoggedOutWhenReceiveIdleProbeFailsLoggedOut(t *testing.T) } lc.wg.Add(1) - lc.pollLoop(context.Background()) + lc.pollLoop(context.Background(), false) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want 2", revisionCalls) @@ -765,7 +765,7 @@ func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(ctx) + lc.pollLoop(ctx, false) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want 2", revisionCalls) @@ -824,7 +824,7 @@ func TestPollLoopReceiveAuthDeadlineIgnoresHeartbeats(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(context.Background()) + lc.pollLoop(context.Background(), false) if heartbeatCalls != 5 { t.Fatalf("heartbeat batches = %d, want 5", heartbeatCalls) @@ -880,7 +880,7 @@ func TestPollLoopReceiveAuthDeadlineSurvivesEOFReconnects(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(context.Background()) + lc.pollLoop(context.Background(), false) if listenCalls != 3 { t.Fatalf("SSE attempts = %d, want 3", listenCalls) @@ -933,7 +933,7 @@ func TestPollLoopSuccessfulReceiveAuthProbePreservesLocalRev(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(ctx) + lc.pollLoop(ctx, false) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want startup plus one successful probe", revisionCalls) @@ -974,7 +974,7 @@ func TestPollLoopParentCancellationDoesNotProbeAuth(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(ctx) + lc.pollLoop(ctx, false) if revisionCalls != 1 { t.Fatalf("revision calls = %d, want startup only", revisionCalls) @@ -1009,6 +1009,42 @@ func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { } } +func TestReceiveAuthErrorStopsPollingOnlyWhenRecoveryIsRejected(t *testing.T) { + oldGetProfile := getProfileWithToken + oldRecover := recoverLineToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + recoverLineToken = oldRecover + }) + getProfileWithToken = func(context.Context, string) (*line.Profile, error) { + return nil, errAuthRequired + } + + tests := []struct { + name string + recoverErr error + wantStopped bool + }{ + {name: "transient failure keeps polling", recoverErr: errRecoveryTimeout}, + {name: "rejected credentials stop polling", recoverErr: errLoginRejected, wantStopped: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + recoverLineToken = func(*LineClient, context.Context) error { + return tt.recoverErr + } + lc := &LineClient{AccessToken: "expired"} + stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("expired"), errors.New("SSE error: 401")) + if stopped != tt.wantStopped { + t.Fatalf("stopped = %v, want %v", stopped, tt.wantStopped) + } + if lc.isSessionInvalidated() != tt.wantStopped || lc.IsLoggedIn() == tt.wantStopped { + t.Fatalf("invalidated = %v, logged in = %v, want invalidated = %v", lc.isSessionInvalidated(), lc.IsLoggedIn(), tt.wantStopped) + } + }) + } +} + func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { oldGetProfile := getProfileWithToken t.Cleanup(func() { From ac22648f27f73250e9283bc8e0c270a6edd9dc09 Mon Sep 17 00:00:00 2001 From: Nick Mills-Barrett Date: Wed, 28 Oct 2026 20:41:55 +0000 Subject: [PATCH 2/4] connector: scope recovery invalidation to session-wide calls and finish startup after token recovery - Only receive, the receive auth probe and the startup token check invalidate the session when recovery fails. Sends and media can get a 403 about the object itself, so a failed recovery there only returns the error, as before. - When the startup token check fails transiently, wait for a working token in the background before running the one-shot startup work (phone notifications, E2EE storage key, block list, chat sync) and starting the workers, instead of running it against a failing token. - Retry the post-PIN account check a few times on network or server errors, so a blip doesn't throw away the approved login and prompt for another PIN. --- pkg/connector/auth_recovery.go | 12 +++- pkg/connector/auth_recovery_test.go | 4 +- pkg/connector/client.go | 101 +++++++++++++++++++++------- pkg/connector/forced_logout_test.go | 2 +- pkg/connector/sync.go | 11 +-- pkg/connector/sync_test.go | 16 ++--- 6 files changed, 102 insertions(+), 44 deletions(-) diff --git a/pkg/connector/auth_recovery.go b/pkg/connector/auth_recovery.go index 26e39839..f08204d8 100644 --- a/pkg/connector/auth_recovery.go +++ b/pkg/connector/auth_recovery.go @@ -163,9 +163,6 @@ func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedCli // case another serialized recovery rotated that token first. return lc.recoverClientAfterAuthError(ctx, newLineAPIClient(recoveryToken), errRecover) } - if ctx.Err() == nil && isRecoveryRejected(errRecover) { - lc.markRecoveryRejected(ctx, recoveryToken, errRecover) - } return nil, errRecover } if ctx.Err() != nil { @@ -180,6 +177,15 @@ func (lc *LineClient) recoverClientAfterAuthError(ctx context.Context, failedCli return lc.newClient(), nil } +func (lc *LineClient) recoverSessionAfterAuthError(ctx context.Context, failedClient *line.Client, err error) (*line.Client, error) { + recoveryToken := lc.getAccessToken() + recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + if errRecover != nil && ctx.Err() == nil && isRecoveryRejected(errRecover) { + lc.markRecoveryRejected(ctx, recoveryToken, errRecover) + } + return recoveredClient, errRecover +} + func (lc *LineClient) callLine(ctx context.Context, call func(*line.Client) error) (*line.Client, error) { return lc.callLineUsing(ctx, nil, call) } diff --git a/pkg/connector/auth_recovery_test.go b/pkg/connector/auth_recovery_test.go index b42c93cf..7388e264 100644 --- a/pkg/connector/auth_recovery_test.go +++ b/pkg/connector/auth_recovery_test.go @@ -69,9 +69,9 @@ func TestRecoveryFailureInvalidatesOnlyRejectedSessions(t *testing.T) { return tt.recoverErr } lc := &LineClient{AccessToken: "current-token"} - retryClient, err := lc.recoverClientAfterAuthError(context.Background(), line.NewClient("current-token"), errAuthRequired) + retryClient, err := lc.recoverSessionAfterAuthError(context.Background(), line.NewClient("current-token"), errAuthRequired) if !errors.Is(err, tt.recoverErr) { - t.Fatalf("recoverClientAfterAuthError error = %v, want %v", err, tt.recoverErr) + t.Fatalf("recoverSessionAfterAuthError error = %v, want %v", err, tt.recoverErr) } if retryClient != nil { t.Fatalf("retry client = %#v, want nil", retryClient) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 58596512..ecce1d5d 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -620,14 +620,11 @@ func (lc *LineClient) Connect(ctx context.Context) { if !started { return } - defer lc.wg.Done() + handedOff := false workersStarted := false defer func() { - if !workersStarted { - if lc.E2EE != nil { - lc.E2EE.Close() - } - run.cancel() + if !handedOff { + lc.finishStartup(run, workersStarted) } }() @@ -686,29 +683,63 @@ func (lc *LineClient) Connect(ctx context.Context) { } // Verify the token is still valid before proceeding - tokenErr := lc.ensureValidToken(ctx) - if tokenErr != nil { + if tokenErr := lc.ensureValidToken(ctx); tokenErr != nil { if ctx.Err() == nil && lc.isLoggedOut(tokenErr) { lc.markLoggedOutByOtherClient(ctx, tokenErr) } - if ctx.Err() != nil || lc.isSessionInvalidated() { + if ctx.Err() != nil || lc.isSessionInvalidated() || !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { return } - lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, continuing so the poll loop can retry") + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, retrying before starting the session") + // Startup work below runs once per connection, so it waits for a working + // token instead of running against a failing one. Connect itself returns so + // other logins aren't held up. + handedOff = true + go func() { + lc.finishStartup(run, lc.waitForValidToken(ctx) && lc.startSession(ctx)) + }() + return + } + workersStarted = lc.startSession(ctx) +} + +func (lc *LineClient) finishStartup(run *lineClientRun, workersStarted bool) { + if !workersStarted { + if lc.E2EE != nil { + lc.E2EE.Close() + } + run.cancel() + } + lc.wg.Done() +} + +func (lc *LineClient) waitForValidToken(ctx context.Context) bool { + var backoff reconnectBackoff + for waitForSSEReconnect(ctx, backoff.next()) { + err := lc.ensureValidToken(ctx) + switch { + case err == nil: + return true + case ctx.Err() != nil: + return false + case lc.isLoggedOut(err): + lc.markLoggedOutByOtherClient(ctx, err) + return false + case lc.isSessionInvalidated(): + return false + } + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Still failing to verify LINE token, retrying") } + return false +} + +func (lc *LineClient) startSession(ctx context.Context) bool { // LINE Chrome disables mobile notifications while its session is active by // default. Explicitly clear that setting once per bridge connection so the // user's phone continues receiving notifications. lc.configurePhoneNotifications(ctx) - if ctx.Err() != nil { - return - } - if tokenErr != nil { - if !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { - return - } - } else if !lc.sendConnectedStateIfCurrent(ctx) { - return + if ctx.Err() != nil || !lc.sendConnectedStateIfCurrent(ctx) { + return false } // Initialize E2EE manager and load keys @@ -765,14 +796,28 @@ func (lc *LineClient) Connect(ctx context.Context) { // sender's existing membership look like a fresh join. lc.syncChatsNow(ctx) if ctx.Err() != nil { - return + return false } lc.wg.Add(3) go lc.syncDMChats(ctx) go lc.prefetchMessages(ctx) - go lc.pollLoop(ctx, tokenErr != nil) - workersStarted = true + go lc.pollLoop(ctx) + return true +} + +func verifyReloginProfile(ctx context.Context, accessToken string) (*line.Profile, error) { + for attempt := 1; ; attempt++ { + profile, err := newLineAPIClient(accessToken).GetProfileContext(ctx) + if err == nil || attempt >= 3 || !isTransientLineError(err) { + return profile, err + } + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(time.Duration(attempt) * 2 * time.Second): + } + } } func (lc *LineClient) tryLogin(ctx context.Context) error { @@ -859,7 +904,7 @@ func (lc *LineClient) tryLogin(ctx context.Context) error { return errLineClientSuperseded } if res.Attempt != nil { - profile, err := newLineAPIClient(accessToken).GetProfileContext(ctx) + profile, err := verifyReloginProfile(ctx, accessToken) if err != nil { return fmt.Errorf("verify re-login account: %w", err) } @@ -1006,8 +1051,14 @@ func (lc *LineClient) refreshLoginE2EEKeys(res *line.LoginResult, meta *UserLogi } func (lc *LineClient) ensureValidToken(ctx context.Context) error { - _, _, err := callLineResult(lc, ctx, func(client *line.Client) (*line.Profile, error) { - return getProfileWithToken(ctx, client.AccessToken) + _, _, err := callLineWithRecovery(ctx, nil, lineCallDeps[*line.Profile]{ + newClient: func() *line.Client { return lc.newClient() }, + recover: lc.recoverSessionAfterAuthError, + isAuthError: lc.isTokenError, + beforeCall: func() error { return lc.checkLineCall(ctx) }, + call: func(client *line.Client) (*line.Profile, error) { + return getProfileWithToken(ctx, client.AccessToken) + }, }) return err } diff --git a/pkg/connector/forced_logout_test.go b/pkg/connector/forced_logout_test.go index 24463ea0..f6b1a33d 100644 --- a/pkg/connector/forced_logout_test.go +++ b/pkg/connector/forced_logout_test.go @@ -70,7 +70,7 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { lc.missingE2EEKeyMu.Lock() } go func() { - _, err := lc.recoverClientAfterAuthError(ctx, line.NewClient("token"), failure) + _, err := lc.recoverSessionAfterAuthError(ctx, line.NewClient("token"), failure) recovered <- err }() if mode == "invalid-key" { diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 3df605f3..cf5e9819 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -1619,7 +1619,7 @@ func waitForSSEReconnect(ctx context.Context, delay time.Duration) bool { } } -func (lc *LineClient) pollLoop(ctx context.Context, sseDisconnected bool) { +func (lc *LineClient) pollLoop(ctx context.Context) { defer lc.wg.Done() var localRev int64 = 0 @@ -1653,6 +1653,7 @@ func (lc *LineClient) pollLoop(ctx context.Context, sseDisconnected bool) { return false } + sseDisconnected := false var backoff reconnectBackoff handler := func(eventType, data string) { backoff.reset() @@ -1784,7 +1785,7 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { return lc.handleReceiveAuthError(ctx, probeClient, probeErr) } - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, probeClient, probeErr) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, probeClient, probeErr) if errRecover != nil { if errors.Is(errRecover, errLineSessionInvalidated) || errors.Is(errRecover, errLineClientSuperseded) || ctx.Err() != nil { return true @@ -1812,7 +1813,7 @@ func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { // the detailed forced-logout envelope before deciding whether recovery is safe. func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient *line.Client, err error) bool { if lc.isLoggedOut(err) { - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, failedClient, err) if errRecover != nil { return true } @@ -1829,7 +1830,7 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return true } if lc.isLoggedOut(profileErr) { - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, profileClient, profileErr) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, profileClient, profileErr) if errRecover != nil { return true } @@ -1839,7 +1840,7 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, failedClient * return false } - recoveredClient, errRecover := lc.recoverClientAfterAuthError(ctx, failedClient, err) + recoveredClient, errRecover := lc.recoverSessionAfterAuthError(ctx, failedClient, err) if recoveredClient != nil { return false } diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 8bd28635..3e149747 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -601,7 +601,7 @@ func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(ctx, false) + lc.pollLoop(ctx) if len(tokens) != 2 { t.Fatalf("SSE attempts = %d, want 2", len(tokens)) @@ -649,7 +649,7 @@ func TestPollLoopMarksLoggedOutWhenReceiveAuthFails(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(context.Background(), false) + lc.pollLoop(context.Background()) if profileCalls != 1 { t.Fatalf("profile calls = %d, want 1", profileCalls) @@ -700,7 +700,7 @@ func TestPollLoopMarksLoggedOutWhenReceiveIdleProbeFailsLoggedOut(t *testing.T) } lc.wg.Add(1) - lc.pollLoop(context.Background(), false) + lc.pollLoop(context.Background()) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want 2", revisionCalls) @@ -765,7 +765,7 @@ func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) { } lc.wg.Add(1) - lc.pollLoop(ctx, false) + lc.pollLoop(ctx) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want 2", revisionCalls) @@ -824,7 +824,7 @@ func TestPollLoopReceiveAuthDeadlineIgnoresHeartbeats(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(context.Background(), false) + lc.pollLoop(context.Background()) if heartbeatCalls != 5 { t.Fatalf("heartbeat batches = %d, want 5", heartbeatCalls) @@ -880,7 +880,7 @@ func TestPollLoopReceiveAuthDeadlineSurvivesEOFReconnects(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(context.Background(), false) + lc.pollLoop(context.Background()) if listenCalls != 3 { t.Fatalf("SSE attempts = %d, want 3", listenCalls) @@ -933,7 +933,7 @@ func TestPollLoopSuccessfulReceiveAuthProbePreservesLocalRev(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(ctx, false) + lc.pollLoop(ctx) if revisionCalls != 2 { t.Fatalf("revision calls = %d, want startup plus one successful probe", revisionCalls) @@ -974,7 +974,7 @@ func TestPollLoopParentCancellationDoesNotProbeAuth(t *testing.T) { }, } lc.wg.Add(1) - lc.pollLoop(ctx, false) + lc.pollLoop(ctx) if revisionCalls != 1 { t.Fatalf("revision calls = %d, want startup only", revisionCalls) From f8cbb835a03f24a3d0719d5ae5ce2cf98feb8a30 Mon Sep 17 00:00:00 2001 From: Nick Mills-Barrett Date: Wed, 28 Oct 2026 20:51:51 +0000 Subject: [PATCH 3/4] connector: only hold startup for auth or transient token check failures Any other getProfile failure (an unexpected 4xx or a decode error) says nothing about the token, so start the session instead of waiting forever. --- pkg/connector/client.go | 33 +++++++++++++++++++++++---------- 1 file changed, 23 insertions(+), 10 deletions(-) diff --git a/pkg/connector/client.go b/pkg/connector/client.go index ecce1d5d..4635b48b 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -687,22 +687,32 @@ func (lc *LineClient) Connect(ctx context.Context) { if ctx.Err() == nil && lc.isLoggedOut(tokenErr) { lc.markLoggedOutByOtherClient(ctx, tokenErr) } - if ctx.Err() != nil || lc.isSessionInvalidated() || !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { + if ctx.Err() != nil || lc.isSessionInvalidated() { return } - lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, retrying before starting the session") - // Startup work below runs once per connection, so it waits for a working - // token instead of running against a failing one. Connect itself returns so - // other logins aren't held up. - handedOff = true - go func() { - lc.finishStartup(run, lc.waitForValidToken(ctx) && lc.startSession(ctx)) - }() - return + if lc.shouldWaitForToken(tokenErr) { + if !lc.sendSSEDisconnectedIfCurrent(ctx, tokenErr) { + return + } + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, retrying before starting the session") + // Startup work below runs once per connection, so it waits for a working + // token instead of running against a failing one. Connect itself returns so + // other logins aren't held up. + handedOff = true + go func() { + lc.finishStartup(run, lc.waitForValidToken(ctx) && lc.startSession(ctx)) + }() + return + } + lc.UserLogin.Bridge.Log.Warn().Err(tokenErr).Msg("Failed to verify LINE token, starting anyway") } workersStarted = lc.startSession(ctx) } +func (lc *LineClient) shouldWaitForToken(err error) bool { + return lc.isTokenError(err) || isTransientLineError(err) +} + func (lc *LineClient) finishStartup(run *lineClientRun, workersStarted bool) { if !workersStarted { if lc.E2EE != nil { @@ -727,6 +737,9 @@ func (lc *LineClient) waitForValidToken(ctx context.Context) bool { return false case lc.isSessionInvalidated(): return false + case !lc.shouldWaitForToken(err): + lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Failed to verify LINE token, starting anyway") + return true } lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("Still failing to verify LINE token, retrying") } From 635f6886d1f6c8f049d79dafa9a473d77c8fb6a2 Mon Sep 17 00:00:00 2001 From: Nick Mills-Barrett Date: Wed, 28 Oct 2026 21:15:08 +0000 Subject: [PATCH 4/4] Drop the tests added in this PR --- pkg/connector/auth_recovery_test.go | 126 +--------------------- pkg/connector/forced_logout_test.go | 21 +--- pkg/connector/phone_notifications_test.go | 3 +- pkg/connector/sync_test.go | 36 ------- 4 files changed, 11 insertions(+), 175 deletions(-) diff --git a/pkg/connector/auth_recovery_test.go b/pkg/connector/auth_recovery_test.go index 7388e264..d74affc0 100644 --- a/pkg/connector/auth_recovery_test.go +++ b/pkg/connector/auth_recovery_test.go @@ -3,9 +3,6 @@ package connector import ( "context" "errors" - "fmt" - "net/url" - "os" "strings" "sync" "sync/atomic" @@ -16,126 +13,13 @@ import ( ) var ( - errAuthRequired = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":119,"reason":"Access token refresh required"}}`) - errLoggedOut = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":8,"reason":"V3_TOKEN_CLIENT_LOGGED_OUT"}}`) - errSenderKey = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":83,"reason":"invalid sender key"}}`) - errNotMember = errors.New(`API error 400: {"code":10051,"data":{"name":"TalkException","code":10,"reason":"not a member"}}`) - errNetwork = errors.New("request failed: dial tcp: i/o timeout") - errRecoveryTimeout = fmt.Errorf("request failed: %w", &url.Error{Op: "Post", URL: "https://line-chrome-gw.line-apps.com/api/auth/tokenRefresh", Err: os.ErrDeadlineExceeded}) - errLoginRejected = errors.New("login failed: loginV2 failed: invalid password (code 1)") + errAuthRequired = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":119,"reason":"Access token refresh required"}}`) + errLoggedOut = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":8,"reason":"V3_TOKEN_CLIENT_LOGGED_OUT"}}`) + errSenderKey = errors.New(`API error 400: {"code":10051,"message":"RESPONSE_ERROR","data":{"name":"TalkException","code":83,"reason":"invalid sender key"}}`) + errNotMember = errors.New(`API error 400: {"code":10051,"data":{"name":"TalkException","code":10,"reason":"not a member"}}`) + errNetwork = errors.New("request failed: dial tcp: i/o timeout") ) -func TestIsTransientLineError(t *testing.T) { - tests := []struct { - name string - err error - want bool - }{ - {name: "network timeout", err: errRecoveryTimeout, want: true}, - {name: "server error", err: &line.HTTPError{StatusCode: 503}, want: true}, - {name: "rate limited", err: &line.HTTPError{StatusCode: 429}, want: true}, - {name: "backed off network failure", err: fmt.Errorf("waiting to retry LINE token recovery: %w", errRecoveryTimeout), want: true}, - {name: "unauthorized", err: &line.HTTPError{StatusCode: 401}, want: false}, - {name: "login rejected", err: errLoginRejected, want: false}, - {name: "unanswered PIN prompt", err: fmt.Errorf("%w: %w", errLinePINNotVerified, errRecoveryTimeout), want: false}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := isTransientLineError(tt.err); got != tt.want { - t.Fatalf("isTransientLineError(%v) = %v, want %v", tt.err, got, tt.want) - } - }) - } -} - -func TestRecoveryFailureInvalidatesOnlyRejectedSessions(t *testing.T) { - oldRecover := recoverLineToken - t.Cleanup(func() { - recoverLineToken = oldRecover - }) - - tests := []struct { - name string - recoverErr error - wantInvalidated bool - }{ - {name: "transient failure keeps the session", recoverErr: errRecoveryTimeout}, - {name: "rejected credentials invalidate the session", recoverErr: errLoginRejected, wantInvalidated: true}, - {name: "unanswered PIN prompt invalidates the session", recoverErr: fmt.Errorf("%w: %w", errLinePINNotVerified, errRecoveryTimeout), wantInvalidated: true}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - recoverLineToken = func(*LineClient, context.Context) error { - return tt.recoverErr - } - lc := &LineClient{AccessToken: "current-token"} - retryClient, err := lc.recoverSessionAfterAuthError(context.Background(), line.NewClient("current-token"), errAuthRequired) - if !errors.Is(err, tt.recoverErr) { - t.Fatalf("recoverSessionAfterAuthError error = %v, want %v", err, tt.recoverErr) - } - if retryClient != nil { - t.Fatalf("retry client = %#v, want nil", retryClient) - } - if lc.isSessionInvalidated() != tt.wantInvalidated || lc.IsLoggedIn() == tt.wantInvalidated { - t.Fatalf("invalidated = %v, logged in = %v, want invalidated = %v", lc.isSessionInvalidated(), lc.IsLoggedIn(), tt.wantInvalidated) - } - }) - } -} - -func TestRunTokenRecoveryBacksOffAfterFailure(t *testing.T) { - lc := &LineClient{} - var calls int - recover := func(context.Context) error { - calls++ - if calls == 1 { - return errRecoveryTimeout - } - return nil - } - - if err := lc.runTokenRecovery(context.Background(), recover); !errors.Is(err, errRecoveryTimeout) { - t.Fatalf("first recovery error = %v, want %v", err, errRecoveryTimeout) - } - lc.recoverRetryAt = time.Now().Add(time.Minute) - if err := lc.runTokenRecovery(context.Background(), recover); !errors.Is(err, errRecoveryTimeout) { - t.Fatalf("backed off recovery error = %v, want %v", err, errRecoveryTimeout) - } - if calls != 1 { - t.Fatalf("recovery calls during backoff = %d, want 1", calls) - } - - lc.recoverRetryAt = time.Now() - if err := lc.runTokenRecovery(context.Background(), recover); err != nil { - t.Fatalf("recovery after backoff returned error: %v", err) - } - if calls != 2 { - t.Fatalf("recovery calls after backoff = %d, want 2", calls) - } - if lc.recoverBackoff.delay != 0 || !lc.recoverRetryAt.IsZero() { - t.Fatal("successful recovery did not reset the backoff") - } -} - -func TestRecoverTokenDoesNotReloginAfterTransientRefreshFailure(t *testing.T) { - lc := &LineClient{} - var reloginCalls int - err := lc.recoverTokenWith( - context.Background(), - func(context.Context) error { return errRecoveryTimeout }, - func(context.Context) error { - reloginCalls++ - return nil - }, - ) - if !errors.Is(err, errRecoveryTimeout) { - t.Fatalf("recoverTokenWith error = %v, want %v", err, errRecoveryTimeout) - } - if reloginCalls != 0 { - t.Fatalf("relogin calls = %d, want 0", reloginCalls) - } -} - func TestCallLineWithRecovery(t *testing.T) { tests := []struct { name string diff --git a/pkg/connector/forced_logout_test.go b/pkg/connector/forced_logout_test.go index f6b1a33d..efe123a3 100644 --- a/pkg/connector/forced_logout_test.go +++ b/pkg/connector/forced_logout_test.go @@ -2,7 +2,6 @@ package connector import ( "context" - "errors" "io" "path/filepath" "testing" @@ -21,7 +20,7 @@ import ( ) func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { - for _, mode := range []string{"ordinary", "invalid-key", "late-invalid-key", "late-stale-key", "rejected-recovery"} { + for _, mode := range []string{"ordinary", "invalid-key", "late-invalid-key", "late-stale-key"} { t.Run(mode, func(t *testing.T) { invalidKey := mode == "invalid-key" || mode == "late-invalid-key" ctx := context.Background() @@ -58,19 +57,12 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { if mode == "invalid-key" { failure = errSenderKey } - var wantRecoverErr error - if mode == "rejected-recovery" { - failure, wantRecoverErr = errAuthRequired, errLoginRejected - oldRecover := recoverLineToken - t.Cleanup(func() { recoverLineToken = oldRecover }) - recoverLineToken = func(*LineClient, context.Context) error { return errLoginRejected } - } recovered := make(chan error, 1) if mode == "invalid-key" { lc.missingE2EEKeyMu.Lock() } go func() { - _, err := lc.recoverSessionAfterAuthError(ctx, line.NewClient("token"), failure) + _, err := lc.recoverClientAfterAuthError(ctx, line.NewClient("token"), failure) recovered <- err }() if mode == "invalid-key" { @@ -82,8 +74,8 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { } lc.missingE2EEKeyMu.Unlock() } - if err := <-recovered; !errors.Is(err, wantRecoverErr) { - t.Fatalf("recovery error = %v, want %v", err, wantRecoverErr) + if err := <-recovered; err != nil { + t.Fatal(err) } if mode == "late-invalid-key" || mode == "late-stale-key" { if !lc.isTokenError(errSenderKey) { @@ -102,8 +94,6 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { wantStateError := status.BridgeStateErrorCode("line-logged-out") if mode == "invalid-key" { wantStateError = "line-e2ee-key-missing" - } else if mode == "rejected-recovery" { - wantStateError = "line-token-expired" } select { case state := <-matrix.states: @@ -144,9 +134,6 @@ func TestInvalidSenderKeyReconnectPersistsFullVerification(t *testing.T) { if invalidKey { wantRestartError = "line-e2ee-key-missing" wantRestartMessage = lineMissingE2EEKeyMessage - } else if mode == "rejected-recovery" { - wantRestartError = "line-token-expired" - wantRestartMessage = status.BridgeStateHumanErrors["line-token-expired"] } if state.StateEvent != status.StateBadCredentials || state.Error != wantRestartError || state.Message != wantRestartMessage || state.UserAction != status.UserActionRelogin { t.Fatalf("unexpected state after bridge restart: %v", state) diff --git a/pkg/connector/phone_notifications_test.go b/pkg/connector/phone_notifications_test.go index 8d91505c..728befd7 100644 --- a/pkg/connector/phone_notifications_test.go +++ b/pkg/connector/phone_notifications_test.go @@ -136,10 +136,11 @@ func TestConfigurePhoneNotificationsTreatsAuthRecoveryFailureAsBestEffort(t *tes ) error { return errAuthRequired } + recoveryErr := errors.New("stored credentials rejected") var recoveries int recoverLineToken = func(*LineClient, context.Context) error { recoveries++ - return errRecoveryTimeout + return recoveryErr } ctx := context.Background() diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 3e149747..7b3f9243 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -1009,42 +1009,6 @@ func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { } } -func TestReceiveAuthErrorStopsPollingOnlyWhenRecoveryIsRejected(t *testing.T) { - oldGetProfile := getProfileWithToken - oldRecover := recoverLineToken - t.Cleanup(func() { - getProfileWithToken = oldGetProfile - recoverLineToken = oldRecover - }) - getProfileWithToken = func(context.Context, string) (*line.Profile, error) { - return nil, errAuthRequired - } - - tests := []struct { - name string - recoverErr error - wantStopped bool - }{ - {name: "transient failure keeps polling", recoverErr: errRecoveryTimeout}, - {name: "rejected credentials stop polling", recoverErr: errLoginRejected, wantStopped: true}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - recoverLineToken = func(*LineClient, context.Context) error { - return tt.recoverErr - } - lc := &LineClient{AccessToken: "expired"} - stopped := lc.handleReceiveAuthError(context.Background(), line.NewClient("expired"), errors.New("SSE error: 401")) - if stopped != tt.wantStopped { - t.Fatalf("stopped = %v, want %v", stopped, tt.wantStopped) - } - if lc.isSessionInvalidated() != tt.wantStopped || lc.IsLoggedIn() == tt.wantStopped { - t.Fatalf("invalidated = %v, logged in = %v, want invalidated = %v", lc.isSessionInvalidated(), lc.IsLoggedIn(), tt.wantStopped) - } - }) - } -} - func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { oldGetProfile := getProfileWithToken t.Cleanup(func() {