AUDIT / src /workers /pyRunner.worker.ts
Arypulka98's picture
feat(audit): deploy full backend cluster node (part 4)
641b62c verified
Raw
History Blame Contribute Delete
18.6 kB
/**
* pyRunner.worker.ts — Pyodide Python execution in a Web Worker
*
* S45: micropip auto-detect — loadPackagesFromImports() + micropip via globals
* S70: package cache dedup — _installedPkgs Set persiste per tutta la vita del worker
* S71: estrazione globals post-run — dopo ogni esecuzione riuscita, estrae le variabili
* utente dal namespace Python e le invia come { type: "vars", id, vars[] }
* S448: Gap 2 — OPFS package cache persistente tra respawn + micropip preload + Pyodide 0.29.4
* - CDN aggiornato a v0.29.4 (Brotli compression migliorata, Safari 17+ supportato)
* - micropip precaricato in parallelo al bootstrap (elimina round-trip CDN separato)
* - _installedPkgs serializzato in OPFS: sopravvive a respawn Worker (respawn ogni 60s idle)
* - Impatto iPhone: run dopo respawn con OPFS piena: 15-20s → 2-3s
*
* NEVER loaded on the main thread — only in this worker.
*/
// loadPyodide viene importato dinamicamente in getPyodide() — ESM-compatible
interface PyodideInstance {
runPythonAsync(code: string): Promise<unknown>;
loadPackage(names: string | string[]): Promise<void>;
loadPackagesFromImports(code: string): Promise<void>;
globals: {
set(key: string, val: unknown): void;
};
isPyProxy(obj: unknown): boolean;
}
// ─── S71: tipo variabile Python serializzata ──────────────────────────────────
export interface PythonVarData {
name: string;
type: string;
preview: string;
value: string;
size?: string;
dtype?: string;
}
export type PyRunnerMsg =
| { type: "run"; id: string; code: string; timeout?: number; packages?: string[] }
| { type: "cancel"; id: string }
| { type: "ping" };
export type PyRunnerResult =
| { type: "stdout"; id: string; text: string }
| { type: "stderr"; id: string; text: string }
| { type: "done"; id: string; exitCode: number; time: number }
| { type: "error"; id: string; message: string }
| { type: "vars"; id: string; vars: PythonVarData[] }
| { type: "loading"; progress: number }
| { type: "ready" };
const PYODIDE_CDN = "https://cdn.jsdelivr.net/pyodide/v0.29.4/full/";
let pyodide: PyodideInstance | null = null;
let loadPromise: Promise<PyodideInstance> | null = null;
const running = new Map<string, { cancelled: boolean }>();
/** S70 — Package cache: persiste per tutta la vita del worker singleton */
const _installedPkgs = new Set<string>();
// ─── S448: OPFS package cache — persiste tra respawn del Worker ───────────────
// Chiave versionata per Pyodide: al cambio versione CDN la cache vecchia viene
// ignorata automaticamente (file handle non trovato → primo avvio pulito).
const OPFS_PKG_CACHE_KEY = "pyodide_pkg_cache_v0294";
async function loadPkgCacheFromOPFS(): Promise<void> {
try {
const root = await navigator.storage.getDirectory();
const fh = await root.getFileHandle(OPFS_PKG_CACHE_KEY, { create: false });
const file = await fh.getFile();
const text = await file.text();
const cached: string[] = JSON.parse(text);
for (const p of cached) _installedPkgs.add(p);
} catch {
// OPFS non supportato (Safari < 16.4) o file non esiste ancora — primo avvio
}
}
function savePkgCacheToOPFS(): void {
// fire-and-forget: non blocca l'esecuzione Python
void (async () => {
try {
const root = await navigator.storage.getDirectory();
const fh = await root.getFileHandle(OPFS_PKG_CACHE_KEY, { create: true });
const writable = await fh.createWritable();
await writable.write(JSON.stringify([..._installedPkgs]));
await writable.close();
} catch {
// non-fatal: cache OPFS non è critica per il funzionamento
}
})();
}
// ─── S71: Python extraction code ──────────────────────────────────────────────
// Dopo ogni runPythonAsync riuscito, questo script estrae le variabili utente
// dal namespace Python globale e le invia tramite __emit_var__(json_string).
// Skippati: nomi che iniziano con "_", nomi iniettati dal framework, moduli built-in.
const EXTRACT_GLOBALS_CODE = `
try:
import json as __json__
def __serialize_var__(name, val):
t = type(val).__name__
preview = ""
value = ""
size = None
dtype = None
try:
# numpy ndarray — check before generic object
if t not in ('bool','int','float','str','list','dict','tuple','set','complex','NoneType','bytes') and hasattr(val,'shape') and hasattr(val,'dtype') and hasattr(val,'flat'):
t = "ndarray"
size = "shape " + str(val.shape)
dtype = str(val.dtype)
flat = list(val.flat[:8])
preview = str(flat)[:80] + ("\\u2026" if val.size > 8 else "")
value = str(val)[:8000]
# pandas DataFrame (ha columns, non la Series)
elif hasattr(val,'to_string') and hasattr(val,'columns'):
t = "DataFrame"
rows_n, cols_n = val.shape
size = str(rows_n) + " righe \\u00d7 " + str(cols_n) + " col"
cols_p = [str(c) for c in val.columns[:6]]
preview = "[" + ", ".join(cols_p) + (", \\u2026" if cols_n > 6 else "") + "]"
value = val.to_string(max_rows=30, max_cols=12)[:8000]
# pandas Series (ha dtype e index ma non columns)
elif hasattr(val,'to_string') and hasattr(val,'dtype') and hasattr(val,'index') and not hasattr(val,'columns'):
t = "Series"
size = str(len(val)) + " el."
dtype = str(val.dtype)
preview = str(val.head(4)).replace('\\n',' ')[:80]
value = val.to_string(max_rows=30)[:8000]
elif isinstance(val, bool):
t = "bool"
preview = str(val)
value = str(val)
elif isinstance(val, int):
preview = str(val)
value = str(val)
elif isinstance(val, float):
preview = format(val, '.6g')
value = repr(val)
elif isinstance(val, complex):
t = "complex"
preview = str(val)
value = str(val)
elif val is None:
t = "NoneType"
preview = "None"
value = "None"
elif isinstance(val, str):
size = str(len(val)) + " char"
preview = repr(val[:60]) + ("\\u2026" if len(val) > 60 else "")
value = repr(val[:8000])
elif isinstance(val, bytes):
t = "bytes"
size = str(len(val)) + " bytes"
preview = repr(val[:40])
value = repr(val[:8000])
elif isinstance(val, (list, tuple)):
size = str(len(val)) + " el."
preview = str(val[:5])[:80] + ("\\u2026" if len(val) > 5 else "")
try:
value = __json__.dumps(val, ensure_ascii=False, default=repr)[:8000]
except Exception:
value = str(val)[:8000]
elif isinstance(val, dict):
size = str(len(val)) + " ch."
keys_p = list(val.keys())[:5]
preview = "{" + ", ".join(repr(k) for k in keys_p) + ("\\u2026" if len(val) > 5 else "") + "}"
try:
value = __json__.dumps(val, ensure_ascii=False, default=repr)[:8000]
except Exception:
value = str(val)[:8000]
elif isinstance(val, set):
size = str(len(val)) + " el."
preview = "{" + ", ".join(repr(v) for v in list(val)[:5]) + ("\\u2026" if len(val) > 5 else "") + "}"
value = str(sorted(str(v) for v in val))[:8000]
elif callable(val):
t = "function"
try:
import inspect as _insp
sig = str(_insp.signature(val))
preview = name + sig
del _insp
except Exception:
preview = "<function " + name + ">"
value = preview
elif hasattr(val,'__module__') and not callable(val):
t = "module"
preview = "<module '" + str(getattr(val,'__name__',name)) + "'>"
value = preview
else:
value = repr(val)[:8000]
preview = value[:80]
except Exception as __se__:
preview = "<err: " + str(__se__) + ">"
value = preview
result = {"name": name, "type": t, "preview": preview, "value": value}
if size is not None: result["size"] = size
if dtype is not None: result["dtype"] = dtype
return result
__skip_names__ = {
'__builtins__','__name__','__doc__','__package__','__loader__',
'__spec__','__import__','__stdout__','__stderr__','_Capture',
'__is_installed__','__mark_installed__','__pip_msg__',
'__code_to_analyze__','__emit_var__','__json__','__serialize_var__',
'__skip_names__','sys','io',
}
for __vk__, __vv__ in list(globals().items()):
if __vk__.startswith('_') or __vk__ in __skip_names__:
continue
try:
__emit_var__(__json__.dumps(__serialize_var__(__vk__, __vv__), ensure_ascii=False))
except Exception:
pass
except Exception as _pip_err:
_err_str = str(_pip_err)
if 'PackageNotFoundError' in _err_str or "Can't find" in _err_str or 'not found' in _err_str.lower():
__pip_msg__(f'[pip-unavailable] {_err_str}\n')
finally:
try:
_g = globals()
for _n in ['__json__','__serialize_var__','__skip_names__','__emit_var__','__vk__','__vv__','_g','_n']:
try: del _g[_n]
except Exception: pass
except Exception:
pass
`;
// ─── Pyodide load ─────────────────────────────────────────────────────────────
async function getPyodide(): Promise<PyodideInstance> {
if (pyodide) return pyodide;
if (loadPromise) return loadPromise;
self.postMessage({ type: "loading", progress: 0 } satisfies PyRunnerResult);
loadPromise = (async () => {
// S448: reidrata i package già installati da sessioni/respawn precedenti
await loadPkgCacheFromOPFS();
// ESM module workers non supportano importScripts — usa dynamic import
// @vite-ignore evita che Vite tenti di bundlare l'URL CDN remoto
const { loadPyodide } = await import(/* @vite-ignore */ `${PYODIDE_CDN}pyodide.mjs`);
self.postMessage({ type: "loading", progress: 30 } satisfies PyRunnerResult);
// S448: micropip precaricato in parallelo al bootstrap — elimina round-trip CDN separato
const py = await loadPyodide({ indexURL: PYODIDE_CDN, packages: ["micropip"] });
self.postMessage({ type: "loading", progress: 80 } satisfies PyRunnerResult);
pyodide = py;
self.postMessage({ type: "ready" } satisfies PyRunnerResult);
return py;
})();
return loadPromise;
}
// ─── S45/S70: auto-install imports con cache ──────────────────────────────────
async function autoInstallImports(
py: PyodideInstance,
code: string,
id: string,
): Promise<void> {
// Step 1 — standard Pyodide packages (Pyodide li cachea internamente)
try {
await py.loadPackagesFromImports(code);
} catch {
// non-fatal
}
// Step 2 — micropip per pacchetti PyPI puri non inclusi in Pyodide
try {
py.globals.set("__code_to_analyze__", code);
py.globals.set("__is_installed__", (pkg: string) => _installedPkgs.has(pkg));
py.globals.set("__mark_installed__", (pkg: string) => {
_installedPkgs.add(pkg);
savePkgCacheToOPFS(); // S448: aggiorna OPFS cache dopo ogni install
});
py.globals.set("__pip_msg__", (text: string) => {
self.postMessage({ type: "stdout", id, text } satisfies PyRunnerResult);
});
await py.runPythonAsync(`
import sys as _sys
try:
import micropip as _micropip
import ast as _ast
_tree = _ast.parse(__code_to_analyze__)
_missing: list[str] = []
for _node in _ast.walk(_tree):
if isinstance(_node, _ast.Import):
for _alias in _node.names:
_pkg = _alias.name.split(".")[0]
try:
__import__(_pkg)
except ImportError:
if _pkg not in _missing:
_missing.append(_pkg)
elif isinstance(_node, _ast.ImportFrom):
if _node.module:
_pkg = _node.module.split(".")[0]
try:
__import__(_pkg)
except ImportError:
if _pkg not in _missing:
_missing.append(_pkg)
_installed_count = 0
_cached_count = 0
for _pkg in _missing:
if __is_installed__(_pkg):
__pip_msg__(f"[pip] \\u2713 {_pkg} (cache)\\n")
_cached_count += 1
else:
__pip_msg__(f"[pip] Installo: {_pkg}\\u2026\\n")
await _micropip.install(_pkg, keep_going=True)
__mark_installed__(_pkg)
_installed_count += 1
if _installed_count > 0:
__pip_msg__(f"[pip] \\u2713 {_installed_count} pacchett{'o' if _installed_count == 1 else 'i'} installat{'o' if _installed_count == 1 else 'i'}\\n")
except Exception:
pass
finally:
try: del __code_to_analyze__, __is_installed__, __mark_installed__, __pip_msg__
except Exception: pass
`);
} catch {
// non-fatal
}
}
// ─── S71: estrazione variabili globali post-run ───────────────────────────────
async function extractAndSendVars(py: PyodideInstance, id: string): Promise<void> {
const vars: PythonVarData[] = [];
try {
py.globals.set("__emit_var__", (varJson: string) => {
try {
vars.push(JSON.parse(varJson) as PythonVarData);
} catch {
// non-fatal: skip malformed var
}
});
await py.runPythonAsync(EXTRACT_GLOBALS_CODE);
if (vars.length > 0) {
self.postMessage({ type: "vars", id, vars } satisfies PyRunnerResult);
}
} catch {
// non-fatal: vars extraction failure non blocca l'esecuzione
}
}
// ─── Message handler ──────────────────────────────────────────────────────────
self.onmessage = async (e: MessageEvent<PyRunnerMsg>) => {
const msg = e.data;
if (msg.type === "ping") {
self.postMessage({ type: "ready" } satisfies PyRunnerResult);
return;
}
if (msg.type === "cancel") {
const ctx = running.get(msg.id);
if (ctx) ctx.cancelled = true;
return;
}
if (msg.type === "run") {
const { id, code, timeout = 30_000, packages = [] } = msg;
const ctx = { cancelled: false };
running.set(id, ctx);
const start = Date.now();
let timedOut = false;
const timer = setTimeout(() => {
timedOut = true;
ctx.cancelled = true;
self.postMessage({ type: "stderr", id, text: "[timeout] Python execution exceeded " + timeout + "ms\n" } satisfies PyRunnerResult);
self.postMessage({ type: "done", id, exitCode: 124, time: Date.now() - start } satisfies PyRunnerResult);
}, timeout);
try {
const py = await getPyodide();
if (ctx.cancelled) { clearTimeout(timer); running.delete(id); return; }
// S45/S70: auto-detect e installa imports (standard Pyodide + micropip con cache)
await autoInstallImports(py, code, id);
if (ctx.cancelled) { clearTimeout(timer); running.delete(id); return; }
// Pacchetti espliciti richiesti dal caller (retrocompatibilità)
if (packages.length > 0) {
const toInstall = packages.filter(p => !_installedPkgs.has(p));
const cached = packages.filter(p => _installedPkgs.has(p));
if (cached.length > 0) {
self.postMessage({ type: "stdout", id, text: `[pip] ✓ ${cached.join(", ")} (cache)\n` } satisfies PyRunnerResult);
}
if (toInstall.length > 0) {
self.postMessage({ type: "stdout", id, text: `[pip] Installo: ${toInstall.join(", ")}…\n` } satisfies PyRunnerResult);
await py.loadPackage(toInstall);
for (const p of toInstall) _installedPkgs.add(p);
savePkgCacheToOPFS(); // S448: persiste anche i package espliciti in OPFS
}
}
py.globals.set("__stdout__", (text: string) => {
if (!ctx.cancelled) self.postMessage({ type: "stdout", id, text } satisfies PyRunnerResult);
});
py.globals.set("__stderr__", (text: string) => {
if (!ctx.cancelled) self.postMessage({ type: "stderr", id, text } satisfies PyRunnerResult);
});
const wrapped = `
import sys, io
class _Capture(io.StringIO):
def __init__(self, fn):
super().__init__()
self._fn = fn
def write(self, s):
if s:
self._fn(s)
return len(s)
def flush(self): pass
sys.stdout = _Capture(__stdout__)
sys.stderr = _Capture(__stderr__)
${code}
`;
await py.runPythonAsync(wrapped);
if (!timedOut && !ctx.cancelled) {
// S71: estrai variabili globali e inviale al main thread
await extractAndSendVars(py, id);
self.postMessage({ type: "done", id, exitCode: 0, time: Date.now() - start } satisfies PyRunnerResult);
}
} catch (err) {
if (!timedOut) {
const errMsg = err instanceof Error ? err.message : String(err);
self.postMessage({ type: "stderr", id, text: errMsg + "\n" } satisfies PyRunnerResult);
// S71: estrai anche in caso di errore parziale (alcune var potrebbero essere definite)
try {
if (pyodide) await extractAndSendVars(pyodide, id);
} catch { /* non-fatal */ }
self.postMessage({ type: "done", id, exitCode: 1, time: Date.now() - start } satisfies PyRunnerResult);
}
} finally {
clearTimeout(timer);
running.delete(id);
}
}
};