Implement cloud policy invalidations using the invalidation service framework.

chrome/browser/policy/cloud/cloud_policy_invalidator.cc

+// Copyright (c) 2013 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/policy/cloud/cloud_policy_invalidator.h"
+
+#include "base/command_line.h"
+#include "base/message_loop/message_loop.h"
+#include "base/metrics/histogram.h"
+#include "base/rand_util.h"
+#include "base/sequenced_task_runner.h"
+#include "base/strings/string_number_conversions.h"
+#include "base/time/time.h"
+#include "base/values.h"
+#include "chrome/browser/invalidation/invalidation_service.h"
+#include "chrome/browser/policy/cloud/cloud_policy_client.h"
+#include "chrome/browser/policy/cloud/cloud_policy_store.h"
+#include "chrome/browser/policy/cloud/enterprise_metrics.h"
+#include "chrome/common/chrome_switches.h"
+#include "policy/policy_constants.h"
+#include "sync/notifier/object_id_invalidation_map.h"
+
+namespace policy {
+
+const int CloudPolicyInvalidator::kMissingPayloadDelay = 5;
+const int CloudPolicyInvalidator::kMaxFetchDelayDefault = 5000;
+const int CloudPolicyInvalidator::kMaxFetchDelayMin = 1000;
+const int CloudPolicyInvalidator::kMaxFetchDelayMax = 300000;
+
+CloudPolicyInvalidator::CloudPolicyInvalidator(
+    invalidation::InvalidationService* invalidation_service,
+    CloudPolicyClient* client,
+    CloudPolicyStore* store,
+    const scoped_refptr<base::SequencedTaskRunner>& task_runner,
+    const base::Closure& invalidate)
+    : invalidation_service_(invalidation_service),
+      client_(client),
+      store_(store),
+      task_runner_(task_runner),
+      invalidate_(invalidate),
+      registered_timestamp_(0),
+      invalid_(false),
+      invalidation_version_(0),
+      unknown_version_invalidation_count_(0),
+      ack_handle_(syncer::AckHandle::InvalidAckHandle()),
+      weak_factory_(this),
+      max_fetch_delay_(kMaxFetchDelayDefault) {
+  DCHECK(invalidation_service);
+  DCHECK(client);
+  DCHECK(store);
+  DCHECK(task_runner.get());
+
+  invalidation_service->RegisterInvalidationHandler(this);
+  OnStoreLoaded(store);
+  store->AddObserver(this);
+}
+
+CloudPolicyInvalidator::~CloudPolicyInvalidator() {
+  invalidation_service_->UnregisterInvalidationHandler(this);
+  store_->RemoveObserver(this);
+}
+
+void CloudPolicyInvalidator::Unregister() {
+  if (invalid_)
+    AcknowledgeInvalidation();
+  invalidation_service_->UpdateRegisteredInvalidationIds(
+      this,
+      syncer::ObjectIdSet());
+  registered_timestamp_ = 0;
+}
+
+void CloudPolicyInvalidator::OnInvalidatorStateChange(
+    syncer::InvalidatorState state) {}
+
+void CloudPolicyInvalidator::OnIncomingInvalidation(
+    const syncer::ObjectIdInvalidationMap& invalidation_map) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  const syncer::ObjectIdInvalidationMap::const_iterator invalidation =
+      invalidation_map.find(object_id_);
+  if (invalidation == invalidation_map.end()) {
+    NOTREACHED();
+    return;
+  }
+  HandleInvalidation(invalidation->second);
+}
+
+void CloudPolicyInvalidator::OnStoreLoaded(CloudPolicyStore* store) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  if (registered_timestamp_) {
+    // Update the kMetricPolicyRefresh histogram. In some cases, this object can
+    // be constructed during an OnStoreLoaded callback, which causes
+    // OnStoreLoaded to be called twice at initialization time, so make sure
+    // that the timestamp does not match the timestamp at which registration
+    // occurred. We only measure changes which occur after registration.
+    if (!store->policy() || !store->policy()->has_timestamp() ||
+        store->policy()->timestamp() != registered_timestamp_) {
+      UMA_HISTOGRAM_ENUMERATION(
+          kMetricPolicyRefresh,
+          GetPolicyRefreshMetric(),
+          kMetricPolicyRefreshSize);
+    }
+
+    // If the policy was invalid and the version stored matches the latest
+    // invalidation version, acknowledge the latest invalidation.
+    if (invalid_ && store->invalidation_version() == invalidation_version_)
+      AcknowledgeInvalidation();
+  }
+
+  UpdateRegistration(store->policy());
+  UpdateMaxFetchDelay(store->policy_map());
+}
+
+void CloudPolicyInvalidator::OnStoreError(CloudPolicyStore* store) {}
+
+void CloudPolicyInvalidator::HandleInvalidation(

[rlarocque, 2013/07/23 17:43:08]

The invalidations framework currently repeatedly re-sends invalidations with an
exponential backoff if the clients don't acknowledge the invalidation right
away.

I'd like to remove this feature, but it seems likely your change will land
before mine does.  If so, you should be prepared for the repeated invalidations.
 Most clients currently just acknowledge them as soon as they're received, in
part to avoid dealing with repeated re-sends.

[Steve Condie, 2013/07/24 01:42:04]

Thanks for pointing this out. My solution is to ignore the "new" invalidation if
the ack handle is the same as my currently outstanding ack handle.

+    const syncer::Invalidation& invalidation) {
+  // If there is still a pending invalidation, acknowledge it, since we only
+  // care about the latest invalidation.
+  if (invalid_)
+    AcknowledgeInvalidation();

[rlarocque, 2013/07/23 17:43:08]

For example, this probably doesn't work so well if an invalidation is
redelivered.

[Steve Condie, 2013/07/24 01:42:04]

Agreed, this was erroneous in that case.

+
+  // Update invalidation state.
+  invalid_ = true;
+  ack_handle_ = invalidation.ack_handle;
+  invalidation_version_ = invalidation.version;
+
+  // When an invalidation with unknown version is received, use negative
+  // numbers based on the number of such invalidations received. This
+  // ensures that the version numbers do not collide with "real" versions
+  // (which are positive) or previous invalidations with unknown version.
+  if (invalidation_version_ == syncer::Invalidation::kUnknownVersion)
+    invalidation_version_ = -(++unknown_version_invalidation_count_);
+
+  // In order to prevent the cloud policy server from becoming overwhelmed when
+  // a policy with many users is modified, delay for a random period of time
+  // before fetching the policy. Delay for at least 20ms so that if multiple
+  // invalidations are received in quick succession, only one fetch will be
+  // performed.
+  base::TimeDelta delay = base::TimeDelta::FromMilliseconds(
+      base::RandInt(20, max_fetch_delay_));
+
+  // If there is a payload, the invalidate callback can run at any time, so set
+  // the version and payload on the client immediately. Otherwise, the callback
+  // must only run after at least kMissingPayloadDelay minutes.
+  const std::string& payload = invalidation.payload;
+  if (!invalidation.payload.empty())
+    client_->SetInvalidationInfo(invalidation_version_, payload);
+  else
+    delay += base::TimeDelta::FromMinutes(kMissingPayloadDelay);
+
+  // Schedule the invalidate callback to run.
+  task_runner_->PostDelayedTask(

[rlarocque, 2013/07/23 17:43:08]

Would you be interested in cancelling any existing tasks before posting a new
one?

If so, you might want to look at base/timer/timer.h's OneShotTimer.  We found it
very useful for implementing sync's delayed fetch logic.

[Steve Condie, 2013/07/24 01:42:04]

The previous tasks are cancelled by virtue of calling AcknowledgeInvalidation. I
did it in this way based how a similar scenario is solved in my team's
CloudPolicyRefreshScheduler class, although OneShotTimer looks like a good
alternative. For now I'd rather leave it since I've manually tested and written
unit tests for it unless you guys have a strong preference.

+      FROM_HERE,
+      base::Bind(
+          &CloudPolicyInvalidator::RunInvalidateCallback,
+          weak_factory_.GetWeakPtr(),
+          payload.empty() /* is_missing_payload */),
+      delay);
+
+  // Update the kMetricPolicyInvalidations histogram.
+  UMA_HISTOGRAM_ENUMERATION(
+      kMetricPolicyInvalidations,
+      payload.empty() ?
+          kMetricPolicyInvalidationsNoPayload :
+          kMetricPolicyInvalidationsPayload,
+      kMetricPolicyInvalidationsSize);
+}
+
+void CloudPolicyInvalidator::UpdateRegistration(
+    const enterprise_management::PolicyData* policy) {
+  // Create the ObjectId based on the policy data.
+  // If the policy does not specify an the ObjectId, then unregister.
+  if (!policy ||
+      !policy->has_timestamp() ||
+      !policy->has_invalidation_source() ||
+      !policy->has_invalidation_name()) {
+    if (registered_timestamp_)
+      Unregister();
+    return;
+  }
+  invalidation::ObjectId object_id(
+      policy->invalidation_source()/*1025*/,
+      policy->invalidation_name()/*"UENUPOL"*/);
+
+  // If the policy object id in the policy data is different from the currently
+  // registered object id, update the object registration.
+  if (!registered_timestamp_ || !(object_id == object_id_)) {
+    if (invalid_)
+      AcknowledgeInvalidation();
+    registered_timestamp_ = policy->timestamp();
+    object_id_ = object_id;
+
+    syncer::ObjectIdSet ids;
+    ids.insert(object_id);
+    invalidation_service_->UpdateRegisteredInvalidationIds(this, ids);
+  }
+}
+
+void CloudPolicyInvalidator::UpdateMaxFetchDelay(const PolicyMap& policy_map) {
+  int delay;
+
+  // Try reading the delay from the policy.
+  const base::Value* delay_policy_value =
+      policy_map.GetValue(key::kMaxInvalidationFetchDelay);
+  if (delay_policy_value && delay_policy_value->GetAsInteger(&delay)) {
+    set_max_fetch_delay(delay);
+    return;
+  }
+
+  // Try reading the delay from the command line switch.
+  std::string delay_string =
+      CommandLine::ForCurrentProcess()->GetSwitchValueASCII(
+          switches::kCloudPolicyInvalidationDelay);
+  if (base::StringToInt(delay_string, &delay)) {
+    set_max_fetch_delay(delay);
+    return;
+  }
+
+  max_fetch_delay_ = kMaxFetchDelayDefault;
+}
+
+void CloudPolicyInvalidator::set_max_fetch_delay(int delay) {
+  if (delay < kMaxFetchDelayMin)
+    max_fetch_delay_ = kMaxFetchDelayMin;
+  else if (delay > kMaxFetchDelayMax)
+    max_fetch_delay_ = kMaxFetchDelayMax;
+  else
+    max_fetch_delay_ = delay;
+}
+
+void CloudPolicyInvalidator::RunInvalidateCallback(bool is_missing_payload) {
+  DCHECK(thread_checker_.CalledOnValidThread());
+  // In the missing payload case, the invalidation version has not been set on
+  // the client yet, so set it now that the required time has elapsed.
+  if (is_missing_payload)
+    client_->SetInvalidationInfo(invalidation_version_, std::string());
+  invalidate_.Run();

[rlarocque, 2013/07/23 17:43:08]

From where does AcknowledgeInvalidation() get called when an invalidation is
successfully handled?

[Steve Condie, 2013/07/24 01:42:04]

The OnStoreLoaded method. That method is invoked after fetching, validating,
storing, and publishing a new policy. At that point we consider the invalidation
handled.

+}
+
+void CloudPolicyInvalidator::AcknowledgeInvalidation() {
+  DCHECK(invalid_);
+  invalid_ = false;
+  client_->SetInvalidationInfo(0, std::string());
+  invalidation_service_->AcknowledgeInvalidation(object_id_, ack_handle_);
+  // Cancel any scheduled invalidate callbacks.
+  weak_factory_.InvalidateWeakPtrs();

[rlarocque, 2013/07/23 17:43:08]

That's a neat trick.  I had no idea that InvalidateWeakPtrs() could be called
several times on the same object.  I think I still prefer the OneShotTimer,
though.

+}
+
+int CloudPolicyInvalidator::GetPolicyRefreshMetric() {
+  if (store_->policy_changed()) {
+    if (invalid_)
+      return kMetricPolicyRefreshInvalidatedChanged;
+    return kMetricPolicyRefreshChanged;
+  }
+  if (invalid_)
+    return kMetricPolicyRefreshInvalidatedUnchanged;
+  return kMetricPolicyRefreshUnchanged;
+}
+
+}  // namespace policy