--- a/scraper/README.md +++ b/scraper/README.md @@ -156,7 +156,7 @@ ## Verified before handover -- `pytest tests` - 86 offline tests (26 collector + 60 for the Oxylabs transport and its provider codes, caps and deadline, watchdog, run budget, cumulative allowance, fulfilment context, probe evidence and acceptance), several of them on real marianos.com payloads: field order and types, price/UPC/gupc +- `pytest tests` - 96 offline tests (27 collector, including three full runs that check retention of unavailable and later-absent products + 69 for the Oxylabs transport and its provider codes, caps and deadline, watchdog, run budget, the shared Blob ledger (lease, If-Match, read-back, reconciliation after an interruption), fulfilment context, probe evidence and acceptance), several of them on real marianos.com payloads: field order and types, price/UPC/gupc rules, displayed PPM, sale vs quantity-offer price, digital coupon wording and stated coupon price, points offers excluded, approximate-weight exclusion, unavailable and unpriced products kept, another location's price never used, failed run never replaces good data, lost store context blocked, wrong store aborts before collecting, interrupted run --- a/scraper/docs/TRANSPORT_OXYLABS.md +++ b/scraper/docs/TRANSPORT_OXYLABS.md @@ -62,9 +62,21 @@ compares `X-Laf-Object` there (case-insensitive) with the header it sent (`laf_echo`: match / missing / mismatch) and reports `store_header_forwarded` (pass / inconclusive / fail). This is supporting evidence only: the store and fulfilment context are proven by each product answer (`laf` and `sourceLocationId`). -* Cumulative allowance (probe, `--ledger FILE --ledger-ceiling 15`): before any request the probe caps itself at what - is left of the ceiling across every earlier execution and writes that reservation; at the end it records the jobs - dispatched. A killed run keeps its full reservation counted; a new folder, session or process cannot reset it. +* Cumulative allowance (probe, `--ledger blob: --ledger-ceiling 15`; v2.2.16, `marianos_scraper/allowance.py`): + the ledger is the blob itself, read and written by the probe with the VM's managed identity. No copy is downloaded + or uploaded by any script, so a stale copy can never overwrite a newer ledger. Every change is one read-modify-write + under an exclusive 30 s blob lease, written with `If-Match` on the ETag that was read, then read back and compared + byte for byte. + + | Step | When | If it cannot be verified | + |---|---|---| + | Lease | before every change; waits at most 30 s for another execution | stop, exit 2, nothing sent | + | Reconcile | at reservation: entries still `reserved` (a run interrupted, killed, or whose final update was not verified) become `unfinished` and stay counted at their full reservation; never lowered automatically | — | + | Reserve | before the first request: `status reserved`, `reserved_jobs` = this run's cap (≤ what is left), `reservation_id` | stop, exit 2, nothing sent (LedgerMismatch / ETag / lease error in `report.json`) | + | Final | at the end (also after a stop or the watchdog), on a fresh read matched by `reservation_id`: `status final`, jobs actually dispatched | exit 2, `allowance.final_update = UNRESOLVED ...`; the reservation stays counted in full; no unverified copy is written to the evidence | + + A missing ledger blob or an entry with neither dispatched jobs nor a reservation stops the probe (fail closed). A + local file (`--ledger `) follows the same protocol with an exclusive lock file and the SHA-256 as ETag. * Probe only: store locator answer ≠ configured store/facility → stop before any product request; a priced product whose `sourceLocationId` ≠ configured store, or any product whose echoed context (`laf`: modality, storeId, facilityId) ≠ the one sent → stop before the coupon request. Full runs reject such a unit too (retried, then the @@ -72,17 +84,27 @@ ## Probe (the small test) +On the Azure VM the session script reads `oxylabs-username` / `oxylabs-password` from Key Vault with the VM's managed +identity, into the launcher's environment only. If that read fails it stops with the Key Vault error (exit 5) and sends +nothing; there is no other credential route. The launcher adds the outer kill: + ``` -read -s OXYLABS_USERNAME && read -s OXYLABS_PASSWORD && export OXYLABS_USERNAME OXYLABS_PASSWORD # typed, never stored +.\infra\run_probe.ps1 -Out evidence\oxylabs-probe- -Upcs ,,,, ` + -Roles ppm-volume,ppm-digital,regular,unavailable,unavailable -MaxRequests 15 -AzureImds ` + -Ledger "blob:https://.blob.core.windows.net//probe/cumulative-usage.json" +``` +which runs +``` python -m marianos_scraper probe --location marianos-vernon-hills --transport oxylabs \ --upcs ,,,, --roles ppm-volume,ppm-digital,regular,unavailable,unavailable \ - --max-requests 15 --max-seconds 1200 --azure-imds --out evidence/oxylabs-probe-$(date -u +%Y%m%dT%H%M%SZ) + --max-requests 15 --max-seconds 1200 --azure-imds --ledger blob: --ledger-ceiling 15 --out ``` Requests in a 5-UPC probe: 1 locator + 1 products (+1 if a UPC is not returned, asked again without the `verified` filter) + 1 coupons per page = 3–4. Retries, if any, count against the same cap. Exit codes: 0 = transport completed and every acceptance check passed; 1 = transport completed but a check failed or -is inconclusive (report.json `acceptance`); 2 = stopped by a cap, an Oxylabs error or a store/fulfilment mismatch; +is inconclusive (report.json `acceptance`); 2 = stopped by a cap, an Oxylabs error, a store/fulfilment mismatch or a +ledger that could not be leased, written or verified; 4 = the site refused. Every attempt's exact answer (failures and every coupon page included) is in `raw/`, indexed in `raw/index.jsonl`. `infra/upload_probe_evidence.ps1` uploads the folder to Blob and reads every file back by SHA-256. --- a/scraper/infra/run_probe.ps1 +++ b/scraper/infra/run_probe.ps1 @@ -10,8 +10,8 @@ -Upcs 0003940001880,0001600040803,0073820310311,0004157305440,0002100006058 ` -Roles ppm-volume,ppm-digital,regular,unavailable,unavailable -AzureImds - -Ledger passes the cumulative-usage ledger shared by every execution: the probe caps itself at what is left of - -LedgerCeiling (15) and records its reservation before any request, so a new folder or process cannot reset it. + -Ledger blob: (or a local file) passes the cumulative-usage ledger shared by every execution. The probe itself + leases it, reserves what is left of -LedgerCeiling (15) before any request and verifies every write by read-back. The Oxylabs credentials are inherited from this PowerShell session's environment (OXYLABS_USERNAME / OXYLABS_PASSWORD) and are never written anywhere. After the child ends, launcher.json is written into the evidence --- a/scraper/marianos_scraper/__init__.py +++ b/scraper/marianos_scraper/__init__.py @@ -1,2 +1,2 @@ """Mariano's location-level product, price, PPM and deal collector (marianos.com, Kroger Public API fallback), built for scheduled runs on Azure.""" -__version__ = "2.2.15" +__version__ = "2.2.16" --- a/scraper/marianos_scraper/__main__.py +++ b/scraper/marianos_scraper/__main__.py @@ -79,7 +79,7 @@ probe.add_argument("--egress-ip", default="", help="this machine's public IP as the operator read it (recorded only)") probe.add_argument("--roles", help="comma-separated role per UPC, same order: ppm-volume, ppm-digital, regular, unavailable, any") probe.add_argument("--azure-imds", action="store_true", help="record which Azure VM ran the probe (Instance Metadata Service)") - probe.add_argument("--ledger", help="cumulative-usage JSON shared by every execution; this run is capped at what is left") + probe.add_argument("--ledger", help="shared cumulative-usage ledger: blob: (leased, ETag-checked, read back; managed identity) or a local file; this run is capped at what is left") probe.add_argument("--ledger-ceiling", type=int, default=15, help="total jobs allowed across all executions") ver = sub.add_parser("verify-location") ver.add_argument("--location", required=True) @@ -122,7 +122,7 @@ try: probe_run = Probe(cfg, loc, site, upcs, out, transport=args.transport, egress_ip=args.egress_ip, roles=roles, azure_imds=args.azure_imds, - ledger=Path(args.ledger) if args.ledger else None, ledger_ceiling=args.ledger_ceiling) + ledger=args.ledger or None, ledger_ceiling=args.ledger_ceiling) except ValueError as exc: parser.error(str(exc)) return probe_run.run() # 0 completed + accepted, 1 completed but not accepted, 2 stopped, 4 site refused --- /dev/null +++ b/scraper/marianos_scraper/allowance.py @@ -0,0 +1,279 @@ +"""Cumulative Oxylabs allowance shared by every probe execution (v2.2.16). + +The ledger is one JSON document. On Azure it is the blob probe/cumulative-usage.json; the probe reads and writes that +blob itself with the VM's managed identity (--ledger blob:), so no local copy exists that could go stale or be +uploaded over a newer version. A local file (--ledger ) gets the same protocol with a lock file. + +Every change is one read-modify-write: + 1. take the exclusive lock (blob: a 30 s lease; file: an O_EXCL lock file), waiting at most `lock_wait` seconds; + 2. read the document and its ETag (file: SHA-256 of the bytes); + 3. write the new document only if the ETag is unchanged (blob: If-Match plus the lease id); + 4. read it back and compare byte for byte. Any difference raises LedgerMismatch: the probe stops, nothing further + is sent, and the ledger is left for a person to resolve. + +Reservation before dispatch: the probe calls reserve() before its first request. The reservation (status "reserved", +reserved_jobs = this run's cap) is durable and verified before anything is sent. finalize() later replaces it, matched +by reservation_id on a fresh read (never on the copy read at the start), with the jobs actually dispatched. + +Reconciliation on restart: reserve() first looks for entries still "reserved" (a run that was interrupted, killed, or +whose final update could not be verified). They become "unfinished" and stay counted at their full reservation. This +is never lowered automatically; only a person who has the evidence may correct it. + +Any ledger error stops the probe before dispatch (exit 2). A missing blob is an error, never a fresh start. +""" +from __future__ import annotations + +import hashlib +import json +import os +import socket +import time +import uuid +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path + + +class LedgerError(Exception): + """The ledger could not be read, locked, changed or verified. The probe stops; nothing more is sent.""" + + +class LedgerMismatch(LedgerError): + """What was read back is not what was written. The probe stops and the ledger stays unresolved.""" + + +def utc() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def jobs_counted(ledger: dict) -> int: + """Jobs counted so far: dispatched where recorded, else the full reservation. An entry with neither stops.""" + total = 0 + for i, e in enumerate(ledger.get("executions") or []): + jobs = e.get("oxylabs_jobs_submitted") + if jobs is None: + jobs = e.get("reserved_jobs") + if jobs is None: + raise LedgerError(f"ledger entry {i} ({e.get('folder')}) has neither dispatched jobs nor a reservation") + total += int(jobs) + return total + + +def encode(data: dict) -> bytes: + return (json.dumps(data, indent=1, ensure_ascii=False) + "\n").encode("utf-8") + + +def sha256(b: bytes) -> str: + return hashlib.sha256(b).hexdigest() + + +def _status(exc: Exception): + return getattr(exc, "status_code", None) or getattr(getattr(exc, "response", None), "status_code", None) + + +class FileStore: + """A local ledger file. The lock is an exclusive lock file next to it; the ETag is the SHA-256 of the bytes.""" + may_create = True + + def __init__(self, path, *, lock_wait: float = 30.0, sleep=time.sleep, clock=time.monotonic): + self.path = Path(path) + self.lock_path = self.path.with_name(self.path.name + ".lock") + self.lock_wait, self.sleep, self.clock = lock_wait, sleep, clock + self._fd = None + + def describe(self) -> str: + return str(self.path) + + def lock(self) -> None: + end = self.clock() + self.lock_wait + while True: + try: + self._fd = os.open(self.lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) + os.write(self._fd, f"{socket.gethostname()} pid {os.getpid()} {utc()}".encode()) + return + except FileExistsError: + if self.clock() >= end: + raise LedgerError(f"ledger lock {self.lock_path} is held by another execution") from None + self.sleep(1.0) + + def unlock(self) -> None: + if self._fd is not None: + os.close(self._fd) + self._fd = None + os.remove(self.lock_path) + + def read(self): + if not self.path.exists(): + return None, None + b = self.path.read_bytes() + return b, sha256(b) + + def write(self, body: bytes, etag) -> str: + _, current = self.read() + if current != etag: + raise LedgerError("ledger changed since it was read (ETag mismatch); nothing written") + tmp = self.path.with_name(self.path.name + ".tmp") + tmp.write_bytes(body) + os.replace(tmp, self.path) + return sha256(body) + + def read_back(self): + return self.read() + + +class BlobStore: + """The ledger blob. Lock = exclusive blob lease; writes carry the lease id and If-Match on the ETag read.""" + may_create = False + + def __init__(self, blob, *, lease_seconds: int = 30, lock_wait: float = 30.0, sleep=time.sleep, clock=time.monotonic): + self.blob = blob # azure.storage.blob.BlobClient (or a test double with the same calls) + self.lease_seconds, self.lock_wait, self.sleep, self.clock = lease_seconds, lock_wait, sleep, clock + self._lease = None + + def describe(self) -> str: + return getattr(self.blob, "url", "blob") + + def lock(self) -> None: + end = self.clock() + self.lock_wait + while True: + try: + self._lease = self.blob.acquire_lease(lease_duration=self.lease_seconds) + return + except Exception as exc: # noqa: BLE001 + code = _status(exc) + if code == 404: + raise LedgerError(f"ledger blob not found ({self.describe()}): refusing to start a new ledger") from exc + if code != 409 or self.clock() >= end: + raise LedgerError(f"could not lease the ledger blob: {code or type(exc).__name__}: {exc}") from exc + self.sleep(2.0) # 409: another execution holds the lease + + def unlock(self) -> None: + if self._lease is not None: + try: + self._lease.release() + except Exception: # noqa: BLE001 - the lease expires on its own within lease_seconds + pass + self._lease = None + + def read(self): + try: + d = self.blob.download_blob(lease=self._lease) + return d.readall(), d.properties.etag + except Exception as exc: # noqa: BLE001 + raise LedgerError(f"could not read the ledger blob: {_status(exc) or type(exc).__name__}: {exc}") from exc + + def write(self, body: bytes, etag) -> str: + try: + from azure.core import MatchConditions + cond = MatchConditions.IfNotModified + except ImportError: # pragma: no cover - the package pins azure-core through azure-storage-blob + cond = "IfNotModified" + try: + r = self.blob.upload_blob(body, overwrite=True, lease=self._lease, etag=etag, match_condition=cond) + return r["etag"] + except Exception as exc: # noqa: BLE001 + code = _status(exc) + what = "ledger changed since it was read (ETag mismatch)" if code == 412 else "could not write the ledger blob" + raise LedgerError(f"{what}: {code or type(exc).__name__}: {exc}") from exc + + def read_back(self): + try: + d = self.blob.download_blob() + return d.readall(), d.properties.etag + except Exception as exc: # noqa: BLE001 + raise LedgerMismatch(f"ledger written but could not be read back: {exc}") from exc + + +def open_store(spec: str, **kw): + """--ledger value: 'blob:https://.blob.core.windows.net//' or a local path.""" + if spec.startswith("blob:"): + from azure.identity import ManagedIdentityCredential + from azure.storage.blob import BlobClient + cred = ManagedIdentityCredential(client_id=os.environ.get("AZURE_CLIENT_ID") or None) # None: the VM's identity + blob = BlobClient.from_blob_url(spec[5:], credential=cred, + connection_timeout=10, read_timeout=15, retry_total=2) + return BlobStore(blob, **kw) + return FileStore(spec, **kw) + + +class Allowance: + def __init__(self, store, ceiling: int = 15): + self.store, self.ceiling = store, ceiling + self.reservation_id: str | None = None + self.last_sha256 = "" + + @contextmanager + def _locked(self): + self.store.lock() + try: + yield + finally: + self.store.unlock() + + def _load(self): + body, etag = self.store.read() + if body is None: + if not self.store.may_create: + raise LedgerError(f"ledger not found ({self.store.describe()})") + return {"ceiling": self.ceiling, "executions": []}, None + try: + return json.loads(body.decode("utf-8")), etag + except ValueError as exc: + raise LedgerError(f"ledger is not valid JSON: {exc}") from exc + + def _commit(self, data: dict, etag) -> dict: + body = encode(data) + written = self.store.write(body, etag) + back, back_etag = self.store.read_back() + if back != body: + raise LedgerMismatch(f"ledger read-back differs: wrote sha256 {sha256(body)}, read " + f"{sha256(back) if back is not None else 'nothing'}; left unresolved") + if written and back_etag and written != back_etag: + raise LedgerMismatch(f"ledger ETag after write {written} but read back {back_etag}; left unresolved") + self.last_sha256 = sha256(body) + return data + + def reserve(self, folder: str, package: str, wanted: int | None) -> dict: + """Reconcile unfinished entries, cap this run at what is left, write and verify the reservation.""" + with self._locked(): + data, etag = self._load() + ceiling = min(int(data.get("ceiling", self.ceiling)), self.ceiling) + reconciled = [] + for e in data.get("executions") or []: + if e.get("status") == "reserved" and e.get("oxylabs_jobs_submitted") is None: + e["status"] = "unfinished" + e["reconciled_utc"] = utc() + e["reconciled_note"] = (f"no verified final update; counted at its full reservation of " + f"{e.get('reserved_jobs')} jobs until a person corrects it from evidence") + reconciled.append(e.get("folder")) + used = jobs_counted(data) + left = ceiling - used + info = {"ledger": self.store.describe(), "ceiling": ceiling, "counted_before": used, "left_before": left, + "reconciled_unfinished": reconciled} + if left <= 0: + if reconciled: + data["total_jobs_counted"] = used + self._commit(data, etag) + info["cap_this_run"] = 0 + return info + cap = left if wanted is None else min(int(wanted), left) + self.reservation_id = uuid.uuid4().hex + data.setdefault("executions", []).append({ + "folder": folder, "package": package, "host": socket.gethostname(), "reservation_id": self.reservation_id, + "status": "reserved", "reserved_utc": utc(), "reserved_jobs": cap, "oxylabs_jobs_submitted": None}) + data["ceiling"] = ceiling + data["total_jobs_counted"] = jobs_counted(data) + self._commit(data, etag) + info.update(cap_this_run=cap, reservation_id=self.reservation_id, ledger_sha256_after_reserve=self.last_sha256) + return info + + def finalize(self, jobs: int, exit_code: int, finished_utc: str) -> dict: + """Record the jobs actually dispatched on a fresh read, matched by reservation id, and verify it.""" + with self._locked(): + data, etag = self._load() + entry = next((e for e in data.get("executions") or [] if e.get("reservation_id") == self.reservation_id), None) + if entry is None: + raise LedgerError(f"reservation {self.reservation_id} is not in the ledger; left unresolved") + entry.update(status="final", oxylabs_jobs_submitted=int(jobs), exit_code=exit_code, finished_utc=finished_utc) + data["total_jobs_counted"] = jobs_counted(data) + return self._commit(data, etag) --- a/scraper/marianos_scraper/probe.py +++ b/scraper/marianos_scraper/probe.py @@ -21,10 +21,12 @@ echoed fulfilment context (laf: modality type, storeId, facilityId) is not the one sent -> stop before the coupon request, exit 2 -Cumulative allowance (--ledger FILE): before any request the probe reads the ledger of every earlier execution, -reserves what is left of the ceiling (default 15) as this run's cap, and writes that reservation; at the end (also after -a stop or the watchdog) it records the jobs actually dispatched. A process that is killed leaves its full reservation -counted. A new folder, session or process can never reset the total. +Cumulative allowance (--ledger blob: or FILE, see allowance.py): before any request the probe takes the ledger's +exclusive lease, reconciles any unfinished reservation (kept at its full count), reserves what is left of the ceiling +(default 15) as this run's cap, writes it with If-Match on the ETag and reads it back. Only a verified reservation lets +the probe send anything. At the end (also after a stop or the watchdog) it records the jobs actually dispatched, on a +fresh read matched by reservation id, verified the same way; if that cannot be verified the run ends with exit 2 and +the reservation stays counted. A new folder, session, process or machine can never reset the total. Session replacement (Oxylabs): a failed session is never reused. When the transport replaced its session after the store was verified, the store is verified again on the new session (counted) and the products are asked again; only @@ -71,6 +73,7 @@ from .log import log from .mapper import FIELD_ORDER, POINTS_OFFER, SiteMapper, fix_text, usd from .runner import code_fingerprint +from .allowance import Allowance, LedgerError, encode, jobs_counted, open_store from .site_source import RequestCapReached, SiteSession, SiteUnavailable, echoed_context try: @@ -144,19 +147,7 @@ "publicIpAddresses": ips} -def _ledger_used(ledger: dict) -> int: - """Jobs counted so far: dispatched where recorded, else the reservation (a killed run keeps its full cap).""" - total = 0 - for e in ledger.get("executions", []): - jobs = e.get("oxylabs_jobs_submitted") - total += int(jobs) if jobs is not None else int(e.get("reserved_jobs", 0)) - return total - - -def _write_json_atomic(path: Path, data: dict) -> None: - tmp = path.with_name(path.name + ".tmp") - tmp.write_text(json.dumps(data, indent=1, ensure_ascii=False), encoding="utf-8") - os.replace(tmp, path) +_ledger_used = jobs_counted def _is_clip_coupon(c: dict) -> bool: @@ -225,7 +216,7 @@ class Probe: def __init__(self, cfg, location, site: SiteSession, upcs: list[str], out: Path, *, transport: str, egress_ip: str = "", roles: list[str] | None = None, azure_imds: bool = False, - ledger: Path | None = None, ledger_ceiling: int = 15): + ledger=None, ledger_ceiling: int = 15): self.cfg, self.loc, self.site, self.upcs = cfg, location, site, [u.strip() for u in upcs if u.strip()] roles = [r.strip() for r in (roles or []) if r.strip()] if roles and len(roles) != len(self.upcs): @@ -238,10 +229,11 @@ self.transport = transport self.egress_ip = egress_ip self.azure_imds = azure_imds - self.ledger_path = Path(ledger) if ledger else None - self.ledger_ceiling = ledger_ceiling - self._ledger: dict | None = None - self._ledger_entry: dict | None = None + if ledger is None or ledger == "": + self.allowance = None + else: + store = open_store(str(ledger)) if isinstance(ledger, (str, Path)) else ledger + self.allowance = Allowance(store, ceiling=ledger_ceiling) self._store_generation = -1 # session generation on which the store was last verified self.mapper = SiteMapper(cfg.settings) self.site.capture = True # keep every attempt with its exact answer for raw/ @@ -251,6 +243,7 @@ self.coupon_origin: dict[str, str] = {} # coupon id -> raw file + index self._lock = threading.Lock() self._finished = False + self._exit_override: int | None = None # set when the ledger's final update cannot be verified self.report: dict = {"version": __version__, "code_fingerprint": code_fingerprint(), "transport_name": transport, "location": location.key, "location_id": location.location_id, "facility_id_expected": location.facility_id, "zipcode": location.zipcode, @@ -292,7 +285,7 @@ if self.azure_imds: self.report["azure_instance"] = _azure_instance() code = 2 - if self.ledger_path and not self._reserve_allowance(): + if self.allowance and not self._reserve_allowance(): self.report["transport"] = "stopped" self._finish(2) if watchdog: @@ -329,30 +322,25 @@ self._finish(code) if watchdog: watchdog.cancel() + if self._exit_override is not None: + code = self._exit_override return code def _reserve_allowance(self) -> bool: - """Read the cumulative ledger, cap this run at what is left, and record the reservation before any request.""" - led = (json.loads(self.ledger_path.read_text(encoding="utf-8")) if self.ledger_path.exists() - else {"ceiling": self.ledger_ceiling, "executions": []}) - ceiling = min(int(led.get("ceiling", self.ledger_ceiling)), self.ledger_ceiling) - used = _ledger_used(led) - left = ceiling - used - self.report["allowance"] = {"ledger": str(self.ledger_path), "ceiling": ceiling, "counted_before": used, "left_before": left} - if left <= 0: - self.report["stop_reason"] = f"allowance: {used} of {ceiling} jobs already counted; nothing sent" + """Reserve this run's cap in the shared ledger, durably and verified, before any request.""" + try: + info = self.allowance.reserve(self.out.name, f"v{__version__}", self.site.max_requests) + except LedgerError as exc: + self.report["allowance"] = {"ledger": self.allowance.store.describe(), "error": str(exc)} + self.report["stop_reason"] = f"allowance: ledger {type(exc).__name__}: {exc}; nothing sent" return False - cap = left if self.site.max_requests is None else min(self.site.max_requests, left) - self.site.max_requests = cap - self.report["caps"]["max_requests"] = cap - self.report["allowance"]["cap_this_run"] = cap - entry = {"folder": self.out.name, "package": f"v{__version__}", "started_utc": _stamp()["utc"], - "reserved_jobs": cap, "oxylabs_jobs_submitted": None} - led.setdefault("executions", []).append(entry) - led["ceiling"] = ceiling - led["total_jobs_counted"] = _ledger_used(led) - _write_json_atomic(self.ledger_path, led) - self._ledger, self._ledger_entry = led, entry + self.report["allowance"] = info + if info["cap_this_run"] <= 0: + self.report["stop_reason"] = (f"allowance: {info['counted_before']} of {info['ceiling']} jobs already " + f"counted; nothing sent") + return False + self.site.max_requests = info["cap_this_run"] + self.report["caps"]["max_requests"] = info["cap_this_run"] return True def _on_verified_session(self, ask): @@ -665,15 +653,22 @@ "session_replacements": self.site.session_generation, "note": "every dispatched attempt counts; errors = every failed attempt (provider or retailer); " "browser start-up page assets are not JSON requests and are not counted"} - if self._ledger is not None: - self._ledger_entry["oxylabs_jobs_submitted"] = self.site.stats.requests - self._ledger_entry["finished_utc"] = self.report["finished"]["utc"] - self._ledger_entry["exit_code"] = code - self._ledger["total_jobs_counted"] = _ledger_used(self._ledger) - _write_json_atomic(self.ledger_path, self._ledger) + if self.allowance is not None and self.allowance.reservation_id: self.report["allowance"]["jobs_this_run"] = self.site.stats.requests - self.report["allowance"]["counted_after"] = self._ledger["total_jobs_counted"] - self._write("cumulative-usage.json", json.dumps(self._ledger, indent=1, ensure_ascii=False)) + try: + led = self.allowance.finalize(self.site.stats.requests, code, self.report["finished"]["utc"]) + self.report["allowance"]["final_update"] = "verified" + self.report["allowance"]["counted_after"] = led["total_jobs_counted"] + self.report["allowance"]["ledger_sha256_after_final"] = self.allowance.last_sha256 + self._write("cumulative-usage.json", encode(led).decode("utf-8")) + except LedgerError as exc: + # the reservation stays counted in full; the run is reported as stopped and the ledger as unresolved + self.report["allowance"]["final_update"] = f"UNRESOLVED: {type(exc).__name__}: {exc}" + self.report["stop_reason"] = (self.report.get("stop_reason") or "") + f"; ledger final update not verified ({exc}): reservation stays counted, ledger unresolved" + self.report["transport"] = "stopped" + code = 2 + self.report["exit_code"] = code + self._exit_override = code index = [] for att in list(self.site.attempts): entry = {k: att.get(k) for k in ("seq", "t", "elapsed_s", "path", "attempt", "transport", "outcome", "status", --- a/scraper/tests/test_oxylabs_probe.py +++ b/scraper/tests/test_oxylabs_probe.py @@ -826,3 +826,221 @@ assert s.outcomes == {"provider_code_15002": 1, "provider_http_400": 1} ids = [{c["key"]: c["value"] for c in p["context"]}["session_id"] for p in http.posts] assert ids[0] != ids[1] # the failed session was not reused for the retry + + +# --- v2.2.16: durable shared ledger in Blob (lease + If-Match + read-back), reconciliation ---------------------- + +from marianos_scraper import allowance as allowance_mod # noqa: E402 +from marianos_scraper.allowance import BlobStore, FileStore, LedgerError # noqa: E402 + +PRIOR = {"ceiling": 15, "executions": [ + {"folder": "probe/oxylabs-probe-20261027T030743Z/", "oxylabs_jobs_submitted": 1}, + {"folder": "probe/oxylabs-probe-20261030T033021Z/", "package": "v2.2.14", "oxylabs_jobs_submitted": 3}], + "total_jobs_submitted": 4} + + +class HttpErr(Exception): + def __init__(self, status): + super().__init__(f"HTTP {status}") + self.status_code = status + + +class FakeBlob: + """The calls BlobStore makes on azure.storage.blob.BlobClient, with Azure's lease and If-Match rules.""" + url = "https://acct.blob.core.windows.net/scraper-test/probe/cumulative-usage.json" + + def __init__(self, data=None): + self.body = None if data is None else allowance_mod.encode(data) + self.etag = 0 + self.lease = None + self.uploads = 0 + self.store_differently = None # callable(body) -> bytes: storage keeps other bytes than were sent + self.read_back_differently = None # callable(body) -> bytes: an unleased read returns other bytes + + class _Lease: + def __init__(self, blob, lid): + self.blob, self.id = blob, lid + + def release(self): + if self.blob.lease == self.id: + self.blob.lease = None + + class _Download: + def __init__(self, body, etag): + self._b = body + self.properties = type("P", (), {"etag": f'"0x{etag}"'})() + + def readall(self): + return self._b + + def acquire_lease(self, lease_duration): + if self.body is None: + raise HttpErr(404) + if self.lease is not None: + raise HttpErr(409) + self.lease = f"lease-{self.etag}-{self.uploads}" + return self._Lease(self, self.lease) + + def download_blob(self, lease=None): + if self.body is None: + raise HttpErr(404) + body = self.body + if lease is None and self.read_back_differently: + body = self.read_back_differently(body) + return self._Download(body, self.etag) + + def upload_blob(self, body, *, overwrite, lease, etag, match_condition): + if self.lease is not None and (lease is None or lease.id != self.lease): + raise HttpErr(412) # LeaseIdMissing / LeaseIdMismatch + if etag != f'"0x{self.etag}"': + raise HttpErr(412) # ConditionNotMet + self.body = self.store_differently(body) if self.store_differently else body + self.etag += 1 + self.uploads += 1 + return {"etag": f'"0x{self.etag}"'} + + def data(self): + return json.loads(self.body.decode("utf-8")) + + +def blob_store(blob, **kw): + return BlobStore(blob, lock_wait=kw.pop("lock_wait", 4), sleep=lambda s: None, **kw) + + +def run_ledger_probe(tmp_path, name, ledger, products, max_requests=14): + s, http = session(lambda p: route(p, products=products), max_requests=max_requests) + code = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / name, + transport="oxylabs", ledger=ledger, ledger_ceiling=15).run() + return code, s, http + + +def test_blob_ledger_reserves_before_dispatch_and_records_the_final_count(tmp_path): + blob = FakeBlob(PRIOR) + products = fixture_products() + s, http = session(lambda p: route(p, products=products), max_requests=14) + posts_at_upload = [] + orig = blob.upload_blob + + def watch(body, **kw): + posts_at_upload.append(len(http.posts)) + return orig(body, **kw) + blob.upload_blob = watch + code = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "e1", + transport="oxylabs", ledger=blob_store(blob)).run() + assert code in (0, 1) and s.max_requests == 11 # 15 - 4 counted, though 14 was asked + assert posts_at_upload == [0, len(http.posts)] # reservation before any POST, final after the last + led = blob.data() + e = led["executions"][-1] + assert e["status"] == "final" and e["reserved_jobs"] == 11 and e["oxylabs_jobs_submitted"] == len(http.posts) + assert led["executions"][:2] == PRIOR["executions"] # earlier executions untouched + assert led["total_jobs_counted"] == 4 + len(http.posts) and blob.lease is None + report = json.loads((tmp_path / "e1" / "report.json").read_text()) + assert report["allowance"]["final_update"] == "verified" and report["allowance"]["reservation_id"] == e["reservation_id"] + assert (tmp_path / "e1" / "cumulative-usage.json").read_bytes() == blob.body # evidence copy = the verified blob + + +def test_interruption_after_dispatch_before_final_update_is_counted_in_full_by_a_fresh_session(tmp_path, monkeypatch): + """A run reserves 5, dispatches jobs, and dies before its final ledger update and before the evidence upload. + A fresh session (new process, folder and Oxylabs session) counts the full reservation and resets nothing.""" + blob = FakeBlob(PRIOR) + products = fixture_products() + s, http = session(lambda p: route(p, products=products), max_requests=5) + p = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "killed", + transport="oxylabs", ledger=blob_store(blob)) + monkeypatch.setattr(p, "_finish", lambda code, lock_timeout=None: None) # the process is killed here + p.run() + assert len(http.posts) >= 2 and not (tmp_path / "killed" / "report.json").exists() + killed = blob.data()["executions"][-1] + assert killed["status"] == "reserved" and killed["reserved_jobs"] == 5 and killed["oxylabs_jobs_submitted"] is None + # the fresh session + code2, s2, http2 = run_ledger_probe(tmp_path, "fresh", blob_store(blob), products) + led = blob.data() + assert led["executions"][2]["status"] == "unfinished" # reconciled, kept at its full reservation + assert led["executions"][2]["oxylabs_jobs_submitted"] is None and led["executions"][2]["reserved_jobs"] == 5 + assert s2.max_requests == 15 - 4 - 5 # 6 left: the jobs it cannot see are not guessed + report = json.loads((tmp_path / "fresh" / "report.json").read_text()) + assert report["allowance"]["reconciled_unfinished"] == ["killed"] and report["allowance"]["counted_before"] == 9 + assert led["total_jobs_counted"] == 9 + len(http2.posts) + # when nothing is left, a further session reconciles, records that, and sends nothing + blob2 = FakeBlob({"ceiling": 15, "executions": PRIOR["executions"] + [ + {"folder": "k", "reservation_id": "r", "status": "reserved", "reserved_jobs": 11, "oxylabs_jobs_submitted": None}]}) + code3, _, http3 = run_ledger_probe(tmp_path, "after", blob_store(blob2), products) + assert code3 == 2 and not http3.posts and blob2.data()["executions"][-1]["status"] == "unfinished" + + +def test_a_held_lease_stops_the_probe_before_anything_is_sent(tmp_path): + blob = FakeBlob(PRIOR) + blob.lease = "held-by-another-execution" + code, s, http = run_ledger_probe(tmp_path, "e", blob_store(blob), fixture_products()) + assert code == 2 and not http.posts and blob.uploads == 0 + report = json.loads((tmp_path / "e" / "report.json").read_text()) + assert "409" in report["stop_reason"] and "nothing sent" in report["stop_reason"] + + +def test_a_stale_copy_cannot_overwrite_a_newer_ledger(): + blob = FakeBlob(PRIOR) + store = blob_store(blob) + store.lock() + body, etag = store.read() + blob.etag += 1 # changed after our read + with pytest.raises(LedgerError, match="ETag"): + store.write(b"{}", etag) + store.unlock() + assert blob.data() == PRIOR + + +def test_a_write_without_the_lease_is_refused(): + blob = FakeBlob(PRIOR) + a = blob_store(blob) + a.lock() + b = blob_store(blob) # never took the lease + with pytest.raises(LedgerError): + b.write(b"{}", f'"0x{blob.etag}"') + a.unlock() + assert blob.data() == PRIOR + + +def test_read_back_mismatch_on_reservation_stops_before_dispatch(tmp_path): + blob = FakeBlob(PRIOR) + blob.read_back_differently = lambda b: b[:-2] + code, s, http = run_ledger_probe(tmp_path, "e", blob_store(blob), fixture_products()) + assert code == 2 and not http.posts + report = json.loads((tmp_path / "e" / "report.json").read_text()) + assert "LedgerMismatch" in report["stop_reason"] and "unresolved" in report["stop_reason"] + + +def test_read_back_mismatch_on_final_update_stops_and_stays_unresolved(tmp_path): + blob = FakeBlob(PRIOR) + products = fixture_products() + s, http = session(lambda p: route(p, products=products), max_requests=14) + p = Probe(CFG, CFG.location("marianos-vernon-hills"), s, [products[0]["id"]], tmp_path / "e", + transport="oxylabs", ledger=blob_store(blob)) + orig = p.allowance.finalize + + def corrupt_then_finalize(*a, **kw): + blob.store_differently = lambda b: b.replace(b'"final"', b'"fin4l"') # storage keeps other bytes + return orig(*a, **kw) + p.allowance.finalize = corrupt_then_finalize + code = p.run() + assert code == 2 and http.posts # it had dispatched; the run is still reported stopped + report = json.loads((tmp_path / "e" / "report.json").read_text()) + assert report["exit_code"] == 2 and report["transport"] == "stopped" + assert report["allowance"]["final_update"].startswith("UNRESOLVED: LedgerMismatch") + assert not (tmp_path / "e" / "cumulative-usage.json").exists() # no unverified copy shown as evidence + + +def test_missing_or_malformed_ledger_fails_closed(tmp_path): + code, _, http = run_ledger_probe(tmp_path, "missing", blob_store(FakeBlob(None)), fixture_products()) + assert code == 2 and not http.posts + assert "not found" in json.loads((tmp_path / "missing" / "report.json").read_text())["stop_reason"] + bad = FakeBlob({"ceiling": 15, "executions": [{"folder": "x"}]}) + code, _, http = run_ledger_probe(tmp_path, "bad", blob_store(bad), fixture_products()) + assert code == 2 and not http.posts and bad.uploads == 0 + + +def test_file_ledger_uses_an_exclusive_lock(tmp_path): + path = tmp_path / "cumulative-usage.json" + path.write_bytes(allowance_mod.encode(PRIOR)) + (tmp_path / "cumulative-usage.json.lock").write_text("other execution") + code, _, http = run_ledger_probe(tmp_path, "e", FileStore(path, lock_wait=0, sleep=lambda s: None), fixture_products()) + assert code == 2 and not http.posts and json.loads(path.read_text()) == PRIOR --- a/scraper/tests/test_scraper.py +++ b/scraper/tests/test_scraper.py @@ -360,4 +360,145 @@ runner._coupon_index = {} unit = {"id": "10-natural-organic-18007", "category": "Natural Organic", "sub_category": "x", "taxonomy": "18007", "expected": 340} res = runner.collect_unit(unit, {"zipcode": "60601", "store_name": "Mariano's", "store_location": "x", "store_logo": "", "location_id": "53100503"}) - assert res["found_upcs"] == 340 and res["api_total"] == 340+ assert res["found_upcs"] == 340 and res["api_total"] == 340 + +# --- v2.2.16: full-run retention across repeat runs ---------------------------------------------------------- + +class ChangingStore: + """The whole website for SiteRunner.run(), offline: every taxonomy root has one leaf of five products. Between + runs the test changes `unavailable` (listed, not sellable) and `absent` (no longer listed at all).""" + STORE = {"locationId": "53100503", "facilityId": "15096", "brand": "MARIANOS", "storeNumber": "00503", + "vanityName": "Mariano's Lakeshore East", + "locale": {"address": {"name": "Lakeshore East", "addressLines": ["333 E Benton Pl"], "cityTown": "CHICAGO", + "stateProvince": "IL", "postalCode": "60601"}}} + + def __init__(self): + import types + self.unavailable, self.absent = set(), set() + self.calls, self.max_requests, self.max_seconds = [], None, None + self.stats = types.SimpleNamespace(requests=0, retries=0, errors=0, total_seconds=0.0) + + @staticmethod + def leaf(root): + return root if len(root) >= 5 else root + "001" + + @staticmethod + def upcs_of(leaf): + return [f"{leaf}{i:02d}".rjust(13, "0") for i in range(1, 6)] + + def listed(self, leaf): + return [u for u in self.upcs_of(leaf) if u not in self.absent] + + def open(self): + pass + + def close(self): + pass + + def store(self, zipcode, location_id): + return self.STORE + + def set_store(self, *a, **kw): + pass + + def check_context(self, products, path): + pass + + def search(self, location_id, taxonomy, offset=0, size=100, brand="", facets=False, brand_facet=False): + if facets: + if len(taxonomy) >= 5: + n = len(self.listed(taxonomy)) if taxonomy in self._leaves() else 0 + return [], n, [{"id": taxonomy, "name": taxonomy, "count": n, "children": []}] + leaf = self.leaf(taxonomy) + n = len(self.listed(leaf)) + return [], n, [{"id": taxonomy, "name": taxonomy, "count": n, + "children": [{"id": leaf, "name": f"Aisle {leaf}", "count": n, "children": []}]}] + ids = self.listed(taxonomy) + return [{"upc": u} for u in ids[offset:offset + size]], len(ids), {} + + def _leaves(self): + import json as _j + tax = _j.loads((Path(CFG.config_dir) / "site_taxonomy.json").read_text(encoding="utf-8"))["categories"] + return {r for spec in tax.values() for r in spec.get("roots", []) if len(r) >= 5} + + def products(self, upcs, verified=True): + import copy + out = [] + for u in upcs: + p = copy.deepcopy(SITE["regular"]) + p["id"] = u + p["item"] = {**p.get("item", {}), "upc": u, "description": f"Roundy's Product {u}"} + p["price"]["storePrices"]["sourceLocationId"] = "53100503" + if u in self.unavailable: + for s in p.get("fulfillmentSummaries") or []: + s.setdefault("availability", {})["sellable"] = False + out.append(p) + return out + + def all_coupons(self, kind): + return ([COUPON] if kind == "standard" else []), {} + + def coupons_for(self, upcs): + return [] + + +def test_unavailable_and_later_absent_products_survive_repeat_runs_with_their_real_last_observation(tmp_path, monkeypatch): + """Three full runs of SiteRunner.run() on one store (LocalStorage, no network): + run 1 (Oct 01) everything listed; run 2 (Oct 02) GONE is no longer listed and OOS is listed but not sellable; + run 3 (Oct 03) GONE is still absent and OOS is absent too. + Every product stays in every published output; a carried record keeps the crawl_date of its last observation + (never today's), availability.jsonl tells fresh from carried, and nothing is invented.""" + from datetime import datetime as _dt + from marianos_scraper import site_runner as sr + from marianos_scraper.site_runner import SiteRunner + for k, val in {"min_records_per_category": 1, "min_ppm_share": 0.0, "min_total_records": 10}.items(): + monkeypatch.setitem(CFG.settings["validation"], k, val) + CFG.storage_root = str(tmp_path) + storage = LocalStorage(str(tmp_path)) + site = ChangingStore() + leaf = site.leaf("02") # Dairy + GONE, OOS = site.upcs_of(leaf)[0], site.upcs_of(leaf)[1] + + day = {"v": "2026-10-01"} + + class FixedDay(_dt): + @classmethod + def now(cls, tz=None): + return _dt.fromisoformat(day["v"] + "T12:00:00+00:00") + monkeypatch.setattr(sr, "datetime", FixedDay) + + def run(d): + day["v"] = d + runner = SiteRunner(CFG, CFG.location("marianos-lakeshore-east"), site=site, api=FakeApi(), storage=storage) + report = runner.run() + assert report["status"] == "published", [c for c in report["validation"]["checks"] if not c["passed"]] + out = f"runs/marianos-lakeshore-east/{report['run_id']}/" + products = {r["upc"]: r for r in json.loads(storage.read_text(out + "products.json"))} + avail = {a["upc"]: a for a in map(json.loads, storage.read_text(out + "availability.jsonl").splitlines())} + latest = {r["upc"] for r in json.loads(storage.read_text("latest/marianos-lakeshore-east/products.json"))} + assert set(products) == latest + return report["run_id"], products, avail + + run1, p1, a1 = run("2026-10-01") + assert {GONE, OOS} <= set(p1) and a1[GONE]["observation"] == a1[OOS]["observation"] == "fresh" + total = len(p1) + + site.absent, site.unavailable = {GONE}, {OOS} + run2, p2, a2 = run("2026-10-02") + assert len(p2) == total # nothing dropped + assert p2[GONE]["crawl_date"] == "2026-10-01" # carried: its real last observation + assert a2[GONE] == {**a2[GONE], "observation": "carried_forward", "last_observed_run": run1, "crawl_date": "2026-10-01"} + assert p2[OOS]["crawl_date"] == "2026-10-02" and a2[OOS]["observation"] == "fresh" # listed: a fresh record + assert a2[OOS]["fulfillment_sellable"] == [] and a2[OOS]["last_observed_run"] == run2 + + site.absent = {GONE, OOS} + run3, p3, a3 = run("2026-10-03") + assert len(p3) == total + assert p3[GONE]["crawl_date"] == "2026-10-01" and a3[GONE]["last_observed_run"] == run1 # not reset by run 2 + assert p3[OOS]["crawl_date"] == "2026-10-02" and a3[OOS]["last_observed_run"] == run2 + assert a3[GONE]["observation"] == a3[OOS]["observation"] == "carried_forward" + fresh = [u for u, a in a3.items() if a["observation"] == "fresh"] + assert len(fresh) == total - 2 and all(p3[u]["crawl_date"] == "2026-10-03" for u in fresh) + assert p3[GONE] == p1[GONE] and p3[OOS] == p2[OOS] # the record itself is the last observation, unchanged + changes = json.loads(storage.read_text(f"runs/marianos-lakeshore-east/{run3}/availability_changes.json")) + assert {c["upc"] for c in changes if c["kept_as"].startswith("carried_forward")} == {GONE, OOS}