diff --git a/src/agent.py b/src/agent.py index 40ae56f..76c4d4c 100644 --- a/src/agent.py +++ b/src/agent.py @@ -206,26 +206,54 @@ class Journal: def __init__(self, path: str): self.path = path + def _torn_tail(self) -> bool: + """Файл не пуст и не заканчивается переводом строки (оборванная запись).""" + try: + if os.path.getsize(self.path) == 0: + return False + except FileNotFoundError: + return False + with open(self.path, "rb") as f: + f.seek(-1, os.SEEK_END) + return f.read(1) != b"\n" + def append(self, record: dict) -> None: record = dict(record) record.setdefault("at", now_iso()) - with open(self.path, "a", encoding="utf-8") as f: - f.write(json.dumps(record, ensure_ascii=False) + "\n") + data = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8") + if self._torn_tail(): + data = b"\n" + data + fd = os.open(self.path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o600) + with os.fdopen(fd, "ab") as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) - def records(self) -> List[dict]: + def _scan(self): + """Возвращает (годные записи, номера битых строк). Годная — dict с command_id.""" + good, bad = [], [] if not os.path.exists(self.path): - return [] - out = [] - with open(self.path, encoding="utf-8") as f: - for n, line in enumerate(f, 1): - line = line.strip() + return good, bad + with open(self.path, "rb") as f: + for n, raw in enumerate(f, 1): + line = raw.strip() if not line: continue try: - out.append(json.loads(line)) + r = json.loads(line.decode("utf-8")) except ValueError: - log("журнал: строка %d не читается, пропускаю" % n) - return out + r = None + if isinstance(r, dict) and "command_id" in r: + good.append(r) + else: + bad.append(n) + return good, bad + + def records(self) -> List[dict]: + good, bad = self._scan() + for n in bad: + log("журнал: строка %d не читается, пропускаю" % n) + return good def started(self, command_id) -> None: self.append({"kind": "started", "command_id": int(command_id)}) @@ -248,34 +276,64 @@ class Journal: return [r for cid, r in last.items() if cid not in acked] def unfinished_started(self) -> List[int]: - started, finished = [], set() + """Порядковый учёт: started открывает id, finished/acked закрывает; порядок — первое открытие.""" + opened = {} # command_id -> открыта ли сейчас; dict сохраняет порядок первого открытия for r in self.records(): - if r.get("kind") == "started": - started.append(r["command_id"]) - elif r.get("kind") in ("finished", "acked"): - finished.add(r["command_id"]) - return [cid for cid in started if cid not in finished] + kind, cid = r.get("kind"), r["command_id"] + if kind == "started": + opened[cid] = True + elif kind in ("finished", "acked") and cid in opened: + opened[cid] = False + return [cid for cid, is_open in opened.items() if is_open] def prune(self, days: int = 30) -> int: - """Удалить записи команд, подтверждённых (acked) раньше, чем days дней назад. Неподтверждённые — остаются.""" - records = self.records() + """Удалить записи id до его последнего acked, если тот старше days дней (возраст — по acked) + и после него у id нет started/finished. Возвращает число удалённых годных записей; битые строки + отбрасываются при перезаписи и учитываются в логе.""" + good, bad = self._scan() + if bad: + log("журнал: удаляю битых строк %d" % len(bad)) cutoff = datetime.now(timezone.utc).timestamp() - days * 86400 - old = set() - for r in records: + last_ack, last_open = {}, {} + for i, r in enumerate(good): + cid = r["command_id"] if r.get("kind") == "acked": - try: - ts = datetime.fromisoformat(r.get("at", "")).timestamp() - except ValueError: - continue - if ts < cutoff: - old.add(r["command_id"]) - keep = [r for r in records if r.get("command_id") not in old] + last_ack[cid] = i + elif r.get("kind") in ("started", "finished"): + last_open[cid] = i + drop_upto = {} # command_id -> индекс последнего acked: всё до него включительно удаляется + for cid, i in last_ack.items(): + if last_open.get(cid, -1) > i: + continue + try: + ts = _parse_at(good[i].get("at")) + except (ValueError, TypeError): + continue + if ts < cutoff: + drop_upto[cid] = i + keep = [r for i, r in enumerate(good) + if not (r["command_id"] in drop_upto and i <= drop_upto[r["command_id"]])] tmp = self.path + ".tmp" - with open(tmp, "w", encoding="utf-8") as f: + try: + os.remove(tmp) + except FileNotFoundError: + pass + fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "wb") as f: for r in keep: - f.write(json.dumps(r, ensure_ascii=False) + "\n") + f.write((json.dumps(r, ensure_ascii=False) + "\n").encode("utf-8")) + f.flush() + os.fsync(f.fileno()) os.replace(tmp, self.path) - return len(records) - len(keep) + return len(good) - len(keep) + + +def _parse_at(value) -> float: + """Метка времени журнала в эпоху; наивное время трактуется как UTC.""" + dt = datetime.fromisoformat(value) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.timestamp() # --------------------------------------------------------------------------- diff --git a/tests/test_journal.py b/tests/test_journal.py index 016134b..9a99603 100644 --- a/tests/test_journal.py +++ b/tests/test_journal.py @@ -1,9 +1,8 @@ -import contextlib -import io import json import os import tempfile import unittest +from unittest import mock from _common import agent @@ -13,6 +12,9 @@ class JournalTest(unittest.TestCase): self.dir = tempfile.TemporaryDirectory() self.path = os.path.join(self.dir.name, "journal.jsonl") self.j = agent.Journal(self.path) + patcher = mock.patch.object(agent, "log") + self.log = patcher.start() + self.addCleanup(patcher.stop) def tearDown(self): self.dir.cleanup() @@ -45,11 +47,8 @@ class JournalTest(unittest.TestCase): with open(self.path, "a", encoding="utf-8") as f: f.write("{не json\n") self.j.finished(1, "done", None, None, 1, 0) - out = io.StringIO() - with contextlib.redirect_stdout(out): - records = self.j.records() - self.assertEqual(len(records), 2) - self.assertIn("строка 2 не читается", out.getvalue()) + self.assertEqual(len(self.j.records()), 2) + self.log.assert_called_with("журнал: строка 2 не читается, пропускаю") def test_prune_keeps_unacked_and_recent(self): self.j.started(1) @@ -73,5 +72,68 @@ class JournalTest(unittest.TestCase): self.assertEqual(json.loads(text.splitlines()[0])["kind"], "finished") + def test_torn_tail_is_not_glued(self): + with open(self.path, "wb") as f: + f.write('{"kind": "star'.encode("utf-8")) + self.j.started(1) + recs = self.j.records() + self.assertEqual([(r["kind"], r["command_id"]) for r in recs], [("started", 1)]) + self.assertEqual(self.j.unfinished_started(), [1]) + + def test_restart_after_acked_reopens(self): + self.j.started(1) + self.j.finished(1, "done", None, None, 1, 0) + self.j.acked(1) + self.assertEqual(self.j.unfinished_started(), []) + self.j.started(1) + self.assertEqual(self.j.unfinished_started(), [1]) + + def test_duplicate_started_is_deduplicated_in_first_open_order(self): + self.j.started(2) + self.j.started(1) + self.j.started(1) + self.assertEqual(self.j.unfinished_started(), [2, 1]) + self.j.finished(2, "done", None, None, 1, 0) + self.assertEqual(self.j.unfinished_started(), [1]) + + def test_prune_keeps_id_with_newer_records(self): + self.j.started(1) + self.j.finished(1, "done", None, None, 1, 0) + self.j.append({"kind": "acked", "command_id": 1, "at": "2020-01-01T00:00:00+00:00"}) + self.j.started(1) + self.j.finished(1, "done", None, None, 1, 0) + before = self.j.records() + self.assertEqual(self.j.prune(days=30), 0) + self.assertEqual(self.j.records(), before) + + def test_non_dict_and_no_command_id_lines_are_skipped(self): + with open(self.path, "wb") as f: + f.write(b'[1, 2]\n5\n{"kind": "started"}\n') + self.j.started(1) + self.j.finished(1, "done", None, None, 1, 0) + self.j.acked(1) + self.assertEqual(len(self.j.records()), 3) + self.assertEqual(self.j.unacked_finished(), []) + self.assertEqual(self.j.unfinished_started(), []) + self.assertEqual(self.j.prune(days=30), 0) + self.log.assert_any_call("журнал: удаляю битых строк 3") + self.assertEqual(len(self.j.records()), 3) + with open(self.path, encoding="utf-8") as f: + self.assertNotIn("[1, 2]", f.read()) + + def test_naive_and_invalid_at_in_prune(self): + self.j.started(1) + self.j.append({"kind": "acked", "command_id": 1, "at": "2020-01-01T00:00:00"}) + self.j.started(2) + self.j.append({"kind": "acked", "command_id": 2, "at": "не дата"}) + self.assertEqual(self.j.prune(days=30), 2) + self.assertEqual([(r["kind"], r["command_id"]) for r in self.j.records()], [("started", 2), ("acked", 2)]) + + def test_journal_file_is_private(self): + self.j.started(1) + mode = os.stat(self.path).st_mode & 0o777 + self.assertEqual(mode & 0o077, 0) + + if __name__ == "__main__": unittest.main()