fix: improve ChromaDB provisioner reliability and dataset deletion safety

- Verify actual OS process before reusing DB-marked-running provisioner
- Wait for ChromaDB HTTP server health before marking provisioner ready
- Use psutil for PID identity validation to prevent wrong-process kills
- Add wait_until_ready() to provisioner protocol with no-op default
- Bundle rapidocr data files in PyInstaller spec (docling dependency)
- Catch chromadb.errors.NotFoundError on collection delete for un-ingested datasets
- Stop ingestion (cancel pending jobs + stop watcher) before dataset deletion
- Add TOCTOU guard: re-check dataset exists before ingest() call

Author: itstauq

backend/syft-space-backend.spec

@@ -23,6 +23,7 @@ PACKAGES_WITH_DATA = [
     'docling_core',
     'docling_ibm_models',
     'docling_parse',
+    'rapidocr',
 ]
 
 # Packages that use importlib.metadata at runtime (need their dist-info).

backend/syft_space/components/dataset_types/chromadb_local/chromadb_provisioner.py

@@ -228,20 +228,50 @@ async def status(cls, state: dict[str, Any]) -> str:
         else:
             return "starting"
 
+    @classmethod
+    async def wait_until_ready(cls, state: dict[str, Any]) -> None:
+        """Wait for ChromaDB HTTP server to be healthy.
+
+        Args:
+            state: State dict from start()
+
+        Raises:
+            TimeoutError: If not healthy within 60s or no port in state
+        """
+        http_port = state.get("httpPort") or state.get("http_port")
+        if not http_port:
+            raise TimeoutError("No HTTP port in provisioner state")
+        await cls._wait_for_healthy(int(http_port))
+
     @classmethod
     async def _is_process_running(cls, pid: int) -> bool:
-        """Check if a process is running.
+        """Check if a ChromaDB process is running.
+
+        Uses psutil to verify the PID belongs to a ChromaDB process
+        (not a reused PID for something else). Falls back to basic
+        os.kill check only if psutil is not installed.
 
         Args:
             pid: Process ID to check
 
         Returns:
-            True if process exists, False otherwise
+            True if process exists and is a ChromaDB process, False otherwise
         """
         try:
-            os.kill(pid, 0)
-            return True
-        except (OSError, ProcessLookupError):
+            import psutil
+        except ImportError:
+            # psutil not available — fall back to basic PID check
+            try:
+                os.kill(pid, 0)
+                return True
+            except (OSError, ProcessLookupError):
+                return False
+
+        try:
+            proc = psutil.Process(pid)
+            cmdline = " ".join(proc.cmdline()).lower()
+            return "chroma" in cmdline
+        except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
             return False
 
     @classmethod

backend/syft_space/components/dataset_types/chromadb_local/chromadb_type.py

@@ -12,6 +12,7 @@
 from typing import Any
 
 from anyio import Path as AsyncPath
+from loguru import logger
 from pydantic import BaseModel, Field, ValidationError, field_validator
 
 from syft_space.components.dataset_types.chunking import (
@@ -34,6 +35,12 @@
 from syft_space.components.shared.utils import ConfigSchemaGenerator
 
 
+try:
+    from chromadb.errors import NotFoundError as _ChromaNotFoundError
+except ImportError:
+    _ChromaNotFoundError = None
+
+
 def _import_chromadb() -> ModuleType:
     try:
         import chromadb
@@ -575,7 +582,14 @@ async def delete(self, ctx: IngestContext) -> None:
         try:
             await client.delete_collection(name=self.collection_name)
         except Exception as e:
-            raise ValueError(f"Error deleting collection: {str(e)}") from e
+            # Collection may not exist if no documents were ever ingested
+            if _ChromaNotFoundError and isinstance(e, _ChromaNotFoundError):
+                logger.info(
+                    f"Collection '{self.collection_name}' does not exist, "
+                    "skipping deletion"
+                )
+            else:
+                raise ValueError(f"Error deleting collection: {str(e)}") from e
 
         # Remove all page images for this collection
         await asyncio.to_thread(

backend/syft_space/components/dataset_types/interfaces.py

@@ -318,6 +318,21 @@ async def is_running(cls, state: dict[str, Any]) -> bool:
         """
         ...
 
+    @classmethod
+    async def wait_until_ready(cls, state: dict[str, Any]) -> None:
+        """Wait until the provisioned resource is ready to accept connections.
+
+        Default is a no-op. Override in subclasses that need startup health
+        checks (e.g., HTTP server readiness).
+
+        Args:
+            state: State dictionary returned from start()
+
+        Raises:
+            TimeoutError: If not ready within implementation-defined timeout
+        """
+        return None
+
     @classmethod
     async def status(cls, state: dict[str, Any]) -> str:
         """Get detailed status of the resource.

backend/syft_space/components/datasets/handlers.py

@@ -91,11 +91,41 @@ async def _ensure_provisioner_running(
             # Remote type - no provisioner needed
             return None
 
-        # Check if already running
+        # Check if already running (verify actual process, not just DB state)
         existing = await self.provisioner_state_repository.get_running_by_dtype(dtype)
         if existing:
-            logger.info(f"Reusing existing provisioner for dtype '{dtype}'")
-            return existing
+            state = existing.state or {}
+            if await provisioner_cls.is_running(state):
+                # Process is alive, but the server may still be booting
+                # (e.g. after an app restart the OS process survives but
+                # ChromaDB's HTTP server hasn't finished starting yet).
+                # Wait for it to be healthy before returning.
+                try:
+                    await provisioner_cls.wait_until_ready(state)
+                except (TimeoutError, Exception) as exc:
+                    logger.warning(
+                        f"Provisioner for '{dtype}' process alive but "
+                        f"not ready ({exc}), restarting..."
+                    )
+                    await provisioner_cls.stop(state)
+                    await self.provisioner_state_repository.upsert_status(
+                        dtype=dtype,
+                        status=ProvisionerStatus.STOPPED,
+                    )
+                    # Fall through to start a new one
+                    existing = None
+                if existing:
+                    logger.info(f"Reusing existing provisioner for dtype '{dtype}'")
+                    return existing
+            else:
+                logger.warning(
+                    f"Provisioner for '{dtype}' marked as running in DB "
+                    f"but process is dead, restarting..."
+                )
+                await self.provisioner_state_repository.upsert_status(
+                    dtype=dtype,
+                    status=ProvisionerStatus.STOPPED,
+                )
 
         # Transition to STARTING (creates or updates, guards checked)
         logger.info(f"Starting provisioner for dtype '{dtype}'")

backend/syft_space/components/datasets/routes.py

@@ -231,13 +231,23 @@ async def delete_dataset(
     ) -> dict[str, str]:
         """Delete a dataset.
 
+        Stops any active/pending ingestion before cleaning up resources.
+
         Args:
             name: Dataset name
             tenant: Current tenant (injected)
 
         Returns:
             Success message
         """
+        # Stop ingestion (cancel pending jobs, stop file watcher) before
+        # deleting so that a queued job cannot recreate the collection after
+        # we clean it up.
+        if ingestion_manager:
+            dataset = await handler.repository.get_by_name(name, tenant.id)
+            if dataset:
+                await ingestion_manager.stop_dataset_ingestion(dataset.id)
+
         return await handler.delete_dataset(name, tenant)
 
     @public_route

backend/syft_space/components/ingestion/manager.py

@@ -512,6 +512,22 @@ async def _process_single_job(self, job: IngestionJob) -> None:
             # Create context (use system context for background ingestion)
             ctx = IngestContext(sender="system@openmined.org", dataset_id=dataset.id)
 
+            # Re-check dataset exists right before ingest to close TOCTOU
+            # window (dataset could have been deleted during file I/O above)
+            dataset = await self._dataset_repository.get_by_id(
+                job.dataset_id, job.tenant_id
+            )
+            if not dataset:
+                logger.info(
+                    f"Dataset deleted during ingestion prep for job {job.id}"
+                )
+                await self._ingestion_repository.update_status(
+                    job.id,
+                    IngestionJobStatus.CANCELLED,
+                    "Dataset deleted during processing",
+                )
+                return
+
             # Call ingest (native async)
             await dataset_type.ingest(ctx, ingest_request)