UN-3295 [FIX] Fix fsspec directory listing cache issue on connectors (#1826)

* UN-3295 [FIX] Fix fsspec directory listing cache issue on connectors

Fixes Azure Blob Storage (and other connectors) not detecting new files or
seeing deleted files. Changes span:

- Disable fsspec directory listing cache: Azure uses listings_expiry_time=1
  due to adlfs bug, S3/Box/Dropbox use use_listings_cache=False
- Eliminate walk() + listdir() double-listing by using walk(detail=True)
- Update type hints from list[str] to Collection[str] for dirs parameter
- Restore success toast for "Clear File History" operation

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3295 [FIX] Fix Ruff B007 lint warnings for unused loop variables

Rename unused `root` loop variable to `_root` in fsspec.walk() calls across filesystem and file discovery modules. Addresses CodeRabbit review feedback.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3295 Address PR review feedback for fsspec cache handling

- Add on_error callback to fs_fsspec.walk() for proper error logging
- Disable Azure listings cache by setting listings_expiry_time=0

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3295 [FIX] Handle on_error callback in SharePointFileSystem.walk()

SharePoint's custom walk() was silently swallowing exceptions without
invoking the on_error callback. Now supports callable, "raise", and
"omit" (default) modes matching fsspec's contract.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* UN-3295 [FIX] Fix invalid type annotation in SharePoint walk()

Use `Callable[[Exception], Any]` instead of builtin `callable` which
is not a valid type for annotations without `__future__.annotations`.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Gayathri <142381512+gaya3-zipstack@users.noreply.github.com>

Author: 3 people

backend/workflow_manager/endpoint_v2/source.py

@@ -3,6 +3,7 @@
 import os
 import shutil
 import uuid
+from collections.abc import Collection
 from hashlib import sha256
 from io import BytesIO
 from itertools import islice
@@ -486,14 +487,14 @@ def _get_matched_files(
         max_depth = int(SourceConstant.MAX_RECURSIVE_DEPTH) if recursive else 1
 
         fs_fsspec = source_fs.get_fsspec_fs()
-        for root, dirs, _ in fs_fsspec.walk(input_directory, maxdepth=max_depth):
-            try:
-                fs_metadata_list: list[dict[str, Any]] = fs_fsspec.listdir(
-                    root
-                )  # Single call for file system metadata
-            except Exception as e:
-                logger.warning(f"Failed to list directory from path: {root}, error: {e}")
-                continue
+
+        def _on_walk_error(error: Exception) -> None:
+            logger.warning(f"Failed to list directory: {error}")
+
+        for _root, dirs, files in fs_fsspec.walk(
+            input_directory, maxdepth=max_depth, detail=True, on_error=_on_walk_error
+        ):
+            fs_metadata_list: list[dict[str, Any]] = list(files.values())
 
             count = self._process_file_fs_directory(
                 fs_metadata_list=fs_metadata_list,
@@ -517,7 +518,7 @@ def _process_file_fs_directory(
         matched_files: dict[str, FileHash],
         patterns: list[str],
         source_fs: UnstractFileSystem,
-        dirs: list[str],
+        dirs: Collection[str],
     ) -> int:
         for fs_metadata in fs_metadata_list:
             if count >= limit:
@@ -708,7 +709,7 @@ def _is_directory(
         source_fs: UnstractFileSystem,
         file_path: str,
         metadata: dict[str, Any],
-        dirs: list[str],
+        dirs: Collection[str],
     ) -> bool:
         """Check if the given path is a directory.
 
@@ -717,7 +718,7 @@ def _is_directory(
                 reading the file.
             file_path (str): The path of the file.
             metadata (dict[str, Any]): The metadata of the file.
-            dirs (list[str]): The list of directories.
+            dirs (Collection[str]): The collection of directory names.
 
         Returns:
             bool: True if the file is a directory, False otherwise.

frontend/src/hooks/useClearFileHistory.js

@@ -35,6 +35,10 @@ const useClearFileHistory = () => {
 
     try {
       await workflowServiceInstance.clearFileMarkers(workflowId);
+      setAlertDetails({
+        type: "success",
+        content: "File history cleared successfully",
+      });
       return true;
     } catch (err) {
       setAlertDetails(handleException(err, "Failed to clear file history"));

unstract/connectors/src/unstract/connectors/filesystems/azure_cloud_storage/azure_cloud_storage.py

@@ -39,8 +39,15 @@ def __init__(self, settings: dict[str, Any]):
         account_name = settings.get("account_name", "")
         access_key = settings.get("access_key", "")
         self.bucket = settings.get("bucket", "")
+        # adlfs._ls_containers() unconditionally reads from DirCache after
+        # populating it — use_listings_cache=False makes the write a no-op,
+        # causing a KeyError. Setting listings_expiry_time=0 allows writes
+        # but expires entries immediately on subsequent reads.
+        # Check https://github.com/fsspec/adlfs/issues/230 for more context.
         self.azure_fs = AzureBlobFileSystem(
-            account_name=account_name, credential=access_key
+            account_name=account_name,
+            credential=access_key,
+            listings_expiry_time=0,
         )
 
     @staticmethod

unstract/connectors/src/unstract/connectors/filesystems/box/box.py

@@ -36,7 +36,9 @@ def __init__(self, settings: dict[str, Any]):
         try:
             oauth = JWTAuth.from_settings_dictionary(settings_dict)
             root_id = 0
-            self.box_fs = BoxFileSystem(oauth=oauth, root_id=root_id)
+            self.box_fs = BoxFileSystem(
+                oauth=oauth, root_id=root_id, use_listings_cache=False
+            )
         except ValueError as e:
             raise ConnectorError(
                 f"Error initialising from Box app settings: {e}",

unstract/connectors/src/unstract/connectors/filesystems/minio/minio.py

@@ -26,6 +26,7 @@ def __init__(self, settings: dict[str, Any]):
             anon=False,
             key=key,
             secret=secret,
+            use_listings_cache=False,
             default_fill_cache=False,
             default_cache_type="none",
             skip_instance_cache=True,

unstract/connectors/src/unstract/connectors/filesystems/sharepoint/sharepoint.py

@@ -11,6 +11,7 @@
 import logging
 import os
 import threading
+from collections.abc import Callable
 from datetime import UTC, datetime
 from typing import Any, NoReturn
 
@@ -415,6 +416,7 @@ def walk(
         self,
         path: str = "",
         maxdepth: int | None = None,
+        on_error: str | Callable[[Exception], Any] = "omit",
         **kwargs: Any,
     ):
         """Walk directory tree."""
@@ -424,7 +426,11 @@ def walk(
         path = self._normalize_path(path)
         try:
             items = self.ls(path, detail=True)
-        except Exception:
+        except Exception as e:
+            if callable(on_error):
+                on_error(e)
+            elif on_error == "raise":
+                raise
             return
 
         dirs = []
@@ -441,7 +447,7 @@ def walk(
         for d in dirs:
             subpath = f"{path}/{d}" if path != "root" else d
             new_depth = maxdepth - 1 if maxdepth is not None else None
-            yield from self.walk(subpath, maxdepth=new_depth, **kwargs)
+            yield from self.walk(subpath, maxdepth=new_depth, on_error=on_error, **kwargs)
 
 
 class SharePointFS(UnstractFileSystem):

unstract/connectors/src/unstract/connectors/filesystems/unstract_file_system.py

@@ -187,23 +187,31 @@ def list_files(
         Returns:
             List of file metadata dictionaries
         """
-        all_files = []
+        all_files: list[dict[str, Any]] = []
         fs_fsspec = self.get_fsspec_fs()
 
-        for root, dirs, _ in fs_fsspec.walk(directory, maxdepth=max_depth):
-            try:
-                fs_metadata_list = fs_fsspec.listdir(root, detail=True)
-                for metadata in fs_metadata_list:
-                    if not include_dirs and self.is_dir_by_metadata(metadata):
-                        continue
-                    all_files.append(metadata)
-
-                    if limit is not None and len(all_files) >= limit:
-                        return all_files
-            except Exception as e:
-                logger.warning(f"Failed to list directory {root}: {e}")
-                self._store_user_error(f"Could not access directory: {root}")
-                continue
+        def _on_walk_error(error: Exception) -> None:
+            """Callback for walk errors — preserves user error reporting."""
+            logger.warning(f"Failed to list directory: {error}")
+            self._store_user_error(f"Could not access directory: {error}")
+
+        for _root, dirs, files in fs_fsspec.walk(
+            directory, maxdepth=max_depth, detail=True, on_error=_on_walk_error
+        ):
+            if include_dirs:
+                items = list(dirs.values()) + list(files.values())
+            else:
+                items = list(files.values())
+
+            for metadata in items:
+                # Safety check: some providers (GCS zero-size objects, Azure
+                # is_directory tag) mis-classify directories as files.
+                if not include_dirs and self.is_dir_by_metadata(metadata):
+                    continue
+                all_files.append(metadata)
+
+                if limit is not None and len(all_files) >= limit:
+                    return all_files
         return all_files
 
     def report_errors_to_user(self) -> list[str]:

unstract/connectors/src/unstract/connectors/filesystems/zs_dropbox/zs_dropbox.py

@@ -16,7 +16,9 @@ def __init__(self, settings: dict[str, Any]):
         from dropboxdrivefs import DropboxDriveFileSystem
 
         super().__init__("Dropbox")
-        self.dropbox_fs = DropboxDriveFileSystem(token=settings["token"])
+        self.dropbox_fs = DropboxDriveFileSystem(
+            token=settings["token"], use_listings_cache=False
+        )
         self.path = "///"
 
     @staticmethod

workers/shared/processing/file_discovery.py

@@ -5,6 +5,7 @@
 """
 
 import time
+from collections.abc import Collection
 from typing import Any
 
 from unstract.connectors.filesystems.unstract_file_system import UnstractFileSystem
@@ -115,22 +116,22 @@ def discover_files_streaming(
 
                 logger.info(f"[StreamingDiscovery] Processing directory: {directory}")
 
+                def _on_walk_error(error: Exception) -> None:
+                    logger.warning(
+                        f"[StreamingDiscovery] Failed to list directory: {error}"
+                    )
+
                 # Walk directory with max depth control
-                for root, dirs, _ in self.fs_fsspec.walk(directory, maxdepth=max_depth):
+                for _root, dirs, files in self.fs_fsspec.walk(
+                    directory, maxdepth=max_depth, detail=True, on_error=_on_walk_error
+                ):
                     metrics["directories_walked"] += 1
 
                     # Check limit before processing directory
                     if len(matched_files) >= file_hard_limit:
                         break
 
-                    try:
-                        # Get all items in directory with metadata
-                        fs_metadata_list: list[dict[str, Any]] = self.fs_fsspec.listdir(
-                            root
-                        )
-                    except Exception as e:
-                        logger.warning(f"Failed to list directory {root}: {e}")
-                        continue
+                    fs_metadata_list: list[dict[str, Any]] = list(files.values())
 
                     # Process files in this directory
                     for fs_metadata in fs_metadata_list:
@@ -395,14 +396,14 @@ def _process_batch(
             )
 
     def _is_directory(
-        self, file_path: str, metadata: dict[str, Any], dirs: list[str]
+        self, file_path: str, metadata: dict[str, Any], dirs: Collection[str]
     ) -> bool:
         """Check if path is a directory using multiple detection methods.
 
         Args:
             file_path: Path to check
             metadata: File metadata from fsspec
-            dirs: List of directories from walk
+            dirs: Collection of directory names from walk
 
         Returns:
             True if path is a directory