Compare commits

..

5 Commits

Author SHA1 Message Date
Maofeng b304b038a4 docs: describe resilient development and telemetry
Document queued rebuilds and artifact-preserving deployment retries for wasmeld dev.

Describe asynchronous invocation telemetry, its durability boundary, graceful flushing, and the default Wasmtime compilation cache location.
2026-07-30 07:18:36 +08:00
Maofeng 0f0effb3c3 refactor(console-ui): split dashboard responsibilities
Reduce the root TanStack route to orchestration and shell state. Move backend-to-view mapping and binary invocation formatting into console-model, presentation views into components/views, and modal workflows into components/dialogs.

Update the rendered artifact test to assert behavior at the new module boundaries without changing the generated UI or routes.
2026-07-30 07:18:29 +08:00
Maofeng 68218b2eae perf(console): decouple invocation telemetry writes
Move Axum management and gateway adapters into a dedicated module while preserving the existing public routers and response contracts.

Replace per-invocation full snapshots with a bounded asynchronous writer that batches metric deltas and events. Serialize it with control snapshots, flush before lifecycle persistence and graceful shutdown, and keep invocation results independent from telemetry failures.

Enable the Runtime compilation cache beside the Console artifact directory and verify metric recovery across a lifecycle flush.
2026-07-30 07:18:25 +08:00
Maofeng 53a8c6b690 feat(runtime): cache compiled components
Enable Wasmtime's persistent compilation cache and expose an optional cache directory in RuntimeConfig.

Create and canonicalize the configured directory before Engine construction, and report cache filesystem and configuration failures through dedicated RuntimeError variants.
2026-07-30 07:18:21 +08:00
Maofeng 38a397fa20 fix(package): make dev deployments retryable
Track the source fingerprint captured before compilation so edits made during a build trigger another cycle.

Keep successful build artifacts pending while the Console is unavailable and retry deployment without recompiling. Make activation retries idempotent by pruning every stale matching development revision after the active revision is switched.
2026-07-30 07:18:17 +08:00
19 changed files with 2759 additions and 2330 deletions
Generated
+100 -18
View File
@@ -673,7 +673,7 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1"
dependencies = [
"thiserror",
"thiserror 2.0.19",
]
[[package]]
@@ -1042,6 +1042,27 @@ dependencies = [
"crypto-common",
]
[[package]]
name = "directories-next"
version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "339ee130d97a610ea5a5872d2bbb130fdf68884ff09d3028b81bec8a1ac23bbc"
dependencies = [
"cfg-if",
"dirs-sys-next",
]
[[package]]
name = "dirs-sys-next"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
dependencies = [
"libc",
"redox_users",
"winapi",
]
[[package]]
name = "displaydoc"
version = "0.2.6"
@@ -1937,7 +1958,7 @@ dependencies = [
"jni-sys",
"log",
"simd_cesu8",
"thiserror",
"thiserror 2.0.19",
"walkdir",
"windows-link",
]
@@ -2057,6 +2078,15 @@ dependencies = [
"cc",
]
[[package]]
name = "libredox"
version = "0.1.18"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c943259e342f1e06ff2da7a83eabdfe7f92ce10262688dbf1895ff0b3e6e4652"
dependencies = [
"libc",
]
[[package]]
name = "linkme"
version = "0.3.37"
@@ -2654,7 +2684,7 @@ dependencies = [
"rustc-hash 2.1.3",
"rustls",
"socket2",
"thiserror",
"thiserror 2.0.19",
"tokio",
"tracing",
"web-time",
@@ -2677,7 +2707,7 @@ dependencies = [
"rustls",
"rustls-pki-types",
"slab",
"thiserror",
"thiserror 2.0.19",
"tinyvec",
"tracing",
"web-time",
@@ -2856,6 +2886,17 @@ dependencies = [
"bitflags",
]
[[package]]
name = "redox_users"
version = "0.4.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
dependencies = [
"getrandom 0.2.17",
"libredox",
"thiserror 1.0.69",
]
[[package]]
name = "regalloc2"
version = "0.13.5"
@@ -3582,7 +3623,7 @@ dependencies = [
"tantivy-stacker",
"tantivy-tokenizer-api",
"tempfile",
"thiserror",
"thiserror 2.0.19",
"time",
"typetag",
"uuid",
@@ -3718,13 +3759,33 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "thiserror"
version = "1.0.69"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52"
dependencies = [
"thiserror-impl 1.0.69",
]
[[package]]
name = "thiserror"
version = "2.0.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9"
dependencies = [
"thiserror-impl",
"thiserror-impl 2.0.19",
]
[[package]]
name = "thiserror-impl"
version = "1.0.69"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
@@ -4084,7 +4145,7 @@ checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c"
dependencies = [
"crossbeam-channel",
"symlink",
"thiserror",
"thiserror 2.0.19",
"time",
"tracing-subscriber",
]
@@ -4152,7 +4213,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6b6c49aecd4abae3ffa88ab1e28b3de7a7497a9732be828e3a6f1855b6ea80fd"
dependencies = [
"mimalloc",
"thiserror",
"thiserror 2.0.19",
"tracing",
"tracing-subscriber",
"turso_core",
@@ -4219,7 +4280,7 @@ dependencies = [
"strum_macros",
"tantivy",
"tempfile",
"thiserror",
"thiserror 2.0.19",
"tracing",
"tracing-subscriber",
"turso_ext",
@@ -4264,7 +4325,7 @@ dependencies = [
"miette",
"strum",
"strum_macros",
"thiserror",
"thiserror 2.0.19",
"turso_macros",
]
@@ -4312,7 +4373,7 @@ dependencies = [
"roaring",
"serde",
"serde_json",
"thiserror",
"thiserror 2.0.19",
"tracing",
"turso_core",
"turso_parser",
@@ -4645,7 +4706,7 @@ dependencies = [
"serde",
"serde_json",
"tempfile",
"thiserror",
"thiserror 2.0.19",
"toasty",
"toasty-driver-turso",
"tokio",
@@ -4668,7 +4729,7 @@ dependencies = [
"serde_json",
"sha2",
"tempfile",
"thiserror",
"thiserror 2.0.19",
"toml",
"wit-component 0.243.0",
"wit-parser 0.243.0",
@@ -4680,7 +4741,7 @@ name = "wasmeld-runtime"
version = "0.1.0"
dependencies = [
"serde",
"thiserror",
"thiserror 2.0.19",
"toml",
"wasmeld-package",
"wasmtime",
@@ -4755,6 +4816,7 @@ dependencies = [
"target-lexicon",
"wasmparser 0.243.0",
"wasmtime-environ",
"wasmtime-internal-cache",
"wasmtime-internal-component-macro",
"wasmtime-internal-component-util",
"wasmtime-internal-cranelift",
@@ -4794,6 +4856,26 @@ dependencies = [
"wasmtime-internal-component-util",
]
[[package]]
name = "wasmtime-internal-cache"
version = "41.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a31a582632f47753e4a9a5412ab5edfbc05a6df9031336a2f2eff46a70a1cac"
dependencies = [
"base64 0.22.1",
"directories-next",
"log",
"postcard",
"rustix 1.1.4",
"serde",
"serde_derive",
"sha2",
"toml",
"wasmtime-environ",
"windows-sys 0.61.2",
"zstd",
]
[[package]]
name = "wasmtime-internal-component-macro"
version = "41.0.0"
@@ -4834,7 +4916,7 @@ dependencies = [
"pulley-interpreter",
"smallvec",
"target-lexicon",
"thiserror",
"thiserror 2.0.19",
"wasmparser 0.243.0",
"wasmtime-environ",
"wasmtime-internal-math",
@@ -4969,7 +5051,7 @@ dependencies = [
"io-lifetimes",
"rustix 1.1.4",
"system-interface",
"thiserror",
"thiserror 2.0.19",
"tokio",
"tracing",
"url",
@@ -5076,7 +5158,7 @@ dependencies = [
"regalloc2",
"smallvec",
"target-lexicon",
"thiserror",
"thiserror 2.0.19",
"wasmparser 0.243.0",
"wasmtime-environ",
"wasmtime-internal-cranelift",
@@ -5540,7 +5622,7 @@ dependencies = [
"flate2",
"indexmap",
"memchr",
"thiserror",
"thiserror 2.0.19",
"zopfli",
]
+1 -1
View File
@@ -35,7 +35,7 @@ toml = "0.9.8"
tower-http = { version = "0.7.0", features = ["cors", "trace"] }
tracing = "0.1.44"
tracing-subscriber = { version = "0.3.23", features = ["env-filter", "fmt"] }
wasmtime = { version = "=41.0.0", default-features = false, features = ["component-model", "cranelift", "runtime", "std"] }
wasmtime = { version = "=41.0.0", default-features = false, features = ["cache", "component-model", "cranelift", "runtime", "std"] }
wasmtime-wasi = { version = "=41.0.0", default-features = false, features = ["p2"] }
wit-bindgen = "=0.41.0"
wit-component = "=0.243.0"
+2 -1
View File
@@ -68,7 +68,8 @@ cargo run -p wasmeld-package --bin wasmeld -- \
首次构建和每次源码变化都会自动完成 WIT 同步、Debug 编译、打包、注册和 Deployment
切换。默认使用 `echo-dev` 这类独立服务 ID,并根据 Component 内容生成
`0.1.0-dev.h<hash>` Revision,不会覆盖正式服务。新版本构建、校验或启动失败时,上一
版本继续运行;切换成功后旧开发 Revision 会被注销并释放。
版本继续运行;切换成功后旧开发 Revision 会被注销并释放。编译期间的新保存会继续
排队构建,Console 暂时不可用时则复用已构建制品重试部署。
管理面每 5 秒刷新一次,可以直接在调用面板预览新版本。只构建并部署一次可使用
`--once`;其它选项包括 `--id``--console``--release``--poll-ms`。管理 API
+379
View File
@@ -0,0 +1,379 @@
import {
ArrowRight,
Clock3,
CloudUpload,
FileCode2,
Package,
Play,
RadioTower,
RefreshCw,
ShieldCheck,
SquareTerminal,
X,
} from "lucide-react";
import { FormEvent, ReactNode, useState } from "react";
import { InvokeResult, RegisterComponentInput } from "../api";
import {
FormattedInvocationOutput,
Service,
formatInvocationOutput,
invocationInputSize,
parseInvocationInput,
} from "../console-model";
function DialogFrame({
title,
description,
icon,
onClose,
closeDisabled = false,
children,
}: {
title: string;
description: string;
icon: ReactNode;
onClose: () => void;
closeDisabled?: boolean;
children: ReactNode;
}) {
return (
<div className="dialog-backdrop">
<dialog
open
className="dialog"
aria-label={title}
onCancel={(event) => {
if (closeDisabled) event.preventDefault();
else onClose();
}}
onKeyDown={(event) => {
if (event.key === "Escape" && !closeDisabled) onClose();
}}
>
<div className="dialog-header">
<div className="dialog-title">
<span>{icon}</span>
<div>
<h2>{title}</h2>
<p>{description}</p>
</div>
</div>
<button
className="icon-button"
type="button"
aria-label="关闭"
title="关闭"
disabled={closeDisabled}
onClick={onClose}
>
<X size={18} />
</button>
</div>
{children}
</dialog>
</div>
);
}
export function ActivateDeploymentDialog({
service,
currentRevision,
submitting,
onClose,
onConfirm,
}: {
service: Service;
currentRevision: string | null;
submitting: boolean;
onClose: () => void;
onConfirm: () => void;
}) {
return (
<DialogFrame
title="切换对外版本"
description="更新该服务在 Gateway 上接收请求的活动版本。"
icon={<RadioTower size={19} />}
onClose={onClose}
closeDisabled={submitting}
>
<div className="deployment-confirmation">
<div className="deployment-route">
<div>
<span></span>
<strong>{currentRevision ?? "尚未部署"}</strong>
</div>
<ArrowRight size={18} />
<div className="deployment-target">
<span></span>
<strong>{service.revision}</strong>
</div>
</div>
<p>
Wasmeld {service.id}@{service.revision} Actor
</p>
</div>
<div className="dialog-footer">
<button className="secondary-button" type="button" disabled={submitting} onClick={onClose}>
</button>
<button className="primary-button" type="button" disabled={submitting} onClick={onConfirm}>
{submitting ? <RefreshCw className="spin" size={16} /> : <RadioTower size={16} />}
{submitting ? "正在切换" : "确认切换"}
</button>
</div>
</DialogFrame>
);
}
export function DeployDialog({
onClose,
onRegister,
}: {
onClose: () => void;
onRegister: (input: RegisterComponentInput) => Promise<void>;
}) {
const [file, setFile] = useState<File | null>(null);
const [submitting, setSubmitting] = useState(false);
const [error, setError] = useState("");
const fileName = file?.name ?? "";
async function submit(event: FormEvent) {
event.preventDefault();
if (!file) return;
if (!file.name.toLowerCase().endsWith(".wasmpkg")) {
setError("请选择 .wasmpkg 组件包");
return;
}
setSubmitting(true);
setError("");
try {
await onRegister({ file });
} catch (submitError) {
setError(submitError instanceof Error ? submitError.message : "组件注册失败");
setSubmitting(false);
}
}
return (
<DialogFrame
title="注册组件包"
description="组件元数据与运行边界由平台自动校验。"
icon={<CloudUpload size={19} />}
onClose={onClose}
>
<form className="dialog-form" onSubmit={submit}>
<label className={fileName ? "upload-zone has-file" : "upload-zone"}>
<input
type="file"
accept=".wasmpkg,application/zip,application/vnd.wasm.component-package"
onChange={(event) => {
setFile(event.target.files?.[0] ?? null);
setError("");
}}
/>
{fileName ? <Package size={28} /> : <CloudUpload size={28} />}
<strong>{fileName || "选择组件包"}</strong>
<span>
{file
? `${(file.size / 1024 / 1024).toFixed(2)} MB · 等待校验`
: ".wasmpkg · 最大 64 MB"}
</span>
</label>
{error && <div className="form-error">{error}</div>}
<div className="dialog-footer">
<button className="secondary-button" type="button" onClick={onClose}>
</button>
<button className="primary-button" type="submit" disabled={submitting || !fileName}>
{submitting ? <RefreshCw className="spin" size={16} /> : <ShieldCheck size={16} />}
{submitting ? "正在注册" : "校验并注册"}
</button>
</div>
</form>
</DialogFrame>
);
}
export function WitPackageDialog({
onClose,
onPublish,
}: {
onClose: () => void;
onPublish: (file: File) => Promise<void>;
}) {
const [file, setFile] = useState<File | null>(null);
const [submitting, setSubmitting] = useState(false);
const [error, setError] = useState("");
const fileName = file?.name ?? "";
async function submit(event: FormEvent) {
event.preventDefault();
if (!file) return;
if (!file.name.toLowerCase().endsWith(".wasm")) {
setError("请选择由 wasmeld wit build 生成的 .wasm 文件");
return;
}
if (file.size > 4 * 1024 * 1024) {
setError("WIT 包不能超过 4 MB");
return;
}
setSubmitting(true);
setError("");
try {
await onPublish(file);
} catch (submitError) {
setError(submitError instanceof Error ? submitError.message : "WIT 包发布失败");
setSubmitting(false);
}
}
return (
<DialogFrame
title="发布 WIT 包"
description="包名、语义版本和直接依赖将从二进制 WIT Package 读取。"
icon={<FileCode2 size={19} />}
onClose={onClose}
>
<form className="dialog-form" onSubmit={submit}>
<label className={fileName ? "upload-zone has-file" : "upload-zone"}>
<input
type="file"
accept=".wasm,application/wasm"
onChange={(event) => {
setFile(event.target.files?.[0] ?? null);
setError("");
}}
/>
{fileName ? <FileCode2 size={28} /> : <CloudUpload size={28} />}
<strong>{fileName || "选择二进制 WIT Package"}</strong>
<span>
{file ? `${(file.size / 1024).toFixed(1)} KB · 等待校验` : ".wasm · 最大 4 MB"}
</span>
</label>
{error && <div className="form-error">{error}</div>}
<div className="dialog-footer">
<button className="secondary-button" type="button" onClick={onClose}>
</button>
<button className="primary-button" type="submit" disabled={submitting || !fileName}>
{submitting ? <RefreshCw className="spin" size={16} /> : <ShieldCheck size={16} />}
{submitting ? "正在发布" : "校验并发布"}
</button>
</div>
</form>
</DialogFrame>
);
}
export function InvokeDialog({
service,
onClose,
onInvoke,
}: {
service: Service;
onClose: () => void;
onInvoke: (service: Service, input: Uint8Array) => Promise<InvokeResult>;
}) {
const [input, setInput] = useState(service.id === "echo" ? "hello wasm" : "");
const [format, setFormat] = useState<"utf8" | "hex">("utf8");
const [output, setOutput] = useState("");
const [outputBytes, setOutputBytes] = useState<number | null>(null);
const [outputFormat, setOutputFormat] = useState<FormattedInvocationOutput | null>(null);
const [running, setRunning] = useState(false);
const [failed, setFailed] = useState(false);
const [latency, setLatency] = useState<number | null>(null);
async function invoke() {
setRunning(true);
setOutput("");
setOutputBytes(null);
setOutputFormat(null);
setFailed(false);
setLatency(null);
try {
const result = await onInvoke(service, parseInvocationInput(input, format));
const formatted = formatInvocationOutput(service, result.output, format);
setOutput(formatted.text);
setOutputBytes(result.outputBytes);
setOutputFormat(formatted);
setLatency(result.latencyMs);
} catch (invokeError) {
setFailed(true);
setOutput(invokeError instanceof Error ? invokeError.message : "组件调用失败");
} finally {
setRunning(false);
}
}
return (
<DialogFrame
title={`调用 ${service.id}@${service.revision}`}
description="请求将进入该实例的串行 mailbox。"
icon={<SquareTerminal size={19} />}
onClose={onClose}
>
<div className="invoke-body">
<div className="invoke-toolbar">
<div className="segmented" aria-label="输入编码">
<button
type="button"
className={format === "utf8" ? "active" : ""}
onClick={() => setFormat("utf8")}
>
UTF-8
</button>
<button
type="button"
className={format === "hex" ? "active" : ""}
onClick={() => setFormat("hex")}
>
HEX
</button>
</div>
<span>{invocationInputSize(input, format)} B</span>
</div>
<label className="code-field">
<span></span>
<textarea
value={input}
onChange={(event) => setInput(event.target.value)}
spellCheck={false}
placeholder={format === "utf8" ? "输入请求内容" : "00 ff a1"}
/>
</label>
<div className={failed ? "invoke-output failed" : "invoke-output"}>
<div>
<span></span>
{output && latency !== null && outputBytes !== null && outputFormat && (
<div className="invoke-output-meta">
<span>
{outputBytes} B · {outputFormat.format}
{outputFormat.automatic ? " · 自动" : ""}
</span>
<small>
<Clock3 size={13} />
{latency} ms
</small>
</div>
)}
</div>
<pre>{running ? "invoking..." : output || "等待调用"}</pre>
</div>
</div>
<div className="dialog-footer">
<button className="secondary-button" type="button" onClick={onClose}>
</button>
<button className="primary-button" type="button" disabled={running} onClick={invoke}>
{running ? <RefreshCw className="spin" size={16} /> : <Play size={16} />}
</button>
</div>
</DialogFrame>
);
}
+986
View File
@@ -0,0 +1,986 @@
import {
Activity,
AlertTriangle,
Archive,
ArrowRight,
Box,
Braces,
Check,
CircleStop,
CloudUpload,
Cpu,
Download,
FileCode2,
Package,
Play,
RadioTower,
RefreshCw,
RotateCcw,
Search,
Server,
SquareTerminal,
X,
Zap,
} from "lucide-react";
import { ReactNode, useState } from "react";
import {
BackendCapability,
BackendRuntime,
BackendWitPackage,
witPackageDownloadUrl,
} from "../api";
import {
EventTone,
RuntimeEvent,
Service,
ServiceStatus,
STATUS_META,
formatDuration,
serviceKey,
} from "../console-model";
function PageHeading({
eyebrow,
title,
description,
actions,
}: {
eyebrow: string;
title: string;
description: string;
actions?: ReactNode;
}) {
return (
<div className="page-heading">
<div>
<span className="eyebrow">{eyebrow}</span>
<h1>{title}</h1>
<p>{description}</p>
</div>
{actions && <div className="heading-actions">{actions}</div>}
</div>
);
}
export function Overview({
services,
filteredServices,
events,
running,
faulted,
totalCalls,
deploymentCount,
runtime,
selectedService,
onSelect,
onDeploy,
onInvoke,
onStatus,
onActivate,
onViewAll,
onRefresh,
onRuntimeAction,
runtimeAction,
deploymentAction,
}: {
services: Service[];
filteredServices: Service[];
events: RuntimeEvent[];
running: number;
faulted: number;
totalCalls: number;
deploymentCount: number;
runtime: BackendRuntime | null;
selectedService: Service | null;
onSelect: (service: Service) => void;
onDeploy: () => void;
onInvoke: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
onViewAll: () => void;
onRefresh: () => void;
onRuntimeAction: (action: "start" | "stop" | "restart") => void;
runtimeAction: "start" | "stop" | "restart" | null;
deploymentAction: string | null;
}) {
const errorCount = services.reduce((sum, service) => sum + service.errors, 0);
const latencies = services
.map((service) => service.latencyMs)
.filter((latency): latency is number => latency !== null);
const latestLatency = latencies.length ? Math.max(...latencies) : null;
return (
<>
<PageHeading
eyebrow="LOCAL RUNTIME"
title="运行概览"
description={`${services.length} 个已注册版本,${running} 个 Actor 正在运行。`}
actions={
<>
<button
className="icon-button"
type="button"
aria-label="刷新"
title="刷新"
onClick={onRefresh}
>
<RefreshCw size={17} />
</button>
<button className="primary-button" type="button" onClick={onDeploy}>
<CloudUpload size={17} />
</button>
</>
}
/>
<section className="metric-grid" aria-label="关键指标">
<Metric
icon={<Package size={18} />}
label="服务版本"
value={services.length.toString()}
note={`${deploymentCount} 个对外服务`}
tone="cyan"
/>
<Metric
icon={<Activity size={18} />}
label="运行实例"
value={running.toString()}
note={`${services.length - running} 个未运行`}
tone="green"
/>
<Metric
icon={<Zap size={18} />}
label="累计调用"
value={totalCalls.toLocaleString("zh-CN")}
note="持久化统计"
tone="amber"
/>
<Metric
icon={<AlertTriangle size={18} />}
label="异常"
value={(faulted + errorCount).toString()}
note={faulted + errorCount ? "需要检查" : "无异常"}
tone="coral"
/>
</section>
<section className="runtime-band">
<div className="runtime-band-title">
<div className="runtime-symbol">
<Cpu size={18} />
</div>
<div>
<strong>Wasmeld Runtime</strong>
<span>
{runtime?.status === "running"
? "Engine 已启用 fuel 与 epoch interruption"
: "Engine 与所有 Actor 已释放"}
</span>
</div>
<div className="runtime-controls" aria-label="Runtime 生命周期">
{runtime?.status === "stopped" ? (
<button
className="icon-button"
type="button"
aria-label="启动 Runtime"
title="启动 Runtime"
disabled={runtimeAction !== null}
onClick={() => onRuntimeAction("start")}
>
<Play size={15} />
</button>
) : (
<>
<button
className="icon-button"
type="button"
aria-label="重启 Runtime"
title="重启 Runtime"
disabled={runtimeAction !== null || !runtime}
onClick={() => onRuntimeAction("restart")}
>
<RotateCcw size={15} />
</button>
<button
className="icon-button runtime-stop-button"
type="button"
aria-label="停止 Runtime"
title="停止 Runtime"
disabled={runtimeAction !== null || !runtime}
onClick={() => onRuntimeAction("stop")}
>
<CircleStop size={15} />
</button>
</>
)}
</div>
</div>
<RuntimeStat
label="已加载版本"
value={(runtime?.registered_services ?? 0).toString()}
detail={`${runtime?.managed_services ?? services.length} 个受管制品`}
/>
<RuntimeStat
label="活跃 Store"
value={(runtime?.running_services ?? running).toString()}
detail="串行 Actor"
/>
<RuntimeStat
label="最近延迟"
value={latestLatency === null ? "—" : `${latestLatency} ms`}
detail="各服务最近调用"
/>
<RuntimeStat
label="运行时间"
value={runtime?.status === "running" ? formatDuration(runtime.uptime_ms) : "—"}
detail="当前 Runtime 实例"
/>
</section>
<div className="overview-grid">
<section className="panel service-panel">
<div className="panel-header">
<div>
<h2></h2>
<p> Actor </p>
</div>
<button className="text-button" type="button" onClick={onViewAll}>
<ArrowRight size={15} />
</button>
</div>
<ServiceTable
services={filteredServices.slice(0, 5)}
selectedKey={selectedService ? serviceKey(selectedService) : ""}
onSelect={onSelect}
onInvoke={onInvoke}
onStatus={onStatus}
onActivate={onActivate}
canActivate={runtime?.status === "running"}
deploymentAction={deploymentAction}
compact
/>
</section>
<aside className="panel event-panel">
<div className="panel-header">
<div>
<h2></h2>
<p>Runtime </p>
</div>
<span className="live-label">
<span />
LIVE
</span>
</div>
<EventList events={events.slice(0, 5)} />
</aside>
</div>
{selectedService && (
<section className="detail-strip">
<div className="detail-identity">
<div className="artifact-icon">
<FileCode2 size={20} />
</div>
<div>
<span></span>
<strong>
{selectedService.id}
<small>@{selectedService.revision}</small>
</strong>
</div>
</div>
<DetailValue label="内存上限" value={`${selectedService.memoryMb} MB`} />
<DetailValue label="Fuel / 调用" value={selectedService.fuel} />
<DetailValue label="Deadline" value={`${selectedService.deadlineMs} ms`} />
<DetailValue label="Mailbox" value={selectedService.mailbox.toString()} />
<div className="detail-actions">
{!selectedService.active && (
<button
className="secondary-button"
type="button"
disabled={runtime?.status !== "running" || deploymentAction !== null}
onClick={() => onActivate(selectedService)}
>
<RadioTower size={16} />
</button>
)}
<button
className="secondary-button"
type="button"
disabled={selectedService.status !== "running"}
onClick={() => onInvoke(selectedService)}
>
<SquareTerminal size={16} />
</button>
</div>
</section>
)}
</>
);
}
function Metric({
icon,
label,
value,
note,
tone,
}: {
icon: ReactNode;
label: string;
value: string;
note: string;
tone: string;
}) {
return (
<article className="metric">
<div className={`metric-icon metric-${tone}`}>{icon}</div>
<div className="metric-copy">
<span>{label}</span>
<strong>{value}</strong>
</div>
<small>{note}</small>
</article>
);
}
function RuntimeStat({ label, value, detail }: { label: string; value: string; detail: string }) {
return (
<div className="runtime-stat">
<span>{label}</span>
<strong>{value}</strong>
<small>{detail}</small>
</div>
);
}
function DetailValue({ label, value }: { label: string; value: string }) {
return (
<div className="detail-value">
<span>{label}</span>
<strong>{value}</strong>
</div>
);
}
function StatusBadge({ status }: { status: ServiceStatus }) {
const meta = STATUS_META[status];
return (
<span className={`status-badge ${meta.className}`}>
<span />
{meta.label}
</span>
);
}
function CapabilitySummary({ capabilities }: { capabilities: BackendCapability[] }) {
if (capabilities.length === 0) {
return <span className="capability-empty"></span>;
}
const [first, ...remaining] = capabilities;
return (
<span
className="capability-summary"
title={capabilities.map((capability) => capability.interface).join("\n")}
>
<Braces size={12} />
<span>
{first.package}/{first.name} @{first.version}
</span>
{remaining.length > 0 && <small>+{remaining.length}</small>}
</span>
);
}
function ServiceTable({
services,
selectedKey,
onSelect,
onInvoke,
onStatus,
onActivate,
canActivate,
deploymentAction,
compact = false,
}: {
services: Service[];
selectedKey?: string;
onSelect: (service: Service) => void;
onInvoke: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
canActivate: boolean;
deploymentAction: string | null;
compact?: boolean;
}) {
if (services.length === 0) {
return (
<div className="empty-state">
<Search size={22} />
<strong></strong>
<span></span>
</div>
);
}
return (
<div className="table-scroll">
<table className={`data-table ${compact ? "service-table-compact" : "service-table-full"}`}>
<thead>
<tr>
<th></th>
<th></th>
{!compact && <th></th>}
{!compact && <th>Host </th>}
<th></th>
<th></th>
<th>
<span className="sr-only"></span>
</th>
</tr>
</thead>
<tbody>
{services.map((service) => (
<tr
key={`${service.id}-${service.revision}`}
className={selectedKey === serviceKey(service) ? "selected" : ""}
>
<td>
<button
className="service-cell service-select"
type="button"
aria-label={`选择 ${service.id}@${service.revision}`}
onClick={() => onSelect(service)}
>
<span className="service-glyph">
<Box size={16} />
</span>
<span>
<span className="service-title">
<strong>{service.id}</strong>
{service.active && (
<span className="active-deployment-badge">
<RadioTower size={10} />
</span>
)}
</span>
<small>
{service.revision} · {service.description}
</small>
</span>
</button>
</td>
<td>
<StatusBadge status={service.status} />
</td>
{!compact && <td className="artifact-cell">{service.artifact}</td>}
{!compact && (
<td>
<CapabilitySummary capabilities={service.capabilities} />
</td>
)}
<td className="numeric-cell">{service.calls.toLocaleString("zh-CN")}</td>
<td className="numeric-cell">
{service.latencyMs ? `${service.latencyMs} ms` : "—"}
</td>
<td>
<div className="row-actions">
<button
className={service.active ? "active-deployment-button" : ""}
type="button"
aria-label={
service.active
? `${service.id}@${service.revision} 是当前对外版本`
: `${service.id}@${service.revision} 设为对外版本`
}
title={
service.active
? "当前对外版本"
: canActivate
? "设为对外版本"
: "请先启动 Runtime"
}
disabled={service.active || !canActivate || deploymentAction !== null}
onClick={() => onActivate(service)}
>
{deploymentAction === serviceKey(service) ? (
<RefreshCw className="spin" size={15} />
) : (
<RadioTower size={15} />
)}
</button>
{service.status === "running" ? (
<>
<button
type="button"
aria-label={`调用 ${service.id}`}
title="测试调用"
onClick={() => onInvoke(service)}
>
<SquareTerminal size={15} />
</button>
<button
type="button"
aria-label={`重启 ${service.id}`}
title="重启"
onClick={() => onStatus(service, "running")}
>
<RotateCcw size={15} />
</button>
<button
type="button"
aria-label={`停止 ${service.id}`}
title="停止"
onClick={() => onStatus(service, "stopped")}
>
<CircleStop size={15} />
</button>
</>
) : (
<button
type="button"
aria-label={`启动 ${service.id}`}
title="启动"
onClick={() => onStatus(service, "running")}
>
<Play size={15} />
</button>
)}
</div>
</td>
</tr>
))}
</tbody>
</table>
</div>
);
}
function EventList({ events }: { events: RuntimeEvent[] }) {
const icons: Record<EventTone, ReactNode> = {
success: <Check size={14} />,
warning: <AlertTriangle size={14} />,
danger: <X size={14} />,
neutral: <CircleStop size={14} />,
};
return (
<div className="event-list">
{events.map((event) => (
<div className="event-row" key={event.id}>
<span className={`event-icon event-${event.tone}`}>{icons[event.tone]}</span>
<div>
<strong>{event.title}</strong>
<span>{event.detail}</span>
</div>
<time>{event.time}</time>
</div>
))}
</div>
);
}
export function ServicesView({
services,
statusFilter,
onFilter,
onDeploy,
onInvoke,
onSelect,
onStatus,
onActivate,
canActivate,
deploymentAction,
}: {
services: Service[];
statusFilter: "all" | ServiceStatus;
onFilter: (status: "all" | ServiceStatus) => void;
onDeploy: () => void;
onInvoke: (service: Service) => void;
onSelect: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
canActivate: boolean;
deploymentAction: string | null;
}) {
return (
<>
<PageHeading
eyebrow="REGISTRY"
title="服务版本"
description="管理不可变 Wasm Component 制品及当前对外版本。"
actions={
<button className="primary-button" type="button" onClick={onDeploy}>
<CloudUpload size={17} />
</button>
}
/>
<section className="panel full-panel">
<div className="filterbar">
<div className="segmented" aria-label="状态筛选">
{[
["all", "全部"],
["running", "运行中"],
["stopped", "已停止"],
["faulted", "故障"],
].map(([value, label]) => (
<button
type="button"
key={value}
className={statusFilter === value ? "active" : ""}
onClick={() => onFilter(value as "all" | ServiceStatus)}
>
{label}
</button>
))}
</div>
<span className="result-count">{services.length} </span>
</div>
<ServiceTable
services={services}
onSelect={onSelect}
onInvoke={onInvoke}
onStatus={onStatus}
onActivate={onActivate}
canActivate={canActivate}
deploymentAction={deploymentAction}
/>
</section>
</>
);
}
export function WitPackagesView({
apiBase,
packages,
onPublish,
}: {
apiBase: string;
packages: BackendWitPackage[];
onPublish: () => void;
}) {
function downloadPackage(packageMetadata: BackendWitPackage) {
const anchor = document.createElement("a");
anchor.href = witPackageDownloadUrl(apiBase, packageMetadata);
anchor.download = `${packageMetadata.name.replace(":", "-")}-${packageMetadata.version}.wasm`;
document.body.append(anchor);
anchor.click();
anchor.remove();
}
return (
<>
<PageHeading
eyebrow="WIT REGISTRY"
title="WIT 包"
description="管理组件开发所依赖的不可变、版本化 WIT Package。"
actions={
<button className="primary-button" type="button" onClick={onPublish}>
<CloudUpload size={17} />
WIT
</button>
}
/>
<section className="panel full-panel">
<div className="filterbar">
<span className="registry-note"></span>
<span className="result-count">{packages.length} </span>
</div>
{packages.length === 0 ? (
<div className="empty-state">
<FileCode2 size={22} />
<strong> WIT </strong>
<span> WIT Package</span>
</div>
) : (
<div className="table-scroll">
<table className="data-table wit-package-table">
<thead>
<tr>
<th>Package</th>
<th></th>
<th>SHA-256</th>
<th>
<span className="sr-only"></span>
</th>
</tr>
</thead>
<tbody>
{packages.map((packageMetadata) => {
const dependencies = packageMetadata.dependencies
.map((dependency) => `${dependency.name}@${dependency.version}`)
.join(", ");
return (
<tr key={`${packageMetadata.name}@${packageMetadata.version}`}>
<td aria-label={`${packageMetadata.name}@${packageMetadata.version}`}>
<div className="service-cell">
<span className="service-glyph">
<FileCode2 size={16} />
</span>
<span>
<strong>{packageMetadata.name}</strong>
<small>{packageMetadata.version}</small>
</span>
</div>
</td>
<td className="artifact-cell" title={dependencies || "无直接依赖"}>
{dependencies || "无直接依赖"}
</td>
<td className="artifact-cell" title={packageMetadata.sha256}>
{packageMetadata.sha256}
</td>
<td>
<div className="row-actions">
<button
type="button"
title="下载 WIT 包"
aria-label={`下载 ${packageMetadata.name}@${packageMetadata.version}`}
onClick={() => downloadPackage(packageMetadata)}
>
<Download size={15} />
</button>
</div>
</td>
</tr>
);
})}
</tbody>
</table>
</div>
)}
</section>
</>
);
}
export function InstancesView({
services,
onInvoke,
onStatus,
onRefresh,
}: {
services: Service[];
onInvoke: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void;
onRefresh: () => void;
}) {
const instances = services.filter(
(service) => service.status === "running" || service.status === "faulted",
);
return (
<>
<PageHeading
eyebrow="ACTORS"
title="运行实例"
description={`${instances.length} 个 Actor 保持独立 Store 与 Instance。`}
actions={
<button
className="icon-button"
type="button"
title="刷新"
aria-label="刷新"
onClick={onRefresh}
>
<RefreshCw size={17} />
</button>
}
/>
<section className="instance-list">
{instances.length === 0 ? (
<output className="instance-placeholder">
<div className="instance-placeholder-icon">
<Cpu size={22} />
</div>
<strong></strong>
<span>0 ACTIVE ACTORS</span>
</output>
) : (
instances.map((service) => (
<article className="instance-row" key={service.id}>
<div className="instance-main">
<span className={`instance-indicator ${service.status}`} />
<div className="artifact-icon">
<Cpu size={19} />
</div>
<div>
<strong>
{service.id}@{service.revision}
</strong>
<span>actor/{service.id}-01</span>
</div>
</div>
<DetailValue
label="Store 内存"
value={
service.status === "running"
? `${Math.round(service.memoryMb * 0.42)} MB`
: "已释放"
}
/>
<DetailValue label="调用总数" value={service.calls.toLocaleString("zh-CN")} />
<DetailValue
label="最近延迟"
value={service.latencyMs ? `${service.latencyMs} ms` : "—"}
/>
<div className="instance-status">
<StatusBadge status={service.status} />
<span>{service.updatedAt}</span>
</div>
<div className="row-actions">
<button
type="button"
title="测试调用"
aria-label={`调用 ${service.id}`}
disabled={service.status !== "running"}
onClick={() => onInvoke(service)}
>
<SquareTerminal size={16} />
</button>
<button
type="button"
title="重启"
aria-label={`重启 ${service.id}`}
onClick={() => onStatus(service, "running")}
>
<RotateCcw size={16} />
</button>
<button
type="button"
title="停止"
aria-label={`停止 ${service.id}`}
disabled={service.status !== "running"}
onClick={() => onStatus(service, "stopped")}
>
<CircleStop size={16} />
</button>
</div>
</article>
))
)}
</section>
</>
);
}
export function ActivityView({
events,
services,
}: {
events: RuntimeEvent[];
services: Service[];
}) {
const calls = services.reduce((sum, service) => sum + service.calls, 0);
const errors = services.reduce((sum, service) => sum + service.errors, 0);
const latencies = services
.map((service) => service.latencyMs)
.filter((latency): latency is number => latency !== null);
const averageLatency = latencies.length
? latencies.reduce((sum, latency) => sum + latency, 0) / latencies.length
: null;
function exportEvents() {
const blob = new Blob([JSON.stringify(events, null, 2)], {
type: "application/json",
});
const url = URL.createObjectURL(blob);
const anchor = document.createElement("a");
anchor.href = url;
anchor.download = "wasmeld-events.json";
anchor.click();
URL.revokeObjectURL(url);
}
return (
<>
<PageHeading
eyebrow="EVENT STREAM"
title="调用记录"
description="Runtime 生命周期、调用结果与沙箱拒绝记录。"
actions={
<button className="secondary-button" type="button" onClick={exportEvents}>
<Archive size={16} />
</button>
}
/>
<section className="panel full-panel">
<div className="activity-summary">
<div>
<span></span>
<strong></strong>
</div>
<div>
<span></span>
<strong>{calls ? `${(((calls - errors) / calls) * 100).toFixed(2)}%` : "—"}</strong>
</div>
<div>
<span></span>
<strong>{averageLatency === null ? "—" : `${averageLatency.toFixed(1)} ms`}</strong>
</div>
<div>
<span></span>
<strong>{events.length}</strong>
</div>
</div>
<EventList events={events} />
</section>
</>
);
}
export function SettingsView({
endpoint,
onSaveEndpoint,
}: {
endpoint: string;
onSaveEndpoint: (value: string) => void;
}) {
const [value, setValue] = useState(endpoint);
return (
<>
<PageHeading
eyebrow="CONFIGURATION"
title="运行设置"
description="本地控制台连接与 Runtime 默认参数。"
/>
<form
className="settings-form"
onSubmit={(event) => {
event.preventDefault();
onSaveEndpoint(value);
}}
>
<section className="settings-section">
<div className="settings-heading">
<Server size={18} />
<div>
<h2></h2>
<p> API 访</p>
</div>
</div>
<div className="form-field wide-field">
<label htmlFor="endpoint">API </label>
<input id="endpoint" value={value} onChange={(event) => setValue(event.target.value)} />
<span></span>
</div>
</section>
<div className="form-footer">
<button className="primary-button" type="submit">
</button>
</div>
</form>
</>
);
}
+208
View File
@@ -0,0 +1,208 @@
import { FileCode2, History, LayoutDashboard, Package, Server, Settings } from "lucide-react";
import { BackendCapability, BackendEvent, BackendService } from "./api";
export type View = "overview" | "services" | "instances" | "wit-packages" | "activity" | "settings";
export type ServiceStatus = "running" | "stopped" | "faulted";
export type EventTone = "success" | "warning" | "danger" | "neutral";
export type Service = {
id: string;
revision: string;
description: string;
artifact: string;
status: ServiceStatus;
updatedAt: string;
memoryMb: number;
fuel: string;
deadlineMs: number;
mailbox: number;
maxInputKb: number;
capabilities: BackendCapability[];
calls: number;
errors: number;
latencyMs: number | null;
active: boolean;
};
export type RuntimeEvent = {
id: number;
time: string;
title: string;
detail: string;
tone: EventTone;
};
export type ConnectionState = "connecting" | "online" | "offline";
export const NAV_ITEMS: Array<{
id: View;
label: string;
icon: typeof LayoutDashboard;
}> = [
{ id: "overview", label: "运行概览", icon: LayoutDashboard },
{ id: "services", label: "服务版本", icon: Package },
{ id: "instances", label: "运行实例", icon: Server },
{ id: "wit-packages", label: "WIT 包", icon: FileCode2 },
{ id: "activity", label: "调用记录", icon: History },
{ id: "settings", label: "运行设置", icon: Settings },
];
export const STATUS_META: Record<ServiceStatus, { label: string; className: string }> = {
running: { label: "运行中", className: "status-running" },
stopped: { label: "已停止", className: "status-stopped" },
faulted: { label: "故障", className: "status-faulted" },
};
export function serviceKey(service: Pick<Service, "id" | "revision">) {
return `${service.id}@${service.revision}`;
}
export function toService(service: BackendService, activeRevision?: string): Service {
return {
id: service.id,
revision: service.revision,
description: "Wasm Component",
artifact: service.artifact.split(/[\\/]/).pop() ?? service.artifact,
status: service.status,
updatedAt: new Date(service.updated_at_ms).toLocaleString("zh-CN", {
hour12: false,
}),
memoryMb: Math.round(service.limits.memory_bytes / 1024 / 1024),
fuel: compactNumber(service.limits.fuel_per_call),
deadlineMs: service.limits.deadline_ms,
mailbox: service.limits.mailbox_capacity,
maxInputKb: Math.round(service.limits.max_input_bytes / 1024),
capabilities: service.capabilities,
calls: service.calls,
errors: service.errors,
latencyMs: service.last_latency_ms,
active: service.revision === activeRevision,
};
}
export function toEvent(event: BackendEvent): RuntimeEvent {
const identity =
event.service_id && event.revision ? `${event.service_id}@${event.revision}` : "Wasmeld";
const meta: Record<BackendEvent["kind"], { label: string; tone: EventTone }> = {
registered: { label: "已注册", tone: "success" },
unregistered: { label: "已注销", tone: "neutral" },
deployed: { label: "部署切换", tone: "success" },
started: { label: "已启动", tone: "success" },
stopped: { label: "已停止", tone: "neutral" },
invoked: { label: "调用完成", tone: "success" },
failed: { label: "操作失败", tone: "danger" },
};
return {
id: event.id,
time: new Date(event.timestamp_ms).toLocaleTimeString("zh-CN", {
hour12: false,
}),
title: `${identity} ${meta[event.kind].label}`,
detail: event.message,
tone: meta[event.kind].tone,
};
}
export function compactNumber(value: number) {
if (value >= 1_000_000) return `${value / 1_000_000}M`;
if (value >= 1_000) return `${value / 1_000}K`;
return value.toString();
}
export function formatDuration(milliseconds: number) {
const totalSeconds = Math.floor(milliseconds / 1000);
const hours = Math.floor(totalSeconds / 3600);
const minutes = Math.floor((totalSeconds % 3600) / 60);
const seconds = totalSeconds % 60;
return hours > 0 ? `${hours}h ${minutes}m` : `${minutes}m ${seconds}s`;
}
export function formatApiHost(value: string) {
try {
return new URL(value).host;
} catch {
return value;
}
}
export function parseInvocationInput(value: string, format: "utf8" | "hex") {
if (format === "utf8") return new TextEncoder().encode(value);
const compact = value.replace(/\s+/g, "");
if (compact.length % 2 !== 0 || /[^0-9a-f]/i.test(compact)) {
throw new Error("HEX 输入必须由成对的十六进制字符组成");
}
return Uint8Array.from(compact.match(/.{2}/g)?.map((byte) => Number.parseInt(byte, 16)) ?? []);
}
export function invocationInputSize(value: string, format: "utf8" | "hex") {
try {
return parseInvocationInput(value, format).byteLength;
} catch {
return 0;
}
}
export type FormattedInvocationOutput = {
text: string;
format: "UTF-8" | "HEX" | "U64 LE";
automatic: boolean;
};
export function formatHex(bytes: Uint8Array) {
return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(" ");
}
export function decodeReadableUtf8(bytes: Uint8Array) {
let text: string;
try {
text = new TextDecoder("utf-8", { fatal: true }).decode(bytes);
} catch {
return null;
}
for (const character of text) {
const codePoint = character.codePointAt(0) ?? 0;
const allowedWhitespace = codePoint === 0x09 || codePoint === 0x0a || codePoint === 0x0d;
if ((codePoint < 0x20 && !allowedWhitespace) || codePoint === 0x7f) {
return null;
}
}
return text;
}
export function formatInvocationOutput(
service: Service,
output: Uint8Array,
format: "utf8" | "hex",
) {
if (service.id === "counter" && output.byteLength === 8) {
return {
text: new DataView(output.buffer, output.byteOffset, output.byteLength)
.getBigUint64(0, true)
.toString(),
format: "U64 LE",
automatic: false,
} satisfies FormattedInvocationOutput;
}
if (format === "hex") {
return {
text: formatHex(output) || "(empty)",
format: "HEX",
automatic: false,
} satisfies FormattedInvocationOutput;
}
const text = decodeReadableUtf8(output);
if (text !== null) {
return {
text: text || "(empty)",
format: "UTF-8",
automatic: false,
} satisfies FormattedInvocationOutput;
}
return {
text: formatHex(output),
format: "HEX",
automatic: true,
} satisfies FormattedInvocationOutput;
}
File diff suppressed because it is too large Load Diff
+19 -6
View File
@@ -83,9 +83,22 @@ test("server-renders the Wasm management console", async () => {
});
test("uses TanStack Start routing and produces Node artifacts", async () => {
const [rootRoute, indexRoute, apiClient, router, packageJson, viteConfig] = await Promise.all([
const [
rootRoute,
indexRoute,
views,
dialogs,
consoleModel,
apiClient,
router,
packageJson,
viteConfig,
] = await Promise.all([
readFile(new URL("../src/routes/__root.tsx", import.meta.url), "utf8"),
readFile(new URL("../src/routes/index.tsx", import.meta.url), "utf8"),
readFile(new URL("../src/components/views.tsx", import.meta.url), "utf8"),
readFile(new URL("../src/components/dialogs.tsx", import.meta.url), "utf8"),
readFile(new URL("../src/console-model.ts", import.meta.url), "utf8"),
readFile(new URL("../src/api.ts", import.meta.url), "utf8"),
readFile(new URL("../src/router.tsx", import.meta.url), "utf8"),
readFile(new URL("../package.json", import.meta.url), "utf8"),
@@ -96,11 +109,11 @@ test("uses TanStack Start routing and produces Node artifacts", async () => {
assert.match(rootRoute, /<HeadContent \/>/);
assert.match(rootRoute, /<Scripts \/>/);
assert.match(indexRoute, /createFileRoute\("\/"\)/);
assert.match(indexRoute, /切换对外版本/);
assert.match(indexRoute, /Host 能力/);
assert.match(indexRoute, /service\.capabilities/);
assert.match(indexRoute, /TextDecoder\("utf-8", \{ fatal: true \}\)/);
assert.match(indexRoute, /outputFormat\.automatic/);
assert.match(dialogs, /切换对外版本/);
assert.match(views, /Host 能力/);
assert.match(views, /service\.capabilities/);
assert.match(consoleModel, /TextDecoder\("utf-8", \{ fatal: true \}\)/);
assert.match(dialogs, /outputFormat\.automatic/);
assert.match(apiClient, /\/api\/v1\/deployments/);
assert.match(apiClient, /capabilities: BackendCapability\[\]/);
assert.match(router, /createRouter/);
+7
View File
@@ -17,10 +17,17 @@ Wasm 制品保存在本地文件系统。服务 manifest、调用计数和最近
Host KV 使用 `(service_id, key)` 复合主键存储在该数据库中:同一服务的 Revision
共享数据,不同服务互相隔离,Runtime 或 Console 重启不会清空 KV。
调用计数、错误数、延迟和调用事件通过容量受限的异步队列批量写入数据库,Gateway
响应不等待 libSQL。队列不可用或进程异常退出时,尚未写入的遥测可能丢失,但不会把
已经成功的 Component 调用改写成失败。注册、Deployment 和生命周期变更仍在响应前
完成持久化。
`wasmeld-console` 启动时会在同一进程内创建 `wasmeld-runtime`、加载 Wasmtime Engine
并重新注册已保存的 Component。Runtime 停止后 Console HTTP 和数据库仍然在线;
再次启动 Runtime 会重新注册 Component。未部署的版本恢复为 `stopped`Deployment
指向的版本会用空初始化配置创建新的 Actor。Actor 的 Store 和线性内存不做快照。
编译后的本地机器码默认缓存在 `var/wasmeld/component-cache`,相同 Component 在
Console 或 Runtime 重启后可由 Wasmtime 直接复用;缓存不包含 Actor 内存。
## 启动
+661
View File
@@ -0,0 +1,661 @@
//! Axum management and public gateway protocol adapters.
use super::*;
use axum::{
Json, Router,
body::{Body, Bytes},
extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection},
http::{HeaderMap, HeaderValue, Method, StatusCode, header},
response::{IntoResponse, Response},
routing::{delete, get, post},
};
use serde_json::json;
use tower_http::{
cors::CorsLayer,
trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
};
use tracing::Level;
use wasmeld_package::wit_package::WitPackageError;
/// Builds the Axum management API around a shared [`Console`].
///
/// The returned router exposes Runtime control, component registration,
/// invocation, events, and WIT Registry endpoints.
pub fn app(console: Arc<Console>, allowed_origins: Vec<HeaderValue>) -> Router {
let max_body_bytes = console
.max_artifact_bytes()
.max(console.max_wit_package_bytes())
.saturating_add(1024 * 1024);
let mut cors = CorsLayer::new()
.allow_methods([Method::GET, Method::POST, Method::DELETE])
.allow_headers([header::CONTENT_TYPE]);
if !allowed_origins.is_empty() {
cors = cors.allow_origin(allowed_origins);
}
Router::new()
.route("/healthz", get(health))
.route("/api/v1/runtime", get(runtime_status))
.route("/api/v1/runtime/start", post(start_runtime))
.route("/api/v1/runtime/stop", post(stop_runtime))
.route("/api/v1/runtime/restart", post(restart_runtime))
.route("/api/v1/services", get(list_services).post(register))
.route(
"/api/v1/services/{id}/{revision}",
delete(unregister_service),
)
.route("/api/v1/deployments", get(list_deployments))
.route("/api/v1/deployments/{id}", get(get_deployment))
.route(
"/api/v1/deployments/{id}/activate",
post(activate_deployment),
)
.route(
"/api/v1/wit/packages",
get(list_wit_packages).post(publish_wit_package),
)
.route(
"/api/v1/wit/packages/{namespace}/{name}/{version}",
get(get_wit_package),
)
.route(
"/api/v1/wit/packages/{namespace}/{name}/{version}/content",
get(download_wit_package),
)
.route(
"/api/v1/services/{id}/{revision}/start",
post(start_service),
)
.route("/api/v1/services/{id}/{revision}/stop", post(stop_service))
.route(
"/api/v1/services/{id}/{revision}/restart",
post(restart_service),
)
.route(
"/api/v1/services/{id}/{revision}/invoke",
post(invoke_service),
)
.route("/api/v1/events", get(list_events))
.layer(DefaultBodyLimit::max(max_body_bytes))
.layer(cors)
.layer(
TraceLayer::new_for_http()
.make_span_with(DefaultMakeSpan::new().level(Level::INFO))
.on_response(DefaultOnResponse::new().level(Level::INFO)),
)
.with_state(console)
}
/// Builds the public data-plane API around the same resident Runtime.
///
/// This router intentionally contains no registration, lifecycle, Registry,
/// or deployment management routes.
pub fn gateway_app(console: Arc<Console>) -> Router {
let max_input_bytes = console.max_gateway_input_bytes();
Router::new()
.route("/healthz", get(health))
.route("/v1/services/{id}/invoke", post(gateway_invoke))
.layer(DefaultBodyLimit::max(max_input_bytes))
.layer(
TraceLayer::new_for_http()
.make_span_with(DefaultMakeSpan::new().level(Level::INFO))
.on_response(DefaultOnResponse::new().level(Level::INFO)),
)
.with_state(console)
}
async fn health() -> Json<serde_json::Value> {
Json(json!({ "status": "ok" }))
}
async fn gateway_invoke(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
headers: HeaderMap,
body: Result<Bytes, BytesRejection>,
) -> Result<Response, GatewayError> {
validate_path_segment("id", &service_id)?;
let content_type = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.split(';').next())
.map(str::trim);
if !content_type.is_some_and(|value| value.eq_ignore_ascii_case("application/octet-stream")) {
return Err(GatewayError::UnsupportedMediaType);
}
let input = body.map_err(|_| GatewayError::PayloadTooLarge {
limit: console.max_gateway_input_bytes(),
})?;
let invocation = run_blocking(console, move |console| {
console.gateway_invoke(&service_id, input.to_vec())
})
.await?;
let mut response_headers = HeaderMap::new();
response_headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/octet-stream"),
);
response_headers.insert(
"x-wasmeld-revision",
HeaderValue::from_str(&invocation.revision).map_err(|_| {
ConsoleError::InvalidRequest("revision is not a valid header".to_owned())
})?,
);
response_headers.insert(
"x-wasmeld-latency-ms",
HeaderValue::from_str(&invocation.result.latency_ms.to_string()).map_err(|_| {
ConsoleError::InvalidRequest("latency is not a valid header".to_owned())
})?,
);
Ok((response_headers, invocation.result.output).into_response())
}
async fn runtime_status(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(console.runtime_view()?))
}
async fn start_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.start_runtime()).await?,
))
}
async fn stop_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.stop_runtime()).await?,
))
}
async fn restart_runtime(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking_and_persist(console, |console| console.restart_runtime()).await?,
))
}
async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<ServiceList>, ApiError> {
Ok(Json(ServiceList {
services: console.services()?,
}))
}
async fn list_deployments(
State(console): State<Arc<Console>>,
) -> Result<Json<DeploymentList>, ApiError> {
Ok(Json(DeploymentList {
deployments: console.deployments()?,
}))
}
async fn get_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
Ok(Json(console.deployment(&service_id)?))
}
async fn activate_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
Json(request): Json<ActivateDeploymentRequest>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
validate_path_segment("revision", &request.revision)?;
let init_config = decode_optional_base64(request.init_config_base64.as_deref())?;
Ok(Json(
run_blocking_and_persist(console, move |console| {
console.activate_deployment(service_id, request.revision, init_config)
})
.await?,
))
}
async fn list_events(State(console): State<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
Ok(Json(EventList {
events: console.events()?,
}))
}
async fn list_wit_packages(
State(console): State<Arc<Console>>,
) -> Result<Json<WitPackageList>, ApiError> {
Ok(Json(WitPackageList {
packages: console.wit_registry.list()?,
}))
}
async fn get_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Json<WitPackageMetadata>, ApiError> {
Ok(Json(
console
.wit_registry
.metadata(&format!("{namespace}:{name}"), &version)?,
))
}
async fn download_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Response, ApiError> {
let (metadata, bytes) = console
.wit_registry
.download(&format!("{namespace}:{name}"), &version)?;
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/wasm")
.header(header::ETAG, format!("\"{}\"", metadata.sha256))
.body(Body::from(bytes))
.map_err(|error| {
ConsoleError::InvalidRequest(format!("failed to build package response: {error}"))
.into()
})
}
async fn publish_wit_package(
State(console): State<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<WitPackageMetadata>), ApiError> {
let mut package = None;
while let Some(field) = multipart
.next_field()
.await
.map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
{
match field.name() {
Some("package") => {
if package.is_some() {
return Err(ConsoleError::InvalidRequest(
"package field must occur exactly once".to_owned(),
)
.into());
}
package = Some(
field
.bytes()
.await
.map_err(|error| {
ConsoleError::InvalidRequest(format!(
"invalid WIT package field: {error}"
))
})?
.to_vec(),
);
}
Some(name) => {
return Err(ConsoleError::InvalidRequest(format!(
"unsupported multipart field {name}"
))
.into());
}
None => {}
}
}
let package = package
.ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
if package.len() > console.max_wit_package_bytes() {
return Err(WitRegistryError::TooLarge {
limit: console.max_wit_package_bytes(),
}
.into());
}
let metadata = run_blocking(console, move |console| {
Ok(console.wit_registry.publish(&package)?)
})
.await?;
Ok((StatusCode::CREATED, Json(metadata)))
}
async fn register(
State(console): State<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<ServiceView>), ApiError> {
let mut package = None;
while let Some(field) = multipart
.next_field()
.await
.map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
{
match field.name() {
Some("package") => {
if package.is_some() {
return Err(ConsoleError::InvalidRequest(
"package field must occur exactly once".to_owned(),
)
.into());
}
package = Some(
field
.bytes()
.await
.map_err(|error| {
ConsoleError::InvalidRequest(format!("invalid package field: {error}"))
})?
.to_vec(),
);
}
Some(name) => {
return Err(ConsoleError::InvalidRequest(format!(
"unsupported multipart field {name}"
))
.into());
}
None => {}
}
}
let package = package
.ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
let service =
run_blocking_and_persist(console, move |console| console.register_package(package)).await?;
Ok((StatusCode::CREATED, Json(service)))
}
async fn start_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
let init_config = decode_optional_base64(
request
.as_ref()
.and_then(|Json(request)| request.init_config_base64.as_deref()),
)?;
Ok(Json(
run_blocking_and_persist(console, move |console| console.start(&key, init_config)).await?,
))
}
async fn stop_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
Ok(Json(
run_blocking_and_persist(console, move |console| console.stop(&key)).await?,
))
}
async fn unregister_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
Ok(Json(
run_blocking_and_persist(console, move |console| console.unregister(&key)).await?,
))
}
async fn restart_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
let init_config = decode_optional_base64(
request
.as_ref()
.and_then(|Json(request)| request.init_config_base64.as_deref()),
)?;
Ok(Json(
run_blocking_and_persist(console, move |console| console.restart(&key, init_config))
.await?,
))
}
async fn invoke_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
Json(request): Json<InvokeRequest>,
) -> Result<Json<InvokeResponse>, ApiError> {
let key = api_key(id, revision)?;
let input = BASE64.decode(request.input_base64).map_err(|error| {
ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}"))
})?;
let result = run_blocking(console, move |console| console.invoke(&key, input)).await?;
Ok(Json(InvokeResponse {
output_base64: BASE64.encode(&result.output),
output_bytes: result.output.len(),
latency_ms: result.latency_ms,
}))
}
async fn run_blocking<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + Send + 'static,
{
// Wasmtime compilation and synchronous Actor calls must not occupy a Tokio
// async worker.
tokio::task::spawn_blocking(move || operation(console))
.await
.map_err(ConsoleError::from)
.and_then(|result| result)
}
async fn run_blocking_and_persist<T, F>(
console: Arc<Console>,
operation: F,
) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + Send + 'static,
{
// Control operations mutate durable state. Persist even on failure because
// an operation may have changed state before returning its error.
let operation_console = Arc::clone(&console);
let result = run_blocking(operation_console, operation).await;
console.flush_invocation_telemetry().await;
let persisted = console.persist_snapshot().await;
match result {
Ok(value) => {
persisted?;
Ok(value)
}
Err(error) => {
if let Err(persistence_error) = persisted {
tracing::error!(
error = %persistence_error,
"failed to persist console state after operation error"
);
}
Err(error)
}
}
}
enum GatewayError {
Console(ConsoleError),
UnsupportedMediaType,
PayloadTooLarge { limit: usize },
}
impl From<ConsoleError> for GatewayError {
fn from(error: ConsoleError) -> Self {
Self::Console(error)
}
}
impl IntoResponse for GatewayError {
fn into_response(self) -> Response {
let (status, code, message) = match self {
Self::UnsupportedMediaType => (
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"unsupported_media_type",
"Content-Type must be application/octet-stream".to_owned(),
),
Self::PayloadTooLarge { limit } => (
StatusCode::PAYLOAD_TOO_LARGE,
"payload_too_large",
format!("request body exceeds the {limit}-byte limit"),
),
Self::Console(error) => {
let (status, code, message) = match &error {
ConsoleError::InvalidRequest(_) => (
StatusCode::BAD_REQUEST,
"invalid_request",
error.to_string(),
),
ConsoleError::DeploymentNotFound(_)
| ConsoleError::ServiceNotFound(_)
| ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => (
StatusCode::NOT_FOUND,
"service_not_found",
"service is not deployed".to_owned(),
),
ConsoleError::RuntimeNotRunning
| ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
| ConsoleError::Runtime(RuntimeError::ActorStopped(_)) => (
StatusCode::SERVICE_UNAVAILABLE,
"service_unavailable",
"service is temporarily unavailable".to_owned(),
),
ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => (
StatusCode::TOO_MANY_REQUESTS,
"service_overloaded",
"service is overloaded".to_owned(),
),
ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) => (
StatusCode::PAYLOAD_TOO_LARGE,
"payload_too_large",
error.to_string(),
),
ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => (
StatusCode::GATEWAY_TIMEOUT,
"deadline_exceeded",
"service execution deadline exceeded".to_owned(),
),
ConsoleError::Runtime(
RuntimeError::OutputTooLarge { .. }
| RuntimeError::ActorFault { .. }
| RuntimeError::ComponentError { .. },
) => (StatusCode::BAD_GATEWAY, "service_error", error.to_string()),
ConsoleError::ArtifactTooLarge { .. }
| ConsoleError::ActiveDeployment(_)
| ConsoleError::Package(_)
| ConsoleError::WitRegistry(_)
| ConsoleError::Storage { .. }
| ConsoleError::ManifestSerialize(_)
| ConsoleError::Database(_)
| ConsoleError::KvBackendStart(_)
| ConsoleError::InvalidDatabaseData(_)
| ConsoleError::Runtime(_)
| ConsoleError::Join(_)
| ConsoleError::LockPoisoned => (
StatusCode::INTERNAL_SERVER_ERROR,
"internal_error",
"internal gateway error".to_owned(),
),
};
(status, code, message)
}
};
(
status,
Json(json!({
"error": {
"code": code,
"message": message,
}
})),
)
.into_response()
}
}
struct ApiError(ConsoleError);
impl From<ConsoleError> for ApiError {
fn from(error: ConsoleError) -> Self {
Self(error)
}
}
impl From<WitRegistryError> for ApiError {
fn from(error: WitRegistryError) -> Self {
Self(error.into())
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let (status, code) = match &self.0 {
ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"),
ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"),
ConsoleError::DeploymentNotFound(_) => (StatusCode::NOT_FOUND, "deployment_not_found"),
ConsoleError::ActiveDeployment(_) => (StatusCode::CONFLICT, "active_deployment"),
ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"),
ConsoleError::ArtifactTooLarge { .. }
| ConsoleError::Package(PackageError::ComponentTooLarge { .. })
| ConsoleError::WitRegistry(WitRegistryError::TooLarge { .. })
| ConsoleError::Runtime(RuntimeError::InputTooLarge { .. })
| ConsoleError::Runtime(RuntimeError::OutputTooLarge { .. }) => {
(StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large")
}
ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => {
(StatusCode::NOT_FOUND, "service_not_registered")
}
ConsoleError::Runtime(RuntimeError::ServiceAlreadyRegistered(_))
| ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
| ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => {
(StatusCode::CONFLICT, "runtime_conflict")
}
ConsoleError::WitRegistry(WitRegistryError::AlreadyExists(_)) => {
(StatusCode::CONFLICT, "wit_package_exists")
}
ConsoleError::WitRegistry(WitRegistryError::NotFound(_)) => {
(StatusCode::NOT_FOUND, "wit_package_not_found")
}
ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => {
(StatusCode::REQUEST_TIMEOUT, "deadline_exceeded")
}
ConsoleError::Runtime(
RuntimeError::InvalidServiceKey
| RuntimeError::InvalidManifest(_)
| RuntimeError::ManifestParse(_)
| RuntimeError::ComponentCompilation(_)
| RuntimeError::UnsupportedImport(_)
| RuntimeError::ActorInitialization(_)
| RuntimeError::ComponentError { .. },
) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_component"),
ConsoleError::Package(_) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_package"),
ConsoleError::WitRegistry(WitRegistryError::InvalidPackage(
WitPackageError::Source { .. }
| WitPackageError::Encode(_)
| WitPackageError::Decode(_)
| WitPackageError::MissingVersion(_),
)) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_wit_package"),
ConsoleError::Storage { .. }
| ConsoleError::ManifestSerialize(_)
| ConsoleError::Database(_)
| ConsoleError::KvBackendStart(_)
| ConsoleError::InvalidDatabaseData(_)
| ConsoleError::Runtime(_)
| ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_))
| ConsoleError::WitRegistry(WitRegistryError::Storage { .. })
| ConsoleError::WitRegistry(WitRegistryError::LockPoisoned)
| ConsoleError::Join(_)
| ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
};
(
status,
Json(json!({
"error": {
"code": code,
"message": self.0.to_string(),
}
})),
)
.into_response()
}
}
@@ -0,0 +1,70 @@
//! Bounded, asynchronous persistence for data-plane invocation telemetry.
use std::sync::Arc;
use tokio::sync::{Mutex, mpsc, oneshot};
use crate::persistence::{InvocationUpdate, Persistence};
const QUEUE_CAPACITY: usize = 1024;
const MAX_BATCH_SIZE: usize = 64;
/// Non-blocking producer used by Wasmtime invocation threads.
pub(crate) struct InvocationPersistence {
sender: mpsc::Sender<Command>,
}
enum Command {
Record(InvocationUpdate),
Flush(oneshot::Sender<()>),
}
impl InvocationPersistence {
/// Starts the single ordered writer on the current Tokio runtime.
pub(crate) fn start(persistence: Persistence, sync: Arc<Mutex<()>>) -> Self {
let (sender, mut receiver) = mpsc::channel(QUEUE_CAPACITY);
tokio::spawn(async move {
while let Some(first) = receiver.recv().await {
let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
let mut flushes = Vec::new();
match first {
Command::Record(update) => batch.push(update),
Command::Flush(flush) => flushes.push(flush),
}
while batch.len() < MAX_BATCH_SIZE {
match receiver.try_recv() {
Ok(Command::Record(update)) => batch.push(update),
Ok(Command::Flush(flush)) => flushes.push(flush),
Err(_) => break,
}
}
if !batch.is_empty() {
let _sync = sync.lock().await;
if let Err(error) = persistence.record_invocations(batch).await {
tracing::error!(%error, "failed to persist invocation telemetry");
}
}
for flush in flushes {
let _ = flush.send(());
}
}
});
Self { sender }
}
/// Enqueues telemetry without extending API response latency.
pub(crate) fn record(&self, update: InvocationUpdate) {
if let Err(error) = self.sender.try_send(Command::Record(update)) {
tracing::warn!(%error, "invocation telemetry queue is unavailable; update dropped");
}
}
/// Waits until all telemetry queued before this call has been attempted.
pub(crate) async fn flush(&self) {
let (sender, receiver) = oneshot::channel();
if self.sender.send(Command::Flush(sender)).await.is_ok() {
let _ = receiver.await;
}
}
}
+62 -662
View File
@@ -6,6 +6,8 @@
//! package Registry. Actor memory remains process-local and is never presented
//! as durable state.
mod api;
mod invocation_persistence;
mod kv_backend;
mod persistence;
mod wit_registry;
@@ -18,34 +20,21 @@ use std::{
time::{Instant, SystemTime, UNIX_EPOCH},
};
use axum::{
Json, Router,
body::{Body, Bytes},
extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection},
http::{HeaderMap, HeaderValue, Method, StatusCode, header},
response::{IntoResponse, Response},
routing::{delete, get, post},
};
use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use serde::{Deserialize, Serialize};
use serde_json::json;
use thiserror::Error;
use tower_http::{
cors::CorsLayer,
trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer},
};
use tracing::Level;
use wasmeld_package::{
ComponentPackage, PackageError, read_package,
wit_package::{WitPackageError, WitPackageMetadata},
ComponentPackage, PackageError, read_package, wit_package::WitPackageMetadata,
};
use wasmeld_runtime::{
CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey,
ServiceManifest,
};
pub use api::{app, gateway_app};
use invocation_persistence::InvocationPersistence;
use kv_backend::ToastyKvBackend;
use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService};
use persistence::{InvocationUpdate, Persistence, StoredDeployment, StoredEvent, StoredService};
use wit_registry::WitRegistry;
pub use wit_registry::WitRegistryError;
@@ -102,7 +91,8 @@ pub struct Console {
lifecycle: Mutex<()>,
state: Mutex<ConsoleState>,
persistence: Persistence,
persistence_sync: tokio::sync::Mutex<()>,
persistence_sync: Arc<tokio::sync::Mutex<()>>,
invocation_persistence: InvocationPersistence,
}
struct ConsoleState {
@@ -270,6 +260,19 @@ impl TryFrom<StoredEvent> for EventView {
}
}
impl From<EventView> for StoredEvent {
fn from(event: EventView) -> Self {
Self {
id: event.id,
timestamp_ms: event.timestamp_ms,
kind: event.kind.as_database_value().to_owned(),
service_id: event.service_id,
revision: event.revision,
message: event.message,
}
}
}
#[derive(Debug, Default, Deserialize)]
#[serde(deny_unknown_fields)]
struct StartRequest {
@@ -422,6 +425,14 @@ impl Console {
}
let persistence = Persistence::open(&config.database_path).await?;
let mut runtime_config = config.runtime;
if runtime_config.component_cache_dir.is_none() {
runtime_config.component_cache_dir = Some(
artifact_dir
.parent()
.unwrap_or(&artifact_dir)
.join("component-cache"),
);
}
if runtime_config.kv_backend.is_none() {
runtime_config.kv_backend = Some(Arc::new(
ToastyKvBackend::start(persistence.clone())
@@ -443,6 +454,9 @@ impl Console {
.collect::<Result<VecDeque<_>, _>>()?;
let runtime = Runtime::new(runtime_config.clone())?;
let persistence_sync = Arc::new(tokio::sync::Mutex::new(()));
let invocation_persistence =
InvocationPersistence::start(persistence.clone(), Arc::clone(&persistence_sync));
let console = Self {
runtime: RwLock::new(Some(runtime)),
runtime_config,
@@ -459,7 +473,8 @@ impl Console {
events,
}),
persistence,
persistence_sync: tokio::sync::Mutex::new(()),
persistence_sync,
invocation_persistence,
};
// The database is authoritative for metrics. Filesystem manifests are
// then used to recover artifacts that predate or missed a DB snapshot.
@@ -476,6 +491,11 @@ impl Console {
self.max_artifact_bytes
}
/// Flushes invocation telemetry accepted before this call.
pub async fn flush_invocation_telemetry(&self) {
self.invocation_persistence.flush().await;
}
/// Returns the configured binary WIT package upload limit.
pub fn max_wit_package_bytes(&self) -> usize {
self.wit_registry.max_package_bytes()
@@ -939,25 +959,34 @@ impl Console {
record.last_latency_ms = Some(latency_ms);
record.updated_at_ms = unix_time_ms();
match result {
Ok(output) => {
drop(state);
self.push_event(
let (failed, event_kind, event_message) = match &result {
Ok(_) => (
false,
EventKind::Invoked,
Some(key),
format!("invoke completed in {latency_ms} ms"),
)?;
Ok(InvocationResult { output, latency_ms })
}
),
Err(error) => {
record.errors = record.errors.saturating_add(1);
if matches!(error, RuntimeError::ActorFault { .. }) {
record.status = ServiceStatus::Faulted;
}
drop(state);
self.push_event(EventKind::Failed, Some(key), error.to_string())?;
Err(error.into())
(true, EventKind::Failed, error.to_string())
}
};
let updated_at_ms = record.updated_at_ms;
drop(state);
let event = self.push_event(event_kind, Some(key), event_message)?;
self.invocation_persistence.record(InvocationUpdate {
service_key: key.to_string(),
failed,
updated_at_ms,
latency_ms,
event: event.into(),
});
match result {
Ok(output) => Ok(InvocationResult { output, latency_ms }),
Err(error) => Err(error.into()),
}
}
@@ -1207,7 +1236,7 @@ impl Console {
kind: EventKind,
key: Option<&ServiceKey>,
message: String,
) -> Result<(), ConsoleError> {
) -> Result<EventView, ConsoleError> {
let mut state = self.state()?;
let event = EventView {
id: state.next_event_id,
@@ -1218,9 +1247,9 @@ impl Console {
message,
};
state.next_event_id = state.next_event_id.saturating_add(1);
state.events.push_front(event);
state.events.push_front(event.clone());
state.events.truncate(MAX_EVENTS);
Ok(())
Ok(event)
}
fn state(&self) -> Result<MutexGuard<'_, ConsoleState>, ConsoleError> {
@@ -1261,447 +1290,6 @@ fn deployment_view(
})
}
/// Builds the Axum management API around a shared [`Console`].
///
/// The returned router exposes Runtime control, component registration,
/// invocation, events, and WIT Registry endpoints.
pub fn app(console: Arc<Console>, allowed_origins: Vec<HeaderValue>) -> Router {
let max_body_bytes = console
.max_artifact_bytes()
.max(console.max_wit_package_bytes())
.saturating_add(1024 * 1024);
let mut cors = CorsLayer::new()
.allow_methods([Method::GET, Method::POST, Method::DELETE])
.allow_headers([header::CONTENT_TYPE]);
if !allowed_origins.is_empty() {
cors = cors.allow_origin(allowed_origins);
}
Router::new()
.route("/healthz", get(health))
.route("/api/v1/runtime", get(runtime_status))
.route("/api/v1/runtime/start", post(start_runtime))
.route("/api/v1/runtime/stop", post(stop_runtime))
.route("/api/v1/runtime/restart", post(restart_runtime))
.route("/api/v1/services", get(list_services).post(register))
.route(
"/api/v1/services/{id}/{revision}",
delete(unregister_service),
)
.route("/api/v1/deployments", get(list_deployments))
.route("/api/v1/deployments/{id}", get(get_deployment))
.route(
"/api/v1/deployments/{id}/activate",
post(activate_deployment),
)
.route(
"/api/v1/wit/packages",
get(list_wit_packages).post(publish_wit_package),
)
.route(
"/api/v1/wit/packages/{namespace}/{name}/{version}",
get(get_wit_package),
)
.route(
"/api/v1/wit/packages/{namespace}/{name}/{version}/content",
get(download_wit_package),
)
.route(
"/api/v1/services/{id}/{revision}/start",
post(start_service),
)
.route("/api/v1/services/{id}/{revision}/stop", post(stop_service))
.route(
"/api/v1/services/{id}/{revision}/restart",
post(restart_service),
)
.route(
"/api/v1/services/{id}/{revision}/invoke",
post(invoke_service),
)
.route("/api/v1/events", get(list_events))
.layer(DefaultBodyLimit::max(max_body_bytes))
.layer(cors)
.layer(
TraceLayer::new_for_http()
.make_span_with(DefaultMakeSpan::new().level(Level::INFO))
.on_response(DefaultOnResponse::new().level(Level::INFO)),
)
.with_state(console)
}
/// Builds the public data-plane API around the same resident Runtime.
///
/// This router intentionally contains no registration, lifecycle, Registry,
/// or deployment management routes.
pub fn gateway_app(console: Arc<Console>) -> Router {
let max_input_bytes = console.max_gateway_input_bytes();
Router::new()
.route("/healthz", get(health))
.route("/v1/services/{id}/invoke", post(gateway_invoke))
.layer(DefaultBodyLimit::max(max_input_bytes))
.layer(
TraceLayer::new_for_http()
.make_span_with(DefaultMakeSpan::new().level(Level::INFO))
.on_response(DefaultOnResponse::new().level(Level::INFO)),
)
.with_state(console)
}
async fn health() -> Json<serde_json::Value> {
Json(json!({ "status": "ok" }))
}
async fn gateway_invoke(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
headers: HeaderMap,
body: Result<Bytes, BytesRejection>,
) -> Result<Response, GatewayError> {
validate_path_segment("id", &service_id)?;
let content_type = headers
.get(header::CONTENT_TYPE)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.split(';').next())
.map(str::trim);
if !content_type.is_some_and(|value| value.eq_ignore_ascii_case("application/octet-stream")) {
return Err(GatewayError::UnsupportedMediaType);
}
let input = body.map_err(|_| GatewayError::PayloadTooLarge {
limit: console.max_gateway_input_bytes(),
})?;
let invocation = run_blocking(console, move |console| {
console.gateway_invoke(&service_id, input.to_vec())
})
.await?;
let mut response_headers = HeaderMap::new();
response_headers.insert(
header::CONTENT_TYPE,
HeaderValue::from_static("application/octet-stream"),
);
response_headers.insert(
"x-wasmeld-revision",
HeaderValue::from_str(&invocation.revision).map_err(|_| {
ConsoleError::InvalidRequest("revision is not a valid header".to_owned())
})?,
);
response_headers.insert(
"x-wasmeld-latency-ms",
HeaderValue::from_str(&invocation.result.latency_ms.to_string()).map_err(|_| {
ConsoleError::InvalidRequest("latency is not a valid header".to_owned())
})?,
);
Ok((response_headers, invocation.result.output).into_response())
}
async fn runtime_status(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(console.runtime_view()?))
}
async fn start_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.start_runtime()).await?,
))
}
async fn stop_runtime(State(console): State<Arc<Console>>) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.stop_runtime()).await?,
))
}
async fn restart_runtime(
State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> {
Ok(Json(
run_blocking(console, |console| console.restart_runtime()).await?,
))
}
async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<ServiceList>, ApiError> {
Ok(Json(ServiceList {
services: console.services()?,
}))
}
async fn list_deployments(
State(console): State<Arc<Console>>,
) -> Result<Json<DeploymentList>, ApiError> {
Ok(Json(DeploymentList {
deployments: console.deployments()?,
}))
}
async fn get_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
Ok(Json(console.deployment(&service_id)?))
}
async fn activate_deployment(
State(console): State<Arc<Console>>,
AxumPath(service_id): AxumPath<String>,
Json(request): Json<ActivateDeploymentRequest>,
) -> Result<Json<DeploymentView>, ApiError> {
validate_path_segment("id", &service_id)?;
validate_path_segment("revision", &request.revision)?;
let init_config = decode_optional_base64(request.init_config_base64.as_deref())?;
Ok(Json(
run_blocking(console, move |console| {
console.activate_deployment(service_id, request.revision, init_config)
})
.await?,
))
}
async fn list_events(State(console): State<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
Ok(Json(EventList {
events: console.events()?,
}))
}
async fn list_wit_packages(
State(console): State<Arc<Console>>,
) -> Result<Json<WitPackageList>, ApiError> {
Ok(Json(WitPackageList {
packages: console.wit_registry.list()?,
}))
}
async fn get_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Json<WitPackageMetadata>, ApiError> {
Ok(Json(
console
.wit_registry
.metadata(&format!("{namespace}:{name}"), &version)?,
))
}
async fn download_wit_package(
State(console): State<Arc<Console>>,
AxumPath((namespace, name, version)): AxumPath<(String, String, String)>,
) -> Result<Response, ApiError> {
let (metadata, bytes) = console
.wit_registry
.download(&format!("{namespace}:{name}"), &version)?;
Response::builder()
.status(StatusCode::OK)
.header(header::CONTENT_TYPE, "application/wasm")
.header(header::ETAG, format!("\"{}\"", metadata.sha256))
.body(Body::from(bytes))
.map_err(|error| {
ConsoleError::InvalidRequest(format!("failed to build package response: {error}"))
.into()
})
}
async fn publish_wit_package(
State(console): State<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<WitPackageMetadata>), ApiError> {
let mut package = None;
while let Some(field) = multipart
.next_field()
.await
.map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
{
match field.name() {
Some("package") => {
if package.is_some() {
return Err(ConsoleError::InvalidRequest(
"package field must occur exactly once".to_owned(),
)
.into());
}
package = Some(
field
.bytes()
.await
.map_err(|error| {
ConsoleError::InvalidRequest(format!(
"invalid WIT package field: {error}"
))
})?
.to_vec(),
);
}
Some(name) => {
return Err(ConsoleError::InvalidRequest(format!(
"unsupported multipart field {name}"
))
.into());
}
None => {}
}
}
let package = package
.ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
if package.len() > console.max_wit_package_bytes() {
return Err(WitRegistryError::TooLarge {
limit: console.max_wit_package_bytes(),
}
.into());
}
let metadata = run_blocking(console, move |console| {
Ok(console.wit_registry.publish(&package)?)
})
.await?;
Ok((StatusCode::CREATED, Json(metadata)))
}
async fn register(
State(console): State<Arc<Console>>,
mut multipart: Multipart,
) -> Result<(StatusCode, Json<ServiceView>), ApiError> {
let mut package = None;
while let Some(field) = multipart
.next_field()
.await
.map_err(|error| ConsoleError::InvalidRequest(format!("invalid multipart body: {error}")))?
{
match field.name() {
Some("package") => {
if package.is_some() {
return Err(ConsoleError::InvalidRequest(
"package field must occur exactly once".to_owned(),
)
.into());
}
package = Some(
field
.bytes()
.await
.map_err(|error| {
ConsoleError::InvalidRequest(format!("invalid package field: {error}"))
})?
.to_vec(),
);
}
Some(name) => {
return Err(ConsoleError::InvalidRequest(format!(
"unsupported multipart field {name}"
))
.into());
}
None => {}
}
}
let package = package
.ok_or_else(|| ConsoleError::InvalidRequest("package field is required".to_owned()))?;
let service = run_blocking(console, move |console| console.register_package(package)).await?;
Ok((StatusCode::CREATED, Json(service)))
}
async fn start_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
let init_config = decode_optional_base64(
request
.as_ref()
.and_then(|Json(request)| request.init_config_base64.as_deref()),
)?;
Ok(Json(
run_blocking(console, move |console| console.start(&key, init_config)).await?,
))
}
async fn stop_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
Ok(Json(
run_blocking(console, move |console| console.stop(&key)).await?,
))
}
async fn unregister_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
Ok(Json(
run_blocking(console, move |console| console.unregister(&key)).await?,
))
}
async fn restart_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
request: Option<Json<StartRequest>>,
) -> Result<Json<ServiceView>, ApiError> {
let key = api_key(id, revision)?;
let init_config = decode_optional_base64(
request
.as_ref()
.and_then(|Json(request)| request.init_config_base64.as_deref()),
)?;
Ok(Json(
run_blocking(console, move |console| console.restart(&key, init_config)).await?,
))
}
async fn invoke_service(
State(console): State<Arc<Console>>,
AxumPath((id, revision)): AxumPath<(String, String)>,
Json(request): Json<InvokeRequest>,
) -> Result<Json<InvokeResponse>, ApiError> {
let key = api_key(id, revision)?;
let input = BASE64.decode(request.input_base64).map_err(|error| {
ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}"))
})?;
let result = run_blocking(console, move |console| console.invoke(&key, input)).await?;
Ok(Json(InvokeResponse {
output_base64: BASE64.encode(&result.output),
output_bytes: result.output.len(),
latency_ms: result.latency_ms,
}))
}
async fn run_blocking<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
where
T: Send + 'static,
F: FnOnce(Arc<Console>) -> Result<T, ConsoleError> + Send + 'static,
{
// Wasmtime compilation and synchronous Actor calls must not occupy a Tokio
// async worker. Persist afterward even on failure because an operation may
// already have recorded metrics or an event before returning its error.
let operation_console = Arc::clone(&console);
let result = tokio::task::spawn_blocking(move || operation(operation_console))
.await
.map_err(ConsoleError::from)
.and_then(|result| result);
let persisted = console.persist_snapshot().await;
match result {
Ok(value) => {
persisted?;
Ok(value)
}
Err(error) => {
if let Err(persistence_error) = persisted {
tracing::error!(
error = %persistence_error,
"failed to persist console state after operation error"
);
}
Err(error)
}
}
}
fn api_key(id: String, revision: String) -> Result<ServiceKey, ConsoleError> {
validate_path_segment("id", &id)?;
validate_path_segment("revision", &revision)?;
@@ -1762,191 +1350,3 @@ fn unix_time_ms() -> u64 {
fn duration_ms(duration: std::time::Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
enum GatewayError {
Console(ConsoleError),
UnsupportedMediaType,
PayloadTooLarge { limit: usize },
}
impl From<ConsoleError> for GatewayError {
fn from(error: ConsoleError) -> Self {
Self::Console(error)
}
}
impl IntoResponse for GatewayError {
fn into_response(self) -> Response {
let (status, code, message) = match self {
Self::UnsupportedMediaType => (
StatusCode::UNSUPPORTED_MEDIA_TYPE,
"unsupported_media_type",
"Content-Type must be application/octet-stream".to_owned(),
),
Self::PayloadTooLarge { limit } => (
StatusCode::PAYLOAD_TOO_LARGE,
"payload_too_large",
format!("request body exceeds the {limit}-byte limit"),
),
Self::Console(error) => {
let (status, code, message) = match &error {
ConsoleError::InvalidRequest(_) => (
StatusCode::BAD_REQUEST,
"invalid_request",
error.to_string(),
),
ConsoleError::DeploymentNotFound(_)
| ConsoleError::ServiceNotFound(_)
| ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => (
StatusCode::NOT_FOUND,
"service_not_found",
"service is not deployed".to_owned(),
),
ConsoleError::RuntimeNotRunning
| ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
| ConsoleError::Runtime(RuntimeError::ActorStopped(_)) => (
StatusCode::SERVICE_UNAVAILABLE,
"service_unavailable",
"service is temporarily unavailable".to_owned(),
),
ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => (
StatusCode::TOO_MANY_REQUESTS,
"service_overloaded",
"service is overloaded".to_owned(),
),
ConsoleError::Runtime(RuntimeError::InputTooLarge { .. }) => (
StatusCode::PAYLOAD_TOO_LARGE,
"payload_too_large",
error.to_string(),
),
ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => (
StatusCode::GATEWAY_TIMEOUT,
"deadline_exceeded",
"service execution deadline exceeded".to_owned(),
),
ConsoleError::Runtime(
RuntimeError::OutputTooLarge { .. }
| RuntimeError::ActorFault { .. }
| RuntimeError::ComponentError { .. },
) => (StatusCode::BAD_GATEWAY, "service_error", error.to_string()),
ConsoleError::ArtifactTooLarge { .. }
| ConsoleError::ActiveDeployment(_)
| ConsoleError::Package(_)
| ConsoleError::WitRegistry(_)
| ConsoleError::Storage { .. }
| ConsoleError::ManifestSerialize(_)
| ConsoleError::Database(_)
| ConsoleError::KvBackendStart(_)
| ConsoleError::InvalidDatabaseData(_)
| ConsoleError::Runtime(_)
| ConsoleError::Join(_)
| ConsoleError::LockPoisoned => (
StatusCode::INTERNAL_SERVER_ERROR,
"internal_error",
"internal gateway error".to_owned(),
),
};
(status, code, message)
}
};
(
status,
Json(json!({
"error": {
"code": code,
"message": message,
}
})),
)
.into_response()
}
}
struct ApiError(ConsoleError);
impl From<ConsoleError> for ApiError {
fn from(error: ConsoleError) -> Self {
Self(error)
}
}
impl From<WitRegistryError> for ApiError {
fn from(error: WitRegistryError) -> Self {
Self(error.into())
}
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let (status, code) = match &self.0 {
ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"),
ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"),
ConsoleError::DeploymentNotFound(_) => (StatusCode::NOT_FOUND, "deployment_not_found"),
ConsoleError::ActiveDeployment(_) => (StatusCode::CONFLICT, "active_deployment"),
ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"),
ConsoleError::ArtifactTooLarge { .. }
| ConsoleError::Package(PackageError::ComponentTooLarge { .. })
| ConsoleError::WitRegistry(WitRegistryError::TooLarge { .. })
| ConsoleError::Runtime(RuntimeError::InputTooLarge { .. })
| ConsoleError::Runtime(RuntimeError::OutputTooLarge { .. }) => {
(StatusCode::PAYLOAD_TOO_LARGE, "payload_too_large")
}
ConsoleError::Runtime(RuntimeError::ServiceNotRegistered(_)) => {
(StatusCode::NOT_FOUND, "service_not_registered")
}
ConsoleError::Runtime(RuntimeError::ServiceAlreadyRegistered(_))
| ConsoleError::Runtime(RuntimeError::ActorUnavailable(_))
| ConsoleError::Runtime(RuntimeError::ActorOverloaded(_)) => {
(StatusCode::CONFLICT, "runtime_conflict")
}
ConsoleError::WitRegistry(WitRegistryError::AlreadyExists(_)) => {
(StatusCode::CONFLICT, "wit_package_exists")
}
ConsoleError::WitRegistry(WitRegistryError::NotFound(_)) => {
(StatusCode::NOT_FOUND, "wit_package_not_found")
}
ConsoleError::Runtime(RuntimeError::DeadlineExceeded { .. }) => {
(StatusCode::REQUEST_TIMEOUT, "deadline_exceeded")
}
ConsoleError::Runtime(
RuntimeError::InvalidServiceKey
| RuntimeError::InvalidManifest(_)
| RuntimeError::ManifestParse(_)
| RuntimeError::ComponentCompilation(_)
| RuntimeError::UnsupportedImport(_)
| RuntimeError::ActorInitialization(_)
| RuntimeError::ComponentError { .. },
) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_component"),
ConsoleError::Package(_) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_package"),
ConsoleError::WitRegistry(WitRegistryError::InvalidPackage(
WitPackageError::Source { .. }
| WitPackageError::Encode(_)
| WitPackageError::Decode(_)
| WitPackageError::MissingVersion(_),
)) => (StatusCode::UNPROCESSABLE_ENTITY, "invalid_wit_package"),
ConsoleError::Storage { .. }
| ConsoleError::ManifestSerialize(_)
| ConsoleError::Database(_)
| ConsoleError::KvBackendStart(_)
| ConsoleError::InvalidDatabaseData(_)
| ConsoleError::Runtime(_)
| ConsoleError::WitRegistry(WitRegistryError::InvalidStorage(_))
| ConsoleError::WitRegistry(WitRegistryError::Storage { .. })
| ConsoleError::WitRegistry(WitRegistryError::LockPoisoned)
| ConsoleError::Join(_)
| ConsoleError::LockPoisoned => (StatusCode::INTERNAL_SERVER_ERROR, "internal_error"),
};
(
status,
Json(json!({
"error": {
"code": code,
"message": self.0.to_string(),
}
})),
)
.into_response()
}
}
+2 -1
View File
@@ -42,7 +42,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
.await?,
);
let management = app(Arc::clone(&console), allowed_origins);
let gateway = gateway_app(console);
let gateway = gateway_app(Arc::clone(&console));
let management_listener = tokio::net::TcpListener::bind(management_address).await?;
let gateway_listener = tokio::net::TcpListener::bind(gateway_address).await?;
let (shutdown_sender, shutdown_receiver) = tokio::sync::watch::channel(false);
@@ -61,6 +61,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
);
signal_task.abort();
result?;
console.flush_invocation_telemetry().await;
Ok(())
}
+79 -5
View File
@@ -54,6 +54,15 @@ pub(crate) struct StoredKvEntry {
pub updated_at_ms: u64,
}
/// One invocation's durable metric delta and event.
pub(crate) struct InvocationUpdate {
pub service_key: String,
pub failed: bool,
pub updated_at_ms: u64,
pub latency_ms: u64,
pub event: StoredEvent,
}
/// Serialized access to the Toasty database connection.
#[derive(Clone)]
pub(crate) struct Persistence {
@@ -131,24 +140,42 @@ impl Persistence {
let mut db = self.db.lock().await;
let mut tx = db.transaction().await?;
let stored_services = StoredService::all().exec(&mut tx).await?;
let service_keys = services
.iter()
.map(|service| service.service_key.clone())
.collect::<BTreeSet<_>>();
for stored in StoredService::all().exec(&mut tx).await? {
let existing_service_keys = stored_services
.iter()
.map(|service| service.service_key.clone())
.collect::<BTreeSet<_>>();
for stored in stored_services {
if !service_keys.contains(&stored.service_key) {
StoredService::delete_by_service_key(&mut tx, &stored.service_key).await?;
}
}
for service in services {
StoredService::upsert_by_service_key(service.service_key)
if existing_service_keys.contains(&service.service_key) {
// Invocation metrics are written as ordered deltas by
// `record_invocations`; control snapshots must not overwrite
// or double-count those concurrent updates.
StoredService::filter_by_service_key(&service.service_key)
.update()
.manifest_toml(service.manifest_toml)
.updated_at_ms(service.updated_at_ms)
.calls(service.calls)
.errors(service.errors)
.last_latency_ms(service.last_latency_ms)
.exec(&mut tx)
.await?;
} else {
StoredService::create()
.service_key(service.service_key)
.manifest_toml(service.manifest_toml)
.updated_at_ms(service.updated_at_ms)
.calls(0)
.errors(0)
.last_latency_ms(None)
.exec(&mut tx)
.await?;
}
}
if let Some(oldest_event) = events.last() {
@@ -189,6 +216,53 @@ impl Persistence {
tx.commit().await
}
/// Atomically records invocation deltas and their bounded event history.
pub(crate) async fn record_invocations(
&self,
updates: Vec<InvocationUpdate>,
) -> toasty::Result<()> {
let mut db = self.db.lock().await;
let mut tx = db.transaction().await?;
let mut newest_event_id = 0;
for update in updates {
if let Some(stored) = StoredService::filter_by_service_key(&update.service_key)
.first()
.exec(&mut tx)
.await?
{
StoredService::filter_by_service_key(&update.service_key)
.update()
.calls(stored.calls.saturating_add(1))
.errors(stored.errors.saturating_add(u64::from(update.failed)))
.updated_at_ms(update.updated_at_ms)
.last_latency_ms(Some(update.latency_ms))
.exec(&mut tx)
.await?;
}
newest_event_id = newest_event_id.max(update.event.id);
StoredEvent::upsert_by_id(update.event.id)
.timestamp_ms(update.event.timestamp_ms)
.kind(update.event.kind)
.service_id(update.event.service_id)
.revision(update.event.revision)
.message(update.event.message)
.exec(&mut tx)
.await?;
}
let oldest_retained_event_id = newest_event_id.saturating_sub(255);
if oldest_retained_event_id > 0 {
StoredEvent::filter(StoredEvent::fields().id().lt(oldest_retained_event_id))
.delete()
.exec(&mut tx)
.await?;
}
tx.commit().await
}
pub(crate) async fn kv_get(
&self,
service_id: &str,
+7
View File
@@ -408,6 +408,7 @@ async fn reloads_manifests_without_restoring_actor_memory() {
assert_eq!(response.status(), StatusCode::OK);
let response = application
.clone()
.oneshot(json_request(
"/api/v1/services/persisted/0.1.0/invoke",
json!({ "input_base64": BASE64.encode(b"persist metrics") }),
@@ -415,6 +416,12 @@ async fn reloads_manifests_without_restoring_actor_memory() {
.await
.expect("invoke request should complete");
assert_eq!(response.status(), StatusCode::OK);
let response = application
.oneshot(empty_post("/api/v1/services/persisted/0.1.0/stop"))
.await
.expect("stop request should flush control state");
assert_eq!(response.status(), StatusCode::OK);
}
let application = test_app(artifact_dir.path()).await;
+3
View File
@@ -80,6 +80,9 @@ cargo +stable run -p wasmeld-package --bin wasmeld -- \
默认服务 ID 为 `<id>-dev`Revision 为 `<version>-dev.h<component-hash>`。正式服务不会
被覆盖。构建、校验或启动失败时不会切换 Deployment,上一版本继续提供服务。
编译过程中再次保存源码会在当前构建结束后立即触发下一轮构建。若 Component 已构建
成功但 Console 暂时不可用,开发循环会保留该 `.wasmpkg` 并重试部署,不会等待新的
源码变化,也不会为每次重试重复编译。
管理面默认每 5 秒刷新一次,也可以直接调用开发服务:
+96 -82
View File
@@ -20,6 +20,7 @@ use crate::{BuildProfile, PackRequest, PackageIdentity, PackedComponent, pack_co
const DEFAULT_CONSOLE: &str = "http://127.0.0.1:8080";
const DEFAULT_POLL_MS: u64 = 350;
const DEPLOY_RETRY_INTERVAL: Duration = Duration::from_secs(1);
struct DevOptions {
manifest_path: PathBuf,
@@ -42,25 +43,21 @@ struct ServiceIdentity {
revision: String,
}
#[derive(Debug, Deserialize)]
struct Deployment {
active_revision: String,
}
pub(crate) fn run(arguments: Vec<String>) -> Result<(), Box<dyn std::error::Error>> {
let options = parse_options(arguments)?;
let manifest_path = fs::canonicalize(&options.manifest_path)?;
let watch_roots = watch_roots(&manifest_path)?;
let client = Client::builder()
.user_agent(format!("wasmeld-dev/{}", env!("CARGO_PKG_VERSION")))
.connect_timeout(Duration::from_secs(2))
.timeout(Duration::from_secs(120))
.build()?;
ensure_console(&client, &options.console)?;
let temporary_dir = env::temp_dir().join(format!("wasmeld-dev-{}", std::process::id()));
fs::create_dir_all(&temporary_dir)?;
let output = temporary_dir.join("component.wasmpkg");
let mut observed = None;
let mut pending_deployment = None;
println!("watching:");
for root in &watch_roots {
@@ -70,16 +67,29 @@ pub(crate) fn run(arguments: Vec<String>) -> Result<(), Box<dyn std::error::Erro
loop {
let current = source_fingerprint(&watch_roots)?;
if observed.as_ref() == Some(&current) {
thread::sleep(options.poll_interval);
continue;
if observed.as_ref() != Some(&current) {
// Record the fingerprint that started this build. A save that
// arrives during compilation will differ on the next loop and
// immediately queue another build.
observed = Some(current);
pending_deployment = None;
println!("change detected; building Component");
match build_component(&options, &manifest_path, &output) {
Ok(packed) => pending_deployment = Some(packed),
Err(error) if options.once => return Err(error),
Err(error) => {
eprintln!("wasmeld dev: {error}");
eprintln!(
"waiting for the next source change; the previous deployment is intact"
);
}
}
}
println!("change detected; building Component");
let result = build_and_deploy(&client, &options, &manifest_path, &output);
observed = Some(source_fingerprint(&watch_roots)?);
match result {
Ok(packed) => {
if let Some(packed) = pending_deployment.as_ref() {
match deploy_component(&client, &options.console, packed) {
Ok(()) => {
let packed = pending_deployment.take().expect("pending deployment");
println!(
"ready: {}@{} ({})",
packed.manifest.id, packed.manifest.revision, packed.manifest.sha256
@@ -91,10 +101,60 @@ pub(crate) fn run(arguments: Vec<String>) -> Result<(), Box<dyn std::error::Erro
Err(error) if options.once => return Err(error),
Err(error) => {
eprintln!("wasmeld dev: {error}");
eprintln!("waiting for the next source change; the previous deployment is intact");
eprintln!("deployment will retry; the previous deployment is intact");
thread::sleep(DEPLOY_RETRY_INTERVAL);
continue;
}
}
}
thread::sleep(options.poll_interval);
}
}
fn build_component(
options: &DevOptions,
manifest_path: &Path,
output: &Path,
) -> Result<PackedComponent, Box<dyn std::error::Error>> {
pack_component(PackRequest {
manifest_path: manifest_path.to_owned(),
output: Some(output.to_owned()),
no_build: false,
locked: options.locked,
profile: if options.release {
BuildProfile::Release
} else {
BuildProfile::Debug
},
identity: PackageIdentity::Development {
id: options.service_id.clone(),
},
})
}
fn deploy_component(
client: &Client,
console: &str,
packed: &PackedComponent,
) -> Result<(), Box<dyn std::error::Error>> {
register(client, console, packed)?;
activate(
client,
console,
&packed.manifest.id,
&packed.manifest.revision,
)?;
for revision in stale_development_revisions(client, console, packed)? {
if let Err(error) = unregister(client, console, &packed.manifest.id, &revision) {
eprintln!(
"wasmeld dev: deployed successfully but could not unregister {}@{}: {error}",
packed.manifest.id, revision
);
}
}
Ok(())
}
fn parse_options(arguments: Vec<String>) -> Result<DevOptions, Box<dyn std::error::Error>> {
@@ -142,70 +202,21 @@ fn parse_options(arguments: Vec<String>) -> Result<DevOptions, Box<dyn std::erro
})
}
fn build_and_deploy(
client: &Client,
options: &DevOptions,
manifest_path: &Path,
output: &Path,
) -> Result<PackedComponent, Box<dyn std::error::Error>> {
let packed = pack_component(PackRequest {
manifest_path: manifest_path.to_owned(),
output: Some(output.to_owned()),
no_build: false,
locked: options.locked,
profile: if options.release {
BuildProfile::Release
} else {
BuildProfile::Debug
},
identity: PackageIdentity::Development {
id: options.service_id.clone(),
},
})?;
let previous = active_revision(client, &options.console, &packed.manifest.id)?;
register(client, &options.console, &packed)?;
activate(
client,
&options.console,
&packed.manifest.id,
&packed.manifest.revision,
)?;
if let Some(previous) = previous
&& previous != packed.manifest.revision
&& is_matching_dev_revision(&previous, &packed.manifest.revision)
&& let Err(error) = unregister(client, &options.console, &packed.manifest.id, &previous)
{
eprintln!(
"wasmeld dev: deployed successfully but could not unregister {}@{}: {error}",
packed.manifest.id, previous
);
}
Ok(packed)
}
fn ensure_console(client: &Client, console: &str) -> Result<(), Box<dyn std::error::Error>> {
let response = client.get(format!("{console}/healthz")).send()?;
if response.status().is_success() {
Ok(())
} else {
Err(http_error("Console health check", response))
}
}
fn active_revision(
fn stale_development_revisions(
client: &Client,
console: &str,
service_id: &str,
) -> Result<Option<String>, Box<dyn std::error::Error>> {
let response = client
.get(format!("{console}/api/v1/deployments/{service_id}"))
.send()?;
match response.status() {
status if status.is_success() => Ok(Some(response.json::<Deployment>()?.active_revision)),
StatusCode::NOT_FOUND => Ok(None),
_ => Err(http_error("deployment lookup", response)),
}
packed: &PackedComponent,
) -> Result<Vec<String>, Box<dyn std::error::Error>> {
Ok(services(client, console)?
.services
.into_iter()
.filter(|service| {
service.id == packed.manifest.id
&& service.revision != packed.manifest.revision
&& is_matching_dev_revision(&service.revision, &packed.manifest.revision)
})
.map(|service| service.revision)
.collect())
}
fn register(
@@ -243,15 +254,18 @@ fn service_exists(
service_id: &str,
revision: &str,
) -> Result<bool, Box<dyn std::error::Error>> {
Ok(services(client, console)?
.services
.iter()
.any(|service| service.id == service_id && service.revision == revision))
}
fn services(client: &Client, console: &str) -> Result<ServiceList, Box<dyn std::error::Error>> {
let response = client.get(format!("{console}/api/v1/services")).send()?;
if !response.status().is_success() {
return Err(http_error("service lookup", response));
}
Ok(response
.json::<ServiceList>()?
.services
.iter()
.any(|service| service.id == service_id && service.revision == revision))
Ok(response.json()?)
}
fn activate(
+11 -1
View File
@@ -1,6 +1,6 @@
//! Runtime error taxonomy exposed to management layers.
use std::time::Duration;
use std::{path::PathBuf, time::Duration};
use thiserror::Error;
@@ -36,6 +36,16 @@ pub enum RuntimeError {
#[error("runtime creation failed: {0}")]
RuntimeCreation(#[source] wasmtime::Error),
#[error("failed to prepare Component compilation cache {path}: {source}")]
CacheDirectory {
path: PathBuf,
#[source]
source: std::io::Error,
},
#[error("invalid Component compilation cache configuration: {0}")]
CacheConfiguration(#[source] wasmtime::Error),
#[error("failed to read component artifact {path}: {source}")]
ArtifactRead {
path: String,
+20 -1
View File
@@ -8,6 +8,7 @@
use std::{
collections::HashMap,
fs,
path::PathBuf,
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
@@ -18,7 +19,7 @@ use std::{
};
use wasmtime::{
Config, Engine, Store, StoreLimits, StoreLimitsBuilder,
Cache, CacheConfig, Config, Engine, Store, StoreLimits, StoreLimitsBuilder,
component::{Component, HasData, Linker, ResourceTable},
};
use wasmtime_wasi::{
@@ -61,6 +62,8 @@ pub struct RuntimeConfig {
pub epoch_tick: Duration,
/// Maximum native stack reservation for WebAssembly execution.
pub max_wasm_stack: usize,
/// Optional persistent cache for compiled Component machine code.
pub component_cache_dir: Option<PathBuf>,
/// Service-scoped storage used by Components importing the KV capability.
pub kv_backend: Option<Arc<dyn KvBackend>>,
}
@@ -70,6 +73,7 @@ impl Default for RuntimeConfig {
Self {
epoch_tick: DEFAULT_EPOCH_TICK,
max_wasm_stack: DEFAULT_MAX_WASM_STACK,
component_cache_dir: None,
kv_backend: None,
}
}
@@ -180,6 +184,21 @@ impl Runtime {
wasmtime_config.consume_fuel(true);
wasmtime_config.epoch_interruption(true);
wasmtime_config.max_wasm_stack(config.max_wasm_stack);
if let Some(cache_dir) = &config.component_cache_dir {
fs::create_dir_all(cache_dir).map_err(|source| RuntimeError::CacheDirectory {
path: cache_dir.clone(),
source,
})?;
let cache_dir =
fs::canonicalize(cache_dir).map_err(|source| RuntimeError::CacheDirectory {
path: cache_dir.clone(),
source,
})?;
let mut cache_config = CacheConfig::new();
cache_config.with_directory(cache_dir);
let cache = Cache::new(cache_config).map_err(RuntimeError::CacheConfiguration)?;
wasmtime_config.cache(Some(cache));
}
let engine =
Arc::new(Engine::new(&wasmtime_config).map_err(RuntimeError::RuntimeCreation)?);