Part deactivated users in the background

One room at a time so we don't take out the whole server with leave
events, and restart at server restart.

synapse/handlers/deactivate_account.py

@@ -12,10 +12,11 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-from twisted.internet import defer
+from twisted.internet import defer, reactor
 
 from ._base import BaseHandler
 from synapse.types import UserID, create_requester
+from synapse.util.logcontext import run_in_background
 
 import logging
 
@@ -30,6 +31,10 @@ class DeactivateAccountHandler(BaseHandler):
         self._device_handler = hs.get_device_handler()
         self._room_member_handler = hs.get_room_member_handler()
 
+        self._user_parter_running = False
+
+        reactor.callWhenRunning(self.start_user_parting)
+
     @defer.inlineCallbacks
     def deactivate_account(self, user_id):
         """Deactivate a user's account
@@ -53,10 +58,38 @@ class DeactivateAccountHandler(BaseHandler):
         yield self.store.user_delete_threepids(user_id)
         yield self.store.user_set_password_hash(user_id, None)
 
+        yield self.store.add_user_pending_deactivation(user_id)
+
+        self.start_user_parting()
+
+    def start_user_parting(self):
+       if not self._user_parter_running:
+            run_in_background(self.user_parter_loop())
+
+    @defer.inlineCallbacks
+    def user_parter_loop(self):
+        self._user_parter_running = True
+        logger.info("Starting user parter")
+        try:
+            while True:
+                user_id = yield self.store.get_user_pending_deactivation()
+                if user_id is None:
+                    break
+                logger.info("User parter parting %r", user_id)
+                yield self.part_user(user_id)
+                yield self.store.del_user_pending_deactivation(user_id)
+                logger.info("User parter finished parting %r", user_id)
+            logger.info("User parter finished: stopping")
+        finally:
+            self._user_parter_running = False
+
+    @defer.inlineCallbacks
+    def part_user(self, user_id):
         user = UserID.from_string(user_id)
 
         rooms_for_user = yield self.store.get_rooms_for_user(user_id)
         for room_id in rooms_for_user:
+            logger.info("User parter parting %r from %r", user_id, room_id)
             yield self._room_member_handler.update_membership(
                 create_requester(user),
                 user,

synapse/storage/registration.py

@@ -526,3 +526,30 @@ class RegistrationStore(RegistrationWorkerStore,
         except self.database_engine.module.IntegrityError:
             ret = yield self.get_3pid_guest_access_token(medium, address)
             defer.returnValue(ret)
+
+    def add_user_pending_deactivation(self, user_id):
+        return self._simple_insert(
+            "users_pending_deactivation",
+            values={
+                "user_id": user_id,
+            },
+            desc="add_user_pending_deactivation",
+        )
+
+    def del_user_pending_deactivation(self, user_id):
+        return self._simple_delete_one(
+            "users_pending_deactivation",
+            keyvalues={
+                "user_id": user_id,
+            },
+            desc="del_user_pending_deactivation",
+        )
+
+    def get_user_pending_deactivation(self):
+        return self._simple_select_one_onecol(
+            "users_pending_deactivation",
+            keyvalues={},
+            retcol="user_id",
+            allow_none=True,
+            desc="get_users_pending_deactivation",
+        )