diff --git a/fs_attachment/models/ir_attachment.py b/fs_attachment/models/ir_attachment.py
index d026fc88f8..50850a2e63 100644
--- a/fs_attachment/models/ir_attachment.py
+++ b/fs_attachment/models/ir_attachment.py
@@ -481,7 +481,7 @@ def _get_fs_storage_for_code(
         code: str,
     ) -> fsspec.AbstractFileSystem | None:
         """Return the filesystem for the given storage code"""
-        fs = self.env["fs.storage"].get_fs_by_code(code)
+        fs = self.env["fs.storage"].sudo().get_fs_by_code(code)
         if not fs:
             raise SystemError(f"No Filesystem storage for code {code}")
         return fs
diff --git a/fs_storage/models/fs_storage.py b/fs_storage/models/fs_storage.py
index b811e59d84..396cf93d46 100644
--- a/fs_storage/models/fs_storage.py
+++ b/fs_storage/models/fs_storage.py
@@ -194,10 +194,32 @@ def get_storage_codes(self):
         return [s.code for s in self.search([])]
 
     @api.model
-    @tools.ormcache("code")
     def get_fs_by_code(self, code):
         """Return the filesystem associated to the given code.
 
+        :param code: the code of the filesystem
+        """
+        fs = self._get_fs_by_code(code)
+
+        if not tools.config["test_enable"]:
+            try:
+                self._check_connection(fs)
+            except Exception as e:
+                _logger.warning(
+                    "Error while connecting to the filesystem storage %s: %s",
+                    code,
+                    e,
+                )
+                # Generate a new fs instance
+                self.env.registry.clear_cache()
+                fs = self._get_fs_by_code(code)
+        return fs
+
+    @api.model
+    @tools.ormcache("code")
+    def _get_fs_by_code(self, code):
+        """Return the filesystem associated to the given code.
+
         :param code: the code of the filesystem
         """
         fs = None
@@ -265,12 +287,38 @@ def _compute_options_properties(self) -> None:
             doc = inspect.getdoc(cls.__init__)
             rec.options_properties = f"__init__{signature}\n{doc}"
 
+    @api.model
+    def _get_marker_file_name(self):
+        return ".odoo_fs_storage.marker"
+
+    @api.model
+    def _marker_file_check_connection(self, fs):
+        marker_file_name = self._get_marker_file_name()
+        try:
+            fs.info(marker_file_name)
+        except FileNotFoundError:
+            fs.touch(marker_file_name)
+
+    @api.model
+    def _check_connection(self, fs):
+        self._marker_file_check_connection(fs)
+        return True
+
     @property
     def fs(self) -> fsspec.AbstractFileSystem:
         """Get the fsspec filesystem for this backend."""
         self.ensure_one()
         if not self.__fs:
             self.__fs = self._get_filesystem()
+        if not tools.config["test_enable"]:
+            # Check whether we need to invalidate FS cache or not.
+            # Use a marker file to limit the scope of the LS command for performance.
+            try:
+                self._check_connection(self.__fs)
+            except Exception as e:
+                self.__fs.clear_instance_cache()
+                self.__fs = None
+                raise e
         return self.__fs
 
     def _get_filesystem_storage_path(self) -> str:
@@ -432,7 +480,7 @@ def delete(self, relative_path) -> None:
 
     def action_test_config(self) -> None:
         try:
-            self.fs.ls("", detail=False)
+            self._check_connection(self.fs)
             title = _("Connection Test Succeeded!")
             message = _("Everything seems properly set up!")
             msg_type = "success"