ninharp/ninlight-w600
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

fix(station): Freigabe atomar verbrauchen, Schreibfehler nicht stillschweigend armed lassen

Browse Source 
Fixrunde 1 zu Task 3 (Migrationsroute):

- C1 (Critical): read_control()/Phasenpruefung/Auslieferung/write_control("listen") liefen
  ohne gemeinsame Sperre - zwei gleichzeitige Anfragen (zwei Lampen, oder Erst- und
  Range-Resume-Verbindung derselben Lampe) konnten beide phase == "armed" sehen und beide
  ausgeliefert bekommen. Neues Station.claim_delivery() liest und verbraucht die Freigabe
  jetzt atomar unter station.lock, vor Beginn der Auslieferung; Station.restore_armed()
  macht das bei einem Abbruch rueckgaengig.
- I2 (Important): eine unerwartet kurz gelesene Datei zaehlte im Sendeloop als "ok" statt
  als "aborted" - relevant geworden, weil dieser Status jetzt entscheidet, ob eine armed-
  Freigabe restauriert wird.
- I3 (Important): ein fehlschlagendes write_control() (z.B. schreibgeschuetztes /run)
  verschluckte den Fehler und liess "armed" dauerhaft scharf. Der Fehler landet jetzt in
  status.json (control_error), und claim_delivery() laesst nur die eine ausloesende
  Anfrage noch durch, bis ein Schreibversuch wieder gelingt.
- I4 (Important): die neuen Server-Tests synchronisierten sich per fester Pause statt wie
  Station.read_state() bis zu einem Praedikat zu pollen. status()/read_control() im
  Testklienten tun das jetzt ebenfalls.
- I5 (Important): resolve_mode()/--mode-file waren ungetestet trotz bindender
  Vorrangregel, auf der Task 4 aufbaut - fuenf neue reine Funktionstests.

Dazu: recent[0] -> recent[-1] in allen betroffenen Tests (Brief-Fehler), ein neuer
Rechte-Test fuer control.json (0644 unter restriktivem umask), und eine neue
Nebenlaeufigkeits-Regression fuer C1.
This commit is contained in:
Michael Sauer
2026-09-13 18:22:49 +02:00
parent b41ed51018
commit d3e608dca0
2 changed files with 265 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/station/server.py
+77 -9
View File
@@ -78,6 +78,13 @@ class Station:
self.last_seen: dict | None = None
self._manifest_mtime: float | None = None
self._manifest_version = "unbekannt"
# Wird True, wenn write_control() zuletzt fehlgeschlagen ist (z.B. volles oder
# schreibgeschuetztes /run) - claim_delivery() liest das, um "armed" nicht
# dauerhaft scharf zu lassen, nur weil das Verbrauchen der Freigabe nicht auf
# Platte geschrieben werden konnte (Fixrunde 1, I3). control_error landet in
# status.json, damit ein Bedienfehler nicht nur auf stdout verschwindet.
self._control_write_broken = False
self.control_error: str | None = None
# Cumulative bytes already delivered to each client while --stall-after is active,
# keyed by client IP. A lamp whose download stalls re-opens a fresh connection and
# resumes with Range right where it left off, so a per-request counter would hand
@@ -150,7 +157,51 @@ class Station:
os.unlink(tmp)
raise
except OSError as e:
print(f"warning: cannot write {self.control_path}: {e}", flush=True)
self._control_write_broken = True
self.control_error = f"cannot write {self.control_path}: {e}"
print(f"warning: {self.control_error}", flush=True)
else:
self._control_write_broken = False
self.control_error = None
def claim_delivery(self, target: str) -> tuple[str, str | None, bool]:
"""Entscheidet unter dem Lock, ob eine Anfrage fuer `target` gerade ausgeliefert
werden darf, und verbraucht - falls die Freigabe dafuer verantwortlich ist - diese
sofort per Compare-and-Set (lesen und auf "listen" setzen, ohne das Lock
dazwischen loszulassen). Ohne das koennten zwei gleichzeitige Anfragen (zwei
Lampen, oder Erst- und Range-Resume-Verbindung derselben Lampe) beide phase ==
"armed" sehen und beide ausgeliefert bekommen (Fixrunde 1, C1) - genau das Szenario,
fuer das die Freigabephase laut Spec 3.3 da ist.
Liefert (phase, armed_target) wie sie VOR dem Verbrauch galten, plus ob die
aktuelle Anfrage ausliefern darf. Bricht die Auslieferung anschliessend ab
(status == "aborted") oder findet der Aufrufer kein Image, muss er bei
phase == "armed" restore_armed(target) aufrufen, damit die Freigabe fuer den
naechsten Bootversuch stehen bleibt.
Schlaegt das Schreiben von "listen" fehl (I3), bekommt nur die eine Anfrage, die
den Fehler zuerst ausloest, noch ihre Auslieferung - jede weitere schlaegt hier
erneut fehl (_control_write_broken bleibt True), bis ein write_control()-Versuch
wieder klappt. Damit bleibt "armed" nicht dauerhaft scharf, nur weil /run gerade
nicht beschreibbar ist, heilt sich aber von selbst, sobald es das wieder ist."""
with self.lock:
phase, armed_target = self.read_control()
if phase == "listen" or (phase == "armed" and armed_target != target):
return phase, armed_target, False
if phase == "armed":
was_broken = self._control_write_broken
self.write_control("listen", None)
if self._control_write_broken:
return phase, armed_target, not was_broken
return phase, armed_target, True
def restore_armed(self, target: str) -> None:
"""Setzt eine per claim_delivery() verbrauchte Freigabe zurueck, wenn die
Auslieferung danach doch nicht (vollstaendig) stattgefunden hat - ein Abbruch darf
die Freigabe nicht verbrauchen, sonst muesste am Geraet erneut von Hand gearmt
werden, obwohl die Lampe es beim naechsten Bootversuch selbst nochmal versuchen
wuerde (Spec 3.3, Brief-Schritt 3.6)."""
self.write_control("armed", target)
def reserve_stall_bytes(self, ip: str, n: int) -> int:
"""How many of the next `n` bytes this client may still receive under --stall-after.
@@ -205,6 +256,7 @@ class Station:
payload["phase"] = phase
payload["armed_target"] = armed_target
payload["last_seen"] = self.last_seen
payload["control_error"] = self.control_error
try:
self.state_path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp = tempfile.mkstemp(dir=self.state_path.parent, prefix=".status-")
@@ -322,9 +374,12 @@ class Handler(BaseHTTPRequestHandler):
target, variant, name = product.target, product.variant, product.name
version = self.station.manifest_version()
phase, armed_target = self.station.read_control()
# claim_delivery() liest die Phase UND verbraucht eine armed-Freigabe atomar unter
# station.lock (Fixrunde 1, C1) - zwischen dem Lesen und dem Verbrauchen darf keine
# zweite gleichzeitige Anfrage hindurchschluepfen, siehe die Methode selbst.
phase, armed_target, allowed = self.station.claim_delivery(target)
if phase == "listen" or (phase == "armed" and armed_target != target):
if not allowed:
# last_seen wird trotzdem gesetzt - das ist der ganze Zweck der Horchphase:
# die Anzeige soll zeigen, welches Modell da ist, auch wenn nichts rausgeht.
self.station.record_migration(ip, target, variant, pid, lamp_version, name,
@@ -335,6 +390,10 @@ class Handler(BaseHTTPRequestHandler):
path = self.station.migrate_image_path(target, variant)
if not path.is_file():
if phase == "armed":
# claim_delivery() hat die Freigabe schon verbraucht, bevor wir wussten,
# dass es gar kein Image gibt - das darf nicht auf Kosten der Freigabe gehen.
self.station.restore_armed(target)
self.station.record_migration(ip, target, variant, pid, lamp_version, name,
version, 0, "notfound")
self.reject(404, "no image for this device")
@@ -342,12 +401,12 @@ class Handler(BaseHTTPRequestHandler):
return
sent, status = self._serve_file(path, started)
if phase == "armed" and status == "ok":
# Erst nach dem vollstaendigen Senden zurueck auf "listen" - _serve_file() ist
# an dieser Stelle bereits fertig. Bricht der Transfer ab (status == "aborted"),
# bleibt die Freigabe stehen, damit die Lampe es beim naechsten Boot noch
# einmal versuchen kann.
self.station.write_control("listen", None)
if phase == "armed" and status == "aborted":
# Bricht der Transfer ab, bleibt die Freigabe stehen, damit die Lampe es beim
# naechsten Boot noch einmal versuchen kann (Spec 3.3). claim_delivery() hat sie
# bereits auf "listen" gesetzt, bevor _serve_file() ueberhaupt anfing zu senden
# (das ist der Punkt von C1) - bei einem Abbruch wird das hier zurueckgenommen.
self.station.restore_armed(target)
self.station.record_migration(ip, target, variant, pid, lamp_version, name, version,
sent, status)
@@ -403,6 +462,15 @@ class Handler(BaseHTTPRequestHandler):
raise ConnectionResetError("peer closed during stall")
chunk = f.read(read_size)
if not chunk:
# Die Datei ist kuerzer als das eingangs per stat() ermittelte
# Content-Length (z.B. von deploy.sh waehrend einer laufenden
# Auslieferung durch eine kleinere ersetzt) - sent < remaining an
# dieser Stelle immer, denn die Schleifenbedingung oben haette sonst
# schon abgebrochen. Das ist ein abgebrochener, kein erfolgreicher
# Transfer (Fixrunde 1, I2): ohne dies wuerde eine im Migrationsmodus
# laufende "armed"-Freigabe faelschlich als verbraucht gelten, obwohl
# die Lampe nie das vollstaendige Image bekommen hat.
status = "aborted"
break
self.wfile.write(chunk)
sent += len(chunk)
tools/station/tests/test_server.py
+188 -31
View File
@@ -472,20 +472,28 @@ class _Response:
class MigrationTestServer:
"""Startet server.py mit --mode/--control und bietet die vom Brief geforderten
Hilfsmethoden. server.py schreibt status.json/control.json fuer eine Anfrage im
selben Handler-Thread, unmittelbar nachdem der Antwortkoerper vollstaendig gesendet
wurde - aber http.client's read() kann zurueckkehren, sobald Content-Length Bytes
angekommen sind, was diesem serverseitigen Schreiben knapp zuvorkommen kann. get()
legt deshalb eine kurze Pause ein, bevor es zurueckkehrt (derselbe Grund, aus dem
Station.read_state() oben mit einem Prädikat auf status.json wartet - hier gibt es
kein generisches Prädikat, das jeder Aufrufer kennt, darum die feste kleine Pause statt
einer Wiederholschleife)."""
Hilfsmethoden.
status()/read_control() folgen demselben Muster wie Station.read_state() oben
(Zeilen 68-90): sie pollen bis zu einigen Sekunden auf ein optionales Praedikat, statt
sich auf eine feste Pause nach get() zu verlassen. Grund ist derselbe wie dort -
server.py schreibt status.json/control.json fuer eine Anfrage erst, nachdem der
Antwortkoerper vollstaendig gesendet ist, aber http.client's read() kann zurueckkehren,
sobald Content-Length Bytes angekommen sind, was diesem Schreiben knapp zuvorkommen
kann. Jeder Aufrufer kennt sein eigenes Praedikat (das ohnehin schon als Assertion
dasteht), daher kein pauschales Warten hier."""
def __init__(self, tmp_path, mode):
self.image_dir = tmp_path / "images"
self.image_dir.mkdir()
self.state_path = tmp_path / "status.json"
self.control_path = tmp_path / "control.json"
# Eigenes Unterverzeichnis fuer die Steuerdatei, getrennt von status.json: ein Test
# fuer I3 (Fixrunde 1) macht genau dieses Verzeichnis schreibgeschuetzt, um einen
# fehlschlagenden write_control() zu erzwingen, und darf dabei nicht auch das
# Schreiben von status.json lahmlegen, ueber das derselbe Test den Fehler beobachtet.
control_dir = tmp_path / "control"
control_dir.mkdir()
self.control_path = control_dir / "control.json"
self.port = free_port()
self.proc = subprocess.Popen(
[sys.executable, str(SERVER), "--image-dir", str(self.image_dir),
@@ -511,30 +519,34 @@ class MigrationTestServer:
conn.request("GET", path, headers=headers or {})
r = conn.getresponse()
body = r.read()
resp = _Response(r.status, dict(r.getheaders()), body)
return _Response(r.status, dict(r.getheaders()), body)
finally:
conn.close()
# See the class docstring: a short pause so the server's own status.json/
# control.json write for this request has landed before we return.
time.sleep(0.05)
return resp
def _read_json(self, path, timeout=2.0):
deadline = time.time() + timeout
def _poll_json(self, path, predicate=None, timeout=5.0):
"""Wie Station.read_state() oben: liest `path` wiederholt (bis zu `timeout`
Sekunden), bis er sich parsen laesst und - falls angegeben - `predicate` erfuellt.
Ohne predicate reicht ein einziges erfolgreiches Parsen, genau wie dort."""
last = None
last_err = None
deadline = time.time() + timeout
while time.time() < deadline:
try:
return json.loads(path.read_text())
last = json.loads(path.read_text())
if predicate is None or predicate(last):
return last
except (OSError, json.JSONDecodeError) as e:
last_err = e
time.sleep(0.02)
raise AssertionError(f"could not read {path}: {last_err}")
time.sleep(0.02)
if last is None:
raise AssertionError(f"could not read {path}: {last_err}")
raise AssertionError(f"{path} never satisfied predicate, last seen: {last!r}")
def status(self):
return self._read_json(self.state_path)
def status(self, predicate=None):
return self._poll_json(self.state_path, predicate)
def read_control(self):
return self._read_json(self.control_path)
def read_control(self, predicate=None):
return self._poll_json(self.control_path, predicate)
def set_control(self, phase, armed_target=None):
tmp = self.control_path.with_name(self.control_path.name + ".tmp")
@@ -577,8 +589,9 @@ def test_migration_auto_serves_known_product(migration_server):
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1.1.1.1800-dev_ota_light.img")
assert r.status == 200
assert r.body == b"IMG" * 1000
assert srv.status()["last_seen"]["target"] == "g95_cw"
assert srv.status()["last_seen"]["variant"] == "dev"
st = srv.status(lambda s: s.get("last_seen") is not None)
assert st["last_seen"]["target"] == "g95_cw"
assert st["last_seen"]["variant"] == "dev"
def test_migration_listen_records_but_serves_nothing(migration_server):
@@ -588,10 +601,10 @@ def test_migration_listen_records_but_serves_nothing(migration_server):
srv.set_control(phase="listen")
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
assert r.status == 404
st = srv.status()
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["last_seen"]["target"] == "g95_cw" # erkannt
assert st["served"] == 0 # aber nichts ausgeliefert
assert st["recent"][0]["status"] == "listen"
assert st["recent"][-1]["status"] == "listen"
def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server):
@@ -601,7 +614,7 @@ def test_arm_serves_exactly_once_then_falls_back_to_listen(migration_server):
srv.set_control(phase="armed", armed_target="g95_cw")
p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
assert srv.get(p).status == 200
assert srv.read_control()["phase"] == "listen"
assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen"
assert srv.get(p).status == 404
@@ -619,14 +632,16 @@ def test_unknown_product_id_is_404_and_logged(migration_server):
srv, _ = migration_server
r = srv.get("/qcx_firmware_-acme.bulb_V1_ota_light.img")
assert r.status == 404
assert srv.status()["recent"][0]["status"] == "unknown"
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["recent"][-1]["status"] == "unknown"
def test_missing_migrate_image_is_404(migration_server):
srv, _ = migration_server
r = srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.e14_V1_ota_light.img")
assert r.status == 404
assert srv.status()["recent"][0]["status"] == "notfound"
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["recent"][-1]["status"] == "notfound"
def test_recovery_path_is_404_in_migration_mode(migration_server):
@@ -678,4 +693,146 @@ def test_version_comes_from_manifest(migration_server):
"ninlight_version": "0.3.9",
"images": [{"target": "g95_cw", "variant": "dev", "file": "g95_cw_dev.img"}]}))
srv.get("/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img")
assert srv.status()["recent"][0]["version"] == "0.3.9"
st = srv.status(lambda s: len(s["recent"]) >= 1)
assert st["recent"][-1]["version"] == "0.3.9"
def test_control_file_is_world_readable_even_under_a_restrictive_umask(tmp_path):
"""Dieselbe 0600-Falle wie bei status.json (siehe
test_state_file_is_world_readable_even_under_a_restrictive_umask oben), aber fuer
control.json: das CLI aus Task 4 muss die Datei lesen UND schreiben koennen."""
old_umask = os.umask(0o077)
try:
srv = MigrationTestServer(tmp_path, "migration")
finally:
os.umask(old_umask)
try:
(srv.image_dir / "migrate").mkdir(exist_ok=True)
(srv.image_dir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="armed", armed_target="g95_cw")
def world_readable():
mode = stat.S_IMODE(srv.control_path.stat().st_mode)
return bool(mode & stat.S_IROTH)
# set_control() oben schreibt direkt aus dem Test, nicht ueber den Server - das
# will hier nicht geprueft werden. Erst die Anfrage unten bringt den Server selbst
# dazu, control.json zu schreiben (armed -> listen), also denselben Codepfad
# (Station.write_control()) wie write_state() fuer status.json.
assert srv.get(
"/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img").status == 200
srv.read_control(lambda _: world_readable())
assert world_readable(), "control.json ist nach dem Schreiben durch den Server " \
"nicht welt-lesbar"
finally:
srv.stop()
def test_armed_serves_exactly_once_under_concurrency(migration_server):
"""C1 (Fixrunde 1): read_control()/Phasenpruefung/Auslieferung/write_control("listen")
liefen ohne gemeinsame Sperre ab - zwei gleichzeitige Anfragen (zwei Lampen, oder Erst-
und Range-Resume-Verbindung derselben Lampe) konnten beide phase == "armed" sehen und
beide die volle Auslieferung bekommen. station.claim_delivery() muss das Lesen und das
Verbrauchen der Freigabe atomar unter einer Sperre machen, damit von n gleichzeitigen
Anfragen gegen dieselbe Freigabe hoechstens eine durchkommt."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG" * 20000)
srv.set_control(phase="armed", armed_target="g95_cw")
p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
n_workers = 20
results = []
results_lock = threading.Lock()
def worker():
r = srv.get(p)
with results_lock:
results.append(r.status)
threads = [threading.Thread(target=worker) for _ in range(n_workers)]
for t in threads:
t.start()
for t in threads:
t.join(timeout=10)
assert len(results) == n_workers, "nicht alle Worker sind fertig geworden"
assert results.count(200) == 1, f"erwartet genau eine 200, bekommen: {results}"
assert results.count(404) == n_workers - 1
assert srv.read_control(lambda c: c.get("phase") == "listen")["phase"] == "listen"
def test_control_write_failure_blocks_further_armed_deliveries(migration_server):
"""I3 (Fixrunde 1): schlaegt das Schreiben von control.json fehl (z.B. volles oder
schreibgeschuetztes /run), darf die Freigabe nicht dauerhaft auf "armed" stehen bleiben
- sonst wuerde jede weitere Lampe des Ziels ebenfalls bedient, genau die Luecke aus C1,
nur diesmal durch einen Schreibfehler statt durch Nebenlaeufigkeit ausgeloest. Die eine
Anfrage, die den Fehler ausloest, bekommt noch ihre Auslieferung (wie eine normale
Freigabe); jede weitere schlaegt fehl, bis das Schreiben wieder klappt. Der Fehler muss
ausserdem in status.json sichtbar sein, sonst wundert sich niemand, warum "arm" nicht
mehr wirkt."""
srv, imgdir = migration_server
(imgdir / "migrate").mkdir(exist_ok=True)
(imgdir / "migrate" / "g95_cw_dev.img").write_bytes(b"IMG")
srv.set_control(phase="armed", armed_target="g95_cw")
p = "/qcx_firmware_-klyqa.lighting.cw-ww.g95-dev_V1_ota_light.img"
control_dir = srv.control_path.parent
old_mode = control_dir.stat().st_mode
os.chmod(control_dir, 0o500) # r-x fuer den Besitzer, kein Schreiben mehr moeglich
try:
r1 = srv.get(p)
assert r1.status == 200, "die eine vorgesehene Auslieferung muss noch rausgehen"
st = srv.status(lambda s: s.get("control_error"))
assert st["control_error"], "Schreibfehler muss in status.json sichtbar sein"
r2 = srv.get(p)
assert r2.status == 404, \
"ein Schreibfehler darf die Freigabe nicht dauerhaft offen lassen"
finally:
os.chmod(control_dir, old_mode)
def _load_server_module():
"""Laedt server.py wie test_log_error_is_not_silenced_by_log_message oben - als
eigenstaendiges Modulobjekt, ohne es in sys.modules einzutragen, damit mehrfache
Aufrufe (einer je resolve_mode-Test) sich nicht in die Quere kommen."""
import importlib.util
spec = importlib.util.spec_from_file_location("ninlight_station_server_rm", SERVER)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def test_resolve_mode_explicit_mode_wins_over_file(tmp_path):
"""I5 (Fixrunde 1): --mode schlaegt --mode-file."""
module = _load_server_module()
mode_file = tmp_path / "mode"
mode_file.write_text("migration")
assert module.resolve_mode("recovery", mode_file) == "recovery"
def test_resolve_mode_reads_the_mode_file(tmp_path):
module = _load_server_module()
mode_file = tmp_path / "mode"
mode_file.write_text("migration")
assert module.resolve_mode(None, mode_file) == "migration"
def test_resolve_mode_missing_file_means_recovery(tmp_path):
module = _load_server_module()
mode_file = tmp_path / "mode" # bewusst nicht angelegt
assert module.resolve_mode(None, mode_file) == "recovery"
def test_resolve_mode_garbage_file_means_recovery(tmp_path):
module = _load_server_module()
mode_file = tmp_path / "mode"
mode_file.write_text("garbage\n")
assert module.resolve_mode(None, mode_file) == "recovery"
def test_resolve_mode_default_without_mode_file_is_recovery():
module = _load_server_module()
assert module.resolve_mode(None, None) == "recovery"