From a28621fde36b916984c040f94d174c6a7417c333 Mon Sep 17 00:00:00 2001 From: IK Date: Mon, 21 Sep 2026 20:47:23 +0700 Subject: [PATCH] fix: release cooldown on pre-request failures --- rf4_research/community_cli.py | 59 +++++++++++++++++++++++--- scripts/fetch-waterbodies-chromium.mjs | 33 ++++++++++---- 2 files changed, 78 insertions(+), 14 deletions(-) diff --git a/rf4_research/community_cli.py b/rf4_research/community_cli.py index 2719dfb..583b2d7 100644 --- a/rf4_research/community_cli.py +++ b/rf4_research/community_cli.py @@ -138,7 +138,7 @@ def mark_fetch(source: str, *, state_file: Path, now: float | None = None) -> No def check_and_reserve( source: str, *, state_file: Path, now: float | None = None, -) -> None: +) -> float: """A02: Atomic check-and-reserve under a single exclusive lock. Opens state file with exclusive lock, reads state, checks cooldown, @@ -183,6 +183,36 @@ def check_and_reserve( sf.flush() os.fsync(sf.fileno()) temp_file.replace(state_file) + return now + finally: + fcntl.flock(lf, fcntl.LOCK_UN) + + +def release_reservation(source: str, *, state_file: Path, reserved_at: float) -> bool: + """Release only the reservation created by this attempt. + + A local failure before an HTTP response must not consume the site window. + The timestamp comparison prevents an older process from deleting a newer + reservation made after its own attempt was superseded. + """ + state_file.parent.mkdir(parents=True, exist_ok=True) + lock_file = state_file.with_suffix(".lock") + lock_file.touch(exist_ok=True) + with open(lock_file, "w") as lf: + fcntl.flock(lf, fcntl.LOCK_EX) + try: + state = _read_state(state_file) + current = state.get(source) + if not isinstance(current, (int, float)) or current != reserved_at: + return False + state.pop(source, None) + temp_file = state_file.with_suffix(".tmp") + with open(temp_file, "w") as sf: + sf.write(json.dumps(state, sort_keys=True)) + sf.flush() + os.fsync(sf.fileno()) + temp_file.replace(state_file) + return True finally: fcntl.flock(lf, fcntl.LOCK_UN) @@ -321,9 +351,18 @@ def main(argv: list[str] | None = None) -> int: "--reserve-only", action="store_true", help="Reserve the shared site cooldown and stop before making an HTTP request", ) + parser.add_argument( + "--release-reservation", action="store_true", + help="Release a reservation that failed before reaching the source", + ) + parser.add_argument("--reserved-at", type=float, help="Timestamp returned by --reserve-only") args = parser.parse_args(argv) if args.html and args.reserve_only: parser.error("--html cannot be combined with --reserve-only") + if args.release_reservation and args.reserved_at is None: + parser.error("--release-reservation requires --reserved-at") + if args.release_reservation and args.html: + parser.error("--release-reservation cannot be combined with --html") if args.source in DETAIL_SOURCES and not args.url: parser.error(f"--url is required for {args.source}") default_url, parse = SOURCES.get(args.source, (None, DETAIL_SOURCES.get(args.source))) @@ -333,12 +372,20 @@ def main(argv: list[str] | None = None) -> int: html = args.html.read_text(encoding="utf-8") else: site_key = fetch_site_key(url) - # Single atomic check-and-reserve before network I/O: failed attempts count toward the limit too. - check_and_reserve(site_key, state_file=args.state_file) - if args.reserve_only: - print(json.dumps({"reserved": True, "source": args.source, "site_key": site_key}, ensure_ascii=False)) + if args.release_reservation: + released = release_reservation(site_key, state_file=args.state_file, reserved_at=args.reserved_at) + print(json.dumps({"released": released, "source": args.source, "site_key": site_key}, ensure_ascii=False)) return 0 - html = fetch_html(url) + reserved_at = check_and_reserve(site_key, state_file=args.state_file) + if args.reserve_only: + print(json.dumps({"reserved": True, "reserved_at": reserved_at, "source": args.source, "site_key": site_key}, ensure_ascii=False)) + return 0 + try: + html = fetch_html(url) + except urllib.error.URLError as exc: + if not isinstance(exc, urllib.error.HTTPError): + release_reservation(site_key, state_file=args.state_file, reserved_at=reserved_at) + raise parsed = ( parse(html, source_url=url) if args.source in DETAIL_SOURCES or args.source == "rf4db-waterbodies" diff --git a/scripts/fetch-waterbodies-chromium.mjs b/scripts/fetch-waterbodies-chromium.mjs index a3fe295..7035fe5 100644 --- a/scripts/fetch-waterbodies-chromium.mjs +++ b/scripts/fetch-waterbodies-chromium.mjs @@ -88,7 +88,18 @@ function reserveCooldown(args) { if (result.stderr) process.stderr.write(result.stderr); throw new Error(`cooldown reservation failed with exit code ${result.status}`); } - process.stdout.write(`${result.stdout.trim()}\n`); + const reservation = JSON.parse(result.stdout); + process.stdout.write(`${JSON.stringify(reservation)}\n`); + return reservation; +} + +function releaseCooldown(args, reservation) { + if (!reservation?.reserved_at) return; + const source = args.mode === "catalog" ? "rf4db-waterbodies" : "rf4db-waterbody"; + const command = ["-m", "rf4_research.community_cli", source, "--url", args.url, "--state-file", args.stateFile, "--release-reservation", "--reserved-at", String(reservation.reserved_at)]; + const result = spawnSync(PYTHON, command, { cwd: ROOT, encoding: "utf8" }); + if (result.stdout) process.stdout.write(`${result.stdout.trim()}\n`); + if (result.status !== 0 && result.stderr) process.stderr.write(result.stderr); } function outputPath(args) { @@ -100,18 +111,24 @@ function outputPath(args) { async function main() { const args = parseArgs(process.argv.slice(2)); - reserveCooldown(args); + const executablePath = await resolveExecutable(args.executable); + if (executablePath) console.log(`Using browser executable: ${executablePath}`); + const reservation = reserveCooldown(args); const output = outputPath(args); const htmlOutput = path.resolve(args.htmlOutput || `${output}.html`); await fs.mkdir(path.dirname(htmlOutput), { recursive: true }); await fs.mkdir(path.dirname(output), { recursive: true }); - const executablePath = await resolveExecutable(args.executable); - if (executablePath) console.log(`Using browser executable: ${executablePath}`); - const context = await chromium.launchPersistentContext(path.resolve(args.profile), { - headless: args.headless, - ...(executablePath ? { executablePath } : {}), - }); + let context; + try { + context = await chromium.launchPersistentContext(path.resolve(args.profile), { + headless: args.headless, + ...(executablePath ? { executablePath } : {}), + }); + } catch (error) { + releaseCooldown(args, reservation); + throw error; + } try { const page = context.pages()[0] || await context.newPage(); await page.goto(args.url, { waitUntil: "domcontentloaded", timeout: 60_000 });