Merge pull request #32888 from ankush/fix_invalidations

fix: Clear persistent cache in running processes

frappe/__init__.py

@@ -1947,14 +1947,14 @@ def get_active_domains():
 
 @request_cache
 def is_setup_complete():
-	is_setup_complete = False
+	setup_complete = False
 	if not frappe.db.table_exists("Installed Application"):
-		return is_setup_complete
+		return setup_complete
 
 	if all(frappe.get_all("Installed Application", {"has_setup_wizard": 1}, pluck="is_setup_complete")):
-		is_setup_complete = True
+		setup_complete = True
 
-	return is_setup_complete
+	return setup_complete
 
 
 @whitelist(allow_guest=True)

frappe/cache_manager.py

@@ -119,6 +119,7 @@ def clear_defaults_cache(user=None):
 
 def clear_doctype_cache(doctype=None):
 	clear_controller_cache(doctype)
+	frappe.client_cache.erase_persistent_caches(doctype=doctype)
 
 	_clear_doctype_cache_from_redis(doctype)
 	if hasattr(frappe.db, "after_commit"):
@@ -172,12 +173,12 @@ def _clear_doctype_cache_from_redis(doctype: str | None = None):
 	frappe.cache.delete_value(to_del)
 
 
-def clear_controller_cache(doctype=None):
+def clear_controller_cache(doctype=None, *, site=None):
 	if not doctype:
-		frappe.controllers.pop(frappe.local.site, None)
+		frappe.controllers.pop(site or frappe.local.site, None)
 		return
 
-	if site_controllers := frappe.controllers.get(frappe.local.site):
+	if site_controllers := frappe.controllers.get(site or frappe.local.site):
 		site_controllers.pop(doctype, None)
 
 

frappe/installer.py

@@ -332,6 +332,7 @@ def install_app(name, verbose=False, set_as_patched=True, force=False):
 	for after_sync in app_hooks.after_sync or []:
 		frappe.get_attr(after_sync)()  #
 
+	frappe.client_cache.erase_persistent_caches()
 	frappe.flags.in_install = False
 
 
@@ -421,6 +422,8 @@ def remove_app(app_name, dry_run=False, yes=False, no_backup=False, force=False)
 	for fn in frappe.get_hooks("after_app_uninstall"):
 		frappe.get_attr(fn)(app_name)
 
+	frappe.client_cache.erase_persistent_caches()
+
 	click.secho(f"Uninstalled App {app_name} from Site {site}", fg="green")
 	frappe.flags.in_uninstall = False
 

frappe/utils/redis_wrapper.py

@@ -1,5 +1,6 @@
 # Copyright (c) 2015, Frappe Technologies Pvt. Ltd. and Contributors
 # License: MIT. See LICENSE
+import json
 import pickle
 import re
 import threading
@@ -14,7 +15,6 @@ from redis.exceptions import ResponseError
 
 import frappe
 from frappe.utils import cstr
-from frappe.utils.data import cint
 
 # 5 is faster than default which is 4.
 # Python uses old protocol for backward compatibility, we don't support anything <3.10.
@@ -585,13 +585,29 @@ class ClientCache:
 
 	def run_invalidator_thread(self):
 		self._watcher = self.invalidator.pubsub()
-		self._watcher.subscribe(**{"__redis__:invalidate": self._handle_invalidation})
+		self._watcher.subscribe(
+			**{
+				"__redis__:invalidate": self._handle_invalidation,
+				"clear_persistent_cache": self._handle_persistent_cache_invalidation,
+			}
+		)
 		return self._watcher.run_in_thread(
 			sleep_time=60,
 			daemon=True,
 			exception_handler=self._exception_handler,
 		)
 
+	def erase_persistent_caches(self, *, doctype=None):
+		"""Send signal to clear all worker-specific caches
+
+		This can include cached controller resolution, @site_cache and any other similar persistent
+		cache.
+		"""
+		self.redis.publish(
+			"clear_persistent_cache",
+			json.dumps({"doctype": doctype, "site": frappe.local.site}),
+		)
+
 	def _handle_invalidation(self, message):
 		if message["data"] is None:
 			# Flushall
@@ -601,6 +617,19 @@ class ClientCache:
 			for key in message["data"]:
 				self.cache.pop(key, None)
 
+	def _handle_persistent_cache_invalidation(self, message):
+		import frappe.utils.caching
+		from frappe.cache_manager import clear_controller_cache
+
+		if message["type"] != "message":
+			return
+
+		payload = frappe._dict(json.loads(message["data"]))
+		clear_controller_cache(payload.doctype, site=payload.site)
+
+		if not payload.doctype:
+			frappe.utils.caching._SITE_CACHE.clear()
+
 	def _exception_handler(self, exc, pubsub, pubsub_thread):
 		if isinstance(exc, (redis.exceptions.ConnectionError)):
 			self.clear_cache()