fix(server-s3): list a member's tmux panes asynchronously
A frozen member (SIGSTOPped tmux server) cost every user one synchronous 3 s block of the event loop per minute (F37's back-off ceiling). Member listing now runs through async execFile (same 3 s / SIGKILL / 1 MiB bounds, same back-off) into the listed cache the synchronous readers (liveSessions, listSessions, listOutput) already use: stale-while- revalidate, 15 s max age, cold or failing = that member's panes UNKNOWN (F38 unchanged). A session this process creates or kills is written into the member's cached answer at once, and a list in flight across that change is dropped. Host tmux calls, local mode and every click's call (new-session, send-keys, kill, has-session) are unchanged. review5's stalled-member test now polls for 4 s past the timeout and asserts each poll < 100 ms, event-loop stall < 500 ms and one sudo (old code: first poll 3157 ms).
This commit is contained in:
+138
-37
@@ -4,8 +4,9 @@
|
||||
* feature. Every dashboard-managed session is named `agenthub-lane-<id>` (see
|
||||
* `pty-run.js`) so a real terminal can attach to the exact same session with
|
||||
* `tmux attach -t agenthub-lane-<id>` (or `agenthub lanes shell`). Never builds a
|
||||
* shell string — every call is `execFileSync("tmux", [...argv])` with an
|
||||
* explicit argument array (matches this repo's rule for git in worktree.js).
|
||||
* shell string — every call is `execFileSync("tmux", [...argv])` (a member's
|
||||
* listing: `execFile`) with an explicit argument array (matches this repo's
|
||||
* rule for git in worktree.js).
|
||||
* No call composes a shell command line at all, `sendCommand` included: it
|
||||
* takes over an idle pane with `respawn-pane -- <argv>` rather than typing a
|
||||
* quoted line into that pane's shell, which keeps an API key passed in `env`
|
||||
@@ -21,13 +22,14 @@
|
||||
* that socket's folder, and a host client on it could be handed a fake server
|
||||
* or replaced by a member's command (`detach-client -E`). sudo logs its
|
||||
* command line, so a session's environment and argv travel in a launch file
|
||||
* (`run-as.launchArgv`) and a paste on stdin. Listing covers every server.
|
||||
* (`run-as.launchArgv`) and a paste on stdin. Listing covers every server;
|
||||
* a member's runs in the background and a poll reads its last answer.
|
||||
* With no member (local mode, S3 not installed, the host's own workspaces)
|
||||
* nothing changes.
|
||||
* @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
|
||||
*/
|
||||
|
||||
const { execFileSync } = require("node:child_process");
|
||||
const { execFile, execFileSync } = require("node:child_process");
|
||||
|
||||
// Test seam: swap the exec implementation so unit tests never invoke a real
|
||||
// tmux binary. Mirrors run-spawner.js's __injectChildForTest/__reset style.
|
||||
@@ -67,39 +69,77 @@ let scopeImpl = realScopeImpl;
|
||||
// stdin). Seams, so tests never sudo.
|
||||
let accountImpl = (name) => require("./run-as").accountForSession(name);
|
||||
// A member's server is theirs: they can stop it (SIGSTOP) or fill it with
|
||||
// sessions, and every call here blocks the event loop. So each call on it is
|
||||
// bounded (time, output), and a server that timed out or flooded us is not
|
||||
// asked again for MEMBER_BACKOFF_MS — its panes read as UNKNOWN meanwhile.
|
||||
// ponytail: a frozen member still costs one MEMBER_TIMEOUT_MS block per
|
||||
// backoff; move member calls to async execFile if that ever shows.
|
||||
// sessions. So each call on it is bounded (time, output), and a server that
|
||||
// timed out or flooded us is not asked again for MEMBER_BACKOFF_MS — its
|
||||
// panes read as UNKNOWN meanwhile. LISTING (every poll: GET /api/run and
|
||||
// /api/lanes, the heal beat, the liveness reaper) never waits for a member:
|
||||
// it runs async (`memberListImpl`) into `listed`, which the poll reads. A
|
||||
// call a user's click makes (new-session, send-keys, kill, has-session)
|
||||
// stays synchronous.
|
||||
// ponytail: a click on a frozen member's pane (and a pane-token / gateway
|
||||
// has-session check) still blocks up to MEMBER_TIMEOUT_MS once per backoff;
|
||||
// make those async too if that ever shows.
|
||||
const MEMBER_TIMEOUT_MS = 3000;
|
||||
const MEMBER_BACKOFF_MS = 60_000;
|
||||
const MEMBER_OPTS = { ...EXEC_OPTS, timeout: MEMBER_TIMEOUT_MS, killSignal: "SIGKILL", maxBuffer: 1 << 20 };
|
||||
const realMemberImpl = (acct, args, input) => {
|
||||
const MEMBER_OPTS = {
|
||||
...EXEC_OPTS,
|
||||
timeout: MEMBER_TIMEOUT_MS,
|
||||
killSignal: "SIGKILL",
|
||||
maxBuffer: 1 << 20,
|
||||
};
|
||||
const { stdio: _piped, ...MEMBER_LIST_OPTS } = MEMBER_OPTS; // execFile always pipes
|
||||
const sudoTmux = (acct, args) => {
|
||||
const runAs = require("./run-as");
|
||||
const [cmd, argv] = runAs.sudoArgv(acct, "tmux", ["-S", runAs.tmuxSocket(acct), ...args]);
|
||||
return runAs.sudoArgv(acct, "tmux", ["-S", runAs.tmuxSocket(acct), ...args]);
|
||||
};
|
||||
const realMemberImpl = (acct, args, input) => {
|
||||
const [cmd, argv] = sudoTmux(acct, args);
|
||||
return execFileSync(cmd, argv, input == null ? MEMBER_OPTS : { ...MEMBER_OPTS, input });
|
||||
};
|
||||
// The same, async, with execFileSync's error shape: `status` = exit code,
|
||||
// ETIMEDOUT / ENOBUFS for a kill at the timeout / a flood.
|
||||
const realMemberListImpl = (acct, args) =>
|
||||
new Promise((resolve, reject) => {
|
||||
const [cmd, argv] = sudoTmux(acct, args);
|
||||
execFile(cmd, argv, MEMBER_LIST_OPTS, (err, out) => {
|
||||
if (!err) return resolve(out);
|
||||
if (typeof err.code === "number") err.status = err.code;
|
||||
if (err.killed) err.code = "ETIMEDOUT";
|
||||
if (err.code === "ERR_CHILD_PROCESS_STDIO_MAXBUFFER") err.code = "ENOBUFS";
|
||||
reject(err);
|
||||
});
|
||||
});
|
||||
let memberImpl = realMemberImpl;
|
||||
let memberListImpl = realMemberListImpl;
|
||||
let unreachable = new Map(); // member login → backoff end
|
||||
function memberCall(acct, args, input) {
|
||||
function reachable(acct) {
|
||||
if ((unreachable.get(acct.login) || 0) > Date.now()) {
|
||||
throw Object.assign(new Error(`${acct.login}: tmux server not answering`), {
|
||||
code: "ETMUXHUNG",
|
||||
});
|
||||
}
|
||||
}
|
||||
function noteHang(acct, err) {
|
||||
if (err.code === "ETIMEDOUT" || err.code === "ENOBUFS") {
|
||||
unreachable.set(acct.login, Date.now() + MEMBER_BACKOFF_MS);
|
||||
}
|
||||
}
|
||||
function memberCall(acct, args, input) {
|
||||
reachable(acct);
|
||||
try {
|
||||
return memberImpl(acct, args, input);
|
||||
} catch (err) {
|
||||
if (err.code === "ETIMEDOUT" || err.code === "ENOBUFS") {
|
||||
unreachable.set(acct.login, Date.now() + MEMBER_BACKOFF_MS);
|
||||
}
|
||||
noteHang(acct, err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
/** Test seam. A sync `member` fake also answers the async listing. */
|
||||
function __setAccountImpl(fn, member) {
|
||||
accountImpl = fn;
|
||||
if (member) memberImpl = member;
|
||||
if (member) {
|
||||
memberImpl = member;
|
||||
memberListImpl = (acct, args) => new Promise((resolve) => resolve(member(acct, args)));
|
||||
}
|
||||
}
|
||||
function accountOf(name) {
|
||||
try {
|
||||
@@ -128,6 +168,7 @@ function __setExecImpl(fn, { exact = true } = {}) {
|
||||
|
||||
exactTargets = exact;
|
||||
listed = new Map();
|
||||
listGen++;
|
||||
}
|
||||
function __setScopeImpl(fn) {
|
||||
scopeImpl = fn;
|
||||
@@ -137,8 +178,10 @@ function __reset() {
|
||||
scopeImpl = realScopeImpl;
|
||||
accountImpl = (name) => require("./run-as").accountForSession(name);
|
||||
memberImpl = realMemberImpl;
|
||||
memberListImpl = realMemberListImpl;
|
||||
exactTargets = null;
|
||||
listed = new Map();
|
||||
listGen++;
|
||||
unreachable = new Map();
|
||||
}
|
||||
|
||||
@@ -149,34 +192,90 @@ function __reset() {
|
||||
// own can read as live for up to a second. "No server" (exit 1) is an answer
|
||||
// too and is shared; any other failure is not.
|
||||
const LIST_TTL_MS = 1000;
|
||||
const LIST_ARGS = ["list-sessions", "-F", "#{session_name}"];
|
||||
let listed = new Map(); // member login ("" = ours) → { at, out } | { at, err }
|
||||
let listGen = 0; // bumped by a change this process makes: drops lists in flight
|
||||
function listOutput(acct = null) {
|
||||
const key = acct ? acct.login : "";
|
||||
const hit = listed.get(key);
|
||||
if (acct) return memberListOutput(acct);
|
||||
const hit = listed.get("");
|
||||
if (hit && Date.now() - hit.at < LIST_TTL_MS) {
|
||||
if (hit.err) throw hit.err;
|
||||
return hit.out;
|
||||
}
|
||||
try {
|
||||
const list = ["list-sessions", "-F", "#{session_name}"];
|
||||
let out = String(acct ? memberCall(acct, list) : execImpl(list));
|
||||
const out = String(execImpl(LIST_ARGS));
|
||||
listed.set("", { at: Date.now(), out });
|
||||
return out;
|
||||
} catch (err) {
|
||||
if (err.status === 1 || err.code === "ENOENT") listed.set("", { at: Date.now(), err });
|
||||
else listed.delete("");
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// A member's list is never waited for: a read answers from `listed` and,
|
||||
// once that is LIST_TTL_MS old, starts one async re-list. Until a first
|
||||
// answer — or once the last is MEMBER_LIST_MAX_AGE_MS old (nobody polled) —
|
||||
// or while their server fails, it throws: their panes are UNKNOWN. A session
|
||||
// this process creates or kills is written into the answer at once (`noteSession`).
|
||||
const MEMBER_LIST_MAX_AGE_MS = 15_000; // > the 10 s heal beat
|
||||
const refreshing = new Set(); // member logins with a list in flight
|
||||
function memberListOutput(acct) {
|
||||
const hit = listed.get(acct.login);
|
||||
const age = hit ? Date.now() - hit.at : Infinity;
|
||||
if (age >= LIST_TTL_MS) refreshList(acct);
|
||||
if (age >= MEMBER_LIST_MAX_AGE_MS) {
|
||||
throw Object.assign(new Error(`${acct.login}: tmux list pending`), { code: "ETMUXPENDING" });
|
||||
}
|
||||
if (hit.err) throw hit.err;
|
||||
return hit.out;
|
||||
}
|
||||
function refreshList(acct) {
|
||||
const key = acct.login;
|
||||
if (refreshing.has(key)) return;
|
||||
refreshing.add(key);
|
||||
const gen = listGen;
|
||||
const hangs = unreachable; // a test's __reset starts a new one
|
||||
const done = (entry) => {
|
||||
refreshing.delete(key);
|
||||
if (gen === listGen) listed.set(key, { at: Date.now(), ...entry });
|
||||
};
|
||||
new Promise((resolve) => {
|
||||
reachable(acct);
|
||||
resolve(memberListImpl(acct, LIST_ARGS));
|
||||
}).then(
|
||||
// A member's server answers for their own workspaces only: a name it
|
||||
// lists that belongs to anyone else (a server the member runs to lie
|
||||
// about another pane) is not believed.
|
||||
if (acct)
|
||||
out = out
|
||||
.split("\n")
|
||||
.filter((n) => ownedBy(n.trim(), acct))
|
||||
.join("\n");
|
||||
listed.set(key, { at: Date.now(), out });
|
||||
return out;
|
||||
} catch (err) {
|
||||
// A member's failure is shared too: a broken server of theirs is not
|
||||
// asked again on every poll.
|
||||
if (acct || err.status === 1 || err.code === "ENOENT") listed.set(key, { at: Date.now(), err });
|
||||
else listed.delete(key);
|
||||
throw err;
|
||||
}
|
||||
(out) =>
|
||||
done({
|
||||
out: String(out)
|
||||
.split("\n")
|
||||
.filter((n) => ownedBy(n.trim(), acct))
|
||||
.join("\n"),
|
||||
}),
|
||||
(err) => {
|
||||
if (hangs === unreachable) noteHang(acct, err);
|
||||
// "No server" is an empty list; any other failure is shared too, so a
|
||||
// broken server of theirs is not asked again on every poll.
|
||||
done(err.status === 1 || err.code === "ENOENT" ? { out: "" } : { err });
|
||||
}
|
||||
);
|
||||
}
|
||||
/** A session this process made or ended: ours re-lists, a member's answer is
|
||||
* corrected in place (an unknown one stays unknown). */
|
||||
function noteSession(name, acct, alive) {
|
||||
listGen++;
|
||||
listed.delete("");
|
||||
const hit = acct && listed.get(acct.login);
|
||||
if (!hit || hit.err) return;
|
||||
const names = hit.out.split("\n").filter((n) => n.trim() && n.trim() !== name);
|
||||
if (alive) names.push(name);
|
||||
listed.set(acct.login, { at: hit.at, out: names.join("\n") });
|
||||
}
|
||||
/** Test seam: resolves once no member list is in flight. */
|
||||
async function __settle() {
|
||||
while (refreshing.size) await new Promise((r) => setImmediate(r));
|
||||
}
|
||||
|
||||
function ownedBy(name, acct) {
|
||||
@@ -273,7 +372,7 @@ function envFlags(env) {
|
||||
function newSession({ name, cwd, argv, env }) {
|
||||
// No argv = tmux's own default shell (a Windows Terminal pane).
|
||||
const cmd = argv.length ? ["--", ...argv] : [];
|
||||
listed = new Map();
|
||||
noteSession(name, null, false);
|
||||
const args = ["new-session", "-d", "-s", name, "-c", cwd, ...envFlags(env), ...cmd];
|
||||
const acct = accountOf(name);
|
||||
if (acct) {
|
||||
@@ -281,6 +380,7 @@ function newSession({ name, cwd, argv, env }) {
|
||||
// the launch file, never on sudo's logged command line.
|
||||
const launch = require("./run-as").launchArgv(acct, env, argv);
|
||||
memberCall(acct, ["new-session", "-d", "-s", name, "-c", cwd, "--", ...launch]);
|
||||
noteSession(name, acct, true);
|
||||
} else if (scopeImpl && process.platform === "linux" && !serverPid()) {
|
||||
try {
|
||||
scopeImpl(args);
|
||||
@@ -449,12 +549,12 @@ function pasteText(name, text) {
|
||||
|
||||
/** Idempotent — a session that's already gone is not an error. */
|
||||
function killSession(name) {
|
||||
listed = new Map();
|
||||
try {
|
||||
onServerOf(name, ["kill-session", "-t", session(name)]);
|
||||
} catch {
|
||||
/* already gone */
|
||||
}
|
||||
noteSession(name, accountOfQuiet(name), false);
|
||||
}
|
||||
|
||||
/** Session names starting with `prefix`. Empty array if tmux has no server running at all. */
|
||||
@@ -567,5 +667,6 @@ module.exports = {
|
||||
memberAttach,
|
||||
__setExecImpl,
|
||||
__setAccountImpl,
|
||||
__settle,
|
||||
__reset,
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user