Separate cache clearing from migrate, add clear_rxdjango_cache command

docs/caching.rst

@@ -161,9 +161,49 @@ regardless of session count or TTL.
 Cache Clearing on Migrate
 =========================
 
-The cache is automatically cleared whenever ``python manage.py migrate`` is
-executed, via a ``post_migrate`` signal handler. This ensures that schema
-changes don't cause stale cached data to be served.
+By default, RxDjango clears the cache whenever ``python manage.py migrate``
+applies migrations to your apps. This ensures that serializer schema changes
+don't cause stale cached data to be served.
+
+The behavior is controlled by the ``RX_CLEAR_CACHE_ON_MIGRATE`` setting:
+
+.. code-block:: python
+
+   # settings.py — default, preserves existing behavior
+   RX_CLEAR_CACHE_ON_MIGRATE = True
+
+When ``True`` (the default), the cache is cleared only when the migration
+``plan`` actually includes migrations for the sender app — no-op runs of
+``migrate`` (nothing to apply) no longer trigger unnecessary cache wipes.
+
+Set to ``False`` to disable automatic clearing entirely and rely on the
+``clear_rxdjango_cache`` management command instead. This is recommended for
+zero-downtime deployments (blue/green, canary, rolling restarts) where the
+shared MongoDB/Redis cache must remain intact while the old deployment is
+still serving traffic.
+
+MongoDB indexes are always created or verified on ``post_migrate``, regardless
+of this setting, so they are never missing after a fresh database setup.
+
+clear_rxdjango_cache Command
+----------------------------
+
+The ``clear_rxdjango_cache`` management command gives operators explicit
+control over cache lifecycle:
+
+.. code-block:: bash
+
+   # Clear all channel caches
+   python manage.py clear_rxdjango_cache
+
+   # Clear a specific channel only
+   python manage.py clear_rxdjango_cache --channel myapp.MyContextChannel
+
+   # Preview what would be cleared (no changes made)
+   python manage.py clear_rxdjango_cache --dry-run
+
+After clearing, the cache is self-healing: the next client connection triggers
+a full state rebuild from the ORM.
 
 Delta Computation
 =================

rxdjango/management/commands/clear_rxdjango_cache.py

@@ -0,0 +1,97 @@
+"""Management command to clear RxDjango's MongoDB and Redis caches.
+
+Clears all cached instance data from MongoDB and resets Redis state keys
+for registered ContextChannel classes. Indexes are preserved and re-ensured
+after clearing.
+
+Use this command when:
+- Deploying serializer schema changes (added/removed/renamed fields)
+- Manually invalidating stale cache after data corrections
+- As part of a blue/green deployment switch
+
+The cache is self-healing: after clearing, the next client connection
+triggers a full state rebuild from the ORM.
+
+Usage::
+
+    # Clear all channel caches
+    python manage.py clear_rxdjango_cache
+
+    # Clear a specific channel
+    python manage.py clear_rxdjango_cache --channel myapp.MyChannel
+
+    # Preview what would be cleared
+    python manage.py clear_rxdjango_cache --dry-run
+"""
+
+from django.core.management.base import BaseCommand, CommandError
+
+
+class Command(BaseCommand):
+    help = 'Clear RxDjango MongoDB and Redis caches for all or specific channels'
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--channel',
+            type=str,
+            help='Fully qualified channel class name (e.g. myapp.MyChannel). '
+                 'If omitted, clears all registered channels.',
+        )
+        parser.add_argument(
+            '--dry-run',
+            action='store_true',
+            help='List channels that would be cleared without clearing them.',
+        )
+
+    def handle(self, *args, **options):
+        channels = self._resolve_channels(options.get('channel'))
+
+        if options['dry_run']:
+            self.stdout.write('Channels that would be cleared:')
+            for channel_class in channels:
+                self.stdout.write(f'  {channel_class.__module__}.{channel_class.__name__}')
+            self.stdout.write(f'\nTotal: {len(channels)} channel(s)')
+            return
+
+        for channel_class in channels:
+            name = f'{channel_class.__module__}.{channel_class.__name__}'
+
+            from rxdjango.mongo import MongoSignalWriter
+            from rxdjango.redis import RedisSession
+
+            mongo = MongoSignalWriter(channel_class)
+            mongo.clear_cache()
+            mongo.ensure_indexes()
+
+            RedisSession.init_database(channel_class)
+
+            channel_class._state_model.clean_active()
+
+            self.stdout.write(self.style.SUCCESS(f'Cleared {name}'))
+
+        self.stdout.write(self.style.SUCCESS(
+            f'\nTotal: {len(channels)} channel(s) cleared'
+        ))
+
+    def _resolve_channels(self, channel_name):
+        """Resolve channel classes from registry, optionally filtered by name."""
+        from rxdjango.channels import ContextChannel
+
+        registry = ContextChannel.get_registered_channels()
+
+        if not channel_name:
+            return sorted(registry, key=lambda c: c.name)
+
+        for channel_class in registry:
+            qualified = f'{channel_class.__module__}.{channel_class.__name__}'
+            simple = channel_class.__name__
+            if channel_name in (qualified, simple):
+                return [channel_class]
+
+        available = ', '.join(
+            f'{c.__module__}.{c.__name__}' for c in registry
+        )
+        raise CommandError(
+            f'Channel "{channel_name}" not found. '
+            f'Available: {available}'
+        )

rxdjango/mongo.py

@@ -272,10 +272,13 @@ def connect(self):
         self.db = client[settings.MONGO_STATE_DB]
         self.collection = self.db[self.channel_class.__name__.lower()]
 
-    def init_database(self):
-        """Drop and recreate the collection with required indexes.
+    def ensure_indexes(self):
+        """Create required indexes if they don't already exist.
+
+        MongoDB's ``create_index`` is idempotent — if the index exists with
+        the same specification, it is a no-op. Safe to call on every startup.
 
-        Called during ``post_migrate`` to reset the cache. Creates two indexes:
+        Creates two indexes:
 
         - ``instance_pkey``: Composite unique index on (anchor_id, user_key,
           instance_type, id) for fast upserts and lookups.
@@ -285,8 +288,6 @@ def init_database(self):
         if self.collection is None:
             self.connect()
 
-        self.collection.drop()
-
         self.collection.create_index(
             [
                 ('_anchor_id', pymongo.ASCENDING),
@@ -306,6 +307,28 @@ def init_database(self):
             name='reconnection_index',
         )
 
+    def clear_cache(self):
+        """Delete all cached documents from the collection.
+
+        Uses ``delete_many`` instead of ``collection.drop()`` to preserve
+        indexes. This avoids a window where queries run without indexes and
+        removes the need to recreate indexes after clearing.
+        """
+        if self.collection is None:
+            self.connect()
+
+        self.collection.delete_many({})
+
+    def init_database(self):
+        """Clear cache and ensure indexes exist.
+
+        Convenience method that combines :meth:`clear_cache` and
+        :meth:`ensure_indexes`. Kept for backward compatibility with test
+        setup code.
+        """
+        self.clear_cache()
+        self.ensure_indexes()
+
     def write_instances(self, anchor_id, instances):
         """Write instances to MongoDB and compute deltas for broadcasting.
 

rxdjango/signal_handler.py

@@ -6,6 +6,7 @@
 from asgiref.sync import async_to_sync
 from collections import defaultdict
 from django.apps import AppConfig
+from django.conf import settings
 from django.db.models.signals import (pre_save, post_save,
                                       pre_delete, post_delete,
                                       post_migrate)
@@ -95,14 +96,30 @@ def setup(self, app_config: AppConfig) -> None:
             return
         self._setup = True
 
-        def init_cache_database(sender, **kwargs):
+        def on_post_migrate(sender, plan=None, **kwargs):
+            # Always ensure indexes exist, regardless of settings
+            self.mongo.ensure_indexes()
+
+            # Only clear cache if the setting is enabled (default True)
+            if not getattr(settings, 'RX_CLEAR_CACHE_ON_MIGRATE', True):
+                return
+
+            # Skip if no migrations were applied for this app
+            if plan is not None:
+                app_label = sender.label
+                has_migrations = any(
+                    migration.app_label == app_label
+                    for migration, rolled_back in plan
+                )
+                if not has_migrations:
+                    return
+
             self.state_model.clean_active()
-            self.mongo.init_database()
+            self.mongo.clear_cache()
             RedisSession.init_database(self.channel_class)
 
-        # Cache is deleted on every migrate
         post_migrate.connect(
-            init_cache_database,
+            on_post_migrate,
             sender=app_config,
             weak=False,
         )