Run a pipeline's WASM jobs in the browser tab (refs andymccall/oxygenflow#80); runner builds shared with the swarm (refs andymccall/oxygenflow#84); 0.9.15 c26267b3aba6

// OxygenFlow's WASM tier in a browser tab (oxygenflow#80): this worker

// unpacks a repository's zip archive into memory and runs a job's WASI

// preview 1 modules over it, as `flow` runs them with wasmtime: the files at

// /src (and as the current directory), the job's arguments and environment,

// stdout and stderr streamed back line by line, and the exit code.

//

// Messages in: { archive: ArrayBuffer, exec: [module, ...args], env: {k: v} }.

// Messages out: { kind: "note"|"out"|"err", text }, then

// { kind: "done", code, ms } or { kind: "failed", text }.

"use strict";

const E = { SUCCESS: 0, ACCES: 2, BADF: 8, EXIST: 20, INVAL: 28, ISDIR: 31, NOENT: 44, NOSYS: 52, NOTDIR: 54, NOTEMPTY: 55, NOTSUP: 58 };

const FT = { UNKNOWN: 0, CHAR: 2, DIR: 3, FILE: 4 };

const O = { CREAT: 1, DIRECTORY: 2, EXCL: 4, TRUNC: 8 };

const FD_APPEND = 1;

const RIGHTS = 0x1fffffffn;

class Exit {

  constructor(code) { this.code = code; }

}

// --- The archive -----------------------------------------------------------

// A zip's files by path, the first directory (the archive's prefix) taken

// off. Read through the central directory, so sizes written after the data

// are known; stored and deflated entries; symbolic links are left out.

async function unzip(buf) {

  const b = new DataView(buf);

  const bytes = new Uint8Array(buf);

  let end = -1;

  for (let i = buf.byteLength - 22; i >= Math.max(0, buf.byteLength - 65557); i--) {

    if (b.getUint32(i, true) === 0x06054b50) { end = i; break; }

  }

  if (end < 0) throw new Error("the archive is not a zip file");

  const count = b.getUint16(end + 10, true);

  let at = b.getUint32(end + 16, true);

  if (at === 0xffffffff) throw new Error("the archive is too large for this page (zip64)");

  const files = new Map();

  const dec = new TextDecoder();

  for (let n = 0; n < count; n++) {

    if (b.getUint32(at, true) !== 0x02014b50) throw new Error("the archive's directory is damaged");

    const method = b.getUint16(at + 10, true);

    const csize = b.getUint32(at + 20, true);

    const nlen = b.getUint16(at + 28, true), xlen = b.getUint16(at + 30, true), clen = b.getUint16(at + 32, true);

    const mode = b.getUint32(at + 38, true) >>> 16;

    const local = b.getUint32(at + 42, true);

    const name = dec.decode(bytes.subarray(at + 46, at + 46 + nlen));

    at += 46 + nlen + xlen + clen;

    if (name.endsWith("/") || (mode & 0o170000) === 0o120000) continue;

    const data = local + 30 + b.getUint16(local + 26, true) + b.getUint16(local + 28, true);

    const raw = bytes.subarray(data, data + csize);

    let content;

    if (method === 0) content = raw.slice();

    else if (method === 8) content = new Uint8Array(await new Response(new Blob([raw]).stream().pipeThrough(new DecompressionStream("deflate-raw"))).arrayBuffer());

    else throw new Error(`${name}: compression method ${method} is not read here`);

    const slash = name.indexOf("/");

    files.set(slash < 0 ? name : name.slice(slash + 1), content);

  }

  return files;

}

// --- The file system ---------------------------------------------------------

class Fs {

  constructor(files) {

    this.files = files; // path -> Uint8Array

    this.dirs = new Set([""]);

    for (const p of files.keys()) this.addParents(p);

  }

  addParents(p) {

    const parts = p.split("/");

    for (let i = 1; i < parts.length; i++) this.dirs.add(parts.slice(0, i).join("/"));

  }

  // `rel` against directory `dir`, or null when it leaves the root.

  join(dir, rel) {

    const out = rel.startsWith("/") ? [] : dir ? dir.split("/") : [];

    for (const part of rel.split("/")) {

      if (part === "" || part === ".") continue;

      if (part === "..") { if (!out.length) return null; out.pop(); } else out.push(part);

    }

    return out.join("/");

  }

  kind(p) {

    if (this.files.has(p)) return FT.FILE;

    if (this.dirs.has(p)) return FT.DIR;

    return null;

  }

  list(dir) {

    const prefix = dir ? dir + "/" : "";

    const names = new Set();

    for (const p of [...this.files.keys(), ...this.dirs]) {

      if (p !== dir && p.startsWith(prefix)) names.add(p.slice(prefix.length).split("/")[0]);

    }

    return [...names].sort();

  }

}

// The modules a job's first `exec` word names: one path, or a `*` pattern

// in its file name, as `flow` resolves them.

function modules(fs, program) {

  const rel = program.replace(/^\/src\//, "").replace(/^\.\//, "");

  if (!rel || rel.startsWith("/") || rel.split("/").includes("..")) throw new Error(`module ${program}: give a path inside the job's files`);

  const cut = rel.lastIndexOf("/");

  const dir = cut < 0 ? "" : rel.slice(0, cut), last = rel.slice(cut + 1);

  if (dir.includes("*")) throw new Error(`module ${program}: only the file name may hold *`);

  if (!last.includes("*")) {

    if (!fs.files.has(rel)) throw new Error(`module ${rel} is not among the repository's files (a build makes it?)`);

    return [rel];

  }

  const re = new RegExp("^" + last.split("*").map((s) => s.replace(/[.+?^${}()|[\]\\]/g, "\\$&")).join(".*") + "$");

  const found = fs.list(dir).filter((n) => re.test(n) && fs.files.has(dir ? `${dir}/${n}` : n)).map((n) => (dir ? `${dir}/${n}` : n));

  if (!found.length) throw new Error(`no module matches ${program}`);

  return found;

}

// --- WASI preview 1 ------------------------------------------------------------

function wasi(fs, argv, env, say) {

  let memory;

  const view = () => new DataView(memory.buffer);

  const u8 = () => new Uint8Array(memory.buffer);

  const enc = new TextEncoder(), dec = new TextDecoder();

  const str = (ptr, len) => dec.decode(u8().subarray(ptr, ptr + len));

  // Open files and directories: 0-2 the streams, 3 /src, 4 ".".

  const fds = new Map([

    [0, { stream: "in" }],

    [1, { stream: "out", partial: "" }],

    [2, { stream: "err", partial: "" }],

    [3, { dir: "", preopen: "/src" }],

    [4, { dir: "", preopen: "." }],

  ]);

  let next = 5;

  const emit = (fd, text) => {

    const f = fds.get(fd);

    f.partial += text;

    let nl;

    while ((nl = f.partial.indexOf("\n")) >= 0) {

      say(f.stream, f.partial.slice(0, nl));

      f.partial = f.partial.slice(nl + 1);

    }

  };

  const flush = () => { for (const fd of [1, 2]) { const f = fds.get(fd); if (f.partial) { say(f.stream, f.partial); f.partial = ""; } } };

  const filestat = (ptr, kind, size) => {

    const v = view();

    for (let i = 0; i < 64; i += 8) v.setBigUint64(ptr + i, 0n, true);

    v.setUint8(ptr + 16, kind);

    v.setBigUint64(ptr + 24, 1n, true);

    v.setBigUint64(ptr + 32, BigInt(size), true);

  };

  const iovs = (ptr, n) => {

    const v = view(), out = [];

    for (let i = 0; i < n; i++) out.push([v.getUint32(ptr + i * 8, true), v.getUint32(ptr + i * 8 + 4, true)]);

    return out;

  };

  const strings = (list) => list.map((s) => enc.encode(s + "\0"));

  const sizes = (list, countPtr, sizePtr) => {

    const v = view(), all = strings(list);

    v.setUint32(countPtr, all.length, true);

    v.setUint32(sizePtr, all.reduce((n, s) => n + s.length, 0), true);

    return E.SUCCESS;

  };

  const fill = (list, ptrs, buf) => {

    const v = view(), m = u8();

    for (const s of strings(list)) { v.setUint32(ptrs, buf, true); ptrs += 4; m.set(s, buf); buf += s.length; }

    return E.SUCCESS;

  };

  const envList = Object.entries(env).map(([k, v]) => `${k}=${v}`);

  const imports = {

    args_sizes_get: (c, s) => sizes(argv, c, s),

    args_get: (p, b) => fill(argv, p, b),

    environ_sizes_get: (c, s) => sizes(envList, c, s),

    environ_get: (p, b) => fill(envList, p, b),

    clock_time_get: (_id, _precision, out) => { view().setBigUint64(out, BigInt(Math.round(performance.timeOrigin + performance.now())) * 1000000n, true); return E.SUCCESS; },

    clock_res_get: (_id, out) => { view().setBigUint64(out, 1000n, true); return E.SUCCESS; },

    random_get: (ptr, len) => { crypto.getRandomValues(u8().subarray(ptr, ptr + len)); return E.SUCCESS; },

    sched_yield: () => E.SUCCESS,

    proc_exit: (code) => { throw new Exit(code); },

    fd_prestat_get: (fd, ptr) => {

      const f = fds.get(fd);

      if (!f || !f.preopen) return E.BADF;

      const v = view();

      v.setUint8(ptr, 0);

      v.setUint32(ptr + 4, enc.encode(f.preopen).length, true);

      return E.SUCCESS;

    },

    fd_prestat_dir_name: (fd, ptr, len) => {

      const f = fds.get(fd);

      if (!f || !f.preopen) return E.BADF;

      u8().set(enc.encode(f.preopen).subarray(0, len), ptr);

      return E.SUCCESS;

    },

    fd_fdstat_get: (fd, ptr) => {

      const f = fds.get(fd);

      if (!f) return E.BADF;

      const v = view();

      v.setUint8(ptr, f.stream ? FT.CHAR : f.dir !== undefined ? FT.DIR : FT.FILE);

      v.setUint16(ptr + 2, f.append ? FD_APPEND : 0, true);

      v.setBigUint64(ptr + 8, RIGHTS, true);

      v.setBigUint64(ptr + 16, RIGHTS, true);

      return E.SUCCESS;

    },

    fd_fdstat_set_flags: () => E.SUCCESS,

    fd_filestat_get: (fd, ptr) => {

      const f = fds.get(fd);

      if (!f) return E.BADF;

      if (f.stream) filestat(ptr, FT.CHAR, 0);

      else if (f.dir !== undefined) filestat(ptr, FT.DIR, 0);

      else filestat(ptr, FT.FILE, fs.files.get(f.path).length);

      return E.SUCCESS;

    },

    path_filestat_get: (fd, _flags, p, len, ptr) => {

      const d = fds.get(fd);

      if (!d || d.dir === undefined) return E.BADF;

      const path = fs.join(d.dir, str(p, len));

      if (path === null) return E.ACCES;

      const kind = fs.kind(path);

      if (kind === null) return E.NOENT;

      filestat(ptr, kind, kind === FT.FILE ? fs.files.get(path).length : 0);

      return E.SUCCESS;

    },

    path_open: (fd, _dirflags, p, len, oflags, _rb, _ri, fdflags, out) => {

      const d = fds.get(fd);

      if (!d || d.dir === undefined) return E.BADF;

      const path = fs.join(d.dir, str(p, len));

      if (path === null) return E.ACCES;

      let kind = fs.kind(path);

      if (oflags & O.DIRECTORY) {

        if (kind === null) return E.NOENT;

        if (kind !== FT.DIR) return E.NOTDIR;

      }

      if (kind === null) {

        if (!(oflags & O.CREAT)) return E.NOENT;

        const parent = path.includes("/") ? path.slice(0, path.lastIndexOf("/")) : "";

        if (fs.kind(parent) !== FT.DIR) return E.NOENT;

        fs.files.set(path, new Uint8Array(0));

        kind = FT.FILE;

      } else if (oflags & O.EXCL) return E.EXIST;

      if (kind === FT.DIR) fds.set(next, { dir: path, entries: null });

      else {

        if (oflags & O.TRUNC) fs.files.set(path, new Uint8Array(0));

        fds.set(next, { path, pos: 0, append: !!(fdflags & FD_APPEND) });

      }

      view().setUint32(out, next, true);

      next += 1;

      return E.SUCCESS;

    },

    path_create_directory: (fd, p, len) => {

      const d = fds.get(fd);

      if (!d || d.dir === undefined) return E.BADF;

      const path = fs.join(d.dir, str(p, len));

      if (path === null) return E.ACCES;

      if (fs.kind(path) !== null) return E.EXIST;

      fs.dirs.add(path);

      return E.SUCCESS;

    },

    path_unlink_file: (fd, p, len) => {

      const d = fds.get(fd);

      if (!d || d.dir === undefined) return E.BADF;

      const path = fs.join(d.dir, str(p, len));

      if (path === null || !fs.files.has(path)) return E.NOENT;

      fs.files.delete(path);

      return E.SUCCESS;

    },

    fd_close: (fd) => {

      if (!fds.has(fd) || fd < 3) return fds.has(fd) ? E.SUCCESS : E.BADF;

      fds.delete(fd);

      return E.SUCCESS;

    },

    fd_read: (fd, ptr, n, out) => {

      const f = fds.get(fd);

      if (!f) return E.BADF;

      if (f.dir !== undefined) return E.ISDIR;

      let total = 0;

      if (!f.stream) {

        const data = fs.files.get(f.path), m = u8();

        for (const [buf, len] of iovs(ptr, n)) {

          const k = Math.min(len, data.length - f.pos);

          if (k <= 0) break;

          m.set(data.subarray(f.pos, f.pos + k), buf);

          f.pos += k;

          total += k;

        }

      }

      view().setUint32(out, total, true);

      return E.SUCCESS;

    },

    fd_write: (fd, ptr, n, out) => {

      const f = fds.get(fd);

      if (!f) return E.BADF;

      if (f.dir !== undefined) return E.ISDIR;

      let total = 0;

      const m = u8();

      for (const [buf, len] of iovs(ptr, n)) {

        const chunk = m.slice(buf, buf + len);

        if (f.stream === "out" || f.stream === "err") emit(fd, dec.decode(chunk, { stream: true }));

        else if (!f.stream) {

          const old = fs.files.get(f.path);

          const at = f.append ? old.length : f.pos;

          const grown = new Uint8Array(Math.max(old.length, at + len));

          grown.set(old);

          grown.set(chunk, at);

          fs.files.set(f.path, grown);

          f.pos = at + len;

        }

        total += len;

      }

      view().setUint32(out, total, true);

      return E.SUCCESS;

    },

    fd_seek: (fd, offset, whence, out) => {

      const f = fds.get(fd);

      if (!f || f.stream || f.dir !== undefined) return E.BADF;

      const size = fs.files.get(f.path).length;

      const base = whence === 0 ? 0 : whence === 1 ? f.pos : size;

      const pos = base + Number(offset);

      if (pos < 0) return E.INVAL;

      f.pos = pos;

      view().setBigUint64(out, BigInt(pos), true);

      return E.SUCCESS;

    },

    fd_tell: (fd, out) => {

      const f = fds.get(fd);

      if (!f || f.stream || f.dir !== undefined) return E.BADF;

      view().setBigUint64(out, BigInt(f.pos), true);

      return E.SUCCESS;

    },

    fd_readdir: (fd, buf, len, cookie, out) => {

      const d = fds.get(fd);

      if (!d || d.dir === undefined) return E.BADF;

      if (!d.entries) d.entries = [".", "..", ...fs.list(d.dir)];

      const v = view(), m = u8();

      let used = 0;

      for (let i = Number(cookie); i < d.entries.length; i++) {

        const name = d.entries[i];

        const bytes = enc.encode(name);

        const child = name === "." || name === ".." ? d.dir : fs.join(d.dir, name);

        const head = new DataView(new ArrayBuffer(24));

        head.setBigUint64(0, BigInt(i + 1), true);

        head.setBigUint64(8, BigInt(i + 1), true);

        head.setUint32(16, bytes.length, true);

        head.setUint8(20, name === "." || name === ".." ? FT.DIR : fs.kind(child) ?? FT.UNKNOWN);

        const entry = new Uint8Array(24 + bytes.length);

        entry.set(new Uint8Array(head.buffer));

        entry.set(bytes, 24);

        const room = len - used;

        m.set(entry.subarray(0, Math.min(room, entry.length)), buf + used);

        used += Math.min(room, entry.length);

        if (used >= len) break;

      }

      v.setUint32(out, used, true);

      return E.SUCCESS;

    },

  };

  return { imports, flush, bind: (mem) => { memory = mem; } };

}

// --- A job -------------------------------------------------------------------------

async function run({ archive, exec, env }) {

  const say = (kind, text) => postMessage({ kind, text });

  const started = performance.now();

  const fs = new Fs(await unzip(archive));

  say("note", `${fs.files.size} files unpacked in ${Math.round(performance.now() - started)} ms`);

  const list = modules(fs, exec[0]);

  let first = 0;

  for (const path of list) {

    say("note", `running ${path}`);

    const name = path.slice(path.lastIndexOf("/") + 1);

    const w = wasi(fs, [name, ...exec.slice(1)], { ...env, PWD: "/src" }, say);

    const module = await WebAssembly.compile(fs.files.get(path));

    const imports = { wasi_snapshot_preview1: {} };

    // What the shim lacks answers ENOSYS rather than refusing the module.

    for (const imp of WebAssembly.Module.imports(module)) {

      if (imp.kind !== "function") continue;

      imports[imp.module] ??= {};

      if (imp.module === "wasi_snapshot_preview1" && w.imports[imp.name]) imports[imp.module][imp.name] = w.imports[imp.name];

      else imports[imp.module][imp.name] ??= () => E.NOSYS;

    }

    const instance = await WebAssembly.instantiate(module, imports);

    w.bind(instance.exports.memory);

    let code = 0;

    try {

      instance.exports._start();

    } catch (e) {

      if (e instanceof Exit) code = e.code;

      else {

        w.flush();

        say("err", `${path} trapped: ${e.message}`);

        code = 134;

      }

    }

    w.flush();

    if (code !== 0 && first === 0) first = code;

  }

  postMessage({ kind: "done", code: first, ms: Math.round(performance.now() - started) });

}

onmessage = (m) => {

  run(m.data).catch((e) => postMessage({ kind: "failed", text: e.message }));

};

// "Run in this tab" (oxygenflow#80): each WASM job of the pipeline in a

// worker of its own (wasi-worker.js), over the bookmark's files fetched once

// as a zip. A job shows its output as it comes, its exit code and its time.

"use strict";

(() => {

  const root = document.getElementById("tab-runner");

  if (!root) return;

  const jobs = JSON.parse(root.dataset.jobs);

  const status = document.getElementById("tab-status");

  const list = document.getElementById("tab-jobs");

  let archive = null;

  const fetchArchive = () => {

    archive ??= fetch(root.dataset.archive, { credentials: "same-origin" }).then((r) => {

      if (!r.ok) throw new Error(`the files did not come (HTTP ${r.status})`);

      return r.arrayBuffer();

    });

    return archive;

  };

  const el = (tag, props, ...kids) => {

    const e = Object.assign(document.createElement(tag), props);

    e.append(...kids);

    return e;

  };

  const rows = jobs.map((job) => {

    const state = el("span", { className: "small mute", textContent: "not run" });

    const out = el("pre", { className: "snippet", hidden: true });

    const run = el("button", { type: "button", textContent: "Run" });

    const stop = el("button", { type: "button", textContent: "Stop", hidden: true });

    const box = el(

      "div",

      { className: "tab-job" },

      el("h3", {}, job.name, " ", state),

      el("p", { className: "small mute" }, el("code", { textContent: job.exec.join(" ") })),

      el("p", {}, run, " ", stop),

      out,

    );

    list.append(box);

    let worker = null;

    const line = (cls, text) => {

      out.hidden = false;

      out.append(el("span", { className: cls, textContent: text + "\n" }));

      out.scrollTop = out.scrollHeight;

    };

    const go = () =>

      new Promise((resolve) => {

        out.replaceChildren();

        state.textContent = "fetching the files…";

        state.className = "small mute";

        run.disabled = true;

        stop.hidden = false;

        const finish = (text, cls, ok) => {

          state.textContent = text;

          state.className = "small " + cls;

          run.disabled = false;

          stop.hidden = true;

          worker?.terminate();

          worker = null;

          resolve(ok);

        };

        stop.onclick = () => finish("stopped", "bad", false);

        fetchArchive()

          .then((buf) => {

            state.textContent = "running…";

            worker = new Worker(root.dataset.worker);

            worker.onmessage = ({ data: m }) => {

              if (m.kind === "done") finish(m.code === 0 ? `passed in ${m.ms} ms` : `failed, exit ${m.code}, in ${m.ms} ms`, m.code === 0 ? "ok" : "bad", m.code === 0);

              else if (m.kind === "failed") finish(`could not run: ${m.text}`, "bad", false);

              else line(m.kind === "out" ? "" : m.kind === "err" ? "bad" : "mute", m.text);

            };

            worker.onerror = (e) => finish(`the worker stopped: ${e.message}`, "bad", false);

            worker.postMessage({ archive: buf.slice(0), exec: job.exec, env: job.env });

          })

          .catch((e) => finish(e.message, "bad", false));

      });

    run.onclick = () => go();

    return go;

  });

  document.getElementById("tab-run-all").onclick = async (e) => {

    e.target.disabled = true;

    let passed = 0;

    for (const go of rows) if (await go()) passed += 1;

    status.textContent = `${passed} of ${rows.length} passed`;

    e.target.disabled = false;

  };

})();

{% extends "base.html" %}

{% import "repo_tabs.html" as m %}

{% block title %}Run in this tab · {{ repo.owner }}/{{ repo.name }}{% endblock %}

{% block bodyattrs %} data-repo="{{ repo.url }}"{% endblock %}

{% block tabs %}{% call m::tabs(repo, "flow") %}{% endblock %}

{% block content %}

<div class="head"><h1>Run in this tab <small><a href="{{ repo.url }}/flow">Flow</a> · WASM jobs, in your browser</small></h1>

<form class="inline" method="get" action="{{ repo.url }}/flow/tab"><label for="tab-bookmark" class="small mute">Jobs of</label> <select id="tab-bookmark" name="bookmark">{% for b in bookmarks %}<option value="{{ b }}"{% if *b == bookmark %} selected{% endif %}>{{ b }}</option>{% endfor %}</select> <button type="submit">Show</button></form>

</div>

<p class="small mute">The pipeline's jobs in a <code>runtime "wasm"</code> sandbox run here as a runner's WASM tier runs them: this tab fetches {{ bookmark }}'s files, and a worker runs each job's modules with them at <code>/src</code>, its arguments and environment, with no network and nothing of your computer. Nothing is reported to the hub: the tab shows the tier works in a browser, before a tab can take jobs as a runner (OxygenFlow #80).</p>

{% if let Some(p) = problem %}

<section class="panel"><p>{{ p }}</p></section>

{% else %}

<section class="panel" id="tab-runner" data-jobs="{{ jobs_json }}" data-archive="{{ archive }}" data-worker="/static/wasi-worker.js?v={{ shell.assets() }}">

  <p><button type="button" class="primary-btn" id="tab-run-all">Run all {{ jobs }}</button> <span class="small mute" id="tab-status"></span></p>

  <div id="tab-jobs"></div>

</section>

<script src="/static/wasm-tab.js?v={{ shell.assets() }}" defer></script>

{% endif %}

{% endblock %}

## 0.9.15 (2026-10-28)

- **Run a pipeline's WASM jobs in your browser (oxygenflow#80, first part).** Flow › *run in this tab* (`/OWNER/NAME/flow/tab`) lists the jobs of a bookmark's pipeline whose sandbox is `runtime "wasm"` and runs them in this tab, as a runner's WASM tier does: a worker fetches the bookmark's files as a zip once, unpacks them in memory and runs each job's modules with a WASI preview 1 shim, the files at `/src` and as the current directory, the job's arguments and environment, its output shown as it comes, its exit code and time; Run all, Run and Stop per job. Nothing is claimed or reported: it shows the tier in a browser before a tab can take jobs as a runner.

- **Runner builds through the swarm (oxygenflow#84).** The first time the hub offers a runner build, it cuts the release asset into its image chunks, as an image named by the build's BLAKE3, so a runner in a swarm (OxygenFlow 0.5.2) fetches it from its site's peers and the hub together and the hub serves it about once per site; Admin › Runners › Images lists it as "runner build flow VERSION".

version = "0.9.15"

version = "0.9.15"

version = "0.9.15"

version = "0.9.15"

    if let Some(o) = &offer {

        crate::runner_updates::share(&app, o);

    }

/// Share an offered build with the swarm (oxygenflow#84): cut the release

/// asset into the hub's image chunks and keep its manifest, named, as an

/// image is, by the build's BLAKE3, so a runner in a swarm fetches it from

/// its site's peers and the hub together, as it fetches images, and the

/// hub serves it about once per site. Done once, in the background, the

/// first time the build is offered; a runner asking before it is done

/// fetches the build over HTTP.

pub fn share(app: &crate::Shared, offer: &UpdateOffer) {

    static SHARING: std::sync::Mutex<BTreeSet<String>> = std::sync::Mutex::new(BTreeSet::new());

    let store = app.images();

    let hash = offer.blake3.clone();

    if !jet::images::is_hash(&hash) || store.manifest(&hash).is_some() {

        return;

    }

    if !SHARING.lock().unwrap_or_else(|p| p.into_inner()).insert(hash.clone()) {

        return;

    }

    let (app, offer) = (app.clone(), offer.clone());

    tokio::spawn(async move {

        let path = crate::web::releases::blob_path(&app, &hash);

        let arch = offer.asset.rsplit('-').next().map(|a| a.trim_end_matches(".exe").to_string());

        let want = hash.clone();

        let made = tokio::task::spawn_blocking(move || -> Result<jet::images::Manifest, String> {

            let m = jet::images::scan_file(&path, arch, |c, bytes| Ok(store.put_chunk(c.hash(), bytes)?))

                .map_err(|e| format!("reading {}: {e}", path.display()))?;

            if m.hash != want {

                return Err(format!("the asset hashes to {}, not {}", &m.hash[..12], &want[..12]));

            }

            store.verify_and_keep(&m)?;

            Ok(m)

        })

        .await

        .unwrap_or_else(|e| Err(e.to_string()));

        match made {

            Ok(m) => match crate::images::record(&app.db, &m, &format!("runner build flow {}", offer.version)).await {

                Ok(()) => tracing::info!(

                    "runner updates: flow {} ({}) shared with the swarm: {} chunks",

                    offer.version,

                    offer.asset,

                    m.chunks.len()

                ),

                Err(e) => tracing::warn!("runner updates: recording flow {} for the swarm: {e}", offer.version),

            },

            Err(e) => tracing::warn!("runner updates: sharing flow {} ({}) with the swarm: {e}", offer.version, offer.asset),

        }

        SHARING.lock().unwrap_or_else(|p| p.into_inner()).remove(&hash);

    });

}

/// A WASM job as the tab runner runs it (oxygenflow#80).

#[derive(serde::Serialize)]

pub struct TabJob {

    name: String,

    exec: Vec<String>,

    env: std::collections::BTreeMap<String, String>,

    memory_mib: u32,

}

#[derive(Template)]

#[template(path = "flow_tab.html")]

pub struct TabPage {

    shell: Shell,

    repo: RepoShell,

    bookmark: String,

    bookmarks: Vec<String>,

    /// The pipeline's WASM jobs, as JSON for the page's script.

    jobs_json: String,

    jobs: usize,

    /// Why there is nothing to run: no pipeline, one that does not read,

    /// or no WASM job in it.

    problem: Option<String>,

    archive: String,

}

#[derive(Deserialize)]

pub struct TabQuery {

    bookmark: Option<String>,

}

/// The WASM jobs of a pipeline (`runtime "wasm"`), each with the

/// environment its sandbox and itself give it.

pub fn wasm_jobs(p: &flow_core::Pipeline) -> Vec<TabJob> {

    p.jobs

        .values()

        .filter_map(|j| {

            let sb = p.sandboxes.get(&j.sandbox)?;

            (sb.runtime == flow_core::Runtime::Wasm && !j.exec.is_empty()).then(|| {

                let mut env = sb.env.clone();

                env.extend(j.env.clone());

                TabJob { name: j.name.clone(), exec: j.exec.clone(), env, memory_mib: sb.memory_mib }

            })

        })

        .collect()

}

/// `GET /{owner}/{name}/flow/tab`: run the pipeline's WASM jobs in this

/// browser tab (oxygenflow#80, first part): a worker unpacks the

/// bookmark's files and runs each job's modules with a WASI shim, as a

/// runner's WASM tier does. Nothing is claimed or reported: it shows the

/// tier works in a tab, before the tab becomes a runner.

pub async fn tab(

    State(app): State<Shared>,

    viewer: Viewer,

    Path((owner, name)): Path<(String, String)>,

    Query(q): Query<TabQuery>,

) -> Result<Page<TabPage>, Error> {

    let shell = Shell::new(&app, &format!("/{owner}/{name}/flow/tab"), &viewer);

    let (repo, row, role) = lookup(&app, &shell, &viewer, &owner, &name).await?;

    let bookmarks = bookmark_names(&app, &row).await;

    let bookmark = q.bookmark.filter(|b| bookmarks.contains(b)).unwrap_or_else(|| row.default_bookmark.clone());

    let b = bookmark.clone();

    let text = app

        .repos

        .read(&repo, move |r, cache| {

            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };

            let heads = r.store().get_state(&st)?.heads;

            let loaded = cache.load(r, &heads)?;

            Ok(match loaded.snapshot.files.get(flow_core::PIPELINE_PATH).map(|f| &f.content) {

                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),

                _ => None,

            })

        })

        .await

        .unwrap_or(None);

    let (jobs, problem) = match text.map(|t| flow_core::Pipeline::parse(&t)) {

        None => (vec![], Some(format!("{bookmark} has no {}.", flow_core::PIPELINE_PATH))),

        Some(Err(e)) => (vec![], Some(format!("The pipeline on {bookmark} does not read: {e}"))),

        Some(Ok(p)) => {

            let jobs = wasm_jobs(&p);

            let problem =

                jobs.is_empty().then(|| format!("The pipeline on {bookmark} has no WASM jobs (a sandbox with runtime \"wasm\")."));

            (jobs, problem)

        }

    };

    Ok(Page(TabPage {

        shell,

        repo: RepoShell::new(&app, &row, "flow").with_role(role),

        archive: format!("/{owner}/{name}/archive?ref={}&format=zip", urlencoding(&bookmark)),

        bookmark,

        bookmarks,

        jobs: jobs.len(),

        jobs_json: serde_json::to_string(&jobs).unwrap_or_else(|_| "[]".into()),

        problem,

    }))

}

        .route("/{owner}/{name}/flow/tab", get(flow::tab))

    Asset { path: "wasi-worker.js", mime: "text/javascript; charset=utf-8", bytes: include_bytes!("../../../../assets/wasi-worker.js") },

    Asset { path: "wasm-tab.js", mime: "text/javascript; charset=utf-8", bytes: include_bytes!("../../../../assets/wasm-tab.js") },

    let build: Vec<u8> = (0..300_000u32).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();

    let build_hash = blake3::hash(&build).to_hex().to_string();

    let at = h.app.config.data.join("assets").join(&build_hash[..2]);

    std::fs::create_dir_all(&at).unwrap();

    std::fs::write(at.join(&build_hash), &build).unwrap();

    db::releases::add_asset(&h.app.db, rel.id, "flow-linux-x86_64", build.len() as i64, &build_hash, "application/octet-stream")

        .await

        .unwrap();

    // Offered, the build is shared with the swarm (oxygenflow#84): an image

    // named by its BLAKE3 whose chunks make it, for runners to fetch.

    let mut shared = None;

    for _ in 0..100 {

        shared = h.app.images().manifest(&build_hash);

        if shared.is_some() {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

    }

    let m = shared.expect("the build is shared with the swarm");

    assert_eq!((m.size, m.arch.as_deref()), (build.len() as u64, Some("x86_64")));

    let mut whole = vec![];

    for c in &m.chunks {

        whole.extend(h.app.images().chunk(c.hash()).unwrap());

    }

    assert_eq!(whole, build);

    let kept = oxygenhub::images::kept(&h.app.db).await.unwrap();

    assert!(kept.iter().any(|k| k.hash == build_hash && k.uploader == "runner build flow 0.2.0"));

    // oxygenflow#80: the page that runs it in a browser tab lists it, with

    // the zip of the bookmark's files its worker reads, and the worker.

    let (st, page) = call(&h.app, "GET", "/andy/wasm/flow/tab", None, None).await;

    assert_eq!(st, StatusCode::OK, "{page}");

    assert!(page.contains("Run all 1") && page.contains(""exec":["count.wasm"]"), "{page}");

    assert!(page.contains("data-archive=\"/andy/wasm/archive?ref=main&format=zip\""), "{page}");

    let (st, zip) = get_bytes(&h.app, "/andy/wasm/archive?ref=main&format=zip").await;

    assert!(st == StatusCode::OK && zip.starts_with(b"PK"), "{st}");

    assert_eq!(get_bytes(&h.app, "/static/wasi-worker.js").await.0, StatusCode::OK);

    let (_, runs) = call(&h.app, "GET", "/andy/wasm/flow", None, None).await;

    assert!(runs.contains("/andy/wasm/flow/tab"), "linked from Flow");

    let (_, none) = call(&h.app, "GET", "/andy/wasm/flow/tab?bookmark=nope", None, None).await;

    assert!(none.contains("Run all 1"), "an unknown bookmark shows the default's");

<div class="head"><h1>Flow <small>pipeline runs · <a href="{{ repo.url }}/deployments">deployments</a> · <a href="{{ repo.url }}/flow/tab" title="Run the pipeline's WASM jobs in your browser">run in this tab</a> · <a href="{{ repo.url }}/insights" target="_blank" rel="noopener" title="Pickup time, jobs, durations and runners, charted">charts ↗</a></small></h1>

/* Run in this tab (oxygenflow#80) */

.tab-job { border-top: 1px solid var(--line); padding: 8px 0; }

.tab-job:first-child { border-top: 0; }

.tab-job h3 { margin: 0; font-size: 15px; }

.tab-job .ok { color: var(--ok); }

.tab-job .bad { color: var(--bad); }

.tab-job pre { max-height: 24rem; overflow: auto; }