[GCM] Add per-app limits to persistent store

Per app limits are enforced at the time an attempt to add a message to the
persistent store is made. By default the limit is 20 messages per app, but
future patches may introduce configurable limits.

BUG=284553

Review URL: https://codereview.chromium.org/99073010

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

5 files changed, 353 insertions, 38 deletions

diff --git a/google_apis/gcm/engine/gcm_store.h b/google_apis/gcm/engine/gcm_store.h
index 05fd431..06189a9 100644
--- a/google_apis/gcm/engine/gcm_store.h
+++ b/google_apis/gcm/engine/gcm_store.h
@@ -71,7 +71,10 @@ class GCM_EXPORT GCMStore {
                                       const UpdateCallback& callback) = 0;
 
   // Unacknowledged outgoing messages handling.
-  virtual void AddOutgoingMessage(const std::string& persistent_id,
+  // Returns false if app has surpassed message limits, else returns true. Note
+  // that the message isn't persisted until |callback| is invoked with
+  // |success| == true.
+  virtual bool AddOutgoingMessage(const std::string& persistent_id,
                                   const MCSMessage& message,
                                   const UpdateCallback& callback) = 0;
   virtual void RemoveOutgoingMessage(const std::string& persistent_id,
diff --git a/google_apis/gcm/engine/gcm_store_impl.cc b/google_apis/gcm/engine/gcm_store_impl.cc
index 3926d71..c82c5df 100644
--- a/google_apis/gcm/engine/gcm_store_impl.cc
+++ b/google_apis/gcm/engine/gcm_store_impl.cc
@@ -27,6 +27,9 @@ namespace gcm {
 
 namespace {
 
+// Limit to the number of outstanding messages per app.
+const int kMessagesPerAppLimit = 20;
+
 // ---- LevelDB keys. ----
 // Key for this device's android id.
 const char kDeviceAIDKey[] = "device_aid_key";
@@ -101,8 +104,10 @@ class GCMStoreImpl::Backend
   void AddOutgoingMessage(const std::string& persistent_id,
                           const MCSMessage& message,
                           const UpdateCallback& callback);
-  void RemoveOutgoingMessages(const PersistentIdList& persistent_ids,
-                              const UpdateCallback& callback);
+  void RemoveOutgoingMessages(
+      const PersistentIdList& persistent_ids,
+      const base::Callback<void(bool, const AppIdToMessageCountMap&)>
+          callback);
   void AddUserSerialNumber(const std::string& username,
                            int64 serial_number,
                            const UpdateCallback& callback);
@@ -137,6 +142,11 @@ GCMStoreImpl::Backend::~Backend() {}
 
 void GCMStoreImpl::Backend::Load(const LoadCallback& callback) {
   LoadResult result;
+  if (db_.get()) {
+    LOG(ERROR) << "Attempting to reload open database.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, result));
+    return;
+  }
 
   leveldb::Options options;
   options.create_if_missing = true;
@@ -193,14 +203,15 @@ void GCMStoreImpl::Backend::Load(const LoadCallback& callback) {
 }
 
 void GCMStoreImpl::Backend::Destroy(const UpdateCallback& callback) {
-  DVLOG(1) << "Destroying RMQ store.";
+  DVLOG(1) << "Destroying GCM store.";
+  db_.reset();
   const leveldb::Status s =
       leveldb::DestroyDB(path_.AsUTF8Unsafe(), leveldb::Options());
   if (s.ok()) {
     foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, true));
     return;
   }
-  LOG(ERROR) << "Destroy failed.";
+  LOG(ERROR) << "Destroy failed: " << s.ToString();
   foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
 }
 
@@ -209,6 +220,12 @@ void GCMStoreImpl::Backend::SetDeviceCredentials(
     uint64 device_security_token,
     const UpdateCallback& callback) {
   DVLOG(1) << "Saving device credentials with AID " << device_android_id;
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
+
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -234,6 +251,12 @@ void GCMStoreImpl::Backend::SetDeviceCredentials(
 void GCMStoreImpl::Backend::AddIncomingMessage(const std::string& persistent_id,
                                                const UpdateCallback& callback) {
   DVLOG(1) << "Saving incoming message with id " << persistent_id;
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
+
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -251,6 +274,11 @@ void GCMStoreImpl::Backend::AddIncomingMessage(const std::string& persistent_id,
 void GCMStoreImpl::Backend::RemoveIncomingMessages(
     const PersistentIdList& persistent_ids,
     const UpdateCallback& callback) {
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -275,6 +303,11 @@ void GCMStoreImpl::Backend::AddOutgoingMessage(const std::string& persistent_id,
                                                const MCSMessage& message,
                                                const UpdateCallback& callback) {
   DVLOG(1) << "Saving outgoing message with id " << persistent_id;
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -293,25 +326,59 @@ void GCMStoreImpl::Backend::AddOutgoingMessage(const std::string& persistent_id,
 
 void GCMStoreImpl::Backend::RemoveOutgoingMessages(
     const PersistentIdList& persistent_ids,
-    const UpdateCallback& callback) {
+    const base::Callback<void(bool, const AppIdToMessageCountMap&)>
+        callback) {
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE,
+                                      base::Bind(callback,
+                                                 false,
+                                                 AppIdToMessageCountMap()));
+    return;
+  }
+  leveldb::ReadOptions read_options;
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
+  AppIdToMessageCountMap removed_message_counts;
+
   leveldb::Status s;
   for (PersistentIdList::const_iterator iter = persistent_ids.begin();
        iter != persistent_ids.end();
        ++iter) {
     DVLOG(1) << "Removing outgoing message with id " << *iter;
+    std::string outgoing_message;
+    leveldb::Slice key_slice = MakeSlice(MakeOutgoingKey(*iter));
+    s = db_->Get(read_options, key_slice, &outgoing_message);
+    if (!s.ok())
+      break;
+    mcs_proto::DataMessageStanza data_message;
+    // Skip the initial tag byte and parse the rest to extract the message.
+    if (data_message.ParseFromString(outgoing_message.substr(1))) {
+      DCHECK(!data_message.from().empty());
+      if (removed_message_counts.count(data_message.from()) != 0)
+        removed_message_counts[data_message.from()]++;
+      else
+        removed_message_counts[data_message.from()] = 1;
+    }
+    DVLOG(1) << "Removing outgoing message with id " << *iter;
+    // Have to create a new slice to perform the deletion.
     s = db_->Delete(write_options, MakeSlice(MakeOutgoingKey(*iter)));
     if (!s.ok())
       break;
   }
   if (s.ok()) {
-    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, true));
+    foreground_task_runner_->PostTask(FROM_HERE,
+                                      base::Bind(callback,
+                                                 true,
+                                                 removed_message_counts));
     return;
   }
   LOG(ERROR) << "LevelDB remove failed: " << s.ToString();
-  foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+  foreground_task_runner_->PostTask(FROM_HERE,
+                                    base::Bind(callback,
+                                               false,
+                                               AppIdToMessageCountMap()));
 }
 
 void GCMStoreImpl::Backend::AddUserSerialNumber(
@@ -319,6 +386,11 @@ void GCMStoreImpl::Backend::AddUserSerialNumber(
     int64 serial_number,
     const UpdateCallback& callback) {
   DVLOG(1) << "Saving username to serial number mapping for user: " << username;
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -337,6 +409,11 @@ void GCMStoreImpl::Backend::AddUserSerialNumber(
 void GCMStoreImpl::Backend::RemoveUserSerialNumber(
     const std::string& username,
     const UpdateCallback& callback) {
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -354,6 +431,11 @@ void GCMStoreImpl::Backend::SetNextSerialNumber(
     const UpdateCallback& callback) {
   DVLOG(1) << "Updating the value of next user serial number to: "
            << next_serial_number;
+  if (!db_.get()) {
+    LOG(ERROR) << "GCMStore db doesn't exist.";
+    foreground_task_runner_->PostTask(FROM_HERE, base::Bind(callback, false));
+    return;
+  }
   leveldb::WriteOptions write_options;
   write_options.sync = true;
 
@@ -512,13 +594,19 @@ GCMStoreImpl::GCMStoreImpl(
     const base::FilePath& path,
     scoped_refptr<base::SequencedTaskRunner> blocking_task_runner)
     : backend_(new Backend(path, base::MessageLoopProxy::current())),
-      blocking_task_runner_(blocking_task_runner) {}
+      blocking_task_runner_(blocking_task_runner),
+      weak_ptr_factory_(this) {}
 
 GCMStoreImpl::~GCMStoreImpl() {}
 
 void GCMStoreImpl::Load(const LoadCallback& callback) {
   blocking_task_runner_->PostTask(
-      FROM_HERE, base::Bind(&GCMStoreImpl::Backend::Load, backend_, callback));
+      FROM_HERE,
+      base::Bind(&GCMStoreImpl::Backend::Load,
+                 backend_,
+                 base::Bind(&GCMStoreImpl::LoadContinuation,
+                            weak_ptr_factory_.GetWeakPtr(),
+                            callback)));
 }
 
 void GCMStoreImpl::Destroy(const UpdateCallback& callback) {
@@ -570,16 +658,31 @@ void GCMStoreImpl::RemoveIncomingMessages(
                  callback));
 }
 
-void GCMStoreImpl::AddOutgoingMessage(const std::string& persistent_id,
+bool GCMStoreImpl::AddOutgoingMessage(const std::string& persistent_id,
                                       const MCSMessage& message,
                                       const UpdateCallback& callback) {
-  blocking_task_runner_->PostTask(
-      FROM_HERE,
-      base::Bind(&GCMStoreImpl::Backend::AddOutgoingMessage,
-                 backend_,
-                 persistent_id,
-                 message,
-                 callback));
+  DCHECK_EQ(message.tag(), kDataMessageStanzaTag);
+  std::string app_id = reinterpret_cast<const mcs_proto::DataMessageStanza*>(
+      &message.GetProtobuf())->from();
+  DCHECK(!app_id.empty());
+  if (app_message_counts_.count(app_id) == 0)
+    app_message_counts_[app_id] = 0;
+  if (app_message_counts_[app_id] < kMessagesPerAppLimit) {
+    app_message_counts_[app_id]++;
+
+    blocking_task_runner_->PostTask(
+        FROM_HERE,
+        base::Bind(&GCMStoreImpl::Backend::AddOutgoingMessage,
+                   backend_,
+                   persistent_id,
+                   message,
+                   base::Bind(&GCMStoreImpl::AddOutgoingMessageContinuation,
+                              weak_ptr_factory_.GetWeakPtr(),
+                              callback,
+                              app_id)));
+    return true;
+  }
+  return false;
 }
 
 void GCMStoreImpl::RemoveOutgoingMessage(const std::string& persistent_id,
@@ -589,7 +692,9 @@ void GCMStoreImpl::RemoveOutgoingMessage(const std::string& persistent_id,
       base::Bind(&GCMStoreImpl::Backend::RemoveOutgoingMessages,
                  backend_,
                  PersistentIdList(1, persistent_id),
-                 callback));
+                 base::Bind(&GCMStoreImpl::RemoveOutgoingMessagesContinuation,
+                            weak_ptr_factory_.GetWeakPtr(),
+                            callback)));
 }
 
 void GCMStoreImpl::RemoveOutgoingMessages(
@@ -600,7 +705,9 @@ void GCMStoreImpl::RemoveOutgoingMessages(
       base::Bind(&GCMStoreImpl::Backend::RemoveOutgoingMessages,
                  backend_,
                  persistent_ids,
-                 callback));
+                 base::Bind(&GCMStoreImpl::RemoveOutgoingMessagesContinuation,
+                            weak_ptr_factory_.GetWeakPtr(),
+                            callback)));
 }
 
 void GCMStoreImpl::SetNextSerialNumber(int64 next_serial_number,
@@ -635,4 +742,57 @@ void GCMStoreImpl::RemoveUserSerialNumber(const std::string& username,
                  callback));
 }
 
+void GCMStoreImpl::LoadContinuation(const LoadCallback& callback,
+                                    const LoadResult& result) {
+  if (!result.success) {
+    callback.Run(result);
+    return;
+  }
+  int num_throttled_apps = 0;
+  for (std::map<std::string, google::protobuf::MessageLite*>::const_iterator
+           iter = result.outgoing_messages.begin();
+       iter != result.outgoing_messages.end(); ++iter) {
+    const mcs_proto::DataMessageStanza* data_message =
+        reinterpret_cast<mcs_proto::DataMessageStanza*>(iter->second);
+    DCHECK(!data_message->from().empty());
+    if (app_message_counts_.count(data_message->from()) == 0)
+      app_message_counts_[data_message->from()] = 1;
+    else
+      app_message_counts_[data_message->from()]++;
+    if (app_message_counts_[data_message->from()] == kMessagesPerAppLimit)
+      num_throttled_apps++;
+  }
+  UMA_HISTOGRAM_COUNTS("GCM.NumThrottledApps", num_throttled_apps);
+  callback.Run(result);
+}
+
+void GCMStoreImpl::AddOutgoingMessageContinuation(
+    const UpdateCallback& callback,
+    const std::string& app_id,
+    bool success) {
+  if (!success) {
+    DCHECK(app_message_counts_[app_id] > 0);
+    app_message_counts_[app_id]--;
+  }
+  callback.Run(success);
+}
+
+void GCMStoreImpl::RemoveOutgoingMessagesContinuation(
+    const UpdateCallback& callback,
+    bool success,
+    const AppIdToMessageCountMap& removed_message_counts) {
+  if (!success) {
+    callback.Run(false);
+    return;
+  }
+  for (AppIdToMessageCountMap::const_iterator iter =
+           removed_message_counts.begin();
+       iter != removed_message_counts.end(); ++iter) {
+    DCHECK_NE(app_message_counts_.count(iter->first), 0U);
+    app_message_counts_[iter->first] -= iter->second;
+    DCHECK_GE(app_message_counts_[iter->first], 0);
+  }
+  callback.Run(true);
+}
+
 }  // namespace gcm
diff --git a/google_apis/gcm/engine/gcm_store_impl.h b/google_apis/gcm/engine/gcm_store_impl.h
index 2037bd5..00af3a2b 100644
--- a/google_apis/gcm/engine/gcm_store_impl.h
+++ b/google_apis/gcm/engine/gcm_store_impl.h
@@ -7,6 +7,7 @@
 
 #include "base/basictypes.h"
 #include "base/memory/ref_counted.h"
+#include "base/memory/weak_ptr.h"
 #include "google_apis/gcm/base/gcm_export.h"
 #include "google_apis/gcm/engine/gcm_store.h"
 
@@ -50,8 +51,7 @@ class GCM_EXPORT GCMStoreImpl : public GCMStore {
                                       const UpdateCallback& callback) OVERRIDE;
 
   // Unacknowledged outgoing messages handling.
-  // TODO(zea): implement per-app limits on the number of outgoing messages.
-  virtual void AddOutgoingMessage(const std::string& persistent_id,
+  virtual bool AddOutgoingMessage(const std::string& persistent_id,
                                   const MCSMessage& message,
                                   const UpdateCallback& callback) OVERRIDE;
   virtual void RemoveOutgoingMessage(const std::string& persistent_id,
@@ -69,11 +69,37 @@ class GCM_EXPORT GCMStoreImpl : public GCMStore {
                                       const UpdateCallback& callback) OVERRIDE;
 
  private:
+  typedef std::map<std::string, int> AppIdToMessageCountMap;
+
+  // Continuation to update the per-app message counts after a load.
+  void LoadContinuation(const LoadCallback& callback,
+                        const LoadResult& result);
+
+  // Continuation to update the per-app message counts when adding messages.
+  // In particular, if a message fails to add, the message count is decremented.
+  void AddOutgoingMessageContinuation(const UpdateCallback& callback,
+                                      const std::string& app_id,
+                                      bool success);
+
+  // Continuation to update the per-app message counts when removing messages.
+  // Note: if doing a read-then-write when removing messages proves expensive,
+  // an in-memory mapping of persisted message id to app could be maintained
+  // instead.
+  void RemoveOutgoingMessagesContinuation(
+      const UpdateCallback& callback,
+      bool success,
+      const std::map<std::string, int>& removed_message_counts);
+
   class Backend;
 
+  // Map of App ids to their message counts.
+  AppIdToMessageCountMap app_message_counts_;
+
   scoped_refptr<Backend> backend_;
   scoped_refptr<base::SequencedTaskRunner> blocking_task_runner_;
 
+  base::WeakPtrFactory<GCMStoreImpl> weak_ptr_factory_;
+
   DISALLOW_COPY_AND_ASSIGN(GCMStoreImpl);
 };
 
diff --git a/google_apis/gcm/engine/gcm_store_impl_unittest.cc b/google_apis/gcm/engine/gcm_store_impl_unittest.cc
index c77f335..5b28ef7 100644
--- a/google_apis/gcm/engine/gcm_store_impl_unittest.cc
+++ b/google_apis/gcm/engine/gcm_store_impl_unittest.cc
@@ -27,6 +27,15 @@ namespace {
 // Number of persistent ids to use in tests.
 const int kNumPersistentIds = 10;
 
+// Number of per-app messages in tests.
+const int kNumMessagesPerApp = 20;
+
+// App name for testing.
+const char kAppName[] = "my_app";
+
+// Category name for testing.
+const char kCategoryName[] = "my_category";
+
 const uint64 kDeviceId = 22;
 const uint64 kDeviceToken = 55;
 
@@ -45,13 +54,15 @@ class GCMStoreImplTest : public testing::Test {
                     const GCMStore::LoadResult& result);
   void UpdateCallback(bool success);
 
- private:
+ protected:
   base::MessageLoop message_loop_;
   base::ScopedTempDir temp_directory_;
+  bool expected_success_;
   scoped_ptr<base::RunLoop> run_loop_;
 };
 
-GCMStoreImplTest::GCMStoreImplTest() {
+GCMStoreImplTest::GCMStoreImplTest()
+    : expected_success_(true) {
   EXPECT_TRUE(temp_directory_.CreateUniqueTempDir());
   run_loop_.reset(new base::RunLoop());
 
@@ -82,7 +93,9 @@ void GCMStoreImplTest::LoadCallback(GCMStore::LoadResult* result_dst,
   run_loop_.reset(new base::RunLoop());
 }
 
-void GCMStoreImplTest::UpdateCallback(bool success) { ASSERT_TRUE(success); }
+void GCMStoreImplTest::UpdateCallback(bool success) {
+  ASSERT_EQ(expected_success_, success);
+}
 
 // Verify creating a new database and loading it.
 TEST_F(GCMStoreImplTest, LoadNew) {
@@ -177,8 +190,8 @@ TEST_F(GCMStoreImplTest, OutgoingMessages) {
   for (int i = 0; i < kNumPersistentIds; ++i) {
     persistent_ids.push_back(GetNextPersistentId());
     mcs_proto::DataMessageStanza message;
-    message.set_from(persistent_ids.back());
-    message.set_category(persistent_ids.back());
+    message.set_from(kAppName + persistent_ids.back());
+    message.set_category(kCategoryName + persistent_ids.back());
     gcm_store->AddOutgoingMessage(
         persistent_ids.back(),
         MCSMessage(message),
@@ -199,8 +212,8 @@ TEST_F(GCMStoreImplTest, OutgoingMessages) {
     const mcs_proto::DataMessageStanza* message =
         reinterpret_cast<mcs_proto::DataMessageStanza*>(
             load_result.outgoing_messages[id]);
-    ASSERT_EQ(message->from(), id);
-    ASSERT_EQ(message->category(), id);
+    ASSERT_EQ(message->from(), kAppName + id);
+    ASSERT_EQ(message->category(), kCategoryName + id);
   }
 
   gcm_store->RemoveOutgoingMessages(
@@ -236,8 +249,8 @@ TEST_F(GCMStoreImplTest, IncomingAndOutgoingMessages) {
     PumpLoop();
 
     mcs_proto::DataMessageStanza message;
-    message.set_from(persistent_ids.back());
-    message.set_category(persistent_ids.back());
+    message.set_from(kAppName + persistent_ids.back());
+    message.set_category(kCategoryName + persistent_ids.back());
     gcm_store->AddOutgoingMessage(
         persistent_ids.back(),
         MCSMessage(message),
@@ -258,8 +271,8 @@ TEST_F(GCMStoreImplTest, IncomingAndOutgoingMessages) {
     const mcs_proto::DataMessageStanza* message =
         reinterpret_cast<mcs_proto::DataMessageStanza*>(
             load_result.outgoing_messages[id]);
-    ASSERT_EQ(message->from(), id);
-    ASSERT_EQ(message->category(), id);
+    ASSERT_EQ(message->from(), kAppName + id);
+    ASSERT_EQ(message->category(), kCategoryName + id);
   }
 
   gcm_store->RemoveIncomingMessages(
@@ -282,6 +295,7 @@ TEST_F(GCMStoreImplTest, IncomingAndOutgoingMessages) {
   ASSERT_TRUE(load_result.outgoing_messages.empty());
 }
 
+// Verify that the next serial number of persisted properly.
 TEST_F(GCMStoreImplTest, NextSerialNumber) {
   const int64 kNextSerialNumber = 77LL;
   scoped_ptr<GCMStore> gcm_store(BuildGCMStore());
@@ -303,6 +317,7 @@ TEST_F(GCMStoreImplTest, NextSerialNumber) {
   EXPECT_EQ(kNextSerialNumber, load_result.next_serial_number);
 }
 
+// Verify that user serial number mappings are persisted properly.
 TEST_F(GCMStoreImplTest, UserSerialNumberMappings) {
   scoped_ptr<GCMStore> gcm_store(BuildGCMStore());
   GCMStore::LoadResult load_result;
@@ -339,6 +354,113 @@ TEST_F(GCMStoreImplTest, UserSerialNumberMappings) {
   EXPECT_EQ(serial_number2, load_result.user_serial_numbers[username2]);
 }
 
+// Test that per-app message limits are enforced, persisted across restarts,
+// and updated as messages are removed.
+TEST_F(GCMStoreImplTest, PerAppMessageLimits) {
+  scoped_ptr<GCMStore> gcm_store(BuildGCMStore());
+  GCMStore::LoadResult load_result;
+  gcm_store->Load(base::Bind(&GCMStoreImplTest::LoadCallback,
+                             base::Unretained(this),
+                             &load_result));
+
+  // Add the initial (below app limit) messages.
+  for (int i = 0; i < kNumMessagesPerApp; ++i) {
+    mcs_proto::DataMessageStanza message;
+    message.set_from(kAppName);
+    message.set_category(kCategoryName);
+    EXPECT_TRUE(gcm_store->AddOutgoingMessage(
+                    base::IntToString(i),
+                    MCSMessage(message),
+                    base::Bind(&GCMStoreImplTest::UpdateCallback,
+                               base::Unretained(this))));
+    PumpLoop();
+  }
+
+  // Attempting to add some more should fail.
+  for (int i = 0; i < kNumMessagesPerApp; ++i) {
+    mcs_proto::DataMessageStanza message;
+    message.set_from(kAppName);
+    message.set_category(kCategoryName);
+    EXPECT_FALSE(gcm_store->AddOutgoingMessage(
+                     base::IntToString(i + kNumMessagesPerApp),
+                     MCSMessage(message),
+                     base::Bind(&GCMStoreImplTest::UpdateCallback,
+                                base::Unretained(this))));
+    PumpLoop();
+  }
+
+  // Tear down and restore the database.
+  gcm_store = BuildGCMStore().Pass();
+  gcm_store->Load(base::Bind(&GCMStoreImplTest::LoadCallback,
+                             base::Unretained(this),
+                             &load_result));
+  PumpLoop();
+
+  // Adding more messages should still fail.
+  for (int i = 0; i < kNumMessagesPerApp; ++i) {
+    mcs_proto::DataMessageStanza message;
+    message.set_from(kAppName);
+    message.set_category(kCategoryName);
+    EXPECT_FALSE(gcm_store->AddOutgoingMessage(
+                     base::IntToString(i + kNumMessagesPerApp),
+                     MCSMessage(message),
+                     base::Bind(&GCMStoreImplTest::UpdateCallback,
+                                base::Unretained(this))));
+    PumpLoop();
+  }
+
+  // Remove the existing messages.
+  for (int i = 0; i < kNumMessagesPerApp; ++i) {
+    gcm_store->RemoveOutgoingMessage(
+        base::IntToString(i),
+        base::Bind(&GCMStoreImplTest::UpdateCallback,
+                   base::Unretained(this)));
+    PumpLoop();
+  }
+
+  // Successfully add new messages.
+  for (int i = 0; i < kNumMessagesPerApp; ++i) {
+    mcs_proto::DataMessageStanza message;
+    message.set_from(kAppName);
+    message.set_category(kCategoryName);
+    EXPECT_TRUE(gcm_store->AddOutgoingMessage(
+                    base::IntToString(i + kNumMessagesPerApp),
+                    MCSMessage(message),
+                    base::Bind(&GCMStoreImplTest::UpdateCallback,
+                               base::Unretained(this))));
+    PumpLoop();
+  }
+}
+
+// When the database is destroyed, all database updates should fail. At the
+// same time, they per-app message counts should not go up, as failures should
+// result in decrementing the counts.
+TEST_F(GCMStoreImplTest, AddMessageAfterDestroy) {
+  scoped_ptr<GCMStore> gcm_store(BuildGCMStore());
+  GCMStore::LoadResult load_result;
+  gcm_store->Load(base::Bind(&GCMStoreImplTest::LoadCallback,
+                             base::Unretained(this),
+                             &load_result));
+  PumpLoop();
+  gcm_store->Destroy(base::Bind(&GCMStoreImplTest::UpdateCallback,
+                               base::Unretained(this)));
+  PumpLoop();
+
+  expected_success_ = false;
+  for (int i = 0; i < kNumMessagesPerApp * 2; ++i) {
+    mcs_proto::DataMessageStanza message;
+    message.set_from(kAppName);
+    message.set_category(kCategoryName);
+    // Because all adds are failing, none should hit the per-app message limits.
+    EXPECT_TRUE(gcm_store->AddOutgoingMessage(
+                    base::IntToString(i),
+                    MCSMessage(message),
+                    base::Bind(&GCMStoreImplTest::UpdateCallback,
+                               base::Unretained(this))));
+    PumpLoop();
+  }
+}
+
 }  // namespace
 
 }  // namespace gcm
diff --git a/google_apis/gcm/engine/mcs_client.cc b/google_apis/gcm/engine/mcs_client.cc
index 43557ae..489bbf8 100644
--- a/google_apis/gcm/engine/mcs_client.cc
+++ b/google_apis/gcm/engine/mcs_client.cc
@@ -237,11 +237,15 @@ void MCSClient::SendMessage(const MCSMessage& message) {
     packet_info->persistent_id = persistent_id;
     SetPersistentId(persistent_id,
                     packet_info->protobuf.get());
-    gcm_store_->AddOutgoingMessage(persistent_id,
-                                   MCSMessage(message.tag(),
-                                              *(packet_info->protobuf)),
-                                   base::Bind(&MCSClient::OnGCMUpdateFinished,
-                                              weak_ptr_factory_.GetWeakPtr()));
+    if (!gcm_store_->AddOutgoingMessage(
+            persistent_id,
+            MCSMessage(message.tag(),
+                       *(packet_info->protobuf)),
+            base::Bind(&MCSClient::OnGCMUpdateFinished,
+                       weak_ptr_factory_.GetWeakPtr()))) {
+      message_sent_callback_.Run("Message queue full.");
+      return;
+    }
   } else if (!connection_factory_->IsEndpointReachable()) {
     DVLOG(1) << "No active connection, dropping message.";
     message_sent_callback_.Run("TTL expired");