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
+`<blog_dir>/_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)

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)

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,10 +58,12 @@ foreground of the `cron` process. Each scheduled invocation calls
                                          +-----------+-------------+
                                                      |
                                                      v
-                                         +-------------------------+
++-------------------------+
                                           | _run_with_retry(...)    |
                                           |  -> _run_once(...)      |
-                                         |     -> state.{load,save}|
+                                          |     -> PostedStore.     |
+                                          |        {is_posted,      |
+                                          |         mark_posted_many}|
                                           +-------------------------+
 ```
 
@@ -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)

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
 

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.

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",
 ]

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
 

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")
+
+
+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",
+]

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:

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")
-    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",
+    ]