conversation: release a cohort scope after a proven stop (row 50, #1536)
releaseCohort sends `release` only to a scope shim that answers hello with the recorded invocation ID, then waits for systemd to drop the unit. The controller releases once per claim, after a proven force stop and on close of a proven-stopped binding; every uncertain path keeps the scope as evidence. The harness's killShims refuses units outside ^mosaic-chat-, R5 checks liveShims positively, and K19's wait on proc.exited is bounded. Dewey's candidate, manifest 375594fc (8 files), approved by Filbert (comment 27053) and Darkwing (comment 27055). Normal engine exit still leaves the scope; the follow-up is #1537. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
Claude Opus 5.5
@@ -232,7 +232,7 @@ live session, and the controller never writes a session file (W11).
|
||||
| `src/terminal.mjs` | the mediated terminal, `node packages/conversation/src/terminal.mjs --socket <path> [--grant <id>]` |
|
||||
| `src/claim.mjs` | `ClaimStore`: the writer claim per seat key and session key, revisions `r<10 digits>.json` |
|
||||
| `src/cohort.mjs` | `ScopeLauncher` (systemd user scope plus `shim.mjs`), `PgroupLauncher`, force stop, cohort and boot proofs |
|
||||
| `src/shim.mjs` | the scope's first process, outside the `engine` cgroup; ops `hello`, `events`, `members`, `term`, `freeze`, `kill`, `release`; ignores SIGTERM |
|
||||
| `src/shim.mjs` | the scope's first process, outside the `engine` cgroup; ops `hello`, `events`, `members`, `term`, `freeze`, `kill`, `release`; ignores SIGTERM and exits on `release` once `engine` is empty |
|
||||
| `src/guard.mjs` | `LiveSessionGuard`: refuses live sessions and paths under live roots |
|
||||
| `src/pi-pin.mjs` | the Pi pin (0.85.1 and its integrity) and the seal |
|
||||
| `src/engine.mjs`, `src/framing.mjs` | the engine link and LF framing |
|
||||
@@ -366,6 +366,21 @@ can disagree with it.
|
||||
fallback can't enumerate a member that left the group, so its force stop
|
||||
ends `uncertain`, never `stopped` (K2). The fake launcher's `forceStop`
|
||||
is fixture-only.
|
||||
- **Scope release.** The shim exits only on `release`, and only while
|
||||
`engine` reads `populated 0`; systemd then collects the empty scope. The
|
||||
controller sends it once the stop is proven, verified and recorded
|
||||
`stopped` on the claim, and again on close of a binding proven stopped,
|
||||
so a release the stop didn't reach still happens; both share one request
|
||||
per claim (R1, R2). The shim must answer for the claim's recorded
|
||||
invocation ID, or nothing is released (R6). Each attempt is in
|
||||
`evidence.releases`. After a stop that ends `uncertain` (evidence
|
||||
unavailable, or a proof the verifier rejects) nothing is released: the
|
||||
scope and its shim are what a later force stop reads, and a collected
|
||||
scope would be an absent observation, not proof that the cohort ended
|
||||
(R3). An engine that exits on its own leaves the shim and scope up and
|
||||
the binding `uncertain`; a proven force stop on the empty cohort then
|
||||
releases them (R4). A controller killed between recording the stop and
|
||||
releasing leaves the scope; a restart doesn't release it.
|
||||
- **Confirmations** bind the target, operation and stop, and are consumed on
|
||||
use (K6). One issued before the stop changed is refused (H17).
|
||||
- **Recovery.** A confirmed `recover` after a proven stop returns a
|
||||
@@ -546,7 +561,7 @@ no prompt:
|
||||
| `claim.test.mjs` | W1–W17, W20, G1–G3: the writer claim, crash barriers, the guard |
|
||||
| `races.test.mjs` | H1–H4, H9–H23: takeover, Interrupt and force stop, retries, incarnations |
|
||||
| `turns.test.mjs` | N1–N25, N24b: the turn tracker against the fake engine's Pi behaviors; the engine seal and environment |
|
||||
| `cohort.test.mjs` | K1–K19: scopes, force stop, proofs, recovery, eligibility, the scope's environment (needs a systemd user manager) |
|
||||
| `cohort.test.mjs` | K1–K19, R1–R6: scopes, force stop, proofs, recovery, eligibility, the scope's environment, scope release (needs a systemd user manager) |
|
||||
| `flows.test.mjs` | S1–S7, P3, E1–E7, the terminal, and a CHAT-01 schema check of every record produced |
|
||||
| `smoke.test.mjs` | the pinned Pi binary, as above |
|
||||
|
||||
|
||||
@@ -14,7 +14,9 @@
|
||||
// every member and wait a bounded grace; freeze `engine` and wait for
|
||||
// `frozen 1`; enumerate every member with pid and start time; write
|
||||
// `cgroup.kill`; wait for `populated 0`. Only when all of that succeeded is
|
||||
// membership complete. Anything unavailable ends the stop `uncertain`.
|
||||
// membership complete. Anything unavailable ends the stop `uncertain`. The
|
||||
// scope stays up after the stop either way; `releaseCohort` ends it once the
|
||||
// controller has recorded a proven stop.
|
||||
|
||||
import { spawn, spawnSync } from "node:child_process";
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
@@ -202,6 +204,28 @@ export async function forceStopCohort({ kind, unitName, invocationId, shimSocket
|
||||
};
|
||||
}
|
||||
|
||||
// Ends a scope whose cohort is proven stopped (#1536). The shim exits on
|
||||
// `release` only while `engine` reads `populated 0`, and systemd then
|
||||
// collects the empty scope. Call it only after a `proven` stop whose claim
|
||||
// is recorded `stopped`: after an `unavailable` one, the scope and its shim
|
||||
// are the evidence a later stop reads, and a collected scope is an absent
|
||||
// observation, never proof. Returns { outcome: "released" | "unavailable",
|
||||
// ... }; `unit` is `absent` once systemd no longer lists the scope.
|
||||
export async function releaseCohort({ kind, unitName, invocationId, shimSocket, waitMs = 3000 }) {
|
||||
if (kind !== "scope") return { outcome: "unavailable", reason: `a ${kind} cohort has no scope to release` };
|
||||
const hello = await shimRequest(shimSocket, "hello");
|
||||
if (!hello.ok) return { outcome: "unavailable", reason: hello.unavailable };
|
||||
if (!invocationId || hello.invocationId !== invocationId) return { outcome: "unavailable", reason: "the shim does not answer for the recorded scope; nothing released" };
|
||||
const released = await shimRequest(shimSocket, "release");
|
||||
if (!released.ok) return { outcome: "unavailable", reason: released.unavailable };
|
||||
const end = Date.now() + waitMs;
|
||||
for (;;) {
|
||||
if (systemdUnits.lookup({ unitName }).state === "absent") return { outcome: "released", unit: "absent" };
|
||||
if (Date.now() >= end) return { outcome: "released", unit: "still listed" };
|
||||
await sleep(20);
|
||||
}
|
||||
}
|
||||
|
||||
export function cohortProof({ binding, stop, result }) {
|
||||
return sealProof(record("cohortProof", {
|
||||
id: newId("cohort-proof"), authority: AUTHORITY, conversation: binding.scope.conversation, execution: binding.execution,
|
||||
|
||||
@@ -23,7 +23,7 @@ import { fileURLToPath } from "node:url";
|
||||
import { randomBytes } from "node:crypto";
|
||||
import { processStart } from "../../discord/src/journal.mjs";
|
||||
import { ALREADY_ACTIVE, ClaimStore, FOREIGN_HOST, UNSAFE_REPLACEMENT, machineId, seatKey, sessionKey } from "./claim.mjs";
|
||||
import { AUTHORITY, PgroupLauncher, bootProof, cohortProof, cohortRefOf, effectReport, forceStopCohort, systemdUnits } from "./cohort.mjs";
|
||||
import { AUTHORITY, PgroupLauncher, bootProof, cohortProof, cohortRefOf, effectReport, forceStopCohort, releaseCohort, systemdUnits } from "./cohort.mjs";
|
||||
import { EngineLink } from "./engine.mjs";
|
||||
import { DIALOG_METHODS, KNOWN_UNSHOWN, NOTIFY_METHODS, deltaBlocks, isFoldedUpdate, messageBlocks, partsOf, roleOf, toolResultText } from "./events.mjs";
|
||||
import { LineSplitter, encodeLine, parseLine } from "./framing.mjs";
|
||||
@@ -224,12 +224,13 @@ export class Controller {
|
||||
this.eligibility = new Map();
|
||||
this.outcomeUnknown = new Set();
|
||||
this.escalating = null;
|
||||
this.scopeReleases = new Map();
|
||||
this.abortWritten = new Set();
|
||||
this.preflightOk = false;
|
||||
this.server = null;
|
||||
this.sockets = new Set();
|
||||
this.events = [];
|
||||
this.evidence = { dropped: { lines: 0, bytes: 0 }, unknownEvents: {}, unshown: {}, folded: 0, dialogs: [], notices: 0, gaps: [], overlaps: [], uncertain: [], stops: [], refusedRevisions: [], internal: [], stderrTail: "" };
|
||||
this.evidence = { dropped: { lines: 0, bytes: 0 }, unknownEvents: {}, unshown: {}, folded: 0, dialogs: [], notices: 0, gaps: [], overlaps: [], uncertain: [], stops: [], releases: [], refusedRevisions: [], internal: [], stderrTail: "" };
|
||||
}
|
||||
|
||||
get binding() {
|
||||
@@ -1555,9 +1556,34 @@ export class Controller {
|
||||
this.#push({ kind: "binding", binding: b });
|
||||
this.#emit("stopped", { stop: stop.id });
|
||||
this.#evidenceAdd("stop", { stop: stop.id, mode: "force-stop", outcome: "stopped", proofs: { cohort: proof.id, effects: effects.id }, resumed });
|
||||
const proven = this.claim;
|
||||
await this.#pause("scope-release", { stop: stop.id });
|
||||
await this.#releaseScope("force-stop", proven);
|
||||
return undefined;
|
||||
}
|
||||
|
||||
// Ends the scope of a claim recorded `stopped` on a cohort proof, once per
|
||||
// claim (#1536). The force stop calls it for the claim it proved, once
|
||||
// that claim is recorded `stopped`; close calls it again, so a release the
|
||||
// stop didn't reach still happens. Every `fail` path above returns before
|
||||
// it: after an uncertain stop the scope and its shim stay as evidence. It
|
||||
// never rejects: a failed release is an `unavailable` entry.
|
||||
#releaseScope(why, claim = this.claim) {
|
||||
const rec = claim?.record;
|
||||
if (!rec || rec.state !== "stopped" || rec.proof?.kind !== "cohortProof" || !rec.shim) return null;
|
||||
let pending = this.scopeReleases.get(claim.claimId);
|
||||
if (!pending) {
|
||||
pending = releaseCohort({ kind: rec.engine?.kind, unitName: rec.unitName, invocationId: rec.invocationId, shimSocket: rec.shim }).catch((err) => ({ outcome: "unavailable", reason: `release: ${err.code ?? err.message}` })).then((result) => {
|
||||
const entry = { kind: "release", claim: claim.claimId, unitName: rec.unitName, why, ...result, at: this.now().toISOString() };
|
||||
this.evidence.releases.push(entry);
|
||||
this.#push({ kind: "evidence", evidence: entry });
|
||||
return result;
|
||||
});
|
||||
this.scopeReleases.set(claim.claimId, pending);
|
||||
}
|
||||
return pending;
|
||||
}
|
||||
|
||||
// check.mjs `stopped`.
|
||||
#stopped(s) {
|
||||
const b = this.b, p = this.stoppedProof;
|
||||
@@ -1659,10 +1685,12 @@ export class Controller {
|
||||
return "revoked";
|
||||
}
|
||||
|
||||
// Test and shutdown hook. Never a release: the claim is unchanged.
|
||||
// Test and shutdown hook. Never a claim release: the claim is unchanged.
|
||||
// A binding proven stopped has its scope released (#1536).
|
||||
async close({ killEngine = false } = {}) {
|
||||
const exec = this.exec;
|
||||
if (exec) exec.closing = true;
|
||||
if (this.b?.state === "stopped" && this.#stopped(this.stops.get(this.b.stop))) await this.#releaseScope("close");
|
||||
if (killEngine && typeof exec?.proc?.kill === "function") exec.proc.kill();
|
||||
else if (killEngine && exec?.proc?.pid) {
|
||||
try {
|
||||
|
||||
@@ -320,7 +320,7 @@ test("W5: SIGKILL between every publication barrier of release; restart finishes
|
||||
const { ch, mark } = await runRelease(probe);
|
||||
ch.send("close");
|
||||
await ch.exited;
|
||||
// The force stop leaves the probe's shim running: nothing sends `release`.
|
||||
// The probe's proven stop releases its scope (#1536); reap the rest now.
|
||||
reap(probe);
|
||||
const points = ch.msgs.slice(mark).filter((m) => m.barrier).map((m) => [m.barrier, m.n]);
|
||||
assert.ok(points.some(([b]) => b === "phase-kill"));
|
||||
@@ -439,7 +439,7 @@ test("W13: crash after the engine spawns, before active; restart finds the live
|
||||
c.close();
|
||||
b.send("close");
|
||||
await b.exited;
|
||||
// The force stop leaves the shim running: nothing sends `release`.
|
||||
// The proven stop releases the scope (#1536); reap the rest now.
|
||||
reap(fx);
|
||||
});
|
||||
|
||||
|
||||
@@ -14,12 +14,12 @@ import { spawn, spawnSync } from "node:child_process";
|
||||
import { dirname, join } from "node:path";
|
||||
import { ClaimStore, FOREIGN_HOST, newClaimId, unitNameFor } from "../src/claim.mjs";
|
||||
import { ConversationClient } from "../src/client.mjs";
|
||||
import { AUTHORITY, PgroupLauncher, ScopeLauncher, scopeAvailable, shimRequest, systemctlShow, systemdUnits } from "../src/cohort.mjs";
|
||||
import { AUTHORITY, PgroupLauncher, ScopeLauncher, forceStopCohort, releaseCohort, scopeAvailable, shimRequest, systemctlShow, systemdUnits } from "../src/cohort.mjs";
|
||||
import { Controller, ELIGIBILITY, TEST_ENGINE } from "../src/controller.mjs";
|
||||
import { ENGINE_ENV, ENGINE_PIN_MISMATCH, engineEnv } from "../src/pi-pin.mjs";
|
||||
import { FixtureVerifier, newId } from "../src/records.mjs";
|
||||
import { ControlClient, FakeLauncher } from "./fake-pi.mjs";
|
||||
import { FAST, REPO, assistantEntry, claimRecords, cleanupAll, controllerFor, fixture, killChildren, killShims, noUnits, reap, receiptState, shimsGone, spawnController, started, tick } from "./harness.mjs";
|
||||
import { FAST, REPO, assistantEntry, claimRecords, cleanupAll, controllerFor, fixture, killChildren, killShims, liveShims, noUnits, reap, receiptState, shimsGone, spawnController, started, tick } from "./harness.mjs";
|
||||
|
||||
const reaped = [];
|
||||
const strays = new Set();
|
||||
@@ -51,6 +51,17 @@ async function until(pred, ms = 4000, what = "condition") {
|
||||
}
|
||||
}
|
||||
|
||||
// `p`'s value, or null if `ms` pass first (Filbert N3 on #1533).
|
||||
async function within(p, ms) {
|
||||
let timer;
|
||||
const late = new Promise((r) => (timer = setTimeout(() => r(null), ms)));
|
||||
try {
|
||||
return await Promise.race([p, late]);
|
||||
} finally {
|
||||
clearTimeout(timer);
|
||||
}
|
||||
}
|
||||
|
||||
// A zombie has exited; only its parent hasn't reaped it yet.
|
||||
function alive(pid) {
|
||||
try {
|
||||
@@ -93,12 +104,12 @@ function gate() {
|
||||
|
||||
// A controller in this process on a real engine process: the fake engine
|
||||
// under ScopeLauncher ("scope") or PgroupLauncher ("pgroup").
|
||||
async function live({ kind = "scope", barrier = null, verifier = new FixtureVerifier({ authorities: [AUTHORITY] }) } = {}) {
|
||||
async function live({ kind = "scope", barrier = null, verifier = new FixtureVerifier({ authorities: [AUTHORITY] }), launcher = null } = {}) {
|
||||
const fx = track(fixture());
|
||||
const control = join(fx.base, "fake.sock");
|
||||
const ctrl = new Controller({
|
||||
fixtureRoot: fx.base, claimRoot: fx.claimRoot, socketDir: fx.socketDir, sessionFile: fx.sessionFile, seat: fx.seat,
|
||||
launcher: kind === "scope" ? new ScopeLauncher() : new PgroupLauncher(),
|
||||
launcher: launcher ?? (kind === "scope" ? new ScopeLauncher() : new PgroupLauncher()),
|
||||
engine: { cwd: fx.proj }, [TEST_ENGINE]: { command: process.execPath, preArgs: [FAKE_PI], env: { ...process.env, FAKE_PI_CONTROL: control, FAKE_PI_LOG: join(fx.base, "fake.log") } },
|
||||
verifier, units: kind === "scope" ? systemdUnits : noUnits, timeouts: FAST, barrier,
|
||||
});
|
||||
@@ -742,12 +753,184 @@ test("K19: a scope launched with only the engine environment still reaches the u
|
||||
const killed = await shimRequest(socketPath, "kill", { timeoutMs: 5000 }, 8000);
|
||||
const released = killed.ok ? await shimRequest(socketPath, "release") : killed;
|
||||
if (!released.ok) spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${unitName}.scope`], { stdio: "ignore", timeout: 5000 });
|
||||
await proc.exited;
|
||||
assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
|
||||
assert.equal(killed.ok, true, JSON.stringify(killed));
|
||||
assert.equal(released.ok, true, JSON.stringify(released));
|
||||
}
|
||||
});
|
||||
|
||||
// ---- scope release after a force stop (#1536) ------------------------------
|
||||
|
||||
const unitGone = (unit) => systemdUnits.lookup({ unitName: unit }).state === "absent";
|
||||
const unitActive = (unit) => systemctlShow(unit)?.activeState === "active";
|
||||
|
||||
test("R1: after a proven force stop the controller releases the scope; the shim and the unit are gone", NEEDS_SCOPE, async () => {
|
||||
const h = await live();
|
||||
const unit = h.rec().unitName, shim = h.rec().shim;
|
||||
try {
|
||||
const shimPid = (await shimRequest(shim, "hello")).shimPid;
|
||||
await forceStop(h);
|
||||
assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
|
||||
await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
|
||||
assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unitName, r.claim]), [["force-stop", "released", unit, h.ctrl.claim.claimId]]);
|
||||
await until(() => unitGone(unit), 8000, "the scope to go");
|
||||
assert.equal(alive(shimPid), false);
|
||||
assert.equal((await shimRequest(shim, "hello")).ok, false);
|
||||
} finally {
|
||||
await h.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("R2: close releases the scope of a binding proven stopped, once, when the stop's own release hasn't run", NEEDS_SCOPE, async () => {
|
||||
const g = gate();
|
||||
const h = await live({ barrier: g.barrier });
|
||||
const unit = h.rec().unitName;
|
||||
try {
|
||||
g.hold("scope-release");
|
||||
await forceStop(h);
|
||||
await g.waitHeld("scope-release");
|
||||
assert.equal(h.ctrl.binding.state, "stopped");
|
||||
assert.ok(unitActive(unit), "held before the stop's release: the scope is still up");
|
||||
assert.deepEqual(h.ctrl.evidence.releases, []);
|
||||
await h.ctrl.close();
|
||||
assert.ok(unitGone(unit), JSON.stringify(systemctlShow(unit)));
|
||||
assert.deepEqual(h.ctrl.evidence.releases.map((r) => [r.why, r.outcome, r.unitName, r.unit]), [["close", "released", unit, "absent"]]);
|
||||
g.release("scope-release");
|
||||
await tick(100);
|
||||
assert.equal(h.ctrl.evidence.releases.length, 1, "the stop's own release reuses the close's");
|
||||
} finally {
|
||||
g.release("scope-release");
|
||||
h.c.close();
|
||||
h.fake.close();
|
||||
reap(h.fx);
|
||||
}
|
||||
});
|
||||
|
||||
// The real stop runs, so the cohort is empty and the shim would accept
|
||||
// `release`; only the outcome says unavailable.
|
||||
class UnavailableAfterStop extends ScopeLauncher {
|
||||
async forceStop(a) {
|
||||
await forceStopCohort(a);
|
||||
return { outcome: "unavailable", reason: "R3: evidence withheld" };
|
||||
}
|
||||
}
|
||||
class RejectingVerifier extends FixtureVerifier {
|
||||
constructor() {
|
||||
super({ authorities: [AUTHORITY] });
|
||||
}
|
||||
cohort() {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
test("R3: no release after an unavailable stop or an unverified proof: the scope and its shim stay as evidence", NEEDS_SCOPE, async () => {
|
||||
for (const [what, opts, reason] of [
|
||||
["unavailable", { launcher: new UnavailableAfterStop() }, /R3: evidence withheld/],
|
||||
["unverified", { verifier: new RejectingVerifier() }, /proof not verified/],
|
||||
]) {
|
||||
const h = await live(opts);
|
||||
const unit = h.rec().unitName, shim = h.rec().shim;
|
||||
try {
|
||||
await forceStop(h);
|
||||
assert.equal(h.ctrl.binding.state, "uncertain", what);
|
||||
assert.match(h.ctrl.evidence.stops.at(-1).reason, reason);
|
||||
await h.ctrl.close();
|
||||
await tick(200);
|
||||
assert.deepEqual(h.ctrl.evidence.releases, [], what);
|
||||
assert.ok(unitActive(unit), `${what}: the scope stays up`);
|
||||
const hello = await shimRequest(shim, "hello");
|
||||
assert.equal(hello.ok, true, `${what}: ${JSON.stringify(hello)}`);
|
||||
assert.equal(hello.invocationId, h.rec().invocationId);
|
||||
} finally {
|
||||
h.c.close();
|
||||
h.fake.close();
|
||||
reap(h.fx);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
// A normal engine exit leaves the shim and the scope: the shim exits only on
|
||||
// `release`. The controller doesn't release then: the binding is uncertain
|
||||
// and the scope is what a force stop reads. The proven stop releases it.
|
||||
test("R4: after a normal engine exit the scope stays until a proven force stop releases it", NEEDS_SCOPE, async () => {
|
||||
const h = await live();
|
||||
const unit = h.rec().unitName, shim = h.rec().shim;
|
||||
try {
|
||||
assert.equal((await h.fake.call("exit")).ok, true);
|
||||
await until(() => h.ctrl.binding.state === "uncertain", 8000, "the binding to see the exit");
|
||||
await until(async () => (await shimRequest(shim, "hello")).engineExit !== null, 4000, "the shim to see the exit");
|
||||
assert.equal((await shimRequest(shim, "hello")).engineExit.code, 0);
|
||||
await tick(200);
|
||||
assert.ok(unitActive(unit), "the shim keeps the scope after the engine exits");
|
||||
assert.deepEqual(h.ctrl.evidence.releases, []);
|
||||
await forceStop(h);
|
||||
assert.equal(h.ctrl.binding.state, "stopped", JSON.stringify(h.ctrl.evidence.stops));
|
||||
assert.deepEqual(h.ctrl.stoppedProof.proof.members, []);
|
||||
await until(() => h.ctrl.evidence.releases.length > 0, 8000, "the release");
|
||||
assert.equal(h.ctrl.evidence.releases[0].outcome, "released");
|
||||
await until(() => unitGone(unit), 8000, "the scope to go");
|
||||
} finally {
|
||||
await h.close();
|
||||
}
|
||||
});
|
||||
|
||||
test("R5: the harness finds a running shim and reaps it, and never signals a shim outside a mosaic-chat- scope (Filbert N1 and N2 on #1533)", NEEDS_SCOPE, async () => {
|
||||
const fx = track(fixture());
|
||||
mkdirSync(fx.socketDir, { recursive: true });
|
||||
{
|
||||
const unitName = unitNameFor(newClaimId());
|
||||
const socketPath = join(fx.socketDir, "r5a.sock");
|
||||
const proc = await new ScopeLauncher().launch({ unitName, socketPath, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
|
||||
const hello = await shimRequest(socketPath, "hello");
|
||||
assert.deepEqual(liveShims(fx.base), [{ pid: hello.shimPid, socket: socketPath, unit: unitName }]);
|
||||
reap(fx);
|
||||
assert.deepEqual(await shimsGone(fx.base), []);
|
||||
assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
|
||||
await until(() => unitGone(unitName), 8000, "the reaped scope to go");
|
||||
}
|
||||
{
|
||||
const unitName = `r5-not-mosaic-${newClaimId().slice(0, 12)}`;
|
||||
const socketPath = join(fx.socketDir, "r5b.sock");
|
||||
const proc = await new ScopeLauncher().launch({ unitName, socketPath, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
|
||||
let released = null;
|
||||
try {
|
||||
const hello = await shimRequest(socketPath, "hello");
|
||||
assert.deepEqual(killShims(fx.base).map((s) => [s.pid, s.unit]), [[hello.shimPid, unitName]]);
|
||||
await tick(200);
|
||||
assert.ok(alive(hello.shimPid), "the refused shim got no signal");
|
||||
assert.ok(unitActive(unitName), "the refused scope got no signal");
|
||||
} finally {
|
||||
const killed = await shimRequest(socketPath, "kill", { timeoutMs: 5000 }, 8000);
|
||||
released = killed.ok ? await shimRequest(socketPath, "release") : killed;
|
||||
if (!released.ok) spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${unitName}.scope`], { stdio: "ignore", timeout: 5000 });
|
||||
assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
|
||||
}
|
||||
assert.equal(released.ok, true, JSON.stringify(released));
|
||||
}
|
||||
});
|
||||
|
||||
test("R6: releaseCohort releases nothing for another invocation ID or a cohort without a scope", NEEDS_SCOPE, async () => {
|
||||
const fx = track(fixture());
|
||||
mkdirSync(fx.socketDir, { recursive: true });
|
||||
const unitName = unitNameFor(newClaimId());
|
||||
const shimSocket = join(fx.socketDir, "r6.sock");
|
||||
const proc = await new ScopeLauncher().launch({ unitName, socketPath: shimSocket, command: "/bin/sleep", args: ["60"], cwd: fx.proj, env: process.env });
|
||||
try {
|
||||
const killed = await shimRequest(shimSocket, "kill", { timeoutMs: 5000 }, 8000);
|
||||
assert.equal(killed.ok, true, JSON.stringify(killed));
|
||||
for (const [kind, invocationId] of [["scope", "another-invocation"], ["scope", null], ["pgroup", proc.invocationId], ["fake", proc.invocationId]]) {
|
||||
const r = await releaseCohort({ kind, unitName, invocationId, shimSocket });
|
||||
assert.equal(r.outcome, "unavailable", `${kind} ${invocationId}: ${JSON.stringify(r)}`);
|
||||
assert.equal((await shimRequest(shimSocket, "hello")).ok, true, `${kind} ${invocationId}: the shim got no release`);
|
||||
}
|
||||
assert.ok(unitActive(unitName));
|
||||
assert.deepEqual(await releaseCohort({ kind: "scope", unitName, invocationId: proc.invocationId, shimSocket }), { outcome: "released", unit: "absent" });
|
||||
assert.notEqual(await within(proc.exited, 10000), null, "the scope's process exited");
|
||||
} finally {
|
||||
reap(fx);
|
||||
}
|
||||
});
|
||||
|
||||
// Last in the file. A shim ignores SIGTERM and outlives its controller, so a
|
||||
// test that leaves one running relies on the after hook's kill (#1533).
|
||||
test("no shim from this file's tests is left running for the after hook", async () => {
|
||||
|
||||
@@ -697,6 +697,8 @@ async function main() {
|
||||
waitPaused: () => fake.waitPaused(req.point),
|
||||
// H19: stop reading stdin so the controller's write fills the pipe.
|
||||
stall: () => void process.stdin.pause(),
|
||||
// A normal engine exit, after the reply is written (#1536).
|
||||
exit: () => void setTimeout(() => process.exit(req.code ?? 0), 20),
|
||||
};
|
||||
if (!ops[req.op]) return sock.write(encodeLine({ id: req.id, ok: false, error: `unknown op ${req.op}` }));
|
||||
const failed = (err) => sock.write(encodeLine({ id: req.id, ok: false, error: String(err.message) }));
|
||||
|
||||
@@ -222,8 +222,9 @@ export function reap(fx) {
|
||||
}
|
||||
|
||||
// The shims under `root` still running, read from /proc. A shim ignores
|
||||
// SIGTERM and only exits on `release` (src/shim.mjs), and a force stop
|
||||
// never sends `release`, so a stopped scope stays up until it is killed.
|
||||
// SIGTERM and only exits on `release` (src/shim.mjs). The controller sends
|
||||
// `release` only after a proven, recorded force stop (#1536), so a scope
|
||||
// left uncertain, or one a test launched itself, stays up until killed.
|
||||
export function liveShims(root = tmp) {
|
||||
const out = [];
|
||||
for (const pid of lsdir("/proc").filter((n) => /^\d+$/.test(n))) {
|
||||
@@ -242,16 +243,23 @@ export function liveShims(root = tmp) {
|
||||
}
|
||||
|
||||
// SIGKILLs each shim's scope, which takes the engine with it, and the shim
|
||||
// itself in case the scope kill doesn't land.
|
||||
// itself in case the scope kill doesn't land. A shim whose unit isn't a
|
||||
// `mosaic-chat-` scope gets no signal at all: it is returned, not killed.
|
||||
export function killShims(root = tmp) {
|
||||
const refused = [];
|
||||
for (const s of liveShims(root)) {
|
||||
if (s.unit) spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${s.unit}.scope`], { stdio: "ignore", timeout: 5000 });
|
||||
if (!/^mosaic-chat-/.test(s.unit ?? "")) {
|
||||
refused.push(s);
|
||||
continue;
|
||||
}
|
||||
spawnSync("systemctl", ["--user", "kill", "--signal=SIGKILL", `${s.unit}.scope`], { stdio: "ignore", timeout: 5000 });
|
||||
try {
|
||||
process.kill(s.pid, "SIGKILL");
|
||||
} catch {
|
||||
// gone
|
||||
}
|
||||
}
|
||||
return refused;
|
||||
}
|
||||
|
||||
// Waits up to `ms` for every shim under `root` to go; returns those left.
|
||||
|
||||
Reference in New Issue
Block a user