Release 3.2.9: fix telemetry spool reclamation and update dashboard dependencies

.env.example

@@ -2,7 +2,7 @@
 # Nonempty GETBIBLE_* application settings override values saved by the menu.
 # Empty optional values leave the saved setting/default in control.
 GETBIBLE_IMAGE_REPOSITORY=ghcr.io/getbible/api
-GETBIBLE_IMAGE_TAG=3.2.8
+GETBIBLE_IMAGE_TAG=3.2.9
 GETBIBLE_DATA_ROOT=/srv/getbible-data
 GETBIBLE_HOSTNAME=getbible
 TZ=UTC

Dockerfile

@@ -1,7 +1,7 @@
 # syntax=docker/dockerfile:1
 FROM ubuntu:24.04
 
-ARG GETBIBLE_VERSION=3.2.8
+ARG GETBIBLE_VERSION=3.2.9
 ARG GETBIBLE_REVISION=unknown
 ARG GETBIBLE_IMAGE_PYTHONS=
 LABEL org.opencontainers.image.title="getBible API" \

VERSION

@@ -1 +1 @@
-3.2.8
+3.2.9

compose.yaml

@@ -4,7 +4,7 @@ services:
   getbible:
     # Production deployment only pulls the prebuilt image. No checkout
     # or Dockerfile is required on the server. See docs/DOCKER.md for login.
-    image: ${GETBIBLE_IMAGE_REPOSITORY:-ghcr.io/getbible/api}:${GETBIBLE_IMAGE_TAG:-3.2.8}
+    image: ${GETBIBLE_IMAGE_REPOSITORY:-ghcr.io/getbible/api}:${GETBIBLE_IMAGE_TAG:-3.2.9}
     restart: unless-stopped
     hostname: ${GETBIBLE_HOSTNAME:-getbible}
     user: "0:0"

docs/LOGGING.md

@@ -256,11 +256,31 @@ reminder or sustained recovery rather than resending an unchanged message at
 every short cooldown. Missing/stale measurements never imply recovery. Inspect
 open incidents and the collector journal when capacity is still saturated.
 
+If unread bytes are zero while committed closed transport occupies the allowance,
+ingestion has caught up and reclamation needs attention. The capacity report shows
+these bytes separately and labels reclamation as pending, blocked or stale. The
+last `spool_cleanup` observation records attempted/removed/retained bytes, reason
+counts and bounded failure details including errno. An open writer points to a
+producer that has not closed its rotated file; inspect nginx reopen failures.
+The collector retries nginx reopening for writer-held nginx archives. For an idle
+runtime producer, Gunicorn's USR1 handling reopens application log files in its
+master and workers. Administrative event producers close their append streams
+after each event, so idle dashboard auditing no longer holds rotated files open.
+Unsupported leases or denied permissions point to the mounted filesystem or
+service identity rather than a need for a larger budget. Inspect current evidence
+with `getbible capacity --json` and `journalctl -u getbible-telemetry --since '1 hour ago'`.
+Safe cleanup retries automatically; do not delete a transport file merely because
+the unread count is zero, since a producer may still append to it.
+
 Recommendations use measured demand/high-water usage, saturation observations and
 available headroom, and explain their headroom assumptions. A growing or stalled
 unread backlog requires fixing collection before relying on a larger spool
 allowance. Advice is withheld when evidence/headroom is insufficient; no limit is
 automatically raised and no unread history is discarded to silence an alarm.
+Transport sizing uses the peak active/pending demand after subtracting committed
+closed spools; the total occupied high-water remains visible as pressure evidence.
+Older observations that lack this distinction do not justify an increased
+allowance: sizing waits for at least 60 observed seconds and six new samples.
 Configured values controlled by deployment environment or host cgroups are shown
 as such. Collector restart retains the incident, while real recovery clears it
 after the configured stability interval.

src/apps/dashboard/ui/package.json

@@ -17,10 +17,10 @@
     "react-dom": "19.3.0"
   },
   "devDependencies": {
-    "@vitejs/plugin-react": "6.1.1",
+    "@vitejs/plugin-react": "6.1.2",
     "bootstrap": "5.3.8",
     "echarts": "6.1.0",
-    "vite": "8.3.2"
+    "vite": "8.3.3"
   },
   "engines": {
     "node": ">=22.12"

src/apps/dashboard/ui/src/capacity.jsx

@@ -28,6 +28,8 @@ export default function Capacity({refresh, execute}) {
                 <div><span>Collection</span><strong>{stale ? 'Unknown — observations are not current' : words(collection.state)}</strong></div>
                 <div><span>Unread backlog</span><strong>{bytes(collection.unread_bytes)} · {number(collection.unread_files)} files</strong></div>
                 <div><span>Transport / retained archives</span><strong>{bytes(collection.budgeted_spool_bytes)} / {bytes(collection.retained_archive_bytes)}</strong></div>
+                <div><span>Committed closed transport</span><strong>{bytes(collection.consumed_rotated_bytes)} · already ingested</strong></div>
+                <div><span>Reclamation</span><div className="text-end"><strong>{stale ? 'Unknown — observations are not current' : words(collection.reclamation?.state)}</strong><small className="d-block">{collection.reclamation?.reason}</small></div></div>
                 <div><span>Producer / collector throughput</span><strong>{rate(collection.producer_bytes_per_second)} / {rate(collection.collector_bytes_per_second)}</strong></div>
                 <div><span>Backlog growth / continuous backlog</span><strong>{rate(collection.backlog_growth_bytes_per_second)} / {number(collection.backlog_observed_seconds)} seconds</strong></div>
                 <div><span>Rate window / peak window</span><strong>{number(collection.rate_window_seconds)} seconds / {number(report.window_seconds / 3600)} hours</strong></div>

src/apps/telemetry/getbible_telemetry/capacity.py

@@ -17,6 +17,38 @@
 HEADROOM = .25
 
 
+def reclamation_status(spool: dict[str, Any], *, now: float | None = None) -> dict[str, Any]:
+    """Explain retained committed transport independently of unread collection."""
+    now = time.time() if now is None else now
+    committed = finite(spool.get("consumed_rotated_bytes"))
+    cleanup = spool.get("problems", {}).get("spool_cleanup", {})
+    reasons = cleanup.get("reason_counts") or {reason: None for reason in cleanup.get("reasons", [])}
+    explanations = {
+        "writer_open": "A producer still has a committed spool open; inspect nginx log reopening and idle runtime producers.",
+        "writer_open_or_lease_unavailable": "A writer is open or the filesystem refused the safety lease.",
+        "lease_unsupported": "The filesystem does not support the safety lease required for reclamation.",
+        "permission_denied": "Lease or removal permission was denied; inspect service identity and mount ownership.",
+        "filesystem_error": "The filesystem refused reclamation; inspect the recorded errno and collector journal.",
+        "source_changed": "A source changed after collection; it is preserved for another ingestion pass.",
+        "source_changed_or_opening": "A source changed or a writer reopened it during reclamation.",
+    }
+    if committed is None:
+        state, reason = "unknown", "Committed transport usage has not been measured."
+    elif not committed:
+        state, reason = "caught_up", "No committed closed transport spools remain."
+    elif finite(cleanup.get("time")) is not None and now - cleanup["time"] > 60:
+        state, reason = "stale", "The last reclamation attempt is over 60 seconds old; inspect the collector journal."
+    elif cleanup.get("retained") or cleanup.get("error"):
+        state = "blocked"
+        reason = " ".join(explanations.get(key, key.replace("_", " ")) for key in reasons)
+        reason = reason or "The reclamation helper failed; inspect spool_cleanup and the collector journal."
+    else:
+        state, reason = "pending", "Committed closed transport spools await a safe cleanup pass."
+    return {"state": state, "reason": reason, "attempted_at": cleanup.get("time"),
+            "removed_bytes": cleanup.get("removed_bytes"), "retained_bytes": cleanup.get("retained_bytes"),
+            "reason_counts": reasons}
+
+
 def finite(value: Any) -> float | None:
     if isinstance(value, bool) or not isinstance(value, (int, float)):
         return None
@@ -121,6 +153,13 @@ def observe(self, sample: dict[str, Any], spool: dict[str, Any], *, now: float |
             record["peak"] = max(record["peak"], value)
             record["samples"] += 1
             record["seconds"] += duration
+            if key == "telemetry_spool" and finite(spool.get("consumed_rotated_bytes")) is not None:
+                # Committed transport occupying disk is reclamation pressure,
+                # not evidence that producers need a larger transport allowance.
+                record["sizing_peak"] = max(record.get("sizing_peak", 0),
+                                             max(0, value - spool["consumed_rotated_bytes"]))
+                record["sizing_samples"] = record.get("sizing_samples", 0) + 1
+                record["sizing_seconds"] = record.get("sizing_seconds", 0) + duration
             if above:
                 record["saturated_seconds"] += duration
                 record["saturated_samples"] += 1
@@ -135,7 +174,12 @@ def observe(self, sample: dict[str, Any], spool: dict[str, Any], *, now: float |
                    "saturated_seconds": sum(item["saturated_seconds"] for item in aggregate),
                    "saturated_samples": sum(item["saturated_samples"] for item in aggregate),
                    "episodes": sum(item["episodes"] for item in aggregate), "saturation_threshold": .9}
-            row["recommendation"] = self._recommend(row, sample, spool)
+            if key == "telemetry_spool":
+                sizing = [item for item in aggregate if "sizing_peak" in item]
+                row.update(sizing_high_water=max((item["sizing_peak"] for item in sizing), default=0) / scale,
+                           sizing_samples=sum(item["sizing_samples"] for item in sizing),
+                           sizing_observed_seconds=sum(item["sizing_seconds"] for item in sizing))
+            row["recommendation"] = self._recommend(row, sample, spool, now=now)
             current.append(row)
         rates = {"producer_bytes_per_second": None, "collector_bytes_per_second": None,
                  "backlog_growth_bytes_per_second": None, "collector_records_per_second": None}
@@ -175,6 +219,8 @@ def observe(self, sample: dict[str, Any], spool: dict[str, Any], *, now: float |
                   "limits": current, "collection": {"state": state, "backlog_observed_seconds": backlog_age,
                   "unread_bytes": spool.get("unread_bytes"), "unread_files": spool.get("unread_files"),
                   "budgeted_spool_bytes": usage, "retained_archive_bytes": spool.get("retained_archive_bytes"),
+                  "consumed_rotated_bytes": spool.get("consumed_rotated_bytes"),
+                  "reclamation": reclamation_status(spool, now=now),
                   "compressed_pending_bytes": spool.get("compressed_pending_bytes"), "rate_window_seconds": 300,
                   "rate_samples": len(observations), **rates}, "problems": spool.get("problems", {}),
                   "note": "Suggested values use observed peak / 0.75. Capped demand is a lower bound, not an ideal-value guarantee. No setting is changed automatically."}
@@ -187,20 +233,30 @@ def observe(self, sample: dict[str, Any], spool: dict[str, Any], *, now: float |
         return result
 
     @staticmethod
-    def _recommend(row: dict[str, Any], sample: dict[str, Any], spool: dict[str, Any]) -> dict[str, Any]:
-        if row["observed_seconds"] < 60 or row["samples"] < 6:
+    def _recommend(row: dict[str, Any], sample: dict[str, Any], spool: dict[str, Any],
+                   *, now: float | None = None) -> dict[str, Any]:
+        if row["id"] == "telemetry_spool":
+            if spool.get("consumed_rotated_bytes", 0) > spool.get("spool_max_bytes", GIB) * .25:
+                cleanup = reclamation_status(spool, now=now)
+                return {"status": "investigate_cleanup", "value": None,
+                        "reason": "Committed transport occupies the allowance but is already ingested. "
+                                  + cleanup["reason"] + " Inspect spool_cleanup before increasing the budget."}
+            seconds = row.get("sizing_observed_seconds", 0)
+            samples = row.get("sizing_samples", 0)
+        else:
+            seconds, samples = row["observed_seconds"], row["samples"]
+        if seconds < 60 or samples < 6:
             return {"status": "insufficient_data", "value": None,
-                    "reason": "Collect at least 60 observed seconds and six samples before sizing."}
+                    "reason": "Collect at least 60 observed seconds and six samples before sizing."
+                              + (" Transport sizing requires fresh measurements separating committed spools from demand."
+                                 if row["id"] == "telemetry_spool" else "")}
         if row["setting"] is None:
             return {"status": "operator_review", "value": None,
                     "reason": "This is a host/filesystem measurement, not an adjustable application allowance. Review physical headroom and workload."}
-        desired = max(row["effective_limit"], row["high_water"] / (1 - HEADROOM))
+        peak = row.get("sizing_high_water", row["high_water"])
+        desired = max(row["effective_limit"], peak / (1 - HEADROOM))
         quantum = .1 if row["unit"] == "cores" else .25 if row["id"] == "memory" else 1
         desired = math.ceil(desired / quantum) * quantum
-        if row["id"] == "telemetry_spool":
-            if spool.get("consumed_rotated_bytes", 0) > spool.get("spool_max_bytes", GIB) * .25 and spool.get("problems", {}).get("spool_cleanup", {}).get("retained", 0):
-                return {"status": "investigate_cleanup", "value": None,
-                        "reason": "Committed transport files are awaiting safe closure or filesystem lease support; inspect spool_cleanup before increasing the budget."}
         if row["id"] in {"telemetry_spool", "telemetry_history"}:
             disks = sample.get("disks", [])
             if not disks:
@@ -215,7 +271,9 @@ def _recommend(row: dict[str, Any], sample: dict[str, Any], spool: dict[str, Any
                 return {"status": "headroom_unknown" if available is None else "insufficient_headroom", "value": None,
                         "reason": "An increased container memory limit needs confirmed host headroom; reserve at least half of currently available host memory."}
         return {"status": "adequate" if desired <= row["effective_limit"] else "suggested",
-                "value": round(desired, 2), "reason": "Observed 24-hour peak with 25% headroom; never automatically lowers the current allowance.",
+                "value": round(desired, 2), "reason": "Observed 24-hour peak with 25% headroom; never automatically lowers the current allowance."
+                    + (" Already committed closed transport is excluded from sizing demand."
+                       if row["id"] == "telemetry_spool" else ""),
                 "assumptions": "Observed demand may be censored by current limits. Confirm workload and physical/provider capacity before applying."}
 
 
@@ -235,6 +293,10 @@ def capacity_report(store, *, now: float | None = None) -> dict[str, Any]:
     result["stale"] = result["age_seconds"] > 60
     result["state"] = "stale" if result["stale"] else "observed"
     if result["stale"]:
+        collection = result.setdefault("collection", {})
+        collection["state"] = "stale"
+        collection["reclamation"] = {**collection.get("reclamation", {}), "state": "stale",
+                                     "reason": "Collection is not providing current reclamation observations."}
         for item in result.get("limits", []):
             item["recommendation"] = {"status": "stale", "value": None, "reason": "Collection is not providing current observations."}
     return result