fix(metadata-sync): persist db_checked flag for deleted models

When a deleted model is checked against the SQLite archive and not found, the `db_checked` flag was set in memory but never saved to disk. This occurred because the save operation was only triggered when `civitai_api_not_found` was True, which is not the case for deleted models (since the CivitAI API is not attempted). As a result, deleted models would be rechecked on every refresh instead of being skipped.

Changes:
- Introduce a `needs_save` flag to track when metadata state is updated
- Save metadata whenever `db_checked` is set to True, regardless of API status
- Ensure `last_checked_at` is set for SQLite-only attempts
- Add regression test to verify the fix
This commit is contained in:
Will Miao
2026-02-03 07:34:41 +08:00
parent bd83f7520e
commit 2c026a2646
2 changed files with 85 additions and 0 deletions

View File

@@ -243,17 +243,27 @@ class MetadataSyncService:
last_error = error or last_error
if civitai_metadata is None or metadata_provider is None:
# Track if we need to save metadata
needs_save = False
if sqlite_attempted:
model_data["db_checked"] = True
needs_save = True
if civitai_api_not_found:
model_data["from_civitai"] = False
model_data["civitai_deleted"] = True
model_data["db_checked"] = sqlite_attempted or (enable_archive and model_data.get("db_checked", False))
model_data["last_checked_at"] = datetime.now().timestamp()
needs_save = True
# Save metadata if any state was updated
if needs_save:
data_to_save = model_data.copy()
data_to_save.pop("folder", None)
# Update last_checked_at for sqlite-only attempts if not already set
if "last_checked_at" not in data_to_save:
data_to_save["last_checked_at"] = datetime.now().timestamp()
await self._metadata_manager.save_metadata(file_path, data_to_save)
default_error = (

View File

@@ -482,6 +482,81 @@ async def test_relink_metadata_raises_when_version_missing():
model_version_id=None,
)
@pytest.mark.asyncio
async def test_fetch_and_update_model_persists_db_checked_when_sqlite_fails(tmp_path):
"""
Regression test: When a deleted model is checked against sqlite and not found,
db_checked=True must be persisted to disk so the model is skipped in future refreshes.
Previously, db_checked was set in memory but never saved because the save_metadata
call was inside the `if civitai_api_not_found:` block, which is False for deleted
models (since the default CivitAI API is never tried).
"""
default_provider = SimpleNamespace(
get_model_by_hash=AsyncMock(),
get_model_version=AsyncMock(),
)
civarchive_provider = SimpleNamespace(
get_model_by_hash=AsyncMock(return_value=(None, "Model not found")),
get_model_version=AsyncMock(),
)
sqlite_provider = SimpleNamespace(
get_model_by_hash=AsyncMock(return_value=(None, "Model not found")),
get_model_version=AsyncMock(),
)
async def select_provider(name: str):
if name == "civarchive_api":
return civarchive_provider
if name == "sqlite":
return sqlite_provider
return default_provider
provider_selector = AsyncMock(side_effect=select_provider)
helpers = build_service(
settings_values={"enable_metadata_archive_db": True},
default_provider=default_provider,
provider_selector=provider_selector,
)
model_path = tmp_path / "model.safetensors"
model_data = {
"civitai_deleted": True,
"db_checked": False,
"from_civitai": False,
"file_path": str(model_path),
"model_name": "Deleted Model",
}
update_cache = AsyncMock()
ok, error = await helpers.service.fetch_and_update_model(
sha256="deadbeef",
file_path=str(model_path),
model_data=model_data,
update_cache_func=update_cache,
)
# The call should fail because neither provider found metadata
assert not ok
assert error is not None
assert "Model not found" in error or "not found in metadata archive DB" in error
# Both providers should have been tried
assert civarchive_provider.get_model_by_hash.await_count == 1
assert sqlite_provider.get_model_by_hash.await_count == 1
# db_checked should be True in memory
assert model_data["db_checked"] is True
# CRITICAL: metadata should have been saved to disk with db_checked=True
helpers.metadata_manager.save_metadata.assert_awaited_once()
saved_call = helpers.metadata_manager.save_metadata.await_args
saved_data = saved_call.args[1]
assert saved_data["db_checked"] is True
assert "folder" not in saved_data # folder should be stripped
assert "last_checked_at" in saved_data # timestamp should be set
@pytest.mark.asyncio
async def test_fetch_and_update_model_does_not_overwrite_api_metadata_with_archive(tmp_path):
helpers = build_service()