# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import threading
import time
import urllib.parse
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from base_python_support import BasePythonSupport
from logger.logger import RaptorLogger
LOG = RaptorLogger(component="raptor-browsertime")
# Artificial /target.html backend delay so prefetch vs. cold-nav is legible.
# 500 ms corresponds to the p75 response time from the pageload event.
TARGET_STALL_MS = 500
# Speculation Rules eagerness levels the landing page can be served with. The
# requested level arrives as the ?eagerness= query parameter; unknown values
# fall back to DEFAULT_EAGERNESS.
ALLOWED_EAGERNESS = ("immediate", "eager", "moderate", "conservative")
DEFAULT_EAGERNESS = "moderate"
# Footer instruction describing how each level's prefetch is triggered.
ACTION_HINTS = {
"immediate": "Prefetch starts on load; click a button to navigate.",
"eager": "Hover a button to trigger prefetch, then click.",
"moderate": "Hover a button to trigger prefetch, then click.",
"conservative": "Press and hold a button to trigger prefetch, then release.",
}
# Doubled braces survive str.format; the JSON braces are emitted literally.
SPECULATION_RULES_TEMPLATE = """
"""
LANDING_HTML_TEMPLATE = """
Speculation Rules Prefetch Demo
{rules}
Speculation Rules Prefetch Demo
Inline <script type="speculationrules"> with
{eagerness} eagerness controls when a matching link is
prefetched. Target pages have a {stall_ms} ms server
stall.
"""
TARGET_HTML = """
Target · Speculation Rules Demo
Arrived
Target page · 500 ms server stall.
- Delivery
- —
- Response time
- —
- FCP
- —
- LCP
- —
- Load time
- —
"""
class _Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def _send(self, body):
body_bytes = body.encode("utf-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body_bytes)))
# Give the cache entry a non-expired freshness lifetime; without it
# nsPrefetchNode::OnStartRequest aborts speculation prefetches with
# NS_BINDING_ABORTED ("document cannot be reused from cache").
self.send_header("Cache-Control", "max-age=60")
self.end_headers()
self.wfile.write(body_bytes)
def _send_404(self):
self.send_response(HTTPStatus.NOT_FOUND)
self.send_header("Content-Length", "0")
self.end_headers()
def do_GET(self):
parsed = urllib.parse.urlparse(self.path)
path = parsed.path
if path in ("/", "/landing.html"):
requested = urllib.parse.parse_qs(parsed.query).get(
"eagerness", [DEFAULT_EAGERNESS]
)[0]
eagerness = (
requested if requested in ALLOWED_EAGERNESS else DEFAULT_EAGERNESS
)
self._send(
LANDING_HTML_TEMPLATE.format(
rules=SPECULATION_RULES_TEMPLATE.format(eagerness=eagerness),
stall_ms=TARGET_STALL_MS,
eagerness=eagerness,
action_hint=ACTION_HINTS[eagerness],
)
)
elif path == "/target.html":
time.sleep(TARGET_STALL_MS / 1000.0)
self._send(TARGET_HTML)
else:
self._send_404()
def log_message(self, fmt, *args):
LOG.info("demo-server: " + fmt % args)
class SpeculationRules(BasePythonSupport):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.httpd = None
self.server_thread = None
self.port = None
def setup_test(self, test, args):
super().setup_test(test, args)
self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), _Handler)
self.port = self.httpd.server_address[1]
self.server_thread = threading.Thread(
target=self.httpd.serve_forever, daemon=True
)
self.server_thread.start()
LOG.info(f"SpeculationRules demo server: http://127.0.0.1:{self.port}/")
def modify_command(self, cmd, test):
cmd += [
"--browsertime.server_url",
f"http://127.0.0.1:{self.port}",
# Target page renders its own Performance-API panel; this overlay
# (desktop default) otherwise obscures it in the recording.
"--videoParams.addTimer",
"false",
]
def handle_result(self, bt_result, raw_result, last_result=False, **kwargs):
extras = raw_result.get("extras", [])
if not extras or not isinstance(extras, list):
return
custom = extras[0].get("custom_data", {})
for key in ("navigation_duration", "response_start"):
val = custom.get(key)
if val is None:
continue
try:
bt_result["measurements"].setdefault(key, []).append(float(val))
except (TypeError, ValueError):
LOG.warning(f"SpeculationRules: non-numeric {key}={val!r}; skipped")
delivery_type = custom.get("delivery_type")
if delivery_type is not None:
LOG.info(f"SpeculationRules delivery_type: {delivery_type!r}")
def summarize_test(self, test, suite, **kwargs):
# Perfherder schema requires subtests to be a list, not a dict.
if suite.get("subtests") == {}:
suite["subtests"] = []
for name, replicates in test["measurements"].items():
if not replicates:
continue
suite["subtests"].append(
self._build_standard_subtest(test, replicates, name, should_alert=False)
)
suite["subtests"].sort(key=lambda st: st["name"])
def clean_up(self):
if self.httpd is not None:
try:
self.httpd.shutdown()
self.httpd.server_close()
except Exception as e:
LOG.warning(f"SpeculationRules shutdown error: {e}")
if self.server_thread is not None:
self.server_thread.join(5)