summaryrefslogtreecommitdiffstats
path: root/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
diff options
context:
space:
mode:
authorChristophe Besson <cbesson@gmail.com>2026-09-18 15:59:24 +0200
committerChristophe Besson <cbesson@gmail.com>2026-09-18 15:59:24 +0200
commit5fa158fab709d3d24a33318b3d910f75c051af2e (patch)
tree937a475e878b248cb3415e4fa23303988ac21221 /packages/meshbay-node/src/meshbay_node/indexer/indexer.py
parent79b8f770ab319cf8d64132a56fc0036dcf0486f7 (diff)
downloadmeshbay-5fa158fab709d3d24a33318b3d910f75c051af2e.tar.gz
fix(node): take the availability poll and every upload write off the loop
The rest of AV9's disk half. Serving a file left the loop in the commit before this one; two paths were still on it. **The availability poll.** `RootSet.refresh_availability` stats every root, and eleven call sites reached it from `async def` — the reconcile loop among them, on a timer. On a sleeping disk that is a stall once per tick, and the stat is also what keeps the disk awake, so a node paid spin-up for a library nobody was reading. All eleven now go through `off_disk`, `Root.is_live` included. **The upload write.** `open`/`write`, and the resolve, the stat, the free-name search, the rename and the unlink around it. This one could not simply be awaited: the handler was synchronous, so nothing could come between the `chunk_index != state.next_index` check and the `advance` that answers it, and that is the whole of the chunk-ordering rule. Awaiting the write opens the gap — chunk 1 arriving while chunk 0 is in the disk thread reads a position that has not moved and is refused as out of order, so an upload would fail on a slow disk and nowhere else. Verified, not assumed: without the lock the new ordering test refuses three chunks of four. So the check, the write and the advance are one critical section again, under a lock held **per group**. Not per session: `partial_uploads` lives in the group context so a reconnecting client finds its upload where it left it, which means two sessions of one member share the position of one `.part` file. Arrival order is preserved by construction — the dispatcher creates one task per message as it arrives, tasks start in creation order, and the lock is the first thing each one waits on, so its waiters queue in arrival order too. `_do_file_upload` is a coroutine now, which is why forty-two test call sites gain an `await`. Their outcomes are unchanged, file by file, against the run before the change. `test_ops.py` asked which public coroutines `ops` exposes and got `off_disk`, imported rather than defined there. It now asks for the ones written in the module, which is what its own docstring means; all forty-three operations are still checked. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/indexer/indexer.py')
-rw-r--r--packages/meshbay-node/src/meshbay_node/indexer/indexer.py15
1 files changed, 8 insertions, 7 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
index c1b151a..887d435 100644
--- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
+++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py
@@ -40,7 +40,7 @@ from meshbay_common.paths import fold, find_fold_collisions, long_path
from meshbay_common.protocol import IndexEntry
from meshbay_node.indexer.cache import IndexCache
from meshbay_node.indexer.group_index import GroupIndex
-from meshbay_node.roots import Root, RootSet
+from meshbay_node.roots import Root, RootSet, off_disk
log = logging.getLogger(__name__)
@@ -387,7 +387,7 @@ class DirectoryIndexer:
await self._initial_scan()
async def _initial_scan(self) -> None:
- self.roots.refresh_availability()
+ await off_disk(self.roots, self.roots.refresh_availability)
total = 0
waiting = [r.name for r in self.roots if r.available]
self._queue(waiting)
@@ -695,7 +695,7 @@ class DirectoryIndexer:
self._index.remove_entry(entry.id)
self.roots = roots
- roots.refresh_availability()
+ await off_disk(roots, roots.refresh_availability)
added = [r for r in roots if r.folded not in old_names and r.available]
self._index.roots = roots.describe()
@@ -794,7 +794,7 @@ class DirectoryIndexer:
return await self._reconcile()
async def _reconcile(self) -> bool:
- changed = self.roots.refresh_availability()
+ changed = await off_disk(self.roots, self.roots.refresh_availability)
touched = False
# Drained before the loop below, because persisting the flag is what
@@ -1042,7 +1042,7 @@ class DirectoryIndexer:
if root is None:
return
root.ejected = False
- root.available = root.is_live()
+ root.available = await off_disk(self.roots, root.is_live)
if not root.available:
await self._finish_plug(None)
return
@@ -1071,7 +1071,8 @@ class DirectoryIndexer:
# Ejected or removed again while it waited. `_rescan_root`
# drops the entries before it walks, so going ahead would
# empty a root that is not there to be read.
- if self._holds(root) and not root.ejected and root.is_live():
+ live = await off_disk(self.roots, root.is_live)
+ if self._holds(root) and not root.ejected and live:
await self._rescan_root(root)
finally:
if waiting:
@@ -1143,7 +1144,7 @@ class DirectoryIndexer:
if root is None:
return
- if deleted and not root.is_live():
+ if deleted and not await off_disk(self.roots, root.is_live):
# The volume went away rather than the file. Freeze: mark the
# root and touch nothing. Every other event for this root
# will arrive here too and be dropped the same way, which is