From fea2fd2c058be49e6c8ed0a38e5590b483fd9790 Mon Sep 17 00:00:00 2001 From: flujo-app Date: Thu, 12 Nov 2026 12:48:46 -0500 Subject: [PATCH] A fleet being stopped or retired opens no new connection, and stop and retire close the one that is open; release 0.7.4 A caller's status read that arrived while the fleet was being stopped opened a private proxy that nothing closed after retire; the fly proxy process then kept the caller alive (Fixes #36). --- package-lock.json | 4 +- package.json | 2 +- src/cloud-lite/fleet.mjs | 28 +++++++--- test/cloud-lite-connection.test.mjs | 81 +++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+), 10 deletions(-) create mode 100644 test/cloud-lite-connection.test.mjs diff --git a/package-lock.json b/package-lock.json index 3f0054a..f80122e 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "flujo-factory", - "version": "0.7.3", + "version": "0.7.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "flujo-factory", - "version": "0.7.3", + "version": "0.7.4", "license": "MIT", "dependencies": { "@modelcontextprotocol/sdk": "1.32.1" diff --git a/package.json b/package.json index 6c7c9bc..163c9eb 100644 --- a/package.json +++ b/package.json @@ -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": { diff --git a/src/cloud-lite/fleet.mjs b/src/cloud-lite/fleet.mjs index 635d9be..7ecbddd 100644 --- a/src/cloud-lite/fleet.mjs +++ b/src/cloud-lite/fleet.mjs @@ -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]; @@ -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 (;;) { @@ -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); @@ -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 } = {}) { @@ -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(() => {}); } } } diff --git a/test/cloud-lite-connection.test.mjs b/test/cloud-lite-connection.test.mjs new file mode 100644 index 0000000..997f82c --- /dev/null +++ b/test/cloud-lite-connection.test.mjs @@ -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); +});