1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
"""Files on disk and the index cache."""
from __future__ import annotations
import asyncio
import logging
from pathlib import Path
from meshbay_node.ops.core import OpError, _group_ctx
log = logging.getLogger("meshbay_node.ops")
# ── Files ────────────────────────────────────────────────────────────────────
async def delete_file(state: dict, group_id: str, file_id: str) -> dict:
"""
Remove a file from a group. Milestone 14.11 — the last operator action that
needed a browser.
Authorization happened in the adapter. On the loopback path that is the
session token, which means physical or SSH access to the machine hosting the
files — an operator who can run this can also `rm` the file, so the check is
not weaker than the alternative.
"""
ctx = _group_ctx(state, group_id)
index = ctx.get("index")
roots = ctx.get("roots")
if not index or not roots:
raise OpError("Group has no index", status=503)
entry = index.get_entry(file_id)
if not entry:
raise OpError("No such file in this group", status=404)
from meshbay_node.roots import entry_abs_path
path = entry_abs_path(roots, entry)
if path is None:
raise OpError(
f"{entry.name!r} is in root {entry.path.split('/')[0]!r}, which is "
f"not readable right now — the file is frozen, not gone", status=409)
try:
path.unlink()
except FileNotFoundError:
# Already gone from disk; drop the stale entry rather than refusing.
log.warning("Index named a file that is not on disk: %s", path)
except OSError as e:
raise OpError(f"Cannot delete {entry.name!r}: {e}", status=500) from e
index.remove_entry(file_id)
log.info("File deleted by operator: %s/%s", entry.path, entry.name)
return {"status": "deleted", "name": entry.name, "path": entry.path,
"group_id": group_id}
# ── Index cache maintenance ───────────────────────────────────────────────────
#
# The (path, size, mtime) -> hash accelerator (indexer/cache.py) is node-wide
# and grows for as long as a path was ever seen — a folder an operator later
# stops sharing (root removed, or every group hosting it is deleted) leaves
# its rows behind forever otherwise. Nothing about correctness needs this:
# a stale row just sits unused (lookup() keys on the live path string, so a
# path nothing scans any more is never looked up). This is disk space
# hygiene the operator can run when they want it, not a background job.
async def index_cache_stats(state: dict) -> dict:
"""Row count only — cheap, safe to call on every dashboard render.
The actual staleness check (prune_index_cache) is not this cheap and
must never run implicitly."""
cache = state.get("index_cache")
return {"count": await cache.count() if cache else 0}
async def prune_index_cache(state: dict) -> dict:
"""
Drop cache rows that cannot be right for anything any more: the path is
not under any group's root at all, or it is under a root that is
available right now and the file is genuinely gone from disk.
Deliberately leaves alone anything under a root that is currently
*unavailable* (a disconnected drive) — indexer.py's own rule is that
such a root freezes rather than empties, precisely so it does not pay a
full rehash the moment it comes back. Pruning through an unavailable
root here would reintroduce exactly that cost via a different door, so
an owning-but-unavailable root wins over "the file isn't there right
now" every time, unconditionally.
A row lost here costs one rehash the next time that path is scanned,
never a wrong answer: lookup() (cache.py) always re-validates size and
mtime against a live stat() before trusting a cached hash.
"""
cache = state.get("index_cache")
if cache is None:
raise OpError("No index cache in this process", status=503)
indexers = list((state.get("indexers") or {}).values())
roots = [root for indexer in indexers for root in indexer.roots]
def _is_stale(path_str: str) -> bool:
path = Path(path_str)
owning = [r for r in roots if r.path in path.parents]
if not owning:
return True
if any(not r.available for r in owning):
return False
return not path.exists()
paths = await cache.all_paths()
stale = await asyncio.to_thread(lambda: [p for p in paths if _is_stale(p)])
await cache.remove_many(stale)
log.info("Index cache pruned: %d stale row(s) removed, %d kept",
len(stale), len(paths) - len(stale))
return {"status": "pruned", "removed": len(stale), "kept": len(paths) - len(stale)}
|