From 7d39f3a33926678b1e54257323e93cad7d8fd6c8 Mon Sep 17 00:00:00 2001 From: m0rph3us1987 Date: Tue, 4 Aug 2026 18:35:28 +0000 Subject: [PATCH] AI Implementation feature(1086): Posted-Anniversary Dedup Store (#5) --- docs/architecture/anniversary-matching.md | 76 ++++++++ docs/architecture/pipeline-runner.md | 63 ++++--- docs/architecture/system-overview.md | 37 ++-- docs/guides/daily-run.md | 20 ++- docs/index.md | 5 +- src/tenbackward/logging_setup.py | 7 + src/tenbackward/main.py | 11 +- src/tenbackward/state.py | 210 ++++++++++++++++++++-- tests/test_run_logging.py | 5 +- tests/test_state.py | 168 +++++++++++++++-- 10 files changed, 516 insertions(+), 86 deletions(-) create mode 100644 docs/architecture/anniversary-matching.md diff --git a/docs/architecture/anniversary-matching.md b/docs/architecture/anniversary-matching.md new file mode 100644 index 0000000..19c93a7 --- /dev/null +++ b/docs/architecture/anniversary-matching.md @@ -0,0 +1,76 @@ +--- +type: architecture +title: Anniversary Matching +description: Rules and file boundaries used to discover Jekyll posts whose publication anniversary is exactly ten years before the current date. +tags: [matching, jekyll, anniversary, posts] +timestamp: 2026-08-04T18:35:00Z +--- + +# Purpose + +`tenbackward.matching` scans the synchronized blog at +`/_posts/blog` and returns posts whose date is exactly ten years +before the current date. The current date is evaluated in the configured +IANA timezone, defaulting to `Europe/Berlin` in the matcher. + +# Discovery Rules + +1. Walk every Markdown file below the configured post root recursively. +2. Parse YAML front matter when present. +3. Require a filename stem shaped as `YYYY-MM-DD-slug` with a valid calendar + date and non-empty slug. +4. Prefer a valid front matter `date`; otherwise use the date in the filename. +5. Match the post year to `today.year - 10` and the month/day to today. +6. Skip posts with `published: false`. +7. Return a `MatchedPost` containing the relative path, title, date, and + canonical site URL. If front matter has no title, use the filename slug. + +# Leap-Day Behavior + +A February 29 post matches February 29 when the current year is a leap year. +In a non-leap year it matches March 1, preserving one anniversary event for +leap-day posts. + +# Output and Integration + +`find_anniversary_matches()` returns `MatchedPost` values for callers that +need metadata. `iter_anniversary_paths()` yields only relative paths; the +pipeline uses those paths as deduplication identifiers in the +[system state store](/architecture/system-overview.md) and persists new IDs in +`posted.json`. + +Unreadable files, malformed front matter, invalid filenames, unpublished +posts, and unexpected per-file errors are skipped with structured warning +events rather than aborting the full scan. + +# Key Files + +| Path | Responsibility | +|---|---| +| `/repo/src/tenbackward/matching.py` | Date parsing, Jekyll front matter handling, anniversary rules, URL construction, and recursive scanning. | +| `/repo/src/tenbackward/main.py` | Supplies the blog post root and site URL, then applies state deduplication. | +| `/repo/tests/test_matching.py` | Covers filename/front matter parsing, date matching, unpublished posts, malformed files, URL output, and leap-day behavior. | +| `/repo/src/tenbackward/blog.py` | Ensures the source repository is cloned or fast-forwarded before matching. | + +# Examples + +## Matching post + +On 2026-08-04, a file named +`_posts/blog/2016/2016-08-04-release.md` is eligible. Its identifier is +`2016/2016-08-04-release.md` and its generated URL is: + +```text +https://example.com/2016/08/04/release/ +``` + +## Skipped post + +A file with `published: false`, an invalid date, or a non-matching +month/day is not yielded and does not enter `posted.json`. + +# Related + +* [Pipeline Runner](/architecture/pipeline-runner.md) +* [System Architecture](/architecture/system-overview.md) +* [Daily Run Guide](/guides/daily-run.md) diff --git a/docs/architecture/pipeline-runner.md b/docs/architecture/pipeline-runner.md index 1ac3d6a..41b1e1d 100644 --- a/docs/architecture/pipeline-runner.md +++ b/docs/architecture/pipeline-runner.md @@ -1,17 +1,19 @@ --- type: architecture title: Pipeline Runner -description: How tenbackward.main turns one cron tick into a startup log line, a retry-wrapped pipeline pass, and a single run_complete summary line. +description: How tenbackward.main synchronizes the blog, discovers anniversary candidates, applies deduplication, retries failures, and emits a run summary. tags: [pipeline, runner, retry] timestamp: 2026-08-04T17:51:00Z --- # Purpose +`tenbackward.main` is the entry point executed by cron. The pipeline first +ensures that the configured blog repository is available and current, then +scans Jekyll posts for today's ten-year anniversary, filters already-recorded +paths, and persists newly discovered identifiers. Job 1086 wires the +anniversary matcher into the runner; Mastodon publishing remains a future +step. -`tenbackward.main` is the entry point executed by cron. Job 1083 -rewrote it to: (a) install JSON logging, (b) execute a retry-wrapped -pipeline pass that is **silent when there are no candidates**, and -(c) emit a single structured `run_complete` summary on success. # Call Flow @@ -20,12 +22,15 @@ main.main() ├── configure_json_logging() ├── load_config() ── raises ConfigError → exit 2 ├── config.data_dir.mkdir(parents=True, exist_ok=True) + ├── ensure_repo(config.data_dir, blog_repo_url, blog_dir, max_retries) + │ └── BlogRepoError → log blog_repo_error → exit 1 ├── log_startup(version, site_url, run_at, tz, hashtags, throwback_prefix) ├── result = _run_with_retry(config) │ ├── for attempt in 1 .. max_retries+1: │ │ try: return _run_once(config) - │ │ except Exception as exc: - │ │ log_error("pipeline_error", exc=exc, attempt=attempt, max_attempts=attempts) + │ │ └── _run_once → ensure_repo → iter_anniversary_paths + │ │ → PostedStore deduplication → mark_posted_many + │ ├── log_error("pipeline_error", exc=exc, attempt=attempt, max_attempts=attempts) │ ├── log_error("pipeline_failed", exc=last_exc, attempts=attempts) │ └── return None ├── if result is None: return 1 @@ -38,10 +43,26 @@ main.main() def _iter_candidates(config: Config) -> Iterable[str]: ... ``` -Job 1083 ships `_iter_candidates` as an **empty iterator** so the -silent-on-no-matches contract is the default behaviour. The future -blog-clone + matching pipeline plugs into this function without -touching `_run_once` or `_run_with_retry`. +Job 1086 wires `_iter_candidates` to the anniversary matcher. The matcher +is now active; it discovers matching Jekyll posts after the repository is +synchronized. + +## Blog Synchronization + +`ensure_repo()` runs before candidate discovery. An empty target is cloned +from `BLOG_REPO_URL`; an existing repository is fetched and fast-forwarded. +Transient Git failures use exponential backoff, while local modifications +fail immediately with `BlogRepoError`. The same synchronization occurs once +in `main()` before `startup` and again inside `_run_once()` as the retryable +pipeline boundary. + + +`_iter_candidates(config)` delegates to `iter_anniversary_paths()` with +`config.blog_dir / "_posts" / "blog"` and `config.site_url`. The matcher +returns relative Markdown paths, such as +`2016/2016-08-04-example.md`, which are used as stable deduplication IDs. +See [Anniversary Matching](/architecture/anniversary-matching.md) for the +file and front matter rules. # `_run_once` — Summary Counters @@ -59,8 +80,11 @@ touching `_run_once` or `_run_with_retry`. | `skipped` | Items that matched but were already in `posted.json` (deduped). | | `posted_ids`| The list of post identifiers written to state. | -`state.save_posted` is called **only** when `posted_ids` is non-empty, -so a no-op run does not touch `posted.json` on disk. +`PostedStore.mark_posted_many` is called **only** when `posted_ids` +is non-empty, so a no-op run does not touch `posted.json` on disk. +The store persists the list under a `"posted"` key +(`{"posted": ["2014/2014-08-04-foo.md", ...]}`) and serialises +concurrent runs with an `fcntl.flock`. # Retry Behaviour (`_run_with_retry`) @@ -73,19 +97,11 @@ so a no-op run does not touch `posted.json` on disk. * When the budget is exhausted, the runner emits `log_error("pipeline_failed", exc=last_exc, attempts=attempts)` and returns `None` so `main` can translate it to `exit 1`. -* `time.sleep(0)` between attempts is the placeholder seam where the - next job can introduce real backoff. # Silent-on-No-Matches Contract -A run with zero candidates produces exactly one line: - -```json -{"ts":"...","level":"INFO","message":"startup","event":"startup",...} -``` - -No `run_complete` line is emitted. Tests assert this by counting log -records (`tests/test_run_logging.py::test_run_summary_silent_when_no_candidates`). +* `run_complete` is emitted for a successful pass, including when all discovered candidates were already posted. +* A run with zero candidates emits only `startup`; no `run_complete` line is emitted. # Example Sequence (Healthy Run) @@ -106,6 +122,7 @@ records (`tests/test_run_logging.py::test_run_summary_silent_when_no_candidates` # Related +* [Anniversary Matching](/architecture/anniversary-matching.md) * [Logging & Run Summary](/architecture/logging.md) * [Config Schema](/architecture/config-schema.md) * [Cron Lifecycle](/operations/cron-lifecycle.md) diff --git a/docs/architecture/system-overview.md b/docs/architecture/system-overview.md index 10593bf..79e9abd 100644 --- a/docs/architecture/system-overview.md +++ b/docs/architecture/system-overview.md @@ -1,7 +1,7 @@ --- type: architecture title: System Architecture -description: Component map of 10Backward — how config, logging, the pipeline runner, state persistence, and container entrypoint are wired together. +description: Component map of 10Backward — how configuration, blog synchronization, anniversary matching, logging, the pipeline runner, state persistence, and container entrypoint are wired together. tags: [architecture, overview] timestamp: 2026-08-04T17:51:00Z --- @@ -9,8 +9,11 @@ timestamp: 2026-08-04T17:51:00Z # Overview `10Backward` is a Mastodon daily-throwback bot that runs as a single -containerised cron job. Job 1083 adds a structured logging layer and a -retry-capable pipeline runner on top of the scaffold shipped by Job 1082. +containerised cron job. Job 1086 adds anniversary matching across Jekyll +posts, backed by the blog clone/pull workflow from Job 1084 and the +structured logging, retry-capable runner, and deduplication state from +Job 1083. The current pipeline identifies matching post paths and records +those identifiers; Mastodon publishing is still a future integration. The container boots, validates environment configuration, renders a `/etc/cron.d/tenbackward` entry that fires once per day at the @@ -33,8 +36,9 @@ foreground of the `cron` process. Each scheduled invocation calls | `tenbackward.main` | CLI entry point; orchestrates startup logging, retry-wrapped pipeline pass, and run summary. | | `tenbackward.config` | Loads `.env` + process env, applies defaults, validates schema, produces a typed `Config` dataclass. | | `tenbackward.logging_setup` | JSON formatter (one log record per line), secret redaction, and `log_startup` / `log_error` / `log_run_summary` helpers. | -| `tenbackward.state` | Read/write of `posted.json` (atomic temp-file replace). | +| `tenbackward.state` | `PostedStore` — self-contained dedup store. `posted.json` holds `{"posted": [str, ...]}`; mutations are serialised by an `fcntl.flock` on a sibling lock file, writes go through a temp-file replace, and `load()` auto-creates an empty list when the file is missing. | | `tenbackward.blog` | GitPython `ensure_repo()` — clones the configured blog repo on first run, fast-forwards it via `pull --ff-only` thereafter, with exponential-backoff retries on transient network errors. | +| `tenbackward.matching` | Walks `_posts/blog/**/*.md`, parses Jekyll front matter and filenames, and yields posts exactly ten years before the current date using Berlin-time and leap-day rules. | | `/etc/cron.d/tenbackward`| Rendered cron file. One daily line that `cd /app` and runs `python -m tenbackward`. | # Communication & Wiring @@ -54,11 +58,13 @@ foreground of the `cron` process. Each scheduled invocation calls +-----------+-------------+ | v - +-------------------------+ - | _run_with_retry(...) | - | -> _run_once(...) | - | -> state.{load,save}| - +-------------------------+ ++-------------------------+ + | _run_with_retry(...) | + | -> _run_once(...) | + | -> PostedStore. | + | {is_posted, | + | mark_posted_many}| + +-------------------------+ ``` * **Env → Config.** `load_config()` merges a `.env` file (when @@ -68,8 +74,11 @@ foreground of the `cron` process. Each scheduled invocation calls * **Cron → main.** Each scheduled tick re-runs `python -m tenbackward`, so every run is a fresh interpreter invocation. * **Pipeline state.** `_run_once` reads `posted.json` via - `state.load_posted()` for dedupe, then writes it back via - `state.save_posted()` only when at least one new post was recorded. + `PostedStore.is_posted()` for dedupe, then writes it back via + `PostedStore.mark_posted_many()` only when at least one new post + was recorded. The store serialises concurrent runs with an + `fcntl.flock` on a sibling lock file and writes via temp-file + rename. # Key Files @@ -79,8 +88,10 @@ foreground of the `cron` process. Each scheduled invocation calls | `/repo/src/tenbackward/main.py` | CLI entry, retry wrapper, pipeline counters. | | `/repo/src/tenbackward/config.py` | Env merging, defaults, validation, `Config` dataclass, `ConfigError`. | | `/repo/src/tenbackward/logging_setup.py` | JSON formatter, secret redaction, structured event helpers. | -| `/repo/src/tenbackward/state.py` | `posted.json` read/write with atomic temp-file replace. | +| `/repo/src/tenbackward/state.py` | `PostedStore` + module helpers (`load_posted`, `is_posted`, `mark_posted`, `mark_posted_many`); list-shaped JSON, `fcntl.flock`, atomic temp-file replace, env-var data dir. | | `/repo/src/tenbackward/blog.py` | `ensure_repo()` — GitPython clone + fast-forward pull with `2^n` retry/backoff; raises `BlogRepoError` on local modifications or exhausted retries. | +| `/repo/src/tenbackward/matching.py` | Jekyll post discovery, front matter parsing, anniversary and leap-day matching, and canonical URL construction. | +| `/repo/src/tenbackward/__main__.py` | Module entrypoint that invokes `tenbackward.main.main()`. | | `/repo/.env.example` | Canonical list of environment variables. | | `/repo/docker-compose.yml` | Service definition; binds env vars from the host `.env`. | | `/repo/Dockerfile` | Builds the runtime image. | @@ -93,6 +104,6 @@ foreground of the `cron` process. Each scheduled invocation calls * [Config Schema](/architecture/config-schema.md) * [Logging & Run Summary](/architecture/logging.md) * [Pipeline Runner](/architecture/pipeline-runner.md) -* [Daily Run Guide](/guides/daily-run.md) +* [Anniversary Matching](/architecture/anniversary-matching.md) * [Environment Variable Setup](/operations/environment-setup.md) * [Cron Lifecycle](/operations/cron-lifecycle.md) diff --git a/docs/guides/daily-run.md b/docs/guides/daily-run.md index 3830843..2d5f129 100644 --- a/docs/guides/daily-run.md +++ b/docs/guides/daily-run.md @@ -1,7 +1,7 @@ --- type: guide title: Daily Run Guide -description: Operator- and tester-focused walkthrough of one daily cron tick — startup, pipeline pass, silent-on-no-matches, and retry behaviour. +description: Operator- and tester-focused walkthrough of one daily cron tick — repository synchronization, anniversary matching, deduplication, logging, and retry behaviour. tags: [guide, run, daily, tester] timestamp: 2026-08-04T17:51:00Z --- @@ -24,6 +24,15 @@ This is the same command the container's cron entry issues once per day at `RUN_AT`. Use it to verify behaviour without waiting for the scheduled tick. +# What the Run Does + +Before scanning, the process ensures the configured blog repository exists +under `/app/data/blog` and is fast-forwarded from `BLOG_REPO_URL`. It then +walks `_posts/blog/**/*.md`, finds posts whose date is exactly ten years +before today, skips IDs already present in `posted.json`, and records new +relative paths. The current implementation records matching IDs only; +Mastodon publication is not yet wired. + # What You Should See The bot uses a **JSON-per-line** logger that writes to @@ -45,8 +54,10 @@ this is the **only** line you should see — the runner emits no | Scenario | Lines in `cron.log` | Container exit | |-------------------------------------------|----------------------------------------------------------------|------------------| -| Config valid, zero candidates (scaffold) | 1× `startup` | `0` | -| Config valid, future blog step posts N≥1 | 1× `startup`, then 1× `run_complete` with `posted=N` | `0` | +| Config valid, no anniversary matches | 1× `startup` | `0` | +| Blog clone or pull fails | 1× `blog_repo_error` | `1` | +| Config valid, matching IDs are new | 1× `startup`, then 1× `run_complete` with `posted=N` | `0` | +| Config valid, all matches already posted | 1× `startup`, then 1× `run_complete` with `posted=0` | `0` | | Pipeline raises once, recovers | 1× `pipeline_error`, then 1× `run_complete` | `0` | | Pipeline keeps raising (budget exhausted) | `MAX_RETRIES + 1` × `pipeline_error`, then 1× `pipeline_failed` | `1` | | Config invalid | 1× `configuration_error` (extras describe what failed) | `2` | @@ -61,7 +72,7 @@ this is the **only** line you should see — the runner emits no | Path (inside container) | When it appears | |-------------------------|--------------------------------------------| | `/app/data/cron.log` | Always (cron appends stdout/stderr here). | -| `/app/data/posted.json` | Created on first persist; only rewritten when a new id is posted. The scaffold run does **not** create this file. | +| `/app/data/posted.json` | Created when the matcher or a state check first loads the store; rewritten only when a new ID is posted. The file contains relative Jekyll paths under a `posted` list. | To verify these from the host: @@ -77,6 +88,7 @@ This is a server-side bot, so there is no UI. The "UI" consists of: * **Log lines on stdout** — one JSON object per `INFO`/`ERROR` event. * **Exit code** — `0` (healthy), `1` (pipeline exhausted), or `2` (config). * **`posted.json`** — JSON state file; its mtime changes whenever a new id is persisted. +* **`/app/data/blog`** — synchronized local clone containing the Jekyll posts scanned for anniversaries. # Examples diff --git a/docs/index.md b/docs/index.md index df81b45..ec49d05 100644 --- a/docs/index.md +++ b/docs/index.md @@ -6,11 +6,12 @@ okf_version: "0.1" * [System Architecture](/architecture/system-overview.md) — Component map of 10Backward: entrypoint, config, logging, runner, state, and cron wiring. * [Config Schema](/architecture/config-schema.md) — Required/optional env vars, validation rules, and the typed Config dataclass. * [Logging & Run Summary](/architecture/logging.md) — JSON formatter, secret redaction, and structured event helpers. -* [Pipeline Runner](/architecture/pipeline-runner.md) — How a cron tick becomes startup + retry-wrapped pass + run_complete. +* [Pipeline Runner](/architecture/pipeline-runner.md) — How a cron tick synchronizes the blog, matches anniversaries, applies deduplication, retries failures, and emits run events. +* [Anniversary Matching](/architecture/anniversary-matching.md) — Jekyll filename/front matter rules, ten-year date matching, leap-day handling, and candidate identifiers. # Operations * [Environment Variable Setup](/operations/environment-setup.md) — Required env vars, renamed keys, and how to populate `.env`. * [Cron Lifecycle](/operations/cron-lifecycle.md) — How `entrypoint.sh` renders `/etc/cron.d/tenbackward` and hands off to `cron -f`. # User Guides -* [Daily Run Guide](/guides/daily-run.md) — Tester/operator walkthrough of a manual daily cron tick and expected log/output behaviour. +* [Daily Run Guide](/guides/daily-run.md) — Tester/operator walkthrough of repository sync, anniversary matching, deduplication, and expected log/output behaviour. diff --git a/src/tenbackward/logging_setup.py b/src/tenbackward/logging_setup.py index 39ba10d..e42cf16 100644 --- a/src/tenbackward/logging_setup.py +++ b/src/tenbackward/logging_setup.py @@ -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", ] diff --git a/src/tenbackward/main.py b/src/tenbackward/main.py index a8cdbd6..16fa1f1 100644 --- a/src/tenbackward/main.py +++ b/src/tenbackward/main.py @@ -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 diff --git a/src/tenbackward/state.py b/src/tenbackward/state.py index dff16e4..6a84a2f 100644 --- a/src/tenbackward/state.py +++ b/src/tenbackward/state.py @@ -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") - with tmp.open("w", encoding="utf-8") as fh: - json.dump(state, fh, sort_keys=True, indent=2) - tmp.replace(path) + + +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(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", +] \ No newline at end of file diff --git a/tests/test_run_logging.py b/tests/test_run_logging.py index b4b7549..874e189 100644 --- a/tests/test_run_logging.py +++ b/tests/test_run_logging.py @@ -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: diff --git a/tests/test_state.py b/tests/test_state.py index bc96d54..8a7bf95 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -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", 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_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("not-json") - assert load_posted(tmp_path) == {} + 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_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_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", + ] \ No newline at end of file