feat: Posted-Anniversary Dedup Store
This commit is contained in:
OpenVelo Agent
@@ -139,6 +139,12 @@ def log_error(event: str, *, exc: BaseException | None = None, **fields: Any) ->
|
||||
logging.getLogger(_LOGGER_NAME).error(event, extra=payload)
|
||||
|
||||
|
||||
def log_warning(event: str, **fields: Any) -> None:
|
||||
"""Emit a single structured warning line, redacting any secret-shaped keys."""
|
||||
payload = _redact_dict({"event": event, **fields})
|
||||
logging.getLogger(_LOGGER_NAME).warning(event, extra=payload)
|
||||
|
||||
|
||||
def log_startup(**fields: Any) -> None:
|
||||
"""Emit a single structured info line at startup."""
|
||||
payload = _redact_dict({"event": "startup", **fields})
|
||||
@@ -152,4 +158,5 @@ __all__ = [
|
||||
"log_error",
|
||||
"log_run_summary",
|
||||
"log_startup",
|
||||
"log_warning",
|
||||
]
|
||||
|
||||
@@ -14,7 +14,7 @@ from .logging_setup import (
|
||||
log_startup,
|
||||
)
|
||||
from .matching import iter_anniversary_paths
|
||||
from .state import load_posted, save_posted
|
||||
from .state import PostedStore
|
||||
|
||||
|
||||
def _iter_candidates(config: Config) -> Iterable[str]:
|
||||
@@ -38,7 +38,7 @@ def _run_once(config: Config) -> tuple[int, int, int, int, list[str]]:
|
||||
max_retries=config.max_retries,
|
||||
)
|
||||
|
||||
already_posted = set(load_posted(config.data_dir).keys())
|
||||
store = PostedStore(config.data_dir)
|
||||
|
||||
scanned = 0
|
||||
matched = 0
|
||||
@@ -46,20 +46,17 @@ def _run_once(config: Config) -> tuple[int, int, int, int, list[str]]:
|
||||
skipped = 0
|
||||
posted_ids: list[str] = []
|
||||
|
||||
state = load_posted(config.data_dir)
|
||||
|
||||
for candidate_id in _iter_candidates(config):
|
||||
scanned += 1
|
||||
matched += 1
|
||||
if candidate_id in already_posted:
|
||||
if store.is_posted(candidate_id):
|
||||
skipped += 1
|
||||
continue
|
||||
posted_ids.append(candidate_id)
|
||||
state[candidate_id] = {"posted_at": _now_iso()}
|
||||
posted += 1
|
||||
|
||||
if posted_ids:
|
||||
save_posted(config.data_dir, state)
|
||||
store.mark_posted_many(posted_ids)
|
||||
|
||||
return scanned, matched, posted, skipped, posted_ids
|
||||
|
||||
|
||||
+191
-17
@@ -1,34 +1,208 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import fcntl
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from typing import Iterable
|
||||
|
||||
from .logging_setup import log_warning
|
||||
|
||||
|
||||
def posted_path(data_dir: Path) -> Path:
|
||||
return Path(data_dir) / "posted.json"
|
||||
DEFAULT_DATA_DIR = Path("./data")
|
||||
_FILE_NAME = "posted.json"
|
||||
_EMPTY_DOCUMENT: dict[str, list[str]] = {"posted": []}
|
||||
|
||||
|
||||
def load_posted(data_dir: Path) -> dict[str, Any]:
|
||||
"""Return the posted-state map, creating an empty one if missing."""
|
||||
path = posted_path(data_dir)
|
||||
if not path.exists():
|
||||
return {}
|
||||
class PostedStoreError(RuntimeError):
|
||||
"""Raised when the posted-state store cannot be used."""
|
||||
|
||||
|
||||
def posted_path(data_dir: Path | None = None) -> Path:
|
||||
"""Resolve the path of ``posted.json``.
|
||||
|
||||
When ``data_dir`` is ``None``, the directory is read from the
|
||||
``DATA_DIR`` environment variable, falling back to ``./data``
|
||||
(the host bind-mount described in the README).
|
||||
"""
|
||||
if data_dir is None:
|
||||
data_dir = Path(os.environ.get("DATA_DIR", str(DEFAULT_DATA_DIR)))
|
||||
return Path(data_dir) / _FILE_NAME
|
||||
|
||||
|
||||
def _coerce_list(value: object) -> list[str]:
|
||||
"""Filter ``value`` down to a list of strings, dropping anything else."""
|
||||
if not isinstance(value, list):
|
||||
return []
|
||||
return [item for item in value if isinstance(item, str)]
|
||||
|
||||
|
||||
def _read_existing(path: Path) -> dict[str, list[str]] | None:
|
||||
"""Return the parsed JSON document at ``path`` or ``None`` on errors.
|
||||
|
||||
A missing file is **not** an error: returns ``None`` so the caller
|
||||
can treat it as the empty document. Malformed JSON or unreadable
|
||||
bytes return ``None`` and a warning is emitted.
|
||||
"""
|
||||
try:
|
||||
with path.open("r", encoding="utf-8") as fh:
|
||||
data = json.load(fh)
|
||||
except json.JSONDecodeError:
|
||||
return {}
|
||||
except FileNotFoundError:
|
||||
return None
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
log_warning("posted_store_corrupt", path=str(path), error=type(exc).__name__)
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return {}
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def save_posted(data_dir: Path, state: dict[str, Any]) -> None:
|
||||
"""Persist the posted-state map atomically-ish."""
|
||||
path = posted_path(data_dir)
|
||||
def _ensure_parent(path: Path) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = path.with_suffix(path.suffix + ".tmp")
|
||||
|
||||
|
||||
def _write_document(path: Path, document: dict[str, list[str]]) -> None:
|
||||
"""Atomically write ``document`` to ``path`` via temp-file + rename."""
|
||||
_ensure_parent(path)
|
||||
pid = os.getpid()
|
||||
tmp = path.with_name(f"{path.name}.tmp.{pid}")
|
||||
try:
|
||||
with tmp.open("w", encoding="utf-8") as fh:
|
||||
json.dump(state, fh, sort_keys=True, indent=2)
|
||||
tmp.replace(path)
|
||||
json.dump(document, fh, sort_keys=False, indent=2)
|
||||
fh.flush()
|
||||
os.fsync(fh.fileno())
|
||||
os.replace(tmp, path)
|
||||
except Exception:
|
||||
try:
|
||||
tmp.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
raise
|
||||
|
||||
|
||||
def _lock_path(path: Path):
|
||||
"""Return an open file descriptor on the lock file inside the data dir.
|
||||
|
||||
The lock file lives in the same directory as ``posted.json`` so the
|
||||
flock is always on a single inode, regardless of whether the JSON
|
||||
file currently exists. The descriptor is opened in append mode so
|
||||
concurrent readers never truncate it.
|
||||
"""
|
||||
_ensure_parent(path)
|
||||
lock = path.with_name(f".{path.name}.lock")
|
||||
fd = os.open(str(lock), os.O_CREAT | os.O_RDWR, 0o644)
|
||||
return lock, fd
|
||||
|
||||
|
||||
class PostedStore:
|
||||
"""Self-contained dedup store persisted as ``posted.json``.
|
||||
|
||||
The on-disk shape is::
|
||||
|
||||
{"posted": ["2014/2014-08-04-foo.md", ...]}
|
||||
|
||||
All mutating operations acquire an exclusive :mod:`fcntl` flock on a
|
||||
sibling lock file so concurrent container runs cannot corrupt the
|
||||
JSON document. The store never raises on a missing file, an
|
||||
unreadable file, or an already-present path — it always reports
|
||||
success in a way that the orchestrator can act on.
|
||||
"""
|
||||
|
||||
def __init__(self, data_dir: Path | None = None) -> None:
|
||||
self._path = posted_path(data_dir)
|
||||
|
||||
@property
|
||||
def path(self) -> Path:
|
||||
return self._path
|
||||
|
||||
def load(self) -> list[str]:
|
||||
"""Read the posted list, creating an empty file when missing."""
|
||||
_ensure_parent(self._path)
|
||||
document = _read_existing(self._path)
|
||||
if document is None:
|
||||
if not self._path.exists():
|
||||
_write_document(self._path, dict(_EMPTY_DOCUMENT))
|
||||
return []
|
||||
return _coerce_list(document.get("posted"))
|
||||
|
||||
def is_posted(self, relative_path: str) -> bool:
|
||||
"""Return ``True`` when ``relative_path`` is already recorded."""
|
||||
posted = self.load()
|
||||
return relative_path in posted
|
||||
|
||||
def _with_lock(self, mutate):
|
||||
lock_path, fd = _lock_path(self._path)
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_EX)
|
||||
return mutate()
|
||||
finally:
|
||||
try:
|
||||
fcntl.flock(fd, fcntl.LOCK_UN)
|
||||
finally:
|
||||
os.close(fd)
|
||||
try:
|
||||
lock_path.unlink()
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
def mark_posted(self, relative_path: str) -> bool:
|
||||
"""Append ``relative_path`` to the stored list. Idempotent.
|
||||
|
||||
Returns ``True`` when the path was newly added, ``False`` when
|
||||
it was already present.
|
||||
"""
|
||||
return self.mark_posted_many([relative_path]) != []
|
||||
|
||||
def mark_posted_many(self, relative_paths: Iterable[str]) -> list[str]:
|
||||
"""Append any new entries from ``relative_paths`` in one write.
|
||||
|
||||
Returns the list of paths that were newly added (possibly
|
||||
empty when the store already contained every supplied path).
|
||||
"""
|
||||
|
||||
candidates = [p for p in relative_paths if isinstance(p, str) and p]
|
||||
if not candidates:
|
||||
return []
|
||||
|
||||
def _mutate() -> list[str]:
|
||||
document = _read_existing(self._path) or dict(_EMPTY_DOCUMENT)
|
||||
existing = _coerce_list(document.get("posted"))
|
||||
added = [p for p in candidates if p not in existing]
|
||||
if not added:
|
||||
return []
|
||||
existing.extend(added)
|
||||
document["posted"] = existing
|
||||
_write_document(self._path, document)
|
||||
return added
|
||||
|
||||
return self._with_lock(_mutate)
|
||||
|
||||
|
||||
def load_posted(data_dir: Path | None = None) -> list[str]:
|
||||
return PostedStore(data_dir).load()
|
||||
|
||||
|
||||
def is_posted(relative_path: str, data_dir: Path | None = None) -> bool:
|
||||
return PostedStore(data_dir).is_posted(relative_path)
|
||||
|
||||
|
||||
def mark_posted(relative_path: str, data_dir: Path | None = None) -> bool:
|
||||
return PostedStore(data_dir).mark_posted(relative_path)
|
||||
|
||||
|
||||
def mark_posted_many(
|
||||
relative_paths: list[str], data_dir: Path | None = None
|
||||
) -> list[str]:
|
||||
return PostedStore(data_dir).mark_posted_many(relative_paths)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"DEFAULT_DATA_DIR",
|
||||
"PostedStore",
|
||||
"PostedStoreError",
|
||||
"is_posted",
|
||||
"load_posted",
|
||||
"mark_posted",
|
||||
"mark_posted_many",
|
||||
"posted_path",
|
||||
]
|
||||
@@ -30,10 +30,9 @@ def _run_lines(buf: io.StringIO) -> list[dict]:
|
||||
|
||||
|
||||
def _seed_state(data_dir: Path, ids: list[str]) -> None:
|
||||
from tenbackward.state import save_posted
|
||||
from tenbackward.state import mark_posted_many
|
||||
|
||||
state = {pid: {"posted_at": "2024-01-01T00:00:00+00:00"} for pid in ids}
|
||||
save_posted(data_dir, state)
|
||||
mark_posted_many(list(ids), data_dir)
|
||||
|
||||
|
||||
def test_run_emits_one_info_summary_on_success(env_setup, data_dir, capture_logger, monkeypatch) -> None:
|
||||
|
||||
+152
-16
@@ -1,31 +1,167 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from tenbackward.state import load_posted, posted_path, save_posted
|
||||
from tenbackward.logging_setup import JsonFormatter
|
||||
from tenbackward.state import (
|
||||
PostedStore,
|
||||
is_posted,
|
||||
load_posted,
|
||||
mark_posted,
|
||||
mark_posted_many,
|
||||
posted_path,
|
||||
)
|
||||
|
||||
|
||||
def test_load_posted_creates_empty_when_missing(tmp_path: Path) -> None:
|
||||
state = load_posted(tmp_path)
|
||||
assert state == {}
|
||||
def _store(tmp_path: Path) -> PostedStore:
|
||||
return PostedStore(tmp_path)
|
||||
|
||||
|
||||
def test_round_trip_persists(tmp_path: Path) -> None:
|
||||
state = {"post-1": {"posted_at": "2024-01-01T00:00:00Z"}}
|
||||
save_posted(tmp_path, state)
|
||||
def test_load_creates_empty_file_when_missing(tmp_path: Path) -> None:
|
||||
store = _store(tmp_path)
|
||||
result = store.load()
|
||||
|
||||
assert result == []
|
||||
assert posted_path(tmp_path).exists()
|
||||
again = load_posted(tmp_path)
|
||||
assert again == state
|
||||
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
||||
assert document == {"posted": []}
|
||||
|
||||
|
||||
def test_load_posted_handles_corrupt_file(tmp_path: Path) -> None:
|
||||
def test_load_returns_empty_when_file_missing_no_crash(tmp_path: Path) -> None:
|
||||
assert _store(tmp_path).load() == []
|
||||
|
||||
|
||||
def test_load_warns_and_returns_empty_on_malformed_json(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
buf = io.StringIO()
|
||||
handler = logging.StreamHandler(buf)
|
||||
handler.setFormatter(JsonFormatter())
|
||||
logger = logging.getLogger("tenbackward")
|
||||
original_handlers = list(logger.handlers)
|
||||
original_propagate = logger.propagate
|
||||
logger.handlers = [handler]
|
||||
logger.setLevel(logging.WARNING)
|
||||
logger.propagate = False
|
||||
try:
|
||||
posted_path(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
posted_path(tmp_path).write_text("not-json")
|
||||
assert load_posted(tmp_path) == {}
|
||||
posted_path(tmp_path).write_text("not-json", encoding="utf-8")
|
||||
result = _store(tmp_path).load()
|
||||
finally:
|
||||
logger.handlers = original_handlers
|
||||
logger.propagate = original_propagate
|
||||
|
||||
assert result == []
|
||||
lines = [
|
||||
json.loads(line) for line in buf.getvalue().splitlines() if line.strip()
|
||||
]
|
||||
events = [line.get("event") for line in lines]
|
||||
assert "posted_store_corrupt" in events
|
||||
|
||||
|
||||
def test_save_posted_creates_parent_dirs(tmp_path: Path) -> None:
|
||||
nested = tmp_path / "deep" / "data"
|
||||
save_posted(nested, {"x": 1})
|
||||
assert posted_path(nested).exists()
|
||||
def test_mark_posted_writes_relative_path_to_file(tmp_path: Path) -> None:
|
||||
_store(tmp_path).mark_posted("2014/2014-08-04-foo.md")
|
||||
|
||||
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
||||
assert document == {"posted": ["2014/2014-08-04-foo.md"]}
|
||||
|
||||
|
||||
def test_is_posted_round_trip(tmp_path: Path) -> None:
|
||||
store = _store(tmp_path)
|
||||
store.mark_posted("2014/2014-08-04-foo.md")
|
||||
|
||||
assert store.is_posted("2014/2014-08-04-foo.md") is True
|
||||
assert store.is_posted("2015/2015-01-01-bar.md") is False
|
||||
|
||||
|
||||
def test_mark_posted_is_idempotent(tmp_path: Path) -> None:
|
||||
store = _store(tmp_path)
|
||||
first = store.mark_posted("2014/2014-08-04-foo.md")
|
||||
second = store.mark_posted("2014/2014-08-04-foo.md")
|
||||
|
||||
assert first is True
|
||||
assert second is False
|
||||
|
||||
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
||||
assert document == {"posted": ["2014/2014-08-04-foo.md"]}
|
||||
|
||||
|
||||
def test_mark_posted_many_persists_in_one_write(
|
||||
tmp_path: Path, monkeypatch
|
||||
) -> None:
|
||||
store = _store(tmp_path)
|
||||
replaces: list[tuple[object, object]] = []
|
||||
real_replace = os.replace
|
||||
|
||||
def counting_replace(src, dst) -> None:
|
||||
replaces.append((src, dst))
|
||||
real_replace(src, dst)
|
||||
|
||||
monkeypatch.setattr("tenbackward.state.os.replace", counting_replace)
|
||||
|
||||
added = store.mark_posted_many(
|
||||
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
||||
)
|
||||
|
||||
assert added == ["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
||||
assert len(replaces) == 1
|
||||
assert Path(replaces[0][1]) == posted_path(tmp_path)
|
||||
|
||||
|
||||
def test_mark_posted_leaves_no_temp_files(tmp_path: Path) -> None:
|
||||
_store(tmp_path).mark_posted("2014/2014-08-04-foo.md")
|
||||
|
||||
leftovers = [
|
||||
p for p in tmp_path.iterdir() if p.name.startswith("posted.json.tmp.")
|
||||
]
|
||||
assert leftovers == []
|
||||
|
||||
|
||||
def test_mark_posted_many_against_existing_is_no_op(tmp_path: Path) -> None:
|
||||
store = _store(tmp_path)
|
||||
store.mark_posted("2014/2014-08-04-foo.md")
|
||||
|
||||
added = store.mark_posted_many(
|
||||
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
||||
)
|
||||
|
||||
assert added == ["2015/2015-01-01-bar.md"]
|
||||
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
||||
assert document == {
|
||||
"posted": ["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
||||
}
|
||||
|
||||
|
||||
def test_posted_path_defaults_via_DATA_DIR(
|
||||
monkeypatch, tmp_path: Path
|
||||
) -> None:
|
||||
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
||||
assert posted_path(None) == tmp_path / "posted.json"
|
||||
|
||||
|
||||
def test_load_filters_non_string_entries(tmp_path: Path) -> None:
|
||||
posted_path(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
posted_path(tmp_path).write_text(
|
||||
json.dumps({"posted": ["good", 42, None, "also-good"]}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
assert _store(tmp_path).load() == ["good", "also-good"]
|
||||
|
||||
|
||||
def test_module_helpers_round_trip(tmp_path: Path) -> None:
|
||||
assert load_posted(tmp_path) == []
|
||||
assert is_posted("2014/2014-08-04-foo.md", tmp_path) is False
|
||||
assert mark_posted("2014/2014-08-04-foo.md", tmp_path) is True
|
||||
added = mark_posted_many(
|
||||
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"], tmp_path
|
||||
)
|
||||
assert added == ["2015/2015-01-01-bar.md"]
|
||||
assert load_posted(tmp_path) == [
|
||||
"2014/2014-08-04-foo.md",
|
||||
"2015/2015-01-01-bar.md",
|
||||
]
|
||||
Reference in New Issue
Block a user