#!/usr/bin/env python3 """Size a backfill, a normal day or a replay against an API budget. Usage: python3 scripts/size_sync.py --records N [--batch 1] [--calls-per-batch 1] [--extra-calls-per-record 0] [--readback-calls-per-batch 0] [--retry-overhead 0.10] [--preset NAME | --window-limit N --window-seconds S] [--daily-limit N] [--remaining N] [--reserve FRACTION | --reserve-calls N] [--other-calls N] [--target-utilization 0.7] [--window-hours H] [--search-results N] [--compare-single] [--json] python3 scripts/size_sync.py --list-presets Computation (references/rate-limits-and-backfills.md): batches = ceil(records / batch) calls = batches x calls_per_batch + records x extra_calls_per_record + batches x readback_calls_per_batch total = ceil(calls x (1 + retry_overhead)) rate = tightest window (limit / seconds) x target utilisation duration = total / rate available = remaining - reserve - other expected consumers (reserve defaults to 20% of the daily allowance when one is known) records that fit now = floor(available / (total / records)) Findings: RL01 FAIL total above available (prints what fits now and what to defer); WARN total above 50% of available, or no reserve could be applied RL02 WARN target utilisation above 0.7; WARN duration above --window-hours RL04 WARN batch above the preset's batch maximum; INFO single-record calls where the endpoint takes batches RL05 FAIL search results above the preset's result cap (prints windows needed at under 9,000 each); WARN at 9,000 or more Presets come from ../assets/rate-limit-presets.json and print their check date and source URL. Salesforce has no preset: its allocation is org-specific, so read DailyApiRequests from the REST Limits resource and pass --daily-limit and --remaining. Exit 1 on any FAIL, 2 on bad input, 0 otherwise. The script calls no API and changes nothing. """ import argparse import json import math import sys from pathlib import Path PRESETS_FILE = Path(__file__).resolve().parent.parent / "assets" / "rate-limit-presets.json" DEFAULT_RESERVE = 0.2 SEARCH_WINDOW_TARGET = 9000 class InputError(Exception): pass def load_presets(): try: with open(PRESETS_FILE, encoding="utf-8") as handle: return json.load(handle) except (OSError, ValueError) as exc: raise InputError(f"cannot read presets file {PRESETS_FILE.name}: {exc}") def fmt(n, digits=0): return f"{n:,.{digits}f}" def ceil6(value): """Ceil after rounding to 6 decimals, so 960 x 1.1 is 1,056, not 1,057.""" return math.ceil(round(value, 6)) def duration_text(seconds): text = f"about {fmt(round(seconds))} seconds" if seconds >= 3600: text += f" ({seconds / 3600:.1f} hours)" elif seconds >= 60: text += f" ({seconds / 60:.1f} minutes)" return text def parse_args(argv): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument("--records", type=int, help="records to send in this load") p.add_argument("--batch", type=int, default=1, help="records per write call (default 1)") p.add_argument("--calls-per-batch", type=float, default=1, help="write calls per batch (default 1)") p.add_argument("--extra-calls-per-record", type=float, default=0, help="lookups or association calls per record") p.add_argument("--readback-calls-per-batch", type=float, default=0, help="verification reads per batch") p.add_argument("--retry-overhead", type=float, default=0.10, help="extra share of calls for retries (default 0.10)") p.add_argument("--preset", help="name from assets/rate-limit-presets.json") p.add_argument("--window-limit", type=float, help="requests allowed per window (custom limits)") p.add_argument("--window-seconds", type=float, help="window length in seconds (custom limits)") p.add_argument("--daily-limit", type=int, help="daily allowance; overrides the preset's figure") p.add_argument("--remaining", type=int, help="calls left in the current daily period, read from headers or the Limits resource") group = p.add_mutually_exclusive_group() group.add_argument("--reserve", type=float, help="fraction of the daily allowance kept for other work (default 0.2)") group.add_argument("--reserve-calls", type=int, help="calls kept for other work, as a number") p.add_argument("--other-calls", type=int, default=0, help="calls other consumers will make in the same period") p.add_argument("--target-utilization", type=float, default=0.7, help="share of the tightest window to plan at (default 0.7)") p.add_argument("--window-hours", type=float, help="hours available for the load") p.add_argument("--search-results", type=int, help="results a search-based recovery must page through") p.add_argument("--compare-single", action="store_true", help="also size the writes one record per call") p.add_argument("--list-presets", action="store_true", help="list presets and exit") p.add_argument("--json", action="store_true", help="print the result as JSON") return p.parse_args(argv) def validate(args): if args.records is None and args.search_results is None: raise InputError("pass --records (and --search-results for a search-based recovery)") checks = [ (args.records is not None and args.records < 1, "--records must be 1 or more"), (args.batch < 1, "--batch must be 1 or more"), (args.calls_per_batch < 0 or args.extra_calls_per_record < 0 or args.readback_calls_per_batch < 0, "call counts cannot be negative"), (args.retry_overhead < 0, "--retry-overhead cannot be negative"), (not 0 < args.target_utilization <= 1, "--target-utilization must be above 0 and at most 1"), (args.reserve is not None and not 0 <= args.reserve < 1, "--reserve is a fraction from 0 to below 1"), ((args.window_limit is None) != (args.window_seconds is None), "pass --window-limit and --window-seconds together"), (args.window_seconds is not None and args.window_seconds <= 0, "--window-seconds must be above 0"), (args.search_results is not None and args.search_results < 0, "--search-results cannot be negative"), ] for failed, message in checks: if failed: raise InputError(message) def resolve_preset(args, data): if not args.preset: return None name = args.preset.strip().lower() if name.startswith("salesforce"): raise InputError("no Salesforce preset: the daily allocation is org-specific and counted over a rolling " "24 hours. Read DailyApiRequests Max and Remaining from the REST Limits resource " "(needs View Setup and Configuration) and pass --daily-limit and --remaining.") presets = data.get("presets", {}) if name not in presets: raise InputError(f"unknown preset '{args.preset}'. Available: {', '.join(sorted(presets))}") return name, presets[name] def main(argv=None): args = parse_args(argv) try: data = load_presets() if args.list_presets: for name, item in sorted(data.get("presets", {}).items()): windows = ", ".join(f"{fmt(w['limit'])} per {fmt(w['seconds'])} s" for w in item["windows"]) print(f"{name}: {item['vendor']}; {windows}; daily {fmt(item['daily_limit']) if item.get('daily_limit') else 'not set'}; " f"checked {item['checked']}") return 0 validate(args) preset = resolve_preset(args, data) except InputError as exc: print(f"error: {exc}", file=sys.stderr) return 2 findings, lines, result = [], [], {} def add(level, rule, message): findings.append({"level": level, "rule": rule, "message": message}) name, item = preset if preset else (None, {}) windows = list(item.get("windows", [])) if args.window_limit is not None: windows = [{"limit": args.window_limit, "seconds": args.window_seconds}] daily = args.daily_limit if args.daily_limit is not None else item.get("daily_limit") if name: lines.append(f"Preset: {name} ({item['vendor']}: {item['scope']}), checked {item['checked']}") lines.append(f" Source: {item['source']}") lines.append(f" Reset: {item['daily_reset']}") if item.get("notes"): lines.append(f" Notes: {item['notes']}") else: lines.append("Preset: none; limits from the command line. Record the vendor's reset rule in the spec " "(Salesforce counts a rolling 24 hours).") result.update(preset=name, preset_checked=item.get("checked"), preset_source=item.get("source")) rate = planned = None if windows: tight = min(windows, key=lambda w: w["limit"] / w["seconds"]) rate = tight["limit"] / tight["seconds"] planned = rate * args.target_utilization lines.append(f"Rate: tightest window {fmt(tight['limit'])} per {fmt(tight['seconds'], 0 if float(tight['seconds']).is_integer() else 1)} s " f"= {rate:.2f}/s; planned at {args.target_utilization:.0%} = {planned:.2f}/s") result.update(rate_per_second=rate, planned_rate_per_second=planned) if args.records is not None: records, batch = args.records, args.batch batches = math.ceil(records / batch) write = batches * args.calls_per_batch extra = records * args.extra_calls_per_record readback = batches * args.readback_calls_per_batch base = write + extra + readback total = ceil6(base * (1 + args.retry_overhead)) lines.insert(0, f"Calls: {fmt(write)} write + {fmt(extra)} extra + {fmt(readback)} read-back = {fmt(base)}; " f"with {args.retry_overhead:.0%} retry overhead = {fmt(total)}") lines.insert(0, f"Load: {fmt(records)} records in batches of {fmt(batch)} = {fmt(batches)} batches") result.update(records=records, batch=batch, batches=batches, write_calls=write, extra_calls=extra, readback_calls=readback, total_calls=total) if planned: seconds = total / planned lines.append(f"Duration: {duration_text(seconds)} at the planned rate") result["duration_seconds"] = round(seconds, 1) if args.window_hours is not None: if seconds > args.window_hours * 3600: add("WARN", "RL02", f"duration {duration_text(seconds)} exceeds the {args.window_hours:g}-hour window; " "split the load across periods or raise the batch size") else: add("PASS", "RL02", f"duration fits the {args.window_hours:g}-hour window") else: lines.append("Duration: not computed (no window limit; pass --preset or --window-limit and --window-seconds)") if daily: share = total / daily lines.append(f"Daily allowance: {fmt(daily)}; this load uses {share:.2%}") result["daily_share"] = round(share, 6) else: lines.append("Daily allowance: not known; pass --daily-limit to see the share") if args.compare_single and batch > 1: single = records * args.calls_per_batch + extra single_total = ceil6(single * (1 + args.retry_overhead)) text = (f"Single-record comparison: {fmt(single)} write calls one record per call, {fmt(single_total)} with " f"retry overhead (read-back excluded)") if daily: text += f"; {single_total / daily:.2%} of the daily allowance" if planned: text += f"; {duration_text(single_total / planned)} at {planned:.2f}/s" lines.append(text) result["single_record_total_calls"] = single_total if args.remaining is not None: if args.reserve_calls is not None: reserve, reserve_text = args.reserve_calls, f"reserve {fmt(args.reserve_calls)}" elif daily: fraction = args.reserve if args.reserve is not None else DEFAULT_RESERVE reserve = math.floor(daily * fraction) reserve_text = f"reserve {fmt(reserve)} ({fraction:.0%} of {fmt(daily)}{', default' if args.reserve is None else ''})" else: reserve, reserve_text = 0, "reserve 0" add("WARN", "RL01", "no reserve applied: no daily allowance is known; pass --reserve-calls with the calls other work needs") available = args.remaining - reserve - args.other_calls lines.append(f"Budget: remaining {fmt(args.remaining)} - {reserve_text} - other consumers {fmt(args.other_calls)} " f"= available {fmt(available)}") result.update(remaining=args.remaining, reserve_calls=reserve, other_calls=args.other_calls, available=available) if total > available: per_record = total / records fits = max(0, math.floor(round(available / per_record, 6))) if available > 0 else 0 add("FAIL", "RL01", f"{fmt(total)} calls exceed the available {fmt(max(available, 0))}; {fmt(fits)} records fit now " f"({per_record:g} calls each), defer {fmt(records - fits)} to the next period after the reset") result.update(records_fit_now=fits, records_deferred=records - fits) elif total > 0.5 * available: add("WARN", "RL01", f"{fmt(total)} calls use {total / available:.0%} of the available {fmt(available)}; " "a retry storm or another consumer could exhaust it") else: add("PASS", "RL01", f"{fmt(total)} calls fit the available {fmt(available)} ({total / available:.2%})") else: add("NOT CHECKED", "RL01", "budget not compared: pass --remaining (from usage headers or the Limits resource)") if windows: if args.target_utilization > 0.7: add("WARN", "RL02", f"planned at {args.target_utilization:.0%} of the tightest window; plan at 70% or less, " "because every app on the allowance shares it") else: add("PASS", "RL02", f"planned at {args.target_utilization:.0%} of the tightest window") else: add("NOT CHECKED", "RL02", "no window limit given") batch_max = item.get("batch_max") if batch_max and batch > batch_max: add("WARN", "RL04", f"batch of {fmt(batch)} is above the {fmt(batch_max)}-record maximum for this preset's batch endpoints") elif batch_max and batch == 1: add("INFO", "RL04", f"this preset's batch endpoints take up to {fmt(batch_max)} records; " f"one record per call costs about {batch_max} times the calls") if args.search_results is not None: cap, page = item.get("result_cap"), item.get("page_size") n = args.search_results if page: pages = math.ceil(n / page) text = f"Search: {fmt(n)} results at {fmt(page)} per page = {fmt(pages)} requests" if planned: text += f", {duration_text(pages / planned)} at {planned:.2f}/s" lines.append(text) result["search_requests"] = pages if not cap: add("NOT CHECKED", "RL05", "the preset has no result cap; check the vendor's search limits") elif n > cap: windows_needed = n // SEARCH_WINDOW_TARGET + 1 add("FAIL", "RL05", f"{fmt(n)} results exceed the {fmt(cap)}-result cap per query; split into {windows_needed} time windows " f"of under {fmt(SEARCH_WINDOW_TARGET)} results each (about {fmt(math.ceil(n / windows_needed))} per window " f"if spread evenly), and overlap them by the BF04 margin") result["search_windows_needed"] = windows_needed elif n >= SEARCH_WINDOW_TARGET: add("WARN", "RL05", f"{fmt(n)} results is within reach of the {fmt(cap)} cap; split the window so each returns under " f"{fmt(SEARCH_WINDOW_TARGET)}") else: add("PASS", "RL05", f"{fmt(n)} results stay under {fmt(SEARCH_WINDOW_TARGET)} per query") findings.sort(key=lambda f: f["rule"]) # stable: keeps insertion order within a rule counts = {lvl: sum(1 for f in findings if f["level"] == lvl) for lvl in ("FAIL", "WARN", "PASS", "INFO", "NOT CHECKED")} summary = f"{counts['FAIL']} FAIL, {counts['WARN']} WARN, {counts['PASS']} PASS" for lvl in ("INFO", "NOT CHECKED"): if counts[lvl]: summary += f", {counts[lvl]} {lvl}" tail = "Verify live before relying on these limits. Nothing was called or changed." if args.json: print(json.dumps({"result": result, "findings": findings, "summary": counts, "note": tail}, indent=2)) else: for line in lines: print(line) for f in findings: print(f"{f['level']} {f['rule']}: {f['message']}") print(f"{summary}. {tail}") return 1 if counts["FAIL"] else 0 if __name__ == "__main__": sys.exit(main())