mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden provider gateway upgrades

Browse Source
This commit is contained in:
Codex
2026-05-06 00:24:17 +00:00
parent af42002f02
commit 2dc46ce056
16 changed files with 1059 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/provider-gateway/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@unidesk/provider-gateway",
"version": "0.2.1",
"version": "0.2.5",
"private": true,
"type": "module",
"scripts": {
src/components/provider-gateway/src/index.ts
+157 -17
View File
@@ -58,6 +58,8 @@ let systemStatusRunning = false;
let previousCpuSample: { idle: number; total: number } | null = null;
let reconnectAttempt = 0;
let stopping = false;
let upgradeSleepUntil = 0;
let upgradeSleepTimer: ReturnType<typeof setTimeout> | null = null;
interface HostSshSession {
proc: ReturnType<typeof Bun.spawn>;
@@ -72,19 +74,29 @@ interface HostSshStdin {
const hostSshSessions = new Map<string, HostSshSession>();
const gatewayMetadata = readGatewayMetadata();
function readGatewayMetadata(): { name: string; version: string } {
function readGatewayMetadataFile(path: string): { name: string; version: string } | null {
try {
const raw = readFileSync(new URL("../package.json", import.meta.url), "utf8");
const raw = readFileSync(path, "utf8");
const parsed = JSON.parse(raw) as { name?: unknown; version?: unknown };
return {
name: typeof parsed.name === "string" && parsed.name.length > 0 ? parsed.name : "@unidesk/provider-gateway",
version: typeof parsed.version === "string" && parsed.version.length > 0 ? parsed.version : "0.0.0-unknown",
};
} catch {
return { name: "@unidesk/provider-gateway", version: "0.0.0-unknown" };
return null;
}
}
function readGatewayMetadata(): { name: string; version: string } {
return readGatewayMetadataFile(new URL("../package.json", import.meta.url).pathname)
?? { name: "@unidesk/provider-gateway", version: "0.0.0-unknown" };
}
function readTargetGatewayMetadata(workspace: string): { name: string; version: string } {
const root = workspace.replace(/\/+$/, "");
return readGatewayMetadataFile(`${root}/src/components/provider-gateway/package.json`) ?? gatewayMetadata;
}
function requiredEnv(name: string): string {
const value = process.env[name];
if (value === undefined || value.length === 0) {
@@ -885,8 +897,14 @@ function safeDockerName(value: string): string {
return value.replace(/[^a-zA-Z0-9_.-]/g, "-").slice(0, 80);
}
function candidateGatewayName(taskId: string): string {
return `unidesk-provider-gateway-${safeDockerName(config.providerId)}-candidate-${safeDockerName(taskId)}`.slice(0, 120);
}
function upgradePlan(taskId: string): Record<string, JsonValue> {
const workspace = config.upgradeWorkspacePath;
const sleepMs = 300_000;
const validationTimeoutMs = 180_000;
const composeBaseCommand = [
"docker",
"compose",
@@ -910,6 +928,8 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
`label=com.docker.compose.project=${config.upgradeComposeProject}`,
"--filter",
`label=com.docker.compose.service=${config.upgradeService}`,
"--filter",
"label=com.docker.compose.oneoff=False",
];
const composeUpCommand = [
...composeBaseCommand,
@@ -920,22 +940,73 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
config.upgradeService,
];
const updaterName = `unidesk-provider-upgrader-${safeDockerName(taskId)}`;
const candidateName = candidateGatewayName(taskId);
const validationNeedleOpen = `"message":"connect_open"`;
const validationNeedleAck = `"requestId":"register"`;
const validationNeedleOk = `"ok":true`;
const validationAttempts = Math.max(1, Math.ceil(validationTimeoutMs / 2000));
const targetGatewayMetadata = readTargetGatewayMetadata(workspace);
const script = [
"set -eu",
"sleep 2",
`cd ${shellQuote(workspace)}`,
`old_ids=$(${listServiceContainersCommand.map(shellQuote).join(" ")})`,
`first_old=""`,
`for old_id in $old_ids; do first_old="$old_id"; break; done`,
`old_name=""`,
`mount_args=""`,
`extra_host_args=""`,
`candidate_env_file=${shellQuote(`/tmp/${candidateName}.env`)}`,
`network_names=""`,
`first_network=""`,
`network_arg=""`,
`if [ -n "$old_ids" ]; then docker update --restart always $old_ids >/dev/null 2>&1 || true; fi`,
`if [ -z "$first_old" ]; then echo "no existing provider-gateway compose container found; cannot perform safe in-place upgrade" >&2; exit 1; fi`,
`old_name=$(docker inspect --format '{{.Name}}' "$first_old")`,
`old_name="\${old_name#/}"`,
`docker inspect --format '{{range .Config.Env}}{{println .}}{{end}}' "$first_old" > "$candidate_env_file"`,
`mount_args=$(docker inspect --format '{{range .Mounts}}{{printf "-v %s:%s" .Source .Destination}}{{if not .RW}}{{printf ":ro"}}{{end}}{{printf "\\n"}}{{end}}' "$first_old")`,
`extra_host_args=$(docker inspect --format '{{range .HostConfig.ExtraHosts}}{{printf "--add-host %s\\n" .}}{{end}}' "$first_old")`,
`network_names=$(docker inspect --format '{{range $name, $_ := .NetworkSettings.Networks}}{{printf "%s\\n" $name}}{{end}}' "$first_old")`,
`for net in $network_names; do first_network="$net"; break; done`,
`if [ -n "$first_network" ]; then network_arg="--network $first_network"; fi`,
composeBuildCommand.map(shellQuote).join(" "),
`ids=$(${listServiceContainersCommand.map(shellQuote).join(" ")})`,
`if [ -n "$ids" ]; then docker rm -f $ids; fi`,
composeUpCommand.map(shellQuote).join(" "),
].join("; ");
`docker rm -f ${shellQuote(candidateName)} >/dev/null 2>&1 || true`,
[
"docker run -d",
`--name ${shellQuote(candidateName)}`,
"--restart no",
"$network_arg",
`--env-file "$candidate_env_file"`,
`--label ${shellQuote(`com.docker.compose.project=${config.upgradeComposeProject}`)}`,
`--label ${shellQuote(`com.docker.compose.service=${config.upgradeService}`)}`,
`--label ${shellQuote("com.docker.compose.oneoff=False")}`,
"$mount_args",
"$extra_host_args",
shellQuote(config.upgradeRunnerImage),
].join(" "),
`if [ -n "$network_names" ]; then for net in $network_names; do if [ "$net" != "$first_network" ]; then docker network connect "$net" ${shellQuote(candidateName)} >/dev/null 2>&1 || true; fi; done; fi`,
`attempt=0`,
`validated=0`,
`while [ "$attempt" -lt "${validationAttempts}" ]; do logs=$(docker logs ${shellQuote(candidateName)} 2>&1 || true); has_open=0; has_ack=0; has_ok=0; case "$logs" in *${shellQuote(validationNeedleOpen)}*) has_open=1;; esac; case "$logs" in *${shellQuote(validationNeedleAck)}*) has_ack=1;; esac; case "$logs" in *${shellQuote(validationNeedleOk)}*) has_ok=1;; esac; if [ "$has_open" = "1" ] && [ "$has_ack" = "1" ] && [ "$has_ok" = "1" ]; then validated=1; break; fi; candidate_running=$(docker inspect --format '{{.State.Running}}' ${shellQuote(candidateName)} 2>/dev/null || true); if [ "$candidate_running" != "true" ]; then break; fi; attempt=$((attempt + 1)); sleep 2; done`,
`if [ "$validated" != "1" ]; then echo "candidate validation failed; old gateway will leave upgrade sleep automatically" >&2; docker logs ${shellQuote(candidateName)} >&2 || true; docker rm -f ${shellQuote(candidateName)} >/dev/null 2>&1 || true; rm -f "$candidate_env_file"; exit 1; fi`,
`docker update --restart always ${shellQuote(candidateName)} >/dev/null`,
`if [ -n "$old_ids" ]; then docker rm -f $old_ids; fi`,
`if [ -n "$old_name" ] && [ "$old_name" != ${shellQuote(candidateName)} ]; then docker rename ${shellQuote(candidateName)} "$old_name" || true; fi`,
`rm -f "$candidate_env_file"`,
`echo "candidate provider-gateway validated and promoted"`,
].join("\n");
const dockerRunCommand = [
"docker",
"run",
"-d",
"--rm",
"--name",
updaterName,
"--restart",
"no",
"--label",
"com.docker.compose.oneoff=True",
"--label",
`unidesk.upgrade.updater=${taskId}`,
"-v",
"/var/run/docker.sock:/var/run/docker.sock",
"-v",
@@ -950,7 +1021,14 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
return {
policy: "always-enabled",
taskId,
providerId: config.providerId,
providerName: config.providerName,
providerGatewayName: gatewayMetadata.name,
providerGatewayVersion: gatewayMetadata.version,
targetProviderGatewayName: targetGatewayMetadata.name,
targetProviderGatewayVersion: targetGatewayMetadata.version,
updaterName,
candidateName,
runnerImage: config.upgradeRunnerImage,
hostProjectRoot: config.upgradeHostProjectRoot,
workspace,
@@ -965,7 +1043,16 @@ function upgradePlan(taskId: string): Record<string, JsonValue> {
listServiceContainersCommand,
composeUpCommand,
replacementStrategy: {
buildBeforeRemove: true,
buildBeforeCandidate: true,
oldGatewaySleepMs: sleepMs,
validationTimeoutMs,
oldGatewayRestartPolicyBeforeSleep: "always",
promoteOnlyAfterCandidateValidation: true,
candidateRestartPolicyAfterPromotion: "always",
candidateUsesOldContainerMounts: true,
candidateUsesOldContainerNetworks: true,
candidateUsesOldContainerExtraHosts: true,
candidateUsesOldContainerEnvironment: true,
removeScope: {
projectLabel: config.upgradeComposeProject,
serviceLabel: config.upgradeService,
@@ -986,9 +1073,15 @@ async function runProviderUpgrade(taskId: string, payload: Record<string, JsonVa
if (!result.ok) {
throw new Error(`provider upgrade scheduler failed with exit ${result.exitCode}: ${result.stderr}`);
}
const sleepMs = Number((plan.replacementStrategy as Record<string, JsonValue>).oldGatewaySleepMs ?? 300_000);
setTimeout(() => enterUpgradeSleep(sleepMs), 250).unref?.();
return {
mode,
message: "provider gateway upgrade scheduled by detached updater container",
message: "provider gateway upgrade scheduled with sleep-and-validate rollback guard",
providerId: config.providerId,
providerName: config.providerName,
providerGatewayVersion: gatewayMetadata.version,
targetProviderGatewayVersion: (plan.targetProviderGatewayVersion as string | undefined) ?? gatewayMetadata.version,
updaterContainerId: result.stdout.trim(),
plan,
stderr: result.stderr.slice(0, 500),
@@ -1197,13 +1290,65 @@ function handleMessage(raw: MessageEvent<string>): void {
function scheduleReconnect(): void {
if (stopping) return;
const sleepRemainingMs = upgradeSleepUntil - Date.now();
if (sleepRemainingMs > 0) {
logger("warn", "reconnect_deferred_for_upgrade_sleep", { sleepRemainingMs });
if (upgradeSleepTimer === null) {
upgradeSleepTimer = setTimeout(() => {
upgradeSleepTimer = null;
logger("warn", "upgrade_sleep_expired_reconnecting", { providerId: config.providerId });
connect();
}, sleepRemainingMs);
}
return;
}
reconnectAttempt += 1;
const delayMs = Math.min(config.reconnectMaxMs, config.reconnectBaseMs * 2 ** Math.min(reconnectAttempt, 8));
logger("warn", "reconnect_scheduled", { reconnectAttempt, delayMs });
setTimeout(connect, delayMs);
}
function clearConnectionTimers(): void {
if (heartbeatTimer !== null) clearInterval(heartbeatTimer);
if (systemStatusTimer !== null) clearInterval(systemStatusTimer);
if (dockerStatusTimer !== null) clearInterval(dockerStatusTimer);
heartbeatTimer = null;
systemStatusTimer = null;
dockerStatusTimer = null;
}
function enterUpgradeSleep(durationMs: number): void {
const sleepMs = Math.max(30_000, Math.min(300_000, Math.floor(durationMs)));
upgradeSleepUntil = Date.now() + sleepMs;
logger("warn", "upgrade_sleep_enter", { providerId: config.providerId, sleepMs, wakeAt: new Date(upgradeSleepUntil).toISOString() });
clearConnectionTimers();
try {
socket?.close(4000, "provider upgrade sleep");
} catch (error) {
logger("warn", "upgrade_sleep_socket_close_failed", { error: String(error) });
}
socket = null;
if (upgradeSleepTimer !== null) clearTimeout(upgradeSleepTimer);
upgradeSleepTimer = setTimeout(() => {
upgradeSleepTimer = null;
logger("warn", "upgrade_sleep_expired_reconnecting", { providerId: config.providerId });
connect();
}, sleepMs);
}
function connect(): void {
if (stopping) return;
const sleepRemainingMs = upgradeSleepUntil - Date.now();
if (sleepRemainingMs > 0) {
logger("warn", "connect_deferred_for_upgrade_sleep", { sleepRemainingMs });
if (upgradeSleepTimer === null) {
upgradeSleepTimer = setTimeout(() => {
upgradeSleepTimer = null;
connect();
}, sleepRemainingMs);
}
return;
}
const url = withToken(config.serverUrl, config.token);
logger("info", "connect_start", { serverUrl: config.serverUrl, providerId: config.providerId });
socket = new WebSocket(url);
@@ -1228,12 +1373,7 @@ function connect(): void {
socket.addEventListener("message", (event) => handleMessage(event as MessageEvent<string>));
socket.addEventListener("close", (event) => {
logger("warn", "connect_close", { code: event.code, reason: event.reason });
if (heartbeatTimer !== null) clearInterval(heartbeatTimer);
if (systemStatusTimer !== null) clearInterval(systemStatusTimer);
if (dockerStatusTimer !== null) clearInterval(dockerStatusTimer);
heartbeatTimer = null;
systemStatusTimer = null;
dockerStatusTimer = null;
clearConnectionTimers();
scheduleReconnect();
});
socket.addEventListener("error", () => {