mirror of
https://github.com/tiennm99/telegram-exporter.git
synced 2026-10-11 03:13:49 +00:00
tiennm99
c37598e008
The session lock was taken after the thing it guards. It lived inside connected_client, so connect() and _login() had already written the shared SQLite session by the time the lock existed. Two runs on the default session each passed their own per-root lock, both opened the same database, and the loser exited 3 *after* causing the corruption its message described. The lock now wraps the whole client lifetime, and the committable-path check runs first so a refused run leaves no lock file next to a session it was never allowed to create. --dry-run is inside that lock too. It writes nothing to the export tree, but it opens the same session file, which is the resource the lock is about - so running it alongside an export now needs its own --session. Refusing to mark an export complete required that *nothing* had succeeded: `failed and not (downloaded or skipped)`. A single already-present file made skipped non-zero and disabled the guard outright, so any resume across a partly-complete export could fail every remaining file and still stamp completed_at with a cursor at end-of-history. A broken export then answered "did my export finish?" with a confident yes. The comparison is now against downloaded + skipped; a legitimate tail of present files still outnumbers its own stray failures and completes normally. Also: - Renewing an expired file reference counted as a failed attempt, so expiry on the final attempt burned the last slot and the fresh reference was never fetched - reported as "exhausted 3 attempts" after two. The refreshed latch already bounds that arm. - Sidecar repair scanned a fixed window back from EOF for the last newline. A partial record larger than the window contains none, so the file was truncated to end-window: still unreadable, one megabyte shorter. The window now grows until a newline is found. - --reset-state made the zeroed cursor durable before rotating the old sidecar, leaving exactly the mixed-generation log that rotating exists to prevent. Rotation now precedes State.open, which is sound only on this path because there is no compatibility check to fail. - Two resets inside one second silently clobbered the first archive through os.replace, and the rename was the one here not fsynced. - title.txt was the only untrusted string written raw. The export root is safe because the title never becomes a path component, but cat title.txt handed ANSI escapes and a right-to-left override to the operator. It is stripped of the same Unicode categories filenames are, from one shared set so the two rules cannot drift, and written atomically. Cost, on the most common path of every resume: - The post directory was fsynced every post, including posts where every file was already present and nothing had been renamed - one fsync per post to re-record a directory entry an earlier run had already made durable. Gated on an actual download. - An already-present file was stat'd three times: exists(), stat(), and again inside _result. One stat now, reused as the recorded size. Moving fsync off the loop is not available: the AST scan forbids to_thread and run_in_executor, deliberately, because parallelism here buys nothing and escalates flood waits. Each fix has a test that fails without it, confirmed by reverting the fix and re-running. 204 tests to 214, coverage unchanged at 93%.
154 lines
5.4 KiB
Python
154 lines
5.4 KiB
Python
"""messages.jsonl: partial-line repair, rotation, and union semantics.
|
|
|
|
Not in the plan's original file list, but phase 5's success criteria name both
|
|
behaviours ("a truncated final line is repaired at startup", "--reset-state
|
|
rotates the sidecar"), and neither belongs in the download-decision table.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from telegram_exporter.sidecar import SIDECAR_NAME, Sidecar
|
|
from telegram_exporter.traversal import Post
|
|
from tests.support import FakeMsg
|
|
|
|
|
|
def lines(root):
|
|
return (root / SIDECAR_NAME).read_text().splitlines()
|
|
|
|
|
|
def test_a_partial_trailing_line_is_truncated_at_startup(tmp_path):
|
|
# A host crash mid-write otherwise leaves a fragment that breaks every JSONL
|
|
# consumer downstream.
|
|
path = tmp_path / SIDECAR_NAME
|
|
path.write_text('{"message_id": 1}\n{"message_id": 2, "fi')
|
|
|
|
with Sidecar(tmp_path):
|
|
pass
|
|
|
|
assert lines(tmp_path) == ['{"message_id": 1}']
|
|
|
|
|
|
def test_a_complete_final_line_without_a_newline_is_kept(tmp_path):
|
|
path = tmp_path / SIDECAR_NAME
|
|
path.write_text('{"message_id": 1}\n{"message_id": 2}')
|
|
|
|
with Sidecar(tmp_path):
|
|
pass
|
|
|
|
assert [json.loads(line)["message_id"] for line in lines(tmp_path)] == [1, 2]
|
|
|
|
|
|
def test_an_intact_file_is_left_alone(tmp_path):
|
|
path = tmp_path / SIDECAR_NAME
|
|
path.write_text('{"message_id": 1}\n')
|
|
|
|
with Sidecar(tmp_path):
|
|
pass
|
|
|
|
assert lines(tmp_path) == ['{"message_id": 1}']
|
|
|
|
|
|
def test_an_empty_or_absent_file_is_not_a_repair_case(tmp_path):
|
|
with Sidecar(tmp_path):
|
|
pass
|
|
assert lines(tmp_path) == []
|
|
|
|
|
|
def test_rotation_preserves_the_old_records_under_a_timestamp(tmp_path):
|
|
(tmp_path / SIDECAR_NAME).write_text('{"message_id": 1}\n')
|
|
sidecar = Sidecar(tmp_path)
|
|
|
|
rotated = sidecar.rotate()
|
|
|
|
assert rotated is not None and rotated.exists()
|
|
assert rotated.name.startswith("messages-") and rotated.suffix == ".jsonl"
|
|
assert not (tmp_path / SIDECAR_NAME).exists()
|
|
# Rotation, not deletion: the current sidecar describes exactly one filter
|
|
# regime while the previous one stays inspectable.
|
|
assert json.loads(rotated.read_text())["message_id"] == 1
|
|
|
|
|
|
def test_rotating_a_missing_sidecar_is_a_no_op(tmp_path):
|
|
assert Sidecar(tmp_path).rotate() is None
|
|
|
|
|
|
def test_records_are_append_only_events_so_consumers_union_them(tmp_path):
|
|
"""Two runs under different filters both record message 1043. Reading only
|
|
the last record would report a narrower file list than what is on disk."""
|
|
msg_photo = FakeMsg(1043, kind="photo", grouped_id=8)
|
|
post = Post(post_id=1042, messages=[msg_photo], max_message_id=1043)
|
|
|
|
for _ in range(2):
|
|
with Sidecar(tmp_path) as sidecar:
|
|
sidecar.append(post, [])
|
|
sidecar.fsync()
|
|
|
|
records = [json.loads(line) for line in lines(tmp_path)]
|
|
assert [r["message_id"] for r in records] == [1043, 1043]
|
|
assert all(r["post_id"] == 1042 and r["grouped_id"] == 8 for r in records)
|
|
|
|
|
|
def test_a_text_only_message_gets_the_same_shape_with_no_files(tmp_path):
|
|
post = Post(post_id=12, messages=[FakeMsg(12, text="hello")], max_message_id=12)
|
|
with Sidecar(tmp_path) as sidecar:
|
|
sidecar.append(post, [])
|
|
sidecar.fsync()
|
|
|
|
record = json.loads(lines(tmp_path)[0])
|
|
assert record["files"] == [] and record["caption"] == "hello"
|
|
assert record["errors"] == []
|
|
|
|
|
|
def test_sender_name_is_only_taken_from_an_already_cached_sender(tmp_path):
|
|
# Never an extra get_entity call: on a 100k-message sweep that is 100k extra
|
|
# RPCs and a flood ban, in exchange for a display string.
|
|
class Sender:
|
|
first_name, last_name, username, title = "Alice", None, "alice", None
|
|
|
|
with_sender = FakeMsg(1, kind="photo", sender=Sender(), sender_id=777)
|
|
without = FakeMsg(2, kind="photo", sender=None, sender_id=778)
|
|
post = Post(post_id=1, messages=[with_sender, without], max_message_id=2)
|
|
|
|
with Sidecar(tmp_path) as sidecar:
|
|
sidecar.append(post, [])
|
|
sidecar.fsync()
|
|
|
|
records = [json.loads(line) for line in lines(tmp_path)]
|
|
assert records[0]["sender_name"] == "Alice"
|
|
assert records[1]["sender_name"] is None
|
|
assert records[1]["sender_id"] == 778
|
|
|
|
|
|
def test_a_partial_record_larger_than_the_scan_window_keeps_earlier_records(tmp_path):
|
|
# Scanning back a fixed window for the last newline finds none when the
|
|
# partial record is bigger than the window. Truncating to `end - window`
|
|
# then cuts inside the partial record and leaves the file just as unreadable,
|
|
# one window shorter - so the window has to grow until a newline is found.
|
|
path = tmp_path / SIDECAR_NAME
|
|
good = '{"message_id": 1}'
|
|
partial = '{"message_id": 2, "caption": "' + "x" * (2 << 20)
|
|
path.write_text(good + "\n" + partial)
|
|
|
|
with Sidecar(tmp_path):
|
|
pass
|
|
|
|
assert lines(tmp_path) == [good]
|
|
|
|
|
|
def test_two_rotations_in_the_same_second_both_stay_inspectable(tmp_path):
|
|
# os.replace onto a one-second stamp silently overwrote the first archive,
|
|
# which is the whole reason --reset-state rotates instead of deleting.
|
|
path = tmp_path / SIDECAR_NAME
|
|
path.write_text('{"message_id": 1}\n')
|
|
first = Sidecar(tmp_path).rotate()
|
|
path.write_text('{"message_id": 2}\n')
|
|
second = Sidecar(tmp_path).rotate()
|
|
|
|
assert first is not None and second is not None
|
|
assert first != second
|
|
assert first.exists() and second.exists()
|
|
assert json.loads(first.read_text())["message_id"] == 1
|
|
assert json.loads(second.read_text())["message_id"] == 2
|