Compare commits

...

8 Commits

Author SHA1 Message Date
Maofeng 72c901c478 fix(console-ui): render binary invocation output safely
- decode UTF-8 responses in fatal mode and reject control-byte payloads
- fall back to hexadecimal output instead of replacement characters
- show response byte count, selected representation, automatic fallback, and latency
- preserve the existing counter u64 representation and cover the wiring in SSR tests
2026-07-29 19:20:15 +08:00
Maofeng 51b452ecc5 docs: explain host capability discovery
- document exact-version import resolution and deny-by-default behavior
- clarify that Component imports remain the capability source of truth
- describe structured capability metadata returned by the Console API
2026-07-29 19:13:51 +08:00
Maofeng bdbf868f75 feat(console-ui): show component host capabilities
- consume structured capability metadata from service responses
- add a stable-width Host Capability column to the full service table
- preserve compact overview density and horizontal overflow on narrow screens
- cover API typing and capability rendering wiring in SSR source tests
2026-07-29 19:13:45 +08:00
Maofeng 72a264e0d1 feat(console): expose component capabilities
- project structured capability package, name, version, and interface fields
- derive capabilities during registration and artifact restoration
- keep Component imports as the source of truth without changing .wasmpkg
- verify empty and Clock capability responses through the management API
2026-07-29 19:13:41 +08:00
Maofeng 5b32886efe feat(runtime): modularize host capabilities
- add an exact-version Capability Registry and structured descriptors
- move Clock bindings, host state, and Linker installation into its own module
- derive required capabilities from Component imports and link only requested interfaces
- cover exact-version resolution and minimal per-component capability sets
2026-07-29 19:13:35 +08:00
Maofeng 3f5bb344f7 docs: document gateway deployment workflow
- explain the separate control-plane and data-plane listeners
- document revision activation and raw binary invocation
- clarify restart restoration and actor-memory semantics
2026-07-29 16:32:21 +08:00
Maofeng 9160362fc0 feat(console-ui): manage active deployments
- load deployment state with the console snapshot
- mark externally routed revisions in service tables
- add guarded activation confirmation and pending states
- cover deployment API wiring in SSR tests
2026-07-29 16:32:16 +08:00
Maofeng 1d3fac4be8 feat(console): add deployment-routed gateway
- persist service-to-revision deployments in libSQL with an additive migration
- restore active actors and atomically switch revisions
- expose an isolated raw-byte gateway on a second listener with stable errors
- cover routing, isolation, revision switching, and restart restoration
2026-07-29 16:32:12 +08:00
17 changed files with 1631 additions and 129 deletions
+21
View File
@@ -4,6 +4,10 @@ Wasmeld 是面向 WebAssembly Component 的服务运行与管理平台。内部
Component,在受限 Wasmtime Sandbox 中以常驻 Actor 运行,并通过稳定 WIT 契约 Component,在受限 Wasmtime Sandbox 中以常驻 Actor 运行,并通过稳定 WIT 契约
暴露能力。 暴露能力。
Runtime 内置版本化 Host Capability Registry。注册 Component 时会直接读取其 WIT
imports,只链接实际请求且版本完全匹配的 Host 能力;未知能力会被拒绝。能力不需要在
`.wasmpkg` 中重复声明,管理面会展示每个服务版本解析出的完整接口标识。
## 结构 ## 结构
```text ```text
@@ -23,6 +27,23 @@ wit/ Wasmeld WIT package 源码
cargo +stable run -p wasmeld-console cargo +stable run -p wasmeld-console
``` ```
后端会启动本地管理 API `127.0.0.1:8080` 和公开数据面
`0.0.0.0:8081`。注册 Component 后,通过管理 API 激活一个版本:
```bash
curl -X POST http://127.0.0.1:8080/api/v1/deployments/echo/activate \
-H 'Content-Type: application/json' \
-d '{"revision":"0.1.0"}'
```
客户端只访问 Gateway,并以原始二进制请求调用活动版本:
```bash
curl http://127.0.0.1:8081/v1/services/echo/invoke \
-H 'Content-Type: application/octet-stream' \
--data-binary 'hello'
```
管理面: 管理面:
```bash ```bash
+30 -2
View File
@@ -4,11 +4,19 @@ const API_BASE_STORAGE_KEY = "wasmeld-api-base";
export type BackendServiceStatus = "running" | "stopped" | "faulted"; export type BackendServiceStatus = "running" | "stopped" | "faulted";
export type BackendCapability = {
interface: string;
package: string;
name: string;
version: string;
};
export type BackendService = { export type BackendService = {
id: string; id: string;
revision: string; revision: string;
artifact: string; artifact: string;
world: string; world: string;
capabilities: BackendCapability[];
status: BackendServiceStatus; status: BackendServiceStatus;
updated_at_ms: number; updated_at_ms: number;
limits: { limits: {
@@ -27,7 +35,7 @@ export type BackendService = {
export type BackendEvent = { export type BackendEvent = {
id: number; id: number;
timestamp_ms: number; timestamp_ms: number;
kind: "registered" | "started" | "stopped" | "invoked" | "failed"; kind: "registered" | "deployed" | "started" | "stopped" | "invoked" | "failed";
service_id: string | null; service_id: string | null;
revision: string | null; revision: string | null;
message: string; message: string;
@@ -41,6 +49,13 @@ export type BackendRuntime = {
running_services: number; running_services: number;
}; };
export type BackendDeployment = {
service_id: string;
active_revision: string;
status: BackendServiceStatus;
updated_at_ms: number;
};
export type BackendWitDependency = { export type BackendWitDependency = {
name: string; name: string;
version: string; version: string;
@@ -80,20 +95,33 @@ export function saveApiBase(value: string): string {
} }
export async function fetchSnapshot(apiBase: string) { export async function fetchSnapshot(apiBase: string) {
const [services, events, runtime, witPackages] = await Promise.all([ const [services, deployments, events, runtime, witPackages] = await Promise.all([
request<{ services: BackendService[] }>(apiBase, "/api/v1/services"), request<{ services: BackendService[] }>(apiBase, "/api/v1/services"),
request<{ deployments: BackendDeployment[] }>(apiBase, "/api/v1/deployments"),
request<{ events: BackendEvent[] }>(apiBase, "/api/v1/events"), request<{ events: BackendEvent[] }>(apiBase, "/api/v1/events"),
request<BackendRuntime>(apiBase, "/api/v1/runtime"), request<BackendRuntime>(apiBase, "/api/v1/runtime"),
request<{ packages: BackendWitPackage[] }>(apiBase, "/api/v1/wit/packages"), request<{ packages: BackendWitPackage[] }>(apiBase, "/api/v1/wit/packages"),
]); ]);
return { return {
services: services.services, services: services.services,
deployments: deployments.deployments,
events: events.events, events: events.events,
runtime, runtime,
witPackages: witPackages.packages, witPackages: witPackages.packages,
}; };
} }
export async function activateDeployment(
apiBase: string,
service: Pick<BackendService, "id" | "revision">,
): Promise<BackendDeployment> {
return request(apiBase, `/api/v1/deployments/${encodeURIComponent(service.id)}/activate`, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ revision: service.revision }),
});
}
export async function registerComponent( export async function registerComponent(
apiBase: string, apiBase: string,
input: RegisterComponentInput, input: RegisterComponentInput,
+289 -20
View File
@@ -18,6 +18,7 @@ import {
LayoutDashboard, LayoutDashboard,
Package, Package,
Play, Play,
RadioTower,
RefreshCw, RefreshCw,
RotateCcw, RotateCcw,
Search, Search,
@@ -30,12 +31,15 @@ import {
} from "lucide-react"; } from "lucide-react";
import { FormEvent, ReactNode, useCallback, useEffect, useMemo, useState } from "react"; import { FormEvent, ReactNode, useCallback, useEffect, useMemo, useState } from "react";
import { import {
BackendCapability,
BackendEvent, BackendEvent,
BackendDeployment,
BackendRuntime, BackendRuntime,
BackendService, BackendService,
BackendWitPackage, BackendWitPackage,
InvokeResult, InvokeResult,
RegisterComponentInput, RegisterComponentInput,
activateDeployment,
changeRuntimeState, changeRuntimeState,
changeServiceState, changeServiceState,
fetchSnapshot, fetchSnapshot,
@@ -63,9 +67,11 @@ type Service = {
deadlineMs: number; deadlineMs: number;
mailbox: number; mailbox: number;
maxInputKb: number; maxInputKb: number;
capabilities: BackendCapability[];
calls: number; calls: number;
errors: number; errors: number;
latencyMs: number | null; latencyMs: number | null;
active: boolean;
}; };
type RuntimeEvent = { type RuntimeEvent = {
@@ -101,7 +107,7 @@ function serviceKey(service: Pick<Service, "id" | "revision">) {
return `${service.id}@${service.revision}`; return `${service.id}@${service.revision}`;
} }
function toService(service: BackendService): Service { function toService(service: BackendService, activeRevision?: string): Service {
return { return {
id: service.id, id: service.id,
revision: service.revision, revision: service.revision,
@@ -116,9 +122,11 @@ function toService(service: BackendService): Service {
deadlineMs: service.limits.deadline_ms, deadlineMs: service.limits.deadline_ms,
mailbox: service.limits.mailbox_capacity, mailbox: service.limits.mailbox_capacity,
maxInputKb: Math.round(service.limits.max_input_bytes / 1024), maxInputKb: Math.round(service.limits.max_input_bytes / 1024),
capabilities: service.capabilities,
calls: service.calls, calls: service.calls,
errors: service.errors, errors: service.errors,
latencyMs: service.last_latency_ms, latencyMs: service.last_latency_ms,
active: service.revision === activeRevision,
}; };
} }
@@ -127,6 +135,7 @@ function toEvent(event: BackendEvent): RuntimeEvent {
event.service_id && event.revision ? `${event.service_id}@${event.revision}` : "Wasmeld"; event.service_id && event.revision ? `${event.service_id}@${event.revision}` : "Wasmeld";
const meta: Record<BackendEvent["kind"], { label: string; tone: EventTone }> = { const meta: Record<BackendEvent["kind"], { label: string; tone: EventTone }> = {
registered: { label: "已注册", tone: "success" }, registered: { label: "已注册", tone: "success" },
deployed: { label: "部署切换", tone: "success" },
started: { label: "已启动", tone: "success" }, started: { label: "已启动", tone: "success" },
stopped: { label: "已停止", tone: "neutral" }, stopped: { label: "已停止", tone: "neutral" },
invoked: { label: "调用完成", tone: "success" }, invoked: { label: "调用完成", tone: "success" },
@@ -182,16 +191,65 @@ function invocationInputSize(value: string, format: "utf8" | "hex") {
} }
} }
type FormattedInvocationOutput = {
text: string;
format: "UTF-8" | "HEX" | "U64 LE";
automatic: boolean;
};
function formatHex(bytes: Uint8Array) {
return Array.from(bytes, (byte) => byte.toString(16).padStart(2, "0")).join(" ");
}
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;
}
function formatInvocationOutput(service: Service, output: Uint8Array, format: "utf8" | "hex") { function formatInvocationOutput(service: Service, output: Uint8Array, format: "utf8" | "hex") {
if (service.id === "counter" && output.byteLength === 8) { if (service.id === "counter" && output.byteLength === 8) {
return new DataView(output.buffer, output.byteOffset, output.byteLength) return {
.getBigUint64(0, true) text: new DataView(output.buffer, output.byteOffset, output.byteLength)
.toString(); .getBigUint64(0, true)
.toString(),
format: "U64 LE",
automatic: false,
} satisfies FormattedInvocationOutput;
} }
if (format === "hex") { if (format === "hex") {
return Array.from(output, (byte) => byte.toString(16).padStart(2, "0")).join(" "); return {
text: formatHex(output) || "(empty)",
format: "HEX",
automatic: false,
} satisfies FormattedInvocationOutput;
} }
return new TextDecoder().decode(output) || "(empty)";
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;
} }
export const Route = createFileRoute("/")({ export const Route = createFileRoute("/")({
@@ -202,6 +260,7 @@ function Home() {
const [view, setView] = useState<View>("overview"); const [view, setView] = useState<View>("overview");
const [services, setServices] = useState<Service[]>([]); const [services, setServices] = useState<Service[]>([]);
const [events, setEvents] = useState<RuntimeEvent[]>([]); const [events, setEvents] = useState<RuntimeEvent[]>([]);
const [deployments, setDeployments] = useState<BackendDeployment[]>([]);
const [witPackages, setWitPackages] = useState<BackendWitPackage[]>([]); const [witPackages, setWitPackages] = useState<BackendWitPackage[]>([]);
const [runtime, setRuntime] = useState<BackendRuntime | null>(null); const [runtime, setRuntime] = useState<BackendRuntime | null>(null);
const [runtimeAction, setRuntimeAction] = useState<"start" | "stop" | "restart" | null>(null); const [runtimeAction, setRuntimeAction] = useState<"start" | "stop" | "restart" | null>(null);
@@ -213,11 +272,15 @@ function Home() {
const [deployOpen, setDeployOpen] = useState(false); const [deployOpen, setDeployOpen] = useState(false);
const [witPublishOpen, setWitPublishOpen] = useState(false); const [witPublishOpen, setWitPublishOpen] = useState(false);
const [invokeKey, setInvokeKey] = useState<string | null>(null); const [invokeKey, setInvokeKey] = useState<string | null>(null);
const [activationKey, setActivationKey] = useState<string | null>(null);
const [deploymentAction, setDeploymentAction] = useState<string | null>(null);
const [toast, setToast] = useState<string | null>(null); const [toast, setToast] = useState<string | null>(null);
const selectedService = const selectedService =
services.find((service) => serviceKey(service) === selectedKey) ?? services[0] ?? null; services.find((service) => serviceKey(service) === selectedKey) ?? services[0] ?? null;
const invokeService = services.find((service) => serviceKey(service) === invokeKey) ?? null; const invokeService = services.find((service) => serviceKey(service) === invokeKey) ?? null;
const activationService =
services.find((service) => serviceKey(service) === activationKey) ?? null;
const filteredServices = useMemo(() => { const filteredServices = useMemo(() => {
const normalized = query.trim().toLowerCase(); const normalized = query.trim().toLowerCase();
@@ -250,8 +313,17 @@ function Home() {
const refreshSnapshot = useCallback(async () => { const refreshSnapshot = useCallback(async () => {
try { try {
const snapshot = await fetchSnapshot(apiBase); const snapshot = await fetchSnapshot(apiBase);
const nextServices = snapshot.services.map(toService); const activeRevisions = new Map(
snapshot.deployments.map((deployment) => [
deployment.service_id,
deployment.active_revision,
]),
);
const nextServices = snapshot.services.map((service) =>
toService(service, activeRevisions.get(service.id)),
);
setServices(nextServices); setServices(nextServices);
setDeployments(snapshot.deployments);
setEvents(snapshot.events.map(toEvent)); setEvents(snapshot.events.map(toEvent));
setRuntime(snapshot.runtime); setRuntime(snapshot.runtime);
setWitPackages(snapshot.witPackages); setWitPackages(snapshot.witPackages);
@@ -265,6 +337,7 @@ function Home() {
setConnection("online"); setConnection("online");
} catch { } catch {
setServices([]); setServices([]);
setDeployments([]);
setEvents([]); setEvents([]);
setRuntime(null); setRuntime(null);
setWitPackages([]); setWitPackages([]);
@@ -318,6 +391,21 @@ function Home() {
} }
} }
async function activateService(service: Service) {
const key = serviceKey(service);
setDeploymentAction(key);
try {
await activateDeployment(apiBase, service);
await refreshSnapshot();
setActivationKey(null);
notify(`${service.id}@${service.revision} 已设为对外版本`);
} catch (error) {
notify(error instanceof Error ? error.message : "Deployment 切换失败");
} finally {
setDeploymentAction(null);
}
}
async function runInvocation(service: Service, input: Uint8Array) { async function runInvocation(service: Service, input: Uint8Array) {
const result = await invokeComponent(apiBase, service, input); const result = await invokeComponent(apiBase, service, input);
await refreshSnapshot(); await refreshSnapshot();
@@ -405,16 +493,19 @@ function Home() {
running={running} running={running}
faulted={faulted} faulted={faulted}
totalCalls={totalCalls} totalCalls={totalCalls}
deploymentCount={deployments.length}
runtime={runtime} runtime={runtime}
selectedService={selectedService} selectedService={selectedService}
onSelect={(service) => setSelectedKey(serviceKey(service))} onSelect={(service) => setSelectedKey(serviceKey(service))}
onDeploy={() => setDeployOpen(true)} onDeploy={() => setDeployOpen(true)}
onInvoke={(service) => setInvokeKey(serviceKey(service))} onInvoke={(service) => setInvokeKey(serviceKey(service))}
onStatus={setServiceStatus} onStatus={setServiceStatus}
onActivate={(service) => setActivationKey(serviceKey(service))}
onViewAll={() => setView("services")} onViewAll={() => setView("services")}
onRefresh={() => void refreshSnapshot()} onRefresh={() => void refreshSnapshot()}
onRuntimeAction={(action) => void setRuntimeStatus(action)} onRuntimeAction={(action) => void setRuntimeStatus(action)}
runtimeAction={runtimeAction} runtimeAction={runtimeAction}
deploymentAction={deploymentAction}
/> />
)} )}
@@ -427,6 +518,9 @@ function Home() {
onInvoke={(service) => setInvokeKey(serviceKey(service))} onInvoke={(service) => setInvokeKey(serviceKey(service))}
onSelect={(service) => setSelectedKey(serviceKey(service))} onSelect={(service) => setSelectedKey(serviceKey(service))}
onStatus={setServiceStatus} onStatus={setServiceStatus}
onActivate={(service) => setActivationKey(serviceKey(service))}
canActivate={runtime?.status === "running"}
deploymentAction={deploymentAction}
/> />
)} )}
@@ -484,6 +578,19 @@ function Home() {
/> />
)} )}
{activationService && (
<ActivateDeploymentDialog
service={activationService}
currentRevision={
services.find((service) => service.id === activationService.id && service.active)
?.revision ?? null
}
submitting={deploymentAction === serviceKey(activationService)}
onClose={() => setActivationKey(null)}
onConfirm={() => void activateService(activationService)}
/>
)}
{toast && ( {toast && (
<output className="toast"> <output className="toast">
<CheckCircle2 size={17} /> <CheckCircle2 size={17} />
@@ -613,16 +720,19 @@ function Overview({
running, running,
faulted, faulted,
totalCalls, totalCalls,
deploymentCount,
runtime, runtime,
selectedService, selectedService,
onSelect, onSelect,
onDeploy, onDeploy,
onInvoke, onInvoke,
onStatus, onStatus,
onActivate,
onViewAll, onViewAll,
onRefresh, onRefresh,
onRuntimeAction, onRuntimeAction,
runtimeAction, runtimeAction,
deploymentAction,
}: { }: {
services: Service[]; services: Service[];
filteredServices: Service[]; filteredServices: Service[];
@@ -630,16 +740,19 @@ function Overview({
running: number; running: number;
faulted: number; faulted: number;
totalCalls: number; totalCalls: number;
deploymentCount: number;
runtime: BackendRuntime | null; runtime: BackendRuntime | null;
selectedService: Service | null; selectedService: Service | null;
onSelect: (service: Service) => void; onSelect: (service: Service) => void;
onDeploy: () => void; onDeploy: () => void;
onInvoke: (service: Service) => void; onInvoke: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void; onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
onViewAll: () => void; onViewAll: () => void;
onRefresh: () => void; onRefresh: () => void;
onRuntimeAction: (action: "start" | "stop" | "restart") => void; onRuntimeAction: (action: "start" | "stop" | "restart") => void;
runtimeAction: "start" | "stop" | "restart" | null; runtimeAction: "start" | "stop" | "restart" | null;
deploymentAction: string | null;
}) { }) {
const errorCount = services.reduce((sum, service) => sum + service.errors, 0); const errorCount = services.reduce((sum, service) => sum + service.errors, 0);
const latencies = services const latencies = services
@@ -677,7 +790,7 @@ function Overview({
icon={<Package size={18} />} icon={<Package size={18} />}
label="服务版本" label="服务版本"
value={services.length.toString()} value={services.length.toString()}
note="来自 Wasmeld" note={`${deploymentCount} 个对外服务`}
tone="cyan" tone="cyan"
/> />
<Metric <Metric
@@ -794,6 +907,9 @@ function Overview({
onSelect={onSelect} onSelect={onSelect}
onInvoke={onInvoke} onInvoke={onInvoke}
onStatus={onStatus} onStatus={onStatus}
onActivate={onActivate}
canActivate={runtime?.status === "running"}
deploymentAction={deploymentAction}
compact compact
/> />
</section> </section>
@@ -832,6 +948,17 @@ function Overview({
<DetailValue label="Deadline" value={`${selectedService.deadlineMs} ms`} /> <DetailValue label="Deadline" value={`${selectedService.deadlineMs} ms`} />
<DetailValue label="Mailbox" value={selectedService.mailbox.toString()} /> <DetailValue label="Mailbox" value={selectedService.mailbox.toString()} />
<div className="detail-actions"> <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 <button
className="secondary-button" className="secondary-button"
type="button" type="button"
@@ -902,12 +1029,35 @@ function StatusBadge({ status }: { status: ServiceStatus }) {
); );
} }
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({ function ServiceTable({
services, services,
selectedKey, selectedKey,
onSelect, onSelect,
onInvoke, onInvoke,
onStatus, onStatus,
onActivate,
canActivate,
deploymentAction,
compact = false, compact = false,
}: { }: {
services: Service[]; services: Service[];
@@ -915,6 +1065,9 @@ function ServiceTable({
onSelect: (service: Service) => void; onSelect: (service: Service) => void;
onInvoke: (service: Service) => void; onInvoke: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void; onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
canActivate: boolean;
deploymentAction: string | null;
compact?: boolean; compact?: boolean;
}) { }) {
if (services.length === 0) { if (services.length === 0) {
@@ -929,12 +1082,13 @@ function ServiceTable({
return ( return (
<div className="table-scroll"> <div className="table-scroll">
<table className="data-table"> <table className={`data-table ${compact ? "service-table-compact" : "service-table-full"}`}>
<thead> <thead>
<tr> <tr>
<th></th> <th></th>
<th></th> <th></th>
{!compact && <th></th>} {!compact && <th></th>}
{!compact && <th>Host </th>}
<th></th> <th></th>
<th></th> <th></th>
<th> <th>
@@ -959,7 +1113,15 @@ function ServiceTable({
<Box size={16} /> <Box size={16} />
</span> </span>
<span> <span>
<strong>{service.id}</strong> <span className="service-title">
<strong>{service.id}</strong>
{service.active && (
<span className="active-deployment-badge">
<RadioTower size={10} />
</span>
)}
</span>
<small> <small>
{service.revision} · {service.description} {service.revision} · {service.description}
</small> </small>
@@ -970,12 +1132,41 @@ function ServiceTable({
<StatusBadge status={service.status} /> <StatusBadge status={service.status} />
</td> </td>
{!compact && <td className="artifact-cell">{service.artifact}</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.calls.toLocaleString("zh-CN")}</td>
<td className="numeric-cell"> <td className="numeric-cell">
{service.latencyMs ? `${service.latencyMs} ms` : "—"} {service.latencyMs ? `${service.latencyMs} ms` : "—"}
</td> </td>
<td> <td>
<div className="row-actions"> <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" ? ( {service.status === "running" ? (
<> <>
<button <button
@@ -1055,6 +1246,9 @@ function ServicesView({
onInvoke, onInvoke,
onSelect, onSelect,
onStatus, onStatus,
onActivate,
canActivate,
deploymentAction,
}: { }: {
services: Service[]; services: Service[];
statusFilter: "all" | ServiceStatus; statusFilter: "all" | ServiceStatus;
@@ -1063,13 +1257,16 @@ function ServicesView({
onInvoke: (service: Service) => void; onInvoke: (service: Service) => void;
onSelect: (service: Service) => void; onSelect: (service: Service) => void;
onStatus: (service: Service, status: ServiceStatus) => void; onStatus: (service: Service, status: ServiceStatus) => void;
onActivate: (service: Service) => void;
canActivate: boolean;
deploymentAction: string | null;
}) { }) {
return ( return (
<> <>
<PageHeading <PageHeading
eyebrow="REGISTRY" eyebrow="REGISTRY"
title="服务版本" title="服务版本"
description="管理 Runtime 已注册的不可变 Wasm Component 制品。" description="管理不可变 Wasm Component 制品及当前对外版本。"
actions={ actions={
<button className="primary-button" type="button" onClick={onDeploy}> <button className="primary-button" type="button" onClick={onDeploy}>
<CloudUpload size={17} /> <CloudUpload size={17} />
@@ -1103,6 +1300,9 @@ function ServicesView({
onSelect={onSelect} onSelect={onSelect}
onInvoke={onInvoke} onInvoke={onInvoke}
onStatus={onStatus} onStatus={onStatus}
onActivate={onActivate}
canActivate={canActivate}
deploymentAction={deploymentAction}
/> />
</section> </section>
</> </>
@@ -1435,12 +1635,14 @@ function DialogFrame({
description, description,
icon, icon,
onClose, onClose,
closeDisabled = false,
children, children,
}: { }: {
title: string; title: string;
description: string; description: string;
icon: ReactNode; icon: ReactNode;
onClose: () => void; onClose: () => void;
closeDisabled?: boolean;
children: ReactNode; children: ReactNode;
}) { }) {
return ( return (
@@ -1449,9 +1651,12 @@ function DialogFrame({
open open
className="dialog" className="dialog"
aria-label={title} aria-label={title}
onCancel={onClose} onCancel={(event) => {
if (closeDisabled) event.preventDefault();
else onClose();
}}
onKeyDown={(event) => { onKeyDown={(event) => {
if (event.key === "Escape") onClose(); if (event.key === "Escape" && !closeDisabled) onClose();
}} }}
> >
<div className="dialog-header"> <div className="dialog-header">
@@ -1467,6 +1672,7 @@ function DialogFrame({
type="button" type="button"
aria-label="关闭" aria-label="关闭"
title="关闭" title="关闭"
disabled={closeDisabled}
onClick={onClose} onClick={onClose}
> >
<X size={18} /> <X size={18} />
@@ -1478,6 +1684,56 @@ function DialogFrame({
); );
} }
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>
);
}
function DeployDialog({ function DeployDialog({
onClose, onClose,
onRegister, onRegister,
@@ -1634,6 +1890,8 @@ function InvokeDialog({
const [input, setInput] = useState(service.id === "echo" ? "hello wasm" : ""); const [input, setInput] = useState(service.id === "echo" ? "hello wasm" : "");
const [format, setFormat] = useState<"utf8" | "hex">("utf8"); const [format, setFormat] = useState<"utf8" | "hex">("utf8");
const [output, setOutput] = useState(""); const [output, setOutput] = useState("");
const [outputBytes, setOutputBytes] = useState<number | null>(null);
const [outputFormat, setOutputFormat] = useState<FormattedInvocationOutput | null>(null);
const [running, setRunning] = useState(false); const [running, setRunning] = useState(false);
const [failed, setFailed] = useState(false); const [failed, setFailed] = useState(false);
const [latency, setLatency] = useState<number | null>(null); const [latency, setLatency] = useState<number | null>(null);
@@ -1641,11 +1899,16 @@ function InvokeDialog({
async function invoke() { async function invoke() {
setRunning(true); setRunning(true);
setOutput(""); setOutput("");
setOutputBytes(null);
setOutputFormat(null);
setFailed(false); setFailed(false);
setLatency(null); setLatency(null);
try { try {
const result = await onInvoke(service, parseInvocationInput(input, format)); const result = await onInvoke(service, parseInvocationInput(input, format));
setOutput(formatInvocationOutput(service, result.output, format)); const formatted = formatInvocationOutput(service, result.output, format);
setOutput(formatted.text);
setOutputBytes(result.outputBytes);
setOutputFormat(formatted);
setLatency(result.latencyMs); setLatency(result.latencyMs);
} catch (invokeError) { } catch (invokeError) {
setFailed(true); setFailed(true);
@@ -1664,7 +1927,7 @@ function InvokeDialog({
> >
<div className="invoke-body"> <div className="invoke-body">
<div className="invoke-toolbar"> <div className="invoke-toolbar">
<div className="segmented"> <div className="segmented" aria-label="输入编码">
<button <button
type="button" type="button"
className={format === "utf8" ? "active" : ""} className={format === "utf8" ? "active" : ""}
@@ -1695,11 +1958,17 @@ function InvokeDialog({
<div className={failed ? "invoke-output failed" : "invoke-output"}> <div className={failed ? "invoke-output failed" : "invoke-output"}>
<div> <div>
<span></span> <span></span>
{output && latency !== null && ( {output && latency !== null && outputBytes !== null && outputFormat && (
<small> <div className="invoke-output-meta">
<Clock3 size={13} /> <span>
{latency} ms {outputBytes} B · {outputFormat.format}
</small> {outputFormat.automatic ? " · 自动" : ""}
</span>
<small>
<Clock3 size={13} />
{latency} ms
</small>
</div>
)} )}
</div> </div>
<pre>{running ? "invoking..." : output || "等待调用"}</pre> <pre>{running ? "invoking..." : output || "等待调用"}</pre>
+155 -2
View File
@@ -768,6 +768,31 @@ button:disabled {
table-layout: fixed; table-layout: fixed;
} }
.service-table-full {
min-width: 980px;
}
.data-table.service-table-full th:first-child {
width: 24%;
}
.service-table-full th:nth-child(2) {
width: 90px;
}
.service-table-full th:nth-child(3) {
width: 150px;
}
.service-table-full th:nth-child(4) {
width: 220px;
}
.service-table-full th:nth-child(5),
.service-table-full th:nth-child(6) {
width: 72px;
}
.data-table th { .data-table th {
height: 34px; height: 34px;
padding: 0 12px; padding: 0 12px;
@@ -785,7 +810,7 @@ button:disabled {
} }
.data-table th:nth-last-child(1) { .data-table th:nth-last-child(1) {
width: 122px; width: 150px;
} }
.data-table td { .data-table td {
@@ -849,6 +874,33 @@ button:disabled {
font-size: 11px; font-size: 11px;
} }
.service-title {
min-width: 0;
display: flex;
align-items: center;
gap: 6px;
}
.service-title strong {
overflow: hidden;
text-overflow: ellipsis;
}
.active-deployment-badge {
flex: 0 0 auto;
height: 17px;
padding: 0 5px;
display: inline-flex;
align-items: center;
gap: 3px;
color: #176b4d;
background: #e4f3ed;
border: 1px solid #c5e4d8;
border-radius: 9px;
font-size: 7px;
font-weight: 700;
}
.service-cell small { .service-cell small {
margin-top: 3px; margin-top: 3px;
color: var(--ink-faint); color: var(--ink-faint);
@@ -864,6 +916,38 @@ button:disabled {
white-space: nowrap; white-space: nowrap;
} }
.capability-summary {
max-width: 100%;
height: 23px;
padding: 0 7px;
display: inline-flex;
align-items: center;
gap: 5px;
color: #315f52;
background: #edf5f2;
border: 1px solid #d2e4de;
border-radius: 4px;
font-family: var(--font-geist-mono), monospace;
font-size: 8px;
}
.capability-summary > span {
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.capability-summary > small {
flex: 0 0 auto;
color: var(--ink-faint);
font-size: 7px;
}
.capability-empty {
color: var(--ink-faint);
font-size: 9px;
}
.numeric-cell { .numeric-cell {
color: var(--ink-soft); color: var(--ink-soft);
font-family: var(--font-geist-mono), monospace; font-family: var(--font-geist-mono), monospace;
@@ -917,7 +1001,7 @@ button:disabled {
} }
.row-actions { .row-actions {
min-width: 100px; min-width: 128px;
display: flex; display: flex;
align-items: center; align-items: center;
justify-content: flex-end; justify-content: flex-end;
@@ -943,6 +1027,14 @@ button:disabled {
border-color: var(--line); border-color: var(--line);
} }
.row-actions .active-deployment-button:disabled {
color: #197253;
background: #e7f4ef;
border-color: #c6e4d9;
cursor: default;
opacity: 1;
}
.event-list { .event-list {
padding: 4px 14px; padding: 4px 14px;
} }
@@ -1094,6 +1186,7 @@ button:disabled {
.detail-actions { .detail-actions {
display: flex; display: flex;
justify-content: flex-end; justify-content: flex-end;
gap: 7px;
} }
.full-panel { .full-panel {
@@ -1597,6 +1690,54 @@ button:disabled {
margin: 18px -18px -18px; margin: 18px -18px -18px;
} }
.deployment-confirmation {
padding: 18px;
}
.deployment-route {
min-height: 72px;
display: grid;
grid-template-columns: minmax(0, 1fr) 24px minmax(0, 1fr);
align-items: center;
gap: 12px;
border-top: 1px solid var(--line);
border-bottom: 1px solid var(--line);
}
.deployment-route > svg {
color: var(--ink-faint);
}
.deployment-route span,
.deployment-route strong {
display: block;
}
.deployment-route span {
margin-bottom: 5px;
color: var(--ink-faint);
font-size: 8px;
}
.deployment-route strong {
overflow: hidden;
font-family: var(--font-geist-mono), monospace;
font-size: 13px;
text-overflow: ellipsis;
white-space: nowrap;
}
.deployment-target {
color: #176b4d;
}
.deployment-confirmation p {
margin: 13px 0 0;
color: var(--ink-soft);
font-size: 9px;
line-height: 16px;
}
.invoke-body { .invoke-body {
padding: 16px; padding: 16px;
} }
@@ -1642,6 +1783,18 @@ button:disabled {
justify-content: space-between; justify-content: space-between;
} }
.invoke-output-meta {
display: flex;
align-items: center;
gap: 12px;
}
.invoke-output-meta > span {
color: var(--ink-faint);
font-family: var(--font-geist-mono), monospace;
font-size: 8px;
}
.invoke-output small { .invoke-output small {
display: flex; display: flex;
align-items: center; align-items: center;
+10 -2
View File
@@ -77,15 +77,16 @@ test("server-renders the Wasm management console", async () => {
assert.match(html, /运行概览/); assert.match(html, /运行概览/);
assert.match(html, /WIT 包/); assert.match(html, /WIT 包/);
assert.match(html, /连接中/); assert.match(html, /连接中/);
assert.match(html, /来自 Wasmeld/); assert.match(html, /0 个对外服务/);
assert.match(html, /Component Model/); assert.match(html, /Component Model/);
assert.doesNotMatch(html, /本地演示|codex-preview|vinext|next\.js/i); assert.doesNotMatch(html, /本地演示|codex-preview|vinext|next\.js/i);
}); });
test("uses TanStack Start routing and produces Node artifacts", async () => { test("uses TanStack Start routing and produces Node artifacts", async () => {
const [rootRoute, indexRoute, router, packageJson, viteConfig] = await Promise.all([ const [rootRoute, indexRoute, apiClient, router, packageJson, viteConfig] = await Promise.all([
readFile(new URL("../src/routes/__root.tsx", import.meta.url), "utf8"), 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/routes/index.tsx", 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("../src/router.tsx", import.meta.url), "utf8"),
readFile(new URL("../package.json", import.meta.url), "utf8"), readFile(new URL("../package.json", import.meta.url), "utf8"),
readFile(new URL("../vite.config.ts", import.meta.url), "utf8"), readFile(new URL("../vite.config.ts", import.meta.url), "utf8"),
@@ -95,6 +96,13 @@ test("uses TanStack Start routing and produces Node artifacts", async () => {
assert.match(rootRoute, /<HeadContent \/>/); assert.match(rootRoute, /<HeadContent \/>/);
assert.match(rootRoute, /<Scripts \/>/); assert.match(rootRoute, /<Scripts \/>/);
assert.match(indexRoute, /createFileRoute\("\/"\)/); 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(apiClient, /\/api\/v1\/deployments/);
assert.match(apiClient, /capabilities: BackendCapability\[\]/);
assert.match(router, /createRouter/); assert.match(router, /createRouter/);
assert.match(packageJson, /"@tanstack\/react-start"/); assert.match(packageJson, /"@tanstack\/react-start"/);
assert.match(packageJson, /"srvx"/); assert.match(packageJson, /"srvx"/);
+43 -7
View File
@@ -6,17 +6,18 @@
- 发布、查询和下载不可变的二进制 WIT Package - 发布、查询和下载不可变的二进制 WIT Package
- 启动、停止和重启内嵌的 `wasmeld-runtime` - 启动、停止和重启内嵌的 `wasmeld-runtime`
- 启动、停止和重启常驻 Actor - 启动、停止和重启常驻 Actor
- 转发二进制调用 - 管理服务 ID 到活动版本的 Deployment
- 通过独立 Gateway 转发原始二进制调用
- 暴露调用计数、状态和最近事件 - 暴露调用计数、状态和最近事件
Wasm 制品保存在本地文件系统。服务 manifest、调用计数和最近 256 条事件通过 Wasm 制品保存在本地文件系统。服务 manifest、调用计数和最近 256 条事件通过
[Toasty](https://github.com/tokio-rs/toasty) 写入本地 libSQL 数据库,默认路径是 [Toasty](https://github.com/tokio-rs/toasty) 写入本地 libSQL 数据库,默认路径是
`var/wasmeld/console.db` `var/wasmeld/console.db`Deployment 也存储在同一数据库中。
`wasmeld-console` 启动时会在同一进程内创建 `wasmeld-runtime`、加载 Wasmtime Engine `wasmeld-console` 启动时会在同一进程内创建 `wasmeld-runtime`、加载 Wasmtime Engine
并重新注册已保存的 Component。Runtime 停止后 Console HTTP 和数据库仍然在线; 并重新注册已保存的 Component。Runtime 停止后 Console HTTP 和数据库仍然在线;
再次启动 Runtime 会重新注册 Component,但 Actor 内存不做快照,服务运行状态统一 再次启动 Runtime 会重新注册 Component。未部署的版本恢复为 `stopped`Deployment
恢复为 `stopped` 指向的版本会用空初始化配置创建新的 Actor。Actor 的 Store 和线性内存不做快照
## 启动 ## 启动
@@ -27,11 +28,18 @@ Wasm 制品保存在本地文件系统。服务 manifest、调用计数和最近
cargo +stable run -p wasmeld-console cargo +stable run -p wasmeld-console
``` ```
默认监听 `127.0.0.1:8080` 进程默认启动两个独立 Listener:
- 管理 API`127.0.0.1:8080`
- Gateway`0.0.0.0:8081`
管理 Router 不会挂载到 Gateway ListenerGateway 也不暴露注册、Runtime 生命周期和
WIT Registry 路由。
可用环境变量: 可用环境变量:
- `WASMELD_ADDR` - `WASMELD_ADDR`
- `WASMELD_GATEWAY_ADDR`
- `WASMELD_ARTIFACT_DIR` - `WASMELD_ARTIFACT_DIR`
- `WASMELD_DATABASE_PATH` - `WASMELD_DATABASE_PATH`
- `WASMELD_WIT_REGISTRY_DIR` - `WASMELD_WIT_REGISTRY_DIR`
@@ -46,19 +54,47 @@ cargo +stable run -p wasmeld-console
| `GET` | `/api/v1/runtime` | Runtime 状态 | | `GET` | `/api/v1/runtime` | Runtime 状态 |
| `POST` | `/api/v1/runtime/start` | 启动 Runtime 并重新注册受管 Component | | `POST` | `/api/v1/runtime/start` | 启动 Runtime 并重新注册受管 Component |
| `POST` | `/api/v1/runtime/stop` | 停止 Runtime 并释放所有 Actor | | `POST` | `/api/v1/runtime/stop` | 停止 Runtime 并释放所有 Actor |
| `POST` | `/api/v1/runtime/restart` | 重建 Runtime服务恢复为停止状态 | | `POST` | `/api/v1/runtime/restart` | 重建 Runtime并重新创建活动 Deployment 的 Actor |
| `GET` | `/api/v1/services` | 服务版本列表 | | `GET` | `/api/v1/services` | 服务版本及精确 Host Capability 列表 |
| `POST` | `/api/v1/services` | multipart 上传单个 `package` 字段(`.wasmpkg` | | `POST` | `/api/v1/services` | multipart 上传单个 `package` 字段(`.wasmpkg` |
| `POST` | `/api/v1/services/{id}/{revision}/start` | 启动 Actor | | `POST` | `/api/v1/services/{id}/{revision}/start` | 启动 Actor |
| `POST` | `/api/v1/services/{id}/{revision}/stop` | 停止 Actor | | `POST` | `/api/v1/services/{id}/{revision}/stop` | 停止 Actor |
| `POST` | `/api/v1/services/{id}/{revision}/restart` | 重启 Actor | | `POST` | `/api/v1/services/{id}/{revision}/restart` | 重启 Actor |
| `POST` | `/api/v1/services/{id}/{revision}/invoke` | Base64 二进制调用 | | `POST` | `/api/v1/services/{id}/{revision}/invoke` | Base64 二进制调用 |
| `GET` | `/api/v1/deployments` | 活动 Deployment 列表 |
| `GET` | `/api/v1/deployments/{id}` | 查询服务的活动版本 |
| `POST` | `/api/v1/deployments/{id}/activate` | 启动目标版本并原子切换活动版本 |
| `GET` | `/api/v1/events` | 最近 256 条持久化事件 | | `GET` | `/api/v1/events` | 最近 256 条持久化事件 |
| `GET` | `/api/v1/wit/packages` | WIT 包版本列表 | | `GET` | `/api/v1/wit/packages` | WIT 包版本列表 |
| `POST` | `/api/v1/wit/packages` | multipart 发布单个二进制 WIT `package` 字段 | | `POST` | `/api/v1/wit/packages` | multipart 发布单个二进制 WIT `package` 字段 |
| `GET` | `/api/v1/wit/packages/{namespace}/{name}/{version}` | WIT 包元数据 | | `GET` | `/api/v1/wit/packages/{namespace}/{name}/{version}` | WIT 包元数据 |
| `GET` | `/api/v1/wit/packages/{namespace}/{name}/{version}/content` | 下载 WIT 包 | | `GET` | `/api/v1/wit/packages/{namespace}/{name}/{version}/content` | 下载 WIT 包 |
激活已注册版本:
```bash
curl -X POST http://127.0.0.1:8080/api/v1/deployments/echo/activate \
-H 'Content-Type: application/json' \
-d '{"revision":"0.1.0"}'
```
Gateway 只提供 `GET /healthz`
`POST /v1/services/{id}/invoke`。调用请求与成功响应的
`Content-Type` 都是 `application/octet-stream`
```bash
curl http://127.0.0.1:8081/v1/services/echo/invoke \
-H 'Content-Type: application/octet-stream' \
--data-binary 'hello'
```
成功响应通过 `X-Wasmeld-Revision` 返回实际路由的版本。Gateway 错误统一返回 JSON
未部署为 `404`Actor 不可用为 `503`,过载为 `429`,执行超时为 `504`
服务响应的 `capabilities` 来自 Component 二进制中的 WIT imports,例如
`wasmeld:clock/monotonic-clock@0.1.0`。Console 不接受手工能力声明,Runtime 只会链接
注册表中存在且版本完全匹配的接口。
组件包的构建与格式说明见 组件包的构建与格式说明见
[`docs/design/wasmeld-component-package.md`](../../docs/design/wasmeld-component-package.md)。 [`docs/design/wasmeld-component-package.md`](../../docs/design/wasmeld-component-package.md)。
WIT 依赖、Registry 和本地 replace 说明见 WIT 依赖、Registry 和本地 replace 说明见
+517 -29
View File
@@ -19,9 +19,9 @@ use std::{
use axum::{ use axum::{
Json, Router, Json, Router,
body::Body, body::{Body, Bytes},
extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State}, extract::{DefaultBodyLimit, Multipart, Path as AxumPath, State, rejection::BytesRejection},
http::{HeaderValue, Method, StatusCode, header}, http::{HeaderMap, HeaderValue, Method, StatusCode, header},
response::{IntoResponse, Response}, response::{IntoResponse, Response},
routing::{get, post}, routing::{get, post},
}; };
@@ -39,16 +39,18 @@ use wasmeld_package::{
wit_package::{WitPackageError, WitPackageMetadata}, wit_package::{WitPackageError, WitPackageMetadata},
}; };
use wasmeld_runtime::{ use wasmeld_runtime::{
ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey, ServiceManifest, CapabilityDescriptor, ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceKey,
ServiceManifest,
}; };
use persistence::{Persistence, StoredEvent, StoredService}; use persistence::{Persistence, StoredDeployment, StoredEvent, StoredService};
use wit_registry::WitRegistry; use wit_registry::WitRegistry;
pub use wit_registry::WitRegistryError; pub use wit_registry::WitRegistryError;
const MAX_EVENTS: usize = 256; const MAX_EVENTS: usize = 256;
const DEFAULT_MAX_ARTIFACT_BYTES: usize = 64 * 1024 * 1024; const DEFAULT_MAX_ARTIFACT_BYTES: usize = 64 * 1024 * 1024;
const DEFAULT_MAX_WIT_PACKAGE_BYTES: usize = 4 * 1024 * 1024; const DEFAULT_MAX_WIT_PACKAGE_BYTES: usize = 4 * 1024 * 1024;
type PersistenceSnapshot = (Vec<StoredService>, Vec<StoredEvent>, Vec<StoredDeployment>);
/// Filesystem, persistence, upload, and Runtime policy used by [`Console`]. /// Filesystem, persistence, upload, and Runtime policy used by [`Console`].
#[derive(Clone, Debug)] #[derive(Clone, Debug)]
@@ -105,12 +107,14 @@ struct ConsoleState {
runtime_started_at: Option<Instant>, runtime_started_at: Option<Instant>,
next_event_id: u64, next_event_id: u64,
services: BTreeMap<ServiceKey, ServiceRecord>, services: BTreeMap<ServiceKey, ServiceRecord>,
deployments: BTreeMap<String, DeploymentRecord>,
events: VecDeque<EventView>, events: VecDeque<EventView>,
} }
#[derive(Clone)] #[derive(Clone)]
struct ServiceRecord { struct ServiceRecord {
manifest: ServiceManifest, manifest: ServiceManifest,
capabilities: Vec<CapabilityView>,
status: ServiceStatus, status: ServiceStatus,
updated_at_ms: u64, updated_at_ms: u64,
calls: u64, calls: u64,
@@ -118,8 +122,14 @@ struct ServiceRecord {
last_latency_ms: Option<u64>, last_latency_ms: Option<u64>,
} }
#[derive(Clone)]
struct DeploymentRecord {
active_revision: String,
updated_at_ms: u64,
}
/// Lifecycle status exposed for one registered service version. /// Lifecycle status exposed for one registered service version.
#[derive(Clone, Copy, Debug, Serialize)] #[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum ServiceStatus { pub enum ServiceStatus {
Running, Running,
@@ -127,6 +137,35 @@ pub enum ServiceStatus {
Faulted, Faulted,
} }
/// API projection of one public service deployment.
#[derive(Clone, Debug, Serialize)]
pub struct DeploymentView {
pub service_id: String,
pub active_revision: String,
pub status: ServiceStatus,
pub updated_at_ms: u64,
}
/// Exact versioned host interface imported by a registered Component.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub struct CapabilityView {
pub interface: String,
pub package: String,
pub name: String,
pub version: String,
}
impl From<CapabilityDescriptor> for CapabilityView {
fn from(capability: CapabilityDescriptor) -> Self {
Self {
interface: capability.interface().to_owned(),
package: capability.package().to_owned(),
name: capability.name().to_owned(),
version: capability.version().to_owned(),
}
}
}
/// API projection of a registered service and its current metrics. /// API projection of a registered service and its current metrics.
#[derive(Clone, Debug, Serialize)] #[derive(Clone, Debug, Serialize)]
pub struct ServiceView { pub struct ServiceView {
@@ -134,6 +173,7 @@ pub struct ServiceView {
pub revision: String, pub revision: String,
pub artifact: String, pub artifact: String,
pub world: String, pub world: String,
pub capabilities: Vec<CapabilityView>,
pub status: ServiceStatus, pub status: ServiceStatus,
pub updated_at_ms: u64, pub updated_at_ms: u64,
pub limits: ResourceLimits, pub limits: ResourceLimits,
@@ -149,6 +189,7 @@ impl ServiceRecord {
revision: self.manifest.revision.clone(), revision: self.manifest.revision.clone(),
artifact: self.manifest.component.clone(), artifact: self.manifest.component.clone(),
world: self.manifest.world.clone(), world: self.manifest.world.clone(),
capabilities: self.capabilities.clone(),
status: self.status, status: self.status,
updated_at_ms: self.updated_at_ms, updated_at_ms: self.updated_at_ms,
limits: self.manifest.limits.clone(), limits: self.manifest.limits.clone(),
@@ -175,6 +216,7 @@ pub struct EventView {
#[serde(rename_all = "snake_case")] #[serde(rename_all = "snake_case")]
pub enum EventKind { pub enum EventKind {
Registered, Registered,
Deployed,
Started, Started,
Stopped, Stopped,
Invoked, Invoked,
@@ -185,6 +227,7 @@ impl EventKind {
fn as_database_value(self) -> &'static str { fn as_database_value(self) -> &'static str {
match self { match self {
Self::Registered => "registered", Self::Registered => "registered",
Self::Deployed => "deployed",
Self::Started => "started", Self::Started => "started",
Self::Stopped => "stopped", Self::Stopped => "stopped",
Self::Invoked => "invoked", Self::Invoked => "invoked",
@@ -195,6 +238,7 @@ impl EventKind {
fn from_database_value(value: &str) -> Result<Self, ConsoleError> { fn from_database_value(value: &str) -> Result<Self, ConsoleError> {
match value { match value {
"registered" => Ok(Self::Registered), "registered" => Ok(Self::Registered),
"deployed" => Ok(Self::Deployed),
"started" => Ok(Self::Started), "started" => Ok(Self::Started),
"stopped" => Ok(Self::Stopped), "stopped" => Ok(Self::Stopped),
"invoked" => Ok(Self::Invoked), "invoked" => Ok(Self::Invoked),
@@ -233,11 +277,23 @@ struct InvokeRequest {
input_base64: String, input_base64: String,
} }
#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ActivateDeploymentRequest {
revision: String,
init_config_base64: Option<String>,
}
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
struct ServiceList { struct ServiceList {
services: Vec<ServiceView>, services: Vec<ServiceView>,
} }
#[derive(Debug, Serialize)]
struct DeploymentList {
deployments: Vec<DeploymentView>,
}
#[derive(Debug, Serialize)] #[derive(Debug, Serialize)]
struct EventList { struct EventList {
events: Vec<EventView>, events: Vec<EventView>,
@@ -271,6 +327,16 @@ struct InvokeResponse {
latency_ms: u64, latency_ms: u64,
} }
struct InvocationResult {
output: Vec<u8>,
latency_ms: u64,
}
struct GatewayInvocation {
revision: String,
result: InvocationResult,
}
/// Errors raised by Console state, persistence, Registry, or Runtime operations. /// Errors raised by Console state, persistence, Registry, or Runtime operations.
#[derive(Debug, Error)] #[derive(Debug, Error)]
pub enum ConsoleError { pub enum ConsoleError {
@@ -280,6 +346,9 @@ pub enum ConsoleError {
#[error("service revision {0} is not managed by Wasmeld")] #[error("service revision {0} is not managed by Wasmeld")]
ServiceNotFound(ServiceKey), ServiceNotFound(ServiceKey),
#[error("service {0} has no active deployment")]
DeploymentNotFound(String),
#[error("Wasmeld Runtime is not running")] #[error("Wasmeld Runtime is not running")]
RuntimeNotRunning, RuntimeNotRunning,
@@ -319,10 +388,10 @@ pub enum ConsoleError {
} }
impl Console { impl Console {
/// Opens durable stores, restores service metadata, and starts a fresh Runtime. /// Opens durable stores, restores control data, and starts a fresh Runtime.
/// ///
/// Persisted actors are deliberately restored as stopped services because /// Actors selected by durable deployments are recreated with empty init
/// their Store and linear memory cannot be reconstructed from control data. /// configuration. Their previous Store and linear memory are not restored.
pub async fn new(config: ConsoleConfig) -> Result<Self, ConsoleError> { pub async fn new(config: ConsoleConfig) -> Result<Self, ConsoleError> {
config.registration_limits.validate()?; config.registration_limits.validate()?;
fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage { fs::create_dir_all(&config.artifact_dir).map_err(|source| ConsoleError::Storage {
@@ -343,7 +412,7 @@ impl Console {
let persistence = Persistence::open(&config.database_path).await?; let persistence = Persistence::open(&config.database_path).await?;
let wit_registry = let wit_registry =
WitRegistry::open(&config.wit_registry_dir, config.max_wit_package_bytes)?; WitRegistry::open(&config.wit_registry_dir, config.max_wit_package_bytes)?;
let (stored_services, mut stored_events) = persistence.load().await?; let (stored_services, mut stored_events, stored_deployments) = persistence.load().await?;
stored_events.sort_unstable_by_key(|event| std::cmp::Reverse(event.id)); stored_events.sort_unstable_by_key(|event| std::cmp::Reverse(event.id));
stored_events.truncate(MAX_EVENTS); stored_events.truncate(MAX_EVENTS);
let next_event_id = stored_events let next_event_id = stored_events
@@ -368,6 +437,7 @@ impl Console {
runtime_started_at: Some(Instant::now()), runtime_started_at: Some(Instant::now()),
next_event_id, next_event_id,
services: BTreeMap::new(), services: BTreeMap::new(),
deployments: BTreeMap::new(),
events, events,
}), }),
persistence, persistence,
@@ -377,6 +447,8 @@ impl Console {
// then used to recover artifacts that predate or missed a DB snapshot. // then used to recover artifacts that predate or missed a DB snapshot.
console.load_stored_services(stored_services)?; console.load_stored_services(stored_services)?;
console.load_manifest_services()?; console.load_manifest_services()?;
console.load_stored_deployments(stored_deployments)?;
console.restore_deployments()?;
console.persist_snapshot().await?; console.persist_snapshot().await?;
Ok(console) Ok(console)
} }
@@ -391,6 +463,11 @@ impl Console {
self.wit_registry.max_package_bytes() self.wit_registry.max_package_bytes()
} }
/// Returns the maximum raw request size accepted by the Gateway.
pub fn max_gateway_input_bytes(&self) -> usize {
self.registration_limits.max_input_bytes
}
/// Returns registered service versions in stable key order. /// Returns registered service versions in stable key order.
pub fn services(&self) -> Result<Vec<ServiceView>, ConsoleError> { pub fn services(&self) -> Result<Vec<ServiceView>, ConsoleError> {
let state = self.state()?; let state = self.state()?;
@@ -403,6 +480,26 @@ impl Console {
Ok(state.events.iter().cloned().collect()) Ok(state.events.iter().cloned().collect())
} }
/// Returns active deployments in stable service id order.
pub fn deployments(&self) -> Result<Vec<DeploymentView>, ConsoleError> {
let state = self.state()?;
state
.deployments
.iter()
.map(|(service_id, deployment)| deployment_view(&state, service_id, deployment))
.collect()
}
/// Returns the active deployment for one public service id.
pub fn deployment(&self, service_id: &str) -> Result<DeploymentView, ConsoleError> {
let state = self.state()?;
let deployment = state
.deployments
.get(service_id)
.ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))?;
deployment_view(&state, service_id, deployment)
}
fn runtime_view(&self) -> Result<RuntimeView, ConsoleError> { fn runtime_view(&self) -> Result<RuntimeView, ConsoleError> {
let runtime = self.runtime()?; let runtime = self.runtime()?;
let state = self.state()?; let state = self.state()?;
@@ -555,24 +652,52 @@ impl Console {
} }
fn build_runtime(&self) -> Result<Runtime, ConsoleError> { fn build_runtime(&self) -> Result<Runtime, ConsoleError> {
let manifests = self let state = self.state()?;
.state()? let manifests = state
.services .services
.values() .values()
.map(|service| service.manifest.clone()) .map(|service| service.manifest.clone())
.collect::<Vec<_>>(); .collect::<Vec<_>>();
let deployments = state
.deployments
.iter()
.map(|(service_id, deployment)| {
ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
})
.collect::<Result<Vec<_>, _>>()?;
drop(state);
let runtime = Runtime::new(self.runtime_config.clone())?; let runtime = Runtime::new(self.runtime_config.clone())?;
for manifest in manifests { for manifest in manifests {
runtime.register_from_file(manifest)?; runtime.register_from_file(manifest)?;
} }
for key in deployments {
runtime.start(&key, Vec::new())?;
}
Ok(runtime) Ok(runtime)
} }
fn mark_runtime_started(&self) -> Result<(), ConsoleError> { fn mark_runtime_started(&self) -> Result<(), ConsoleError> {
let mut state = self.state()?; let mut state = self.state()?;
let deployed = state
.deployments
.iter()
.map(|(service_id, deployment)| {
ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
})
.collect::<Result<Vec<_>, _>>()?;
let updated_at_ms = unix_time_ms();
state.runtime_started_at = Some(Instant::now()); state.runtime_started_at = Some(Instant::now());
for service in state.services.values_mut() { for (key, service) in &mut state.services {
service.status = ServiceStatus::Stopped; let status = if deployed.contains(key) {
ServiceStatus::Running
} else {
ServiceStatus::Stopped
};
if service.status != status {
service.status = status;
service.updated_at_ms = updated_at_ms;
}
} }
Ok(()) Ok(())
} }
@@ -642,8 +767,9 @@ impl Console {
let _ = fs::remove_file(&manifest_path); let _ = fs::remove_file(&manifest_path);
return Err(error.into()); return Err(error.into());
} }
let capabilities = runtime_capabilities(runtime, &key)?;
let view = self.insert_service(manifest, ServiceStatus::Stopped)?; let view = self.insert_service(manifest, capabilities, ServiceStatus::Stopped)?;
self.push_event( self.push_event(
EventKind::Registered, EventKind::Registered,
Some(&key), Some(&key),
@@ -686,7 +812,55 @@ impl Console {
Ok(view) Ok(view)
} }
fn invoke(&self, key: &ServiceKey, input: Vec<u8>) -> Result<InvokeResponse, ConsoleError> { fn activate_deployment(
&self,
service_id: String,
revision: String,
init_config: Vec<u8>,
) -> Result<DeploymentView, ConsoleError> {
validate_path_segment("id", &service_id)?;
validate_path_segment("revision", &revision)?;
let key = ServiceKey::new(service_id.clone(), revision.clone())?;
let _lifecycle = self
.lifecycle
.lock()
.map_err(|_| ConsoleError::LockPoisoned)?;
let runtime = self.runtime()?;
let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
self.ensure_managed(&key)?;
// Starting is idempotent. The old revision stays resident so calls
// that resolved immediately before this switch can finish safely and
// rollback does not require recompilation.
runtime.start(&key, init_config)?;
let updated_at_ms = unix_time_ms();
let view = {
let mut state = self.state()?;
let service = state
.services
.get_mut(&key)
.ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
service.status = ServiceStatus::Running;
service.updated_at_ms = updated_at_ms;
let deployment = DeploymentRecord {
active_revision: revision,
updated_at_ms,
};
state
.deployments
.insert(service_id.clone(), deployment.clone());
deployment_view(&state, &service_id, &deployment)?
};
self.push_event(
EventKind::Deployed,
Some(&key),
"deployment activated".to_owned(),
)?;
Ok(view)
}
fn invoke(&self, key: &ServiceKey, input: Vec<u8>) -> Result<InvocationResult, ConsoleError> {
let runtime = self.runtime()?; let runtime = self.runtime()?;
let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?; let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
self.ensure_managed(key)?; self.ensure_managed(key)?;
@@ -711,11 +885,7 @@ impl Console {
Some(key), Some(key),
format!("invoke completed in {latency_ms} ms"), format!("invoke completed in {latency_ms} ms"),
)?; )?;
Ok(InvokeResponse { Ok(InvocationResult { output, latency_ms })
output_base64: BASE64.encode(&output),
output_bytes: output.len(),
latency_ms,
})
} }
Err(error) => { Err(error) => {
record.errors = record.errors.saturating_add(1); record.errors = record.errors.saturating_add(1);
@@ -729,6 +899,24 @@ impl Console {
} }
} }
fn gateway_invoke(
&self,
service_id: &str,
input: Vec<u8>,
) -> Result<GatewayInvocation, ConsoleError> {
let revision = {
let state = self.state()?;
state
.deployments
.get(service_id)
.map(|deployment| deployment.active_revision.clone())
.ok_or_else(|| ConsoleError::DeploymentNotFound(service_id.to_owned()))?
};
let key = ServiceKey::new(service_id.to_owned(), revision.clone())?;
let result = self.invoke(&key, input)?;
Ok(GatewayInvocation { revision, result })
}
fn load_stored_services( fn load_stored_services(
&self, &self,
stored_services: Vec<StoredService>, stored_services: Vec<StoredService>,
@@ -747,10 +935,12 @@ impl Console {
))); )));
} }
runtime.register_from_file(manifest.clone())?; runtime.register_from_file(manifest.clone())?;
let capabilities = runtime_capabilities(runtime, &key)?;
self.state()?.services.insert( self.state()?.services.insert(
key, key,
ServiceRecord { ServiceRecord {
manifest, manifest,
capabilities,
status: ServiceStatus::Stopped, status: ServiceStatus::Stopped,
updated_at_ms: stored.updated_at_ms, updated_at_ms: stored.updated_at_ms,
calls: stored.calls, calls: stored.calls,
@@ -793,22 +983,79 @@ impl Console {
continue; continue;
} }
runtime.register_from_file(manifest.clone())?; runtime.register_from_file(manifest.clone())?;
self.insert_service(manifest, ServiceStatus::Stopped)?; let capabilities = runtime_capabilities(runtime, &key)?;
self.insert_service(manifest, capabilities, ServiceStatus::Stopped)?;
} }
Ok(()) Ok(())
} }
fn load_stored_deployments(
&self,
stored_deployments: Vec<StoredDeployment>,
) -> Result<(), ConsoleError> {
let mut state = self.state()?;
for stored in stored_deployments {
validate_path_segment("id", &stored.service_id)?;
validate_path_segment("revision", &stored.active_revision)?;
let key = ServiceKey::new(stored.service_id.clone(), stored.active_revision.clone())?;
if !state.services.contains_key(&key) {
return Err(ConsoleError::InvalidDatabaseData(format!(
"deployment for {:?} references unmanaged revision {key}",
stored.service_id
)));
}
state.deployments.insert(
stored.service_id,
DeploymentRecord {
active_revision: stored.active_revision,
updated_at_ms: stored.updated_at_ms,
},
);
}
Ok(())
}
fn restore_deployments(&self) -> Result<(), ConsoleError> {
let deployments = {
let state = self.state()?;
state
.deployments
.iter()
.map(|(service_id, deployment)| {
ServiceKey::new(service_id.clone(), deployment.active_revision.clone())
})
.collect::<Result<Vec<_>, _>>()?
};
{
let runtime = self.runtime()?;
let runtime = runtime.as_ref().ok_or(ConsoleError::RuntimeNotRunning)?;
for key in &deployments {
runtime.start(key, Vec::new())?;
}
}
let mut state = self.state()?;
for key in deployments {
let service = state
.services
.get_mut(&key)
.ok_or_else(|| ConsoleError::ServiceNotFound(key.clone()))?;
service.status = ServiceStatus::Running;
}
Ok(())
}
async fn persist_snapshot(&self) -> Result<(), ConsoleError> { async fn persist_snapshot(&self) -> Result<(), ConsoleError> {
// Serialize snapshots so a slower earlier write cannot overwrite a // Serialize snapshots so a slower earlier write cannot overwrite a
// newer state captured by a concurrent HTTP operation. // newer state captured by a concurrent HTTP operation.
let _sync = self.persistence_sync.lock().await; let _sync = self.persistence_sync.lock().await;
let (services, events) = self.persistence_snapshot()?; let (services, events, deployments) = self.persistence_snapshot()?;
self.persistence.sync(services, events).await?; self.persistence.sync(services, events, deployments).await?;
Ok(()) Ok(())
} }
fn persistence_snapshot(&self) -> Result<(Vec<StoredService>, Vec<StoredEvent>), ConsoleError> { fn persistence_snapshot(&self) -> Result<PersistenceSnapshot, ConsoleError> {
let state = self.state()?; let state = self.state()?;
let services = state let services = state
.services .services
@@ -836,18 +1083,29 @@ impl Console {
message: event.message.clone(), message: event.message.clone(),
}) })
.collect(); .collect();
Ok((services, events)) let deployments = state
.deployments
.iter()
.map(|(service_id, deployment)| StoredDeployment {
service_id: service_id.clone(),
active_revision: deployment.active_revision.clone(),
updated_at_ms: deployment.updated_at_ms,
})
.collect();
Ok((services, events, deployments))
} }
fn insert_service( fn insert_service(
&self, &self,
manifest: ServiceManifest, manifest: ServiceManifest,
capabilities: Vec<CapabilityView>,
status: ServiceStatus, status: ServiceStatus,
) -> Result<ServiceView, ConsoleError> { ) -> Result<ServiceView, ConsoleError> {
let key = manifest.key()?; let key = manifest.key()?;
let mut state = self.state()?; let mut state = self.state()?;
let record = ServiceRecord { let record = ServiceRecord {
manifest, manifest,
capabilities,
status, status,
updated_at_ms: unix_time_ms(), updated_at_ms: unix_time_ms(),
calls: 0, calls: 0,
@@ -912,6 +1170,35 @@ impl Console {
} }
} }
fn runtime_capabilities(
runtime: &Runtime,
key: &ServiceKey,
) -> Result<Vec<CapabilityView>, ConsoleError> {
Ok(runtime
.capabilities(key)?
.into_iter()
.map(Into::into)
.collect())
}
fn deployment_view(
state: &ConsoleState,
service_id: &str,
deployment: &DeploymentRecord,
) -> Result<DeploymentView, ConsoleError> {
let key = ServiceKey::new(service_id.to_owned(), deployment.active_revision.clone())?;
let service = state
.services
.get(&key)
.ok_or_else(|| ConsoleError::ServiceNotFound(key))?;
Ok(DeploymentView {
service_id: service_id.to_owned(),
active_revision: deployment.active_revision.clone(),
status: service.status,
updated_at_ms: deployment.updated_at_ms,
})
}
/// Builds the Axum management API around a shared [`Console`]. /// Builds the Axum management API around a shared [`Console`].
/// ///
/// The returned router exposes Runtime control, component registration, /// The returned router exposes Runtime control, component registration,
@@ -935,6 +1222,12 @@ pub fn app(console: Arc<Console>, allowed_origins: Vec<HeaderValue>) -> Router {
.route("/api/v1/runtime/stop", post(stop_runtime)) .route("/api/v1/runtime/stop", post(stop_runtime))
.route("/api/v1/runtime/restart", post(restart_runtime)) .route("/api/v1/runtime/restart", post(restart_runtime))
.route("/api/v1/services", get(list_services).post(register)) .route("/api/v1/services", get(list_services).post(register))
.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( .route(
"/api/v1/wit/packages", "/api/v1/wit/packages",
get(list_wit_packages).post(publish_wit_package), get(list_wit_packages).post(publish_wit_package),
@@ -971,10 +1264,71 @@ pub fn app(console: Arc<Console>, allowed_origins: Vec<HeaderValue>) -> Router {
.with_state(console) .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> { async fn health() -> Json<serde_json::Value> {
Json(json!({ "status": "ok" })) 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( async fn runtime_status(
State(console): State<Arc<Console>>, State(console): State<Arc<Console>>,
) -> Result<Json<RuntimeView>, ApiError> { ) -> Result<Json<RuntimeView>, ApiError> {
@@ -1007,6 +1361,38 @@ async fn list_services(State(console): State<Arc<Console>>) -> Result<Json<Servi
})) }))
} }
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> { async fn list_events(State(console): State<Arc<Console>>) -> Result<Json<EventList>, ApiError> {
Ok(Json(EventList { Ok(Json(EventList {
events: console.events()?, events: console.events()?,
@@ -1200,9 +1586,12 @@ async fn invoke_service(
let input = BASE64.decode(request.input_base64).map_err(|error| { let input = BASE64.decode(request.input_base64).map_err(|error| {
ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}")) ConsoleError::InvalidRequest(format!("input_base64 is invalid: {error}"))
})?; })?;
Ok(Json( let result = run_blocking(console, move |console| console.invoke(&key, input)).await?;
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> async fn run_blocking<T, F>(console: Arc<Console>, operation: F) -> Result<T, ConsoleError>
@@ -1298,6 +1687,104 @@ fn duration_ms(duration: std::time::Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) 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::Package(_)
| ConsoleError::WitRegistry(_)
| ConsoleError::Storage { .. }
| ConsoleError::ManifestSerialize(_)
| ConsoleError::Database(_)
| 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); struct ApiError(ConsoleError);
impl From<ConsoleError> for ApiError { impl From<ConsoleError> for ApiError {
@@ -1317,6 +1804,7 @@ impl IntoResponse for ApiError {
let (status, code) = match &self.0 { let (status, code) = match &self.0 {
ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"), ConsoleError::InvalidRequest(_) => (StatusCode::BAD_REQUEST, "invalid_request"),
ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"), ConsoleError::ServiceNotFound(_) => (StatusCode::NOT_FOUND, "service_not_found"),
ConsoleError::DeploymentNotFound(_) => (StatusCode::NOT_FOUND, "deployment_not_found"),
ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"), ConsoleError::RuntimeNotRunning => (StatusCode::CONFLICT, "runtime_not_running"),
ConsoleError::ArtifactTooLarge { .. } ConsoleError::ArtifactTooLarge { .. }
| ConsoleError::Package(PackageError::ComponentTooLarge { .. }) | ConsoleError::Package(PackageError::ComponentTooLarge { .. })
+33 -9
View File
@@ -1,11 +1,11 @@
//! Standalone Wasmeld management server entry point. //! Standalone Wasmeld control-plane and data-plane server entry point.
use std::{env, net::SocketAddr, path::PathBuf, sync::Arc}; use std::{env, net::SocketAddr, path::PathBuf, sync::Arc};
use axum::http::HeaderValue; use axum::http::HeaderValue;
use tracing::info; use tracing::info;
use tracing_subscriber::EnvFilter; use tracing_subscriber::EnvFilter;
use wasmeld_console::{Console, ConsoleConfig, app}; use wasmeld_console::{Console, ConsoleConfig, app, gateway_app};
#[tokio::main] #[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> { async fn main() -> Result<(), Box<dyn std::error::Error>> {
@@ -16,9 +16,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
) )
.init(); .init();
let address = env::var("WASMELD_ADDR") let management_address = env::var("WASMELD_ADDR")
.unwrap_or_else(|_| "127.0.0.1:8080".to_owned()) .unwrap_or_else(|_| "127.0.0.1:8080".to_owned())
.parse::<SocketAddr>()?; .parse::<SocketAddr>()?;
let gateway_address = env::var("WASMELD_GATEWAY_ADDR")
.unwrap_or_else(|_| "0.0.0.0:8081".to_owned())
.parse::<SocketAddr>()?;
let artifact_dir = env::var_os("WASMELD_ARTIFACT_DIR") let artifact_dir = env::var_os("WASMELD_ARTIFACT_DIR")
.map(PathBuf::from) .map(PathBuf::from)
.unwrap_or_else(|| PathBuf::from("var/wasmeld/components")); .unwrap_or_else(|| PathBuf::from("var/wasmeld/components"));
@@ -38,13 +41,26 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}) })
.await?, .await?,
); );
let application = app(console, allowed_origins); let management = app(Arc::clone(&console), allowed_origins);
let listener = tokio::net::TcpListener::bind(address).await?; let gateway = gateway_app(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);
let signal_task = tokio::spawn(async move {
shutdown_signal().await;
let _ = shutdown_sender.send(true);
});
info!(%address, "Wasmeld listening"); info!(%management_address, "Wasmeld management API listening");
axum::serve(listener, application) info!(%gateway_address, "Wasmeld Gateway listening");
.with_graceful_shutdown(shutdown_signal()) let result = tokio::try_join!(
.await?; axum::serve(management_listener, management)
.with_graceful_shutdown(wait_for_shutdown(shutdown_receiver.clone())),
axum::serve(gateway_listener, gateway)
.with_graceful_shutdown(wait_for_shutdown(shutdown_receiver)),
);
signal_task.abort();
result?;
Ok(()) Ok(())
} }
@@ -63,3 +79,11 @@ async fn shutdown_signal() {
tracing::warn!("failed to install Ctrl+C handler"); tracing::warn!("failed to install Ctrl+C handler");
} }
} }
async fn wait_for_shutdown(mut receiver: tokio::sync::watch::Receiver<bool>) {
while !*receiver.borrow() {
if receiver.changed().await.is_err() {
break;
}
}
}
+47 -6
View File
@@ -1,7 +1,8 @@
//! Toasty models and transactional libSQL snapshots for Console control data. //! Toasty models and transactional libSQL snapshots for Console control data.
//! //!
//! Component bytes and WIT packages remain filesystem artifacts. This module //! Component bytes and WIT packages remain filesystem artifacts. This module
//! persists only service manifests, metrics, and the bounded event history. //! persists only service manifests, deployments, metrics, and the bounded
//! event history.
use std::path::Path; use std::path::Path;
@@ -31,31 +32,62 @@ pub(crate) struct StoredEvent {
pub message: String, pub message: String,
} }
#[derive(Clone, Debug, toasty::Model)]
pub(crate) struct StoredDeployment {
#[key]
pub service_id: String,
pub active_revision: String,
pub updated_at_ms: u64,
}
/// Serialized access to the Toasty database connection. /// Serialized access to the Toasty database connection.
pub(crate) struct Persistence { pub(crate) struct Persistence {
db: Mutex<Db>, db: Mutex<Db>,
} }
impl Persistence { impl Persistence {
/// Opens the database and installs the schema for a new file. /// Opens the database and installs or advances its schema.
pub async fn open(path: &Path) -> toasty::Result<Self> { pub async fn open(path: &Path) -> toasty::Result<Self> {
let is_new_database = !path.exists(); let is_new_database = !path.exists();
let db = Db::builder() let db = Db::builder()
.models(toasty::models!(StoredService, StoredEvent)) .models(toasty::models!(
StoredService,
StoredEvent,
StoredDeployment
))
.build(Turso::file(path)) .build(Turso::file(path))
.await?; .await?;
if is_new_database { if is_new_database {
db.push_schema().await?; db.push_schema().await?;
} else {
// This project predates Toasty's migration history. Keep the
// additive bootstrap explicit so existing databases can be opened
// without recreating the service and event tables.
let mut db = db;
toasty::sql::statement(
r#"CREATE TABLE IF NOT EXISTS "stored_deployments" (
"service_id" TEXT NOT NULL,
"active_revision" TEXT NOT NULL,
"updated_at_ms" INTEGER NOT NULL,
PRIMARY KEY ("service_id")
)"#,
)
.exec(&mut db)
.await?;
return Ok(Self { db: Mutex::new(db) });
} }
Ok(Self { db: Mutex::new(db) }) Ok(Self { db: Mutex::new(db) })
} }
/// Loads the complete service set and retained event history. /// Loads the complete service and deployment sets plus retained events.
pub async fn load(&self) -> toasty::Result<(Vec<StoredService>, Vec<StoredEvent>)> { pub async fn load(
&self,
) -> toasty::Result<(Vec<StoredService>, Vec<StoredEvent>, Vec<StoredDeployment>)> {
let mut db = self.db.lock().await; let mut db = self.db.lock().await;
let services = StoredService::all().exec(&mut *db).await?; let services = StoredService::all().exec(&mut *db).await?;
let events = StoredEvent::all().exec(&mut *db).await?; let events = StoredEvent::all().exec(&mut *db).await?;
Ok((services, events)) let deployments = StoredDeployment::all().exec(&mut *db).await?;
Ok((services, events, deployments))
} }
/// Atomically upserts a Console snapshot and prunes expired events. /// Atomically upserts a Console snapshot and prunes expired events.
@@ -63,6 +95,7 @@ impl Persistence {
&self, &self,
services: Vec<StoredService>, services: Vec<StoredService>,
events: Vec<StoredEvent>, events: Vec<StoredEvent>,
deployments: Vec<StoredDeployment>,
) -> toasty::Result<()> { ) -> toasty::Result<()> {
let mut db = self.db.lock().await; let mut db = self.db.lock().await;
let mut tx = db.transaction().await?; let mut tx = db.transaction().await?;
@@ -96,6 +129,14 @@ impl Persistence {
.await?; .await?;
} }
for deployment in deployments {
StoredDeployment::upsert_by_service_id(deployment.service_id)
.active_revision(deployment.active_revision)
.updated_at_ms(deployment.updated_at_ms)
.exec(&mut tx)
.await?;
}
tx.commit().await tx.commit().await
} }
} }
+248 -2
View File
@@ -15,7 +15,7 @@ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use serde_json::{Value, json}; use serde_json::{Value, json};
use tempfile::TempDir; use tempfile::TempDir;
use tower::ServiceExt; use tower::ServiceExt;
use wasmeld_console::{Console, ConsoleConfig, app}; use wasmeld_console::{Console, ConsoleConfig, app, gateway_app};
use wasmeld_package::{ use wasmeld_package::{
module::{ModuleLock, sync_dependencies}, module::{ModuleLock, sync_dependencies},
wit_package::build_wit_package, wit_package::build_wit_package,
@@ -40,6 +40,7 @@ async fn manages_a_resident_component_over_http() {
assert_eq!(status, StatusCode::CREATED, "{registered}"); assert_eq!(status, StatusCode::CREATED, "{registered}");
assert_eq!(registered["id"], "echo"); assert_eq!(registered["id"], "echo");
assert_eq!(registered["status"], "stopped"); assert_eq!(registered["status"], "stopped");
assert_eq!(registered["capabilities"], json!([]));
let response = application let response = application
.clone() .clone()
@@ -89,6 +90,234 @@ async fn manages_a_resident_component_over_http() {
assert!(events["events"].as_array().unwrap().len() >= 5); assert!(events["events"].as_array().unwrap().len() >= 5);
} }
#[tokio::test]
async fn reports_exact_component_host_capabilities() {
let artifact_dir = TempDir::new().expect("temporary artifact directory");
let application = test_app(artifact_dir.path()).await;
let component = fs::read(component_artifact("clock_probe_component.wasm"))
.expect("clock probe component should be readable");
let response = application
.clone()
.oneshot(package_request("clock-probe", "0.1.0", &component))
.await
.expect("register request should complete");
assert_eq!(response.status(), StatusCode::CREATED);
let registered = response_json(response).await;
assert_eq!(
registered["capabilities"],
json!([{
"interface": "wasmeld:clock/monotonic-clock@0.1.0",
"package": "wasmeld:clock",
"name": "monotonic-clock",
"version": "0.1.0"
}])
);
let response = application
.oneshot(get_request("/api/v1/services"))
.await
.expect("service list request should complete");
let services = response_json(response).await;
assert_eq!(
services["services"][0]["capabilities"],
registered["capabilities"]
);
}
#[tokio::test]
async fn routes_raw_gateway_calls_through_the_active_deployment() {
let artifact_dir = TempDir::new().expect("temporary artifact directory");
let (management, gateway) = test_apps(artifact_dir.path()).await;
let component = fs::read(echo_component()).expect("echo component should be readable");
let response = gateway
.clone()
.oneshot(raw_request(
"/v1/services/echo/invoke",
"application/octet-stream",
b"not deployed",
))
.await
.expect("gateway request should complete");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
assert_eq!(
response_json(response).await["error"]["code"],
"service_not_found"
);
let response = gateway
.clone()
.oneshot(get_request("/api/v1/services"))
.await
.expect("gateway route isolation request should complete");
assert_eq!(response.status(), StatusCode::NOT_FOUND);
for revision in ["0.1.0", "0.2.0"] {
let response = management
.clone()
.oneshot(package_request("echo", revision, &component))
.await
.expect("register request should complete");
assert_eq!(response.status(), StatusCode::CREATED);
}
let response = management
.clone()
.oneshot(json_request(
"/api/v1/deployments/echo/activate",
json!({ "revision": "0.1.0" }),
))
.await
.expect("activation request should complete");
assert_eq!(response.status(), StatusCode::OK);
let deployment = response_json(response).await;
assert_eq!(deployment["active_revision"], "0.1.0");
assert_eq!(deployment["status"], "running");
let response = gateway
.clone()
.oneshot(raw_request(
"/v1/services/echo/invoke",
"application/octet-stream",
b"public bytes",
))
.await
.expect("gateway invocation should complete");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
response.headers()["content-type"],
"application/octet-stream"
);
assert_eq!(response.headers()["x-wasmeld-revision"], "0.1.0");
assert_eq!(
to_bytes(response.into_body(), usize::MAX).await.unwrap(),
b"public bytes".as_slice()
);
let response = management
.clone()
.oneshot(json_request(
"/api/v1/deployments/echo/activate",
json!({ "revision": "0.2.0" }),
))
.await
.expect("switch request should complete");
assert_eq!(response.status(), StatusCode::OK);
let response = gateway
.clone()
.oneshot(raw_request(
"/v1/services/echo/invoke",
"application/octet-stream",
b"switched",
))
.await
.expect("gateway invocation should complete");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["x-wasmeld-revision"], "0.2.0");
let response = gateway
.clone()
.oneshot(raw_request(
"/v1/services/echo/invoke",
"application/json",
b"{}",
))
.await
.expect("unsupported media request should complete");
assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
assert_eq!(
response_json(response).await["error"]["code"],
"unsupported_media_type"
);
let response = management
.clone()
.oneshot(empty_post("/api/v1/services/echo/0.2.0/stop"))
.await
.expect("active actor stop request should complete");
assert_eq!(response.status(), StatusCode::OK);
let response = gateway
.oneshot(raw_request(
"/v1/services/echo/invoke",
"application/octet-stream",
b"stopped",
))
.await
.expect("stopped deployment request should complete");
assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(
response_json(response).await["error"]["code"],
"service_unavailable"
);
let response = management
.oneshot(get_request("/api/v1/deployments"))
.await
.expect("deployment list request should complete");
let deployments = response_json(response).await;
assert_eq!(deployments["deployments"].as_array().unwrap().len(), 1);
assert_eq!(deployments["deployments"][0]["active_revision"], "0.2.0");
}
#[tokio::test]
async fn restores_the_active_deployment_with_a_fresh_actor() {
let artifact_dir = TempDir::new().expect("temporary artifact directory");
let component = fs::read(echo_component()).expect("echo component should be readable");
{
let (management, gateway) = test_apps(artifact_dir.path()).await;
let response = management
.clone()
.oneshot(package_request("restored", "1.0.0", &component))
.await
.expect("register request should complete");
assert_eq!(response.status(), StatusCode::CREATED);
let response = management
.oneshot(json_request(
"/api/v1/deployments/restored/activate",
json!({ "revision": "1.0.0" }),
))
.await
.expect("activation request should complete");
assert_eq!(response.status(), StatusCode::OK);
let response = gateway
.oneshot(raw_request(
"/v1/services/restored/invoke",
"application/octet-stream",
b"before restart",
))
.await
.expect("gateway invocation should complete");
assert_eq!(response.status(), StatusCode::OK);
}
let (management, gateway) = test_apps(artifact_dir.path()).await;
let response = management
.oneshot(get_request("/api/v1/deployments/restored"))
.await
.expect("restored deployment request should complete");
let deployment = response_json(response).await;
assert_eq!(deployment["active_revision"], "1.0.0");
assert_eq!(deployment["status"], "running");
let response = gateway
.oneshot(raw_request(
"/v1/services/restored/invoke",
"application/octet-stream",
b"after restart",
))
.await
.expect("restored gateway invocation should complete");
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(
to_bytes(response.into_body(), usize::MAX).await.unwrap(),
b"after restart".as_slice()
);
}
#[tokio::test] #[tokio::test]
async fn reloads_manifests_without_restoring_actor_memory() { async fn reloads_manifests_without_restoring_actor_memory() {
let artifact_dir = TempDir::new().expect("temporary artifact directory"); let artifact_dir = TempDir::new().expect("temporary artifact directory");
@@ -473,6 +702,10 @@ async fn client_fetches_transitive_wit_dependencies_from_the_registry() {
} }
async fn test_app(artifact_dir: &Path) -> Router { async fn test_app(artifact_dir: &Path) -> Router {
test_apps(artifact_dir).await.0
}
async fn test_apps(artifact_dir: &Path) -> (Router, Router) {
let console = Console::new(ConsoleConfig { let console = Console::new(ConsoleConfig {
artifact_dir: artifact_dir.to_path_buf(), artifact_dir: artifact_dir.to_path_buf(),
wit_registry_dir: artifact_dir.join("wit-packages"), wit_registry_dir: artifact_dir.join("wit-packages"),
@@ -481,13 +714,16 @@ async fn test_app(artifact_dir: &Path) -> Router {
}) })
.await .await
.expect("console should start"); .expect("console should start");
app(Arc::new(console), Vec::new()) let console = Arc::new(console);
(app(Arc::clone(&console), Vec::new()), gateway_app(console))
} }
fn package_request(id: &str, revision: &str, component: &[u8]) -> Request<Body> { fn package_request(id: &str, revision: &str, component: &[u8]) -> Request<Body> {
let mut package = Cursor::new(Vec::new()); let mut package = Cursor::new(Vec::new());
let world = if id == "counter" { let world = if id == "counter" {
"component:counter/counter-component@0.1.0" "component:counter/counter-component@0.1.0"
} else if id == "clock-probe" {
"component:clock-probe/clock-probe-component@0.1.0"
} else { } else {
"component:echo/echo-component@0.1.0" "component:echo/echo-component@0.1.0"
}; };
@@ -552,6 +788,15 @@ fn json_request(uri: &str, value: Value) -> Request<Body> {
.unwrap() .unwrap()
} }
fn raw_request(uri: &str, content_type: &str, body: &[u8]) -> Request<Body> {
Request::builder()
.method("POST")
.uri(uri)
.header(header::CONTENT_TYPE, content_type)
.body(Body::from(body.to_vec()))
.unwrap()
}
fn empty_post(uri: &str) -> Request<Body> { fn empty_post(uri: &str) -> Request<Body> {
Request::builder() Request::builder()
.method("POST") .method("POST")
@@ -580,6 +825,7 @@ fn component_artifact(name: &str) -> PathBuf {
for manifest in [ for manifest in [
"components/echo/Cargo.toml", "components/echo/Cargo.toml",
"components/counter/Cargo.toml", "components/counter/Cargo.toml",
"components/clock-probe/Cargo.toml",
] { ] {
let status = Command::new("rustup") let status = Command::new("rustup")
.args([ .args([
-9
View File
@@ -5,12 +5,3 @@ wasmtime::component::bindgen!({
path: "../../wit/service", path: "../../wit/service",
world: "service-component", world: "service-component",
}); });
pub(crate) mod clock {
// Host capabilities are generated separately so adding a capability does
// not expand the baseline service contract for every component.
wasmtime::component::bindgen!({
path: "../../wit/clock",
world: "clock-host",
});
}
@@ -0,0 +1,44 @@
//! `wasmeld:clock/monotonic-clock@0.1.0` host implementation.
use std::time::Instant;
use wasmtime::component::Linker;
use crate::{RuntimeError, runtime::HostState};
wasmtime::component::bindgen!({
path: "../../wit/clock",
world: "clock-host",
});
pub(crate) const INTERFACE: &str = "wasmeld:clock/monotonic-clock@0.1.0";
pub(crate) const PACKAGE: &str = "wasmeld:clock";
pub(crate) const NAME: &str = "monotonic-clock";
pub(crate) const VERSION: &str = "0.1.0";
pub(crate) struct ClockState {
origin: Instant,
}
impl ClockState {
pub(crate) fn new() -> Self {
Self {
origin: Instant::now(),
}
}
pub(crate) fn now(&self) -> u64 {
u64::try_from(self.origin.elapsed().as_nanos()).unwrap_or(u64::MAX)
}
}
impl wasmeld::clock::monotonic_clock::Host for HostState {
fn now(&mut self) -> u64 {
self.capabilities.monotonic_clock_now()
}
}
pub(crate) fn add_to_linker(linker: &mut Linker<HostState>) -> Result<(), RuntimeError> {
wasmeld::clock::monotonic_clock::add_to_linker::<HostState, HostState>(linker, |state| state)
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))
}
@@ -0,0 +1,29 @@
//! Versioned host capabilities available to WebAssembly Components.
mod clock;
mod registry;
pub use registry::CapabilityDescriptor;
pub(crate) use registry::{Capability, CapabilityRegistry};
/// Per-Actor state owned by enabled host capabilities.
pub(crate) struct HostCapabilities {
clock: Option<clock::ClockState>,
}
impl HostCapabilities {
pub(crate) fn new(capabilities: &[Capability]) -> Self {
Self {
clock: capabilities
.contains(&Capability::MonotonicClock)
.then(clock::ClockState::new),
}
}
fn monotonic_clock_now(&self) -> u64 {
self.clock
.as_ref()
.expect("the Clock interface is linked only when its state is enabled")
.now()
}
}
@@ -0,0 +1,120 @@
//! Exact interface lookup and Linker installation for host capabilities.
use wasmtime::component::Linker;
use super::clock;
use crate::{RuntimeError, runtime::HostState};
/// Stable description of one versioned WIT interface used by a Component.
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct CapabilityDescriptor {
interface: &'static str,
package: &'static str,
name: &'static str,
version: &'static str,
}
impl CapabilityDescriptor {
const fn new(
interface: &'static str,
package: &'static str,
name: &'static str,
version: &'static str,
) -> Self {
Self {
interface,
package,
name,
version,
}
}
/// Returns the fully qualified and versioned WIT interface identity.
pub const fn interface(&self) -> &'static str {
self.interface
}
/// Returns the versioned WIT package identity without the version suffix.
pub const fn package(&self) -> &'static str {
self.package
}
/// Returns the WIT interface name inside the package.
pub const fn name(&self) -> &'static str {
self.name
}
/// Returns the exact semantic version implemented by the Runtime.
pub const fn version(&self) -> &'static str {
self.version
}
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub(crate) enum Capability {
MonotonicClock,
}
impl Capability {
pub(crate) const fn descriptor(self) -> CapabilityDescriptor {
match self {
Self::MonotonicClock => CapabilityDescriptor::new(
clock::INTERFACE,
clock::PACKAGE,
clock::NAME,
clock::VERSION,
),
}
}
pub(crate) fn add_to_linker(self, linker: &mut Linker<HostState>) -> Result<(), RuntimeError> {
match self {
Self::MonotonicClock => clock::add_to_linker(linker),
}
}
}
/// Registry of the exact WIT interface versions implemented by this Runtime.
pub(crate) struct CapabilityRegistry;
impl CapabilityRegistry {
/// Resolves an exact Component import to a supported host capability.
pub(crate) fn resolve(interface: &str) -> Option<Capability> {
match interface {
clock::INTERFACE => Some(Capability::MonotonicClock),
_ => None,
}
}
/// Installs only the capabilities requested by one Component.
pub(crate) fn add_to_linker(
linker: &mut Linker<HostState>,
capabilities: &[Capability],
) -> Result<(), RuntimeError> {
for capability in capabilities {
capability.add_to_linker(linker)?;
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::{Capability, CapabilityRegistry};
#[test]
fn resolves_only_the_exact_supported_interface_version() {
assert_eq!(
CapabilityRegistry::resolve("wasmeld:clock/monotonic-clock@0.1.0"),
Some(Capability::MonotonicClock)
);
assert_eq!(
CapabilityRegistry::resolve("wasmeld:clock/monotonic-clock@0.2.0"),
None
);
assert_eq!(
CapabilityRegistry::resolve("wasmeld:clock/other@0.1.0"),
None
);
}
}
+2
View File
@@ -6,10 +6,12 @@
//! service exports, and invokes it through one serial Actor. //! service exports, and invokes it through one serial Actor.
mod bindings; mod bindings;
mod capability;
mod error; mod error;
mod manifest; mod manifest;
mod runtime; mod runtime;
pub use capability::CapabilityDescriptor;
pub use error::RuntimeError; pub use error::RuntimeError;
pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest}; pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest};
pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats}; pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats};
+35 -41
View File
@@ -30,8 +30,9 @@ use wasmtime_wasi::{
}; };
use crate::{ use crate::{
RuntimeError, ServiceKey, ServiceManifest, CapabilityDescriptor, RuntimeError, ServiceKey, ServiceManifest,
bindings::{ServiceComponent, ServiceError, clock as clock_bindings}, bindings::{ServiceComponent, ServiceError},
capability::{Capability, CapabilityRegistry, HostCapabilities},
manifest::ResourceLimits, manifest::ResourceLimits,
}; };
@@ -40,7 +41,6 @@ const MAX_EPOCH_TICK: Duration = Duration::from_millis(10);
const DEFAULT_MAX_WASM_STACK: usize = 512 * 1024; const DEFAULT_MAX_WASM_STACK: usize = 512 * 1024;
const ACTOR_RESOURCE_COUNT_LIMIT: usize = 16; const ACTOR_RESOURCE_COUNT_LIMIT: usize = 16;
const ACTOR_TABLE_ELEMENT_LIMIT: usize = 16 * 1024; const ACTOR_TABLE_ELEMENT_LIMIT: usize = 16 * 1024;
const MONOTONIC_CLOCK_IMPORT: &str = "wasmeld:clock/monotonic-clock@0.1.0";
const ALLOWED_WASI_IMPORTS: &[&str] = &[ const ALLOWED_WASI_IMPORTS: &[&str] = &[
"wasi:cli/environment@", "wasi:cli/environment@",
"wasi:cli/exit@", "wasi:cli/exit@",
@@ -162,12 +162,7 @@ impl Drop for EpochTicker {
struct RegisteredService { struct RegisteredService {
manifest: ServiceManifest, manifest: ServiceManifest,
component: Arc<Component>, component: Arc<Component>,
capabilities: Vec<HostCapability>, capabilities: Vec<Capability>,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum HostCapability {
MonotonicClock,
} }
impl Runtime { impl Runtime {
@@ -327,6 +322,26 @@ impl Runtime {
actor.invoke(input) actor.invoke(input)
} }
/// Returns the exact versioned host capabilities imported by a service.
pub fn capabilities(
&self,
key: &ServiceKey,
) -> Result<Vec<CapabilityDescriptor>, RuntimeError> {
let services = self
.inner
.services
.lock()
.map_err(|_| RuntimeError::LockPoisoned)?;
let service = services
.get(key)
.ok_or_else(|| RuntimeError::ServiceNotRegistered(key.clone()))?;
Ok(service
.capabilities
.iter()
.map(|capability| capability.descriptor())
.collect())
}
/// Stops one Actor and releases its Store and Component Instance. /// Stops one Actor and releases its Store and Component Instance.
pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> { pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> {
let _lifecycle = self let _lifecycle = self
@@ -557,15 +572,15 @@ impl ActorStatus {
} }
} }
struct HostState { pub(crate) struct HostState {
store_limits: StoreLimits, store_limits: StoreLimits,
wasi: WasiCtx, wasi: WasiCtx,
wasi_table: ResourceTable, wasi_table: ResourceTable,
clock_origin: Instant, pub(crate) capabilities: HostCapabilities,
} }
impl HostState { impl HostState {
fn new(limits: &ResourceLimits) -> Self { fn new(limits: &ResourceLimits, capabilities: &[Capability]) -> Self {
let store_limits = StoreLimitsBuilder::new() let store_limits = StoreLimitsBuilder::new()
.memory_size(limits.memory_bytes) .memory_size(limits.memory_bytes)
.table_elements(ACTOR_TABLE_ELEMENT_LIMIT) .table_elements(ACTOR_TABLE_ELEMENT_LIMIT)
@@ -585,7 +600,7 @@ impl HostState {
store_limits, store_limits,
wasi: wasi.build(), wasi: wasi.build(),
wasi_table: ResourceTable::new(), wasi_table: ResourceTable::new(),
clock_origin: Instant::now(), capabilities: HostCapabilities::new(capabilities),
} }
} }
} }
@@ -594,12 +609,6 @@ impl HasData for HostState {
type Data<'a> = &'a mut HostState; type Data<'a> = &'a mut HostState;
} }
impl clock_bindings::wasmeld::clock::monotonic_clock::Host for HostState {
fn now(&mut self) -> u64 {
u64::try_from(self.clock_origin.elapsed().as_nanos()).unwrap_or(u64::MAX)
}
}
impl WasiView for HostState { impl WasiView for HostState {
fn ctx(&mut self) -> WasiCtxView<'_> { fn ctx(&mut self) -> WasiCtxView<'_> {
WasiCtxView { WasiCtxView {
@@ -639,12 +648,12 @@ impl ActorWorker {
}); });
} }
let mut store = Store::new(&engine, HostState::new(&limits)); let mut store = Store::new(&engine, HostState::new(&limits, &service.capabilities));
store.limiter(|state| &mut state.store_limits); store.limiter(|state| &mut state.store_limits);
let mut linker = Linker::new(&engine); let mut linker = Linker::new(&engine);
add_restricted_wasi(&mut linker)?; add_restricted_wasi(&mut linker)?;
add_host_capabilities(&mut linker, &service.capabilities)?; CapabilityRegistry::add_to_linker(&mut linker, &service.capabilities)?;
let epoch_ticks = ticks_for(limits.deadline(), epoch_tick); let epoch_ticks = ticks_for(limits.deadline(), epoch_tick);
configure_call_budget(&mut store, &limits, epoch_ticks)?; configure_call_budget(&mut store, &limits, epoch_ticks)?;
@@ -758,33 +767,16 @@ fn link_wasi(result: wasmtime::Result<()>) -> Result<(), RuntimeError> {
result.map_err(|error| RuntimeError::ActorInitialization(error.to_string())) result.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))
} }
fn add_host_capabilities(
linker: &mut Linker<HostState>,
capabilities: &[HostCapability],
) -> Result<(), RuntimeError> {
for capability in capabilities {
match capability {
HostCapability::MonotonicClock => link_wasi(
clock_bindings::wasmeld::clock::monotonic_clock::add_to_linker::<
HostState,
HostState,
>(linker, |state| state),
)?,
}
}
Ok(())
}
fn validate_component_imports( fn validate_component_imports(
component: &Component, component: &Component,
engine: &Engine, engine: &Engine,
) -> Result<Vec<HostCapability>, RuntimeError> { ) -> Result<Vec<Capability>, RuntimeError> {
// Import validation is deny-by-default. A host function is linked only // Import validation is deny-by-default. A host function is linked only
// after its exact WIT identity or WASI family has passed this allowlist. // after its exact WIT identity or WASI family has passed this allowlist.
let mut capabilities = Vec::new(); let mut capabilities = Vec::new();
for (name, _) in component.component_type().imports(engine) { for (name, _) in component.component_type().imports(engine) {
if name == MONOTONIC_CLOCK_IMPORT { if let Some(capability) = CapabilityRegistry::resolve(name) {
capabilities.push(HostCapability::MonotonicClock); capabilities.push(capability);
} else if !ALLOWED_WASI_IMPORTS } else if !ALLOWED_WASI_IMPORTS
.iter() .iter()
.any(|allowed| name.starts_with(allowed)) .any(|allowed| name.starts_with(allowed))
@@ -793,6 +785,8 @@ fn validate_component_imports(
} }
} }
capabilities.sort_unstable();
capabilities.dedup();
Ok(capabilities) Ok(capabilities)
} }
@@ -16,6 +16,7 @@ fn echo_component_round_trips_bytes() {
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
let (key, actor) = start_component(&runtime, "echo", "echo_component.wasm"); let (key, actor) = start_component(&runtime, "echo", "echo_component.wasm");
assert!(runtime.capabilities(&key).unwrap().is_empty());
assert_eq!(actor.invoke(b"hello wasm".to_vec()).unwrap(), b"hello wasm"); assert_eq!(actor.invoke(b"hello wasm".to_vec()).unwrap(), b"hello wasm");
runtime.stop(&key).unwrap(); runtime.stop(&key).unwrap();
@@ -171,6 +172,13 @@ fn explicit_clock_capability_is_linked() {
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start"); let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
let (key, actor) = start_component(&runtime, "clock-probe", "clock_probe_component.wasm"); let (key, actor) = start_component(&runtime, "clock-probe", "clock_probe_component.wasm");
let capabilities = runtime.capabilities(&key).unwrap();
assert_eq!(capabilities.len(), 1);
assert_eq!(
capabilities[0].interface(),
"wasmeld:clock/monotonic-clock@0.1.0"
);
let first = actor.invoke(b"first".to_vec()).unwrap(); let first = actor.invoke(b"first".to_vec()).unwrap();
let second = actor.invoke(b"second".to_vec()).unwrap(); let second = actor.invoke(b"second".to_vec()).unwrap();
assert_eq!(&first[8..], b"first"); assert_eq!(&first[8..], b"first");