spencermandClaude Opus 5.5
6adaf7bf25
FILE checks the archive folder and copies the packet inside the filing transaction. A failed check or copy rolls the filing back, so the packet stays in its previous stage; the person filing gets 503 archive_failed, and in its own transaction the server writes an archive.filing_blocked audit row and raises the critical archive_failed condition to Admin for that packet (who tried, folder, error). The next successful filing clears it. The archive-copy job now only catches up filings made while the archive was off. PLAN question 53, CHANGELOG, FEED and SECURITY updated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
104 lines
4.6 KiB
Python
104 lines
4.6 KiB
Python
"""The archive copy (Gitea issue 7) for seals filed while the archive was off, every 15 minutes and on request.
|
|
|
|
FILE copies its own packet in the filing transaction and refuses to file when the copy fails (Gitea issue 57; see
|
|
domain/archive.py). This job catches up the rest: off when ARCHIVE_ROOT is unset, and idle until the Admin saves a
|
|
folder. Each run checks the folder (the probe), queues every unbroken seal with a data book that has no
|
|
`archive_copies` row (so turning the archive on later copies earlier filings too), then copies the pending rows and
|
|
the failed rows that are due. A failed copy waits 5, 15, 60 minutes, then an hour between tries. The `archive_failed`
|
|
condition (subject `archive`) is raised when a copy has failed three times or the folder check fails, and cleared when
|
|
no copy is failed. These packets are already filed, so a failed copy here never touches the packet.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from ...domain import archive
|
|
from ...feed.config import FeedSettings
|
|
from .. import jobs
|
|
from . import Counts, JobContext, JobSpec
|
|
|
|
NAME = archive.JOB
|
|
INTERVAL_SECONDS = 15 * 60
|
|
BATCH = 200
|
|
TITLE = "Archive copy failing"
|
|
|
|
|
|
def _check_changed(ctx: JobContext, result: archive.CheckResult, archive_dir: str) -> None:
|
|
"""Audit the job's own folder check only when its result differs from the last check on record."""
|
|
last = archive.last_check(ctx.conn)
|
|
if last is None or bool(last["ok"]) != result.ok or last.get("archive_dir") != archive_dir:
|
|
archive.audit_check(ctx.conn, result, archive_dir=archive_dir, source="job")
|
|
|
|
|
|
def run(ctx: JobContext) -> Counts:
|
|
root = archive.root_of(ctx.settings)
|
|
counts: Counts = {"enabled": root is not None, "queued": 0, "copied": 0, "failed": 0}
|
|
if root is None:
|
|
return counts
|
|
archive_dir, _, _, _ = archive.load(ctx.conn)
|
|
counts["archive_dir"] = archive_dir
|
|
if archive_dir is None:
|
|
counts["skipped"] = "no archive folder saved"
|
|
return counts
|
|
rel, result = archive.check(root, archive_dir)
|
|
_check_changed(ctx, result, archive_dir)
|
|
if not result.ok or rel is None:
|
|
counts["check"] = result.message
|
|
ctx.error(f"archive folder check failed: {result.message}")
|
|
jobs.raise_event(
|
|
ctx.conn,
|
|
archive.EVENT,
|
|
subject=archive.EVENT_SUBJECT,
|
|
title=TITLE,
|
|
body=f"The archive folder check failed: {result.message}",
|
|
detail={"archive_dir": archive_dir, "message": result.message},
|
|
severity="critical",
|
|
condition=True,
|
|
)
|
|
return counts
|
|
base = root.joinpath(*rel.split("/"))
|
|
with ctx.conn.transaction():
|
|
counts["queued"] = archive.queue_missing(ctx.conn)
|
|
for row in archive.due_rows(ctx.conn, BATCH):
|
|
try:
|
|
folder, sha, files = archive.copy_seal(ctx.conn, ctx.settings, base, rel, int(row["seal_id"]))
|
|
except archive.CopyError as e:
|
|
attempts = archive.mark_failed(ctx.conn, row, str(e))
|
|
counts["failed"] += 1
|
|
ctx.error(f"seal {row['seal_id']}: {e} (try {attempts})", degrade=True)
|
|
continue
|
|
except Exception as e: # anything unexpected is one failed copy, never a stopped run
|
|
attempts = archive.mark_failed(ctx.conn, row, f"{type(e).__name__}: {e}")
|
|
counts["failed"] += 1
|
|
ctx.log.exception("archive copy of seal %s failed", row["seal_id"])
|
|
ctx.error(f"seal {row['seal_id']}: {type(e).__name__} (try {attempts})", degrade=True)
|
|
continue
|
|
archive.mark_done(ctx.conn, row, folder, sha, files)
|
|
counts["copied"] += 1
|
|
st = archive.copies_status(ctx.conn)
|
|
counts.update(pending=st["pending"], failed_total=st["failed"], done_total=st["done"])
|
|
stuck = archive.stuck_failures(ctx.conn)
|
|
if stuck:
|
|
lines = [f"{f['packet_label']} (seal {f['seal_id']}): {f['last_error']}" for f in stuck[:10]]
|
|
jobs.raise_event(
|
|
ctx.conn,
|
|
archive.EVENT,
|
|
subject=archive.EVENT_SUBJECT,
|
|
title=TITLE,
|
|
body=f"{len(stuck)} filed packet(s) could not be copied to the archive:\n" + "\n".join(lines),
|
|
detail={"failed": st["failed"], "seal_ids": [f["seal_id"] for f in stuck]},
|
|
severity="critical",
|
|
condition=True,
|
|
)
|
|
elif st["failed"] == 0:
|
|
jobs.resolve_event(ctx.conn, archive.EVENT, subject=archive.EVENT_SUBJECT, title=TITLE)
|
|
return counts
|
|
|
|
|
|
def cadence(_: FeedSettings, now_site: datetime) -> int:
|
|
return INTERVAL_SECONDS
|
|
|
|
|
|
SPEC = JobSpec(name=NAME, run=run, cadence=cadence, pausable=False)
|