Fast Updater

This commit is contained in:
firestar5683
2026-10-03 21:22:38 -05:00
parent a8915ca858
commit 39c1767f65
14 changed files with 964 additions and 85 deletions
+98
View File
@@ -0,0 +1,98 @@
import hashlib
import json
import os
from pathlib import Path
import stat
import subprocess
import sys
RECEIPT = 'prebuilt.json'
RUNTIME_ROOTS = ('openpilot/system/ui/', 'openpilot/selfdrive/ui/', 'openpilot/starpilot/ui/',
'openpilot/starpilot/software/', 'openpilot/starpilot/galaxy/')
def protected(path):
if path in ('prebuilt', RECEIPT):
return False
if path.endswith('.py') and path.startswith(RUNTIME_ROOTS):
return False
if path.startswith('openpilot/starpilot/galaxy/') and path.endswith(('.html', '.css', '.js', '.svg', '.png', '.jpg', '.woff2')):
return False
return True
def inventory_digest(entries):
inventory = sorted((path, mode, blob) for path, mode, blob in entries if protected(path))
return hashlib.sha256(json.dumps(inventory, separators=(',', ':'), ensure_ascii=True).encode()).hexdigest()
def revision_digest(git, revision):
entries = []
for record in git('ls-tree', '-rz', '--full-tree', revision).split('\0'):
if record:
metadata, path = record.split('\t', 1)
mode, kind, blob = metadata.split()
if kind != 'blob':
raise ValueError('Prebuilt inventory contains non-file source')
entries.append((path, mode, blob))
return inventory_digest(entries)
def valid_receipt(git, revision):
if git('ls-tree', revision, '--', RECEIPT).strip().split()[:1] != ['100644']:
return False
try:
receipt = json.loads(git('show', f'{revision}:{RECEIPT}'))
return (type(receipt) is dict and type(receipt.get('schema_version')) is int and receipt['schema_version'] == 1 and
receipt.get('protected_inventory_sha256') == revision_digest(git, revision))
except (ValueError, TypeError, KeyError):
return False
def write_receipt(repo):
repo = Path(repo).resolve()
def git(*args, input=None):
return subprocess.check_output(['git', '-c', 'core.hooksPath=/dev/null', '-C', str(repo), *args],
input=input, timeout=30).decode()
entries = []
regular = []
for record in git('ls-files', '--stage', '-z').split('\0'):
if not record:
continue
metadata, path = record.split('\t', 1)
indexed_mode, _, stage = metadata.split()
if stage != '0' or indexed_mode == '160000':
raise ValueError('Prebuilt source has unresolved or nested Git entries')
if not protected(path):
continue
file = repo / path
info = file.lstat()
if stat.S_ISLNK(info.st_mode):
mode = '120000'
blob = git('hash-object', '--stdin', input=os.fsencode(os.readlink(file))).strip()
elif stat.S_ISREG(info.st_mode):
mode = '100755' if info.st_mode & 0o111 else '100644'
regular.append((path, mode))
continue
else:
raise ValueError('Prebuilt source contains a non-regular file')
entries.append((path, mode, blob))
for offset in range(0, len(regular), 256):
batch = regular[offset:offset + 256]
blobs = git('hash-object', '--no-filters', '--', *(path for path, _ in batch)).splitlines()
if len(blobs) != len(batch):
raise ValueError('Prebuilt source hashing was incomplete')
entries.extend((path, mode, blob) for (path, mode), blob in zip(batch, blobs, strict=True))
receipt = dict(schema_version=1, protected_inventory_sha256=inventory_digest(entries))
temporary = repo / (RECEIPT + '.tmp')
try:
temporary.write_text(json.dumps(receipt, sort_keys=True) + '\n')
os.replace(temporary, repo / RECEIPT)
finally:
temporary.unlink(missing_ok=True)
return receipt
if __name__ == '__main__':
write_receipt(sys.argv[1])
@@ -96,8 +96,8 @@ class UpdaterProcess:
except (SoftwareOperationError, UpdaterControlError, OSError, RuntimeError, ValueError):
return False
def send(self, action: str) -> None:
if action not in ("check", "download"):
def send(self, action: str, *, branch=None, commit=None) -> None:
if action not in ("check", "download", "fast"):
raise SoftwareOperationError("Invalid updater command", 400)
with self.lock:
pid = self._manager_pid()
@@ -105,7 +105,10 @@ class UpdaterProcess:
if self._manager_pid() != pid or self._proc_identity(pid) != start:
raise SoftwareOperationError("Updater process changed", 503)
try:
self.control.send(pid, start, action)
if action == "fast":
self.control.send(pid, start, action, branch=branch, commit=commit)
else:
self.control.send(pid, start, action)
except (UpdaterControlError, OSError, RuntimeError, ValueError):
raise SoftwareOperationError("Updater control is unavailable", 503) from None
@@ -158,6 +161,7 @@ class SoftwareOperations:
self.request: dict | None = None
self._baseline: tuple | None = None
self._started = 0.0
self._fast_commit = None
def _param(self, key: str, limit: int = 256) -> str | None:
try:
@@ -236,15 +240,25 @@ class SoftwareOperations:
if request is None or request["state"] != "pending" or request["action"] == "install":
return
state = status["updater"]["state"]
if request["target"] is not None and status["updater"]["targetBranch"] != request["target"]:
if request["action"] != "fast" and request["target"] is not None and status["updater"]["targetBranch"] != request["target"]:
request.update(state="failed", error="Updater target changed")
elif state == "idle" and status["updater"]["failedCount"] is not None and self._baseline[2] is not None and \
status["updater"]["failedCount"] > self._baseline[2]:
request.update(state="failed", error="Updater reported a failure")
error = "Updater reported a failure"
if request["action"] == "fast":
try:
error = self._param("LastUpdateException", 4096) or error
except SoftwareOperationError:
pass
request.update(state="failed", error=error)
elif state == "idle" and status["updater"]["failedCount"] == 0 and \
status["updater"]["lastSuccessAt"] is not None and \
status["updater"]["lastSuccessAt"] != self._baseline[0]:
if request["action"] == "check" or status["updater"]["lastFetchAt"] != self._baseline[1]:
if request["action"] == "fast":
if status["updater"]["lastFetchAt"] != self._baseline[1]:
restarting = self._flag("DoReboot") is True or status["installed"]["commit"] != self._fast_commit
request.update(state="complete", error=None, outcome="restarting" if restarting else "up_to_date")
elif request["action"] == "check" or status["updater"]["lastFetchAt"] != self._baseline[1]:
request.update(state="complete", error=None)
elif self._ready(status, request["target"]):
request.update(state="complete", error=None)
@@ -284,7 +298,7 @@ class SoftwareOperations:
reason = "Updater is not running"
return {
"parked": parked, "availableBranches": branches, "selectedTarget": target,
"canCheck": usable, "canSelect": usable and bool(branches),
"canCheck": usable, "canFastUpdate": usable and self._valid_branch(status["installed"]["branch"]), "canSelect": usable and bool(branches),
"canDownload": usable and selected, "canInstall": usable and selected and self._ready(status, target),
"reason": reason, "request": dict(self.request) if self.request is not None else None,
"automaticDownloads": automatic_downloads(self.params), "canConfigure": parked and not reboot,
@@ -334,7 +348,7 @@ class SoftwareOperations:
return self.snapshot()
def action(self, action: str, payload: object, *, authorized: Callable[[], bool]) -> dict:
if type(payload) is not dict or type(action) is not str or action not in ("check", "download", "select", "install", "preferences"):
if type(payload) is not dict or type(action) is not str or action not in ("check", "download", "select", "install", "preferences", "fast"):
raise SoftwareOperationError("Invalid software action", 400)
if payload.get("action") != action or any(type(key) is not str for key in payload):
raise SoftwareOperationError("Invalid software action", 400)
@@ -357,10 +371,15 @@ class SoftwareOperations:
raise SoftwareOperationError("Vehicle is not safely parked", 409)
if view["reason"] is not None:
raise SoftwareOperationError(view["reason"], 503 if view["reason"] == "Updater is not running" else 409)
if branch is not None and branch not in view["availableBranches"]:
if action != "fast" and branch is not None and branch not in view["availableBranches"]:
raise SoftwareOperationError("Branch is not in the current updater list", 409)
if action != "check" and action != "select" and (branch != view["selectedTarget"] or not view["canDownload"]):
if action not in ("check", "select", "fast") and (branch != view["selectedTarget"] or not view["canDownload"]):
raise SoftwareOperationError("Selected updater target changed", 409)
if action == "fast":
installed = self.status.snapshot()["installed"]
identity = self.git_identity(self.installed)
if not view["canFastUpdate"] or branch != installed["branch"] or identity is None or identity != (branch, installed["commit"]):
raise SoftwareOperationError("Installed branch changed; refresh before updating", 409)
if action == "install" and not view["canInstall"]:
raise SoftwareOperationError("Finalized update does not match selected branch", 409)
if not authorized() or not self._parked() or self._flag("IsOnroad") is not False:
@@ -368,7 +387,8 @@ class SoftwareOperations:
current = self.status.snapshot()
if (current["updater"]["state"] != "idle" or self._flag("DisableUpdates") is not False or
self._flag("DoReboot") is not False or not self.process.available() or
(branch is not None and branch not in self._branches())):
(action != "fast" and branch is not None and branch not in self._branches()) or
(action == "fast" and current["installed"]["branch"] != branch)):
raise SoftwareOperationError("Updater state changed", 409)
if action == "select":
self.params.put("UpdaterTargetBranch", branch, block=True)
@@ -387,7 +407,15 @@ class SoftwareOperations:
status["updater"]["failedCount"])
if not authorized() or not self._parked() or self._flag("IsOnroad") is not False:
raise SoftwareOperationError("Session or parked state changed", 403)
self.process.send(action)
if action == "fast":
identity = self.git_identity(self.installed)
if identity is None or identity != (branch, status["installed"]["commit"]) or status["installed"]["branch"] != branch:
raise SoftwareOperationError("Installed branch changed; refresh before updating", 409)
self._fast_commit = status["installed"]["commit"]
if action == "fast":
self.process.send(action, branch=identity[0], commit=identity[1])
else:
self.process.send(action)
self.request = {"id": secrets.token_hex(8), "action": action,
"target": branch if branch is not None else status["updater"]["targetBranch"],
"state": "pending", "error": None}
@@ -16,14 +16,17 @@ class FakeProcess:
def __init__(self):
self.running = True
self.sent = []
self.fast_requests = []
def available(self):
return self.running
def send(self, sig):
def send(self, sig, **identity):
if not self.running:
raise SoftwareOperationError("Updater stopped", 503)
self.sent.append(sig)
if sig == "fast":
self.fast_requests.append(identity)
def close(self):
pass
@@ -265,6 +268,46 @@ class SoftwareOperationsTest(unittest.TestCase):
self.assertEqual(reader.call_count, 4)
def test_fast_current_branch_without_branch_list_or_selected_target(self):
self.params.remove("UpdaterAvailableBranches")
self.params.put("UpdaterTargetBranch", "release", block=True)
self.assertTrue(self.owner.snapshot()["canFastUpdate"])
result = self.act("fast", "main")
self.assertEqual(result["request"]["target"], "main")
self.assertEqual(self.process.sent, ["fast"])
self.assertEqual(self.process.fast_requests, [{"branch": "main", "commit": "a" * 40}])
self.assertEqual(self.params.get("UpdaterTargetBranch"), "release")
self.assertFalse(self.params.get_bool("DoReboot"))
self.params.put("LastUpdateTime", datetime(2026, 10, 3, 12), block=True)
self.params.put("UpdaterLastFetchTime", datetime(2026, 10, 3, 12), block=True)
result = self.owner.snapshot()["request"]
self.assertEqual(result["state"], "complete")
self.assertEqual(result["outcome"], "up_to_date")
def test_fast_rejects_other_or_stale_installed_branch(self):
self.denied("fast", 409, "release")
self.git[self.installed] = ("main", "b" * 40)
self.denied("fast", 409, "main")
self.git[self.installed] = ("release", "a" * 40)
self.denied("fast", 409, "main")
self.assertEqual(self.process.sent, [])
self.assertFalse(self.params.get_bool("DoReboot"))
def test_fast_keeps_existing_admission_and_observes_failure(self):
for key in ("DisableUpdates", "IsOnroad", "DoReboot"):
self.params.put_bool(key, True, block=True)
self.denied("fast", 409, "main")
self.params.put_bool(key, False, block=True)
self.authorized = False
self.denied("fast", 403, "main")
self.authorized = True
self.act("fast", "main")
self.denied("fast", 409, "main")
self.params.put("UpdateFailedCount", 1, block=True)
self.assertEqual(self.owner.snapshot()["request"]["state"], "failed")
self.assertEqual(self.process.sent, ["fast"])
class UpdaterIdentityTest(unittest.TestCase):
def test_fresh_manager_pid_and_same_proc_start_before_signal(self):
with tempfile.TemporaryDirectory() as temporary:
@@ -324,3 +324,40 @@ assert(validSoftwareSnapshot(snapshot({ operations: { ...operations, automaticDo
assert(!validSoftwareSnapshot(snapshot({ operations: { ...operations, history: { ...history, installed: [{ hash: '--help' }] } } })))
assert(!validSoftwareSnapshot(snapshot({ operations: { ...operations, automaticDownloads: 'true' } })))
assert.doesNotMatch(SoftwarePage.template, /v-html/)
const fast = fixture()
fast.feed.start()
await fast.reply(0, snapshot({ operations: { ...operations, availableBranches: [], selectedTarget: "beta", canFastUpdate: true } }))
fast.feed.action("fast", "beta")
assert.equal(fast.requests.length, 1)
fast.feed.action("fast", installed.branch)
assert.deepEqual(JSON.parse(fast.requests[1].options.body), { action: "fast", branch: installed.branch })
await fast.reply(1, snapshot({ operations: { ...operations, canFastUpdate: false, request: request("fast", "pending", installed.branch) }, updater: { ...updater, state: "updating..." } }))
fast.feed.action("fast", installed.branch)
assert.equal(fast.requests.length, 2)
fast.fire(1000)
await fast.reply(2, snapshot({ operations: { ...operations, canFastUpdate: true, request: { ...request("fast", "complete", installed.branch), outcome: "up_to_date" } } }))
assert.equal(fast.feed.installBaseline, null)
assert.match(fast.feed.notice, /already up to date/)
fast.feed.stop()
const fastLost = fixture()
fastLost.feed.start()
await fastLost.reply(0, snapshot({ operations: { ...operations, canFastUpdate: true } }))
fastLost.feed.action("fast", installed.branch)
await fastLost.fail(1)
assert.equal(fastLost.feed.uncertain, true)
fastLost.feed.action("fast", installed.branch)
assert.equal(fastLost.requests.length, 2)
fastLost.fire(2000)
await fastLost.reply(2, snapshot({ installed: { ...installed, commit: "b".repeat(40) }, operations: { ...operations, canFastUpdate: true } }))
assert.equal(fastLost.feed.installBaseline, null)
assert.equal(fastLost.feed.uncertain, false)
assert.match(fastLost.feed.notice, /verified after reconnect/)
fastLost.feed.stop()
const fastPage = { actionDisabled: false, operations: { canFastUpdate: true, selectedTarget: "beta" }, data: { installed }, dialog: null }
SoftwarePage.methods.askFastUpdate.call(fastPage)
assert.deepEqual(fastPage.dialog, { action: "fast", branch: "Dom", title: "Fast Update", message: "Download latest version of Dom and restart?", label: "Fast Update" })
assert.equal(validSoftwareSnapshot(snapshot({ operations: { ...operations, canFastUpdate: "yes" } })), false)
assert.equal(validSoftwareSnapshot(snapshot({ operations: { ...operations, request: { ...request("fast"), outcome: "invented" } } })), false)
@@ -1,11 +1,11 @@
import { decodeLayoutBackup, encodeLayoutBackup, MAX_LAYOUT_BACKUP_BYTES } from "./layout-backup.js"
const ACTIONS = new Set(["check", "download", "select", "install", "preferences"])
const ACTIONS = new Set(["check", "download", "select", "install", "preferences", "fast"])
const REQUEST_STATES = new Set(["pending", "complete", "failed"])
const UPDATER_ACTIVE = new Set(["checking...", "downloading...", "finalizing update..."])
const UPDATER_ACTIVE = new Set(["checking...", "downloading...", "finalizing update...", "updating..."])
const PRIMARY_BRANCHES = ["StarPilot", "Dom"]
const unavailableOperations = () => ({ parked: false, availableBranches: [], selectedTarget: null,
canCheck: false, canDownload: false, canSelect: false, canInstall: false,
canCheck: false, canFastUpdate: false, canDownload: false, canSelect: false, canInstall: false,
reason: "Update controls are unavailable", request: null })
export function validSoftwareSnapshot(data) {
@@ -22,12 +22,14 @@ export function validSoftwareSnapshot(data) {
operations.availableBranches.length <= 256 && operations.availableBranches.every((branch) =>
typeof branch === "string" && branch.length > 0 && branch.length <= 128) &&
text(operations.selectedTarget) && text(operations.reason) &&
(operations.canFastUpdate === undefined || typeof operations.canFastUpdate === "boolean") &&
(operations.automaticDownloads === undefined || flag(operations.automaticDownloads)) &&
(operations.canConfigure === undefined || typeof operations.canConfigure === "boolean") &&
(operations.history === undefined || validHistory(operations.history)) &&
[operations.canCheck, operations.canDownload, operations.canSelect, operations.canInstall].every((value) => typeof value === "boolean") &&
(request === null || !!request && typeof request.id === "string" && request.id.length <= 100 &&
ACTIONS.has(request.action) && text(request.target) && REQUEST_STATES.has(request.state) && text(request.error)))
ACTIONS.has(request.action) && text(request.target) && REQUEST_STATES.has(request.state) && text(request.error) &&
(request.outcome === undefined || ["up_to_date", "restarting"].includes(request.outcome))))
}
export function validHistory(value) {
@@ -147,7 +149,9 @@ export class SoftwareStatusFeed {
this.uncertain = false
this.uncertainAction = null
this.actionPriorRequestId = null
if (body.action === "install") {
if (body.action === "fast") {
this.notice = `Updating ${body.branch}. Waiting for completion or reconnect.`
} else if (body.action === "install") {
this.installBaseline = this.installBaseline || { commit: payload.installed.commit, branch: payload.installed.branch, target: body.branch }
this.notice = "Restart requested. Waiting to reconnect and verify the installed build."
} else this.notice = body.action === "select" ? `Target branch set to ${body.branch}. Check and download the update when ready.` :
@@ -166,6 +170,13 @@ export class SoftwareStatusFeed {
this.uncertainAction = null
this.actionPriorRequestId = null
}
if (this.installBaseline?.action === "fast" && observed?.action === "fast" && observed.target === this.installBaseline.target &&
["complete", "failed"].includes(observed.state)) {
if (observed.state === "failed" || observed.outcome === "up_to_date") {
this.installBaseline = null
this.notice = observed.state === "failed" ? "" : "Installed branch is already up to date. No restart needed."
} else this.notice = "Update installed. Waiting to reconnect and verify the installed build."
}
if (this.installBaseline && payload.installed.branch === this.installBaseline.target &&
(payload.installed.commit && payload.installed.commit !== this.installBaseline.commit ||
payload.installed.branch !== this.installBaseline.branch)) {
@@ -186,7 +197,7 @@ export class SoftwareStatusFeed {
if (body !== null) {
this.uncertain = !error?.rejected
this.uncertainAction = this.uncertain ? body : null
if (error?.rejected && body.action === "install") this.installBaseline = null
if (error?.rejected && ["install", "fast"].includes(body.action)) this.installBaseline = null
}
this.error = error?.message || "Software request failed. Refresh to try again."
this.emit(this.data ? "ready" : this.attempted ? "unavailable" : "loading")
@@ -214,14 +225,16 @@ export class SoftwareStatusFeed {
action(action, branch = null) {
const operations = this.data?.operations
const capability = { check: "canCheck", download: "canDownload", select: "canSelect", install: "canInstall" }[action]
const capability = { check: "canCheck", download: "canDownload", select: "canSelect", install: "canInstall", fast: "canFastUpdate" }[action]
if (!this.active || !capability || this.mutating || this.blocked || this.uncertain || !operations?.parked ||
operations[capability] !== true || operations.request?.state === "pending") return null
if (action === "select" && (branch === "other:" || !operations.availableBranches.includes(branch) || branch === operations.selectedTarget)) return null
if (["download", "install"].includes(action) && (!branch || branch !== operations.selectedTarget)) return null
if (action === "fast" && (!branch || branch !== this.data.installed.branch)) return null
this.actionPriorRequestId = operations.request?.id ?? null
if (action === "install") this.installBaseline = { commit: this.data.installed.commit,
branch: this.data.installed.branch, target: branch }
if (action === "fast") this.installBaseline = { commit: this.data.installed.commit, branch: this.data.installed.branch, target: branch, action }
return this.run(action === "check" ? { action } : { action, branch })
}
}
@@ -345,6 +358,9 @@ export const SoftwarePage = {
requestMessage(request) {
if (!request) return ""
if (request.state === "failed") return request.error || "Update request failed. Refresh and try again."
if (request.action === "fast") return request.state === "pending" ? `Updating ${request.target}…` :
request.outcome === "up_to_date" ? "Installed branch is already up to date. No restart needed." :
"Update installed. Waiting to reconnect and verify the installed build."
if (request.action === "install") return "Restart requested. Waiting to reconnect and verify the installed build."
if (request.state === "pending") return request.action === "check" ? "Checking for updates…" : "Downloading and preparing update…"
return request.action === "check" ? "Update check finished." : request.action === "download" ? "Download finished." : "Target branch saved."
@@ -354,6 +370,12 @@ export const SoftwarePage = {
this.dialog = { action: "select", branch: this.draftBranch, title: "Change target branch",
message: `Set ${this.draftBranch} as the target branch? This only stages the choice. Check for updates and download separately.`, label: "Set target branch" }
},
askFastUpdate() {
const branch = this.data?.installed?.branch
if (this.actionDisabled || this.operations?.canFastUpdate !== true || !branch) return
this.dialog = { action: "fast", branch, title: "Fast Update",
message: `Download latest version of ${branch} and restart?`, label: "Fast Update" }
},
askInstall() {
if (this.actionDisabled || !this.operations?.canInstall || !this.operations.selectedTarget) return
this.dialog = { action: "install", branch: this.operations.selectedTarget, title: "Restart and install update",
@@ -412,7 +434,8 @@ export const SoftwarePage = {
<p v-else-if="data.updater.state">{{ data.updater.state }}</p>
<p v-if="data.updater.targetChangeFound === true">An update was found for the target branch.</p>
<p v-else-if="data.updater.targetChangeFound === false">No target change reported by the last check.</p>
<div class="gx-software-actions"><button type="button" class="gx-btn" :disabled="actionDisabled || !operations.canDownload || !operations.selectedTarget" @click="feed.action('download', operations.selectedTarget)">Download update</button>
<div class="gx-software-actions"><button type="button" class="gx-btn" :disabled="actionDisabled || operations.canFastUpdate !== true || !data.installed.branch" @click="askFastUpdate">Fast Update</button>
<button type="button" class="gx-btn" :disabled="actionDisabled || !operations.canDownload || !operations.selectedTarget" @click="feed.action('download', operations.selectedTarget)">Download update</button>
<button type="button" class="gx-btn" :disabled="actionDisabled || !operations.canInstall || !operations.selectedTarget" @click="askInstall">Restart &amp; install</button></div>
<dl><dt>Last checked</dt><dd>{{ reported(data.updater.lastSuccessAt) }}</dd><dt>Last download</dt><dd>{{ reported(data.updater.lastFetchAt) }}</dd></dl></section>
<section v-if="operations.automaticDownloads !== undefined" class="gx-card gx-software-card">
@@ -0,0 +1,83 @@
import os
from pathlib import Path
import re
import subprocess
from openpilot.common.vendor_manifest import validate_revision
from openpilot.common.prebuilt_manifest import revision_digest, valid_receipt
class FastUpdateError(RuntimeError):
pass
def run(command, cwd):
environment = dict(os.environ, GIT_LFS_SKIP_SMUDGE='1', GIT_TERMINAL_PROMPT='0')
return subprocess.check_output(command, cwd=cwd, env=environment, stderr=subprocess.STDOUT,
timeout=120 if "fetch" in command or "reset" in command else 30).decode()
def fast_update(repo, branch, *, params, parked, expected_commit, current_os=None, invalidate=None, run=run):
repo = Path(repo)
def git(*arguments):
return run(['git', '-c', 'gc.auto=0', '-c', 'maintenance.auto=false', '-c', 'core.hooksPath=/dev/null', '-c', 'submodule.recurse=false', '-c', 'filter.lfs.required=false',
'-c', 'filter.lfs.smudge=', '-c', 'filter.lfs.process=', *arguments], cwd=str(repo))
def admitted():
if not parked():
raise FastUpdateError('Fast Update requires fresh parked state')
if git('symbolic-ref', '--short', 'HEAD').strip() != branch:
raise FastUpdateError('Fast Update only updates the current branch')
if git('status', '--porcelain', '--untracked-files=no').strip():
raise FastUpdateError('Tracked source has local changes')
git('check-ref-format', '--branch', branch)
admitted()
previous = git('rev-parse', 'HEAD^{commit}').strip()
if not isinstance(expected_commit, str) or not re.fullmatch(r'[0-9a-f]{40}', expected_commit) or previous != expected_commit:
raise FastUpdateError('Approved source revision changed; refresh and try again')
git('fetch', '--depth=1', '--no-tags', '--no-recurse-submodules', 'origin', f'refs/heads/{branch}')
target = git('rev-parse', 'FETCH_HEAD^{commit}').strip()
validate_revision(repo, target)
launch = git('show', f'{target}:launch_env.sh')
versions = re.findall(r'^\s*export\s+AGNOS_VERSION=["\']([^"\'\n]+)["\']\s*$', launch, re.MULTILINE)
if len(versions) != 1 or not current_os or versions[0] != current_os:
raise FastUpdateError('Update requires a different or unverified operating system')
if target == previous:
return 'up-to-date'
for revision in (previous, target):
if git('ls-tree', revision, '--', 'prebuilt').strip().split()[:1] != ['100644']:
raise FastUpdateError('Fast Update requires prebuilt releases; use a full validated update')
if not valid_receipt(git, target) and revision_digest(git, previous) != revision_digest(git, target):
raise FastUpdateError('Native build inputs or artifacts changed without a matching build receipt; use a full validated update')
tracked = set(git('ls-tree', '-rz', '--name-only', target).split('\0')) - {''}
untracked = set(git('ls-files', '--others', '-z').split('\0')) - {''}
directories = {str(parent) for name in tracked for parent in Path(name).parents if str(parent) != '.'}
for name in untracked:
if name in tracked or name in directories or any(str(parent) in tracked for parent in Path(name).parents):
raise FastUpdateError('Update would overwrite untracked data')
admitted()
if git('rev-parse', 'HEAD^{commit}').strip() != previous:
raise FastUpdateError('Source changed during download')
if invalidate is None:
raise FastUpdateError('Update staging invalidation is unavailable')
git('update-ref', 'refs/starpilot/previous', previous)
invalidate()
admitted()
if git('rev-parse', 'HEAD^{commit}').strip() != previous:
raise FastUpdateError('Source changed before installation')
try:
git('reset', '--hard', '--no-recurse-submodules', target)
if git('rev-parse', 'HEAD^{commit}').strip() != target:
raise FastUpdateError('Applied source revision does not match download')
validate_revision(repo, 'HEAD')
if not parked():
raise FastUpdateError('Parked state changed before restart')
params.put_bool('DoReboot', True, block=True)
except Exception:
git('reset', '--hard', '--no-recurse-submodules', previous)
if git('rev-parse', 'HEAD^{commit}').strip() != previous:
raise FastUpdateError('Source rollback failed')
raise
return 'reboot-requested'
@@ -0,0 +1,279 @@
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[3]
def load(name, path):
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
sys.modules[name] = module
spec.loader.exec_module(module)
return module
vendor = load('openpilot.common.vendor_manifest', ROOT / 'common/vendor_manifest.py')
prebuilt = load('openpilot.common.prebuilt_manifest', ROOT / 'common/prebuilt_manifest.py')
owner = load('fast_update_owner', ROOT / 'starpilot/software/fast_update.py')
class Params:
def __init__(self):
self.calls = []
def put_bool(self, *args, **kwargs):
self.calls.append((args, kwargs))
class TestFastUpdate(unittest.TestCase):
def git(self, repo, *args):
return subprocess.check_output(['git', '-C', str(repo), *args], stderr=subprocess.STDOUT).decode().strip()
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
base = Path(self.temp.name)
self.remote, self.repo = base / 'remote', base / 'repo'
self.remote.mkdir()
self.git(self.remote, 'init', '-b', 'Dom')
self.git(self.remote, 'config', 'user.name', 'Test')
self.git(self.remote, 'config', 'user.email', 'test@example.invalid')
dependencies = []
for name in sorted(vendor.DEPENDENCIES):
(self.remote / name).mkdir()
(self.remote / name / 'source').write_text('vendored')
dependencies.append(dict(path=name, commit='1' * 40, tree='2' * 40, url='https://example.invalid/source', exclude=[]))
manifest = dict(schema_version=1, upstream=dict(commit='3' * 40, url='https://example.invalid/upstream'), dependencies=dependencies)
(self.remote / 'upstream-sync.json').write_text(json.dumps(manifest))
(self.remote / 'launch_env.sh').write_text('export AGNOS_VERSION="19.8.1"\n')
(self.remote / 'prebuilt').write_text('')
(self.remote / 'openpilot/starpilot/ui').mkdir(parents=True)
(self.remote / 'openpilot/starpilot/ui/source.py').write_text('old')
self.git(self.remote, 'add', '.')
self.git(self.remote, 'commit', '-m', 'initial')
self.git(base, 'clone', '--no-local', str(self.remote), str(self.repo))
self.previous = self.git(self.repo, 'rev-parse', 'HEAD')
(self.remote / 'openpilot/starpilot/ui/source.py').write_text('new')
self.git(self.remote, 'commit', '-am', 'update')
self.target = self.git(self.remote, 'rev-parse', 'HEAD')
self.params = Params()
self.invalidations = []
def update(self, **kwargs):
arguments = dict(params=self.params, parked=lambda: True, expected_commit=self.git(self.repo, 'rev-parse', 'HEAD'), current_os='19.8.1',
invalidate=lambda: self.invalidations.append(True))
arguments.update(kwargs)
return owner.fast_update(self.repo, 'Dom', **arguments)
def test_applies_exact_revision_and_retains_untracked_models(self):
(self.repo / 'models').mkdir()
(self.repo / 'models' / 'download').write_bytes(b'model')
commands = []
def traced(command, cwd):
commands.append(command)
return owner.run(command, cwd)
self.assertEqual(self.update(run=traced), 'reboot-requested')
fetch = next(command for command in commands if 'fetch' in command)
self.assertIn('gc.auto=0', fetch)
self.assertIn('maintenance.auto=false', fetch)
self.assertIn('--depth=1', fetch)
self.assertIn('--no-tags', fetch)
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.target)
self.assertEqual(self.git(self.repo, 'rev-parse', 'refs/starpilot/previous'), self.previous)
self.assertEqual((self.repo / 'models' / 'download').read_bytes(), b'model')
self.assertEqual(self.params.calls, [(('DoReboot', True), dict(block=True))])
self.assertEqual(self.update(), 'up-to-date')
self.assertEqual(len(self.invalidations), 1)
def test_expected_commit_changed_rejected_before_fetch(self):
commands = []
def traced(command, cwd):
commands.append(command)
return owner.run(command, cwd)
with self.assertRaises(owner.FastUpdateError):
self.update(expected_commit='1' * 40, run=traced)
self.assertFalse(any('fetch' in command for command in commands))
def test_source_only_ui_update_admitted(self):
folder = self.remote / 'openpilot/starpilot/ui'
folder.mkdir(parents=True, exist_ok=True)
(folder / 'runtime_app.py').write_text('value = 1')
self.git(self.remote, 'add', '.')
self.git(self.remote, 'commit', '-m', 'UI update')
self.assertEqual(self.update(), 'reboot-requested')
def test_native_source_change_rejected_without_artifact_proof(self):
(self.remote / 'native.cc').write_text('int main() {}')
self.git(self.remote, 'add', '.')
self.git(self.remote, 'commit', '-m', 'native update')
with self.assertRaisesRegex(owner.FastUpdateError, 'full validated update'):
self.update()
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.invalidations)
def receipt_commit(self):
prebuilt.write_receipt(self.remote)
self.git(self.remote, 'add', 'prebuilt.json')
self.git(self.remote, 'commit', '-m', 'validated receipt')
def test_matching_receipt_admits_native_source_and_artifacts(self):
(self.remote / 'native.cc').write_text('int source;')
(self.remote / 'native.so').write_bytes(b'compiled artifact fixture')
self.git(self.remote, 'add', '.')
self.receipt_commit()
self.assertEqual(self.update(), 'reboot-requested')
def test_receipt_stale_source_rejected(self):
(self.remote / 'native.cc').write_text('int source;')
self.git(self.remote, 'add', '.')
self.receipt_commit()
(self.remote / 'native.cc').write_text('int changed;')
self.git(self.remote, 'commit', '-am', 'stale source')
with self.assertRaisesRegex(owner.FastUpdateError, 'matching build receipt'):
self.update()
self.assertFalse(self.invalidations)
def test_receipt_stale_artifact_rejected(self):
(self.remote / 'native.so').write_bytes(b'artifact')
self.git(self.remote, 'add', '.')
self.receipt_commit()
(self.remote / 'native.so').write_bytes(b'changed artifact')
self.git(self.remote, 'commit', '-am', 'stale artifact')
with self.assertRaisesRegex(owner.FastUpdateError, 'matching build receipt'):
self.update()
self.assertFalse(self.invalidations)
def test_receipt_working_content_and_modes_match_committed_tree(self):
(self.remote / 'native.cc').write_text('int unstaged;')
self.git(self.remote, 'add', '.')
(self.remote / 'native.cc').write_text('int working;')
(self.remote / 'native.cc').chmod(0o755)
(self.remote / 'native-link').symlink_to('native.cc')
self.git(self.remote, 'add', 'native-link')
prebuilt.write_receipt(self.remote)
self.git(self.remote, 'add', '.')
self.git(self.remote, 'commit', '-m', 'receipt content and modes')
git = lambda *args: subprocess.check_output(['git', '-C', str(self.remote), *args]).decode()
self.assertTrue(prebuilt.valid_receipt(git, 'HEAD'))
def test_missing_prebuilt_rejected_without_removing_local_marker(self):
self.git(self.remote, 'rm', 'prebuilt')
self.git(self.remote, 'commit', '-m', 'nonprebuilt')
with self.assertRaisesRegex(owner.FastUpdateError, 'prebuilt'):
self.update()
self.assertTrue((self.repo / 'prebuilt').exists())
def test_wrong_os_rejected_before_mutation(self):
with self.assertRaises(owner.FastUpdateError):
self.update(current_os='19.8.2')
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.invalidations)
self.assertFalse(self.params.calls)
def test_dirty_tracked_source_rejected(self):
(self.repo / 'openpilot/starpilot/ui/source.py').write_text('local repair')
with self.assertRaises(owner.FastUpdateError):
self.update()
self.assertEqual((self.repo / 'openpilot/starpilot/ui/source.py').read_text(), 'local repair')
def test_staged_source_rejected(self):
(self.repo / 'openpilot/starpilot/ui/source.py').write_text('staged repair')
self.git(self.repo, 'add', 'openpilot/starpilot/ui/source.py')
with self.assertRaises(owner.FastUpdateError):
self.update()
self.assertEqual((self.repo / 'openpilot/starpilot/ui/source.py').read_text(), 'staged repair')
self.assertFalse(self.invalidations)
def test_reboot_write_failure_rolls_back(self):
def failing(*args, **kwargs):
raise RuntimeError('params write failed')
self.params.put_bool = failing
with self.assertRaisesRegex(RuntimeError, 'params write failed'):
self.update()
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
def test_fetch_failure_does_not_mutate(self):
def failing(command, cwd):
if 'fetch' in command:
raise RuntimeError('fetch failed')
return owner.run(command, cwd)
with self.assertRaisesRegex(RuntimeError, 'fetch failed'):
self.update(run=failing)
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.invalidations)
def test_apply_failure_rolls_back(self):
failed = False
def failing(command, cwd):
nonlocal failed
if 'reset' in command and command[-1] == self.target and not failed:
failed = True
owner.run(command, cwd)
raise RuntimeError('apply failed after reset')
return owner.run(command, cwd)
with self.assertRaisesRegex(RuntimeError, 'apply failed'):
self.update(run=failing)
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertEqual((self.repo / 'openpilot/starpilot/ui/source.py').read_text(), 'old')
self.assertFalse(self.params.calls)
def test_lost_parked_before_reboot_rolls_back(self):
states = iter([True, True, True, False])
with self.assertRaises(owner.FastUpdateError):
self.update(parked=lambda: next(states))
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.params.calls)
def test_lost_parked_before_source_mutation(self):
states = iter([True, False])
with self.assertRaises(owner.FastUpdateError):
self.update(parked=lambda: next(states))
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.invalidations)
def test_invalid_vendor_target_rejected_before_checkout(self):
(self.remote / 'upstream-sync.json').write_text('{}')
self.git(self.remote, 'commit', '-am', 'invalid')
with self.assertRaises(ValueError):
self.update()
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.invalidations)
def test_other_branch_rejected(self):
self.git(self.repo, 'checkout', '-b', 'Other')
with self.assertRaises(owner.FastUpdateError):
self.update()
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
def test_untracked_collision_rejected(self):
(self.repo / 'new-model').write_bytes(b'model')
(self.remote / 'new-model').write_text('tracked update')
self.git(self.remote, 'add', 'new-model')
self.receipt_commit()
with self.assertRaisesRegex(owner.FastUpdateError, 'overwrite untracked data'):
self.update()
self.assertEqual((self.repo / 'new-model').read_bytes(), b'model')
self.assertFalse(self.invalidations)
def test_post_apply_validation_failure_rolls_back(self):
original = owner.validate_revision
def validate(repo, revision):
if revision == 'HEAD':
raise ValueError('post-check failed')
return original(repo, revision)
with patch.object(owner, 'validate_revision', validate):
with self.assertRaisesRegex(ValueError, 'post-check failed'):
self.update()
self.assertEqual(self.git(self.repo, 'rev-parse', 'HEAD'), self.previous)
self.assertFalse(self.params.calls)
if __name__ == '__main__':
unittest.main()
@@ -14,6 +14,47 @@ class UpdaterControlTests(unittest.TestCase):
with self.assertRaises(update_control.UpdaterControlError):
update_control.send(0, 0, 'install')
def test_fast_wire_pins_branch_and_commit_without_changing_legacy_commands(self):
commit = 'a' * 40
with mock.patch.object(update_control, '_exchange') as exchange:
update_control.send(123, 456, 'fast', branch='SecretGoodStarPilot', commit=commit)
exchange.assert_called_once_with(123, 456, b'fast SecretGoodStarPilot ' + commit.encode())
self.assertEqual(update_control._parse(b'fast SecretGoodStarPilot ' + commit.encode() + b'\n'),
('fast', 'SecretGoodStarPilot', commit))
for action in ('check', 'download'):
self.assertEqual(update_control._command(action), action.encode())
self.assertEqual(update_control._parse(action.encode() + b'\n'), (action, None, None))
self.assertLessEqual(len(update_control._command('fast', 'x' * 128, commit)) + 1, update_control.MAX_COMMAND)
for raw in (b'fast\n', b'fast Dom bad\n', b'fast Dom ' + commit.encode(),
b'fast Dom ' + commit.encode() + b' extra\n', b'fast Dom\ncheck ' + commit.encode() + b'\n'):
with self.subTest(raw=raw), self.assertRaises(update_control.UpdaterControlError):
update_control._parse(raw)
for branch, revision in ((None, commit), ('Dom', None), ('Dom\ncheck', commit), ('x' * 129, commit), ('Dom', 'A' * 40)):
with self.subTest(branch=branch), self.assertRaises(update_control.UpdaterControlError):
update_control.send(123, 456, 'fast', branch=branch, commit=revision)
with self.assertRaises(update_control.UpdaterControlError):
update_control.send(123, 456, 'check', branch='Dom', commit=commit)
def test_server_delivers_fast_identity_and_legacy_callback_shape(self):
for raw, expected in ((b'fast Dom ' + b'a' * 40 + b'\n', mock.call('fast', branch='Dom', commit='a' * 40)),
(b'check\n', mock.call('check')), (b'download\n', mock.call('download'))):
with self.subTest(raw=raw):
client, connection = socket.socketpair()
try:
client.settimeout(1)
client.sendall(raw)
callback = mock.Mock()
listener = mock.Mock()
listener.accept.side_effect = [(connection, None), OSError()]
server = update_control.UpdaterControlServer(callback)
with mock.patch.object(update_control, '_peer', return_value=(os.getpid(), os.geteuid())):
server._run(listener)
self.assertEqual(client.recv(16), b'ok\n')
self.assertEqual(callback.call_args, expected)
finally:
client.close()
connection.close()
@unittest.skipUnless(sys.platform.startswith('linux') and hasattr(socket, 'SO_PEERCRED'), 'Linux peer credentials required')
def test_same_process_control_and_stale_identity(self):
actions = []
+39 -10
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
import os
import re
import socket
import struct
import threading
@@ -11,7 +12,9 @@ from pathlib import Path
TIMEOUT = 0.5
MAX_COMMAND = 16
MAX_COMMAND = 256
_BRANCH = re.compile(r"[A-Za-z0-9_][A-Za-z0-9._/+@-]{0,127}\Z")
_COMMIT = re.compile(r"[0-9a-f]{40}\Z")
_CREDENTIALS = struct.Struct('3i')
@@ -78,14 +81,32 @@ def available(pid: int, start: int) -> bool:
return False
def send(pid: int, start: int, action: str) -> None:
if action not in ('check', 'download'):
raise UpdaterControlError('Invalid updater control action')
_exchange(pid, start, action.encode('ascii'))
def _command(action: str, branch=None, commit=None) -> bytes:
if action in ('check', 'download') and branch is None and commit is None:
return action.encode('ascii')
if action == 'fast' and type(branch) is str and type(commit) is str and _BRANCH.fullmatch(branch) and _COMMIT.fullmatch(commit):
return f'fast {branch} {commit}'.encode('ascii')
raise UpdaterControlError('Invalid updater control action')
def _parse(command: bytes):
if command in (b'status\n', b'check\n', b'download\n'):
return command[:-1].decode('ascii'), None, None
try:
action, branch, commit = command[:-1].decode('ascii').split(' ')
except (UnicodeError, ValueError):
raise UpdaterControlError('Invalid updater control message') from None
if not command.endswith(b'\n') or action != 'fast' or _command(action, branch, commit) + b'\n' != command:
raise UpdaterControlError('Invalid updater control message')
return action, branch, commit
def send(pid: int, start: int, action: str, *, branch=None, commit=None) -> None:
_exchange(pid, start, _command(action, branch, commit))
class UpdaterControlServer:
def __init__(self, request: Callable[[str], None]):
def __init__(self, request: Callable[..., None]):
self.request = request
self.socket: socket.socket | None = None
self.thread: threading.Thread | None = None
@@ -123,13 +144,21 @@ class UpdaterControlServer:
try:
_, uid = _peer(connection)
command = _line(connection)
if uid != os.geteuid() or command not in (b'status\n', b'check\n', b'download\n'):
if uid != os.geteuid():
connection.sendall(b'error\n')
else:
if command != b'status\n':
self.request(command[:-1].decode('ascii'))
action, branch, commit = _parse(command)
if action == 'fast':
self.request(action, branch=branch, commit=commit)
elif action != 'status':
self.request(action)
connection.sendall(b'ok\n')
except (OSError, RuntimeError, ValueError):
except (RuntimeError, ValueError):
try:
connection.sendall(b'error\n')
except OSError:
pass
except OSError:
pass
def close(self) -> None:
@@ -0,0 +1,152 @@
from pathlib import Path
import tempfile
import threading
import sys
from types import ModuleType
import unittest
from unittest.mock import Mock, patch
from openpilot.system.updated.tests.test_vendored_update import load_updater
class StopLoop(Exception):
pass
class TestFastRequest(unittest.TestCase):
def setUp(self):
self.updated = load_updater()
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
root = Path(self.directory.name)
self.updated.LOCK_FILE = str(root / 'lock')
self.updated.STAGING_ROOT = str(root / 'staging')
self.updated.OVERLAY_INIT = root / 'overlay'
self.params = Mock()
self.values = {'InstallDate': '2026-10-03', 'UpdateFailedCount': 2}
self.writes = []
self.params.get.side_effect = lambda key, **kwargs: self.values.get(key)
self.params.get_bool.side_effect = lambda key: bool(self.values.get(key, False))
def put(key, value, **kwargs):
self.writes.append((key, value))
self.values[key] = value
self.params.put.side_effect = put
self.params.put_bool.side_effect = put
self.params.remove.side_effect = lambda key: self.values.pop(key, None)
self.updated.Params = Mock(return_value=self.params)
self.helper = self.updated.WaitTimeHelper.__new__(self.updated.WaitTimeHelper)
self.helper.ready_event = threading.Event()
self.helper.request_lock = threading.RLock()
self.helper.request_generation = 0
self.helper.fast_target = None
self.helper.user_request = self.updated.UserRequest.NONE
self.helper._control_request('fast', branch='SecretGoodStarPilot', commit='a' * 40)
self.helper.sleep = Mock(side_effect=StopLoop)
self.updated.WaitTimeHelper = Mock(return_value=self.helper)
self.updater = Mock()
self.updater.fast_update.return_value = 'up-to-date'
self.updated.Updater = Mock(return_value=self.updater)
self.updated.init_overlay = Mock()
self.updated.set_consistent_flag = Mock()
def run_cycle(self):
with self.assertRaises(StopLoop):
self.updated.main()
self.updated.init_overlay.assert_not_called()
self.updater.check_for_update.assert_not_called()
self.updater.fetch_update.assert_not_called()
self.updater.set_params.assert_not_called()
self.updated.system_time_valid.assert_not_called()
self.updater.fast_update.assert_called_once_with('SecretGoodStarPilot', 'a' * 40)
def test_fast_request_skips_staging_and_startup_delay(self):
self.run_cycle()
self.assertEqual(self.values['UpdaterState'], 'idle')
self.assertEqual(self.values['UpdateFailedCount'], 0)
self.assertIn('UpdaterLastFetchTime', self.values)
self.assertIn('LastUpdateTime', self.values)
self.assertNotIn('DoReboot', self.values)
self.assertLess(self.writes.index(('UpdateFailedCount', 0)), len(self.writes) - 1)
self.assertEqual(self.writes[-1], ('UpdaterState', 'idle'))
def test_fast_failure_reports_error_without_reboot(self):
self.updater.fast_update.side_effect = RuntimeError('Tracked source has local changes')
self.run_cycle()
self.assertEqual(self.values['UpdaterState'], 'idle')
self.assertEqual(self.values['UpdateFailedCount'], 3)
self.assertEqual(self.values['LastUpdateException'], 'Tracked source has local changes')
self.assertNotIn('DoReboot', self.values)
self.assertNotIn('LastUpdateTime', self.values)
def test_long_failure_remains_readable_by_galaxy(self):
self.updater.fast_update.side_effect = RuntimeError('download failed: ' + 'é' * 5000)
self.run_cycle()
self.assertLessEqual(len(self.values['LastUpdateException'].encode('utf-8')), 4096)
self.assertTrue(self.values['LastUpdateException'].startswith('download failed: '))
self.assertEqual(self.values['UpdaterState'], 'idle')
def test_success_waits_for_manager_restart(self):
def apply(branch, commit):
self.params.put_bool('DoReboot', True, block=True)
return 'reboot-requested'
self.updater.fast_update.side_effect = apply
self.run_cycle()
self.assertEqual(self.values['UpdaterState'], 'updating...')
self.assertTrue(self.values['DoReboot'])
def test_new_request_is_not_lost_when_previous_work_finishes(self):
_, generation, target = self.helper.current_request()
self.assertEqual(target, ('SecretGoodStarPilot', 'a' * 40))
self.helper._control_request('check')
self.assertFalse(self.helper.finish_request(generation))
self.assertTrue(self.helper.ready_event.is_set())
request, generation, target = self.helper.current_request()
self.assertIsNone(target)
self.assertEqual(request, self.updated.UserRequest.CHECK)
self.assertTrue(self.helper.finish_request(generation))
self.assertEqual(self.helper.user_request, self.updated.UserRequest.NONE)
self.assertFalse(self.helper.ready_event.is_set())
class TestFastAdapter(unittest.TestCase):
def test_source_admission_and_overlay_invalidation_are_owned_by_updater(self):
updated = load_updater()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
updated.FINALIZED = str(root / 'finalized')
Path(updated.FINALIZED).mkdir()
marker = Path(updated.FINALIZED, '.overlay_consistent')
marker.touch()
updated.OVERLAY_INIT = root / '.overlay_init'
updated.OVERLAY_INIT.touch()
updated.BASEDIR = str(root)
updated.dismount_overlay = Mock()
updated.HARDWARE.get_os_version.return_value = '19.8.1'
updater = updated.Updater()
updater.params = Mock()
updater.params.get_bool.side_effect = lambda key: key == 'IsOffroad'
source = Mock()
source.allowed.return_value = True
source.effective.return_value = False
physical = ModuleType('openpilot.starpilot.drive_state.evidence')
physical.PhysicalSource = Mock(return_value=source)
owner = ModuleType('openpilot.starpilot.software.fast_update')
def install(repo, branch, **options):
self.assertEqual((repo, branch, options['expected_commit'], options['current_os']),
(root, 'SecretGoodStarPilot', 'b' * 40, '19.8.1'))
self.assertTrue(options['parked']())
self.assertTrue(marker.exists())
options['invalidate']()
self.assertFalse(marker.exists())
self.assertFalse(updated.OVERLAY_INIT.exists())
updated.dismount_overlay.assert_called_once()
return 'reboot-requested'
owner.fast_update = install
with patch.dict(sys.modules, {physical.__name__: physical, owner.__name__: owner}):
self.assertEqual(updater.fast_update('SecretGoodStarPilot', 'b' * 40), 'reboot-requested')
source.close.assert_called_once()
updater.params.put_bool.assert_called_once_with('UpdateAvailable', False, block=True)
if __name__ == '__main__':
unittest.main()
@@ -38,7 +38,7 @@ class TestUnpublishedBranch(unittest.TestCase):
def run(command, cwd):
self.assertEqual(cwd, str(self.merged))
self.commands.append(command)
if command == ["git", "ls-remote", "--heads"]:
if command == ["git", "ls-remote", "--heads", "origin"]:
return self.remote_heads
if command == ["git", "ls-remote", "origin", "HEAD"]:
return f"{REMOTE}\tHEAD\n"
@@ -55,7 +55,7 @@ class TestUnpublishedBranch(unittest.TestCase):
self.assertFalse(self.updater.update_ready)
self.assertFalse(self.updater.fetch_update())
self.assertNotIn(TARGET, self.updater.branches)
self.assertEqual(self.commands.count(["git", "ls-remote", "--heads"]), 1)
self.assertEqual(self.commands.count(["git", "ls-remote", "--heads", "origin"]), 1)
self.updated.set_consistent_flag.assert_not_called()
self.assertNotIn("UpdaterState", self.updater.params.values)
@@ -94,7 +94,7 @@ class TestUnpublishedBranch(unittest.TestCase):
self.assertTrue(self.updater.update_available)
def failed_run(command, cwd):
if command == ["git", "ls-remote", "--heads"]:
if command == ["git", "ls-remote", "--heads", "origin"]:
raise subprocess.CalledProcessError(128, command, output="network failed")
return f"{REMOTE}\tHEAD\n"
+115 -51
View File
@@ -45,11 +45,15 @@ class UserRequest:
NONE = 0
CHECK = 1
FETCH = 2
FAST = 3
class WaitTimeHelper:
def __init__(self):
self.ready_event = threading.Event()
self.user_request = UserRequest.NONE
self.request_lock = threading.RLock()
self.request_generation = 0
self.fast_target = None
signal.signal(signal.SIGHUP, self.update_now)
signal.signal(signal.SIGUSR1, self.check_now)
self.control = UpdaterControlServer(self._control_request)
@@ -60,24 +64,39 @@ class WaitTimeHelper:
else:
atexit.register(self.control.close)
def _control_request(self, action: str) -> None:
if action == 'check':
self.user_request = UserRequest.CHECK
elif action == 'download':
self.user_request = UserRequest.FETCH
else:
return
self.ready_event.set()
def _request(self, request: int, *, branch=None, commit=None) -> None:
with self.request_lock:
self.user_request = request
self.fast_target = (branch, commit) if request == UserRequest.FAST else None
self.request_generation += 1
self.ready_event.set()
def _control_request(self, action: str, *, branch=None, commit=None) -> None:
request = {'check': UserRequest.CHECK, 'download': UserRequest.FETCH, 'fast': UserRequest.FAST}.get(action)
if request is not None:
self._request(request, branch=branch, commit=commit)
def update_now(self, signum: int, frame) -> None:
cloudlog.info("caught SIGHUP, attempting to downloading update")
self.user_request = UserRequest.FETCH
self.ready_event.set()
self._request(UserRequest.FETCH)
def check_now(self, signum: int, frame) -> None:
cloudlog.info("caught SIGUSR1, checking for updates")
self.user_request = UserRequest.CHECK
self.ready_event.set()
self._request(UserRequest.CHECK)
def current_request(self) -> tuple[int, int, tuple[str, str] | None]:
with self.request_lock:
self.ready_event.clear()
return self.user_request, self.request_generation, self.fast_target
def finish_request(self, generation: int) -> bool:
with self.request_lock:
if self.request_generation != generation:
return False
self.user_request = UserRequest.NONE
self.fast_target = None
self.ready_event.clear()
return True
def sleep(self, t: float) -> None:
self.ready_event.wait(timeout=t)
@@ -374,15 +393,14 @@ class Updater:
excluded_branches = ('release2', 'release2-staging')
try:
run(["git", "ls-remote", "origin", "HEAD"], OVERLAY_MERGED)
self._has_internet = True
except subprocess.CalledProcessError:
self._has_internet = False
self._branches_checked = False
setup_git_options(OVERLAY_MERGED)
output = run(["git", "ls-remote", "--heads"], OVERLAY_MERGED)
try:
output = run(["git", "ls-remote", "--heads", "origin"], OVERLAY_MERGED)
except subprocess.CalledProcessError:
self._has_internet = False
raise
self._has_internet = True
self.branches.clear()
for line in output.split('\n'):
@@ -403,6 +421,35 @@ class Updater:
else:
cloudlog.info(f"up to date on {cur_branch} ({str(cur_commit)[:7]})")
def fast_update(self, branch: str, commit: str) -> str:
from openpilot.starpilot.drive_state.evidence import PhysicalSource
from openpilot.starpilot.software.fast_update import fast_update
source = PhysicalSource()
def parked():
deadline = time.monotonic() + 0.75
while True:
if (self.params.get_bool("IsOffroad") and not self.params.get_bool("IsOnroad") and
source.allowed() and source.effective() is False):
return True
if time.monotonic() >= deadline:
return False
time.sleep(0.02)
def invalidate():
Path(FINALIZED, ".overlay_consistent").unlink(missing_ok=True)
OVERLAY_INIT.unlink(missing_ok=True)
dismount_overlay()
self.params.put_bool("UpdateAvailable", False, block=True)
self._branches_checked = False
try:
return fast_update(Path(BASEDIR), branch, expected_commit=commit, params=self.params, parked=parked,
current_os=HARDWARE.get_os_version(), invalidate=invalidate)
finally:
source.close()
def fetch_update(self) -> bool:
if not self._branches_checked:
self.check_for_update()
@@ -475,7 +522,7 @@ def main() -> None:
params.put("InstallDate", t, block=True)
updater = Updater()
update_failed_count = 0 # TODO: Load from param?
update_failed_count = max(0, params.get("UpdateFailedCount") or 0)
wait_helper = WaitTimeHelper()
# invalidate old finalized update
@@ -487,39 +534,47 @@ def main() -> None:
# Run the update loop
first_run = True
while True:
wait_helper.ready_event.clear()
requested, request_generation, fast_target = wait_helper.current_request()
# Attempt an update
exception = None
fast_requested = requested == UserRequest.FAST
try:
# TODO: reuse overlay from previous updated instance if it looks clean
init_overlay()
# ensure we have some params written soon after startup
updater.set_params(False, update_failed_count, exception)
if not system_time_valid() or first_run:
first_run = False
wait_helper.sleep(60)
continue
update_failed_count += 1
# check for update
params.put("UpdaterState", "checking...", block=True)
updater.check_for_update()
# download update
last_fetch = params.get("UpdaterLastFetchTime")
timed_out = last_fetch is None or (datetime.datetime.now(datetime.UTC).replace(tzinfo=None) - last_fetch > datetime.timedelta(days=3))
user_requested_fetch = wait_helper.user_request == UserRequest.FETCH
if params.get_bool("NetworkMetered") and not timed_out and not user_requested_fetch:
cloudlog.info("skipping fetch, connection metered")
elif not download_permitted(params, manual=user_requested_fetch, check_only=wait_helper.user_request == UserRequest.CHECK):
cloudlog.info("skipping fetch, check-only request or automatic downloads disabled")
if fast_requested:
update_failed_count += 1
params.put("UpdaterState", "updating...", block=True)
updater.fast_update(*fast_target)
write_time_to_param(params, "UpdaterLastFetchTime")
write_time_to_param(params, "LastUpdateTime")
else:
if updater.fetch_update():
write_time_to_param(params, "UpdaterLastFetchTime")
# TODO: reuse overlay from previous updated instance if it looks clean
init_overlay()
# ensure we have some params written soon after startup
updater.set_params(False, update_failed_count, exception)
if not system_time_valid() or first_run:
first_run = False
wait_helper.sleep(60)
continue
update_failed_count += 1
# check for update
params.put("UpdaterState", "checking...", block=True)
updater.check_for_update()
# download update
last_fetch = params.get("UpdaterLastFetchTime")
timed_out = last_fetch is None or (datetime.datetime.now(datetime.UTC).replace(tzinfo=None) - last_fetch > datetime.timedelta(days=3))
user_requested_fetch = requested == UserRequest.FETCH
if params.get_bool("NetworkMetered") and not timed_out and not user_requested_fetch:
cloudlog.info("skipping fetch, connection metered")
elif not download_permitted(params, manual=user_requested_fetch, check_only=requested == UserRequest.CHECK):
cloudlog.info("skipping fetch, check-only request or automatic downloads disabled")
else:
if updater.fetch_update():
write_time_to_param(params, "UpdaterLastFetchTime")
update_failed_count = 0
except subprocess.CalledProcessError as e:
cloudlog.event(
@@ -536,14 +591,23 @@ def main() -> None:
OVERLAY_INIT.unlink(missing_ok=True)
try:
params.put("UpdaterState", "idle", block=True)
update_successful = (update_failed_count == 0)
updater.set_params(update_successful, update_failed_count, exception)
if fast_requested:
params.put("UpdateFailedCount", update_failed_count, block=True)
if exception is None:
params.remove("LastUpdateException")
else:
params.put("LastUpdateException", exception.encode("utf-8")[:4096].decode("utf-8", errors="ignore"), block=True)
params.put("UpdaterState", "updating..." if params.get_bool("DoReboot") else "idle", block=True)
else:
params.put("UpdaterState", "idle", block=True)
updater.set_params(update_successful, update_failed_count, exception)
except Exception:
cloudlog.exception("uncaught updated exception while setting params, shouldn't happen")
# infrequent attempts if we successfully updated recently
wait_helper.user_request = UserRequest.NONE
if not wait_helper.finish_request(request_generation):
continue
wait_helper.sleep(5*60 if update_failed_count > 0 else 1.5*60*60)
+1
View File
@@ -616,6 +616,7 @@ run_larch64_build() {
# QCOM-compiled driving/DM models and both camera warp families.
python3 -m tools.laptop_device_build.package_model_chunks --source "${ROOT_DIR}/openpilot/selfdrive/modeld/models" --destination "${ROOT_DIR}/openpilot/selfdrive/modeld/models"
verify_device_artifacts
python3 "${ROOT_DIR}/openpilot/common/prebuilt_manifest.py" "${ROOT_DIR}"
touch "${ROOT_DIR}/prebuilt"
}
+1
View File
@@ -55,6 +55,7 @@ case "${mode}" in
fi
python3 -m tools.laptop_device_build.package_model_chunks --source "${ROOT_DIR}/openpilot/selfdrive/modeld/models" --destination "${ROOT_DIR}/openpilot/selfdrive/modeld/models"
python3 "${ROOT_DIR}/tools/laptop_device_build/validate_artifacts.py" "${ROOT_DIR}"
python3 "${ROOT_DIR}/openpilot/common/prebuilt_manifest.py" "${ROOT_DIR}"
touch prebuilt
;;
*)