diff options
4 files changed, 79 insertions, 15 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 0015f1a..667e1eb 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -160,8 +160,17 @@ class NodeDaemon: # restart retries everything, matching the "disposable, rebuildable" # stance the rest of this cache takes (docs/mediacenter.md §1/§2). # Shared across the video and audio enrichment paths — content- - # addressed ids never collide between the two. - self._enriched_attempted: set[str] = set() + # addressed ids never collide between the two. Keyed by + # (group_id, entry.id), not entry.id alone: the id is a content + # hash, so the same physical file shared into two different groups + # (found live — overlapping test libraries across several demo + # groups) produces the same id in both. A bare-id set marked the + # second group's copy "already attempted" the moment the first + # group's enrichment ran, even though nothing had ever populated + # *that* group's own index — every file in the second group stayed + # at duration 0 with no artist/album, permanently, since nothing + # ever revisits an id already in this set. + self._enriched_attempted: set[tuple[str, str]] = set() self._roster: Roster | None = None self._indexers: list[DirectoryIndexer] = [] self._tasks: list[asyncio.Task] = [] @@ -1098,14 +1107,14 @@ class NodeDaemon: if not video_root: return for entry in entries: - if entry.type != "video" or entry.id in self._enriched_attempted: + if entry.type != "video" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_video_root(entry.path, video_root): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue - self._enriched_attempted.add(entry.id) + self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) @@ -1151,7 +1160,7 @@ class NodeDaemon: old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue - self._enriched_attempted.discard(entry.id) + self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_video_entries(indexer, updates) async def _enrich_new_audio_entries(self, indexer: DirectoryIndexer, entries: list) -> None: @@ -1172,14 +1181,14 @@ class NodeDaemon: return root_boundary = indexer.roots.resolve(audio_root, require_available=False) for entry in entries: - if entry.type != "audio" or entry.id in self._enriched_attempted: + if entry.type != "audio" or (indexer.group_id, entry.id) in self._enriched_attempted: continue if not _under_audio_root(entry.path, audio_root): continue file_path = entry_abs_path(indexer.roots, entry) if not file_path or not file_path.exists(): continue - self._enriched_attempted.add(entry.id) + self._enriched_attempted.add((indexer.group_id, entry.id)) async def on_done(file_id: str, fields: dict, _indexer=indexer) -> None: await self._on_enriched(_indexer, file_id, fields) @@ -1223,7 +1232,7 @@ class NodeDaemon: old = previous.get_entry(entry.id) if old is None or (old.name == entry.name and old.path == entry.path): continue - self._enriched_attempted.discard(entry.id) + self._enriched_attempted.discard((indexer.group_id, entry.id)) await self._enrich_new_audio_entries(indexer, updates) async def _on_enriched(self, indexer: DirectoryIndexer, file_id: str, fields: dict) -> None: diff --git a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py index f088f5e..deab1bd 100644 --- a/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py +++ b/packages/meshbay-node/tests/test_audio_root_gates_enrichment.py @@ -111,8 +111,8 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path): await asyncio.sleep(0.05) by_name = {e.name: e for e in indexer.index.entries} - assert by_name["in-root.mp3"].id in daemon._enriched_attempted - assert by_name["outside.mp3"].id not in daemon._enriched_attempted, ( + assert (group_id, by_name["in-root.mp3"].id) in daemon._enriched_attempted + assert (group_id, by_name["outside.mp3"].id) not in daemon._enriched_attempted, ( "a file outside the configured audio_root must never be enriched") finally: await _teardown(daemon) @@ -149,8 +149,63 @@ async def test_setting_the_audio_root_sweeps_what_it_already_contains(tmp_path): await asyncio.sleep(0.05) # let the fire-and-forget sweep actually run entry = next(iter(indexer.index.entries)) - assert entry.id in daemon._enriched_attempted, ( + assert (group_id, entry.id) in daemon._enriched_attempted, ( "a file already sitting in the newly-chosen root must be picked " "up by the sweep, not wait for some unrelated future change") finally: await _teardown(daemon) + + +async def test_the_same_file_shared_into_two_groups_enriches_in_both(tmp_path): + """ + Regression, found live: `entry.id` is a content hash, so the exact same + physical file (byte-for-byte, e.g. a test/demo library reused across + several groups) produces the *same id* wherever it's indexed. + `_enriched_attempted` used to be keyed by that bare id alone — global + across every group this node hosts — so the moment group A's copy got + enriched, group B's otherwise-identical copy read as "already + attempted" and was skipped forever, even though nothing had ever + populated *group B's own* index. Every file in the group manifested as + duration 0 with no artist/album, permanently — nothing else ever + revisits an id once it's in this set. Now keyed by (group_id, id). + """ + shared_a = tmp_path / "shared_a" + shared_a.mkdir() + (shared_a / "Music").mkdir() + (shared_a / "Music" / "track.mp3").write_bytes(_AUDIO_BYTES) + + shared_b = tmp_path / "shared_b" + shared_b.mkdir() + (shared_b / "Music").mkdir() + (shared_b / "Music" / "track.mp3").write_bytes(_AUDIO_BYTES) # identical content + + group_a, group_b = "a" * 32, "b" * 32 + daemon = await _make_daemon(tmp_path, shared_a, group_a) + try: + indexer_a = DirectoryIndexer( + roots=one_root(shared_a), group_id=group_a, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek()) + indexer_b = DirectoryIndexer( + roots=one_root(shared_b), group_id=group_b, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek()) + await indexer_a.initial_scan() + await indexer_b.initial_scan() + + entry_a = next(iter(indexer_a.index.entries)) + entry_b = next(iter(indexer_b.index.entries)) + assert entry_a.id == entry_b.id, ( + "the fixture itself must produce identical content hashes — " + "otherwise this test isn't exercising the collision at all") + + await daemon._roster.set_audio_root(group_a, "shared_a/Music", set_by="op") + await daemon._roster.set_audio_root(group_b, "shared_b/Music", set_by="op") + + await daemon._enrich_new_audio_entries(indexer_a, list(indexer_a.index.entries)) + await daemon._enrich_new_audio_entries(indexer_b, list(indexer_b.index.entries)) + + assert (group_a, entry_a.id) in daemon._enriched_attempted + assert (group_b, entry_b.id) in daemon._enriched_attempted, ( + "group B's copy must be enriched independently of group A's, " + "even though the two entries share the exact same id") + finally: + await _teardown(daemon) diff --git a/packages/meshbay-node/tests/test_startup_scan_enrichment.py b/packages/meshbay-node/tests/test_startup_scan_enrichment.py index cdadad9..65b9728 100644 --- a/packages/meshbay-node/tests/test_startup_scan_enrichment.py +++ b/packages/meshbay-node/tests/test_startup_scan_enrichment.py @@ -84,7 +84,7 @@ async def test_a_file_already_on_disk_at_startup_gets_enrichment_scheduled(tmp_p await asyncio.sleep(0.05) # let the coalescing timer fire _broadcast_index_change entry = next(iter(indexer.index.entries)) - assert entry.id in daemon._enriched_attempted, ( + assert ("a" * 32, entry.id) in daemon._enriched_attempted, ( "a file already on disk at startup must get enrichment scheduled the " "first time its group's index is broadcast, not only on a later " "watchdog-detected change to it") diff --git a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py index df86a79..9cfb819 100644 --- a/packages/meshbay-node/tests/test_video_root_gates_enrichment.py +++ b/packages/meshbay-node/tests/test_video_root_gates_enrichment.py @@ -108,8 +108,8 @@ async def test_only_entries_under_the_configured_root_are_enriched(tmp_path): await asyncio.sleep(0.05) by_name = {e.name: e for e in indexer.index.entries} - assert by_name["in-root.mkv"].id in daemon._enriched_attempted - assert by_name["outside.mkv"].id not in daemon._enriched_attempted, ( + assert (group_id, by_name["in-root.mkv"].id) in daemon._enriched_attempted + assert (group_id, by_name["outside.mkv"].id) not in daemon._enriched_attempted, ( "a file outside the configured video_root must never be enriched") finally: await _teardown(daemon) @@ -146,7 +146,7 @@ async def test_setting_the_video_root_sweeps_what_it_already_contains(tmp_path): await asyncio.sleep(0.05) # let the fire-and-forget sweep actually run entry = next(iter(indexer.index.entries)) - assert entry.id in daemon._enriched_attempted, ( + assert (group_id, entry.id) in daemon._enriched_attempted, ( "a file already sitting in the newly-chosen root must be picked " "up by the sweep, not wait for some unrelated future change") finally: |