mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-30 05:02:58 +00:00
feat(hosts): bind replica and session lifetime to origin authority
This commit is contained in:
Mohamed Boudra
@@ -28,6 +28,10 @@
|
||||
"./node": {
|
||||
"types": "./dist/node/index.d.ts",
|
||||
"default": "./dist/node/index.js"
|
||||
},
|
||||
"./internal/host-connection": {
|
||||
"types": "./dist/hosts/connection.d.ts",
|
||||
"default": "./dist/hosts/connection.js"
|
||||
}
|
||||
},
|
||||
"publishConfig": {
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
import { safeRandomId } from "../daemon-client-transport-utils.js";
|
||||
import { DaemonClient } from "../daemon-client.js";
|
||||
import type { DaemonTransportFactory } from "../daemon-client-transport-types.js";
|
||||
|
||||
export function createHostChannelTransport(
|
||||
origin: DaemonClient,
|
||||
serverId: string,
|
||||
scopeId: string,
|
||||
initialRouteId: string | null,
|
||||
link = false,
|
||||
): DaemonTransportFactory {
|
||||
let first: string | null = initialRouteId;
|
||||
return () => {
|
||||
const opens = new Set<() => void>();
|
||||
const closes = new Set<(event?: unknown) => void>();
|
||||
const messages = new Set<(data: unknown, binary: boolean) => void>();
|
||||
const errors = new Set<(error: unknown) => void>();
|
||||
let routeId: string | null = null;
|
||||
let closed = false;
|
||||
const cleanups: (() => void)[] = [];
|
||||
const close = (error?: unknown) => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
if (error instanceof Error) Object.assign(error, { serverId });
|
||||
for (const cleanup of cleanups) cleanup();
|
||||
try {
|
||||
if (routeId) origin.closeHostRoute(routeId);
|
||||
else origin.cancelHostRoute(scopeId);
|
||||
} catch {
|
||||
/* Transport already closed. */
|
||||
}
|
||||
if (error) for (const handler of errors) handler(error);
|
||||
for (const handler of closes) handler(error);
|
||||
};
|
||||
cleanups.push(
|
||||
origin.subscribeRawMessages((message) => {
|
||||
if (message.type === "hosts.route.frame" && message.payload.subscriptionId === routeId) {
|
||||
for (const handler of messages) handler(message.payload.data, false);
|
||||
}
|
||||
if (message.type === "hosts.route.closed" && message.payload.subscriptionId === routeId)
|
||||
close(
|
||||
Object.assign(new Error(message.payload.message), {
|
||||
code: message.payload.code,
|
||||
serverId,
|
||||
}),
|
||||
);
|
||||
}),
|
||||
);
|
||||
cleanups.push(
|
||||
origin.subscribeHostFrames((frame) => {
|
||||
if (frame.routeId === routeId) for (const handler of messages) handler(frame.payload, true);
|
||||
}),
|
||||
);
|
||||
cleanups.push(
|
||||
origin.subscribeConnectionStatus((connectionState) => {
|
||||
if (connectionState.status !== "connected") close();
|
||||
}),
|
||||
);
|
||||
let opening: Promise<string>;
|
||||
if (first) opening = Promise.resolve(first);
|
||||
else opening = link ? origin.openHostLink(scopeId) : origin.openHostRoute(serverId, scopeId);
|
||||
first = null;
|
||||
void opening
|
||||
.then((id) => {
|
||||
if (closed) {
|
||||
origin.closeHostRoute(id);
|
||||
return;
|
||||
}
|
||||
routeId = id;
|
||||
for (const handler of opens) handler();
|
||||
return undefined;
|
||||
})
|
||||
.catch(close);
|
||||
return {
|
||||
send(data) {
|
||||
if (closed || !routeId)
|
||||
throw Object.assign(new Error(`Host ${serverId} is offline`), {
|
||||
code: "host_offline",
|
||||
serverId,
|
||||
});
|
||||
origin.sendHostFrame(
|
||||
routeId,
|
||||
typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data),
|
||||
);
|
||||
},
|
||||
close: () => close(),
|
||||
onOpen(handler) {
|
||||
opens.add(handler);
|
||||
return () => opens.delete(handler);
|
||||
},
|
||||
onClose(handler) {
|
||||
closes.add(handler);
|
||||
return () => closes.delete(handler);
|
||||
},
|
||||
onError(handler) {
|
||||
errors.add(handler);
|
||||
return () => errors.delete(handler);
|
||||
},
|
||||
onMessage(handler) {
|
||||
messages.add(handler);
|
||||
return () => messages.delete(handler);
|
||||
},
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
/** Internal adapter for clients that own a low-level replica pipeline. */
|
||||
export interface HostConnectionLease {
|
||||
readonly client: DaemonClient;
|
||||
release(): Promise<void>;
|
||||
}
|
||||
|
||||
/** The origin alone resolves the destination; leases never probe another authority. */
|
||||
export async function acquireHostConnection(
|
||||
origin: DaemonClient,
|
||||
serverId: string,
|
||||
signal: AbortSignal,
|
||||
): Promise<HostConnectionLease> {
|
||||
const assertActive = () => {
|
||||
if (signal.aborted) throw new Error("Host connection acquisition is disposed");
|
||||
};
|
||||
assertActive();
|
||||
origin.requireConnectedHosts();
|
||||
if (
|
||||
origin.getServerId() === serverId &&
|
||||
origin.getLastServerInfoMessage()?.features?.hostSessionScopes !== true
|
||||
)
|
||||
throw Object.assign(new Error("Update the managing daemon to acquire owned session scopes"), {
|
||||
code: "unsupported_capability",
|
||||
serverId,
|
||||
});
|
||||
|
||||
const scopeId = safeRandomId();
|
||||
let client: DaemonClient | null = null;
|
||||
let release: Promise<void> | null = null;
|
||||
const cancel = () => {
|
||||
origin.cancelHostRoute(scopeId);
|
||||
if (client) void client.close().catch(() => {});
|
||||
};
|
||||
signal.addEventListener("abort", cancel, { once: true });
|
||||
const close = () => {
|
||||
if (!release) {
|
||||
signal.removeEventListener("abort", cancel);
|
||||
release = client ? client.close() : Promise.resolve();
|
||||
}
|
||||
return release;
|
||||
};
|
||||
try {
|
||||
const routeId = await origin.openHostRoute(serverId, scopeId);
|
||||
if (signal.aborted) {
|
||||
origin.closeHostRoute(routeId);
|
||||
assertActive();
|
||||
}
|
||||
client = new DaemonClient({
|
||||
url: "ws://routed-host/ws",
|
||||
clientId: scopeId,
|
||||
transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId),
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
await client.connect();
|
||||
assertActive();
|
||||
client.setReconnectEnabled(true);
|
||||
return { client, release: close };
|
||||
} catch (error) {
|
||||
await close().catch(() => {});
|
||||
throw Object.assign(error instanceof Error ? error : new Error("Host handshake failed"), {
|
||||
code: error instanceof Error && "code" in error ? error.code : "host_offline",
|
||||
serverId,
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -1,10 +1,10 @@
|
||||
import { safeRandomId } from "../daemon-client-transport-utils.js";
|
||||
import { DaemonClient } from "../daemon-client.js";
|
||||
import type { DaemonTransportFactory } from "../daemon-client-transport-types.js";
|
||||
import type { OwnedSubscription } from "../connection/index.js";
|
||||
import type { PaseoApi, createPaseoApi } from "../index.js";
|
||||
import type { HostAdd, HostUpdate, HostSummary, HostDirectory } from "@getpaseo/protocol/hosts";
|
||||
import { HostRuntime } from "./runtime.js";
|
||||
import { acquireHostConnection } from "./connection.js";
|
||||
export { createHostChannelTransport } from "./connection.js";
|
||||
|
||||
export interface PaseoHostActions {
|
||||
list(): Promise<HostDirectory>;
|
||||
@@ -16,108 +16,6 @@ export interface PaseoHostActions {
|
||||
runtime(): HostRuntime;
|
||||
}
|
||||
|
||||
export function createHostChannelTransport(
|
||||
origin: DaemonClient,
|
||||
serverId: string,
|
||||
scopeId: string,
|
||||
initialRouteId: string | null,
|
||||
link = false,
|
||||
): DaemonTransportFactory {
|
||||
let first: string | null = initialRouteId;
|
||||
return () => {
|
||||
const opens = new Set<() => void>();
|
||||
const closes = new Set<(event?: unknown) => void>();
|
||||
const messages = new Set<(data: unknown, binary: boolean) => void>();
|
||||
const errors = new Set<(error: unknown) => void>();
|
||||
let routeId: string | null = null;
|
||||
let closed = false;
|
||||
const cleanups: (() => void)[] = [];
|
||||
const close = (error?: unknown) => {
|
||||
if (closed) return;
|
||||
closed = true;
|
||||
if (error instanceof Error) Object.assign(error, { serverId });
|
||||
for (const cleanup of cleanups) cleanup();
|
||||
try {
|
||||
if (routeId) origin.closeHostRoute(routeId);
|
||||
else origin.cancelHostRoute(scopeId);
|
||||
} catch {
|
||||
/* Transport already closed. */
|
||||
}
|
||||
if (error) for (const handler of errors) handler(error);
|
||||
for (const handler of closes) handler(error);
|
||||
};
|
||||
cleanups.push(
|
||||
origin.subscribeRawMessages((message) => {
|
||||
if (message.type === "hosts.route.frame" && message.payload.subscriptionId === routeId) {
|
||||
for (const handler of messages) handler(message.payload.data, false);
|
||||
}
|
||||
if (message.type === "hosts.route.closed" && message.payload.subscriptionId === routeId)
|
||||
close(
|
||||
Object.assign(new Error(message.payload.message), {
|
||||
code: message.payload.code,
|
||||
serverId,
|
||||
}),
|
||||
);
|
||||
}),
|
||||
);
|
||||
cleanups.push(
|
||||
origin.subscribeHostFrames((frame) => {
|
||||
if (frame.routeId === routeId) for (const handler of messages) handler(frame.payload, true);
|
||||
}),
|
||||
);
|
||||
cleanups.push(
|
||||
origin.subscribeConnectionStatus((connectionState) => {
|
||||
if (connectionState.status !== "connected") close();
|
||||
}),
|
||||
);
|
||||
let opening: Promise<string>;
|
||||
if (first) opening = Promise.resolve(first);
|
||||
else opening = link ? origin.openHostLink(scopeId) : origin.openHostRoute(serverId, scopeId);
|
||||
first = null;
|
||||
void opening
|
||||
.then((id) => {
|
||||
if (closed) {
|
||||
origin.closeHostRoute(id);
|
||||
return;
|
||||
}
|
||||
routeId = id;
|
||||
for (const handler of opens) handler();
|
||||
return undefined;
|
||||
})
|
||||
.catch(close);
|
||||
return {
|
||||
send(data) {
|
||||
if (closed || !routeId)
|
||||
throw Object.assign(new Error(`Host ${serverId} is offline`), {
|
||||
code: "host_offline",
|
||||
serverId,
|
||||
});
|
||||
origin.sendHostFrame(
|
||||
routeId,
|
||||
typeof data === "string" || data instanceof Uint8Array ? data : new Uint8Array(data),
|
||||
);
|
||||
},
|
||||
close: () => close(),
|
||||
onOpen(handler) {
|
||||
opens.add(handler);
|
||||
return () => opens.delete(handler);
|
||||
},
|
||||
onClose(handler) {
|
||||
closes.add(handler);
|
||||
return () => closes.delete(handler);
|
||||
},
|
||||
onError(handler) {
|
||||
errors.add(handler);
|
||||
return () => errors.delete(handler);
|
||||
},
|
||||
onMessage(handler) {
|
||||
messages.add(handler);
|
||||
return () => messages.delete(handler);
|
||||
},
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
export function createHostActions(
|
||||
origin: DaemonClient,
|
||||
owner: {
|
||||
@@ -150,60 +48,23 @@ export function createHostActions(
|
||||
async client(serverId) {
|
||||
assertActive();
|
||||
origin.requireConnectedHosts();
|
||||
if (origin.getLastServerInfoMessage()?.serverId === serverId) {
|
||||
const api = owner.api(origin, { signal: owner.signal });
|
||||
const scope = owner.own(() => ({ release: () => api.dispose() }));
|
||||
return { ...api, dispose: () => scope.release() };
|
||||
}
|
||||
const scopeId = safeRandomId();
|
||||
const cancel = () => origin.cancelHostRoute(scopeId);
|
||||
owner.signal.addEventListener("abort", cancel, { once: true });
|
||||
let routeId: string;
|
||||
const lease = await acquireHostConnection(origin, serverId, owner.signal);
|
||||
try {
|
||||
routeId = await origin.openHostRoute(serverId, scopeId);
|
||||
} catch (error) {
|
||||
throw Object.assign(error instanceof Error ? error : new Error(String(error)), {
|
||||
serverId,
|
||||
});
|
||||
} finally {
|
||||
owner.signal.removeEventListener("abort", cancel);
|
||||
}
|
||||
if (owner.signal.aborted) {
|
||||
origin.closeHostRoute(routeId);
|
||||
assertActive();
|
||||
}
|
||||
const client = new DaemonClient({
|
||||
url: "ws://routed-host/ws",
|
||||
clientId: scopeId,
|
||||
transportFactory: createHostChannelTransport(origin, serverId, scopeId, routeId),
|
||||
// Initial acquisition has a deadline; established scopes restore demand on loss.
|
||||
reconnect: { enabled: false },
|
||||
});
|
||||
const api = owner.api(client, { signal: owner.signal });
|
||||
const scope = owner.own(() => ({
|
||||
release: async () => {
|
||||
try {
|
||||
await api.dispose();
|
||||
} finally {
|
||||
await client.close();
|
||||
}
|
||||
},
|
||||
}));
|
||||
try {
|
||||
await client.connect();
|
||||
client.setReconnectEnabled(true);
|
||||
assertActive();
|
||||
return { ...api, dispose: () => scope.release() };
|
||||
} catch (error) {
|
||||
const failure = Object.assign(
|
||||
error instanceof Error ? error : new Error("Host handshake failed"),
|
||||
{
|
||||
code: error instanceof Error && "code" in error ? error.code : "host_offline",
|
||||
serverId,
|
||||
const api = owner.api(lease.client, { signal: owner.signal });
|
||||
const scope = owner.own(() => ({
|
||||
release: async () => {
|
||||
try {
|
||||
await api.dispose();
|
||||
} finally {
|
||||
await lease.release();
|
||||
}
|
||||
},
|
||||
);
|
||||
await scope.release();
|
||||
throw failure;
|
||||
}));
|
||||
return { ...api, dispose: () => scope.release() };
|
||||
} catch (error) {
|
||||
await lease.release().catch(() => {});
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
runtime() {
|
||||
|
||||
@@ -23,14 +23,14 @@ export function createScheduleActions(
|
||||
return method(...args);
|
||||
};
|
||||
return {
|
||||
create: action(client.scheduleCreate.bind(client)),
|
||||
list: action(client.scheduleList.bind(client)),
|
||||
inspect: action(client.scheduleInspect.bind(client)),
|
||||
update: action(client.scheduleUpdate.bind(client)),
|
||||
pause: action(client.schedulePause.bind(client)),
|
||||
resume: action(client.scheduleResume.bind(client)),
|
||||
delete: action(client.scheduleDelete.bind(client)),
|
||||
logs: action(client.scheduleLogs.bind(client)),
|
||||
runOnce: action(client.scheduleRunOnce.bind(client)),
|
||||
create: action((...args) => client.scheduleCreate(...args)),
|
||||
list: action((...args) => client.scheduleList(...args)),
|
||||
inspect: action((...args) => client.scheduleInspect(...args)),
|
||||
update: action((...args) => client.scheduleUpdate(...args)),
|
||||
pause: action((...args) => client.schedulePause(...args)),
|
||||
resume: action((...args) => client.scheduleResume(...args)),
|
||||
delete: action((...args) => client.scheduleDelete(...args)),
|
||||
logs: action((...args) => client.scheduleLogs(...args)),
|
||||
runOnce: action((...args) => client.scheduleRunOnce(...args)),
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user