From 6193ed242d8255952e5a50bc90e8e88cf750595a Mon Sep 17 00:00:00 2001 From: changukyu Date: Sat, 11 Jul 2026 16:06:28 +0900 Subject: [PATCH] fix: serialize store read-modify-write and harden seed/queue paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Under ThreadingHTTPServer several store writers ran unlocked read-modify-write cycles, so concurrent operator actions could silently lose each other's updates or revert an in-flight change. Concurrency (_write_lock; external I/O kept OUTSIDE the lock): - record_decision now takes _write_lock so a decision isn't reverted by a concurrent rescore/sweep. - manual_search runs the network search first, then re-reads the submission under the lock before the queryHistory RMW (was clobbering concurrent edits). - _apply_provider_usage_delta merges only usage/status onto the fresh row instead of rewriting a stale pre-search snapshot (kept an in-flight emergency disable from being reverted). - _sync_provider_payloads preserves an operator's disable via an operatorDisabled flag (connectivity-only disables still reconnect). - set_provider_enabled / emergency_disable / _sync_submission_provider_state / collect_keyword_candidates now lock (+ single transaction where applicable). Robustness: - seed: a zero-byte/corrupt image now yields a 'failed' submission + audit instead of an IndexError that aborted the whole import. - ensure_queue: key the UPDATE/SELECT on the existing row id on a folder_path match (was matching zero rows and returning None -> seed TypeError); audit queue activation / legacy migration. - store_enrichment: lastFailure uses the first failed item; auto-search dedup reads only the submission's evidence; log a swallowed face-crop load error. - image_store: documents the pre-existing same-stem id limitation (a prior sibling-dependent fix was reverted — it resurrected decided submissions). Tests: operator-disable survives reseed, usage delta keeps an in-flight disable, seed survives an unanalyzable image, ensure_queue mismatched-id, dismiss persistence. --- src/rights_filter/server/image_store.py | 8 + src/rights_filter/server/sqlite_store.py | 129 +++++++++++----- src/rights_filter/server/store_enrichment.py | 106 +++++++++---- src/rights_filter/server/store_operations.py | 125 ++++++++++----- src/rights_filter/server/store_persistence.py | 21 +-- src/rights_filter/server/store_queue.py | 29 +++- .../rights_filter/server/test_sqlite_store.py | 144 ++++++++++++++++++ 7 files changed, 441 insertions(+), 121 deletions(-) diff --git a/src/rights_filter/server/image_store.py b/src/rights_filter/server/image_store.py index 3d08ea7..046adeb 100644 --- a/src/rights_filter/server/image_store.py +++ b/src/rights_filter/server/image_store.py @@ -78,6 +78,14 @@ class LocalSubmissionImageStore: return normalized def _scan_records(self) -> list[dict[str, Any]]: + # NOTE: a scanned file's id is its bare stem. Two same-stem files + # (a.jpg + a.png) collide and one silently merges into the other — a + # known, pre-existing limitation. A prior fix that appended the extension + # on collision was reverted because the id then depended on sibling files, + # so adding a same-stem sibling later re-derived ids and resurrected + # already-decided submissions as duplicates. A correct fix must preserve + # each incumbent id at seed time (where existing ids are known) — not in + # this pure scan — and carry a one-time asset-path migration. if not self.root.exists(): return [] records: list[dict[str, Any]] = [] diff --git a/src/rights_filter/server/sqlite_store.py b/src/rights_filter/server/sqlite_store.py index 5ac9dd6..d748a1d 100644 --- a/src/rights_filter/server/sqlite_store.py +++ b/src/rights_filter/server/sqlite_store.py @@ -282,7 +282,30 @@ class CopyrighterStore( batch.run(submissions) for record in records: - run = repository.analysis_runs_for_submission(record["id"])[-1] + runs = repository.analysis_runs_for_submission(record["id"]) + if not runs: + # BatchAnalyzer saves no run when preprocessing fails (an + # unreadable / zero-byte / corrupt image). Persist a 'failed' + # submission with an audit trail and move on, instead of raising + # IndexError here — which previously aborted the whole seed after + # a partial commit and re-failed on every subsequent reload. + failed_submission = _submission_payload( + record, + 0, + "failed", + ["이미지 분석 실패 — 파일이 손상되었거나 비어 있습니다."], + self.provider_runtime.provider_payloads, + ) + failed_submission["queueId"] = queue_id + self._put("submissions", failed_submission["id"], failed_submission) + self.add_audit_event( + "rights.system", + "Analysis run created", + str(record["id"]), + "이미지 분석 실패 — 손상되었거나 빈 파일", + ) + continue + run = runs[-1] score = run.score submission = _submission_payload( record, @@ -566,22 +589,26 @@ class CopyrighterStore( if decision == "rejected" and not memo.strip(): raise ValueError("reject decision requires memo") - with self._transaction() as conn: - submission = self._get("submissions", submission_id, conn=conn) - submission["decisionStatus"] = decision - submission["applicantStatus"] = "승인 대기" if decision == "approved" else "반려 예정" if decision == "rejected" else "검토 중" - self._put("submissions", submission_id, submission, conn=conn) + # Serialize with the other submission writers (rescore, provider sweeps) + # so an operator decision is not silently reverted by a concurrent + # lock-holding re-put of a stale submission payload. + with self._write_lock: + with self._transaction() as conn: + submission = self._get("submissions", submission_id, conn=conn) + submission["decisionStatus"] = decision + submission["applicantStatus"] = "승인 대기" if decision == "approved" else "반려 예정" if decision == "rejected" else "검토 중" + self._put("submissions", submission_id, submission, conn=conn) - if decision in {"held", "rejected"}: - self._create_or_update_watchlist_entry(submission_id, decision, memo, image_store, conn=conn) + if decision in {"held", "rejected"}: + self._create_or_update_watchlist_entry(submission_id, decision, memo, image_store, conn=conn) - self.add_audit_event( - actor="rights.ops", - event="Operator decision created", - object_id=submission_id, - change=f"{decision} · {memo or 'memo optional'}", - conn=conn, - ) + self.add_audit_event( + actor="rights.ops", + event="Operator decision created", + object_id=submission_id, + change=f"{decision} · {memo or 'memo optional'}", + conn=conn, + ) return self.review(submission_id) def mark_evidence_status( @@ -627,35 +654,55 @@ class CopyrighterStore( return self._all("providers") def set_provider_enabled(self, provider_id: str, enabled: bool) -> list[dict[str, Any]]: - provider = self._get("providers", provider_id) - if enabled and provider_id != "internal": - runtime_provider = self.provider_runtime.provider_payloads.get(provider_id, {}) - if not runtime_provider.get("enabled", False): - reason = str(runtime_provider.get("lastFailure", "")) or f"{provider_id} provider not connected" - raise ValueError(reason) - provider["enabled"] = bool(enabled) - self._put("providers", provider_id, provider) - self._sync_submission_provider_state() - self.add_audit_event( - actor="admin.ops", - event="Provider setting changed", - object_id=provider["name"], - change="enabled" if enabled else "disabled", - ) + with self._write_lock: + provider = self._get("providers", provider_id) + if enabled and provider_id != "internal": + runtime_provider = self.provider_runtime.provider_payloads.get(provider_id, {}) + if not runtime_provider.get("enabled", False): + reason = str(runtime_provider.get("lastFailure", "")) or f"{provider_id} provider not connected" + raise ValueError(reason) + provider["enabled"] = bool(enabled) + # Record that this on/off was an explicit operator action so a later + # reload does not silently flip it back to the connectivity default. + provider["operatorDisabled"] = not bool(enabled) + with self._transaction() as conn: + self._put("providers", provider_id, provider, conn=conn) + self.add_audit_event( + actor="admin.ops", + event="Provider setting changed", + object_id=provider["name"], + change="enabled" if enabled else "disabled", + conn=conn, + ) + self._sync_submission_provider_state() return self.providers() def emergency_disable_external_providers(self) -> list[dict[str, Any]]: - for provider in self.providers(): - if provider["id"] != "internal": - provider["enabled"] = False - self._put("providers", provider["id"], provider) - external_provider_ids = [provider["id"] for provider in self.providers() if provider["id"] != "internal"] - for submission in self._all("submissions"): - provider_state = submission.setdefault("providerState", {}) - for provider_id in external_provider_ids: - provider_state[provider_id] = "disabled" - self._put("submissions", submission["id"], submission) - self.add_audit_event("admin.ops", "Provider setting changed", "external providers", "emergency disabled") + with self._write_lock: + providers = self.providers() + external_provider_ids = [ + provider["id"] for provider in providers if provider["id"] != "internal" + ] + submissions = self._all("submissions") + with self._transaction() as conn: + for provider in providers: + if provider["id"] != "internal": + provider["enabled"] = False + # Deliberate admin stop — must persist across reloads. + provider["operatorDisabled"] = True + self._put("providers", provider["id"], provider, conn=conn) + for submission in submissions: + provider_state = submission.setdefault("providerState", {}) + for provider_id in external_provider_ids: + provider_state[provider_id] = "disabled" + self._put("submissions", submission["id"], submission, conn=conn) + self.add_audit_event( + "admin.ops", + "Provider setting changed", + "external providers", + "emergency disabled", + conn=conn, + ) return self.providers() def audit_events(self, limit: int = 500) -> list[dict[str, Any]]: diff --git a/src/rights_filter/server/store_enrichment.py b/src/rights_filter/server/store_enrichment.py index 23d6904..c31d891 100644 --- a/src/rights_filter/server/store_enrichment.py +++ b/src/rights_filter/server/store_enrichment.py @@ -10,10 +10,13 @@ rescoring provided by the host class. Behavior unchanged. from __future__ import annotations import json +import logging from dataclasses import replace from datetime import datetime from typing import Any +logger = logging.getLogger(__name__) + from rights_filter.analysis.face_person_detection import HeuristicFacePersonDetector from rights_filter.analysis.fingerprints import FingerprintService from rights_filter.analysis.internal_analyzer import InternalAnalyzer @@ -44,31 +47,54 @@ from rights_filter.server.store_serialization import ( class StoreEnrichmentMixin: def _sync_provider_payloads(self) -> None: - existing = {provider["id"]: provider for provider in self._all("providers")} - for provider_id, payload in self.provider_runtime.provider_payloads.items(): - merged = { - **payload, - "usage": existing.get(provider_id, {}).get("usage", payload["usage"]), - } - self._put("providers", provider_id, merged) - self._sync_submission_provider_state() + with self._write_lock: + existing = {provider["id"]: provider for provider in self._all("providers")} + for provider_id, payload in self.provider_runtime.provider_payloads.items(): + prior = existing.get(provider_id) + runtime_enabled = bool(payload.get("enabled", False)) + # `operatorDisabled` distinguishes a deliberate operator/admin + # disable from a connectivity disable. Only the former survives a + # reconnect; a provider that was merely missing credentials is + # re-enabled once the adapter connects (e.g. env change adds keys). + operator_disabled = bool(prior.get("operatorDisabled", False)) if prior else False + if not runtime_enabled: + # Adapter not connected — cannot be enabled regardless. + enabled = False + elif operator_disabled: + # Operator deliberately disabled a connected provider; do not + # let reload/collection silently re-enable it (no audit trail). + enabled = False + else: + enabled = runtime_enabled + merged = { + **payload, + "usage": existing.get(provider_id, {}).get("usage", payload["usage"]), + "enabled": enabled, + "operatorDisabled": operator_disabled, + } + self._put("providers", provider_id, merged) + self._sync_submission_provider_state() def _sync_submission_provider_state(self, queue_id: str | None = None) -> None: - provider_payloads = {provider["id"]: provider for provider in self._all("providers")} - evidence_by_submission = self._evidence_by_submission(queue_id=queue_id) - for submission in self._all("submissions", queue_id=queue_id): - provider_state = submission.get("providerState", {}) - evidence = evidence_by_submission.get(str(submission["id"]), []) - provider_state["internal"] = "ok" - for provider_id in _external_provider_ids(provider_payloads): - provider_state[provider_id] = _external_provider_state_for_submission( - provider_payloads, - provider_id, - submission, - evidence, - ) - submission["providerState"] = provider_state - self._put("submissions", submission["id"], submission) + # Full-table read-modify-write sweep; serialize it so a concurrent + # operator write to an unrelated submission is not reverted by a stale + # copy read at the top of this loop. + with self._write_lock: + provider_payloads = {provider["id"]: provider for provider in self._all("providers")} + evidence_by_submission = self._evidence_by_submission(queue_id=queue_id) + for submission in self._all("submissions", queue_id=queue_id): + provider_state = submission.get("providerState", {}) + evidence = evidence_by_submission.get(str(submission["id"]), []) + provider_state["internal"] = "ok" + for provider_id in _external_provider_ids(provider_payloads): + provider_state[provider_id] = _external_provider_state_for_submission( + provider_payloads, + provider_id, + submission, + evidence, + ) + submission["providerState"] = provider_state + self._put("submissions", submission["id"], submission) def _normalize_saved_google_weak_labels_and_rescore(self, queue_id: str | None = None) -> None: changed_submissions: set[str] = set() @@ -150,6 +176,9 @@ class StoreEnrichmentMixin: try: original = image_store.image_payload(submission_id) except Exception: + # A missing/unreadable image is expected for some rows, but log it so + # a real disk/permission failure is not invisible. + logger.warning("face-crop sync skipped: could not load image for %s", submission_id, exc_info=True) return signal = HeuristicFacePersonDetector().detect(original) @@ -308,7 +337,14 @@ class StoreEnrichmentMixin: item.source in {EvidenceSource.FAILURE, EvidenceSource.EXTERNAL_SKIPPED} for item in domain_evidence ): - google_provider["lastFailure"] = domain_evidence[0].reason if domain_evidence else "Google image search failed" + google_provider["lastFailure"] = next( + ( + item.reason + for item in domain_evidence + if item.source in {EvidenceSource.FAILURE, EvidenceSource.EXTERNAL_SKIPPED} + ), + "Google image search failed", + ) else: google_provider["lastSuccess"] = _now_label() google_provider["lastFailure"] = "없음" @@ -408,7 +444,7 @@ class StoreEnrichmentMixin: return existing_signatures = _existing_google_custom_query_signatures( - self._evidence_by_submission().get(submission_id, []) + self._evidence_for_submission(submission_id) ) can_compare_search_images = self._can_compare_search_result_images( submission_id, @@ -542,7 +578,14 @@ class StoreEnrichmentMixin: item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} for item in all_domain_evidence ): - google_search_provider["lastFailure"] = all_domain_evidence[0].reason if all_domain_evidence else "Google custom search failed" + google_search_provider["lastFailure"] = next( + ( + item.reason + for item in all_domain_evidence + if item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} + ), + "Google custom search failed", + ) else: google_search_provider["lastSuccess"] = _now_label() google_search_provider["lastFailure"] = "없음" @@ -574,7 +617,7 @@ class StoreEnrichmentMixin: return existing_signatures = _existing_naver_query_signatures( - self._evidence_by_submission().get(submission_id, []) + self._evidence_for_submission(submission_id) ) can_compare_search_images = self._can_compare_search_result_images( submission_id, @@ -777,7 +820,14 @@ class StoreEnrichmentMixin: item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} for item in all_domain_evidence ): - naver_provider["lastFailure"] = all_domain_evidence[0].reason if all_domain_evidence else "Naver search failed" + naver_provider["lastFailure"] = next( + ( + item.reason + for item in all_domain_evidence + if item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} + ), + "Naver search failed", + ) else: naver_provider["lastSuccess"] = _now_label() naver_provider["lastFailure"] = "없음" diff --git a/src/rights_filter/server/store_operations.py b/src/rights_filter/server/store_operations.py index cd172c5..a65ebe4 100644 --- a/src/rights_filter/server/store_operations.py +++ b/src/rights_filter/server/store_operations.py @@ -318,47 +318,58 @@ class StoreOperationsMixin: query: str, image_store: LocalSubmissionImageStore | None = None, ) -> dict[str, Any]: - submission = self._get("submissions", submission_id) provider_payload = self._get("providers", provider) if not provider_payload["enabled"]: raise ValueError(f"{provider} provider disabled") + # External search runs OUTSIDE the write lock (it can take seconds); only + # the read-modify-write of the submission is serialized below. domain_evidence, provider_call_count = self._manual_search_evidence( submission_id, provider, query, image_store, ) - for item in domain_evidence: - evidence = _evidence_payload(submission_id, item) - evidence["status"] = "manual" - self._put("evidence", evidence["id"], evidence) - submission.setdefault("queryHistory", []).insert( - 0, - { - "provider": provider, - "query": query, - "status": "manual", - "timestamp": _now_label(), - "count": len(domain_evidence), - }, - ) - self._put("submissions", submission_id, submission) - if any( - item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} - for item in domain_evidence - ): - provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} search failed" - else: - provider_payload["lastSuccess"] = _now_label() - provider_payload["lastFailure"] = "없음" - self._apply_provider_usage_delta(provider, provider_call_count, provider_payload) + with self._write_lock: + for item in domain_evidence: + evidence = _evidence_payload(submission_id, item) + evidence["status"] = "manual" + self._put("evidence", evidence["id"], evidence) + + # Re-read the submission AFTER the search so an operator decision or + # other update made while the search was in flight is not clobbered + # by a stale snapshot. + submission = self._get("submissions", submission_id) + submission.setdefault("queryHistory", []).insert( + 0, + { + "provider": provider, + "query": query, + "status": "manual", + "timestamp": _now_label(), + "count": len(domain_evidence), + }, + ) + self._put("submissions", submission_id, submission) + if any( + item.source in {EvidenceSource.SEARCH_SKIPPED, EvidenceSource.ENRICHMENT_FAILURE} + for item in domain_evidence + ): + provider_payload["lastFailure"] = domain_evidence[0].reason if domain_evidence else f"{provider} search failed" + else: + provider_payload["lastSuccess"] = _now_label() + provider_payload["lastFailure"] = "없음" + self._apply_provider_usage_delta(provider, provider_call_count, provider_payload) + + # LLM narration may call out to Ollama — keep it outside the lock. self._ensure_llm_summary(submission_id) - if image_store is not None: - self._rescore_submission(submission_id) - self._sync_submission_provider_state() - self.add_audit_event("rights.ops", "Provider called", f"{provider} / {submission_id}", f"manual text query: {query}") + + with self._write_lock: + if image_store is not None: + self._rescore_submission(submission_id) + self._sync_submission_provider_state() + self.add_audit_event("rights.ops", "Provider called", f"{provider} / {submission_id}", f"manual text query: {query}") return self.review(submission_id) def _ensure_llm_summaries_for_existing_source_evidence(self, queue_id: str | None = None) -> None: @@ -581,7 +592,11 @@ class StoreOperationsMixin: if provider == "google_search" and self.provider_runtime.google_custom_search_adapter is None: raise ValueError("google_search provider not connected") - self._clear_collection_candidates() + # Run the search cascade OUTSIDE the write lock and accumulate the + # candidates in memory. The clear + repopulate is then committed under the + # lock in one transaction, so it can't interleave with a concurrent + # promote (which would 404 on cleared rows) or a second collect (which + # would resurrect the previous query's rows). if provider == "google_search": domain_evidence = self.provider_runtime.google_custom_search_adapter.search_images( "candidate-collection", @@ -595,9 +610,10 @@ class StoreOperationsMixin: self.provider_runtime.search_policy, ) collected = 0 + pending: list[dict[str, Any]] = [] provider_call_count = 1 if any(item.source != EvidenceSource.SEARCH_SKIPPED for item in domain_evidence) else 0 for candidate in self._collection_candidates_from_evidence(query, domain_evidence, provider): - self._put("collection_candidates", candidate["id"], candidate) + pending.append(candidate) collected += 1 if provider == "google_search" and collected == 0: @@ -610,7 +626,7 @@ class StoreOperationsMixin: provider_call_count += 1 domain_evidence.extend(web_evidence) for candidate in self._collection_candidates_from_evidence(query, web_evidence, provider): - self._put("collection_candidates", candidate["id"], candidate) + pending.append(candidate) collected += 1 if provider == "naver" and collected == 0: page_evidence = self.provider_runtime.naver_adapter.search_pages( @@ -622,7 +638,7 @@ class StoreOperationsMixin: provider_call_count += 1 domain_evidence.extend(page_evidence) for candidate in self._collection_candidates_from_evidence(query, page_evidence, provider): - self._put("collection_candidates", candidate["id"], candidate) + pending.append(candidate) collected += 1 if provider == "naver" and collected == 0: web_evidence = self.provider_runtime.naver_adapter.search_web_pages( @@ -634,7 +650,7 @@ class StoreOperationsMixin: provider_call_count += 1 domain_evidence.extend(web_evidence) for candidate in self._collection_candidates_from_evidence(query, web_evidence, provider): - self._put("collection_candidates", candidate["id"], candidate) + pending.append(candidate) collected += 1 if any( @@ -645,8 +661,20 @@ class StoreOperationsMixin: else: provider_payload["lastSuccess"] = _now_label() provider_payload["lastFailure"] = "없음" - self._apply_provider_usage_delta(provider, provider_call_count, provider_payload) - self.add_audit_event("rights.ops", "Keyword candidates collected", provider, f"{query} · {collected} candidates") + + with self._write_lock: + with self._transaction() as conn: + conn.execute("delete from collection_candidates") + for candidate in pending: + self._put("collection_candidates", candidate["id"], candidate, conn=conn) + self.add_audit_event( + "rights.ops", + "Keyword candidates collected", + provider, + f"{query} · {collected} candidates", + conn=conn, + ) + self._apply_provider_usage_delta(provider, provider_call_count, provider_payload) payload = self.bootstrap() payload["collected"] = collected return payload @@ -691,6 +719,31 @@ class StoreOperationsMixin: ) return self.bootstrap() + def dismiss_collection_candidate(self, candidate_id: str) -> dict[str, Any]: + # Persist the dismissal so the candidate stays gone across refreshes + # instead of resurrecting on the next bootstrap. Idempotent: a repeat / + # double-click on an already-removed candidate is a no-op. + with self._write_lock: + try: + candidate = self._get("collection_candidates", candidate_id) + except KeyError: + return self.bootstrap() + if candidate.get("status") == "promoted": + raise ValueError("promoted candidate cannot be dismissed") + query = str(candidate.get("query", "")) + with self._transaction() as conn: + conn.execute( + "delete from collection_candidates where id = ?", (candidate_id,) + ) + self.add_audit_event( + "rights.ops", + "Keyword candidate dismissed", + candidate_id, + f"dismissed candidate for query: {query}", + conn=conn, + ) + return self.bootstrap() + def promote_collection_candidates(self, payload: dict[str, Any]) -> dict[str, Any]: candidate_ids = _unique_texts(_text_list(payload.get("candidate_ids", payload.get("candidateIds")))) if not candidate_ids: diff --git a/src/rights_filter/server/store_persistence.py b/src/rights_filter/server/store_persistence.py index 03b3b93..73bf8e6 100644 --- a/src/rights_filter/server/store_persistence.py +++ b/src/rights_filter/server/store_persistence.py @@ -231,18 +231,19 @@ class StorePersistenceMixin: def _apply_provider_usage_delta( self, provider_id: str, count: int, payload: dict[str, Any] ) -> None: - # Re-read current usage under the write lock and apply the delta so - # concurrent provider calls don't lose increments. The rest of `payload` - # (lastSuccess / lastFailure) is last-writer-wins, which is acceptable for - # status fields. + # Start from the freshly-read row and merge only the fields this call + # actually produced (usage delta + call status). The caller's `payload` + # was snapshotted before a multi-second external search, so writing it + # wholesale would revert any operator change made meanwhile — in + # particular an emergency disable (`enabled`). Merging keeps that change. with self._write_lock: current = self._get("providers", provider_id) - payload["usage"] = int(current.get("usage", 0) or 0) + int(count) - self._put("providers", provider_id, payload) - - def _clear_collection_candidates(self) -> None: - with self._connect() as conn: - conn.execute("delete from collection_candidates") + merged = dict(current) + for key in ("lastSuccess", "lastFailure", "lastRun"): + if key in payload: + merged[key] = payload[key] + merged["usage"] = int(current.get("usage", 0) or 0) + int(count) + self._put("providers", provider_id, merged) def _all(self, table: str, queue_id: str | None = None) -> list[dict[str, Any]]: _validate_table(table) diff --git a/src/rights_filter/server/store_queue.py b/src/rights_filter/server/store_queue.py index e4de0d6..c40958b 100644 --- a/src/rights_filter/server/store_queue.py +++ b/src/rights_filter/server/store_queue.py @@ -109,14 +109,31 @@ class StoreQueueMixin: """, (queue_id, str(folder), queue_label, now.isoformat(" ", "seconds"), int(now.timestamp())), ) + effective_id = queue_id else: + # A row matched by folder_path may carry a different (older) id + # than the freshly-computed queue_id — keep the existing row's id + # as the effective key so the UPDATE, the deactivate-others sweep, + # and the final SELECT all target the real row instead of silently + # matching zero rows and returning None. + effective_id = existing["id"] conn.execute( "update submission_queues set is_active = 1, folder_path = ?, label = ? where id = ?", - (str(folder), queue_label, queue_id), + (str(folder), queue_label, effective_id), + ) + conn.execute("update submission_queues set is_active = 0 where id != ?", (effective_id,)) + migrated_legacy = self._has_queueless_submissions(conn=conn) + if migrated_legacy: + self._migrate_legacy_submissions_to_queue(effective_id, conn=conn) + # Audit only meaningful transitions (a brand-new queue or a legacy + # migration), not every reload, so the audit trail isn't spammed. + if existing is None or migrated_legacy: + change = f"folder {folder}" + ( + "; legacy queueless submissions migrated" if migrated_legacy else "; new queue" + ) + self.add_audit_event( + "rights.system", "Submission queue activated", effective_id, change, conn=conn ) - conn.execute("update submission_queues set is_active = 0 where id != ?", (queue_id,)) - if self._has_queueless_submissions(conn=conn): - self._migrate_legacy_submissions_to_queue(queue_id, conn=conn) queue_row = conn.execute( """ select id, folder_path, label, is_active, created_at, created_epoch, @@ -124,10 +141,10 @@ class StoreQueueMixin: from submission_queues where id = ? """, - (queue_id,), + (effective_id,), ).fetchone() if queue_row is None: - return self._queue_row_by_id(queue_id) + return self._queue_row_by_id(effective_id) return { "id": queue_row["id"], "folderPath": queue_row["folder_path"], diff --git a/tests/rights_filter/server/test_sqlite_store.py b/tests/rights_filter/server/test_sqlite_store.py index 31d4e91..902781c 100644 --- a/tests/rights_filter/server/test_sqlite_store.py +++ b/tests/rights_filter/server/test_sqlite_store.py @@ -1228,6 +1228,150 @@ def test_sqlite_store_emergency_disable_marks_google_custom_search_state_disable assert submission["providerState"]["llm"] == "disabled" +_CONNECTED_PROVIDER_ENV = { + "NAVER_CLIENT_ID": "naver-id", + "NAVER_CLIENT_SECRET": "naver-secret", + "GOOGLE_CLOUD_VISION_API_KEY": "google-key", + "GOOGLE_CUSTOM_SEARCH_API_KEY": "google-search-key", + "GOOGLE_CUSTOM_SEARCH_CX": "search-engine-id", + "OLLAMA_MODEL": "qwen2.5:0.5b-instruct", +} + + +def test_ensure_queue_handles_folder_match_with_mismatched_id(tmp_path: Path): + # 2026-07-11 #13: a stored queue row whose id differs from the freshly + # computed queue_id (e.g. after a path-normalization change) must still be + # found by folder_path and updated in place — the old code keyed the UPDATE + # and final SELECT on the computed id, matched zero rows, and returned None + # (which then crashed seed_from_image_store on str(queue["id"])). + store = CopyrighterStore( + tmp_path / "copyrighter.sqlite3", + provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})), + ) + store.initialize() + folder = (tmp_path / "queue-folder").resolve() + folder.mkdir() + with store._connect() as conn: + conn.execute( + """ + insert into submission_queues + (id, folder_path, label, is_active, created_at, created_epoch, last_imported_epoch, last_imported_at) + values (?, ?, ?, 1, '', 0, 0, '') + """, + ("queue-LEGACYID", str(folder), "legacy"), + ) + conn.commit() + + result = store.ensure_queue(folder) + assert result is not None + assert result["id"] == "queue-LEGACYID" + assert result["isActive"] is True + + +def test_dismiss_collection_candidate_persists_and_is_idempotent(tmp_path: Path): + # 2026-07-11 H9: dismissing a candidate must persist (delete the row) so it + # does not resurrect on the next bootstrap; a repeat dismiss is a no-op. + store = CopyrighterStore( + tmp_path / "copyrighter.sqlite3", + provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})), + ) + store.initialize() + candidate = { + "id": "cand-1", + "provider": "naver", + "query": "IU", + "title": "IU", + "status": "candidate", + "sourceUrl": "https://example.test/iu", + "collectedEpoch": 0, + "sampleFingerprints": [], + } + store._put("collection_candidates", "cand-1", candidate) + assert any(c["id"] == "cand-1" for c in store.bootstrap()["collectionCandidates"]) + + store.dismiss_collection_candidate("cand-1") + assert all(c["id"] != "cand-1" for c in store.bootstrap()["collectionCandidates"]) + + store.dismiss_collection_candidate("cand-1") # idempotent no-op + + +def test_seed_survives_unanalyzable_image(tmp_path: Path): + # 2026-07-11 H5: a single zero-byte/corrupt image must not abort the whole + # seed with IndexError (which left a partial commit that re-failed on every + # reload). It becomes a 'failed' submission; the good ones still seed. + root = tmp_path / "store" + image_dir = root / "images" + image_dir.mkdir(parents=True) + (image_dir / "good.svg").write_text("", encoding="utf-8") + (image_dir / "bad.png").write_bytes(b"") + (root / "submissions.json").write_text( + json.dumps( + [ + {"id": "SUB-GOOD", "title": "good", "file": "images/good.svg", "width": 800, "height": 600, "submitted_at": "2026-07-11 10:00"}, + {"id": "SUB-BAD", "title": "bad", "file": "images/bad.png", "width": 800, "height": 600, "submitted_at": "2026-07-11 10:00"}, + ] + ), + encoding="utf-8", + ) + image_store = LocalSubmissionImageStore(root) + store = CopyrighterStore( + tmp_path / "copyrighter.sqlite3", + provider_runtime=build_provider_runtime(env={}, transport=FakeJsonTransport({})), + ) + store.initialize() + store.seed_from_image_store(image_store) # must not raise + + subs = {s["id"]: s for s in store.bootstrap()["submissions"]} + assert "SUB-GOOD" in subs + assert subs["SUB-BAD"]["riskBand"] == "failed" + + +def test_operator_disabled_provider_survives_reseed(tmp_path: Path): + # 2026-07-11 H3: a provider an operator deliberately disabled must not be + # silently re-enabled by the next reload/seed just because the adapter is + # connected. A provider disabled only for connectivity still reconnects. + runtime = build_provider_runtime( + env=_CONNECTED_PROVIDER_ENV, transport=FakeJsonTransport({"items": []}) + ) + store = CopyrighterStore(tmp_path / "copyrighter.sqlite3", provider_runtime=runtime) + store.initialize() + image_store = _image_store(tmp_path / "submissions") + store.seed_from_image_store(image_store) + + store.set_provider_enabled("naver", False) + store.seed_from_image_store(image_store) # reload + + naver = next(p for p in store.providers() if p["id"] == "naver") + assert naver["enabled"] is False + google = next(p for p in store.providers() if p["id"] == "google") + assert google["enabled"] is True + + +def test_provider_usage_delta_does_not_revert_inflight_disable(tmp_path: Path): + # 2026-07-11 H2: _apply_provider_usage_delta must merge only usage/status onto + # the fresh row. If it rewrote a stale pre-search snapshot, an emergency + # disable that landed while the search was in flight would be reverted. + runtime = build_provider_runtime( + env=_CONNECTED_PROVIDER_ENV, transport=FakeJsonTransport({"items": []}) + ) + store = CopyrighterStore(tmp_path / "copyrighter.sqlite3", provider_runtime=runtime) + store.initialize() + store.seed_from_image_store(_image_store(tmp_path / "submissions")) + + stale = next(p for p in store.providers() if p["id"] == "naver") + assert stale["enabled"] is True + + store.set_provider_enabled("naver", False) # operator disables mid-search + + stale["lastSuccess"] = "2026-07-11 00:00:00" + store._apply_provider_usage_delta("naver", 1, stale) # search finishes + + naver = next(p for p in store.providers() if p["id"] == "naver") + assert naver["enabled"] is False + assert naver["usage"] >= 1 + assert naver["lastSuccess"] == "2026-07-11 00:00:00" + + def test_sqlite_store_manual_naver_search_uses_real_env_client(tmp_path: Path): transport = FakeJsonTransport( {