fix: circuit-breaker on failure streaks; stop binning rate-limited tasks
Two bugs, both found by autopsy of the djsim_phonewall run (2026-07-08): 15 calendar prompts queued, 2 generated, then THIRTEEN consecutive "Timed out — no new asset" at a metronomic 3m30s apart. 45 minutes of the driver politely feeding prompts into a Flow that had stopped generating. 1. NO CIRCUIT BREAKER. Each timeout threw, was reported failed, and the loop moved straight on. When Flow goes quiet it does not spontaneously start again. Now: 3 consecutive failures -> clear isRunning, log a loud system message, leave the queue intact for a hand-resume. (The abuse-flag path already had consecutiveRateLimits; failures had nothing.) 2. done_ids() COUNTED 'ratelimited' AS DONE. handleRateLimit() writes that status with the explicit comment "report the task back as pending" — and the server then binned the task forever. Now excluded. 'failed' is still counted: a timeout may have burned credits invisibly, so nothing retries automatically. requeue.py: the deliberate way to retry. --list / --dry / by id or prefix / --failed. Writes a .bak, never re-runs a success. Used it to make the 13 cal79 prompts pending again. Note for whoever reads this next: the LIVE server + queue live in ~/Library/Application Support/flowrinse/, not this repo. HERE is derived from the script's own dir, so running the repo copy silently serves the stale Jul-6 queue. Copy the script over, run it from there. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
parent
8464cb5e21
commit
3cff97df93
19
content.js
19
content.js
@ -35,6 +35,8 @@ let pollTimeout = null;
|
|||||||
let generationInProgress = false;
|
let generationInProgress = false;
|
||||||
let successCount = 0;
|
let successCount = 0;
|
||||||
let consecutiveRateLimits = 0;
|
let consecutiveRateLimits = 0;
|
||||||
|
let consecutiveFails = 0;
|
||||||
|
const FAIL_STREAK_STOP = 3; // when Flow goes quiet, it stays quiet. Stop; don't eat the bank.
|
||||||
|
|
||||||
function init() {
|
function init() {
|
||||||
chrome.storage.local.get({ ...settings, pendingHarvest: null }, (data) => {
|
chrome.storage.local.get({ ...settings, pendingHarvest: null }, (data) => {
|
||||||
@ -480,17 +482,28 @@ async function executeTask(task) {
|
|||||||
else if (tiles.length) log('info', `New tile detected but no direct media URL — reporting the tile link (recover later). ${gridCensus()}`);
|
else if (tiles.length) log('info', `New tile detected but no direct media URL — reporting the tile link (recover later). ${gridCensus()}`);
|
||||||
await reportCompletion(task.id, 'success', null, fresh);
|
await reportCompletion(task.id, 'success', null, fresh);
|
||||||
chrome.storage.local.remove('requeue_' + task.id);
|
chrome.storage.local.remove('requeue_' + task.id);
|
||||||
successCount++; consecutiveRateLimits = 0;
|
successCount++; consecutiveRateLimits = 0; consecutiveFails = 0;
|
||||||
if (settings.restEvery && successCount % settings.restEvery === 0) {
|
if (settings.restEvery && successCount % settings.restEvery === 0) {
|
||||||
setCooldown(settings.restMs || 15 * 60 * 1000, `Scheduled rest after ${successCount} gens`);
|
setCooldown(settings.restMs || 15 * 60 * 1000, `Scheduled rest after ${successCount} gens`);
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
log('error', `Task ${task.id} failed: ${err.message}`);
|
log('error', `Task ${task.id} failed: ${err.message}`);
|
||||||
await reportCompletion(task.id, 'failed', err.message, []);
|
await reportCompletion(task.id, 'failed', err.message, []);
|
||||||
|
// CIRCUIT BREAKER. 2026-07-08: Flow went quiet after two images and this loop
|
||||||
|
// politely fed it 13 more prompts over 45 minutes, marking every one failed.
|
||||||
|
// When Flow stops producing it does not spontaneously start again — STOP, and
|
||||||
|
// leave the queue intact so the run can be resumed by hand.
|
||||||
|
if (++consecutiveFails >= FAIL_STREAK_STOP) {
|
||||||
|
settings.isRunning = false;
|
||||||
|
chrome.storage.local.set({ isRunning: false });
|
||||||
|
log('system', `⛔ ${consecutiveFails} failures in a row — STOPPING. `
|
||||||
|
+ `Flow isn't generating. Check the tab (credits? signed out? queue jammed?), `
|
||||||
|
+ `then re-queue the failed ids and press Start.`);
|
||||||
|
}
|
||||||
} finally {
|
} finally {
|
||||||
generationInProgress = false;
|
generationInProgress = false;
|
||||||
chrome.storage.local.set({ status: settings.isRunning ? 'running' : 'idle' });
|
chrome.storage.local.set({ status: settings.isRunning ? 'running' : 'idle' });
|
||||||
scheduleNext();
|
scheduleNext(); // no-ops when the breaker above cleared isRunning
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -530,7 +543,7 @@ async function runPostReloadHarvest(ph) {
|
|||||||
chrome.storage.local.remove('pendingHarvest');
|
chrome.storage.local.remove('pendingHarvest');
|
||||||
chrome.storage.local.remove('requeue_' + ph.id);
|
chrome.storage.local.remove('requeue_' + ph.id);
|
||||||
await reportCompletion(ph.id, 'success', null, fresh);
|
await reportCompletion(ph.id, 'success', null, fresh);
|
||||||
successCount++; consecutiveRateLimits = 0;
|
successCount++; consecutiveRateLimits = 0; consecutiveFails = 0;
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
log('error', `Post-refresh harvest failed: ${err.message}`);
|
log('error', `Post-refresh harvest failed: ${err.message}`);
|
||||||
chrome.storage.local.remove('pendingHarvest');
|
chrome.storage.local.remove('pendingHarvest');
|
||||||
|
|||||||
@ -82,13 +82,24 @@ def load_queue():
|
|||||||
return rows
|
return rows
|
||||||
|
|
||||||
def done_ids():
|
def done_ids():
|
||||||
|
"""Ids that must NOT be handed out again.
|
||||||
|
|
||||||
|
'ratelimited' is deliberately absent. handleRateLimit() writes that status meaning
|
||||||
|
"report the task back as pending" — no gen ran, the abuse flag just froze us mid-task.
|
||||||
|
Counting it as done silently binned the task forever, contradicting its own comment.
|
||||||
|
|
||||||
|
'failed' IS counted: a timeout may have burned credits invisibly, so nothing is retried
|
||||||
|
automatically. To re-run a failed task, delete its line from results.jsonl (requeue.py).
|
||||||
|
"""
|
||||||
ids = set()
|
ids = set()
|
||||||
if os.path.exists(RESULTS_FILE):
|
if os.path.exists(RESULTS_FILE):
|
||||||
with open(RESULTS_FILE) as f:
|
with open(RESULTS_FILE) as f:
|
||||||
for line in f:
|
for line in f:
|
||||||
line = line.strip()
|
line = line.strip()
|
||||||
if line:
|
if line:
|
||||||
ids.add(json.loads(line).get("id"))
|
row = json.loads(line)
|
||||||
|
if row.get("status") != "ratelimited":
|
||||||
|
ids.add(row.get("id"))
|
||||||
return ids
|
return ids
|
||||||
|
|
||||||
class Handler(BaseHTTPRequestHandler):
|
class Handler(BaseHTTPRequestHandler):
|
||||||
|
|||||||
76
requeue.py
Normal file
76
requeue.py
Normal file
@ -0,0 +1,76 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
"""Make finished tasks pending again by removing their rows from results.jsonl.
|
||||||
|
|
||||||
|
The queue server treats any id in results.jsonl as done (except 'ratelimited'), so a task
|
||||||
|
that failed is never retried automatically — a timeout may have burned credits invisibly
|
||||||
|
and we don't re-bill by accident. This is the deliberate, explicit way to retry.
|
||||||
|
|
||||||
|
python3 requeue.py --list # what's failed, grouped
|
||||||
|
python3 requeue.py --dry cal79_ # what WOULD be requeued (prefix match)
|
||||||
|
python3 requeue.py cal79_ # do it (writes a .bak first)
|
||||||
|
python3 requeue.py --failed # every failed task
|
||||||
|
python3 requeue.py cal79_disco cal79_tmi # exact ids
|
||||||
|
|
||||||
|
Stop the queue server first, or it may rewrite the file under you.
|
||||||
|
"""
|
||||||
|
import collections
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
import shutil
|
||||||
|
import sys
|
||||||
|
|
||||||
|
HERE = os.path.expanduser('~/Library/Application Support/flowrinse')
|
||||||
|
RESULTS = os.path.join(HERE, 'results.jsonl')
|
||||||
|
|
||||||
|
|
||||||
|
def rows():
|
||||||
|
with open(RESULTS) as f:
|
||||||
|
return [json.loads(l) for l in f if l.strip()]
|
||||||
|
|
||||||
|
|
||||||
|
def main():
|
||||||
|
args = [a for a in sys.argv[1:]]
|
||||||
|
if not args:
|
||||||
|
print(__doc__)
|
||||||
|
return 1
|
||||||
|
dry = '--dry' in args
|
||||||
|
only_failed = '--failed' in args
|
||||||
|
pats = [a for a in args if not a.startswith('--')]
|
||||||
|
rs = rows()
|
||||||
|
|
||||||
|
if '--list' in args:
|
||||||
|
c = collections.Counter((r['status'], r['id'].rsplit('_', 1)[0]) for r in rs)
|
||||||
|
for (st, grp), n in sorted(c.items()):
|
||||||
|
print(f'{st:<12} {grp:<24} {n}')
|
||||||
|
return 0
|
||||||
|
|
||||||
|
def hit(r):
|
||||||
|
if r['status'] == 'success':
|
||||||
|
return False # never re-run a success; that's just burning credits
|
||||||
|
if only_failed and r['status'] == 'failed':
|
||||||
|
return True
|
||||||
|
return any(r['id'] == p or r['id'].startswith(p) for p in pats)
|
||||||
|
|
||||||
|
doomed = [r for r in rs if hit(r)]
|
||||||
|
if not doomed:
|
||||||
|
print('nothing matched (successes are never requeued)')
|
||||||
|
return 1
|
||||||
|
for r in doomed:
|
||||||
|
print(f" requeue {r['id']:<24} {r['status']:<10} {str(r.get('error',''))[:46]}")
|
||||||
|
print(f"\n{len(doomed)} task(s){' [DRY RUN]' if dry else ''}")
|
||||||
|
if dry:
|
||||||
|
return 0
|
||||||
|
|
||||||
|
shutil.copy2(RESULTS, RESULTS + '.bak')
|
||||||
|
keep = [r for r in rs if r not in doomed]
|
||||||
|
with open(RESULTS + '.tmp', 'w') as f:
|
||||||
|
for r in keep:
|
||||||
|
f.write(json.dumps(r) + '\n')
|
||||||
|
os.replace(RESULTS + '.tmp', RESULTS) # a kill mid-write can't truncate the real file
|
||||||
|
print(f'wrote {len(keep)} rows (backup: results.jsonl.bak). Restart the queue server; '
|
||||||
|
'they are pending again.')
|
||||||
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
sys.exit(main())
|
||||||
Loading…
Reference in New Issue
Block a user