shifathosn
/
effectcraft
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #224 from storytold/fix/215-216-218-web-workers 

Web: replaced files reach reused workers, failed worker starts fail their job, failed storage writes retry (#215, #216, #218)
This commit is contained in:
BFlat 2026-10-08 05:33:59 +07:00 committed by GitHub
parent da6cb73142 1ca78821b9
commit e38a92e792
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 259 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
apps/effectcraft-web/js/host.js
View File

@ -278,8 +278,8 @@ export function audioStop() {
// ------------------------------------------------------------------ workers
// Idle workers are reused; each keeps the files it was sent (path → size) so a file goes to a
// worker once.
// Idle workers are reused; each keeps the files it was sent (path → version) so a file goes to a
// worker once, and again when it was replaced (even by one of the same size, #218).
const idle = [];
const running = new Map(); // job id → {worker, done}
// Job workers open a WebGPU device of their own (unless `?nogpuworkers`): their jobs' frames
@ -310,7 +310,8 @@ function spawn(base) {
return w;
}
/// Run a job in a worker. `files`: [[path, Uint8Array]] the job reads; `onMessage(type, json,
/// Run a job in a worker. `files`: [[path, Uint8Array, version]] the job reads (`version`
/// changes with every write of the file); `onMessage(type, json,
/// path, bytes)` receives "reply" (json), "file" (path, bytes: a rendered file to download),
/// "store" (path, bytes: a file to keep in browser storage) and "error" (json = message).
export function workerRun(id, json, files, base, onMessage) {
@ -339,14 +340,17 @@ export function workerRun(id, json, files, base, onMessage) {
onMessage("error", String(e.message || "worker error"), null, null);
w.worker.terminate();
};
for (const [path, bytes] of files) {
if (w.files.get(path) === bytes.length) continue;
w.files.set(path, bytes.length);
for (const [path, bytes, version] of files) {
const tag = version ?? bytes.length;
if (w.files.get(path) === tag) continue;
w.files.set(path, tag);
w.worker.postMessage({ type: "file", path, bytes });
}
w.worker.postMessage({ type: "job", json });
}, (e) => {
// The worker didn't start (#216): the job fails and the worker goes.
running.delete(id);
w.worker.terminate();
onMessage("error", String(e.message || e), null, null);
});
}

3
apps/effectcraft-web/src/api.rs
View File

@ -148,7 +148,8 @@ pub fn remove_stored(path: String) -> bool {
crate::files::STORE.remove_file(&path)
}
/// Resolves once every change is written to browser storage.
/// Resolves once every change is written to browser storage; rejects when a write failed (it is
/// tried again).
#[wasm_bindgen]
pub fn flush() -> js_sys::Promise {
crate::persist::flushed()

6
apps/effectcraft-web/src/files.rs
View File

@ -35,6 +35,12 @@ pub fn get(path: &str) -> Option<Arc<[u8]>> {
STORE.file(path)
}
/// A file's bytes and version (new with every write: workers resend a file whose version they
/// haven't seen).
pub fn get_versioned(path: &str) -> Option<(Arc<[u8]>, u64)> {
STORE.file_versioned(path)
}
/// `(path, size)` of every file in the table.
pub fn list() -> Vec<(String, usize)> {
STORE.files().into_iter().map(|(p, n, _, _)| (p, n)).collect()

18
apps/effectcraft-web/src/frames.rs
View File

@ -19,7 +19,7 @@
//! frame times. Without WebGPU, frames render on the worker's CPU.
use std::cell::RefCell;
use std::collections::HashSet;
use std::collections::HashMap;
use std::sync::Arc;
use effectcraft_engine::offload::footage_files;
@ -45,7 +45,8 @@ extern "C" {
struct Fw {
index: u32,
mirror: Mirror,
files: HashSet<String>,
/// The footage files it was sent, with their versions (a replaced file goes again).
files: HashMap<String, u64>,
busy: Option<(u64, RemoteDone)>,
alive: bool,
/// Its GPU adapter (after it reported), or why it has none.
@ -99,7 +100,7 @@ impl WebFrames {
let mut s = s.borrow_mut();
for _ in 0..n {
let index = frame_worker_start(&crate::page_url(""), on.as_ref().unchecked_ref(), gpu);
s.workers.push(Fw { index, mirror: Mirror::default(), files: HashSet::new(), busy: None, alive: true, gpu: None });
s.workers.push(Fw { index, mirror: Mirror::default(), files: HashMap::new(), busy: None, alive: true, gpu: None });
}
});
HANDLER.with(|h| *h.borrow_mut() = Some(on));
@ -116,14 +117,13 @@ impl WebFrames {
// Footage the worker hasn't got yet.
let w = &mut s.workers[k];
for p in footage_files(&job.project) {
if w.files.contains(&p) {
let Some((d, version)) = crate::files::get_versioned(&p) else { continue };
if w.files.get(&p) == Some(&version) {
continue;
}
if let Some(d) = crate::files::get(&p) {
let bytes = js_sys::Uint8Array::from(&d[..]);
frame_worker_post(w.index, &obj(&[("type", "file".into()), ("path", p.as_str().into()), ("bytes", bytes.into())]), &js_sys::Array::new());
w.files.insert(p);
}
let bytes = js_sys::Uint8Array::from(&d[..]);
frame_worker_post(w.index, &obj(&[("type", "file".into()), ("path", p.as_str().into()), ("bytes", bytes.into())]), &js_sys::Array::new());
w.files.insert(p, version);
}
let sync = w.mirror.sync(job.revision, &job.project);
let index = w.index;

73
apps/effectcraft-web/src/persist.rs
View File

@ -3,7 +3,8 @@
//!
//! - [`load`] (before the app starts) opens the storage and reads every entry into the mirror.
//! - Changes flush in the background ([`schedule_flush`], wired as the store's change hook):
//! one write per changed key, in order, with nothing written twice concurrently.
//! one write per changed key, in order, with nothing written twice concurrently. A write that
//! fails (quota exceeded…) is kept and tried again, after 2 s, then longer, up to a minute.
//! - The session snapshot ([`snapshot`]) records the open project and editor state every second
//! or so while it changes, so a reload (or a crash) comes back to where the user was
//! ([`restore`]).
@ -37,11 +38,18 @@ thread_local! {
static FLUSHING: Cell<bool> = const { Cell::new(false) };
static BACKEND: std::cell::RefCell<String> = const { std::cell::RefCell::new(String::new()) };
static LAST_ERROR: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
static FLUSH_WAITERS: std::cell::RefCell<Vec<js_sys::Function>> = const { std::cell::RefCell::new(Vec::new()) };
/// [`flushed`] promises waiting for the running flush: (resolve, reject).
static FLUSH_WAITERS: std::cell::RefCell<Vec<(js_sys::Function, js_sys::Function)>> = const { std::cell::RefCell::new(Vec::new()) };
/// Delay before retrying failed writes (ms; doubles while they keep failing).
static RETRY_MS: Cell<i32> = const { Cell::new(RETRY_FIRST_MS) };
static RETRY_PENDING: Cell<bool> = const { Cell::new(false) };
static SNAP: Cell<(u64, f64)> = const { Cell::new((u64::MAX, 0.0)) };
static LAST_META: std::cell::RefCell<String> = const { std::cell::RefCell::new(String::new()) };
}
const RETRY_FIRST_MS: i32 = 2_000;
const RETRY_MAX_MS: i32 = 60_000;
/// The storage in use: `opfs`, `indexeddb` or `memory`.
pub fn backend() -> String {
BACKEND.with(|b| b.borrow().clone())
@ -86,8 +94,11 @@ pub fn schedule_flush() {
return; // the running flush picks the new changes up
}
wasm_bindgen_futures::spawn_local(async {
// Keys whose write failed in this flush, and the last failure.
let mut failed: Vec<String> = vec![];
let mut error = None;
loop {
let pending = STORE.lock().take_pending();
let pending: Vec<_> = STORE.lock().take_pending().into_iter().filter(|p| !failed.contains(&p.key)).collect();
if pending.is_empty() {
break;
}
@ -99,24 +110,68 @@ pub fn schedule_flush() {
if let Err(e) = r {
let msg = format!("saving {} to browser storage failed: {}", p.key, js_err(e));
log::warn!("{msg}");
LAST_ERROR.with(|l| *l.borrow_mut() = Some(msg));
LAST_ERROR.with(|l| *l.borrow_mut() = Some(msg.clone()));
error = Some(msg);
failed.push(p.key);
}
}
}
// Failed writes stay pending (#215): flushed later, and the waiters hear of the failure
// instead of being told the changes are stored.
{
let mut m = STORE.lock();
for k in &failed {
m.retry(k);
}
}
FLUSHING.with(|f| f.set(false));
for w in FLUSH_WAITERS.with(|w| std::mem::take(&mut *w.borrow_mut())) {
let _ = w.call0(&JsValue::NULL);
let waiters = FLUSH_WAITERS.with(|w| std::mem::take(&mut *w.borrow_mut()));
match &error {
None => {
RETRY_MS.with(|r| r.set(RETRY_FIRST_MS));
for (resolve, _) in waiters {
let _ = resolve.call0(&JsValue::NULL);
}
}
Some(msg) => {
let e: JsValue = js_sys::Error::new(msg).into();
for (_, reject) in waiters {
let _ = reject.call1(&JsValue::NULL, &e);
}
retry_later();
}
}
});
}
/// A promise that resolves once every change so far is in the browser storage.
/// Flush again after the retry delay (once; the delay doubles up to a minute while writes keep
/// failing).
fn retry_later() {
if RETRY_PENDING.with(|r| r.replace(true)) {
return;
}
let ms = RETRY_MS.with(|r| {
let ms = r.get();
r.set(ms.saturating_mul(2).min(RETRY_MAX_MS));
ms
});
let again = Closure::once_into_js(|| {
RETRY_PENDING.with(|r| r.set(false));
schedule_flush();
});
if let Some(w) = web_sys::window() {
let _ = w.set_timeout_with_callback_and_timeout_and_arguments_0(again.unchecked_ref(), ms);
}
}
/// A promise that resolves once every change so far is in the browser storage, and rejects
/// when a write failed (the change stays pending and is tried again).
pub fn flushed() -> js_sys::Promise {
js_sys::Promise::new(&mut |resolve, _| {
js_sys::Promise::new(&mut |resolve, reject| {
if !FLUSHING.with(|f| f.get()) && !STORE.lock().has_pending() {
let _ = resolve.call0(&JsValue::NULL);
} else {
FLUSH_WAITERS.with(|w| w.borrow_mut().push(resolve));
FLUSH_WAITERS.with(|w| w.borrow_mut().push((resolve, reject)));
schedule_flush();
}
})

58
apps/effectcraft-web/src/store.rs
View File

@ -42,6 +42,9 @@ pub struct Entry {
pub modified: f64,
/// Written to the browser's storage (render outputs are not).
pub persist: bool,
/// Which write this is: new with every write of the key, so a file replaced by another of
/// the same size still reads as changed (workers are sent it again, #218).
pub version: u64,
}
/// A queued write for the browser storage: `data` = `None` deletes the key.
@ -58,17 +61,21 @@ pub struct Mirror {
entries: BTreeMap<String, Entry>,
/// Keys changed since the last [`Mirror::take_pending`], in first-change order.
dirty: Vec<String>,
/// The last [`Entry::version`] given.
version: u64,
}
impl Mirror {
/// An entry read from the browser's storage at startup (nothing to flush).
pub fn load(&mut self, key: &str, data: Arc<[u8]>, modified: f64) {
self.entries.insert(key.to_string(), Entry { data, modified, persist: true });
let version = self.next_version();
self.entries.insert(key.to_string(), Entry { data, modified, persist: true, version });
}
pub fn put(&mut self, key: &str, data: Arc<[u8]>, modified: f64, persist: bool) {
let was_persisted = self.entries.get(key).is_some_and(|e| e.persist);
self.entries.insert(key.to_string(), Entry { data, modified, persist });
let version = self.next_version();
self.entries.insert(key.to_string(), Entry { data, modified, persist, version });
if persist || was_persisted {
self.touch(key);
}
@ -86,6 +93,16 @@ impl Mirror {
}
}
fn next_version(&mut self) -> u64 {
self.version = self.version.wrapping_add(1);
self.version
}
/// A write of `key` that failed: the next flush writes the key's latest state again (#215).
pub fn retry(&mut self, key: &str) {
self.touch(key);
}
fn touch(&mut self, key: &str) {
if !self.dirty.iter().any(|k| k == key) {
self.dirty.push(key.to_string());
@ -172,6 +189,11 @@ impl Store {
self.lock().get(&file_key(path)).map(|e| e.data.clone())
}
/// The file under `path` with its [`Entry::version`].
pub fn file_versioned(&self, path: &str) -> Option<(Arc<[u8]>, u64)> {
self.lock().get(&file_key(path)).map(|e| (e.data.clone(), e.version))
}
pub fn remove_file(&self, path: &str) -> bool {
let r = self.lock().remove(&file_key(path));
self.changed();
@ -412,6 +434,38 @@ mod tests {
assert_eq!(m.list("files").len(), 1);
}
/// A file replaced by different bytes of the same size gets a new version, so workers are
/// sent it again (#218).
#[test]
fn versions_change_with_every_write() {
let s = store();
s.put_file("/collision.png", Arc::from(&b"red!"[..]), true);
let (_, a) = s.file_versioned("/collision.png").unwrap();
s.put_file("/collision.png", Arc::from(&b"blue"[..]), true);
let (data, b) = s.file_versioned("/collision.png").unwrap();
assert_ne!(a, b);
assert_eq!(&data[..], b"blue");
assert_eq!(s.file_versioned("/collision.png").map(|f| f.1), Some(b), "reading doesn't change it");
}
/// A write that failed is flushed again with the key's latest state (#215).
#[test]
fn failed_writes_retry() {
let s = store();
s.put_file("/collision.png", Arc::from(&b"blue"[..]), true);
let mut m = s.lock();
let p = m.take_pending();
assert_eq!(p.len(), 1);
assert!(!m.has_pending());
m.retry(&p[0].key);
assert_eq!(m.take_pending(), p, "the same write again");
// A newer change wins over the failed one.
m.retry("files/collision.png");
m.put("files/collision.png", Arc::from(&b"gold"[..]), 2.0, true);
let again = m.take_pending();
assert_eq!((again.len(), again[0].data.as_deref()), (1, Some(&b"gold"[..])));
}
#[test]
fn keys_and_paths_round_trip() {
assert_eq!(file_key("/a b/c.mov"), "files/a b/c.mov");

4
apps/effectcraft-web/src/worker.rs
View File

@ -60,8 +60,8 @@ impl Offload for WorkerOffload {
let id = req.id;
let files = js_sys::Array::new();
for p in &req.files {
if let Some(d) = crate::files::get(p) {
files.push(&js_sys::Array::of2(&JsValue::from_str(p), &js_sys::Uint8Array::from(&d[..])));
if let Some((d, version)) = crate::files::get_versioned(p) {
files.push(&js_sys::Array::of3(&JsValue::from_str(p), &js_sys::Uint8Array::from(&d[..]), &JsValue::from_f64(version as f64)));
}
}
let json = serde_json::to_string(&req).map_err(|e| e.to_string())?;

101
apps/effectcraft-web/tests/workers.mjs Normal file
View File

@ -0,0 +1,101 @@
// Job-worker plumbing of the web app (js/host.js, web/worker.js) under Node with a fake Worker,
// no browser or Wasm build needed:
//
// node apps/effectcraft-web/tests/workers.mjs
//
// - A reused worker is sent a file again when it was replaced, even by one of the same size
// (#218), and not when it is unchanged.
// - A worker whose start fails fails its job and is terminated, instead of the job waiting for
// "ready" forever (#216): host.js on the worker's error, worker.js re-raising the failed start.
// Exits non-zero on failure.
import assert from "node:assert/strict";
import { copyFileSync, mkdtempSync, writeFileSync } from "node:fs";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
const here = dirname(fileURLToPath(import.meta.url));
const tick = () => new Promise((r) => setTimeout(r, 0));
globalThis.self = globalThis;
// A Worker that starts (or fails to) on "init" and finishes each job at once.
let failStart = false;
const workers = [];
class FakeWorker {
constructor() {
this.posted = [];
this.terminated = false;
workers.push(this);
}
postMessage(m) {
this.posted.push(m);
setTimeout(() => {
if (m.type === "init") {
if (failStart) this.onerror?.({ message: "worker failed to start" });
else this.onmessage?.({ data: { type: "ready" } });
} else if (m.type === "job") {
this.onmessage?.({ data: { type: "reply", json: "{}", done: true } });
}
});
}
terminate() {
this.terminated = true;
}
}
globalThis.Worker = FakeWorker;
const host = await import(pathToFileURL(join(here, "../js/host.js")));
const run = (id, files) =>
new Promise((resolve) => host.workerRun(id, "{}", files, "http://localhost/", (kind, json) => kind !== "file" && resolve({ kind, json })));
const sent = (w) => w.posted.filter((m) => m.type === "file").map((m) => m.bytes[0]);
// #218: equal-length replacement.
const red = new Uint8Array(512).fill(1);
const blue = new Uint8Array(512).fill(2);
assert.equal((await run(1, [["/collision.png", red, 1]])).kind, "reply");
assert.equal((await run(2, [["/collision.png", blue, 2]])).kind, "reply");
assert.equal((await run(3, [["/collision.png", blue, 2]])).kind, "reply");
assert.equal(workers.length, 1, "the worker is reused");
assert.deepEqual(sent(workers[0]), [1, 2], "the replacement is sent once, the unchanged file not again");
// #216: a worker that fails to start.
failStart = true;
const slow = run(4, []); // the idle worker takes this one
const failed = await run(5, []);
await slow;
assert.equal(failed.kind, "error");
assert.ok(workers[1].terminated, "the failed worker is terminated");
assert.equal(host.workerStats().running, 0);
// #216, worker side: a rejected start is re-raised where the page's `onerror` sees it.
const dir = mkdtempSync(join(tmpdir(), "ec-worker-"));
copyFileSync(join(here, "../web/worker.js"), join(dir, "worker.js"));
writeFileSync(
join(dir, "effectcraft_web.js"),
`export default async function init() {}
export function workerInit() {}
export async function workerJobInit() { throw new Error("no device"); }
export async function workerFrameInit() {}`,
);
const posted = [];
globalThis.postMessage = (m) => posted.push(m);
// A browser's Worker `error` event only sees thrown errors, not unhandled rejections (Node would
// turn one into an uncaught exception unless it is listened to).
const unhandled = [];
process.on("unhandledRejection", (e) => unhandled.push(e));
const raised = new Promise((resolve) => {
process.once("uncaughtException", resolve);
setTimeout(() => resolve(new Error("nothing was raised")), 1000);
});
await import(pathToFileURL(join(dir, "worker.js")));
self.onmessage({ data: { type: "init", base: "http://localhost/" } });
const err = await raised;
await tick();
// (So a failed assertion below fails the run.)
process.removeAllListeners("unhandledRejection");
process.removeAllListeners("uncaughtException");
assert.deepEqual(unhandled, [], "the failed start is handled, not left as an unhandled rejection");
assert.equal(err.message, "no device");
assert.ok(!posted.some((m) => m.type === "ready"), "no ready after a failed start");
console.log("workers: ok");

3
apps/effectcraft-web/web/worker.js
View File

@ -30,6 +30,9 @@ self.onmessage = (e) => {
else await ec.workerJobInit(!!m.gpu);
self.postMessage({ type: "ready" });
})();
// A failed start is re-raised outside the promise so the page's `onerror` sees it: its job
// fails instead of waiting for "ready" forever (#216).
ready.catch((err) => setTimeout(() => { throw err; }));
return;
}
// A failure (a panic) is re-raised outside the queue so the page's `onerror` sees it, as

7
docs/web.md
View File

@ -384,7 +384,7 @@ with the method's `result`, errors reject with the message.
| `effectcraft.saveToBrowser(path?)` | save the project to browser storage without downloading it (default: its path, or `/<name>.ecproj`); resolves with `{path, bytes}` |
| `effectcraft.listStored()` | `{backend, usage, quota, persisted, pending, files: [{path, size, modified}], config: [name]}` |
| `effectcraft.removeStored(path)` | delete a stored file |
| `effectcraft.flush()` | resolves once every change is written to browser storage |
| `effectcraft.flush()` | resolves once every change is written to browser storage; rejects when a write failed (quota exceeded…: the change stays pending and is retried) |
| `effectcraft.info()` | graphics backend, `gpu`, `storage`, `audio` (`{state, sampleRate, backend, posted, played, underruns}`), `workers`, `frameWorkers`, `diskCache`, `restored`, `webgpu`, `serviceWorker`, version, `crossOriginIsolated`, load timings |
| `effectcraft.workerFrameCheck(params)` | a frame rendered in a frame worker (its GPU by default) against the page's CPU render: `{width, height, maxDiff, meanDiff, over4, workerMs}` |
@ -413,6 +413,11 @@ cargo xtask web --serve 8765 &
node apps/effectcraft-web/tests/smoke.mjs --url http://127.0.0.1:8765/ --out target/web/smoke
```
`apps/effectcraft-web/tests/workers.mjs` checks the job-worker plumbing (`js/host.js`,
`web/worker.js`) under Node with a fake `Worker`, without a build: a replaced file is sent to a
reused worker again even at the same size, and a worker that fails to start fails its job and is
terminated (`node apps/effectcraft-web/tests/workers.mjs`).
It also checks the M13.10 paths: viewer frames rendered in frame workers while scrubbing with the
CPU renderer (project synced as diffs, event-loop gaps under 400 ms), a `wait: true` render that
replies when done while the page keeps running, and the Media Browser over browser storage; and