diff --git a/dikte/config.py b/dikte/config.py index 7229553..24b7a20 100644 --- a/dikte/config.py +++ b/dikte/config.py @@ -5,6 +5,8 @@ import hashlib import json import os import sys +import threading +import time from . import api from . import ggml @@ -525,6 +527,30 @@ TRANSCRIBERS = { "openrouter_base_url", "openrouter_transcribe_model"), } +# One lock for the history file and the meeting index both, rather than one +# each: the files are a few kilobytes, the writes happen a handful of times an +# hour, and a second lock would only add a way to take them in the wrong order. +_FILES_LOCK = threading.Lock() + + +def _replace_with_retry(tmp, target): + """The atomic swap, tried again briefly when the target is held. + + On Windows an antivirus or sync tool opens a freshly written file to look + at it, and a rename over the file fails for as long as it is held. The + hold lasts milliseconds, so three tries with a short sleep cover it; a + file held longer than that is a real error and is raised as one. + """ + for attempt in range(3): + try: + tmp.replace(target) + return + except OSError: + if attempt == 2: + raise + time.sleep(0.05) + + # Corners used to be stored with Turkish names. _CORNER_MIGRATION = { "sol-alt": "bottom-left", "sağ-alt": "bottom-right", @@ -545,7 +571,19 @@ class Config: self.data.update({k: v for k, v in stored.items() if k in DEFAULTS}) except FileNotFoundError: pass - except (json.JSONDecodeError, OSError) as exc: + except json.JSONDecodeError as exc: + # Set aside rather than left in place: the next save would write + # the defaults over it, and whatever broke the file deserves to + # still be there to look at. Best effort; a rename that fails + # changes nothing about falling back to the defaults. + broken = CONFIG_FILE.with_suffix(".json.broken") + try: + CONFIG_FILE.replace(broken) + except OSError: + pass + print(f"dikte: could not read settings ({exc}), using defaults; " + f"the unreadable file was kept as {broken}") + except OSError as exc: print(f"dikte: could not read settings ({exc}), using defaults") self.data["overlay_corner"] = _CORNER_MIGRATION.get( self.data["overlay_corner"], self.data["overlay_corner"] @@ -560,8 +598,13 @@ class Config: tmp = CONFIG_FILE.with_suffix(".json.tmp") with open(tmp, "w", encoding="utf-8") as fh: json.dump(self.data, fh, ensure_ascii=False, indent=2) + # Pushed to the disk before the rename: swapping in a file that + # still lives in the page cache turns a power cut into a settings + # wipe, which the atomic replace exists to prevent. + fh.flush() + os.fsync(fh.fileno()) os.chmod(tmp, 0o600) - tmp.replace(CONFIG_FILE) + _replace_with_retry(tmp, CONFIG_FILE) i18n.set_language(self.data["ui_language"]) def __getitem__(self, key): @@ -727,12 +770,22 @@ def default_assistant_prompt(): def append_history(entry): DATA_DIR.mkdir(parents=True, exist_ok=True) - with open(HISTORY_FILE, "a", encoding="utf-8") as fh: - fh.write(json.dumps(entry, ensure_ascii=False) + "\n") + with _FILES_LOCK: + with open(HISTORY_FILE, "a", encoding="utf-8") as fh: + fh.write(json.dumps(entry, ensure_ascii=False) + "\n") def read_history(limit=None): """Newest last. A limit of None (or 0) reads the whole file.""" + # Locked even though the rewrites are atomic: it costs nothing, and a read + # that waits out a rewrite in flight hands back the settled file rather + # than whichever side of the swap it happened to land on. + with _FILES_LOCK: + return _read_history(limit) + + +def _read_history(limit=None): + """The body of read_history, for callers already holding the lock.""" try: with open(HISTORY_FILE, encoding="utf-8") as fh: lines = fh.readlines() @@ -755,40 +808,66 @@ def _write_history(lines): tmp = HISTORY_FILE.with_suffix(".jsonl.tmp") with open(tmp, "w", encoding="utf-8") as fh: fh.writelines(lines) - tmp.replace(HISTORY_FILE) + fh.flush() + os.fsync(fh.fileno()) + _replace_with_retry(tmp, HISTORY_FILE) def trim_history(limit): """Drop the oldest entries once the file passes `limit` rows. 0 means keep all.""" if not limit or limit < 0: return - try: - with open(HISTORY_FILE, encoding="utf-8") as fh: - lines = fh.readlines() - except OSError: - return - if len(lines) <= limit: - return - _write_history(lines[-limit:]) + # Read and rewrite under one lock, so a dictation appended in between the + # two is not erased by a rewrite that never saw it. + with _FILES_LOCK: + try: + with open(HISTORY_FILE, encoding="utf-8") as fh: + lines = fh.readlines() + except OSError: + return + if len(lines) <= limit: + return + _write_history(lines[-limit:]) def _row_key(row): return json.dumps(row, ensure_ascii=False, sort_keys=True) +def amend_history(entry, **changes): + """Patch one entry in place, matched on its whole content like delete_history. + + For the caller that learns something after its row is already written: the + row goes in before the paste is attempted, and a paste that then fails + still has to end up in the record. None when the row is gone, which a trim + in between can legitimately make true.""" + wanted = _row_key(entry) + with _FILES_LOCK: + rows = _read_history() + for row in rows: + if _row_key(row) == wanted: + row.update(changes) + _write_history([json.dumps(r, ensure_ascii=False) + "\n" + for r in rows]) + return row + return None + + def delete_history(rows): """Remove the given entries, matched on their whole content rather than on a line number: the worker may have appended a new one since the list was read.""" doomed = {_row_key(row) for row in rows} if not doomed: return - kept = [json.dumps(row, ensure_ascii=False) + "\n" - for row in read_history() if _row_key(row) not in doomed] - _write_history(kept) + with _FILES_LOCK: + kept = [json.dumps(row, ensure_ascii=False) + "\n" + for row in _read_history() if _row_key(row) not in doomed] + _write_history(kept) def clear_history(): - HISTORY_FILE.unlink(missing_ok=True) + with _FILES_LOCK: + HISTORY_FILE.unlink(missing_ok=True) # --- meetings ------------------------------------------------------------- @@ -804,6 +883,12 @@ def meeting_paths(base): def read_meetings(): """Newest last.""" + with _FILES_LOCK: + return _read_meetings() + + +def _read_meetings(): + """The body of read_meetings, for callers already holding the lock.""" try: with open(MEETINGS_FILE, encoding="utf-8") as fh: lines = fh.readlines() @@ -826,29 +911,33 @@ def _write_meetings(rows): with open(tmp, "w", encoding="utf-8") as fh: for row in rows: fh.write(json.dumps(row, ensure_ascii=False) + "\n") - tmp.replace(MEETINGS_FILE) + fh.flush() + os.fsync(fh.fileno()) + _replace_with_retry(tmp, MEETINGS_FILE) def save_meeting(entry): """Insert the row, or replace the one with the same base.""" - rows = read_meetings() - for index, row in enumerate(rows): - if row["base"] == entry["base"]: - rows[index] = entry - break - else: - rows.append(entry) - _write_meetings(rows) + with _FILES_LOCK: + rows = _read_meetings() + for index, row in enumerate(rows): + if row["base"] == entry["base"]: + rows[index] = entry + break + else: + rows.append(entry) + _write_meetings(rows) def update_meeting(base, **changes): """Patch one row and hand it back, or None when it is gone.""" - rows = read_meetings() - for row in rows: - if row["base"] == base: - row.update(changes) - _write_meetings(rows) - return row + with _FILES_LOCK: + rows = _read_meetings() + for row in rows: + if row["base"] == base: + row.update(changes) + _write_meetings(rows) + return row return None @@ -857,7 +946,9 @@ def delete_meetings(bases): doomed = set(bases) if not doomed: return - _write_meetings([row for row in read_meetings() if row["base"] not in doomed]) + with _FILES_LOCK: + _write_meetings([row for row in _read_meetings() + if row["base"] not in doomed]) for base in doomed: for path in meeting_paths(base): try: diff --git a/tests/test_config.py b/tests/test_config.py index 9d8e372..a7b6211 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -8,7 +8,10 @@ config and now shadows the default. import json import os +import pathlib import sys +import threading +import time import unittest from unittest import mock @@ -44,6 +47,21 @@ class Loading(DikteTest): conf = cfg.Config() self.assertEqual(conf["cleanup_model"], cfg.DEFAULTS["cleanup_model"]) + def test_a_config_that_is_not_json_is_set_aside_as_evidence(self): + """Left in place it would be overwritten by the very next save.""" + cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True) + cfg.CONFIG_FILE.write_text("{not json", encoding="utf-8") + broken = cfg.CONFIG_FILE.with_suffix(".json.broken") + with mock.patch("builtins.print") as told: + conf = cfg.Config() + self.assertEqual(broken.read_text(encoding="utf-8"), "{not json") + self.assertFalse(cfg.CONFIG_FILE.exists()) + self.assertIn(str(broken), told.call_args[0][0]) + conf.save() + self.assertEqual(broken.read_text(encoding="utf-8"), "{not json") + self.assertEqual(self.read_config_file()["cleanup_model"], + cfg.DEFAULTS["cleanup_model"]) + def test_a_config_that_is_json_but_not_an_object(self): cfg.CONFIG_DIR.mkdir(parents=True, exist_ok=True) cfg.CONFIG_FILE.write_text("[1, 2]", encoding="utf-8") @@ -113,6 +131,41 @@ class Saving(DikteTest): conf.save() self.assertEqual(i18n.language(), "tr") + def test_the_settings_hit_the_disk_before_the_swap(self): + """Renaming a file still in the page cache into place makes a power + cut a settings wipe, which is what the atomic replace exists to stop.""" + with mock.patch("os.fsync") as fsync: + cfg.Config().save() + fsync.assert_called_once() + + def test_a_file_held_briefly_by_a_scanner_does_not_fail_the_save(self): + """Antivirus and sync tools on Windows hold a fresh file for a moment, + and the rename over it fails until they let go.""" + attempts = [] + real_replace = pathlib.Path.replace + + def flaky(path, target): + attempts.append(str(target)) + if len(attempts) < 3: + raise PermissionError("held by a scanner") + return real_replace(path, target) + + with mock.patch.object(pathlib.Path, "replace", flaky), \ + mock.patch("time.sleep"): + cfg.Config().save() + self.assertEqual(len(attempts), 3) + self.assertEqual(self.read_config_file()["language"], + cfg.DEFAULTS["language"]) + + def test_a_file_held_for_good_still_raises(self): + def held(path, target): + raise PermissionError("never let go") + + with mock.patch.object(pathlib.Path, "replace", held), \ + mock.patch("time.sleep"): + with self.assertRaises(PermissionError): + cfg.Config().save() + class Keys(DikteTest): def test_a_stored_key_is_used(self): @@ -350,6 +403,23 @@ class History(DikteTest): cfg.delete_history([]) self.assertEqual(len(cfg.read_history()), 1) + def test_amending_matches_on_content_and_patches_in_place(self): + rows = [self.entry("a"), self.entry("b")] + for row in rows: + cfg.append_history(row) + patched = cfg.amend_history(rows[0], cleanup_error="could not paste") + self.assertEqual(patched["cleanup_error"], "could not paste") + kept = cfg.read_history() + self.assertEqual([row["text"] for row in kept], ["a", "b"]) + self.assertEqual(kept[0]["cleanup_error"], "could not paste") + + def test_amending_a_row_a_trim_took_away_is_a_no_op(self): + row = self.entry("gone") + cfg.append_history(row) + cfg.clear_history() + self.assertIsNone(cfg.amend_history(row, cleanup_error="x")) + self.assertEqual(cfg.read_history(), []) + def test_clearing(self): cfg.append_history(self.entry("a")) cfg.clear_history() @@ -358,6 +428,40 @@ class History(DikteTest): def test_clearing_a_history_that_is_not_there(self): cfg.clear_history() # must not raise + def test_an_append_during_a_trim_is_not_lost(self): + """Trim is read, cut, rewrite; a dictation appended between the read + and the rewrite must wait rather than be erased by a rewrite that + never saw it. The rewrite is slowed down to hold the race open.""" + for index in range(10): + cfg.append_history(self.entry(str(index))) + real_write = cfg._write_history + rewriting = threading.Event() + + def slow_write(lines): + rewriting.set() + time.sleep(0.1) + real_write(lines) + + with mock.patch.object(cfg, "_write_history", slow_write): + trimmer = threading.Thread(target=cfg.trim_history, args=(3,)) + trimmer.start() + # The trim now holds the lock inside its read-cut-rewrite window. + self.assertTrue(rewriting.wait(5)) + appender = threading.Thread(target=cfg.append_history, + args=(self.entry("late"),)) + appender.start() + trimmer.join() + appender.join() + self.assertEqual([row["text"] for row in cfg.read_history()], + ["7", "8", "9", "late"]) + + def test_the_rewrite_hits_the_disk_before_the_swap(self): + for index in range(5): + cfg.append_history(self.entry(str(index))) + with mock.patch("os.fsync") as fsync: + cfg.trim_history(2) + fsync.assert_called_once() + class Meetings(DikteTest): def entry(self, base, **changes): @@ -430,6 +534,27 @@ class Meetings(DikteTest): cfg.delete_meetings([]) self.assertEqual(len(cfg.read_meetings()), 1) + def test_the_index_hits_the_disk_before_the_swap(self): + with mock.patch("os.fsync") as fsync: + cfg.save_meeting(self.entry("a")) + fsync.assert_called_once() + + def test_an_index_held_briefly_by_a_scanner_is_still_written(self): + real_replace = pathlib.Path.replace + attempts = [] + + def flaky(path, target): + attempts.append(str(target)) + if len(attempts) < 3: + raise PermissionError("held by a scanner") + return real_replace(path, target) + + with mock.patch.object(pathlib.Path, "replace", flaky), \ + mock.patch("time.sleep"): + cfg.save_meeting(self.entry("a")) + self.assertEqual(len(attempts), 3) + self.assertEqual([row["base"] for row in cfg.read_meetings()], ["a"]) + class Defaults(unittest.TestCase): """The table itself, which every command line and settings tab reads."""