diff --git a/backend/app/database/self_writes.py b/backend/app/database/self_writes.py new file mode 100644 index 000000000..f1bfd8101 --- /dev/null +++ b/backend/app/database/self_writes.py @@ -0,0 +1,154 @@ +""" +A short-lived record of files PictoPy modified itself. + +The sync microservice watches every registered folder and treats any file change +as a user edit, so a write of our own comes straight back as a full folder +resync. Recording what we wrote lets the watcher tell the two apart. + +The path format here is a contract with `sync-microservice/app/database/ +self_writes.py`, which reads this table: both sides key on +`os.path.normcase(os.path.abspath(path))`. +""" + +import os +import sqlite3 +import time +from typing import List, Set, Tuple + +from app.config.settings import DATABASE_PATH +from app.logging.setup_logging import get_logger + +logger = get_logger(__name__) + +# An entry the watcher never claims is one it missed -- it was stopped, or the +# change was coalesced away. Expiring them keeps the table bounded and limits +# how long a stale row can mask a real edit to the same path. +SELF_WRITE_TTL_SECONDS = 3600 + +# (path, file_size, file_mtime) as observed on disk right now. +ObservedFile = Tuple[str, int, int] + + +def self_write_key(path: str) -> str: + """Normalise a path to the form both services store and look up by.""" + return os.path.normcase(os.path.abspath(path)) + + +def _connect() -> sqlite3.Connection: + conn = sqlite3.connect(DATABASE_PATH) + # This table stands alone, keyed by path, so nothing here depends on it + # today. Set anyway to match every other _connect() in this package. + conn.execute("PRAGMA foreign_keys = ON") + return conn + + +def db_create_self_writes_table() -> None: + conn = _connect() + cursor = conn.cursor() + try: + cursor.execute( + """ + CREATE TABLE IF NOT EXISTS self_writes ( + path TEXT PRIMARY KEY, + file_size INTEGER NOT NULL, + file_mtime INTEGER NOT NULL, + -- Epoch seconds rather than CURRENT_TIMESTAMP: this column only + -- exists to be subtracted from, and text timestamps make that + -- arithmetic a timezone question. + written_at INTEGER NOT NULL + ) + """ + ) + conn.commit() + finally: + conn.close() + + +def db_record_self_write(path: str, file_size: int, file_mtime: int) -> bool: + """ + Note that PictoPy is about to leave a file in this exact state. + + Callers record before the bytes land, because the watcher can fire the + instant they do. + """ + conn = _connect() + cursor = conn.cursor() + now = int(time.time()) + + try: + cursor.execute( + """ + INSERT INTO self_writes (path, file_size, file_mtime, written_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + file_size=excluded.file_size, + file_mtime=excluded.file_mtime, + written_at=excluded.written_at + """, + (self_write_key(path), file_size, file_mtime, now), + ) + # Pruning here avoids needing a scheduler for a table this small. + cursor.execute( + "DELETE FROM self_writes WHERE written_at < ?", + (now - SELF_WRITE_TTL_SECONDS,), + ) + conn.commit() + return True + except sqlite3.Error as e: + logger.error(f"Error recording self write for {path}: {e}") + conn.rollback() + return False + finally: + conn.close() + + +def db_take_matching_self_writes(observed: List[ObservedFile]) -> Set[str]: + """ + Return the observed paths that match a recorded write, and forget them. + + Claiming an entry as it matches means a second event for the same write is + treated as a real change. That is the safe direction to be wrong in: it + costs one redundant rescan, where holding the entry could swallow a genuine + edit that happened to land in the same second at the same size. + """ + if not observed: + return set() + + conn = _connect() + cursor = conn.cursor() + matched: Set[str] = set() + + try: + by_key = {self_write_key(path): path for path, _, _ in observed} + placeholders = ",".join("?" for _ in observed) + cursor.execute( + f""" + SELECT path, file_size, file_mtime + FROM self_writes + WHERE path IN ({placeholders}) + """, + list(by_key), + ) + recorded = {row[0]: (row[1], row[2]) for row in cursor.fetchall()} + + claimed = [] + for path, size, mtime in observed: + key = self_write_key(path) + if recorded.get(key) == (size, mtime): + matched.add(path) + claimed.append(key) + + if claimed: + cursor.executemany( + "DELETE FROM self_writes WHERE path = ?", + [(key,) for key in claimed], + ) + conn.commit() + + return matched + except sqlite3.Error as e: + # Failing open means a redundant rescan, never a missed change. + logger.error(f"Error matching self writes: {e}") + return set() + finally: + conn.close() diff --git a/backend/app/utils/self_write.py b/backend/app/utils/self_write.py new file mode 100644 index 000000000..33dc25f4d --- /dev/null +++ b/backend/app/utils/self_write.py @@ -0,0 +1,70 @@ +""" +Replacing a file in a watched folder without the change reading as a user edit. + +Anything that rewrites a file inside a registered folder should go through here +rather than opening the path directly. +""" + +import os +import tempfile +from typing import Optional + +from app.database.self_writes import db_record_self_write +from app.logging.setup_logging import get_logger + +logger = get_logger(__name__) + +# The temp file lands in the watched folder too, so the watcher recognises and +# ignores it by name. Mirrored in sync-microservice/app/utils/watcher.py. +SELF_WRITE_TEMP_PREFIX = ".pictopy-write-" + + +def _discard(temp_path: Optional[str]) -> None: + """Best-effort cleanup; a leftover temp is noise, not a failure worth raising.""" + if not temp_path: + return + try: + os.unlink(temp_path) + except OSError: + logger.warning(f"Could not remove temporary file {temp_path}") + + +def self_write_util_replace(path: str, data: bytes) -> bool: + """ + Atomically replace a file's contents and record the write for the watcher. + + Writes to a sibling temp file, records the size and mtime that file already + has, then renames it into place. The rename carries both across unchanged, + so the ledger row is in the database before the new bytes are visible at the + watched path -- which matters, because the watcher can fire the moment they + are. + + Returns False if the file could not be replaced, leaving the original as is. + """ + target = os.path.abspath(path) + directory = os.path.dirname(target) + temp_path: Optional[str] = None + + try: + # Same directory, so the rename stays on one filesystem and stays atomic. + handle_fd, temp_path = tempfile.mkstemp( + dir=directory, prefix=SELF_WRITE_TEMP_PREFIX, suffix=".tmp" + ) + with os.fdopen(handle_fd, "wb") as handle: + handle.write(data) + handle.flush() + # Without this the rename can land before the bytes do, which on a + # crash leaves a photo that is neither the old one nor the new one. + os.fsync(handle.fileno()) + + stats = os.stat(temp_path) + db_record_self_write(target, stats.st_size, int(stats.st_mtime)) + + os.replace(temp_path, target) + return True + except OSError as e: + # A locked or read-only file is ordinary on Windows; the caller retries + # on a later pass rather than treating it as a failed run. + logger.error(f"Could not replace {path}: {e}") + _discard(temp_path) + return False diff --git a/backend/main.py b/backend/main.py index 071bb79cd..366be51d0 100644 --- a/backend/main.py +++ b/backend/main.py @@ -25,6 +25,7 @@ db_clear_stale_processing_flags, ) from app.database.metadata import db_create_metadata_table +from app.database.self_writes import db_create_self_writes_table from app.database.semantic_labels import db_create_semantic_labels_table from app.database.image_embeddings import db_create_image_embeddings_table from app.database.video_frames import db_create_video_frames_tables @@ -83,6 +84,7 @@ async def lifespan(app: FastAPI): db_create_album_images_table() db_create_metadata_table() db_create_memories_table() # References images(id) and videos(id) + db_create_self_writes_table() # Standalone: keyed by path, no foreign keys # Nothing is indexing or tagging yet, so anything still flagged busy is # left over from a previous session and would block memory generation. db_clear_stale_processing_flags() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index ccf14b4ae..61d071547 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -10,6 +10,7 @@ from app.database.albums import db_create_albums_table, db_create_album_images_table from app.database.folders import db_create_folders_table from app.database.metadata import db_create_metadata_table +from app.database.self_writes import db_create_self_writes_table from app.database.semantic_labels import db_create_semantic_labels_table from app.database.image_embeddings import db_create_image_embeddings_table from app.database.video_frames import db_create_video_frames_tables @@ -39,6 +40,7 @@ def setup_before_all_tests(): db_create_video_frames_tables() db_create_metadata_table() db_create_memories_table() # References images(id) and videos(id) + db_create_self_writes_table() # Standalone: keyed by path, no foreign keys print("All database tables created successfully") except Exception as e: print(f"Error creating database tables: {e}") diff --git a/backend/tests/test_self_write.py b/backend/tests/test_self_write.py new file mode 100644 index 000000000..a3242f6bd --- /dev/null +++ b/backend/tests/test_self_write.py @@ -0,0 +1,216 @@ +""" +The watcher treats any file change as a user edit, so a write of PictoPy's own +would come back as a folder resync. These cover the ledger that tells them +apart, and the atomic replace that populates it. +""" + +import os +import sqlite3 +import tempfile +import time +from typing import Iterator + +import pytest + +from app.database.self_writes import ( + SELF_WRITE_TTL_SECONDS, + db_create_self_writes_table, + db_record_self_write, + db_take_matching_self_writes, + self_write_key, +) +from app.utils.self_write import SELF_WRITE_TEMP_PREFIX, self_write_util_replace + + +@pytest.fixture(scope="function") +def test_db(monkeypatch: pytest.MonkeyPatch) -> Iterator[str]: + db_fd, db_path = tempfile.mkstemp() + os.close(db_fd) + + monkeypatch.setattr("app.config.settings.DATABASE_PATH", db_path) + monkeypatch.setattr("app.database.self_writes.DATABASE_PATH", db_path) + db_create_self_writes_table() + + yield db_path + + os.unlink(db_path) + + +def _observe(path: str): + stats = os.stat(path) + return (path, stats.st_size, int(stats.st_mtime)) + + +class TestLedger: + def test_a_recorded_write_is_claimed(self, test_db, tmp_path): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + path, size, mtime = _observe(str(photo)) + + db_record_self_write(path, size, mtime) + + assert db_take_matching_self_writes([(path, size, mtime)]) == {path} + + def test_an_unrecorded_file_is_left_alone(self, test_db, tmp_path): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + assert db_take_matching_self_writes([_observe(str(photo))]) == set() + + def test_a_later_edit_to_the_same_path_is_not_claimed(self, test_db, tmp_path): + """The user changing the file we wrote is a real change, not our echo.""" + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + path, size, mtime = _observe(str(photo)) + db_record_self_write(path, size, mtime) + + photo.write_bytes(b"y" * 4096) + + assert db_take_matching_self_writes([_observe(str(photo))]) == set() + + def test_an_entry_is_claimed_only_once(self, test_db, tmp_path): + """A second event for one write must read as a real change.""" + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + observed = _observe(str(photo)) + db_record_self_write(*observed) + + assert db_take_matching_self_writes([observed]) == {observed[0]} + assert db_take_matching_self_writes([observed]) == set() + + def test_claiming_one_path_leaves_the_others(self, test_db, tmp_path): + ours = tmp_path / "ours.jpg" + theirs = tmp_path / "theirs.jpg" + ours.write_bytes(b"x" * 100) + theirs.write_bytes(b"y" * 100) + db_record_self_write(*_observe(str(ours))) + + matched = db_take_matching_self_writes( + [_observe(str(ours)), _observe(str(theirs))] + ) + + assert matched == {str(ours)} + + def test_lookup_survives_a_differently_spelled_path( + self, test_db, tmp_path, monkeypatch + ): + """ + The watcher reports whatever the OS hands it, so the two sides agree on a + normalised key rather than on the exact string. Case folding is part of + that on Windows but not on Linux, where paths really are case-sensitive; + absolute-vs-relative is the part that holds everywhere. + """ + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + _, size, mtime = _observe(str(photo)) + db_record_self_write(str(photo), size, mtime) + + monkeypatch.chdir(tmp_path) + + assert db_take_matching_self_writes([("a.jpg", size, mtime)]) == {"a.jpg"} + + def test_nothing_observed_queries_nothing(self, test_db): + assert db_take_matching_self_writes([]) == set() + + def test_stale_entries_are_pruned_on_the_next_write(self, test_db, tmp_path): + """An entry the watcher never claimed was one it missed; it must not linger.""" + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + path, size, mtime = _observe(str(photo)) + db_record_self_write(path, size, mtime) + + conn = sqlite3.connect(test_db) + conn.execute( + "UPDATE self_writes SET written_at = ?", + (int(time.time()) - SELF_WRITE_TTL_SECONDS - 60,), + ) + conn.commit() + conn.close() + + db_record_self_write(str(tmp_path / "unrelated.jpg"), 1, 1) + + assert db_take_matching_self_writes([(path, size, mtime)]) == set() + + def test_an_unreadable_ledger_suppresses_nothing(self, test_db, tmp_path): + """Failing open costs a rescan; failing closed would drop a real change.""" + photo = tmp_path / "a.jpg" + photo.write_bytes(b"x" * 100) + observed = _observe(str(photo)) + db_record_self_write(*observed) + + conn = sqlite3.connect(test_db) + conn.execute("DROP TABLE self_writes") + conn.commit() + conn.close() + + assert db_take_matching_self_writes([observed]) == set() + + +class TestAtomicReplace: + def test_the_new_bytes_land(self, test_db, tmp_path): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"original") + + assert self_write_util_replace(str(photo), b"replaced") is True + assert photo.read_bytes() == b"replaced" + + def test_the_write_is_recorded_so_the_watcher_ignores_it(self, test_db, tmp_path): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"original") + + self_write_util_replace(str(photo), b"replaced") + + assert db_take_matching_self_writes([_observe(str(photo))]) == {str(photo)} + + def test_the_ledger_is_written_before_the_bytes_are_visible( + self, test_db, tmp_path, monkeypatch + ): + """ + The watcher can fire the instant the rename lands, so recording after it + would leave a window where our own write reads as a user edit. + """ + photo = tmp_path / "a.jpg" + photo.write_bytes(b"original") + + def explode(src, dst): + raise OSError("rename interrupted") + + monkeypatch.setattr("app.utils.self_write.os.replace", explode) + + assert self_write_util_replace(str(photo), b"replaced") is False + + conn = sqlite3.connect(test_db) + rows = conn.execute("SELECT path FROM self_writes").fetchall() + conn.close() + assert rows == [(self_write_key(str(photo)),)] + + def test_a_failed_replace_leaves_the_original_and_no_scratch_file( + self, test_db, tmp_path, monkeypatch + ): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"original") + + def explode(src, dst): + raise OSError("rename interrupted") + + monkeypatch.setattr("app.utils.self_write.os.replace", explode) + self_write_util_replace(str(photo), b"replaced") + + assert photo.read_bytes() == b"original" + assert [p.name for p in tmp_path.iterdir()] == ["a.jpg"] + + def test_an_unwritable_location_reports_failure(self, test_db, tmp_path): + missing = tmp_path / "no-such-dir" / "a.jpg" + assert self_write_util_replace(str(missing), b"data") is False + + def test_success_leaves_no_scratch_file_behind(self, test_db, tmp_path): + photo = tmp_path / "a.jpg" + photo.write_bytes(b"original") + + self_write_util_replace(str(photo), b"replaced") + + leftovers = [ + p.name + for p in tmp_path.iterdir() + if p.name.startswith(SELF_WRITE_TEMP_PREFIX) + ] + assert leftovers == [] diff --git a/sync-microservice/app/database/self_writes.py b/sync-microservice/app/database/self_writes.py new file mode 100644 index 000000000..4e74e1704 --- /dev/null +++ b/sync-microservice/app/database/self_writes.py @@ -0,0 +1,75 @@ +""" +Reads the self-write ledger the primary backend maintains. + +The table is created and populated by `backend/app/database/self_writes.py`; +this service only claims entries. The path format is a contract with that +module: both sides key on `os.path.normcase(os.path.abspath(path))`. +""" + +import os +import sqlite3 +from typing import List, Set, Tuple + +from app.config.settings import DATABASE_PATH +from app.logging.setup_logging import get_sync_logger + +logger = get_sync_logger(__name__) + +# (path, file_size, file_mtime) as observed on disk right now. +ObservedFile = Tuple[str, int, int] + + +def _self_write_key(path: str) -> str: + return os.path.normcase(os.path.abspath(path)) + + +def db_take_matching_self_writes(observed: List[ObservedFile]) -> Set[str]: + """ + Return the observed paths PictoPy wrote itself, and claim them. + + Claiming as we match means a second event for one write reads as a real + change. That is the safe direction: it costs a redundant rescan, where + keeping the entry could swallow a genuine edit to the same path. + """ + if not observed: + return set() + + conn = None + matched: Set[str] = set() + + try: + conn = sqlite3.connect(DATABASE_PATH) + cursor = conn.cursor() + + keys = {_self_write_key(path) for path, _, _ in observed} + placeholders = ",".join("?" for _ in keys) + cursor.execute( + f""" + SELECT path, file_size, file_mtime + FROM self_writes + WHERE path IN ({placeholders}) + """, + list(keys), + ) + recorded = {row[0]: (row[1], row[2]) for row in cursor.fetchall()} + + claimed = [] + for path, size, mtime in observed: + key = _self_write_key(path) + if recorded.get(key) == (size, mtime): + matched.add(path) + claimed.append((key,)) + + if claimed: + cursor.executemany("DELETE FROM self_writes WHERE path = ?", claimed) + conn.commit() + + return matched + except sqlite3.Error as e: + # Failing open costs a redundant rescan; failing closed would drop a + # real change, so an unreadable ledger must never suppress anything. + logger.error(f"Error matching self writes: {e}") + return set() + finally: + if conn is not None: + conn.close() diff --git a/sync-microservice/app/utils/watcher.py b/sync-microservice/app/utils/watcher.py index 902a50cf5..6117efd54 100644 --- a/sync-microservice/app/utils/watcher.py +++ b/sync-microservice/app/utils/watcher.py @@ -2,10 +2,11 @@ import threading import time import logging -from typing import List, Tuple, Dict, Optional +from typing import List, Set, Tuple, Dict, Optional from watchfiles import watch, Change import httpx from app.database.folders import db_get_all_folders_with_ids +from app.database.self_writes import db_take_matching_self_writes from app.config.settings import PRIMARY_BACKEND_URL from app.logging.setup_logging import get_sync_logger @@ -21,6 +22,10 @@ FolderIdPath = Tuple[str, str] +# Mirrors backend/app/utils/self_write.py, which names its temp files this way +# so they can be recognised here. +SELF_WRITE_TEMP_PREFIX = ".pictopy-write-" + # Global variables to track watcher state watcher_thread: Optional[threading.Thread] = None stop_event = threading.Event() @@ -49,6 +54,36 @@ def watcher_util_get_folder_id_if_watched(file_path: str) -> Optional[str]: return None +def watcher_util_is_own_temp_file(file_path: str) -> bool: + """ + True for the scratch file a PictoPy write leaves in the watched folder. + + It has to be created beside its target for the rename to stay atomic, so the + watcher sees it appear and disappear. Prefix mirrors + backend/app/utils/self_write.py. + """ + return os.path.basename(file_path).startswith(SELF_WRITE_TEMP_PREFIX) + + +def watcher_util_drop_self_writes(file_paths: List[str]) -> Set[str]: + """ + Of the given paths, those matching a write PictoPy just made itself. + + Batched into one query because a metadata sync pass touches many files at + once. + """ + observed = [] + for file_path in file_paths: + try: + stats = os.stat(file_path) + except OSError: + # Gone or unreadable: not something we can claim as our own write. + continue + observed.append((file_path, stats.st_size, int(stats.st_mtime))) + + return db_take_matching_self_writes(observed) + + def watcher_util_handle_file_changes(changes: set) -> None: """ Handle file changes detected by watchfiles. @@ -60,14 +95,30 @@ def watcher_util_handle_file_changes(changes: set) -> None: affected_folders = {} # folder_path -> folder_id mapping + # PictoPy writing metadata into a photo is a file modification like any + # other, and resyncing the folder it lives in would undo the write's whole + # point. Both passes below run before anything is attributed to a folder. + surviving = [ + (change, file_path) + for change, file_path in changes + if not watcher_util_is_own_temp_file(file_path) + ] + self_written = watcher_util_drop_self_writes( + [file_path for change, file_path in surviving if change != Change.deleted] + ) + if self_written: + logger.debug(f"Ignoring {len(self_written)} change(s) PictoPy made itself") + # First pass - count changes and identify affected folders - for change, file_path in changes: + for change, file_path in surviving: # Process deletions if change == Change.deleted: deleted_folder_id = watcher_util_get_folder_id_if_watched(file_path) if deleted_folder_id: deleted_folder_ids.append(deleted_folder_id) continue + elif file_path in self_written: + continue # Find affected folder closest_folder = watcher_util_find_closest_parent_folder(