diff --git a/pkg/connector/auth_recovery_test.go b/pkg/connector/auth_recovery_test.go index f72c59d..07e8b01 100644 --- a/pkg/connector/auth_recovery_test.go +++ b/pkg/connector/auth_recovery_test.go @@ -245,6 +245,102 @@ func TestRunTokenRecoveryRejectsInvalidatedSessionBeforeRecentRecovery(t *testin } } +func TestRunTokenRecoveryRejectsSupersededClient(t *testing.T) { + lc := &LineClient{} + lc.retire() + var calls int + err := lc.runTokenRecovery(context.Background(), func(context.Context) error { + calls++ + return nil + }) + if !errors.Is(err, errLineClientSuperseded) { + t.Fatalf("err = %v, want errLineClientSuperseded", err) + } + if calls != 0 { + t.Fatalf("recovery calls = %d, want 0", calls) + } +} + +func TestRecoverTokenDoesNotReloginAfterForcedLogoutRefresh(t *testing.T) { + lc := &LineClient{} + var reloginCalls int + err := lc.recoverTokenWith( + context.Background(), + func(context.Context) error { return errLoggedOut }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + if !line.IsLoggedOut(err) { + t.Fatalf("recoverTokenWith error = %v, want logged-out error", err) + } + if reloginCalls != 0 { + t.Fatalf("relogin calls = %d, want 0", reloginCalls) + } +} + +func TestRecoverTokenDoesNotReloginAfterCancellation(t *testing.T) { + lc := &LineClient{} + ctx, cancel := context.WithCancel(context.Background()) + var reloginCalls int + err := lc.recoverTokenWith( + ctx, + func(context.Context) error { + cancel() + return context.Canceled + }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + if !errors.Is(err, context.Canceled) { + t.Fatalf("recoverTokenWith error = %v, want context.Canceled", err) + } + if reloginCalls != 0 { + t.Fatalf("relogin calls = %d, want 0", reloginCalls) + } +} + +func TestForcedLogoutWinsOverInFlightRecovery(t *testing.T) { + lc := &LineClient{AccessToken: "old-token"} + recoveryStarted := make(chan struct{}) + allowRecovery := make(chan struct{}) + recoveryDone := make(chan error, 1) + go func() { + recoveryDone <- lc.runTokenRecovery(context.Background(), func(context.Context) error { + close(recoveryStarted) + <-allowRecovery + lc.setTokens("recovered-token", "") + return nil + }) + }() + <-recoveryStarted + + logoutDone := make(chan struct{}) + go func() { + lc.markLoggedOutByOtherClient(context.Background(), errLoggedOut) + close(logoutDone) + }() + close(allowRecovery) + + if err := <-recoveryDone; err != nil { + t.Fatalf("recovery returned error: %v", err) + } + select { + case <-logoutDone: + case <-time.After(time.Second): + t.Fatal("forced logout did not complete after recovery") + } + if lc.hasAccessToken() { + t.Fatal("in-flight recovery resurrected the invalidated session") + } + if !lc.isSessionInvalidated() { + t.Fatal("session was not invalidated after in-flight recovery") + } +} + func TestRunTokenRecoverySerializesConcurrentRecovery(t *testing.T) { var lc LineClient var calls int32 diff --git a/pkg/connector/client.go b/pkg/connector/client.go index 4be4ecf..62fc5a3 100644 --- a/pkg/connector/client.go +++ b/pkg/connector/client.go @@ -8,6 +8,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "maunium.net/go/mautrix/bridgev2" @@ -18,7 +19,10 @@ import ( "github.com/highesttt/matrix-line-messenger/pkg/line" ) -var errLineSessionInvalidated = errors.New("LINE session invalidated by another client") +var ( + errLineSessionInvalidated = errors.New("LINE session invalidated by another client") + errLineClientSuperseded = errors.New("LINE client was superseded") +) const lineMissingE2EEKeyMessage = "LINE encryption keys are unavailable. Reconnect LINE in Beeper to restore message decryption." @@ -29,8 +33,8 @@ var loginWithCredentials = func(email, password, certificate string) (*line.Logi return newLineAPIClient("").Login(email, password, certificate) } -var getProfileWithToken = func(token string) (*line.Profile, error) { - return newLineAPIClient(token).GetProfile() +var getProfileWithToken = func(ctx context.Context, token string) (*line.Profile, error) { + return newLineAPIClient(token).GetProfileContext(ctx) } type LineClient struct { @@ -49,12 +53,19 @@ type LineClient struct { tokenMu sync.RWMutex recoverMu sync.Mutex recoverTime time.Time + recoveryStopped bool missingE2EEKeyMu sync.Mutex missingE2EEKeyMarked bool // sessionInvalidated is set when LINE forcefully logs out this Chrome-style // session. It prevents background calls from re-logging in before the user // clicks Reconnect. sessionInvalidated bool + runMu sync.Mutex + activeRun *lineClientRun + stopped bool + superseded atomic.Bool + forcedLogoutMu sync.Mutex + forcedLogoutSent bool // cacheMu protects peerKeys, blockedUsers, contactCache, mediaFlowCache, // noE2EEGroups, groupMemberCache, generatedGroupNameCache, and knownMemberChatMIDs. @@ -73,6 +84,10 @@ type LineClient struct { wg sync.WaitGroup } +type lineClientRun struct { + cancel context.CancelFunc +} + type cachedMediaFlow struct { flowMap map[string]int cachedAt time.Time @@ -138,6 +153,50 @@ func (lc *LineClient) isSessionInvalidated() bool { return lc.sessionInvalidated } +func (lc *LineClient) beginRun(parent context.Context) (context.Context, *lineClientRun, bool) { + ctx, cancel := context.WithCancel(parent) + run := &lineClientRun{cancel: cancel} + lc.runMu.Lock() + if lc.stopped { + lc.runMu.Unlock() + cancel() + return ctx, run, false + } + // Reserve one count for the full Connect invocation before publishing the + // run. Disconnect cannot observe zero and return while startup is in flight. + lc.wg.Add(1) + previous := lc.activeRun + lc.activeRun = run + lc.runMu.Unlock() + if previous != nil { + previous.cancel() + } + return ctx, run, true +} + +func (lc *LineClient) cancelActiveRun() { + lc.runMu.Lock() + run := lc.activeRun + lc.runMu.Unlock() + if run != nil { + run.cancel() + } +} + +func (lc *LineClient) retire() { + lc.Disconnect() +} + +func (lc *LineClient) claimForcedLogoutState() bool { + lc.forcedLogoutMu.Lock() + defer lc.forcedLogoutMu.Unlock() + if lc.forcedLogoutSent { + return false + } + lc.forcedLogoutSent = true + return true +} + func (lc *LineClient) newClient() *line.Client { return newLineAPIClient(lc.getAccessToken()) } @@ -259,7 +318,7 @@ func (lc *LineClient) shouldAttemptTokenRecovery(ctx context.Context, err error) if err == nil { return false } - if lc.isSessionInvalidated() { + if ctx.Err() != nil || lc.superseded.Load() || lc.isSessionInvalidated() { return false } if lc.isLoggedOut(err) { @@ -270,24 +329,35 @@ func (lc *LineClient) shouldAttemptTokenRecovery(ctx context.Context, err error) } func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) { + // Serialize invalidation with token recovery. If a refresh/re-login was + // already in flight, it may finish first, but this transition always runs + // afterward so recovery cannot resurrect a forcefully logged-out session. + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + if lc.UserLogin == nil { lc.invalidateAccessToken() line.InvalidateOBSTokenCache() + lc.cancelActiveRun() return } - if lc.UserLogin.Client != nil && lc.UserLogin.Client != lc { + // superseded covers connector-ordered retirement. The identity check is a + // final guard for bridgev2 client swaps that happen outside that lifecycle. + if lc.superseded.Load() || (lc.UserLogin.Client != nil && lc.UserLogin.Client != lc) { if lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Debug().Err(err).Msg("Ignoring forced logout from stale LINE client") } + lc.cancelActiveRun() return } + sendState := lc.claimForcedLogoutState() lc.invalidateAccessToken() line.InvalidateOBSTokenCache() lc.saveSessionInvalidated(ctx) - if lc.UserLogin.Bridge != nil { + if sendState && lc.UserLogin.Bridge != nil { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("LINE session invalidated by another client; marking login bad credentials") } - if lc.UserLogin.BridgeState != nil { + if sendState && lc.UserLogin.BridgeState != nil { lc.UserLogin.BridgeState.Send(status.BridgeState{ StateEvent: status.StateBadCredentials, Error: "line-logged-out", @@ -295,6 +365,22 @@ func (lc *LineClient) markLoggedOutByOtherClient(ctx context.Context, err error) UserAction: status.UserActionRelogin, }) } + lc.cancelActiveRun() +} + +func (lc *LineClient) sendConnectedStateIfCurrent(ctx context.Context) bool { + // Serialize the final startup state with forced logout and token recovery. + // Whichever transition gets this lock last determines the visible state. + lc.recoverMu.Lock() + defer lc.recoverMu.Unlock() + if ctx.Err() != nil || lc.superseded.Load() || lc.isSessionInvalidated() || !lc.hasAccessToken() { + return false + } + lc.UserLogin.Bridge.Log.Info().Int("token_len", len(lc.getAccessToken())).Msg("LINE client connected; notifying bridge") + lc.UserLogin.BridgeState.Send(status.BridgeState{ + StateEvent: status.StateConnected, + }) + return true } func (lc *LineClient) saveSessionInvalidated(ctx context.Context) { @@ -358,13 +444,32 @@ 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) +} + +func (lc *LineClient) recoverTokenWith( + ctx context.Context, + refresh func(context.Context) error, + relogin func(context.Context) error, +) error { return lc.runTokenRecovery(ctx, func(ctx context.Context) error { - if err := lc.refreshAndSave(ctx); err == nil { + if err := refresh(ctx); err == nil { + if ctx.Err() != nil { + return ctx.Err() + } lc.UserLogin.Bridge.Log.Info().Msg("Token recovered via refresh") return nil + } else if ctx.Err() != nil { + return ctx.Err() + } else if lc.isLoggedOut(err) || errors.Is(err, errLineSessionInvalidated) { + return err } lc.UserLogin.Bridge.Log.Info().Msg("Refresh failed, attempting re-login with stored credentials...") - return lc.tryLogin(ctx) + err := relogin(ctx) + if ctx.Err() != nil { + return ctx.Err() + } + return err }) } @@ -372,6 +477,12 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context lc.recoverMu.Lock() defer lc.recoverMu.Unlock() + if ctx.Err() != nil { + return ctx.Err() + } + if lc.recoveryStopped || lc.superseded.Load() { + return errLineClientSuperseded + } if lc.isSessionInvalidated() { return errLineSessionInvalidated } @@ -383,11 +494,32 @@ func (lc *LineClient) runTokenRecovery(ctx context.Context, recover func(context if err := recover(ctx); err != nil { return err } + if ctx.Err() != nil { + return ctx.Err() + } + if lc.superseded.Load() { + return errLineClientSuperseded + } + if lc.isSessionInvalidated() { + return errLineSessionInvalidated + } lc.recoverTime = time.Now() return nil } func (lc *LineClient) Connect(ctx context.Context) { + ctx, run, started := lc.beginRun(ctx) + if !started { + return + } + defer lc.wg.Done() + workersStarted := false + defer func() { + if !workersStarted { + run.cancel() + } + }() + lc.cacheMu.Lock() if lc.peerKeys == nil { lc.peerKeys = make(map[string]peerKeyInfo) @@ -421,7 +553,14 @@ func (lc *LineClient) Connect(ctx context.Context) { lc.markLoggedOutByOtherClient(ctx, errLineSessionInvalidated) return } - if err := lc.tryLogin(ctx); err != nil { + if err := lc.runTokenRecovery(ctx, lc.tryLogin); err != nil { + if ctx.Err() != nil || lc.superseded.Load() || errors.Is(err, errLineSessionInvalidated) { + return + } + if lc.isLoggedOut(err) { + lc.markLoggedOutByOtherClient(ctx, err) + return + } lc.UserLogin.BridgeState.Send(status.BridgeState{ StateEvent: status.StateBadCredentials, Error: "line-login-failed", @@ -434,6 +573,9 @@ 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 + } if lc.isLoggedOut(err) { lc.markLoggedOutByOtherClient(ctx, err) return @@ -446,11 +588,9 @@ func (lc *LineClient) Connect(ctx context.Context) { }) return } - - lc.UserLogin.Bridge.Log.Info().Int("token_len", len(lc.getAccessToken())).Msg("LINE client connected; notifying bridge") - lc.UserLogin.BridgeState.Send(status.BridgeState{ - StateEvent: status.StateConnected, - }) + if !lc.sendConnectedStateIfCurrent(ctx) { + return + } // Initialize E2EE manager and load keys mgr, err := e2ee.NewManager() @@ -505,11 +645,15 @@ func (lc *LineClient) Connect(ctx context.Context) { // sender's existing membership look like a fresh join. lc.wg.Add(1) lc.syncChats(ctx) + if ctx.Err() != nil { + return + } lc.wg.Add(3) go lc.syncDMChats(ctx) go lc.prefetchMessages(ctx) go lc.pollLoop(ctx) + workersStarted = true } func (lc *LineClient) tryLogin(ctx context.Context) error { @@ -632,10 +776,30 @@ func (lc *LineClient) refreshLoginE2EEKeys(res *line.LoginResult, meta *UserLogi } func (lc *LineClient) ensureValidToken(ctx context.Context) error { - _, err := getProfileWithToken(lc.getAccessToken()) + return lc.ensureValidTokenWith( + ctx, + func(ctx context.Context) error { + _, err := getProfileWithToken(ctx, lc.getAccessToken()) + return err + }, + lc.refreshAndSave, + lc.tryLogin, + ) +} + +func (lc *LineClient) ensureValidTokenWith( + ctx context.Context, + profile func(context.Context) error, + refresh func(context.Context) error, + relogin func(context.Context) error, +) error { + err := profile(ctx) if err == nil { return nil } + if ctx.Err() != nil { + return ctx.Err() + } if lc.isLoggedOut(err) { return err @@ -647,19 +811,28 @@ func (lc *LineClient) ensureValidToken(ctx context.Context) error { } lc.UserLogin.Bridge.Log.Info().Msg("Access token expired, attempting refresh...") - if errRefresh := lc.refreshAndSave(ctx); errRefresh == nil { - lc.UserLogin.Bridge.Log.Info().Msg("Token refreshed successfully") - return nil - } else { - lc.UserLogin.Bridge.Log.Warn().Err(errRefresh).Msg("Token refresh failed") - } - - lc.UserLogin.Bridge.Log.Info().Msg("Attempting re-login with stored credentials...") - return lc.tryLogin(ctx) + return lc.recoverTokenWith(ctx, refresh, relogin) } func (lc *LineClient) Disconnect() { + // Disconnect is terminal for this NetworkAPI instance. Framework reconnects + // create a replacement client, so late handlers on this one must not mutate + // the shared UserLogin or start token recovery. + lc.superseded.Store(true) + lc.runMu.Lock() + lc.stopped = true + run := lc.activeRun + if run != nil { + run.cancel() + } + lc.runMu.Unlock() lc.wg.Wait() + // markLoggedOutByOtherClient and token recovery both hold recoverMu while + // applying/persisting session state. Drain any call that began before the + // superseded flag was visible before a replacement installs new metadata. + lc.recoverMu.Lock() + lc.recoveryStopped = true + lc.recoverMu.Unlock() } func (lc *LineClient) IsLoggedIn() bool { return lc.hasAccessToken() } diff --git a/pkg/connector/client_lifecycle_test.go b/pkg/connector/client_lifecycle_test.go new file mode 100644 index 0000000..e4d1a86 --- /dev/null +++ b/pkg/connector/client_lifecycle_test.go @@ -0,0 +1,310 @@ +package connector + +import ( + "context" + "errors" + "io" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/rs/zerolog" + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/database" + "maunium.net/go/mautrix/bridgev2/networkid" +) + +func TestLineClientDisconnectCancelsActiveRun(t *testing.T) { + lc := &LineClient{} + runCtx, _, started := lc.beginRun(context.Background()) + if !started { + t.Fatal("beginRun unexpectedly rejected startup") + } + workerStarted := make(chan struct{}) + lc.wg.Add(1) + go func() { + defer lc.wg.Done() + close(workerStarted) + <-runCtx.Done() + }() + <-workerStarted + + disconnected := make(chan struct{}) + go func() { + lc.Disconnect() + close(disconnected) + }() + select { + case <-runCtx.Done(): + case <-time.After(time.Second): + t.Fatal("Disconnect did not cancel the active run") + } + select { + case <-disconnected: + t.Fatal("Disconnect returned before the Connect reservation was released") + default: + } + lc.wg.Done() // release the reservation owned by the simulated Connect call + + select { + case <-disconnected: + case <-time.After(time.Second): + t.Fatal("Disconnect did not cancel and join the active run") + } + if !errors.Is(runCtx.Err(), context.Canceled) { + t.Fatalf("run context error = %v, want context.Canceled", runCtx.Err()) + } +} + +func TestLineClientDisconnectBeforeConnectRejectsStartup(t *testing.T) { + lc := &LineClient{} + lc.Disconnect() + lc.recoverMu.Lock() + recoveryStopped := lc.recoveryStopped + lc.recoverMu.Unlock() + if !recoveryStopped { + t.Fatal("Disconnect did not stop token recovery") + } + runCtx, _, started := lc.beginRun(context.Background()) + if started { + lc.wg.Done() + t.Fatal("beginRun started after Disconnect completed") + } + if !errors.Is(runCtx.Err(), context.Canceled) { + t.Fatalf("rejected run context error = %v, want context.Canceled", runCtx.Err()) + } +} + +func TestCreateLoginSharesFinalizationLock(t *testing.T) { + connector := &LineConnector{} + firstProcess, err := connector.CreateLogin(context.Background(), nil, "") + if err != nil { + t.Fatalf("first CreateLogin returned error: %v", err) + } + secondProcess, err := connector.CreateLogin(context.Background(), nil, "") + if err != nil { + t.Fatalf("second CreateLogin returned error: %v", err) + } + first := firstProcess.(*LineEmailLogin) + second := secondProcess.(*LineEmailLogin) + if first.finalizeMu == nil || first.finalizeMu != second.finalizeMu || first.finalizeMu != &connector.loginFinalizeMu { + t.Fatal("login processes do not share the connector finalization lock") + } +} + +func TestFindReusedLineLogin(t *testing.T) { + logins := make(map[string]*bridgev2.UserLogin, 3) + for _, loginID := range []string{"override", "target", "unrelated"} { + logins[loginID] = &bridgev2.UserLogin{ + UserLogin: &database.UserLogin{ID: networkid.UserLoginID(loginID)}, + } + } + tests := []struct { + name string + overrideID string + detectedID string + wantID string + }{ + { + name: "ordinary same-ID login", + detectedID: "target", + wantID: "target", + }, + { + name: "ordinary new account", + detectedID: "new-account", + }, + { + name: "same-ID override", + overrideID: "target", + detectedID: "target", + wantID: "target", + }, + { + name: "override resolves to another existing account", + overrideID: "override", + detectedID: "target", + wantID: "target", + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var override *bridgev2.UserLogin + if tc.overrideID != "" { + override = logins[tc.overrideID] + } + got := findReusedLineLogin( + override, + []*bridgev2.UserLogin{logins["override"], logins["target"], logins["unrelated"]}, + networkid.UserLoginID(tc.detectedID), + ) + if tc.wantID == "" { + if got != nil { + t.Fatalf("reused login = %s, want nil", got.ID) + } + } else if got != logins[tc.wantID] { + t.Fatalf("reused login = %p, want %p (%s)", got, logins[tc.wantID], tc.wantID) + } + }) + } +} + +func TestSameIDReplacementRetiresOldClientBeforeMetadataChange(t *testing.T) { + oldMeta := &UserLoginMetadata{AccessToken: "old-token", Mid: "target"} + login := &bridgev2.UserLogin{ + UserLogin: &database.UserLogin{ + ID: "target", + Metadata: oldMeta, + }, + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + } + stale := &LineClient{AccessToken: "old-token", UserLogin: login} + login.Client = stale + + retireLineLogin(login) + if !stale.superseded.Load() { + t.Fatal("same-ID client was not retired before metadata replacement") + } + freshMeta := &UserLoginMetadata{AccessToken: "fresh-token", Mid: "target"} + login.Metadata = freshMeta + fresh := &LineClient{AccessToken: "fresh-token", UserLogin: login} + login.Client = fresh + if login.Client != fresh { + t.Fatal("replacement client was not published") + } + + stale.markLoggedOutByOtherClient(context.Background(), errLoggedOut) + + if freshMeta.AccessToken != "fresh-token" || freshMeta.SessionInvalidated { + t.Fatalf("retired client changed replacement metadata: %#v", freshMeta) + } +} + +func TestForcedLogoutCancelsActiveRun(t *testing.T) { + lc := &LineClient{AccessToken: "stale"} + runCtx, _, started := lc.beginRun(context.Background()) + if !started { + t.Fatal("beginRun unexpectedly rejected startup") + } + defer lc.wg.Done() + + lc.markLoggedOutByOtherClient(runCtx, errLoggedOut) + + select { + case <-runCtx.Done(): + case <-time.After(time.Second): + t.Fatal("forced logout did not cancel the active run") + } + if lc.hasAccessToken() { + t.Fatal("forced logout did not clear the access token") + } + if !lc.isSessionInvalidated() { + t.Fatal("forced logout did not invalidate the session") + } +} + +func TestStaleForcedLogoutCancelsOnlyStaleClient(t *testing.T) { + login := &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + } + stale := &LineClient{AccessToken: "stale-token", UserLogin: login} + current := &LineClient{AccessToken: "current-token", UserLogin: login} + login.Client = current + staleCtx, _, staleStarted := stale.beginRun(context.Background()) + currentCtx, _, currentStarted := current.beginRun(context.Background()) + if !staleStarted || !currentStarted { + t.Fatal("beginRun unexpectedly rejected startup") + } + defer stale.wg.Done() + defer current.wg.Done() + defer current.cancelActiveRun() + + stale.markLoggedOutByOtherClient(staleCtx, errLoggedOut) + + select { + case <-staleCtx.Done(): + case <-time.After(time.Second): + t.Fatal("stale client run was not canceled") + } + select { + case <-currentCtx.Done(): + t.Fatal("stale client logout canceled the replacement client") + default: + } + if current.getAccessToken() != "current-token" || current.isSessionInvalidated() { + t.Fatal("stale client logout changed replacement session state") + } + if stale.getAccessToken() != "stale-token" || stale.isSessionInvalidated() { + t.Fatal("stale client logout was not rejected by client ownership") + } +} + +func TestRetiredClientCannotRestart(t *testing.T) { + lc := &LineClient{} + lc.retire() + if !lc.superseded.Load() { + t.Fatal("retired client was not marked superseded") + } + runCtx, _, started := lc.beginRun(context.Background()) + if started { + lc.wg.Done() + t.Fatal("retired client accepted a new run") + } + if !errors.Is(runCtx.Err(), context.Canceled) { + t.Fatalf("retired run context error = %v, want context.Canceled", runCtx.Err()) + } +} + +func TestRepeatedForcedLogoutStillInvalidatesAndCancels(t *testing.T) { + lc := &LineClient{ + AccessToken: "stale-token", + forcedLogoutSent: true, + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + runCtx, _, started := lc.beginRun(context.Background()) + if !started { + t.Fatal("beginRun unexpectedly rejected startup") + } + defer lc.wg.Done() + + lc.markLoggedOutByOtherClient(runCtx, errLoggedOut) + + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("repeated forced logout did not keep the session invalidated") + } + select { + case <-runCtx.Done(): + case <-time.After(time.Second): + t.Fatal("repeated forced logout did not cancel the active run") + } +} + +func TestConnectedStateRejectedAfterSessionInvalidation(t *testing.T) { + lc := &LineClient{AccessToken: "stale-token", sessionInvalidated: true} + if lc.sendConnectedStateIfCurrent(context.Background()) { + t.Fatal("invalidated session was allowed to emit CONNECTED") + } +} + +func TestClaimForcedLogoutStateOnlyOnce(t *testing.T) { + lc := &LineClient{} + var claims atomic.Int32 + var wg sync.WaitGroup + for range 32 { + wg.Add(1) + go func() { + defer wg.Done() + if lc.claimForcedLogoutState() { + claims.Add(1) + } + }() + } + wg.Wait() + if got := claims.Load(); got != 1 { + t.Fatalf("forced logout state claims = %d, want 1", got) + } +} diff --git a/pkg/connector/connector.go b/pkg/connector/connector.go index efe640c..4dd7d81 100644 --- a/pkg/connector/connector.go +++ b/pkg/connector/connector.go @@ -26,12 +26,17 @@ const ( ) type LineConnector struct { - br *bridgev2.Bridge + br *bridgev2.Bridge + loginFinalizeMu sync.Mutex } var _ bridgev2.NetworkConnector = (*LineConnector)(nil) func (lc *LineConnector) Init(bridge *bridgev2.Bridge) { + // Keep connection state in Beeper's bridge status API only. Framework status + // notices call GetManagementRoom, which creates the LINE bot room when the + // user doesn't already have one. + bridge.Config.BridgeStatusNotices = "none" lc.br = bridge } @@ -146,7 +151,7 @@ func (lc *LineConnector) GetLoginFlows() []bridgev2.LoginFlow { } func (lc *LineConnector) CreateLogin(ctx context.Context, user *bridgev2.User, flowID string) (bridgev2.LoginProcess, error) { - return &LineEmailLogin{User: user}, nil + return &LineEmailLogin{User: user, finalizeMu: &lc.loginFinalizeMu}, nil } type LineEmailLogin struct { @@ -159,11 +164,13 @@ type LineEmailLogin struct { NoE2EE bool // True when login fell back to non-E2EE (LSOFF account) ExistingMetadata *UserLoginMetadata + ExistingLogin *bridgev2.UserLogin pollResult chan *line.LoginResult pollErr chan error polling bool mu sync.Mutex + finalizeMu *sync.Mutex } var _ bridgev2.LoginProcessUserInput = (*LineEmailLogin)(nil) @@ -201,6 +208,7 @@ func (ll *LineEmailLogin) StartWithOverride(ctx context.Context, override *bridg ll.Password = meta.Password ll.Certificate = meta.Certificate ll.ExistingMetadata = meta + ll.ExistingLogin = override if ll.Email == "" || ll.Password == "" { return ll.loginErrorStep("No stored LINE credentials are available. Please enter your LINE email and password to reconnect."), nil @@ -458,29 +466,50 @@ func (ll *LineEmailLogin) finishLogin(ctx context.Context, res *line.LoginResult } detectedLineID := networkid.UserLoginID(mid) + // bridgev2 reuses same-ID UserLogin objects and replaces their metadata + // before replacing Client. Serialize that handoff so concurrent login + // completions cannot orphan each other's clients. + if ll.finalizeMu != nil { + ll.finalizeMu.Lock() + defer ll.finalizeMu.Unlock() + } + targetLogin := findReusedLineLogin(ll.ExistingLogin, ll.User.GetUserLogins(), detectedLineID) + retireLineLogin(targetLogin) + var newClient *LineClient ul, err := ll.User.NewLogin(ctx, &database.UserLogin{ ID: detectedLineID, RemoteName: displayName, Metadata: meta, }, &bridgev2.NewLoginParams{ LoadUserLogin: func(ctx context.Context, login *bridgev2.UserLogin) error { - login.Client = &LineClient{ + newClient = &LineClient{ UserLogin: login, AccessToken: token, RefreshToken: refreshToken, Mid: mid, HTTPClient: &http.Client{Timeout: 10 * time.Second}, } + login.Client = newClient return nil }, }) if err != nil { + if newClient != nil { + newClient.Disconnect() + } return nil, fmt.Errorf("failed to create user login: %w", err) } + if newClient == nil { + return nil, fmt.Errorf("failed to create LINE client") + } + if ll.ExistingLogin != nil && ll.ExistingLogin.UserLogin != nil && ll.ExistingLogin.ID != ul.ID { + // A different-ID override remains valid until the new login is safely + // installed. Retire it only after NewLogin succeeds. + retireLineLogin(ll.ExistingLogin) + } - go ul.Client.Connect(context.Background()) - ul.BridgeState.Send(status.BridgeState{StateEvent: status.StateConnected}) + go newClient.Connect(context.Background()) return &bridgev2.LoginStep{ Type: bridgev2.LoginStepTypeComplete, @@ -490,6 +519,31 @@ func (ll *LineEmailLogin) finishLogin(ctx context.Context, res *line.LoginResult }, nil } +func retireLineLogin(login *bridgev2.UserLogin) { + if login == nil { + return + } + if client, ok := login.Client.(*LineClient); ok { + client.retire() + } +} + +func findReusedLineLogin( + override *bridgev2.UserLogin, + current []*bridgev2.UserLogin, + detectedID networkid.UserLoginID, +) *bridgev2.UserLogin { + if override != nil && override.UserLogin != nil && override.ID == detectedID { + return override + } + for _, login := range current { + if login != nil && login.UserLogin != nil && login.ID == detectedID { + return login + } + } + return nil +} + func exportLoginE2EEKeys(res *line.LoginResult, client *line.Client) (*e2ee.Manager, map[string]string, error) { if res.EncryptedKeyChain == "" || res.E2EEPublicKey == "" { return nil, nil, nil diff --git a/pkg/connector/connector_test.go b/pkg/connector/connector_test.go new file mode 100644 index 0000000..fcff61f --- /dev/null +++ b/pkg/connector/connector_test.go @@ -0,0 +1,166 @@ +package connector + +import ( + "context" + "errors" + "io" + "sync/atomic" + "testing" + "time" + + "github.com/rs/zerolog" + + "maunium.net/go/mautrix" + "maunium.net/go/mautrix/bridgev2" + "maunium.net/go/mautrix/bridgev2/bridgeconfig" + "maunium.net/go/mautrix/bridgev2/database" + "maunium.net/go/mautrix/bridgev2/status" + "maunium.net/go/mautrix/event" + "maunium.net/go/mautrix/id" +) + +type bridgeStateTestMatrix struct { + bridgev2.MatrixConnector + states chan status.BridgeState +} + +func (matrix *bridgeStateTestMatrix) GetCapabilities() *bridgev2.MatrixCapabilities { + return &bridgev2.MatrixCapabilities{AutoJoinInvites: true} +} + +func (matrix *bridgeStateTestMatrix) NewUserIntent(_ context.Context, _ id.UserID, accessToken string) (bridgev2.MatrixAPI, string, error) { + return nil, accessToken, nil +} + +func (matrix *bridgeStateTestMatrix) SendBridgeStatus(ctx context.Context, state *status.BridgeState) error { + select { + case matrix.states <- *state: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +type bridgeStateTestBot struct { + bridgev2.MatrixAPI + createRoomCalls atomic.Int32 + sendMessageCalls atomic.Int32 +} + +func (bot *bridgeStateTestBot) GetMXID() id.UserID { + return "@linebot:example.com" +} + +func (bot *bridgeStateTestBot) CreateRoom(context.Context, *mautrix.ReqCreateRoom) (id.RoomID, error) { + bot.createRoomCalls.Add(1) + return "", errors.New("unexpected management room creation") +} + +func (bot *bridgeStateTestBot) SendMessage( + context.Context, + id.RoomID, + event.Type, + *event.Content, + *bridgev2.MatrixSendExtra, +) (*mautrix.RespSendEvent, error) { + bot.sendMessageCalls.Add(1) + return &mautrix.RespSendEvent{}, nil +} + +func TestInitDisablesManagementRoomBridgeStatusNotices(t *testing.T) { + for _, configuredValue := range []string{"errors", "all", "none"} { + t.Run(configuredValue, func(t *testing.T) { + bridge := &bridgev2.Bridge{ + Config: &bridgeconfig.BridgeConfig{BridgeStatusNotices: configuredValue}, + } + connector := &LineConnector{} + + connector.Init(bridge) + + if connector.br != bridge { + t.Fatal("Init did not retain the bridge") + } + if bridge.Config.BridgeStatusNotices != "none" { + t.Fatalf("BridgeStatusNotices = %q, want none", bridge.Config.BridgeStatusNotices) + } + }) + } +} + +func TestBridgeStateUpdatesDoNotUseManagementRoom(t *testing.T) { + tests := map[string]id.RoomID{ + "room must not be created": "", + "existing room stays silent": "!management:example.com", + } + + for name, managementRoom := range tests { + t.Run(name, func(t *testing.T) { + log := zerolog.New(io.Discard) + matrix := &bridgeStateTestMatrix{states: make(chan status.BridgeState, 2)} + bot := &bridgeStateTestBot{} + connector := &LineConnector{} + bridge := &bridgev2.Bridge{ + Config: &bridgeconfig.BridgeConfig{BridgeStatusNotices: "errors"}, + Matrix: matrix, + Bot: bot, + Network: connector, + Log: log, + BackgroundCtx: context.Background(), + } + connector.Init(bridge) + + userMXID := id.UserID("@user:example.com") + user := &bridgev2.User{ + User: &database.User{ + MXID: userMXID, + ManagementRoom: managementRoom, + }, + Bridge: bridge, + Log: log, + } + login := &bridgev2.UserLogin{ + UserLogin: &database.UserLogin{ + UserMXID: userMXID, + ID: "line-user", + RemoteName: "LINE User", + }, + Bridge: bridge, + User: user, + Log: log, + } + login.BridgeState = bridge.NewBridgeStateQueue(login) + t.Cleanup(login.BridgeState.Destroy) + + wantStates := []status.BridgeState{ + { + StateEvent: status.StateBadCredentials, + Error: "line-logged-out", + Message: "LINE logged this Chrome Extension session out because another LINE client connected.", + UserAction: status.UserActionRelogin, + }, + {StateEvent: status.StateConnected}, + } + for _, state := range wantStates { + login.BridgeState.Send(state) + } + + for _, want := range wantStates { + select { + case got := <-matrix.states: + if got.StateEvent != want.StateEvent || got.Error != want.Error || got.UserAction != want.UserAction { + t.Fatalf("bridge state = %#v, want event=%s error=%s action=%s", got, want.StateEvent, want.Error, want.UserAction) + } + case <-time.After(2 * time.Second): + t.Fatalf("timed out waiting for %s bridge state", want.StateEvent) + } + } + + if calls := bot.createRoomCalls.Load(); calls != 0 { + t.Fatalf("bot CreateRoom calls = %d, want 0", calls) + } + if calls := bot.sendMessageCalls.Load(); calls != 0 { + t.Fatalf("bot SendMessage calls = %d, want 0", calls) + } + }) + } +} diff --git a/pkg/connector/forced_logout_test.go b/pkg/connector/forced_logout_test.go index 9268ed4..572d09e 100644 --- a/pkg/connector/forced_logout_test.go +++ b/pkg/connector/forced_logout_test.go @@ -2,11 +2,15 @@ package connector import ( "context" + "io" "testing" + "time" "maunium.net/go/mautrix/bridgev2" "maunium.net/go/mautrix/bridgev2/database" + "github.com/rs/zerolog" + "github.com/highesttt/matrix-line-messenger/pkg/line" ) @@ -20,7 +24,7 @@ func TestEnsureValidTokenReturnsLoggedOutWithoutRelogin(t *testing.T) { var profileCalls int var loginCalls int - getProfileWithToken = func(token string) (*line.Profile, error) { + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { profileCalls++ if token != "expired" { t.Fatalf("profile token = %q, want expired", token) @@ -45,6 +49,82 @@ func TestEnsureValidTokenReturnsLoggedOutWithoutRelogin(t *testing.T) { } } +func TestEnsureValidTokenDoesNotReloginAfterLoggedOutRefresh(t *testing.T) { + lc := &LineClient{ + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + var reloginCalls int + err := lc.ensureValidTokenWith( + context.Background(), + func(context.Context) error { return errAuthRequired }, + func(context.Context) error { return errLoggedOut }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + if !line.IsLoggedOut(err) { + t.Fatalf("ensureValidTokenWith error = %v, want logged-out error", err) + } + if reloginCalls != 0 { + t.Fatalf("relogin calls = %d, want 0", reloginCalls) + } +} + +func TestForcedLogoutWinsOverEnsureValidTokenRefresh(t *testing.T) { + lc := &LineClient{ + AccessToken: "old-token", + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + refreshStarted := make(chan struct{}) + allowRefresh := make(chan struct{}) + ensureDone := make(chan error, 1) + var reloginCalls int + go func() { + ensureDone <- lc.ensureValidTokenWith( + context.Background(), + func(context.Context) error { return errAuthRequired }, + func(context.Context) error { + close(refreshStarted) + <-allowRefresh + lc.setTokens("recovered-token", "") + return nil + }, + func(context.Context) error { + reloginCalls++ + return nil + }, + ) + }() + <-refreshStarted + + logoutDone := make(chan struct{}) + go func() { + lc.markLoggedOutByOtherClient(context.Background(), errLoggedOut) + close(logoutDone) + }() + close(allowRefresh) + + if err := <-ensureDone; err != nil { + t.Fatalf("ensureValidTokenWith returned error: %v", err) + } + if reloginCalls != 0 { + t.Fatalf("relogin calls = %d, want 0", reloginCalls) + } + select { + case <-logoutDone: + case <-time.After(time.Second): + t.Fatal("forced logout did not complete after startup refresh") + } + if lc.hasAccessToken() || !lc.isSessionInvalidated() { + t.Fatal("startup refresh resurrected the forcefully logged-out session") + } +} + func TestStartWithOverrideUsesStoredCredentials(t *testing.T) { oldLogin := loginWithCredentials t.Cleanup(func() { @@ -72,7 +152,8 @@ func TestStartWithOverrideUsesStoredCredentials(t *testing.T) { }, } - step, err := (&LineEmailLogin{}).StartWithOverride(context.Background(), override) + login := &LineEmailLogin{} + step, err := login.StartWithOverride(context.Background(), override) if err != nil { t.Fatalf("StartWithOverride returned error: %v", err) } @@ -85,4 +166,7 @@ func TestStartWithOverrideUsesStoredCredentials(t *testing.T) { if step.StepID != "dev.highest.matrix.line.enter_pin" { t.Fatalf("step ID = %q, want enter PIN", step.StepID) } + if login.ExistingLogin != override { + t.Fatal("override login was not retained for retirement before replacement") + } } diff --git a/pkg/connector/sync.go b/pkg/connector/sync.go index 2d4f3b6..5fad3b5 100644 --- a/pkg/connector/sync.go +++ b/pkg/connector/sync.go @@ -26,22 +26,30 @@ import ( ) const ( - prefetchMessagesConcurrency = 4 - messageBoxPageLimit = 100 - startupBackfillMessageLimit = 50 - unblockBackfillFallbackDelay = 10 * time.Second - groupPortalCreateWait = 30 * time.Second - beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline" + prefetchMessagesConcurrency = 4 + messageBoxPageLimit = 100 + startupBackfillMessageLimit = 50 + unblockBackfillFallbackDelay = 10 * time.Second + groupPortalCreateWait = 30 * time.Second + beeperExcludeFromTimelineKey = "com.beeper.exclude_from_timeline" + defaultReceiveAuthProbeInterval = 150 * time.Second ) +var errReceiveAuthProbeDue = errors.New("receive auth probe due") + var ( - getLastOpRevisionWithClient = func(client *line.Client) (int64, error) { - return client.GetLastOpRevision() + getLastOpRevisionWithClient = func(ctx context.Context, client *line.Client) (int64, error) { + return client.GetLastOpRevisionContext(ctx) } listenSSEWithClient = func(client *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error { return client.ListenSSE(ctx, localRev, handler) } - sseReconnectDelay = 3 * time.Second + sseReconnectDelay = 3 * time.Second + receiveAuthProbeInterval = defaultReceiveAuthProbeInterval + receiveAuthProbeNow = time.Now + newReceiveAuthProbeContext = func(parent context.Context, deadline time.Time) (context.Context, context.CancelFunc) { + return context.WithDeadlineCause(parent, deadline, errReceiveAuthProbeDue) + } ) func (lc *LineClient) getMessageBoxesWithRecovery(ctx context.Context, opts line.MessageBoxesOptions) (*line.MessageBoxesResponse, error) { @@ -1153,6 +1161,40 @@ func (lc *LineClient) refreshGroupsForContact(ctx context.Context, mid string) { } } +func startReceiveAuthProbeContext(parent context.Context, startedAt time.Time) (context.Context, context.CancelFunc, time.Time) { + if receiveAuthProbeInterval <= 0 { + return parent, func() {}, time.Time{} + } + + nextProbeAt := startedAt.Add(receiveAuthProbeInterval) + if now := receiveAuthProbeNow(); !nextProbeAt.After(now) { + nextProbeAt = now.Add(receiveAuthProbeInterval) + } + receiveCtx, cancel := newReceiveAuthProbeContext(parent, nextProbeAt) + return receiveCtx, cancel, nextProbeAt +} + +func isReceiveAuthProbeDue(receiveCtx context.Context, nextProbeAt time.Time) bool { + if receiveAuthProbeInterval <= 0 { + return false + } + return errors.Is(context.Cause(receiveCtx), errReceiveAuthProbeDue) || !receiveAuthProbeNow().Before(nextProbeAt) +} + +func waitForSSEReconnect(ctx context.Context) bool { + if sseReconnectDelay <= 0 { + return ctx.Err() == nil + } + timer := time.NewTimer(sseReconnectDelay) + defer timer.Stop() + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + func (lc *LineClient) pollLoop(ctx context.Context) { defer lc.wg.Done() @@ -1160,7 +1202,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { client := lc.newClient() lc.UserLogin.Bridge.Log.Info().Msg("Starting LINE SSE loop...") - rev, err := getLastOpRevisionWithClient(client) + rev, err := getLastOpRevisionWithClient(ctx, client) if err != nil && lc.isLoggedOut(err) { lc.markLoggedOutByOtherClient(ctx, err) return @@ -1168,7 +1210,7 @@ func (lc *LineClient) pollLoop(ctx context.Context) { if err != nil && lc.shouldAttemptTokenRecovery(ctx, err) { if errRecover := lc.recoverToken(ctx); errRecover == nil { client = lc.newClient() - rev, err = getLastOpRevisionWithClient(client) + rev, err = getLastOpRevisionWithClient(ctx, client) } else { lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token for getLastOpRevision") } @@ -1180,6 +1222,20 @@ func (lc *LineClient) pollLoop(ctx context.Context) { lc.UserLogin.Bridge.Log.Info().Int64("local_rev", localRev).Msg("Seeded local revision from getLastOpRevision") } + receiveCtx, cancelReceive, nextProbeAt := startReceiveAuthProbeContext(ctx, receiveAuthProbeNow()) + defer func() { + cancelReceive() + }() + probeAndReschedule := func() bool { + probeStartedAt := receiveAuthProbeNow() + if lc.handleReceiveAuthProbe(ctx) || ctx.Err() != nil { + return true + } + cancelReceive() + receiveCtx, cancelReceive, nextProbeAt = startReceiveAuthProbeContext(ctx, probeStartedAt) + return false + } + handler := func(eventType, data string) { // handle keep alives if eventType == "ping" || eventType == "connInfoRevision" { @@ -1233,20 +1289,41 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } for { + if ctx.Err() != nil { + return + } + if isReceiveAuthProbeDue(receiveCtx, nextProbeAt) { + if probeAndReschedule() { + return + } + continue + } + client = lc.newClient() - err := listenSSEWithClient(client, ctx, localRev, handler) + err := listenSSEWithClient(client, receiveCtx, localRev, handler) if err != nil { + if ctx.Err() != nil { + return + } + if isReceiveAuthProbeDue(receiveCtx, nextProbeAt) { + if probeAndReschedule() { + return + } + continue + } if errors.Is(err, context.Canceled) { return } - if err.Error() != "EOF" { + if !errors.Is(err, io.EOF) { lc.UserLogin.Bridge.Log.Warn().Err(err).Msg("SSE Disconnected") if line.IsSSEIdleTimeout(err) { - if lc.handleReceiveIdleTimeout(ctx) { + if probeAndReschedule() { return } - time.Sleep(sseReconnectDelay) + if !waitForSSEReconnect(receiveCtx) { + continue + } continue } @@ -1261,21 +1338,33 @@ func (lc *LineClient) pollLoop(ctx context.Context) { } } } - time.Sleep(sseReconnectDelay) + if !waitForSSEReconnect(receiveCtx) { + continue + } } } } -// 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 { +// handleReceiveAuthProbe checks Talk auth independently of SSE activity. The +// caller reconnects from the existing localRev after every probe so operations +// that arrived during the check are replayed rather than skipped. +func (lc *LineClient) handleReceiveAuthProbe(ctx context.Context) bool { + if ctx.Err() != nil || lc.isSessionInvalidated() { + return true + } + if lc.superseded.Load() { + return true + } + // 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()) + _, probeErr := getLastOpRevisionWithClient(ctx, lc.newClient()) if probeErr == nil { return false } + if ctx.Err() != nil { + return true + } if lc.isLoggedOut(probeErr) { lc.markLoggedOutByOtherClient(ctx, probeErr) @@ -1293,14 +1382,14 @@ func (lc *LineClient) handleReceiveIdleTimeout(ctx context.Context) bool { 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") + lc.UserLogin.Bridge.Log.Warn().Err(errRecover).Msg("Failed to recover token after receive auth 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") + lc.UserLogin.Bridge.Log.Warn().Err(probeErr).Msg("Receive auth probe failed; reconnecting SSE") } return false } @@ -1314,7 +1403,10 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, err error) boo return true } - _, profileErr := getProfileWithToken(lc.getAccessToken()) + _, profileErr := getProfileWithToken(ctx, lc.getAccessToken()) + if ctx.Err() != nil { + return true + } if lc.isLoggedOut(profileErr) { lc.markLoggedOutByOtherClient(ctx, profileErr) return true @@ -1328,6 +1420,9 @@ func (lc *LineClient) handleReceiveAuthError(ctx context.Context, err error) boo } if errRecover := lc.recoverToken(ctx); errRecover != nil { + if ctx.Err() != nil { + return true + } if errors.Is(errRecover, errLineSessionInvalidated) || lc.isLoggedOut(errRecover) { lc.markLoggedOutByOtherClient(ctx, errRecover) return true diff --git a/pkg/connector/sync_test.go b/pkg/connector/sync_test.go index 31251f1..4336109 100644 --- a/pkg/connector/sync_test.go +++ b/pkg/connector/sync_test.go @@ -13,6 +13,87 @@ import ( "github.com/highesttt/matrix-line-messenger/pkg/line" ) +func installManualReceiveAuthProbeDeadline(t *testing.T) <-chan context.CancelCauseFunc { + t.Helper() + oldInterval := receiveAuthProbeInterval + oldNow := receiveAuthProbeNow + oldNewContext := newReceiveAuthProbeContext + t.Cleanup(func() { + receiveAuthProbeInterval = oldInterval + receiveAuthProbeNow = oldNow + newReceiveAuthProbeContext = oldNewContext + }) + + fixedNow := time.Unix(1_700_000_000, 0) + receiveAuthProbeInterval = 150 * time.Second + receiveAuthProbeNow = func() time.Time { return fixedNow } + cancels := make(chan context.CancelCauseFunc, 8) + newReceiveAuthProbeContext = func(parent context.Context, _ time.Time) (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancelCause(parent) + cancels <- cancel + return ctx, func() { cancel(context.Canceled) } + } + return cancels +} + +func TestDefaultReceiveAuthProbeIntervalIs150Seconds(t *testing.T) { + if defaultReceiveAuthProbeInterval != 150*time.Second { + t.Fatalf("default receive auth probe interval = %s, want 150s", defaultReceiveAuthProbeInterval) + } + oldInterval := receiveAuthProbeInterval + oldNow := receiveAuthProbeNow + oldNewContext := newReceiveAuthProbeContext + t.Cleanup(func() { + receiveAuthProbeInterval = oldInterval + receiveAuthProbeNow = oldNow + newReceiveAuthProbeContext = oldNewContext + }) + + startedAt := time.Unix(1_700_000_000, 0) + receiveAuthProbeInterval = defaultReceiveAuthProbeInterval + receiveAuthProbeNow = func() time.Time { return startedAt } + var capturedDeadline time.Time + newReceiveAuthProbeContext = func(parent context.Context, deadline time.Time) (context.Context, context.CancelFunc) { + capturedDeadline = deadline + return context.WithCancel(parent) + } + _, cancel, nextProbeAt := startReceiveAuthProbeContext(context.Background(), startedAt) + cancel() + + want := startedAt.Add(150 * time.Second) + if !capturedDeadline.Equal(want) || !nextProbeAt.Equal(want) { + t.Fatalf("probe deadline = %s / %s, want %s", capturedDeadline, nextProbeAt, want) + } +} + +func TestReceiveAuthProbeContextExpiresOnSchedule(t *testing.T) { + oldInterval := receiveAuthProbeInterval + oldNow := receiveAuthProbeNow + oldNewContext := newReceiveAuthProbeContext + t.Cleanup(func() { + receiveAuthProbeInterval = oldInterval + receiveAuthProbeNow = oldNow + newReceiveAuthProbeContext = oldNewContext + }) + + receiveAuthProbeInterval = 10 * time.Millisecond + receiveAuthProbeNow = time.Now + newReceiveAuthProbeContext = func(parent context.Context, deadline time.Time) (context.Context, context.CancelFunc) { + return context.WithDeadlineCause(parent, deadline, errReceiveAuthProbeDue) + } + receiveCtx, cancel, _ := startReceiveAuthProbeContext(context.Background(), receiveAuthProbeNow()) + defer cancel() + + select { + case <-receiveCtx.Done(): + case <-time.After(time.Second): + t.Fatal("receive auth probe context did not reach its deadline") + } + if !errors.Is(context.Cause(receiveCtx), errReceiveAuthProbeDue) { + t.Fatalf("context cause = %v, want errReceiveAuthProbeDue", context.Cause(receiveCtx)) + } +} + func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) { oldGetLastOpRevision := getLastOpRevisionWithClient oldListenSSE := listenSSEWithClient @@ -23,7 +104,7 @@ func TestPollLoopRebuildsSSEClientAfterReconnect(t *testing.T) { sseReconnectDelay = oldReconnectDelay }) - getLastOpRevisionWithClient = func(*line.Client) (int64, error) { + getLastOpRevisionWithClient = func(context.Context, *line.Client) (int64, error) { return 1234, nil } sseReconnectDelay = time.Millisecond @@ -73,12 +154,12 @@ func TestPollLoopMarksLoggedOutWhenReceiveAuthFails(t *testing.T) { getProfileWithToken = oldGetProfile }) - getLastOpRevisionWithClient = func(*line.Client) (int64, error) { + getLastOpRevisionWithClient = func(context.Context, *line.Client) (int64, error) { return 1234, nil } var profileCalls int - getProfileWithToken = func(token string) (*line.Profile, error) { + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { profileCalls++ if token != "stale" { t.Fatalf("profile token = %q, want stale", token) @@ -123,7 +204,7 @@ func TestPollLoopMarksLoggedOutWhenReceiveIdleProbeFailsLoggedOut(t *testing.T) }) var revisionCalls int - getLastOpRevisionWithClient = func(client *line.Client) (int64, error) { + getLastOpRevisionWithClient = func(_ context.Context, client *line.Client) (int64, error) { if client.AccessToken != "stale" { t.Fatalf("revision probe token = %q, want stale", client.AccessToken) } @@ -176,7 +257,7 @@ func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) { }) var revisionCalls int - getLastOpRevisionWithClient = func(client *line.Client) (int64, error) { + getLastOpRevisionWithClient = func(_ context.Context, client *line.Client) (int64, error) { if client.AccessToken != "valid" { t.Fatalf("revision probe token = %q, want valid", client.AccessToken) } @@ -233,13 +314,216 @@ func TestPollLoopReconnectsWhenReceiveIdleProbeSucceeds(t *testing.T) { } } +func TestPollLoopReceiveAuthDeadlineIgnoresHeartbeats(t *testing.T) { + deadlineCancels := installManualReceiveAuthProbeDeadline(t) + oldGetLastOpRevision := getLastOpRevisionWithClient + oldListenSSE := listenSSEWithClient + t.Cleanup(func() { + getLastOpRevisionWithClient = oldGetLastOpRevision + listenSSEWithClient = oldListenSSE + }) + + var revisionCalls int + getLastOpRevisionWithClient = func(_ context.Context, 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 + } + + var heartbeatCalls int + listenSSEWithClient = func(_ *line.Client, ctx context.Context, localRev int64, handler func(eventType, data string)) error { + if localRev != 1234 { + t.Fatalf("localRev = %d, want 1234", localRev) + } + for range 5 { + heartbeatCalls++ + handler("ping", "null") + handler("connInfoRevision", "42") + } + (<-deadlineCancels)(errReceiveAuthProbeDue) + <-ctx.Done() + return ctx.Err() + } + + lc := &LineClient{ + AccessToken: "stale", + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + lc.wg.Add(1) + lc.pollLoop(context.Background()) + + if heartbeatCalls != 5 { + t.Fatalf("heartbeat batches = %d, want 5", heartbeatCalls) + } + if revisionCalls != 2 { + t.Fatalf("revision calls = %d, want startup plus one deadline probe", revisionCalls) + } + if lc.hasAccessToken() { + t.Fatal("access token was not invalidated after deadline probe") + } + if !lc.isSessionInvalidated() { + t.Fatal("session was not marked invalidated after deadline probe") + } +} + +func TestPollLoopReceiveAuthDeadlineSurvivesEOFReconnects(t *testing.T) { + deadlineCancels := installManualReceiveAuthProbeDeadline(t) + oldGetLastOpRevision := getLastOpRevisionWithClient + oldListenSSE := listenSSEWithClient + oldReconnectDelay := sseReconnectDelay + t.Cleanup(func() { + getLastOpRevisionWithClient = oldGetLastOpRevision + listenSSEWithClient = oldListenSSE + sseReconnectDelay = oldReconnectDelay + }) + sseReconnectDelay = 0 + + var revisionCalls int + getLastOpRevisionWithClient = func(context.Context, *line.Client) (int64, error) { + revisionCalls++ + if revisionCalls == 1 { + return 1234, nil + } + return 0, errLoggedOut + } + + var listenCalls int + listenSSEWithClient = func(_ *line.Client, _ context.Context, localRev int64, _ func(eventType, data string)) error { + if localRev != 1234 { + t.Fatalf("localRev = %d, want 1234", localRev) + } + listenCalls++ + if listenCalls == 3 { + (<-deadlineCancels)(errReceiveAuthProbeDue) + } + return io.EOF + } + + lc := &LineClient{ + AccessToken: "stale", + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + lc.wg.Add(1) + lc.pollLoop(context.Background()) + + if listenCalls != 3 { + t.Fatalf("SSE attempts = %d, want 3", listenCalls) + } + if revisionCalls != 2 { + t.Fatalf("revision calls = %d, want startup plus one deadline probe", revisionCalls) + } +} + +func TestPollLoopSuccessfulReceiveAuthProbePreservesLocalRev(t *testing.T) { + deadlineCancels := installManualReceiveAuthProbeDeadline(t) + oldGetLastOpRevision := getLastOpRevisionWithClient + oldListenSSE := listenSSEWithClient + t.Cleanup(func() { + getLastOpRevisionWithClient = oldGetLastOpRevision + listenSSEWithClient = oldListenSSE + }) + + var revisionCalls int + getLastOpRevisionWithClient = func(context.Context, *line.Client) (int64, error) { + revisionCalls++ + if revisionCalls == 1 { + return 1234, nil + } + return 5678, nil + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var listenCalls int + listenSSEWithClient = func(_ *line.Client, receiveCtx context.Context, localRev int64, _ func(eventType, data string)) error { + if localRev != 1234 { + t.Fatalf("localRev = %d, want health probe result to remain ignored", localRev) + } + listenCalls++ + if listenCalls == 1 { + (<-deadlineCancels)(errReceiveAuthProbeDue) + <-receiveCtx.Done() + return receiveCtx.Err() + } + cancel() + <-receiveCtx.Done() + return receiveCtx.Err() + } + + 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 startup plus one successful probe", revisionCalls) + } + if listenCalls != 2 { + t.Fatalf("SSE attempts = %d, want reconnect after deadline probe", listenCalls) + } + if !lc.hasAccessToken() || lc.isSessionInvalidated() { + t.Fatal("successful health probe changed valid session state") + } +} + +func TestPollLoopParentCancellationDoesNotProbeAuth(t *testing.T) { + installManualReceiveAuthProbeDeadline(t) + oldGetLastOpRevision := getLastOpRevisionWithClient + oldListenSSE := listenSSEWithClient + t.Cleanup(func() { + getLastOpRevisionWithClient = oldGetLastOpRevision + listenSSEWithClient = oldListenSSE + }) + + var revisionCalls int + getLastOpRevisionWithClient = func(context.Context, *line.Client) (int64, error) { + revisionCalls++ + return 1234, nil + } + ctx, cancel := context.WithCancel(context.Background()) + listenSSEWithClient = func(_ *line.Client, receiveCtx context.Context, _ int64, _ func(eventType, data string)) error { + cancel() + <-receiveCtx.Done() + return receiveCtx.Err() + } + + lc := &LineClient{ + AccessToken: "valid", + UserLogin: &bridgev2.UserLogin{ + Bridge: &bridgev2.Bridge{Log: zerolog.New(io.Discard)}, + }, + } + lc.wg.Add(1) + lc.pollLoop(ctx) + + if revisionCalls != 1 { + t.Fatalf("revision calls = %d, want startup only", revisionCalls) + } + if !lc.hasAccessToken() || lc.isSessionInvalidated() { + t.Fatal("parent cancellation changed session state") + } +} + func TestReceiveRequestNeedLoginMarksLoggedOutImmediately(t *testing.T) { oldGetProfile := getProfileWithToken t.Cleanup(func() { getProfileWithToken = oldGetProfile }) - getProfileWithToken = func(token string) (*line.Profile, error) { + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { t.Fatal("REQUEST_NEED_LOGIN should be handled without probing profile") return nil, nil } @@ -265,7 +549,7 @@ func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { }) var profileCalls int - getProfileWithToken = func(token string) (*line.Profile, error) { + getProfileWithToken = func(_ context.Context, token string) (*line.Profile, error) { profileCalls++ if token != "valid" { t.Fatalf("profile token = %q, want valid", token) @@ -289,3 +573,41 @@ func TestReceiveAuthErrorWithValidProfileDoesNotRecover(t *testing.T) { t.Fatal("valid session was marked invalidated") } } + +func TestReceiveAuthErrorCancellationDuringProfileDoesNotInvalidate(t *testing.T) { + oldGetProfile := getProfileWithToken + t.Cleanup(func() { + getProfileWithToken = oldGetProfile + }) + + profileStarted := make(chan struct{}) + getProfileWithToken = func(ctx context.Context, token string) (*line.Profile, error) { + if token != "valid" { + t.Fatalf("profile token = %q, want valid", token) + } + close(profileStarted) + <-ctx.Done() + return nil, errLoggedOut + } + + ctx, cancel := context.WithCancel(context.Background()) + lc := &LineClient{AccessToken: "valid"} + result := make(chan bool, 1) + go func() { + result <- lc.handleReceiveAuthError(ctx, errors.New("SSE error: 401")) + }() + <-profileStarted + cancel() + + select { + case stopped := <-result: + if !stopped { + t.Fatal("canceled receive auth handler should stop") + } + case <-time.After(time.Second): + t.Fatal("receive auth handler did not stop after cancellation") + } + if !lc.hasAccessToken() || lc.isSessionInvalidated() { + t.Fatal("ordinary cancellation was misclassified as forced logout") + } +} diff --git a/pkg/line/client.go b/pkg/line/client.go index d65f06e..feed183 100644 --- a/pkg/line/client.go +++ b/pkg/line/client.go @@ -311,7 +311,11 @@ func (c *Client) GetRSAKeyInfo() (*RSAKeyInfo, error) { } func (c *Client) callRPC(service, method string, args ...interface{}) ([]byte, error) { - return c.callRPCWithBaseURL(BaseURL, service, method, args...) + return c.callRPCContext(context.Background(), service, method, args...) +} + +func (c *Client) callRPCContext(ctx context.Context, service, method string, args ...interface{}) ([]byte, error) { + return c.callRPCWithBaseURLContext(ctx, BaseURL, service, method, args...) } func (c *Client) callShopRPC(service, method string, args ...interface{}) ([]byte, error) { @@ -319,6 +323,10 @@ func (c *Client) callShopRPC(service, method string, args ...interface{}) ([]byt } func (c *Client) callRPCWithBaseURL(baseURL, service, method string, args ...interface{}) ([]byte, error) { + return c.callRPCWithBaseURLContext(context.Background(), baseURL, service, method, args...) +} + +func (c *Client) callRPCWithBaseURLContext(ctx context.Context, baseURL, service, method string, args ...interface{}) ([]byte, error) { url := fmt.Sprintf("%s/%s/%s", baseURL, service, method) var bodyBytes []byte @@ -332,7 +340,7 @@ func (c *Client) callRPCWithBaseURL(baseURL, service, method string, args ...int } } - req, err := http.NewRequest("POST", url, bytes.NewBuffer(bodyBytes)) + req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(bodyBytes)) if err != nil { return nil, fmt.Errorf("failed to create request: %w", err) } diff --git a/pkg/line/methods.go b/pkg/line/methods.go index b976573..6c2da2a 100644 --- a/pkg/line/methods.go +++ b/pkg/line/methods.go @@ -1,6 +1,7 @@ package line import ( + "context" "encoding/base64" "encoding/json" "fmt" @@ -100,7 +101,11 @@ func (c *Client) LoginV2WithVerifier(verifier string) (*LoginResult, error) { // GetProfile fetches the user's profile information func (c *Client) GetProfile() (*Profile, error) { - resp, err := c.callRPC("TalkService", "getProfile", 2) + return c.GetProfileContext(context.Background()) +} + +func (c *Client) GetProfileContext(ctx context.Context) (*Profile, error) { + resp, err := c.callRPCContext(ctx, "TalkService", "getProfile", 2) if err != nil { return nil, err } @@ -611,7 +616,11 @@ func (c *Client) GetChats(mids []string, withMembers, withInvitees bool) (*GetCh } func (c *Client) GetLastOpRevision() (int64, error) { - resp, err := c.callRPC("TalkService", "getLastOpRevision") + return c.GetLastOpRevisionContext(context.Background()) +} + +func (c *Client) GetLastOpRevisionContext(ctx context.Context) (int64, error) { + resp, err := c.callRPCContext(ctx, "TalkService", "getLastOpRevision") if err != nil { return 0, err } diff --git a/pkg/line/methods_context_test.go b/pkg/line/methods_context_test.go new file mode 100644 index 0000000..b6b0bcf --- /dev/null +++ b/pkg/line/methods_context_test.go @@ -0,0 +1,58 @@ +package line + +import ( + "context" + "errors" + "net/http" + "testing" + "time" +) + +func testRPCContextCancellation(t *testing.T, method string, call func(*Client, context.Context) error) { + t.Helper() + requestStarted := make(chan struct{}) + client := NewClient("valid-token") + client.HTTPClient = &http.Client{ + Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) { + close(requestStarted) + <-req.Context().Done() + return nil, req.Context().Err() + }), + } + + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + result <- call(client, ctx) + }() + + select { + case <-requestStarted: + case <-time.After(time.Second): + t.Fatalf("%s request did not start", method) + } + cancel() + + select { + case err := <-result: + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + case <-time.After(time.Second): + t.Fatalf("%s did not stop after context cancellation", method) + } +} + +func TestGetLastOpRevisionContextCancellation(t *testing.T) { + testRPCContextCancellation(t, "getLastOpRevision", func(client *Client, ctx context.Context) error { + _, err := client.GetLastOpRevisionContext(ctx) + return err + }) +} + +func TestGetProfileContextCancellation(t *testing.T) { + testRPCContextCancellation(t, "getProfile", func(client *Client, ctx context.Context) error { + _, err := client.GetProfileContext(ctx) + return err + }) +} diff --git a/pkg/line/sse_test.go b/pkg/line/sse_test.go index 7827f53..fdac72c 100644 --- a/pkg/line/sse_test.go +++ b/pkg/line/sse_test.go @@ -72,3 +72,66 @@ func TestListenSSEIdleTimeout(t *testing.T) { t.Fatalf("err = %v, want ErrSSEIdleTimeout", err) } } + +func TestListenSSEContextCancellationWithContinuousHeartbeats(t *testing.T) { + oldClient := sseHTTPClient + oldTimeout := sseIdleTimeout + t.Cleanup(func() { + sseHTTPClient = oldClient + sseIdleTimeout = oldTimeout + }) + + sseIdleTimeout = time.Second + pr, pw := io.Pipe() + 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.WithCancelCause(context.Background()) + defer cancel(context.Canceled) + listenResult := make(chan error, 1) + go func() { + listenResult <- NewClient("stale-token").ListenSSE(ctx, 0, func(event, data string) {}) + }() + + heartbeatsWritten := make(chan struct{}) + writerDone := make(chan struct{}) + go func() { + defer close(writerDone) + for range 5 { + if _, err := io.WriteString(pw, "event: ping\ndata: null\n\n"); err != nil { + return + } + } + close(heartbeatsWritten) + <-ctx.Done() + _ = pw.Close() + }() + + select { + case <-heartbeatsWritten: + case <-time.After(time.Second): + t.Fatal("SSE reader did not consume heartbeat frames") + } + cancel(errors.New("receive auth probe due")) + + select { + case err := <-listenResult: + if !errors.Is(err, context.Canceled) { + t.Fatalf("err = %v, want context.Canceled", err) + } + case <-time.After(time.Second): + t.Fatal("ListenSSE did not stop after context cancellation") + } + select { + case <-writerDone: + case <-time.After(time.Second): + t.Fatal("heartbeat writer did not stop after context cancellation") + } +}