Hook up ClientToServerResponse::THROTTLED to the client sync loop.

TEST=Added SyncerThreadWithSyncerTest.Throttling

Review URL: http://codereview.chromium.org/269101

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@29236 0039d316-1c4b-4281-b951-d872f2087c98

15 files changed, 159 insertions, 36 deletions

diff --git a/chrome/browser/sync/engine/all_status.cc b/chrome/browser/sync/engine/all_status.cc
index d5b2a2b1..07551a1 100644
--- a/chrome/browser/sync/engine/all_status.cc
+++ b/chrome/browser/sync/engine/all_status.cc
@@ -101,7 +101,8 @@ AllStatus::Status AllStatus::CalcSyncing(const SyncerEvent &event) const {
   }
   status.syncing |= syncerStatus.syncing();
   // Show a syncer as syncing if it's got stalled updates.
-  status.syncing = event.last_session->ShouldSyncAgain();
+  status.syncing = event.last_session->HasMoreToSync() &&
+      event.last_session->silenced_until().is_null();
   status.initial_sync_ended |= syncerStatus.IsShareUsable();
   status.syncer_stuck |= syncerStatus.syncer_stuck();
   if (syncerStatus.consecutive_errors() > status.max_consecutive_errors)
diff --git a/chrome/browser/sync/engine/sync_process_state.cc b/chrome/browser/sync/engine/sync_process_state.cc
index 0171a19..32ee2a6 100644
--- a/chrome/browser/sync/engine/sync_process_state.cc
+++ b/chrome/browser/sync/engine/sync_process_state.cc
@@ -44,7 +44,6 @@ SyncProcessState::SyncProcessState(syncable::DirectoryManager* dirman,
       resolver_(resolver),
       model_safe_worker_(model_safe_worker),
       syncer_event_channel_(syncer_event_channel),
-      silenced_until_(0),
       error_rate_(0),
       current_sync_timestamp_(0),
       servers_latest_timestamp_(0),
@@ -141,7 +140,7 @@ void SyncProcessState::increment_num_sync_cycles() {
   ++num_sync_cycles_;
 }
 
-void SyncProcessState::set_silenced_until(const time_t val) {
+void SyncProcessState::set_silenced_until(const base::TimeTicks& val) {
   UpdateDirty(val != silenced_until_);
   silenced_until_ = val;
 }
diff --git a/chrome/browser/sync/engine/sync_process_state.h b/chrome/browser/sync/engine/sync_process_state.h
index 9e0d2ff..de48e17 100644
--- a/chrome/browser/sync/engine/sync_process_state.h
+++ b/chrome/browser/sync/engine/sync_process_state.h
@@ -19,6 +19,7 @@
 #include "base/atomicops.h"
 #include "base/basictypes.h"
 #include "base/port.h"
+#include "base/time.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer_types.h"
 #include "chrome/browser/sync/syncable/syncable_id.h"
@@ -184,8 +185,8 @@ class SyncProcessState {
   void set_num_sync_cycles(const int val);
   void increment_num_sync_cycles();
 
-  time_t silenced_until() const { return silenced_until_; }
-  void set_silenced_until(const time_t val);
+  base::TimeTicks silenced_until() const { return silenced_until_; }
+  void set_silenced_until(const base::TimeTicks& val);
 
   // Info that is tracked purely for status reporting.
 
@@ -297,7 +298,6 @@ class SyncProcessState {
         resolver_(NULL),
         model_safe_worker_(NULL),
         syncer_event_channel_(NULL),
-        silenced_until_(0),
         error_rate_(0),
         current_sync_timestamp_(0),
         servers_latest_timestamp_(0),
@@ -335,7 +335,7 @@ class SyncProcessState {
   std::set<ConflictSet*> conflict_sets_;
 
   // When we're over bandwidth quota, we don't update until past this time.
-  time_t silenced_until_;
+  base::TimeTicks silenced_until_;
 
   // Status information, as opposed to state info that may also be exposed for
   // status reporting purposes.
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index b0eaa5b..099f831 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -1395,11 +1395,11 @@ void SyncManager::SyncInternal::HandleSyncerEvent(const SyncerEvent& event) {
   // Notifications are sent at the end of every sync cycle, regardless of
   // whether we should sync again.
   if (event.what_happened == SyncerEvent::SYNC_CYCLE_ENDED) {
-    if (!event.last_session->ShouldSyncAgain()) {
+    if (!event.last_session->HasMoreToSync()) {
       observer_->OnSyncCycleCompleted();
     }
 
-    // TODO(chron): Consider changing this back to track ShouldSyncAgain
+    // TODO(chron): Consider changing this back to track HasMoreToSync
     // Only notify peers if a commit has occurred and change the bookmark model.
     if (event.last_session && event.last_session->items_committed()) {
       notification_pending_ = true;
diff --git a/chrome/browser/sync/engine/syncer.cc b/chrome/browser/sync/engine/syncer.cc
index a297dca..9274edf 100644
--- a/chrome/browser/sync/engine/syncer.cc
+++ b/chrome/browser/sync/engine/syncer.cc
@@ -57,7 +57,6 @@ Syncer::Syncer(
       max_commit_batch_size_(kDefaultMaxCommitBatchSize),
       connection_manager_(connection_manager),
       dirman_(dirman),
-      silenced_until_(0),
       command_channel_(NULL),
       model_safe_worker_(model_safe_worker),
       updates_source_(sync_pb::GetUpdatesCallerInfo::UNKNOWN),
@@ -94,7 +93,7 @@ bool Syncer::SyncShare(SyncProcessState* process_state) {
   session.set_source(TestAndSetUpdatesSource());
   session.set_notifications_enabled(notifications_enabled());
   SyncShare(&session, SYNCER_BEGIN, SYNCER_END);
-  return session.ShouldSyncAgain();
+  return session.HasMoreToSync();
 }
 
 bool Syncer::SyncShare(SyncerStep first_step, SyncerStep last_step) {
@@ -104,7 +103,7 @@ bool Syncer::SyncShare(SyncerStep first_step, SyncerStep last_step) {
                          model_safe_worker());
   SyncerSession session(&cycle_state, &state);
   SyncShare(&session, first_step, last_step);
-  return session.ShouldSyncAgain();
+  return session.HasMoreToSync();
 }
 
 void Syncer::SyncShare(SyncerSession* session) {
@@ -116,6 +115,9 @@ void Syncer::SyncShare(SyncerSession* session,
                        const SyncerStep last_step) {
   SyncerStep current_step = first_step;
 
+  // Reset silenced_until_, it is the callers responsibility to honor throttles.
+  silenced_until_ = session->silenced_until();
+
   SyncerStep next_step;
   while (!ExitRequested()) {
     switch (current_step) {
diff --git a/chrome/browser/sync/engine/syncer.h b/chrome/browser/sync/engine/syncer.h
index 408af6a..2d6ba6a 100644
--- a/chrome/browser/sync/engine/syncer.h
+++ b/chrome/browser/sync/engine/syncer.h
@@ -11,6 +11,7 @@
 
 #include "base/basictypes.h"
 #include "base/scoped_ptr.h"
+#include "base/time.h"
 #include "chrome/browser/sync/engine/client_command_channel.h"
 #include "chrome/browser/sync/engine/conflict_resolver.h"
 #include "chrome/browser/sync/engine/syncer_types.h"
@@ -84,7 +85,11 @@ class Syncer {
   void RequestEarlyExit() { early_exit_requested_ = true; }
 
   // SyncShare(...) variants cause one sync cycle to occur.  The return value
-  // indicates whether we should sync again.
+  // indicates whether we should sync again.  If we should not sync again,
+  // it doesn't necessarily mean everything is OK; we could be throttled for
+  // example.  Like a good parent, it is the caller's responsibility to clean up
+  // after the syncer when it finishes a sync share operation and honor
+  // server mandated throttles.
   // The zero-argument version of SyncShare is provided for unit tests.
   // When |sync_process_state| is provided, it is used as the syncer state
   // for the sync cycle.  It is treated as an input/output parameter.
@@ -140,6 +145,8 @@ class Syncer {
     notifications_enabled_ = state;
   }
 
+  base::TimeTicks silenced_until() const { return silenced_until_; }
+  bool is_silenced() const { return !silenced_until_.is_null(); }
  private:
   void RequestNudge(int milliseconds);
 
@@ -162,7 +169,7 @@ class Syncer {
   syncable::DirectoryManager* const dirman_;
 
   // When we're over bandwidth quota, we don't update until past this time.
-  time_t silenced_until_;
+  base::TimeTicks silenced_until_;
 
   scoped_ptr<SyncerEventChannel> syncer_event_channel_;
   scoped_ptr<ShutdownChannel> shutdown_channel_;
diff --git a/chrome/browser/sync/engine/syncer_end_command.cc b/chrome/browser/sync/engine/syncer_end_command.cc
index f25cec8..300f7f4 100644
--- a/chrome/browser/sync/engine/syncer_end_command.cc
+++ b/chrome/browser/sync/engine/syncer_end_command.cc
@@ -22,7 +22,7 @@ void SyncerEndCommand::ExecuteImpl(SyncerSession* session) {
   SyncerStatus status(session);
   status.set_syncing(false);
 
-  if (!session->ShouldSyncAgain()) {
+  if (!session->HasMoreToSync()) {
     // This might be the first time we've fully completed a sync cycle.
     DCHECK(session->got_zero_updates());
 
diff --git a/chrome/browser/sync/engine/syncer_proto_util.cc b/chrome/browser/sync/engine/syncer_proto_util.cc
index 49f7c3c..4896c59 100644
--- a/chrome/browser/sync/engine/syncer_proto_util.cc
+++ b/chrome/browser/sync/engine/syncer_proto_util.cc
@@ -158,7 +158,8 @@ bool SyncerProtoUtil::PostClientToServerMessage(ClientToServerMessage* msg,
         break;
       case ClientToServerResponse::THROTTLED:
         LOG(WARNING) << "Client silenced by server.";
-        session->set_silenced_until(time(0) + kSyncDelayAfterThrottled);
+        session->set_silenced_until(base::TimeTicks::Now() +
+            base::TimeDelta::FromSeconds(kSyncDelayAfterThrottled));
         rv = false;
         break;
       default:
diff --git a/chrome/browser/sync/engine/syncer_session.h b/chrome/browser/sync/engine/syncer_session.h
index a8c389f..c75458f 100644
--- a/chrome/browser/sync/engine/syncer_session.h
+++ b/chrome/browser/sync/engine/syncer_session.h
@@ -14,6 +14,7 @@
 #include <utility>
 #include <vector>
 
+#include "base/time.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/sync_cycle_state.h"
 #include "chrome/browser/sync/engine/sync_process_state.h"
@@ -122,11 +123,11 @@ class SyncerSession {
     return sync_process_state_->conflicting_updates();
   }
 
-  time_t silenced_until() const {
+  base::TimeTicks silenced_until() const {
     return sync_process_state_->silenced_until();
   }
 
-  void set_silenced_until(time_t silenced_until) const {
+  void set_silenced_until(base::TimeTicks silenced_until) const {
     sync_process_state_->set_silenced_until(silenced_until);
   }
 
@@ -321,7 +322,7 @@ class SyncerSession {
   // TODO(chron): Unit test for this method.
   // returns true iff this session contains data that should go through the
   // sync engine again.
-  bool ShouldSyncAgain() const {
+  bool HasMoreToSync() const {
     return (HasRemainingItemsToCommit() &&
               sync_process_state_->successful_commits() > 0) ||
            conflict_sets_built() ||
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index 2ab6106..073b5ea 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -136,12 +136,6 @@ void SyncerThread::NudgeSyncer(int milliseconds_from_now, NudgeSource source) {
     return;
   }
 
-  if (vault_.current_wait_interval_.had_nudge_during_backoff) {
-    // Drop nudges on the floor if we've already had one since starting this
-    // stage of exponential backoff.
-    return;
-  }
-
   NudgeSyncImpl(milliseconds_from_now, source);
 }
 
@@ -343,7 +337,11 @@ void SyncerThread::ThreadMainLoop() {
 
     const TimeTicks next_poll = last_sync_time +
         vault_.current_wait_interval_.poll_delta;
-    const TimeTicks end_wait =
+    bool throttled = vault_.current_wait_interval_.mode ==
+        WaitInterval::THROTTLED;
+    // If we are throttled, we must wait.  Otherwise, wait until either the next
+    // nudge (if one exists) or the poll interval.
+    const TimeTicks end_wait = throttled ? next_poll :
         !vault_.nudge_queue_.empty() &&
         vault_.nudge_queue_.top().first < next_poll ?
         vault_.nudge_queue_.top().first : next_poll;
@@ -372,7 +370,7 @@ void SyncerThread::ThreadMainLoop() {
 
     // Handle a nudge, caused by either a notification or a local bookmark
     // event.  This will also update the source of the following SyncMain call.
-    bool nudged = UpdateNudgeSource(continue_sync_cycle,
+    bool nudged = UpdateNudgeSource(throttled, continue_sync_cycle,
                                     &initial_sync_for_thread);
 
     LOG(INFO) << "Calling Sync Main at time " << Time::Now().ToInternalValue();
@@ -398,6 +396,16 @@ SyncerThread::WaitInterval SyncerThread::CalculatePollingWaitTime(
     bool was_nudged) {
   lock_.AssertAcquired();  // We access 'vault' in here, so we need the lock.
   WaitInterval return_interval;
+
+  // Server initiated throttling trumps everything.
+  if (vault_.syncer_ && vault_.syncer_->is_silenced()) {
+    // We don't need to reset other state, it can continue where it left off.
+    return_interval.mode = WaitInterval::THROTTLED;
+    return_interval.poll_delta = vault_.syncer_->silenced_until() -
+        TimeTicks::Now();
+    return return_interval;
+  }
+
   bool is_continuing_sync_cyle = *continue_sync_cycle;
   *continue_sync_cycle = false;
 
@@ -473,13 +481,14 @@ void SyncerThread::ThreadMain() {
 void SyncerThread::SyncMain(Syncer* syncer) {
   CHECK(syncer);
   AutoUnlock unlock(lock_);
-  while (syncer->SyncShare()) {
+  while (syncer->SyncShare() && !syncer->is_silenced()) {
     LOG(INFO) << "Looping in sync share";
   }
   LOG(INFO) << "Done looping in sync share";
 }
 
-bool SyncerThread::UpdateNudgeSource(bool continue_sync_cycle,
+bool SyncerThread::UpdateNudgeSource(bool was_throttled,
+                                     bool continue_sync_cycle,
                                      bool* initial_sync) {
   bool nudged = false;
   NudgeSource nudge_source = kUnknown;
@@ -491,7 +500,7 @@ bool SyncerThread::UpdateNudgeSource(bool continue_sync_cycle,
   // previous sync cycle.
   while (!vault_.nudge_queue_.empty() &&
          TimeTicks::Now() >= vault_.nudge_queue_.top().first) {
-    if (!nudged) {
+    if (!was_throttled && !nudged) {
       nudge_source = vault_.nudge_queue_.top().second;
       nudged = true;
     }
@@ -618,6 +627,13 @@ int SyncerThread::CalculateSyncWaitTime(int last_interval, int user_idle_ms) {
 // Called with mutex_ already locked.
 void SyncerThread::NudgeSyncImpl(int milliseconds_from_now,
                                  NudgeSource source) {
+  if (vault_.current_wait_interval_.mode == WaitInterval::THROTTLED ||
+      vault_.current_wait_interval_.had_nudge_during_backoff) {
+    // Drop nudges on the floor if we've already had one since starting this
+    // stage of exponential backoff or we are throttled.
+    return;
+  }
+
   const TimeTicks nudge_time = TimeTicks::Now() +
       TimeDelta::FromMilliseconds(milliseconds_from_now);
   NudgeObject nudge_object(nudge_time, source);
diff --git a/chrome/browser/sync/engine/syncer_thread.h b/chrome/browser/sync/engine/syncer_thread.h
index e68ac70..f0287d4 100644
--- a/chrome/browser/sync/engine/syncer_thread.h
+++ b/chrome/browser/sync/engine/syncer_thread.h
@@ -71,6 +71,7 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread> {
   FRIEND_TEST(SyncerThreadTest, CalculatePollingWaitTime);
   FRIEND_TEST(SyncerThreadWithSyncerTest, Polling);
   FRIEND_TEST(SyncerThreadWithSyncerTest, Nudge);
+  FRIEND_TEST(SyncerThreadWithSyncerTest, Throttling);
   friend class SyncerThreadWithSyncerTest;
   friend class SyncerThreadFactory;
 public:
@@ -87,6 +88,9 @@ public:
       // backoff.
       // EXPONENTIAL_BACKOFF intervals are nudge-rate limited to 1 per interval.
       EXPONENTIAL_BACKOFF,
+      // A server-initiated throttled interval.  We do not allow any syncing
+      // during such an interval.
+      THROTTLED,
     };
 
     Mode mode;
@@ -252,9 +256,13 @@ public:
 
   // Sets the source value of the controlled syncer's updates_source value.
   // The initial sync boolean is updated if read as a sentinel.  The following
-  // two methods work in concert to achieve this goal.  Returns true if it
-  // determines a nudge actually occurred.
-  bool UpdateNudgeSource(bool continue_sync_cycle,
+  // two methods work in concert to achieve this goal.
+  // If |was_throttled| was true, this still discards elapsed nudges, but we
+  // treat the request as a periodic poll rather than a nudge from a source.
+  // TODO(timsteele/code reviewer): The first poll after a throttle period
+  // will appear as a periodic request.  Do we want to be more specific?
+  // Returns true if it determines a nudge actually occurred.
+  bool UpdateNudgeSource(bool was_throttled, bool continue_sync_cycle,
                          bool* initial_sync);
   void SetUpdatesSource(bool nudged, NudgeSource nudge_source,
                         bool* initial_sync);
diff --git a/chrome/browser/sync/engine/syncer_thread_unittest.cc b/chrome/browser/sync/engine/syncer_thread_unittest.cc
index b80fa30..0701384 100644
--- a/chrome/browser/sync/engine/syncer_thread_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_thread_unittest.cc
@@ -57,14 +57,26 @@ class SyncerThreadWithSyncerTest : public testing::Test {
   DISALLOW_COPY_AND_ASSIGN(SyncerThreadWithSyncerTest);
 };
 
-class SyncShareIntercept : public MockConnectionManager::MidCommitObserver {
+class SyncShareIntercept : public MockConnectionManager::ThrottleRequestVisitor,
+                           public MockConnectionManager::MidCommitObserver {
  public:
-  SyncShareIntercept() : sync_occured_(false, false) {}
+  SyncShareIntercept() : sync_occured_(false, false),
+                         allow_multiple_interceptions_(true) {}
   virtual ~SyncShareIntercept() {}
   virtual void Observe() {
+    if (!allow_multiple_interceptions_ && !times_sync_occured_.empty())
+      FAIL() << "Multiple sync shares occured.";
     times_sync_occured_.push_back(TimeTicks::Now());
     sync_occured_.Signal();
   }
+
+  // ThrottleRequestVisitor implementation.
+  virtual void VisitAtomically() {
+    // Server has told the client to throttle.  We should not see any syncing.
+    allow_multiple_interceptions_ = false;
+    times_sync_occured_.clear();
+  }
+
   void WaitForSyncShare(int at_least_this_many, TimeDelta max_wait) {
     while (at_least_this_many-- > 0)
       sync_occured_.TimedWait(max_wait);
@@ -75,6 +87,7 @@ class SyncShareIntercept : public MockConnectionManager::MidCommitObserver {
  private:
   std::vector<TimeTicks> times_sync_occured_;
   base::WaitableEvent sync_occured_;
+  bool allow_multiple_interceptions_;
   DISALLOW_COPY_AND_ASSIGN(SyncShareIntercept);
 };
 
@@ -561,4 +574,33 @@ TEST_F(SyncerThreadWithSyncerTest, Nudge) {
   EXPECT_TRUE(syncer_thread()->Stop(2000));
 }
 
+TEST_F(SyncerThreadWithSyncerTest, Throttling) {
+  SyncShareIntercept interceptor;
+  connection()->SetMidCommitObserver(&interceptor);
+  const TimeDelta poll_interval = TimeDelta::FromMilliseconds(10);
+  syncer_thread()->SetSyncerShortPollInterval(poll_interval);
+
+  EXPECT_TRUE(syncer_thread()->Start());
+  metadb()->Open();
+
+  // Wait for some healthy syncing.
+  interceptor.WaitForSyncShare(4, poll_interval + poll_interval);
+
+  // Tell the server to throttle a single request, which should be all it takes
+  // to silence our syncer (for 2 hours, so we shouldn't hit that in this test).
+  // This will atomically visit the interceptor so it can switch to throttled
+  // mode and fail on multiple requests.
+  connection()->ThrottleNextRequest(&interceptor);
+
+  // Try to trigger a sync (we have a really short poll interval already).
+  syncer_thread()->NudgeSyncer(0, SyncerThread::kUnknown);
+  syncer_thread()->NudgeSyncer(0, SyncerThread::kUnknown);
+
+  // Stick around for several poll intervals for good measure. Any sync is
+  // a failure.
+  interceptor.WaitForSyncShare(1, poll_interval * 10);
+
+  EXPECT_TRUE(syncer_thread()->Stop(2000));
+}
+
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/engine/syncer_types.h b/chrome/browser/sync/engine/syncer_types.h
index 36563d0..4a96677 100644
--- a/chrome/browser/sync/engine/syncer_types.h
+++ b/chrome/browser/sync/engine/syncer_types.h
@@ -99,7 +99,7 @@ struct SyncerEvent {
 
     // We have reached the SYNCER_END state in the main sync loop.
     // Check the SyncerSession for information like whether we need to continue
-    // syncing (SyncerSession::ShouldSyncAgain).
+    // syncing (SyncerSession::HasMoreToSync).
     SYNC_CYCLE_ENDED,
   };
 
diff --git a/chrome/test/sync/engine/mock_server_connection.cc b/chrome/test/sync/engine/mock_server_connection.cc
index 46ec870..372d408 100644
--- a/chrome/test/sync/engine/mock_server_connection.cc
+++ b/chrome/test/sync/engine/mock_server_connection.cc
@@ -44,6 +44,8 @@ MockConnectionManager::MockConnectionManager(DirectoryManager* dirmgr,
       mid_commit_callback_function_(NULL),
       mid_commit_observer_(NULL),
       client_command_(NULL),
+      throttling_(false),
+      fail_non_periodic_get_updates_(false),
       next_position_in_parent_(2) {
     server_reachable_ = true;
 };
@@ -105,6 +107,7 @@ bool MockConnectionManager::PostBufferToPath(const PostBufferParams* params,
   EXPECT_TRUE(!store_birthday_sent_ || post.has_store_birthday() ||
               post.message_contents() == ClientToServerMessage::AUTHENTICATE);
   store_birthday_sent_ = true;
+
   if (post.message_contents() == ClientToServerMessage::COMMIT) {
     ProcessCommit(&post, &response);
   } else if (post.message_contents() == ClientToServerMessage::GET_UPDATES) {
@@ -118,6 +121,15 @@ bool MockConnectionManager::PostBufferToPath(const PostBufferParams* params,
   if (client_command_.get()) {
     response.mutable_client_command()->CopyFrom(*client_command_.get());
   }
+
+  {
+    AutoLock throttle_lock(throttle_lock_);
+    if (throttling_) {
+        response.set_error_code(ClientToServerResponse::THROTTLED);
+        throttling_ = false;
+    }
+  }
+
   response.SerializeToString(params->buffer_out);
   if (mid_commit_callback_function_) {
     if (mid_commit_callback_function_(directory))
@@ -126,6 +138,7 @@ bool MockConnectionManager::PostBufferToPath(const PostBufferParams* params,
   if (mid_commit_observer_) {
     mid_commit_observer_->Observe();
   }
+
   return result;
 }
 
@@ -248,6 +261,11 @@ void MockConnectionManager::ProcessGetUpdates(ClientToServerMessage* csm,
   ASSERT_EQ(csm->message_contents(), ClientToServerMessage::GET_UPDATES);
   const GetUpdatesMessage& gu = csm->get_updates();
   EXPECT_TRUE(gu.has_from_timestamp());
+  if (fail_non_periodic_get_updates_) {
+    EXPECT_EQ(sync_pb::GetUpdatesCallerInfo::PERIODIC,
+              gu.caller_info().source());
+  }
+
   // TODO(sync): filter results dependant on timestamp? or check limits?
   response->mutable_get_updates()->CopyFrom(updates_);
   ResetUpdates();
@@ -380,3 +398,11 @@ const CommitMessage& MockConnectionManager::last_sent_commit() const {
   DCHECK(!commit_messages_.empty());
   return *commit_messages_.back();
 }
+
+void MockConnectionManager::ThrottleNextRequest(
+    ThrottleRequestVisitor* visitor) {
+  AutoLock lock(throttle_lock_);
+  throttling_ = true;
+  if (visitor)
+    visitor->VisitAtomically();
+}
\ No newline at end of file
diff --git a/chrome/test/sync/engine/mock_server_connection.h b/chrome/test/sync/engine/mock_server_connection.h
index 1c19a5b..e203081 100644
--- a/chrome/test/sync/engine/mock_server_connection.h
+++ b/chrome/test/sync/engine/mock_server_connection.h
@@ -117,6 +117,18 @@ class MockConnectionManager : public browser_sync::ServerConnectionManager {
 
   void FailNextPostBufferToPathCall() { fail_next_postbuffer_ = true; }
 
+  // A visitor class to allow a test to change some monitoring state atomically
+  // with the action of throttling requests (for example, so you can say
+  // "ThrottleNextRequest, and assert no more requests are made once throttling
+  // is in effect" in one step.
+  class ThrottleRequestVisitor {
+   public:
+    // Called with throttle parameter lock acquired.
+    virtual void VisitAtomically() = 0;
+  };
+  void ThrottleNextRequest(ThrottleRequestVisitor* visitor);
+  void FailNonPeriodicGetUpdates() { fail_non_periodic_get_updates_ = true; }
+
   // Simple inspectors.
   bool client_stuck() const { return client_stuck_; }
 
@@ -204,6 +216,14 @@ class MockConnectionManager : public browser_sync::ServerConnectionManager {
   // What we use to determine if we should return SUCCESS or BAD_AUTH_TOKEN.
   std::string valid_auth_token_;
 
+  // Whether we are faking a server mandating clients to throttle requests.
+  // Protected by |throttle_lock_|.
+  bool throttling_;
+  Lock throttle_lock_;
+
+  // True if we are only accepting GetUpdatesCallerInfo::PERIODIC requests.
+  bool fail_non_periodic_get_updates_;
+
   scoped_ptr<sync_pb::ClientCommand> client_command_;
 
   // The next value to use for the position_in_parent property.