From 8b7df8cb65ae8d3b89db4ef8315fdafeb8822c0d Mon Sep 17 00:00:00 2001 From: chick Date: Sat, 5 Sep 2026 19:52:27 +0800 Subject: [PATCH] fix(instances): release the hub binding before forgetting a node Deletion now follows the published contract: prepare returns the one-time token and the console spends it on DELETE /api/v1/instances/:id with If-Match, where the generic execute endpoint had been rejecting it as an operation the safe transport cannot replay. An online device is told to release its own binding first. A refusal keeps the record and reports UNBIND_FAILED on the job item so the console can say why the node is still there; a node that never answers a heartbeat is forgotten locally. --- .../connections/fleet-heartbeat.test.ts | 115 +++++++++++++++++- .../connections/fleet-heartbeat.ts | 87 ++++++++++++- .../application/notifications/smtp-sender.ts | 15 ++- .../delete-instance-operation.test.ts | 92 +++++++++++++- .../operations/delete-instance-operation.ts | 101 ++++++++++++++- apps/api/src/control-plane.test.ts | 80 ++++++++++++ apps/api/src/control-plane.ts | 72 +++++++++-- .../instance-api-data-source.test.ts | 94 ++++++++++++-- .../src/instances/instance-api-data-source.ts | 56 ++++++++- apps/web/src/instances/instance-crud.test.tsx | 17 +++ apps/web/src/instances/instance-crud.tsx | 7 +- 11 files changed, 691 insertions(+), 45 deletions(-) diff --git a/apps/api/src/application/connections/fleet-heartbeat.test.ts b/apps/api/src/application/connections/fleet-heartbeat.test.ts index 58e7d3f..3f41df8 100644 --- a/apps/api/src/application/connections/fleet-heartbeat.test.ts +++ b/apps/api/src/application/connections/fleet-heartbeat.test.ts @@ -2,9 +2,9 @@ import { afterEach, describe, expect, it, vi } from 'vitest'; import type { Instance, InstancePage } from '@multi-simadmin/contracts'; import type { InstanceService } from '../instances/instance-service.js'; -import type { ConnectionProbe } from './connection-probe.js'; +import type { ConnectionState, ConnectionProbe } from './connection-probe.js'; import type { ConnectionSettingsService } from './connection-settings-service.js'; -import { FleetHeartbeatCoordinator } from './fleet-heartbeat.js'; +import { FleetHeartbeatCoordinator, type FleetStateTransition } from './fleet-heartbeat.js'; function instance(id: string): Instance { return { @@ -28,6 +28,11 @@ function coordinator(options: { readonly probe?: (id: string) => Promise; readonly heartbeatMs?: number; readonly concurrency?: number; + /** Reachability the snapshot journal already holds when the first beat starts. */ + readonly before?: ReadonlyMap; + /** Lets a test flip a node to reachable-but-unauthenticated after it answers. */ + readonly authenticated?: (id: string) => boolean; + readonly onTransition?: (transitions: readonly FleetStateTransition[]) => Promise; }) { const listed: string[] = []; const instances = { @@ -39,12 +44,30 @@ function coordinator(options: { }, } as unknown as InstanceService; const probed: string[] = []; + const journal = new Map(options.before ?? []); const probe = { async test(id: string) { probed.push(id); - await options.probe?.(id); - return { instanceId: id, authenticated: true, checkedAt: '2026-09-05T00:00:00.000Z' }; + try { + await options.probe?.(id); + } catch { + // A failed probe leaves the previous snapshot in place and lets it expire on its own, + // which is exactly what the offline window is for. + throw new Error('probe failed'); + } + const authenticated = options.authenticated?.(id) ?? true; + journal.set(id, { + reachable: true, + authenticated, + checkedAt: '2026-09-05T00:00:00.000Z', + }); + return { + instanceId: id, + authenticated, + checkedAt: '2026-09-05T00:00:00.000Z', + }; }, + reachability: () => journal, } as unknown as ConnectionProbe; const settings = { heartbeatMs: options.heartbeatMs ?? 30_000, @@ -56,8 +79,9 @@ function coordinator(options: { pageSize: options.pageSize, now: () => new Date('2026-09-05T00:00:00.000Z'), ...(options.concurrency === undefined ? {} : { concurrency: options.concurrency }), + ...(options.onTransition === undefined ? {} : { onTransition: options.onTransition }), }); - return { beat, listed, probed }; + return { beat, listed, probed, journal }; } afterEach(() => { @@ -76,6 +100,7 @@ describe('FleetHeartbeatCoordinator', () => { await expect(beat.runOnce()).resolves.toEqual({ probed: 3, failed: 0, + transitions: [], startedAt: '2026-09-05T00:00:00.000Z', finishedAt: '2026-09-05T00:00:00.000Z', }); @@ -83,6 +108,86 @@ describe('FleetHeartbeatCoordinator', () => { expect([...probed].sort()).toEqual(['a', 'b', 'c']); }); + it('records a baseline on the first beat instead of announcing every node as recovered', async () => { + const announced: (readonly FleetStateTransition[])[] = []; + const { beat } = coordinator({ + pageSize: 10, + pages: new Map([[1, [instance('a'), instance('b')]]]), + onTransition: async (transitions) => { + announced.push(transitions); + }, + }); + await expect(beat.runOnce()).resolves.toMatchObject({ transitions: [] }); + await expect(beat.runOnce()).resolves.toMatchObject({ transitions: [] }); + expect(announced).toEqual([]); + }); + + it('announces a node only once its snapshot has actually expired', async () => { + const down: ConnectionState = { + reachable: false, + authenticated: false, + checkedAt: '2026-09-04T00:00:00.000Z', + }; + const up: ConnectionState = { + reachable: true, + authenticated: true, + checkedAt: '2026-09-04T00:00:00.000Z', + }; + const announced: (readonly FleetStateTransition[])[] = []; + const { beat, journal } = coordinator({ + pageSize: 10, + concurrency: 1, + pages: new Map([[1, [instance('a'), instance('b')]]]), + before: new Map([ + ['a', up], + ['b', down], + ]), + probe: async (id) => { + // b never answers; a answers on the first beat and goes quiet afterwards. + if (id === 'b' || !journal.get('a')?.reachable) throw new Error('ECONNREFUSED'); + }, + onTransition: async (transitions) => { + announced.push(transitions); + }, + }); + + // a is reachable and b is still down, so nothing moved against the baseline. + await expect(beat.runOnce()).resolves.toMatchObject({ + probed: 1, + failed: 1, + transitions: [], + }); + + // a's snapshot ages out. A single lost probe would not be enough on its own; the beat + // reports offline only once the offline window the operator configured has passed. + journal.set('a', down); + await expect(beat.runOnce()).resolves.toMatchObject({ + transitions: [{ instanceId: 'a', from: 'online', to: 'offline' }], + }); + expect(announced).toHaveLength(1); + }); + + it('announces an expired device session as a state change', async () => { + const announced: (readonly FleetStateTransition[])[] = []; + let sessionValid = true; + const { beat } = coordinator({ + pageSize: 10, + concurrency: 1, + pages: new Map([[1, [instance('a')]]]), + authenticated: () => sessionValid, + onTransition: async (transitions) => { + announced.push(transitions); + }, + }); + // The first beat only records that the node is healthy. + await expect(beat.runOnce()).resolves.toMatchObject({ transitions: [] }); + sessionValid = false; + await expect(beat.runOnce()).resolves.toMatchObject({ + transitions: [{ instanceId: 'a', from: 'online', to: 'auth-required' }], + }); + expect(announced).toHaveLength(1); + }); + it('stops paging on a short page and probes a duplicated id once', async () => { const { beat, listed, probed } = coordinator({ pageSize: 2, diff --git a/apps/api/src/application/connections/fleet-heartbeat.ts b/apps/api/src/application/connections/fleet-heartbeat.ts index e5ca780..be370c0 100644 --- a/apps/api/src/application/connections/fleet-heartbeat.ts +++ b/apps/api/src/application/connections/fleet-heartbeat.ts @@ -1,10 +1,42 @@ -import type { ConnectionProbe } from './connection-probe.js'; +import type { ConnectionProbe, ConnectionState } from './connection-probe.js'; import type { ConnectionSettingsService } from './connection-settings-service.js'; import type { InstanceService } from '../instances/instance-service.js'; +/** Reachability class the console and the notification rules both speak in. */ +export type FleetObservedState = 'online' | 'auth-required' | 'offline'; + +/** + * Operator-facing copy shared by the notification templates so every channel reads alike. + * `FLEET_STATE_NAMES` describes a steady state, `FLEET_STATE_EVENTS` describes arriving at it. + */ +export const FLEET_STATE_NAMES: Readonly> = Object.freeze({ + online: '在线', + 'auth-required': '需要认证', + offline: '离线', +}); + +export const FLEET_STATE_EVENTS: Readonly> = Object.freeze({ + online: '恢复在线', + 'auth-required': '设备会话已失效', + offline: '已离线', +}); + +export const FLEET_STATE_CODES: Readonly> = Object.freeze({ + online: 'online', + 'auth-required': 'auth_required', + offline: 'offline', +}); + +export interface FleetStateTransition { + readonly instanceId: string; + readonly from: FleetObservedState; + readonly to: FleetObservedState; +} + export interface FleetHeartbeatSummary { readonly probed: number; readonly failed: number; + readonly transitions: readonly FleetStateTransition[]; readonly startedAt: string; readonly finishedAt: string; } @@ -17,12 +49,23 @@ export interface FleetHeartbeatOptions { /** Devices touched at the same time; the control plane talks to LAN hosts, not the internet. */ readonly concurrency?: number; readonly pageSize?: number; + /** + * Called once per beat with every state change the beat revealed. Reachability is derived + * from snapshot expiry, so a single lost probe never fires anything; a node has to stay + * down for the whole offline window before it counts as a transition. + */ + readonly onTransition?: (transitions: readonly FleetStateTransition[]) => Promise; } const DEFAULT_CONCURRENCY = 6; const DEFAULT_PAGE_SIZE = 100; const MAX_PAGES = 100; +function observedState(state: ConnectionState | undefined): FleetObservedState { + if (!state?.reachable) return 'offline'; + return state.authenticated ? 'online' : 'auth-required'; +} + /** * Keeps device online state honest without waiting for an operator to open a page: every beat * probes each instance once and lets the snapshot journal carry the result until it expires. @@ -34,9 +77,14 @@ export class FleetHeartbeatCoordinator { readonly #now: () => Date; readonly #concurrency: number; readonly #pageSize: number; + readonly #onTransition: + | ((transitions: readonly FleetStateTransition[]) => Promise) + | undefined; #timer: ReturnType | undefined; #running: Promise | undefined; #stopped = false; + /** States seen by the previous beat; undefined until the first beat establishes a baseline. */ + #previous: ReadonlyMap | undefined; constructor(options: FleetHeartbeatOptions) { this.#instances = options.instances; @@ -51,6 +99,7 @@ export class FleetHeartbeatCoordinator { throw new RangeError('pageSize must be between 1 and 100'); this.#concurrency = concurrency; this.#pageSize = pageSize; + this.#onTransition = options.onTransition; } start(): void { @@ -95,6 +144,7 @@ export class FleetHeartbeatCoordinator { for (const instance of current.items) if (!ids.includes(instance.id)) ids.push(instance.id); if (current.items.length < this.#pageSize) break; } + const before = this.#previous; let probed = 0; let failed = 0; let cursor = 0; @@ -115,6 +165,39 @@ export class FleetHeartbeatCoordinator { }, ); await Promise.all(workers); - return { probed, failed, startedAt, finishedAt: this.#now().toISOString() }; + const transitions = this.#diff(before, ids); + if (transitions.length > 0) await this.#onTransition?.(transitions); + return { probed, failed, transitions, startedAt, finishedAt: this.#now().toISOString() }; + } + + /** Snapshot of the given instances, including ones that have never been probed. */ + #states(ids: readonly string[]): ReadonlyMap { + const reachability = this.#probe.reachability(); + const states = new Map(); + for (const id of ids) { + const state = reachability.get(id); + states.set(id, observedState(state)); + } + return states; + } + + /** + * The first beat only records a baseline. Without one, every instance would start as + * "offline" and then announce itself as recovered, so a restart would flood the queue. + */ + #diff( + before: ReadonlyMap | undefined, + ids: readonly string[], + ): readonly FleetStateTransition[] { + const after = this.#states(ids); + this.#previous = after; + if (before === undefined) return []; + const transitions: FleetStateTransition[] = []; + for (const [id, to] of after) { + const from = before.get(id); + if (from === undefined || from === to) continue; + transitions.push({ instanceId: id, from, to }); + } + return transitions; } } diff --git a/apps/api/src/application/notifications/smtp-sender.ts b/apps/api/src/application/notifications/smtp-sender.ts index 24b394a..7becf3e 100644 --- a/apps/api/src/application/notifications/smtp-sender.ts +++ b/apps/api/src/application/notifications/smtp-sender.ts @@ -178,11 +178,20 @@ function expect(reply: { code: number; text: string }, codes: readonly number[]) function plainSocket(host: string, port: number, timeoutMs: number): Promise { return new Promise((resolve, reject) => { const socket = createConnection({ host, port }); - socket.setTimeout(timeoutMs); - socket.once('connect', () => resolve(socket)); - socket.once('timeout', () => { + const onTimeout = (): void => { socket.destroy(); reject(new Error('SMTP 连接超时')); + }; + socket.setTimeout(timeoutMs); + socket.once('timeout', onTimeout); + socket.once('connect', () => { + // The timer only guards the handshake. Every command afterwards carries its own reply + // deadline, and leaving an idle timer armed would destroy the session at the same + // instant that deadline expires, making the reported detail a coin flip between + // "connection closed" and "timed out". + socket.removeListener('timeout', onTimeout); + socket.setTimeout(0); + resolve(socket); }); socket.once('error', reject); }); diff --git a/apps/api/src/application/operations/delete-instance-operation.test.ts b/apps/api/src/application/operations/delete-instance-operation.test.ts index ef75c81..51d7fad 100644 --- a/apps/api/src/application/operations/delete-instance-operation.test.ts +++ b/apps/api/src/application/operations/delete-instance-operation.test.ts @@ -6,6 +6,7 @@ import { migrateDatabase } from '../../infrastructure/database/migrations.js'; import type { SecretStore } from '../../infrastructure/secrets/secret-store.js'; import { DELETE_INSTANCE_PARAMETER_SCHEMA_ID, + type BindingRelease, DeleteInstanceOperation, DeleteInstanceOperationError, } from './delete-instance-operation.js'; @@ -26,7 +27,9 @@ afterEach(() => { for (const database of databases.splice(0)) database.close(); }); -function fixture() { +function fixture( + releaseBinding?: (instanceId: string, requestId: string) => Promise, +) { const db = new Database(':memory:'); db.pragma('foreign_keys=ON'); migrateDatabase(db); @@ -43,6 +46,7 @@ function fixture() { db, instances, now, + ...(releaseBinding === undefined ? {} : { releaseBinding }), idFactory: () => `operation-${++sequence}`, tokenFactory: () => Buffer.alloc(32, 7).toString('base64url'), }); @@ -396,4 +400,90 @@ describe('DeleteInstanceOperation', () => { ).rejects.toMatchObject({ code: 'CONFIRMATION_INVALID' }); expect(db.prepare('SELECT COUNT(*) count FROM jobs').get()).toEqual({ count: 0 }); }); + + it('keeps the node when the device refuses to release its binding', async () => { + const calls: string[] = []; + const { instances, operation } = fixture(async (instanceId, requestId) => { + calls.push(`${instanceId}:${requestId}`); + return { status: 'failed' }; + }); + const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); + const prepared = await operation.prepare({ + operationId: 'deleteInstance', + targets: [{ instanceId: instance.id, revision: 1 }], + parameters: { parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] }, + }); + const job = await operation.executeDelete({ + instanceId: instance.id, + revision: 1, + preparationId: prepared.id, + confirmationToken: prepared.confirmationToken, + actor: 'loopback-control-plane', + requestId: 'request-refused', + }); + expect(calls).toEqual([`${instance.id}:request-refused`]); + expect(job.status).toBe('failed'); + expect(job.items).toEqual([ + { + id: expect.any(String), + targetId: instance.id, + state: 'failed', + error: { + type: 'about:blank', + title: 'Job item failed', + status: 502, + detail: 'The job item did not complete successfully.', + code: 'UNBIND_FAILED', + requestId: 'request-refused', + }, + }, + ]); + expect(await instances.get(instance.id)).toBeDefined(); + }); + + it('forgets a node that never answered, exactly as an offline device is dropped', async () => { + const { instances, operation } = fixture(async () => ({ status: 'skipped' })); + const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); + const prepared = await operation.prepare({ + operationId: 'deleteInstance', + targets: [{ instanceId: instance.id, revision: 1 }], + parameters: { parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] }, + }); + const job = await operation.executeDelete({ + instanceId: instance.id, + revision: 1, + preparationId: prepared.id, + confirmationToken: prepared.confirmationToken, + actor: 'loopback-control-plane', + requestId: 'request-offline', + }); + expect(job.status).toBe('succeeded'); + expect(job.items[0]).not.toHaveProperty('error'); + expect(await instances.get(instance.id)).toBeUndefined(); + }); + + it('treats a transport that throws while the node was online as a refusal', async () => { + const { db, instances, operation } = fixture(async () => { + throw new Error('ECONNREFUSED'); + }); + const instance = await instances.create({ name: 'A', origin: 'http://192.168.1.10' }); + const prepared = await operation.prepare({ + operationId: 'deleteInstance', + targets: [{ instanceId: instance.id, revision: 1 }], + parameters: { parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] }, + }); + const job = await operation.executeDelete({ + instanceId: instance.id, + revision: 1, + preparationId: prepared.id, + confirmationToken: prepared.confirmationToken, + actor: 'loopback-control-plane', + requestId: 'request-thrown', + }); + expect(job.items[0]).toMatchObject({ state: 'failed', error: { code: 'UNBIND_FAILED' } }); + expect(db.prepare('SELECT result_code FROM job_items').get()).toEqual({ + result_code: 'UNBIND_FAILED', + }); + expect(await instances.get(instance.id)).toBeDefined(); + }); }); diff --git a/apps/api/src/application/operations/delete-instance-operation.ts b/apps/api/src/application/operations/delete-instance-operation.ts index ef30915..29b70de 100644 --- a/apps/api/src/application/operations/delete-instance-operation.ts +++ b/apps/api/src/application/operations/delete-instance-operation.ts @@ -6,6 +6,8 @@ import { InstanceService, InstanceServiceError } from '../instances/instance-ser export const DELETE_INSTANCE_PARAMETER_SCHEMA_ID = 'deleteInstance.parameters.v1' as const; const OPERATION_ID = 'deleteInstance'; const ACTOR = 'loopback-control-plane'; +/** Job item code for "the device still holds its central binding", surfaced to the console. */ +export const BINDING_REFUSAL_CODE = 'UNBIND_FAILED' as const; const TTL_MS = 5 * 60 * 1000; const PARAMETERS_DIGEST = createHash('sha256') .update(JSON.stringify({ parameterSchemaId: DELETE_INSTANCE_PARAMETER_SCHEMA_ID, fields: [] })) @@ -53,9 +55,23 @@ interface ExecuteInput { readonly actor: string; readonly requestId: string; } + +/** + * Answer to "release the central binding on this device before we forget it". + * `skipped` means the node never answered a heartbeat, so there is nobody to notify and the + * console deletes its own record; the device re-binds if it ever comes back. `failed` keeps the + * record, because a device that refused or timed out must not be silently orphaned. + */ +export type BindingRelease = + | { readonly status: 'released' } + | { readonly status: 'skipped' } + | { readonly status: 'failed' }; + interface Options { readonly db: Database.Database; readonly instances: InstanceService; + /** Optional so a control plane without a device transport can still delete. */ + readonly releaseBinding?: (instanceId: string, requestId: string) => Promise; readonly now?: () => Date; readonly idFactory?: () => string; readonly tokenFactory?: () => string; @@ -73,6 +89,7 @@ const validRevision = (value: unknown): value is number => export class DeleteInstanceOperation { private readonly db: Database.Database; private readonly instances: InstanceService; + private readonly releaseBinding: Options['releaseBinding']; private readonly clock: () => Date; private readonly id: () => string; private readonly token: () => string; @@ -80,6 +97,7 @@ export class DeleteInstanceOperation { constructor(options: Options) { this.db = options.db; this.instances = options.instances; + this.releaseBinding = options.releaseBinding; this.clock = options.now ?? (() => new Date()); this.id = options.idFactory ?? randomUUID; this.token = options.tokenFactory ?? (() => randomBytes(32).toString('base64url')); @@ -294,6 +312,14 @@ export class DeleteInstanceOperation { 'Confirmation could not be accepted', ); + // An online device is told to release its central binding first, and a refusal keeps the + // record so the operator can retry instead of losing sight of a device that is still + // reporting to a hub. Only a node that cannot be reached at all is forgotten locally. + const release = await this.releaseBindingPort(input.instanceId, input.requestId); + if (release?.status === 'failed') { + this.closeAsFailed(ids, input.requestId, BINDING_REFUSAL_CODE); + return this.job(ids.job); + } try { await this.instances.delete(input.instanceId, input.revision, { allowJobHistory: true }); const finished = this.clock().toISOString(); @@ -339,6 +365,53 @@ export class DeleteInstanceOperation { return this.job(ids.job); } + /** A transport that throws while the node was reachable counts as a refusal, never a pass. */ + private async releaseBindingPort( + instanceId: string, + requestId: string, + ): Promise { + if (!this.releaseBinding) return undefined; + try { + return await this.releaseBinding(instanceId, requestId); + } catch { + return { status: 'failed' }; + } + } + + /** Closes the job without touching the instance row, so the device stays listed and retryable. */ + private closeAsFailed( + ids: { job: string; item: string; attempt: string }, + requestId: string, + resultCode: string, + ): void { + const finished = this.clock().toISOString(); + this.db.transaction(() => { + this.db + .prepare( + `UPDATE job_items + SET status='failed',result_code=?,error_json=?,finished_at=?,updated_at=? + WHERE id=? AND status='running'`, + ) + .run( + resultCode, + JSON.stringify({ status: 502, code: resultCode, requestId }), + finished, + finished, + ids.item, + ); + this.db + .prepare( + "UPDATE job_attempts SET status='failed',finished_at=? WHERE id=? AND status='running'", + ) + .run(finished, ids.attempt); + this.db + .prepare( + "UPDATE jobs SET status='failed',finished_at=?,updated_at=? WHERE id=? AND status='running'", + ) + .run(finished, finished, ids.job); + })(); + } + reconcileInterruptedJobs(): number { const finished = this.clock().toISOString(); return this.db.transaction(() => { @@ -369,7 +442,7 @@ export class DeleteInstanceOperation { private job(id: string): Job { const row = this.db .prepare( - 'SELECT operation_id,status,root_job_id,retry_of_job_id,created_at FROM jobs WHERE id=?', + 'SELECT operation_id,status,root_job_id,retry_of_job_id,created_at,request_id FROM jobs WHERE id=?', ) .get(id) as { operation_id: string; @@ -377,13 +450,17 @@ export class DeleteInstanceOperation { root_job_id: string; retry_of_job_id: string | null; created_at: string; + request_id: string; }; const items = this.db - .prepare('SELECT id,instance_id,status FROM job_items WHERE job_id=? ORDER BY created_at,id') + .prepare( + 'SELECT id,instance_id,status,result_code FROM job_items WHERE job_id=? ORDER BY created_at,id', + ) .all(id) as Array<{ id: string; instance_id: string; status: 'succeeded' | 'failed' | 'unknown-result'; + result_code: string | null; }>; const attempts = this.db .prepare( @@ -401,7 +478,25 @@ export class DeleteInstanceOperation { status: row.status, ...(row.retry_of_job_id ? { retryOfJobId: row.retry_of_job_id } : {}), rootJobId: row.root_job_id, - items: items.map((item) => ({ id: item.id, targetId: item.instance_id, state: item.status })), + items: items.map((item) => ({ + id: item.id, + targetId: item.instance_id, + state: item.status, + // The console reads this code straight out of the execute response to explain why the + // node it just confirmed is still in the list. + ...(item.result_code === BINDING_REFUSAL_CODE + ? { + error: { + type: 'about:blank', + title: 'Job item failed', + status: 502, + detail: 'The job item did not complete successfully.', + code: BINDING_REFUSAL_CODE, + requestId: row.request_id, + }, + } + : {}), + })), attempts: attempts.map((attempt) => ({ id: attempt.id, state: attempt.status, diff --git a/apps/api/src/control-plane.test.ts b/apps/api/src/control-plane.test.ts index bf42e3a..6efa2ba 100644 --- a/apps/api/src/control-plane.test.ts +++ b/apps/api/src/control-plane.test.ts @@ -1216,4 +1216,84 @@ describe('buildControlPlaneApp', () => { expect(body.total).toBe(1); await app.close(); }); + + it('keeps a node that refuses to release its hub binding and drops one that is unreachable', async () => { + const db = new Database(':memory:'); + db.pragma('foreign_keys=ON'); + migrateDatabase(db); + dbs.push(db); + const unbinds: string[] = []; + const app = buildControlPlaneApp({ + db, + store: new Store(), + upstream: { + get: async (url) => { + if (url.includes('192.168.1.21')) throw new Error('ECONNREFUSED'); + return { status: 200, headers: {}, body: '' }; + }, + request: async ({ url, method }) => { + if (method === 'POST' && url.endsWith('/api/hub/unbind')) { + unbinds.push(url); + return { status: 409, headers: {}, body: '{"status":"error","message":"busy"}' }; + } + return { status: 200, headers: {}, body: '{"status":"success"}' }; + }, + postNetworkRegisterAuto: async () => ({ status: 200 }), + postServiceRestart: async () => ({ status: 200 }), + postBasebandRestart: async () => ({ status: 200 }), + postSystemReboot: async () => ({ status: 200 }), + }, + }); + async function remove(instanceId: string): Promise> { + const prepared = await app.inject({ + method: 'POST', + url: '/api/v1/operations/prepare', + payload: { + operationId: 'deleteInstance', + targets: [{ instanceId, revision: 1 }], + parameters: { parameterSchemaId: 'deleteInstance.parameters.v1', fields: [] }, + }, + }); + expect(prepared.statusCode).toBe(200); + const executed = await app.inject({ + method: 'DELETE', + url: `/api/v1/instances/${encodeURIComponent(instanceId)}`, + headers: { + 'if-match': '"rev-1"', + 'x-preparation-id': prepared.json().id as string, + 'x-confirmation-token': prepared.json().confirmationToken as string, + }, + }); + expect(executed.statusCode).toBe(202); + return executed.json() as Record; + } + + const created: string[] = []; + for (const [name, host] of [ + ['Refusing', '192.168.1.20'], + ['Unreachable', '192.168.1.21'], + ]) { + const response = await app.inject({ + method: 'POST', + url: '/api/v1/instances', + payload: { name, origin: `http://${host}:8080` }, + }); + expect(response.statusCode).toBe(201); + created.push(response.json().id as string); + } + const [refused, gone] = created; + + expect(await remove(refused!)).toMatchObject({ + status: 'failed', + items: [{ state: 'failed', error: { code: 'UNBIND_FAILED' } }], + }); + expect(await remove(gone!)).toMatchObject({ status: 'succeeded' }); + + const listed = await app.inject('/api/v1/instances'); + expect((listed.json().items as Array<{ id: string }>).map((item) => item.id)).toEqual([ + refused, + ]); + expect(unbinds).toEqual([`http://192.168.1.20:8080/api/hub/unbind`]); + await app.close(); + }); }); diff --git a/apps/api/src/control-plane.ts b/apps/api/src/control-plane.ts index a9d80ee..3aef71f 100644 --- a/apps/api/src/control-plane.ts +++ b/apps/api/src/control-plane.ts @@ -13,7 +13,12 @@ import { type ConnectionTransport, } from './application/connections/connection-probe.js'; import { ConnectionSettingsService } from './application/connections/connection-settings-service.js'; -import { FleetHeartbeatCoordinator } from './application/connections/fleet-heartbeat.js'; +import { + FLEET_STATE_CODES, + FLEET_STATE_EVENTS, + FLEET_STATE_NAMES, + FleetHeartbeatCoordinator, +} from './application/connections/fleet-heartbeat.js'; import { InstanceCredentialResolver } from './application/connections/instance-credential-resolver.js'; import { InstanceLoginService } from './application/connections/instance-login-service.js'; import { @@ -23,7 +28,10 @@ import { } from './application/connections/upstream-session-client.js'; import { InstanceService } from './application/instances/instance-service.js'; import { DeviceDiscoveryService } from './application/instances/device-discovery-service.js'; -import { DeleteInstanceOperation } from './application/operations/delete-instance-operation.js'; +import { + type BindingRelease, + DeleteInstanceOperation, +} from './application/operations/delete-instance-operation.js'; import type { SecretStore } from './infrastructure/secrets/secret-store.js'; import { registerInstanceRoutes } from './interface/http/instance-routes.js'; import { registerEventRoutes } from './interface/http/event-routes.js'; @@ -181,12 +189,6 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane connectionLogs, snapshotTtlMs: () => connectionSettings.snapshotTtlMs, }); - const heartbeat = new FleetHeartbeatCoordinator({ - instances, - probe: connections, - settings: connectionSettings, - ...(options.now ? { now: options.now } : {}), - }); const sessions = new InstanceSessionStore(); const discovery = options.now ? new DeviceDiscoveryService({ transport: options.upstream, instances, now: options.now }) @@ -238,6 +240,31 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane store: options.store, ...(options.now ? { now: options.now } : {}), }); + // The heartbeat owns the reachability loop, so it also owns telling the rule engine when a + // node went offline, came back, or lost its device session. + const heartbeat = new FleetHeartbeatCoordinator({ + instances, + probe: connections, + settings: connectionSettings, + ...(options.now ? { now: options.now } : {}), + onTransition: async (transitions) => { + for (const transition of transitions) { + const instance = await instances.get(transition.instanceId); + const label = instance?.name ?? transition.instanceId; + await centralNotifications.enqueueEvent('system', { + instanceId: transition.instanceId, + instanceTags: instance?.tags ?? [], + fields: { + title: `${label} ${FLEET_STATE_EVENTS[transition.to]}`, + status: FLEET_STATE_CODES[transition.to], + content: `节点 ${label} 从「${FLEET_STATE_NAMES[transition.from]}」变为「${ + FLEET_STATE_NAMES[transition.to] + }」（${FLEET_STATE_EVENTS[transition.to]}）。`, + }, + }); + } + }, + }); const hubMessages = new HubMessageService(options.db, { instances, messages, @@ -269,9 +296,34 @@ export function buildControlPlaneApp(options: ControlPlaneOptions): ControlPlane ...(options.now ? { now: options.now } : {}), }); smsOutbox.reconcileInterrupted(); + // Removal follows the hub rule the operator already knows: an online device is told to release + // its own binding first and a refusal keeps the record, while a device that cannot be reached + // at all is forgotten locally. The closure is late-bound because the device action service is + // built further down, next to the session store it needs. + const releaseBinding = async (instanceId: string, requestId: string): Promise => { + try { + await connections.test(instanceId); + } catch { + return { status: 'skipped' }; + } + const result = await deviceActions.execute( + instanceId, + 'hub.unbind', + {}, + { actor: 'loopback-control-plane', requestId, confirm: true }, + ); + // A device too old to expose the endpoint never had a binding to release. + if (result.status === 404 || result.status === 405) return { status: 'released' }; + return result.ok ? { status: 'released' } : { status: 'failed' }; + }; const deletion = options.now - ? new DeleteInstanceOperation({ db: options.db, instances, now: options.now }) - : new DeleteInstanceOperation({ db: options.db, instances }); + ? new DeleteInstanceOperation({ + db: options.db, + instances, + releaseBinding, + now: options.now, + }) + : new DeleteInstanceOperation({ db: options.db, instances, releaseBinding }); deletion.reconcileInterruptedJobs(); const listHubDevices = createHubDeviceReader(options.db, instances); const resolveOperationCookie = async ( diff --git a/apps/web/src/instances/instance-api-data-source.test.ts b/apps/web/src/instances/instance-api-data-source.test.ts index 751d5ec..221a4aa 100644 --- a/apps/web/src/instances/instance-api-data-source.test.ts +++ b/apps/web/src/instances/instance-api-data-source.test.ts @@ -5,6 +5,7 @@ import { createInstanceApiDataSource, passwordUpdate, tagsFromInput, + BindingRefusalError, } from './instance-api-data-source.js'; afterEach(() => vi.unstubAllGlobals()); @@ -86,30 +87,97 @@ describe('instance API data source', () => { ); }); - it('prepares and executes R3 deletion through the durable operation endpoint', async () => { + it('prepares R3 deletion and executes it on the owner route with the one-time token', async () => { const token = 'a'.repeat(32); const fetch = vi .fn() + .mockResolvedValueOnce(response(instance, 200, '"rev-3"')) .mockResolvedValueOnce(response({ id: 'prep-1', confirmationToken: token })) .mockResolvedValueOnce(response({ id: 'job-1' }, 202)); vi.stubGlobal('fetch', fetch); + const source = createInstanceApiDataSource(); - await createInstanceApiDataSource().delete('owner', 3); - expect(JSON.parse(fetch.mock.calls[0]?.[1]?.body as string)).toEqual({ + await source.get('owner/id'); + await source.delete('owner/id', 3); + expect(JSON.parse(fetch.mock.calls[1]?.[1]?.body as string)).toEqual({ operationId: 'deleteInstance', - targets: [{ instanceId: 'owner', revision: 3 }], + targets: [{ instanceId: 'owner/id', revision: 3 }], parameters: { parameterSchemaId: 'deleteInstance.parameters.v1', fields: [] }, }); - expect(fetch.mock.calls[1]?.[0]).toBe('/api/v1/operations/execute'); - expect(fetch.mock.calls[1]?.[1]).toMatchObject({ - method: 'POST', - body: JSON.stringify({ preparationId: 'prep-1', confirmationToken: token }), + expect(fetch.mock.calls[2]?.[0]).toBe('/api/v1/instances/owner%2Fid'); + expect(fetch.mock.calls[2]?.[1]).toMatchObject({ + method: 'DELETE', + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + 'If-Match': '"rev-3"', + 'X-Preparation-Id': 'prep-1', + 'X-Confirmation-Token': token, + }, }); - expect( - fetch.mock.calls.some( - ([url, init]) => init?.method === 'DELETE' || String(url).includes(token), - ), - ).toBe(false); + expect(fetch.mock.calls[2]?.[1]?.body).toBeUndefined(); + // The confirmation is a bearer secret: it belongs in a header, never in a URL or a body. + expect(fetch.mock.calls.every(([url]) => !String(url).includes(token))).toBe(true); + }); + + it('refetches the ETag when deletion is confirmed without a warm cache', async () => { + const token = 'c'.repeat(32); + const fetch = vi + .fn() + .mockResolvedValueOnce(response({ id: 'prep-1', confirmationToken: token })) + .mockResolvedValueOnce(response(instance, 200, '"rev-3"')) + .mockResolvedValueOnce(response({ id: 'job-1' }, 202)); + vi.stubGlobal('fetch', fetch); + + await createInstanceApiDataSource().delete('owner/id', 3); + + expect(fetch.mock.calls[1]?.[1]?.method).toBeUndefined(); + expect(fetch.mock.calls[2]?.[1]).toMatchObject({ + method: 'DELETE', + headers: expect.objectContaining({ 'If-Match': '"rev-3"' }), + }); + }); + + it('drops a deletion whose confirmation no longer matches the recorded revision', async () => { + const token = 'd'.repeat(32); + const fetch = vi + .fn() + .mockResolvedValueOnce(response({ id: 'prep-1', confirmationToken: token })) + .mockResolvedValueOnce(response({ ...instance, revision: 4 }, 200, '"rev-4"')); + vi.stubGlobal('fetch', fetch); + + await expect(createInstanceApiDataSource().delete('owner/id', 3)).rejects.toThrow(/dropped/i); + expect(fetch.mock.calls.some(([, init]) => init?.method === 'DELETE')).toBe(false); + }); + + it('reports a device that kept its binding instead of pretending the node is gone', async () => { + const token = 'b'.repeat(32); + const fetch = vi + .fn() + .mockResolvedValueOnce(response({ id: 'prep-1', confirmationToken: token })) + .mockResolvedValueOnce(response(instance, 200, '"rev-3"')) + .mockResolvedValueOnce( + response( + { + id: 'job-1', + status: 'failed', + items: [ + { + id: 'item-1', + targetId: 'owner/id', + state: 'failed', + error: { code: 'UNBIND_FAILED', status: 502 }, + }, + ], + }, + 202, + ), + ); + vi.stubGlobal('fetch', fetch); + + await expect(createInstanceApiDataSource().delete('owner/id', 3)).rejects.toBeInstanceOf( + BindingRefusalError, + ); }); it('rejects mismatched owners and does not render arbitrary response text as an error', async () => { diff --git a/apps/web/src/instances/instance-api-data-source.ts b/apps/web/src/instances/instance-api-data-source.ts index f51fc80..5c3082d 100644 --- a/apps/web/src/instances/instance-api-data-source.ts +++ b/apps/web/src/instances/instance-api-data-source.ts @@ -39,6 +39,18 @@ interface Preparation { const DELETE_SCHEMA = 'deleteInstance.parameters.v1'; const SAFE_FALLBACK = 'The instance operation could not be completed.'; +/** + * The device kept its central binding, so the control plane refused to forget it. This is the + * one deletion failure the operator can act on, and the row stays in the list, so the console + * shows the reason instead of the generic fallback. + */ +export class BindingRefusalError extends Error { + constructor() { + super('设备仍持有回连绑定，节点已保留。请先在设备面板处理绑定，或在节点离线后重试删除。'); + this.name = 'BindingRefusalError'; + } +} + function ownerPath(instanceId: string, suffix = ''): string { return `/api/v1/instances/${encodeURIComponent(instanceId)}${suffix}`; } @@ -109,6 +121,21 @@ export function createInstanceApiDataSource(): InstanceDataSource { etags.set(value.id, { revision: value.revision, value: strongEtag(result.response) }); return value; }; + /** + * Deletion is an If-Match route, so the console needs the strong ETag for the exact revision + * the operator confirmed. A warm cache is preferred; a cold or stale one costs one GET rather + * than letting the browser guess at a header the server is about to check. + */ + const etagFor = async (instanceId: string, revision: number): Promise => { + const cached = etags.get(instanceId); + if (cached !== undefined && cached.revision === revision) return cached.value; + const fresh = await instanceResponse(ownerPath(instanceId), instanceId); + const stored = etags.get(fresh.id); + if (stored === undefined) throw new Error('The server did not return a valid strong ETag.'); + if (fresh.revision !== revision) + throw new Error('The instance changed before deletion, so the confirmation was dropped.'); + return stored.value; + }; return { async get(instanceId) { return instanceResponse(ownerPath(instanceId), instanceId); @@ -157,13 +184,30 @@ export function createInstanceApiDataSource(): InstanceDataSource { ) throw new Error('The server returned an invalid deletion confirmation.'); const confirmation = prepared as unknown as Preparation; - await requestJson('/api/v1/operations/execute', { - method: 'POST', - body: JSON.stringify({ - preparationId: confirmation.id, - confirmationToken: confirmation.confirmationToken, - }), + // The generic execute endpoint only accepts the operations the safe transport can replay, + // and deletion is not one of them: it goes to the owner route with the one-time token in a + // header, where the token is consumed against the revision the preparation was bound to. + const match = await etagFor(instanceId, revision); + const executed = await requestJson(ownerPath(instanceId), { + method: 'DELETE', + headers: { + 'If-Match': match, + 'X-Preparation-Id': confirmation.id, + 'X-Confirmation-Token': confirmation.confirmationToken, + }, }); + // Deletion is accepted as a job, and the job is where a device that kept its binding + // reports back. The row survives a refresh in that case, so the console has to say why. + if (isRecord(executed) && Array.isArray(executed.items)) { + const refused = executed.items.some( + (item) => + isRecord(item) && + item.state === 'failed' && + isRecord(item.error) && + item.error.code === 'UNBIND_FAILED', + ); + if (refused) throw new BindingRefusalError(); + } etags.delete(instanceId); }, }; diff --git a/apps/web/src/instances/instance-crud.test.tsx b/apps/web/src/instances/instance-crud.test.tsx index c875724..2059ab4 100644 --- a/apps/web/src/instances/instance-crud.test.tsx +++ b/apps/web/src/instances/instance-crud.test.tsx @@ -4,6 +4,7 @@ import userEvent from '@testing-library/user-event'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { InstanceEditor, type InstanceDataSource, type ManagedInstance } from './instance-crud.js'; +import { BindingRefusalError } from './instance-api-data-source.js'; const owner: ManagedInstance = { id: 'owner', @@ -111,4 +112,20 @@ describe('Instance CRUD form', () => { await user.click(screen.getByRole('button', { name: '确认删除' })); expect(source.delete).toHaveBeenCalledWith('owner', 3); }); + + it('explains why a device that kept its binding is still listed', async () => { + const user = userEvent.setup(); + const source = dataSource({ + delete: vi.fn(async () => { + throw new BindingRefusalError(); + }), + }); + render(); + await screen.findByDisplayValue('Owner modem'); + await user.click(screen.getByRole('button', { name: '删除实例' })); + await user.type(screen.getByLabelText('输入 owner 以确认'), 'owner'); + await user.click(screen.getByRole('button', { name: '确认删除' })); + + expect((await screen.findByRole('alert')).textContent).toContain('设备仍持有回连绑定'); + }); }); diff --git a/apps/web/src/instances/instance-crud.tsx b/apps/web/src/instances/instance-crud.tsx index 848ebd5..10233e1 100644 --- a/apps/web/src/instances/instance-crud.tsx +++ b/apps/web/src/instances/instance-crud.tsx @@ -3,6 +3,7 @@ import { Button, Card, Title } from 'animal-island-ui'; import { DeviceDiscoveryPanel, type DeviceSelection } from './device-discovery-panel.js'; import { + BindingRefusalError, createInstanceApiDataSource, passwordUpdate, tagsFromInput, @@ -137,8 +138,10 @@ export function InstanceEditor({ setStatus('实例删除请求已接受。'); setError(undefined); setConfirming(false); - } catch { - setError(message()); + } catch (cause) { + // A device that kept its binding stays in the list, so the operator needs the real reason. + if (cause instanceof BindingRefusalError) setError(cause.message); + else setError(message()); } finally { setBusy(false); }