PerUserTopicSubscriptionManager: Fix handling of 401 (unauthorized) errors

On a 401 (HTTP unauthorized) error, the correct response it to get a
new access token and try again (most likely reason for the error is that
the previous access token was expired).
PerUserTopicSubscriptionManager did this, but SubscriptionEntry's
SubscriptionFinishedCallback was a OnceCallback, and it wasn't re-set
for the second subscription attempt. So the result of the second attempt
was simply ignored: If it succeeded, we'd now be subscribed, but not
know about it (likely resulting in a redundant subscription request at
some later time). If it failed, it wouldn't get retried anymore, so
invalidations would be broken until at least the next Chrome restart.

This CL fixes this by simply making the callback a RepeatingCallback. It
also adds a test.

Bug: 1020117
Change-Id: Ifaf29a6db21ce9f37888603b534ef50970ffc94f
Reviewed-on: https://chromium-review.googlesource.com/c/chromium/src/+/2000720
Commit-Queue: Marc Treib <treib@chromium.org>
Reviewed-by: Tim Schumann <tschumann@chromium.org>
Reviewed-by: Mikel Astiz <mastiz@chromium.org>
Cr-Commit-Position: refs/heads/master@{#732055}

components/invalidation/impl/per_user_topic_subscription_manager.cc

@@ -54,11 +54,13 @@ const char kFCMOAuthScope[] =
 // Note: Taking |topic| and |private_topic_name| by value (rather than const
 // ref) because the caller (in practice, SubscriptionEntry) may be destroyed by
 // the callback.
-using SubscriptionFinishedCallback =
-    base::OnceCallback<void(Topic topic,
-                            Status code,
-                            std::string private_topic_name,
-                            PerUserTopicSubscriptionRequest::RequestType type)>;
+// This is a RepeatingCallback because in case of failure, the request will get
+// retried, so it might actually run multiple times.
+using SubscriptionFinishedCallback = base::RepeatingCallback<void(
+    Topic topic,
+    Status code,
+    std::string private_topic_name,
+    PerUserTopicSubscriptionRequest::RequestType type)>;
 
 static const net::BackoffEntry::Policy kBackoffPolicy = {
     // Number of initial errors (in sequence) to ignore before applying
@@ -201,8 +203,7 @@ PerUserTopicSubscriptionManager::SubscriptionEntry::~SubscriptionEntry() {}
 void PerUserTopicSubscriptionManager::SubscriptionEntry::SubscriptionFinished(
     const Status& code,
     const std::string& topic_name) {
-  if (completion_callback)
-    std::move(completion_callback).Run(topic, code, topic_name, type);
+  completion_callback.Run(topic, code, topic_name, type);
 }
 
 void PerUserTopicSubscriptionManager::SubscriptionEntry::Cancel() {
@@ -287,7 +288,7 @@ void PerUserTopicSubscriptionManager::UpdateSubscribedTopics(
 
       pending_subscriptions_[topic.first] = std::make_unique<SubscriptionEntry>(
           topic.first,
-          base::BindOnce(
+          base::BindRepeating(
               &PerUserTopicSubscriptionManager::SubscriptionFinishedForTopic,
               base::Unretained(this)),
           PerUserTopicSubscriptionRequest::SUBSCRIBE, topic.second.is_public);
@@ -305,7 +306,7 @@ void PerUserTopicSubscriptionManager::UpdateSubscribedTopics(
       // topic, we should probably cancel it first?
       pending_subscriptions_[topic] = std::make_unique<SubscriptionEntry>(
           topic,
-          base::BindOnce(
+          base::BindRepeating(
               &PerUserTopicSubscriptionManager::SubscriptionFinishedForTopic,
               base::Unretained(this)),
           PerUserTopicSubscriptionRequest::UNSUBSCRIBE);
@@ -406,9 +407,6 @@ void PerUserTopicSubscriptionManager::ActOnSuccessfulSubscription(
 
 void PerUserTopicSubscriptionManager::ScheduleRequestForRepetition(
     const Topic& topic) {
-  pending_subscriptions_[topic]->completion_callback = base::BindOnce(
-      &PerUserTopicSubscriptionManager::SubscriptionFinishedForTopic,
-      base::Unretained(this));
   pending_subscriptions_[topic]->request_backoff_.InformOfRequest(false);
   pending_subscriptions_[topic]->request_retry_timer_.Start(
       FROM_HERE,
@@ -473,6 +471,9 @@ void PerUserTopicSubscriptionManager::RequestAccessToken() {
   // Only one active request at a time.
   if (access_token_fetcher_ != nullptr)
     return;
+  // TODO(crbug.com/1020117): If the timer is already running, then this method
+  // should probably early-out instead of starting a request immediately. As it
+  // is, this might bypass the exponential backoff.
   request_access_token_retry_timer_.Stop();
   OAuth2AccessTokenManager::ScopeSet oauth2_scopes = {kFCMOAuthScope};
   // Invalidate previous token, otherwise the identity provider will return the
@@ -499,6 +500,8 @@ void PerUserTopicSubscriptionManager::OnAccessTokenRequestCompleted(
 void PerUserTopicSubscriptionManager::OnAccessTokenRequestSucceeded(
     const std::string& access_token) {
   // Reset backoff time after successful response.
+  // TODO(crbug.com/1020117): This should probably be .InformOfRequest(true)
+  // rather than .Reset().
   request_access_token_backoff_.Reset();
   access_token_ = access_token;
   // Emit ENABLED when successfully got the token.

components/invalidation/impl/per_user_topic_subscription_manager_unittest.cc

@@ -185,6 +185,10 @@ class PerUserTopicSubscriptionManagerTest : public testing::Test {
     task_environment_.FastForwardBy(delta);
   }
 
+  signin::IdentityTestEnvironment* identity_test_env() {
+    return &identity_test_env_;
+  }
+
  private:
   base::test::SingleThreadTaskEnvironment task_environment_{
       base::test::TaskEnvironment::TimeSource::MOCK_TIME};
@@ -296,6 +300,56 @@ TEST_F(PerUserTopicSubscriptionManagerTest, ShouldRepeatRequestsOnFailure) {
       per_user_topic_subscription_manager->HaveAllRequestsFinishedForTest());
 }
 
+TEST_F(PerUserTopicSubscriptionManagerTest,
+       ShouldInvalidateAccessTokenOnUnauthorized) {
+  // For this test, we need to manually control when access tokens are returned.
+  identity_test_env()->SetAutomaticIssueOfAccessTokens(false);
+
+  auto ids = GetSequenceOfTopics(kInvalidationObjectIdsCount);
+
+  auto per_user_topic_subscription_manager = BuildRegistrationManager();
+  ASSERT_TRUE(per_user_topic_subscription_manager->GetSubscribedTopicsForTest()
+                  .empty());
+
+  // The first subscription attempt will fail with an "unauthorized" error.
+  AddCorrectSubscriptionResponce(
+      /*private_topic=*/std::string(), kFakeInstanceIdToken,
+      net::HTTP_UNAUTHORIZED);
+
+  per_user_topic_subscription_manager->UpdateSubscribedTopics(
+      ids, kFakeInstanceIdToken);
+  // This should have resulted in a request for an access token. Return one
+  // (which is considered invalid, e.g. already expired).
+  identity_test_env()->WaitForAccessTokenRequestIfNecessaryAndRespondWithToken(
+      "invalid_access_token", base::Time::Now());
+
+  // Now the subscription requests should be scheduled.
+  ASSERT_FALSE(
+      per_user_topic_subscription_manager->HaveAllRequestsFinishedForTest());
+
+  // Wait for the subscription requests to happen.
+  base::RunLoop().RunUntilIdle();
+
+  // Since the subscriptions failed, the requests should still be pending.
+  ASSERT_FALSE(
+      per_user_topic_subscription_manager->HaveAllRequestsFinishedForTest());
+
+  // A new access token should have been requested. (Note: We'd really want to
+  // check that the previous token got invalidated before a new one was
+  // requested, but the identity test code doesn't expose that.)
+  // Serving a new access token will trigger another subscription attempt; let
+  // this one succeed.
+  AddCorrectSubscriptionResponce();
+  identity_test_env()->WaitForAccessTokenRequestIfNecessaryAndRespondWithToken(
+      "valid_access_token", base::Time::Max());
+  base::RunLoop().RunUntilIdle();
+
+  EXPECT_FALSE(per_user_topic_subscription_manager->GetSubscribedTopicsForTest()
+                   .empty());
+  EXPECT_TRUE(
+      per_user_topic_subscription_manager->HaveAllRequestsFinishedForTest());
+}
+
 TEST_F(PerUserTopicSubscriptionManagerTest, ShouldRepeatRequestsOnForbidden) {
   auto ids = GetSequenceOfTopics(kInvalidationObjectIdsCount);