Capture and display per-run transfer failures

app/__init__.py

@@ -111,14 +111,24 @@ def _check_runs():
         with app.app_context():
             job_runner.check_and_update_runs()
 
+    def _collect_errors():
+        with app.app_context():
+            job_runner.collect_active_errors()
+
     monitor.add_job(_check_runs, "interval", seconds=10)
+    # rclone only remembers its last ~100 completed transfers, so failures
+    # must be picked up quickly. Overlap is harmless: entries are de-duplicated.
+    monitor.add_job(
+        _collect_errors, "interval", seconds=3, max_instances=1, coalesce=True
+    )
     if app.config["BACKGROUND_TASKS_ENABLED"]:
         monitor.start()
 
     # Register blueprints
     from app.routes.history import history_bp
     from app.routes.jobs import jobs_bp
     from app.routes.operations import ops_bp
+    from app.routes.run_errors import run_errors_bp
     from app.routes.schedules import schedules_bp
     from app.routes.settings import settings_bp
     from app.routes.sse import sse_bp
@@ -127,6 +137,7 @@ def _check_runs():
     app.register_blueprint(ops_bp, url_prefix="/ops")
     app.register_blueprint(schedules_bp, url_prefix="/schedules")
     app.register_blueprint(history_bp, url_prefix="/history")
+    app.register_blueprint(run_errors_bp, url_prefix="/runs")
     app.register_blueprint(sse_bp)
     app.register_blueprint(settings_bp, url_prefix="/settings")
 

app/models.py

@@ -120,3 +120,48 @@ class RunHistory(db.Model):
     retry_of = db.relationship(
         "RunHistory", remote_side="RunHistory.id", uselist=False
     )
+    # (``errors`` is already the int count of failures rclone reported.)
+    error_log = db.relationship(
+        "RunError", backref="run", lazy="dynamic", cascade="all,delete-orphan"
+    )
+
+
+class RunError(db.Model):
+    """One failed transfer or check captured while a run was in progress.
+
+    rclone's RC API only keeps the most recent ~100 completed transfers, so the
+    runner copies failures here as they happen. Runs can fail thousands of
+    files, hence the indexes for grouping and cursor paging.
+    """
+
+    __table_args__ = (
+        db.UniqueConstraint("run_id", "fingerprint", name="uq_run_error"),
+        db.Index("ix_run_error_run_category", "run_id", "category"),
+    )
+
+    id = db.Column(db.Integer, primary_key=True)
+    run_id = db.Column(
+        db.Integer, db.ForeignKey("run_history.id"), nullable=False, index=True
+    )
+    occurred_at = db.Column(db.DateTime, default=datetime.utcnow)
+    path = db.Column(db.String(1000), nullable=True)
+    kind = db.Column(db.String(10), default="transfer")  # transfer | check
+    size = db.Column(db.BigInteger, nullable=True)
+    message = db.Column(db.Text, nullable=False)
+    # ``message`` with ids/urls/quoted names stripped, so thousands of failures
+    # with the same cause collapse into one group.
+    category = db.Column(db.String(200), nullable=False)
+    fingerprint = db.Column(db.String(40), nullable=False)
+
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "occurred_at": (
+                self.occurred_at.isoformat() + "Z" if self.occurred_at else None
+            ),
+            "path": self.path,
+            "kind": self.kind,
+            "size": self.size,
+            "message": self.message,
+            "category": self.category,
+        }

app/rclone_client.py

@@ -72,8 +72,12 @@ def get_stats(self, group=None):
             params["group"] = group
         return self._call("core/stats", params)
 
-    def get_transferred(self):
-        return self._call("core/transferred")
+    def get_transferred(self, group=None):
+        """Recently completed transfers (rclone keeps only the last ~100)."""
+        params = {}
+        if group:
+            params["group"] = group
+        return self._call("core/transferred", params)
 
     def set_bwlimit(self, rate):
         return self._call("core/bwlimit", {"rate": rate})

app/routes/history.py

@@ -1,6 +1,6 @@
 from flask import Blueprint, render_template
 
-from app.models import RunHistory
+from app.models import RunError, RunHistory
 
 history_bp = Blueprint("history", __name__)
 
@@ -18,4 +18,7 @@ def list_history():
 @history_bp.route("/<int:run_id>")
 def detail(run_id):
     run = RunHistory.query.get_or_404(run_id)
-    return render_template("history/detail.html", run=run)
+    error_count = RunError.query.filter_by(run_id=run.id).count()
+    return render_template(
+        "history/detail.html", run=run, error_count=error_count
+    )

app/routes/run_errors.py

@@ -0,0 +1,148 @@
+import csv
+import io
+
+from flask import Blueprint, Response, jsonify, request, stream_with_context
+from sqlalchemy import func, or_
+
+from app.extensions import db
+from app.models import RunError, RunHistory
+
+run_errors_bp = Blueprint("run_errors", __name__)
+
+DEFAULT_LIMIT = 100
+MAX_LIMIT = 500
+
+
+def _filtered(run_id):
+    """Errors of a run narrowed by the ?category= and ?q= query parameters."""
+    query = RunError.query.filter(RunError.run_id == run_id)
+    category = request.args.get("category")
+    if category:
+        query = query.filter(RunError.category == category)
+    needle = (request.args.get("q") or "").strip()
+    if needle:
+        escaped = (
+            needle.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        )
+        pattern = f"%{escaped}%"
+        query = query.filter(
+            or_(
+                RunError.path.ilike(pattern, escape="\\"),
+                RunError.message.ilike(pattern, escape="\\"),
+            )
+        )
+    return query
+
+
+def _int_arg(name, default=None):
+    try:
+        return int(request.args.get(name, ""))
+    except ValueError:
+        return default
+
+
+@run_errors_bp.route("/<int:run_id>/errors")
+def list_errors(run_id):
+    """Failures of a run as JSON, newest first, paged by id cursors.
+
+    ``?after=<id>`` returns what failed since that id (for the live feed) and
+    ``?before=<id>`` the next older page. Cursors rather than offsets keep
+    paging cheap and stable while new failures keep arriving. With
+    ``?summary=1`` the reply also groups the run's failures by cause.
+    """
+    run = RunHistory.query.get_or_404(run_id)
+    limit = max(1, min(_int_arg("limit", DEFAULT_LIMIT), MAX_LIMIT))
+    after = _int_arg("after")
+    before = _int_arg("before")
+
+    query = _filtered(run.id)
+    matched = query.count() if request.args.get("count") else None
+    if after is not None:
+        query = query.filter(RunError.id > after)
+    if before is not None:
+        query = query.filter(RunError.id < before)
+
+    rows = query.order_by(RunError.id.desc()).limit(limit + 1).all()
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+
+    reply = {
+        "errors": [row.to_dict() for row in rows],
+        "has_more": has_more,
+        "total": RunError.query.filter_by(run_id=run.id).count(),
+        "last_id": db.session.query(func.max(RunError.id))
+        .filter(RunError.run_id == run.id)
+        .scalar()
+        or 0,
+        "status": run.status,
+    }
+    if matched is not None:
+        reply["matched"] = matched
+    if request.args.get("summary"):
+        reply["categories"] = _categories(run.id)
+    return jsonify(reply)
+
+
+def _categories(run_id):
+    groups = (
+        db.session.query(
+            RunError.category,
+            func.count(RunError.id),
+            func.min(RunError.occurred_at),
+            func.max(RunError.occurred_at),
+        )
+        .filter(RunError.run_id == run_id)
+        .group_by(RunError.category)
+        .order_by(func.count(RunError.id).desc(), RunError.category)
+        .limit(100)
+        .all()
+    )
+    return [
+        {
+            "category": category,
+            "count": count,
+            "first": first.isoformat() + "Z" if first else None,
+            "last": last.isoformat() + "Z" if last else None,
+        }
+        for category, count, first, last in groups
+    ]
+
+
+def _spreadsheet_safe(value):
+    """Stop file names like '=cmd|...' from being run as formulas when opened."""
+    text = "" if value is None else str(value)
+    return "'" + text if text[:1] in ("=", "+", "-", "@", "\t", "\r") else text
+
+
+@run_errors_bp.route("/<int:run_id>/errors.csv")
+def export_errors(run_id):
+    """Every stored failure of a run (honouring filters) as a CSV download."""
+    run = RunHistory.query.get_or_404(run_id)
+    query = _filtered(run.id).order_by(RunError.id)
+
+    def generate():
+        buffer = io.StringIO()
+        writer = csv.writer(buffer)
+        writer.writerow(["time_utc", "kind", "path", "size", "message"])
+        yield buffer.getvalue()
+        for row in query.yield_per(500):
+            buffer.seek(0)
+            buffer.truncate()
+            writer.writerow(
+                [
+                    row.occurred_at.isoformat() if row.occurred_at else "",
+                    row.kind,
+                    _spreadsheet_safe(row.path),
+                    row.size if row.size is not None else "",
+                    _spreadsheet_safe(row.message),
+                ]
+            )
+            yield buffer.getvalue()
+
+    return Response(
+        stream_with_context(generate()),
+        mimetype="text/csv",
+        headers={
+            "Content-Disposition": f'attachment; filename="run-{run.id}-errors.csv"'
+        },
+    )

app/services/job_runner.py

@@ -3,6 +3,7 @@
 
 from app.extensions import db
 from app.models import Job, RunHistory
+from app.services.run_errors import record_failures
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +71,7 @@ def stop_job(self, run):
                 pass
 
         # Capture final stats before marking stopped
+        self.collect_errors(run)
         self._save_final_stats(run)
 
         run.status = "stopped"
@@ -93,19 +95,45 @@ def _parse_rclone_time(self, time_str):
     def _save_final_stats(self, run):
         """Fetch per-group stats from rclone and persist them on the run."""
         if not run.stats_group:
-            return
+            return None
         try:
             stats = self.rclone.get_stats(group=run.stats_group)
             run.bytes_transferred = stats.get("bytes", 0)
             run.files_transferred = stats.get("transfers", 0)
             run.errors = stats.get("errors", 0)
+            return stats
         except Exception as exc:
             logger.warning(
                 "Failed to fetch final stats for Run #%d (group=%s): %s",
                 run.id,
                 run.stats_group,
                 exc,
             )
+        return None
+
+    def collect_errors(self, run):
+        """Copy failed transfers/checks of a run out of rclone into the DB."""
+        if not run.stats_group:
+            return 0
+        try:
+            reply = self.rclone.get_transferred(group=run.stats_group)
+            return record_failures(run, (reply or {}).get("transferred"))
+        except Exception as exc:
+            db.session.rollback()
+            logger.warning(
+                "Failed to collect errors for Run #%d (group=%s): %s",
+                run.id,
+                run.stats_group,
+                exc,
+            )
+            return 0
+
+    def collect_active_errors(self):
+        """Poll every tracked run; called often since rclone forgets old entries."""
+        for run_id in list(self._active_runs):
+            run = db.session.get(RunHistory, run_id)
+            if run:
+                self.collect_errors(run)
 
     def check_and_update_runs(self):
         for run_id, rclone_jobid in list(self._active_runs.items()):
@@ -131,7 +159,18 @@ def check_and_update_runs(self):
                 if status.get("error"):
                     run.error_message = str(status["error"])
 
-                self._save_final_stats(run)
+                self.collect_errors(run)
+                stats = self._save_final_stats(run)
+                # Failures not tied to a file (listing, deleting, ...) only show
+                # up as rclone's last error; keep it if nothing else explains
+                # the failure.
+                if (
+                    not success
+                    and not run.error_message
+                    and stats
+                    and stats.get("lastError")
+                ):
+                    run.error_message = str(stats["lastError"])
 
                 db.session.commit()
                 del self._active_runs[run_id]