m0rph3us1987/10Backward
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI Implementation feature(1089): Scheduled Execution with Cron and Retry Handling (#9)

Browse Source
This commit was merged in pull request #9.
This commit is contained in:
m0rph3us1987
2026-08-04 20:48:16 +00:00
parent 3545b81546
commit 5f6ddf76a4
27 changed files with 644 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tenbackward/blog.py
+103 -48
View File
@@ -26,7 +26,24 @@ _LOCAL_MODIFICATION_MARKERS = (
class BlogRepoError(RuntimeError):
"""Raised when the local blog working tree cannot be ensured."""
"""Raised when the local blog working tree cannot be ensured.
Surface this as a fatal pipeline error (do not retry) so configuration
problems fail immediately and transient network failures get a chance to
recover via the retry loop.
"""
class BlogTransientError(RuntimeError):
"""Wraps a transient HTTPS clone/pull failure that should be retried.
``operation`` records which Git action triggered the failure so the
runner can label its log lines with "git clone" or "git pull".
"""
def __init__(self, message: str, *, operation: str = "git pull") -> None:
super().__init__(message)
self.operation = operation
def blog_dir(data_dir: Path) -> Path:
@@ -52,54 +69,57 @@ def _is_local_modification_error(exc: GitCommandError) -> bool:
return any(marker in message for marker in _LOCAL_MODIFICATION_MARKERS)
def _fetch_and_pull(
repo: Repo,
*,
fetch: Callable[[], object],
pull: Callable[[], object],
sleep: Callable[[float], None],
max_retries: int,
def _log_retry(
logger: logging.Logger | None,
*,
operation: str,
attempt: int,
error: BaseException,
) -> None:
"""Fetch + fast-forward pull with retry/backoff for transient network errors.
if logger is None:
return
logger.error(
"blog_retry",
extra={
"event": "blog_retry",
"operation": operation,
"attempt": attempt,
"error": str(error),
},
)
Local-modification errors raise ``BlogRepoError`` immediately, without retry.
def _retry_loop(
*,
operation: str,
attempts: int,
sleep: Callable[[float], None],
logger: logging.Logger | None,
action: Callable[[], None],
) -> None:
"""Invoke ``action`` up to ``attempts`` times with ``2 ** (attempt - 1)``
second backoff between failures.
Transient errors are retried; :class:`BlogRepoError` (fatal local or
validation problems) propagates immediately. On exhaustion, raises
:class:`BlogTransientError` carrying the failing operation label.
"""
attempts = max(1, max_retries + 1)
last_exc: GitCommandError | None = None
last_exc: BaseException | None = None
for attempt in range(1, attempts + 1):
try:
fetch()
pull()
action()
return
except GitCommandError as exc:
except BlogRepoError:
raise
except Exception as exc: # noqa: BLE001 — boundary retry hook
last_exc = exc
if _is_local_modification_error(exc):
if logger is not None:
logger.error(
"blog_local_modifications",
extra={"event": "blog_local_modifications", "error": str(exc)},
)
raise BlogRepoError(
"local modifications detected in blog working tree; aborting run"
) from exc
if logger is not None:
logger.error(
"blog_fetch_error",
extra={
"event": "blog_fetch_error",
"attempt": attempt,
"max_attempts": attempts,
"error": str(exc),
},
)
_log_retry(logger, operation=operation, attempt=attempt, error=exc)
if attempt < attempts:
sleep(BACKOFF_BASE_SECONDS ** attempt)
sleep(BACKOFF_BASE_SECONDS ** (attempt - 1))
assert last_exc is not None
raise BlogRepoError(
f"blog fetch/pull failed after {attempts} attempt(s): {last_exc}"
raise BlogTransientError(
f"{operation} failed after {attempts} attempt(s): {last_exc}",
operation=operation,
) from last_exc
@@ -111,17 +131,22 @@ def ensure_repo(
max_retries: int | None = None,
logger: logging.Logger | None = None,
sleep: Callable[[float], None] = time.sleep,
clone_impl: Callable[[str, str], Repo] | None = None,
fetch_impl: Callable[[Repo], Callable[[], object]] | None = None,
pull_impl: Callable[[Repo], Callable[[], object]] | None = None,
) -> Repo:
"""Ensure a local working clone of the blog repository exists and is up to date.
Returns the :class:`git.Repo` pointing at an up-to-date working tree.
Raises :class:`BlogRepoError` on irrecoverable failures.
Raises :class:`BlogRepoError` on irrecoverable failures and
:class:`BlogTransientError` after a transient failure exhausts the
retry budget (so the caller can label and re-attempt at the pipeline
level).
"""
url = repo_url or blog_repo_url()
target = blog_path or blog_dir(data_dir)
retries = MAX_RETRIES_DEFAULT if max_retries is None else max_retries
attempts = max(1, retries)
target.parent.mkdir(parents=True, exist_ok=True)
@@ -133,13 +158,28 @@ def ensure_repo(
"blog_clone_start",
extra={"event": "blog_clone_start", "url": url, "path": str(target)},
)
cloned = Repo.clone_from(url, str(target))
cloned_holder: dict[str, Repo] = {}
def _do_clone() -> None:
r = (clone_impl or Repo.clone_from)(url, str(target))
cloned_holder["repo"] = r
_retry_loop(
operation="git clone",
attempts=attempts,
sleep=sleep,
logger=logger,
action=_do_clone,
)
if logger is not None:
logger.info(
"blog_clone_complete",
extra={"event": "blog_clone_complete", "path": str(target)},
)
return cloned
if "repo" in cloned_holder:
return cloned_holder["repo"]
return Repo(str(target))
try:
repo = Repo(str(target))
@@ -151,13 +191,28 @@ def ensure_repo(
fetch = (fetch_impl or _default_fetch)(repo)
pull = (pull_impl or _default_pull)(repo)
_fetch_and_pull(
repo,
fetch=fetch,
pull=pull,
def _do_pull() -> None:
try:
fetch()
pull()
except GitCommandError as exc:
if _is_local_modification_error(exc):
if logger is not None:
logger.error(
"blog_local_modifications",
extra={"event": "blog_local_modifications", "error": str(exc)},
)
raise BlogRepoError(
"local modifications detected in blog working tree; aborting run"
) from exc
raise BlogTransientError(str(exc), operation="git pull") from exc
_retry_loop(
operation="git pull",
attempts=attempts,
sleep=sleep,
max_retries=retries,
logger=logger,
action=_do_pull,
)
return repo
@@ -171,4 +226,4 @@ def _default_fetch(repo: Repo) -> Callable[[], object]:
def _default_pull(repo: Repo) -> Callable[[], object]:
def _do() -> object:
return repo.git.pull("--ff-only")
return _do
return _do
src/tenbackward/config.py
+47 -19
View File
@@ -13,10 +13,10 @@ from dotenv import dotenv_values, load_dotenv
REQUIRED_KEYS = (
"MASTODON_BASE_URL",
"MASTODON_ACCESS_TOKEN",
"VISIBILITY",
"MASTODON_VISIBILITY",
"SITE_URL",
"HASHTAGS",
"THROWNBACK_PREFIX",
"THROWBACK_PREFIX",
"MAX_RETRIES",
"RUN_AT",
"TZ",
@@ -25,11 +25,7 @@ REQUIRED_KEYS = (
OPTIONAL_KEYS: tuple[str, ...] = ("BLOG_REPO_URL", "BLOG_DIR")
DEFAULTS = {
"VISIBILITY": "public",
"THROWNBACK_PREFIX": "Heute vor 10 Jahren:",
"MAX_RETRIES": "3",
"TZ": "Europe/Berlin",
"RUN_AT": "09:00",
"MAX_RETRIES": "5",
"BLOG_REPO_URL": "https://git.chaospott.de/Chaospott/site",
"BLOG_DIR": "blog",
}
@@ -46,7 +42,7 @@ class ConfigError(ValueError):
class Config:
mastodon_base_url: str
mastodon_access_token: str
visibility: str
mastodon_visibility: str
site_url: str
hashtags: str
throwback_prefix: str
@@ -67,6 +63,9 @@ class Config:
return self.run_at.split(":", 1)[1]
DEFAULT_DOTENV_PATH = Path("/app/.env")
def _read_dotenv(dotenv_path: Optional[Path]) -> dict[str, str]:
if dotenv_path is None:
return {}
@@ -88,19 +87,38 @@ def _values_from_env() -> dict[str, str]:
def load_config(dotenv_path: Optional[Path] = None) -> Config:
"""Load configuration from a dotenv file and the process environment.
Environment variables take precedence over the dotenv file so a mounted
`.env` can be supplemented by Compose-level overrides.
The .env file is loaded via python-dotenv at startup so the bind-mounted
runtime secrets are visible to the process. Environment variables take
precedence over the dotenv file so a mounted `.env` can be supplemented
by Compose-level overrides.
Required keys are validated against the **raw** merged map (dotenv +
process env) before any defaults are applied. Defaults are only used
for the retry budget and the optional Jekyll blog source so a missing
Mastodon/secret/URL/etc. fails fast with a clear error.
"""
if dotenv_path is None:
dotenv_path = DEFAULT_DOTENV_PATH if DEFAULT_DOTENV_PATH.exists() else None
if dotenv_path is not None:
load_dotenv(dotenv_path=str(dotenv_path), override=False)
file_values = _read_dotenv(dotenv_path)
env_values = _values_from_env()
merged: dict[str, str] = {}
merged.update(file_values)
merged.update(env_values)
raw: dict[str, str] = {}
raw.update(file_values)
raw.update(env_values)
# MAX_RETRIES is the only required key that has a permitted default
# (5 attempts per the Job). Apply it before the required-key check so a
# missing value uses the documented default rather than failing fast.
if not raw.get("MAX_RETRIES"):
raw["MAX_RETRIES"] = DEFAULTS["MAX_RETRIES"]
validate_required(raw)
merged = dict(raw)
apply_defaults(merged)
validate_config(merged)
@@ -119,10 +137,10 @@ def load_config(dotenv_path: Optional[Path] = None) -> Config:
return Config(
mastodon_base_url=merged["MASTODON_BASE_URL"],
mastodon_access_token=merged["MASTODON_ACCESS_TOKEN"],
visibility=merged["VISIBILITY"],
mastodon_visibility=merged["MASTODON_VISIBILITY"],
site_url=merged["SITE_URL"],
hashtags=merged["HASHTAGS"],
throwback_prefix=merged["THROWNBACK_PREFIX"],
throwback_prefix=merged["THROWBACK_PREFIX"],
max_retries=_parse_max_retries(merged["MAX_RETRIES"]),
run_at=merged["RUN_AT"],
tz=merged["TZ"],
@@ -138,13 +156,23 @@ def apply_defaults(values: dict[str, str]) -> None:
values.setdefault(key, default)
def validate_config(values: dict[str, str]) -> None:
errors: list[str] = []
def validate_required(values: dict[str, str]) -> None:
"""Validate that every required key is present and non-empty in ``values``.
Called against the raw (no-default) merged map so a missing key always
surfaces as a startup error rather than being silently substituted.
"""
errors: list[str] = []
for key in REQUIRED_KEYS:
raw = values.get(key, "")
if not raw:
errors.append(f"missing required configuration key: {key}")
if errors:
raise ConfigError("; ".join(errors))
def validate_config(values: dict[str, str]) -> None:
errors: list[str] = []
run_at = values.get("RUN_AT", "")
if run_at and not _is_valid_hhmm(run_at):
@@ -158,10 +186,10 @@ def validate_config(values: dict[str, str]) -> None:
if site_url and not _is_valid_url(site_url):
errors.append(f"SITE_URL={site_url!r} must be a valid http(s) URL")
visibility = values.get("VISIBILITY", "")
visibility = values.get("MASTODON_VISIBILITY", "")
if visibility and not _is_valid_visibility(visibility):
errors.append(
f"VISIBILITY={visibility!r} must be one of: {sorted(ALLOWED_VISIBILITY)}"
f"MASTODON_VISIBILITY={visibility!r} must be one of: {sorted(ALLOWED_VISIBILITY)}"
)
blog_repo_url = values.get("BLOG_REPO_URL", "")
src/tenbackward/logging_setup.py
+13
View File
@@ -139,6 +139,18 @@ def log_error(event: str, *, exc: BaseException | None = None, **fields: Any) ->
logging.getLogger(_LOGGER_NAME).error(event, extra=payload)
def log_event(event: str, **fields: Any) -> None:
"""Emit a structured event line at the level appropriate to ``event``.
Unlike :func:`log_error`, this helper does not attach ``exc_type``; it
forwards arbitrary fields (including ``error=str(exc)``) through the
same redaction path so callers can include the underlying exception
text without leaking secret-shaped keys.
"""
payload = _redact_dict({"event": event, **fields})
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})
@@ -156,6 +168,7 @@ __all__ = [
"JsonFormatter",
"configure_json_logging",
"log_error",
"log_event",
"log_run_summary",
"log_startup",
"log_warning",
src/tenbackward/main.py
+93 -46
View File
@@ -1,23 +1,28 @@
from __future__ import annotations
import logging
import sys
import time
from typing import Iterable
from typing import Callable, Iterable
from . import __version__
from .blog import BlogRepoError, ensure_repo
from .blog import BlogRepoError, BlogTransientError, ensure_repo
from .config import Config, ConfigError, load_config
from .logging_setup import (
configure_json_logging,
log_error,
log_event,
log_run_summary,
log_startup,
)
from .matching import MatchedPost, find_anniversary_matches
from .publishing import publish_mastodon
from .publishing import PublishError, PublishFatalError, publish_mastodon
from .state import PostedStore
_logger = logging.getLogger("tenbackward")
def _iter_candidates(config: Config) -> Iterable[MatchedPost]:
"""Yield candidate :class:`MatchedPost` objects whose anniversary is
exactly 10 years before today.
@@ -26,16 +31,97 @@ def _iter_candidates(config: Config) -> Iterable[MatchedPost]:
return find_anniversary_matches(post_root, config.site_url)
def _run_once(config: Config) -> tuple[int, int, int, int, list[str]]:
def _log_retry_attempt(operation: str, attempt: int, exc: BaseException) -> None:
log_event(
"retry_attempt",
operation=operation,
attempt=attempt,
error=str(exc),
exc_type=type(exc).__name__,
)
def _log_final_failure(operation: str, exc: BaseException, attempts: int) -> None:
log_error(
"retry_exhausted",
exc=exc,
operation=operation,
attempts=attempts,
error=str(exc),
)
def _operation_label(exc: BaseException) -> str:
if isinstance(exc, BlogTransientError):
return exc.operation
if isinstance(exc, PublishError):
return "mastodon post"
if isinstance(exc, BlogRepoError):
return "git pull"
return "pipeline"
def _run_with_retry(
config: Config,
*,
sleep: Callable[[float], None] = time.sleep,
logger: logging.Logger | None = None,
) -> tuple[int, int, int, int, list[str]] | None:
"""Execute the pipeline with retries. Returns the summary counters on
success or ``None`` when the retry budget is exhausted.
A single retry budget governs both the Git and Mastodon operations —
``ensure_repo`` is invoked with ``max_retries=0`` so the runner, not the
blog layer, decides whether to re-attempt. Configuration and fatal
validation errors (``ConfigError``, :class:`BlogRepoError`,
:class:`PublishFatalError`) are not retried.
"""
attempts = max(1, config.max_retries)
active_logger = logger or _logger
last_exc: BaseException | None = None
for attempt in range(1, attempts + 1):
try:
return _run_once(config, logger=active_logger, sleep=sleep)
except (ConfigError, BlogRepoError, PublishFatalError) as exc:
last_exc = exc
_log_final_failure(type(exc).__name__, exc, attempt)
return None
except (BlogTransientError, PublishError) as exc:
last_exc = exc
_log_retry_attempt(_operation_label(exc), attempt, exc)
if attempt < attempts:
sleep(2 ** (attempt - 1))
except Exception as exc: # noqa: BLE001 — opportunistic retry boundary
last_exc = exc
_log_retry_attempt("pipeline", attempt, exc)
if attempt < attempts:
sleep(2 ** (attempt - 1))
assert last_exc is not None
_log_final_failure(_operation_label(last_exc), last_exc, attempts)
return None
def _run_once(
config: Config,
*,
logger: logging.Logger | None = None,
sleep: Callable[[float], None] = time.sleep,
) -> tuple[int, int, int, int, list[str]]:
"""Execute one pipeline pass and return the summary counters.
Returns ``(scanned, matched, posted, skipped, posted_ids)``.
The blog layer's retry budget is disabled here (``max_retries=0``);
the runner owns the unified retry policy so we never nest two retry
budgets back-to-back.
"""
ensure_repo(
config.data_dir,
repo_url=config.blog_repo_url,
blog_path=config.blog_dir,
max_retries=config.max_retries,
max_retries=0,
logger=logger,
sleep=sleep,
)
store = PostedStore(config.data_dir)
@@ -61,35 +147,6 @@ def _run_once(config: Config) -> tuple[int, int, int, int, list[str]]:
return scanned, matched, posted, skipped, posted_ids
def _now_iso() -> str:
from datetime import datetime, timezone
return datetime.now(tz=timezone.utc).isoformat()
def _run_with_retry(config: Config) -> tuple[int, int, int, int, list[str]] | None:
"""Execute the pipeline with retries. Returns the summary counters on
success or ``None`` when the retry budget is exhausted."""
attempts = max(1, config.max_retries + 1)
last_exc: BaseException | None = None
for attempt in range(1, attempts + 1):
try:
return _run_once(config)
except Exception as exc: # noqa: BLE001 — broad on purpose
last_exc = exc
log_error(
"pipeline_error",
exc=exc,
attempt=attempt,
max_attempts=attempts,
)
if attempt < attempts:
time.sleep(0)
assert last_exc is not None
log_error("pipeline_failed", exc=last_exc, attempts=attempts)
return None
def main() -> int:
configure_json_logging()
@@ -100,17 +157,7 @@ def main() -> int:
return 2
config.data_dir.mkdir(parents=True, exist_ok=True)
try:
ensure_repo(
config.data_dir,
repo_url=config.blog_repo_url,
blog_path=config.blog_dir,
max_retries=config.max_retries,
)
except BlogRepoError as exc:
log_error("blog_repo_error", exc=exc)
return 1
config.blog_dir.mkdir(parents=True, exist_ok=True)
log_startup(
version=__version__,
@@ -121,7 +168,7 @@ def main() -> int:
throwback_prefix=config.throwback_prefix,
)
result = _run_with_retry(config)
result = _run_with_retry(config, logger=_logger)
if result is None:
return 1
src/tenbackward/publishing.py
+32 -9
View File
@@ -26,6 +26,13 @@ class PublishError(RuntimeError):
"""
class PublishFatalError(PublishError):
"""Raised for fatal configuration or validation failures (missing token,
invalid instance URL, malformed status). The pipeline orchestrator must
not retry these — they will fail the same way on every attempt.
"""
def _normalize_hashtags(hashtags: str) -> str:
parts = [token.strip() for token in (hashtags or "").split(",")]
parts = [token for token in parts if token]
@@ -57,7 +64,7 @@ def build_status_text(
regardless of the upstream ordering.
"""
if not posts:
raise PublishError("empty_posts: cannot compose status without posts")
raise PublishFatalError("empty_posts: cannot compose status without posts")
ordered = sorted(posts, key=lambda m: (m.date, m.path))
blocks: list[str] = []
@@ -73,17 +80,30 @@ def build_status_text(
def validate_status(status: str, limit: int = MASTODON_STATUS_LIMIT) -> None:
"""Raise :class:`PublishError` when ``status`` exceeds ``limit``.
"""Raise :class:`PublishFatalError` when ``status`` exceeds ``limit``.
Never truncates: the spec requires failing safely rather than
shortening content.
"""
if len(status) > limit:
raise PublishError(
raise PublishFatalError(
f"status_too_long: len={len(status)} limit={limit}"
)
def _validate_publish_config(config: Config) -> None:
if not config.mastodon_base_url or not config.mastodon_base_url.startswith(
("http://", "https://")
):
raise PublishFatalError(
f"publish_failed: mastodon auth: invalid MASTODON_BASE_URL={config.mastodon_base_url!r}"
)
if not config.mastodon_access_token:
raise PublishFatalError(
"publish_failed: mastodon auth: MASTODON_ACCESS_TOKEN is empty"
)
def _post_status_via_mastodon_py(
status: str,
*,
@@ -106,10 +126,12 @@ def publish_mastodon(
"""Compose, validate, and publish ``posts`` to Mastodon.
Returns the composed status text on success. Raises
:class:`PublishError` on any failure (composition, length, or API).
The original exception is chained via ``raise ... from exc`` so the
caller can inspect the underlying cause.
:class:`PublishFatalError` for configuration/validation failures (no
retry), or :class:`PublishError` (transient) for network/HTTP errors
that the runner may retry.
"""
_validate_publish_config(config)
try:
status = build_status_text(
config.throwback_prefix, posts, config.hashtags
@@ -119,14 +141,14 @@ def publish_mastodon(
status,
base_url=config.mastodon_base_url,
access_token=config.mastodon_access_token,
visibility=config.visibility,
visibility=config.mastodon_visibility,
client_factory=client_factory,
)
except PublishError:
except PublishFatalError:
raise
except Exception as exc: # noqa: BLE001 — third-party boundary
raise PublishError(
f"publish_failed: {type(exc).__name__}: {exc}"
f"publish_failed: mastodon post: {type(exc).__name__}: {exc}"
) from exc
return status
@@ -134,6 +156,7 @@ def publish_mastodon(
__all__ = [
"MASTODON_STATUS_LIMIT",
"PublishError",
"PublishFatalError",
"build_status_text",
"publish_mastodon",
"slugify_title",