fix(server): close residual review findings F-5..F-8
- Continuous endpoints reject non-list channels (400), null no longer wipes stored list (F-5) - '@'-prefixed legacy channel values normalized on load/save in manager and store (F-6) - _run_loop guarded: unexpected exceptions logged, backoff retry, no silent thread death (F-7) - +8 tests: continuous validation, normalization, loop survival, join-timeout/tombstone (F-8) - 58 tests passing; REVIEW.md updated
This commit is contained in:
+135
-67
@@ -1055,12 +1055,20 @@ class PerAccountContinuousScrapeManager:
|
||||
|
||||
def _load_config(self) -> Dict[str, Any]:
|
||||
acc_state = load_account(DATA_DIR, self.account_id)
|
||||
return dict(acc_state.get("continuous_scraping", {
|
||||
cfg = dict(acc_state.get("continuous_scraping", {
|
||||
"enabled": True,
|
||||
"interval_minutes": 1,
|
||||
"channels": [],
|
||||
"run_all_tracked": True,
|
||||
}))
|
||||
# F-6: normalize @-prefixed channel values that were written to disk
|
||||
# by import/migration/older code — they must match tracked ids (no @).
|
||||
cfg["channels"] = [
|
||||
str(ch).strip().lstrip("@")
|
||||
for ch in cfg.get("channels", [])
|
||||
if str(ch).strip()
|
||||
]
|
||||
return cfg
|
||||
|
||||
def _save_config(self) -> None:
|
||||
store = get_account_store(DATA_DIR, self.account_id)
|
||||
@@ -1069,7 +1077,7 @@ class PerAccountContinuousScrapeManager:
|
||||
"enabled": bool(self.config.get("enabled", True)),
|
||||
"interval_minutes": max(1, int(self.config.get("interval_minutes", 1) or 1)),
|
||||
"channels": [
|
||||
str(item).strip()
|
||||
str(item).strip().lstrip("@")
|
||||
for item in self.config.get("channels", [])
|
||||
if str(item).strip()
|
||||
],
|
||||
@@ -1090,7 +1098,7 @@ class PerAccountContinuousScrapeManager:
|
||||
self.config["enabled"] = bool(disk_cfg.get("enabled", False))
|
||||
self.config["interval_minutes"] = max(1, int(disk_cfg.get("interval_minutes", 1) or 1))
|
||||
self.config["channels"] = [
|
||||
str(item).strip()
|
||||
str(item).strip().lstrip("@")
|
||||
for item in disk_cfg.get("channels", [])
|
||||
if str(item).strip()
|
||||
]
|
||||
@@ -1234,65 +1242,82 @@ class PerAccountContinuousScrapeManager:
|
||||
def _run_loop(self) -> None:
|
||||
try:
|
||||
while not self.stop_event.is_set():
|
||||
# Refresh config from disk so channel / setting changes take effect
|
||||
self.refresh_config()
|
||||
# Bail promptly if refresh/cancel requested the stop so join()
|
||||
# usually returns quickly instead of waiting out a full scrape.
|
||||
if self.stop_event.is_set():
|
||||
break
|
||||
# F-7: a top-level guard keeps the worker thread alive even if
|
||||
# something outside the scrape iteration itself raises (e.g.
|
||||
# refresh_config/auth_status_for). A dead worker thread while
|
||||
# enabled=True is the failure mode F-3 was fixing.
|
||||
try:
|
||||
# Refresh config from disk so channel / setting changes take effect
|
||||
self.refresh_config()
|
||||
# Bail promptly if refresh/cancel requested the stop so join()
|
||||
# usually returns quickly instead of waiting out a full scrape.
|
||||
if self.stop_event.is_set():
|
||||
break
|
||||
|
||||
# Check auth — don't iterate if account isn't authorized
|
||||
auth_info = auth_status_for(self.account_id)
|
||||
if auth_info.get("status") not in ("ready", "authorized"):
|
||||
self._log(
|
||||
f"Account not authorized (status={auth_info.get('status')}), "
|
||||
"skipping iteration",
|
||||
"warn",
|
||||
)
|
||||
sleep_seconds = max(5, 60)
|
||||
# Check auth — don't iterate if account isn't authorized
|
||||
auth_info = auth_status_for(self.account_id)
|
||||
if auth_info.get("status") not in ("ready", "authorized"):
|
||||
self._log(
|
||||
f"Account not authorized (status={auth_info.get('status')}), "
|
||||
"skipping iteration",
|
||||
"warn",
|
||||
)
|
||||
sleep_seconds = max(5, 60)
|
||||
interrupted = self.stop_event.wait(timeout=sleep_seconds)
|
||||
if interrupted:
|
||||
break
|
||||
continue
|
||||
|
||||
channels = self._resolve_channels()
|
||||
cfg = self.snapshot()["config"]
|
||||
interval_minutes = cfg.get("interval_minutes", 1)
|
||||
|
||||
if not channels:
|
||||
self._log("No channels configured for continuous scraping.", "warn")
|
||||
else:
|
||||
self._log(f"Starting iteration for {len(channels)} channel(s).", "info")
|
||||
buffer = io.StringIO()
|
||||
try:
|
||||
with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer):
|
||||
run_job("scrape_selected", {
|
||||
"channels": channels,
|
||||
"account_id": self.account_id,
|
||||
})
|
||||
output = buffer.getvalue().strip()
|
||||
if output:
|
||||
for line in output.splitlines():
|
||||
self._log(line)
|
||||
with self.lock:
|
||||
self.status["last_iteration_at"] = utc_now_iso()
|
||||
self.status["last_finished_at"] = utc_now_iso()
|
||||
self.status["last_error"] = None
|
||||
self._log("Iteration finished.", "success")
|
||||
except Exception as exc:
|
||||
output = buffer.getvalue().strip()
|
||||
if output:
|
||||
for line in output.splitlines():
|
||||
self._log(line)
|
||||
self._log(f"Iteration failed: {exc}", "error")
|
||||
with self.lock:
|
||||
self.status["last_error"] = str(exc)
|
||||
|
||||
sleep_seconds = max(5, interval_minutes * 60)
|
||||
self._log(f"Sleeping for {interval_minutes} minute(s).", "debug")
|
||||
interrupted = self.stop_event.wait(timeout=sleep_seconds)
|
||||
if interrupted:
|
||||
break
|
||||
continue
|
||||
|
||||
channels = self._resolve_channels()
|
||||
cfg = self.snapshot()["config"]
|
||||
interval_minutes = cfg.get("interval_minutes", 1)
|
||||
|
||||
if not channels:
|
||||
self._log("No channels configured for continuous scraping.", "warn")
|
||||
else:
|
||||
self._log(f"Starting iteration for {len(channels)} channel(s).", "info")
|
||||
buffer = io.StringIO()
|
||||
try:
|
||||
with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer):
|
||||
run_job("scrape_selected", {
|
||||
"channels": channels,
|
||||
"account_id": self.account_id,
|
||||
})
|
||||
output = buffer.getvalue().strip()
|
||||
if output:
|
||||
for line in output.splitlines():
|
||||
self._log(line)
|
||||
with self.lock:
|
||||
self.status["last_iteration_at"] = utc_now_iso()
|
||||
self.status["last_finished_at"] = utc_now_iso()
|
||||
self.status["last_error"] = None
|
||||
self._log("Iteration finished.", "success")
|
||||
except Exception as exc:
|
||||
output = buffer.getvalue().strip()
|
||||
if output:
|
||||
for line in output.splitlines():
|
||||
self._log(line)
|
||||
self._log(f"Iteration failed: {exc}", "error")
|
||||
with self.lock:
|
||||
self.status["last_error"] = str(exc)
|
||||
|
||||
sleep_seconds = max(5, interval_minutes * 60)
|
||||
self._log(f"Sleeping for {interval_minutes} minute(s).", "debug")
|
||||
interrupted = self.stop_event.wait(timeout=sleep_seconds)
|
||||
if interrupted:
|
||||
break
|
||||
except Exception:
|
||||
# Unexpected failure outside the scrape iteration: log it,
|
||||
# surface it in status, and back off briefly instead of
|
||||
# letting the worker thread die silently.
|
||||
logger.exception("Continuous scrape loop error for account %s", self.account_id)
|
||||
with self.lock:
|
||||
self.status["last_error"] = "Unexpected loop error (see logs)"
|
||||
self.status["last_iteration_at"] = utc_now_iso()
|
||||
self._log("Unexpected loop error; retrying in 10s.", "error")
|
||||
interrupted = self.stop_event.wait(timeout=10)
|
||||
if interrupted:
|
||||
break
|
||||
finally:
|
||||
# Only mark running=False once the thread has truly exited so the
|
||||
# status reflects reality (a still-running scrape is not "stopped").
|
||||
@@ -2533,16 +2558,43 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
|
||||
try:
|
||||
enabled = parse_bool(body.get("enabled"))
|
||||
interval_minutes = int(body.get("interval_minutes", 1))
|
||||
channels, dropped = clean_continuous_channels(body.get("channels", []))
|
||||
# F-5: if `channels` key is present but not a list, reject
|
||||
# with 400 instead of silently wiping the stored list.
|
||||
raw_channels = body.get("channels")
|
||||
if raw_channels is not None and not isinstance(raw_channels, list):
|
||||
return self.send_error_json(
|
||||
HTTPStatus.BAD_REQUEST,
|
||||
"`channels` must be a list if provided",
|
||||
)
|
||||
if raw_channels is not None:
|
||||
channels, dropped = clean_continuous_channels(raw_channels)
|
||||
else:
|
||||
# Key absent → keep existing stored channel list.
|
||||
channels = None
|
||||
dropped = []
|
||||
run_all_tracked = parse_bool(body.get("run_all_tracked", True), default=True)
|
||||
if self.app.legacy_account_id:
|
||||
payload = self.app.continuous_orchestrator.update_for(
|
||||
account_id=self.app.legacy_account_id,
|
||||
enabled=enabled,
|
||||
interval_minutes=interval_minutes,
|
||||
channels=channels,
|
||||
run_all_tracked=run_all_tracked,
|
||||
)
|
||||
if channels is not None:
|
||||
payload = self.app.continuous_orchestrator.update_for(
|
||||
account_id=self.app.legacy_account_id,
|
||||
enabled=enabled,
|
||||
interval_minutes=interval_minutes,
|
||||
channels=channels,
|
||||
run_all_tracked=run_all_tracked,
|
||||
)
|
||||
else:
|
||||
# channels absent: read existing from disk, do not overwrite.
|
||||
existing = load_account(DATA_DIR, self.app.legacy_account_id)
|
||||
existing_channels, _ = clean_continuous_channels(
|
||||
existing.get("continuous_scraping", {}).get("channels", [])
|
||||
)
|
||||
payload = self.app.continuous_orchestrator.update_for(
|
||||
account_id=self.app.legacy_account_id,
|
||||
enabled=enabled,
|
||||
interval_minutes=interval_minutes,
|
||||
channels=existing_channels,
|
||||
run_all_tracked=run_all_tracked,
|
||||
)
|
||||
else:
|
||||
payload = {"config": {}, "status": {"running": False, "logs": [], "log_entries": []}}
|
||||
except Exception as exc:
|
||||
@@ -3078,7 +3130,23 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
|
||||
try:
|
||||
enabled = parse_bool(body.get("enabled"))
|
||||
interval_minutes = int(body.get("interval_minutes", 1))
|
||||
channels, dropped = clean_continuous_channels(body.get("channels", []))
|
||||
# F-5: if `channels` key is present but not a list, reject
|
||||
# with 400 instead of silently wiping the stored list.
|
||||
raw_channels = body.get("channels")
|
||||
if raw_channels is not None and not isinstance(raw_channels, list):
|
||||
return self.send_error_json(
|
||||
HTTPStatus.BAD_REQUEST,
|
||||
"`channels` must be a list if provided",
|
||||
)
|
||||
if raw_channels is not None:
|
||||
channels, dropped = clean_continuous_channels(raw_channels)
|
||||
else:
|
||||
# Key absent → keep existing stored channel list.
|
||||
existing = load_account(DATA_DIR, account_id)
|
||||
channels, _ = clean_continuous_channels(
|
||||
existing.get("continuous_scraping", {}).get("channels", [])
|
||||
)
|
||||
dropped = []
|
||||
run_all_tracked = parse_bool(body.get("run_all_tracked", True), default=True)
|
||||
payload = self.app.continuous_orchestrator.update_for(
|
||||
account_id=account_id,
|
||||
|
||||
Reference in New Issue
Block a user