Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions package-lock.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "flujo-factory",
"version": "0.7.3",
"version": "0.7.4",
"description": "Durable FLUJO swarm engine and Factory SDK, CLI, and MCP server",
"license": "MIT",
"repository": {
Expand Down
28 changes: 21 additions & 7 deletions src/cloud-lite/fleet.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,9 @@ export class FactoryCloudLiteFleet {
}
async request(path, body) {
if (!this.connection) {
// A fleet being stopped or retired has no coordinator to reach: never open a proxy for it (a caller's late status
// read would otherwise leave a private proxy process running after the fleet is gone).
if (this.closed()) throw new Error('The fleet is being stopped or is retired: no connection to its coordinator is opened.');
const upstream = await this.upstream();
const inventory = JSON.parse(await this.fly(['machine', 'list', '--app', this.state.plan.app, '--json']));
const first = this.state.machines[0];
Expand All @@ -300,6 +303,8 @@ export class FactoryCloudLiteFleet {
}
if (owned.state !== 'started') throw new Error('Original coordinator Machine is not started.');
this.connection = await upstream.proxy({ app: this.state.plan.app, org: this.state.plan.org ?? 'personal', machineId: first.id, localPort: await upstream.unusedLoopbackPort() });
// the fleet was stopped while this proxy was being opened: close it again at once
if (this.closed()) { await this.disconnect(); throw new Error('The fleet was stopped while its connection was being opened.'); }
const deadline = Date.now() + 30000;
try {
for (;;) {
Expand All @@ -323,9 +328,12 @@ export class FactoryCloudLiteFleet {
if (!response.ok) throw new Error(`Fleet coordinator ${path} returned ${response.status}.`); return await response.json();
} catch (error) { await this.disconnect(); throw error; }
}
async disconnect() { if (this.connection) { await this.connection.stop(); this.connection = null; } }
async disconnect() { if (this.connection) { const connection = this.connection; this.connection = null; await connection.stop(); } }
/** Being stopped here, or retired: from then on no new connection to the coordinator is opened (start() opens it up again). */
closed() { return !!this.closing || this.state?.status === 'retired'; }
async start({ attempt } = {}) {
if (!/^[a-z0-9-]{3,64}$/.test(attempt ?? '')) throw new TypeError('Select a stable recovery attempt identity.');
this.closing = false;
for (const machine of this.state.machines) await this.stage(`${machine.id}.start.${attempt}`, async () => {
const inventory = JSON.parse(await this.fly(['machine', 'list', '--app', this.state.plan.app, '--json']));
const original = inventory.find(value => value.id === machine.id);
Expand Down Expand Up @@ -548,10 +556,14 @@ export class FactoryCloudLiteFleet {
if (complete) this.state.artifactsCollectedAt = new Date().toISOString(); this.save(); return { directory: artifacts, complete, collected: receipts.length, expected: this.state.machines.length };
}
async stop() {
for (const machine of [...this.state.machines].reverse()) await this.stage(`${machine.id}.stop`, async () => {
await this.fly(['machine', 'stop', machine.id, '--app', this.state.plan.app]); return { stopped: true, machineId: machine.id };
});
this.state.status = 'stopped'; this.save();
// no new connection from here on, and the one that is open goes with the machines (also when a stop fails)
this.closing = true;
try {
for (const machine of [...this.state.machines].reverse()) await this.stage(`${machine.id}.stop`, async () => {
await this.fly(['machine', 'stop', machine.id, '--app', this.state.plan.app]); return { stopped: true, machineId: machine.id };
});
this.state.status = 'stopped'; this.save();
} finally { await this.disconnect().catch(() => {}); }
}
/** Quiesce original sessions, verify remote immutable evidence, then stop exactly the owned machines. */
async verifiedShutdown({ attempt = 'final', timeoutMs = 120000, evidenceTimeoutMs = 300000, providerTimeoutMs = 120000, retire = false } = {}) {
Expand Down Expand Up @@ -667,7 +679,9 @@ export class FactoryCloudLiteFleet {
throw new Error('Retirement requires the exact original Machine and volume inventory; retain resources whose ownership differs.');
}
}
await this.stage('app.retire', async () => { await this.fly(['apps', 'destroy', this.state.plan.app, '--yes']); return { app: this.state.plan.app, retired: true }; });
this.state.status = 'retired'; this.save();
try {
await this.stage('app.retire', async () => { await this.fly(['apps', 'destroy', this.state.plan.app, '--yes']); return { app: this.state.plan.app, retired: true }; });
this.state.status = 'retired'; this.save();
} finally { await this.disconnect().catch(() => {}); }
}
}
81 changes: 81 additions & 0 deletions test/cloud-lite-connection.test.mjs
Original file line number Diff line number Diff line change
@@ -0,0 +1,81 @@
// Found in a real Greenswarm run (Nov 12, flujo-app/FACTORY#36): a caller's status read that arrived while the fleet was being
// stopped opened a new private proxy, and nothing closed it after retire; the `fly proxy` process then kept the caller alive.
// A fleet being stopped or retired opens no new connection, and stop and retire close the one that is open.
import test from 'node:test';
import assert from 'node:assert/strict';
import { FactoryCloudLiteFleet } from '../src/cloud-lite/fleet.mjs';
import { validateCloudLitePlan } from '../src/cloud-lite/protocol.mjs';
import { tempDir } from './support/temp-dir.mjs';

const plan = () => validateCloudLitePlan({ runId: 'lite-conn', app: 'factory-lite-conn', repository: 'mario-andreschak/FLUJO', goal: 'Close what you open', machines: 1, agentsPerMachine: 1 });

/** A fleet with one started coordinator Machine, a stand-in for fly and for the private proxy that counts what it opens. */
function fleetWithProxy({ onMachineStop = async () => {}, slowProxy = null } = {}) {
const fleet = new FactoryCloudLiteFleet({ directory: tempDir('factory-cloud-lite-conn-') });
fleet.state = { plan: plan(), status: 'provisioned', machines: [{ id: 'm-1', name: 'factory-lite-conn-m0', volumeId: 'v-1' }], operations: {} };
fleet.save();
const proxies = { opened: 0, stopped: 0 };
fleet.execute = async (_bin, args) => {
if (args[0] === 'machine' && args[1] === 'list') return JSON.stringify([{ id: 'm-1', name: 'factory-lite-conn-m0', state: 'started', config: { metadata: { 'factory.run': 'lite-conn', 'factory.machine': 'm0' } } }]);
if (args[0] === 'machine' && args[1] === 'stop') { await onMachineStop(); return ''; }
return '';
};
fleet.infrastructure = {
unusedLoopbackPort: async () => 1,
proxy: async () => {
if (slowProxy) await slowProxy();
proxies.opened++;
return { origin: 'http://127.0.0.1:1', check() {}, stop: async () => { proxies.stopped++; } };
},
};
return { fleet, proxies };
}

test('a status read that arrives while the fleet is being stopped opens no proxy', async () => {
let late = null;
const { fleet, proxies } = fleetWithProxy({ onMachineStop: async () => { late = fleet.request('/operator/status'); } });
await fleet.stop();
await assert.rejects(late, /being stopped/);
assert.equal(proxies.opened, 0);
assert.ok(!fleet.connection);
});

test('stop closes the connection that is open, and retire leaves none behind', async () => {
const { fleet, proxies } = fleetWithProxy();
fleet.connection = { origin: 'http://127.0.0.1:1', check() {}, stop: async () => { proxies.stopped++; } };
await fleet.stop();
assert.equal(proxies.stopped, 1);
assert.ok(!fleet.connection);
// a retire whose app removal was already done (journal) still ends without a connection, and after it nothing reconnects
fleet.state.operations['app.retire'] = { state: 'succeeded', result: { app: 'factory-lite-conn', retired: true } };
fleet.save();
fleet.connection = { origin: 'http://127.0.0.1:1', check() {}, stop: async () => { proxies.stopped++; } };
await fleet.retire();
assert.equal(fleet.state.status, 'retired');
assert.ok(!fleet.connection);
await assert.rejects(fleet.request('/operator/status'), /retired/);
assert.equal(proxies.opened, 0);
});

test('a proxy that was being opened when the stop began is closed again at once', async () => {
let release;
const { fleet, proxies } = fleetWithProxy({ slowProxy: () => new Promise((r) => { release = r; }) });
const opening = fleet.request('/operator/status');
while (!release) await new Promise((r) => setImmediate(r));
const stopping = fleet.stop();
release();
await assert.rejects(opening, /stopped while its connection was being opened/);
await stopping;
assert.equal(proxies.opened, 1);
assert.equal(proxies.stopped, 1);
assert.ok(!fleet.connection);
});

test('a fleet started again after a stop connects as before', async () => {
const { fleet } = fleetWithProxy();
await fleet.stop();
fleet.state.operations = {};
fleet.save();
await fleet.start({ attempt: 'again-1' }).catch(() => {});
assert.equal(fleet.closed(), false);
});
Loading