fix(node): plug an auto-ejected removable root back once its files return

A node started with the desktop session runs before the session has mounted
its USB drives. The safety net then auto-ejected every removable root and
persisted it exactly like an operator's eject, so after each reboot those
roots stayed ejected until someone plugged them by hand (seen on a node whose
/media drives were mounted a minute after it started).

An auto-eject is now stored as such ("auto" in roster.db). At startup and at
every reconcile, an auto-ejected root whose path is readable again is checked
against a few files the hash cache knows under it, at the same path with the
same size and mtime; one found and the root is plugged back and rescanned.
An empty mount point or another drive in its place is not recognised and
stays ejected. An operator's eject is never undone automatically.

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

1 files changed, 64 insertions, 1 deletions

diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index f0505f7..5e139bb 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -389,6 +389,7 @@ class DirectoryIndexer:
 
     async def _initial_scan(self) -> None:
         await off_disk(self.roots, self.roots.refresh_availability)
+        await self._plug_back_recognised()
         total = 0
         waiting = [r.name for r in self.roots if r.available]
         self._queue(waiting)
@@ -825,6 +826,8 @@ class DirectoryIndexer:
                 except Exception:
                     log.exception("Could not persist the auto-eject of root %r", name)
 
+        changed += [(root, True) for root in await self._plug_back_recognised()]
+
         for root, available in changed:
             if available:
                 log.info("Root %r is back — rescanning", root.name)
@@ -1032,6 +1035,65 @@ class DirectoryIndexer:
             self._observer = None
         self._start_observer()
 
+    # How many of a root's known files are looked for before it is plugged back
+    # automatically. One found is enough: another drive, or an empty mount
+    # point, holds none of them at the same path with the same size and mtime.
+    RECOGNISE_SAMPLE = 5
+
+    async def _plug_back_recognised(self) -> list[Root]:
+        """
+        Un-eject the roots the safety net ejected, once they hold their own files
+        again: a drive not mounted yet when the node started, or unplugged and
+        plugged back. An operator's eject is never undone here.
+        """
+        back = []
+        for root in [r for r in self.roots if r.ejected and r.auto]:
+            if not await off_disk(self.roots, self._recognises, root,
+                                  await self._known_files(root)):
+                continue
+            root.ejected = root.auto = False
+            root.available = True
+            log.info("Root %r is back with its files — plugged automatically", root.name)
+            if self.on_root_ejected:
+                try:
+                    await self.on_root_ejected(root.name, False)
+                except Exception:
+                    log.exception("Could not persist the return of root %r", root.name)
+            back.append(root)
+        return back
+
+    async def _known_files(self, root: Root) -> list[tuple[str, int, float | None]]:
+        """Files this root is known to hold, as (path, size, mtime or None).
+
+        The hash cache first: it survives a restart, when an ejected root has no
+        entry in the index at all. The index is the fallback for a node without
+        a cache.
+        """
+        if self._cache is not None:
+            known = await self._cache.sample_under(str(root.path), self.RECOGNISE_SAMPLE)
+            if known:
+                return known
+        return [(str(path), e.size, None)
+                for e in self._entries_under(root)[:self.RECOGNISE_SAMPLE]
+                if (path := self._entry_path(root, e)) is not None]
+
+    @staticmethod
+    def _recognises(root: Root, known: list[tuple[str, int, float | None]]) -> bool:
+        """Blocking: is this the root's own content, readable again?"""
+        if not root.is_live():
+            return False
+        if not known:
+            # Nothing known under it, so nothing a wrong disk could pass for.
+            return True
+        for path, size, mtime in known:
+            try:
+                st = Path(path).stat()
+            except OSError:
+                continue
+            if st.st_size == size and (mtime is None or st.st_mtime == mtime):
+                return True
+        return False
+
     def eject_root(self, root_name: str) -> None:
         """Stop watching a root without touching its entries."""
         from meshbay_common.paths import fold
@@ -1039,6 +1101,7 @@ class DirectoryIndexer:
         for root in self.roots:
             if fold(root.name) == target:
                 root.ejected = True
+                root.auto = False
                 root.available = False
                 frozen = len(self._entries_under(root))
                 log.info("Root %r ejected — %d entries frozen", root.name, frozen)
@@ -1058,7 +1121,7 @@ class DirectoryIndexer:
                 break
         if root is None:
             return
-        root.ejected = False
+        root.ejected = root.auto = False
         root.available = await off_disk(self.roots, root.is_live)
         if not root.available:
             await self._finish_plug(None)