test(quality): add concurrency and real browser gates

2026-07-18 03:20:42 +08:00
parent 88d41a7a18
commit a884033352
5 changed files with 643 additions and 14 deletions
@@ -1,5 +1,9 @@
 import Database from 'better-sqlite3';
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
 import { afterEach, describe, expect, it } from 'vitest';
+import { openDatabase } from '../../infrastructure/database/database.js';
 import { migrateDatabase } from '../../infrastructure/database/migrations.js';
 import type { SecretStore } from '../../infrastructure/secrets/secret-store.js';
 import type { Instance } from '@multi-simadmin/contracts';
@@ -38,8 +42,11 @@ class MemorySecrets implements SecretStore {
  }
 }
 const dbs: Database.Database[] = [];
+const temporaryDirectories: string[] = [];
 afterEach(() => {
  for (const db of dbs.splice(0)) db.close();
+  for (const directory of temporaryDirectories.splice(0))
+    rmSync(directory, { recursive: true, force: true });
 });
 function fixture(ids = ['id-1', 'slot-1', 'ref-1', 'id-2', 'slot-2', 'ref-2']) {
  const db = new Database(':memory:');
@@ -56,6 +63,30 @@ function fixture(ids = ['id-1', 'slot-1', 'ref-1', 'id-2', 'slot-2', 'ref-2']) {
  });
  return { db, store, service };
 }
+function crossConnectionFixture() {
+  const directory = mkdtempSync(join(tmpdir(), 'instance-cas-'));
+  temporaryDirectories.push(directory);
+  const path = join(directory, 'state.sqlite');
+  const firstDb = openDatabase(path);
+  migrateDatabase(firstDb);
+  const secondDb = openDatabase(path);
+  dbs.push(firstDb, secondDb);
+  const store = new MemorySecrets();
+  const options = { store, now: () => new Date('2026-07-16T12:00:00.000Z') };
+  const firstIds = ['id-1', 'first-slot', 'first-reference'];
+  const secondIds = ['second-slot', 'second-reference'];
+  const first = new InstanceService({
+    ...options,
+    db: firstDb,
+    idFactory: () => firstIds.shift()!,
+  });
+  const second = new InstanceService({
+    ...options,
+    db: secondDb,
+    idFactory: () => secondIds.shift()!,
+  });
+  return { firstDb, secondDb, store, first, second };
+}
 const basic = { name: ' Alpha ', origin: 'http://192.168.1.10:8080/', tags: [' z ', 'a', 'a'] };
 const code = async (promise: Promise<unknown>) => {
  try {
@@ -222,6 +253,23 @@ describe('InstanceService', () => {
    });
  });

+  it('rejects stale update and delete CAS across managed connections', async () => {
+    const { firstDb, first, second } = crossConnectionFixture();
+    await first.create(basic);
+    await second.update('id-1', 1, { name: 'Committed elsewhere' });
+
+    await expect(first.update('id-1', 1, { name: 'Stale update' })).rejects.toMatchObject({
+      code: 'REVISION_CONFLICT',
+    });
+    await expect(first.delete('id-1', 1)).rejects.toMatchObject({ code: 'REVISION_CONFLICT' });
+    expect(
+      firstDb.prepare('SELECT name,config_revision FROM instances WHERE id=?').get('id-1'),
+    ).toEqual({
+      name: 'Committed elsewhere',
+      config_revision: 2,
+    });
+  });
+
  it('preserves, replaces and clears credentials without exposing them', async () => {
    const { db, store, service } = fixture();
    await service.create({ ...basic, password: { action: 'set', password: 'first-secret' } });
@@ -6,6 +6,7 @@ import { UpstreamError } from '../../infrastructure/transport/upstream-error.js'
 import {
  StatusSnapshotError,
  StatusSnapshotService,
+  type HealthSnapshotInstance,
  type HealthTransportResponse,
 } from './status-snapshot-service.js';

@@ -17,7 +18,12 @@ afterEach(() => {
 const instant = (value: string) => () => new Date(value);

 function fixture(
-  options: { response?: HealthTransportResponse; failure?: unknown; now?: () => Date } = {},
+  options: {
+    response?: HealthTransportResponse;
+    failure?: unknown;
+    now?: () => Date;
+    idFactory?: () => string;
+  } = {},
 ) {
  const db = new Database(':memory:');
  db.pragma('foreign_keys=ON');
@@ -27,11 +33,10 @@ function fixture(
  db.prepare(
    'INSERT INTO instances (id,name,base_url,auth_mode,enabled,config_revision,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?)',
  ).run('instance-1', 'LAN', 'http://192.168.1.20:3000', 'none', 1, 1, created, created);
-  const instances = {
-    get: vi.fn(async (id: string) =>
-      id === 'instance-1' ? { id, origin: 'http://192.168.1.20:3000' } : undefined,
-    ),
-  };
+  const owners = new Map<string, HealthSnapshotInstance>([
+    ['instance-1', { id: 'instance-1', origin: 'http://192.168.1.20:3000', revision: 1 }],
+  ]);
+  const instances = { get: vi.fn(async (id: string) => owners.get(id)) };
  const transport = {
    get: vi.fn(async (url: string) => {
      expect(url).toBe('http://192.168.1.20:3000/api/health');
@@ -51,9 +56,9 @@ function fixture(
    ttlMs: 60_000,
    maxStaleMs: 300_000,
    now: options.now ?? instant('2026-07-17T12:00:00.000Z'),
-    idFactory: () => 'snapshot-id',
+    idFactory: options.idFactory ?? (() => 'snapshot-id'),
  });
-  return { db, instances, transport, service };
+  return { db, instances, owners, transport, service };
 }

 const row = (db: Database.Database) =>
@@ -257,6 +262,129 @@ describe('StatusSnapshotService health snapshots', () => {
    });
  });

+  it('lets the latest overlapping probe replace an equal-time payload', async () => {
+    const first = fixture();
+    const secondTransport = { get: vi.fn<() => Promise<HealthTransportResponse>>() };
+    const second = new StatusSnapshotService({
+      db: first.db,
+      instances: first.instances,
+      transport: secondTransport,
+      ttlMs: 60_000,
+      maxStaleMs: 300_000,
+      now: instant('2026-07-17T12:00:00.000Z'),
+      idFactory: () => 'second-snapshot-id',
+    });
+    let resolveOlder!: (value: HealthTransportResponse) => void;
+    let resolveLatest!: (value: HealthTransportResponse) => void;
+    first.transport.get.mockImplementationOnce(
+      async () => new Promise((resolve) => (resolveOlder = resolve)),
+    );
+    secondTransport.get.mockImplementationOnce(
+      async () => new Promise((resolve) => (resolveLatest = resolve)),
+    );
+
+    const older = first.service.refreshHealth('instance-1');
+    await vi.waitFor(() => expect(first.transport.get).toHaveBeenCalledTimes(1));
+    const latest = second.refreshHealth('instance-1');
+    await vi.waitFor(() => expect(secondTransport.get).toHaveBeenCalledTimes(1));
+    resolveOlder({ status: 200, body: '{"status":"older"}' });
+    await older;
+    expect(payload(first.db)).toMatchObject({ data: { status: 'older' } });
+    resolveLatest({ status: 200, body: '{"status":"latest"}' });
+    await latest;
+
+    expect(row(first.db)).toMatchObject({ observed_at: '2026-07-17T12:00:00.000Z' });
+    expect(payload(first.db)).toMatchObject({ data: { status: 'latest' } });
+  });
+
+  it.each([
+    [{ id: 'instance-1', origin: 'http://192.168.1.20:3000', revision: 0 }],
+    [
+      {
+        id: 'instance-1',
+        origin: 'http://192.168.1.20:3000',
+        revision: Number.MAX_SAFE_INTEGER + 1,
+      },
+    ],
+    [{ id: 'different-id', origin: 'http://192.168.1.20:3000', revision: 1 }],
+    [{ id: 'instance-1', origin: 'http://192.168.1.20:3000/', revision: 1 }],
+  ] as const)('fails closed for an invalid owner token %#', async (invalidOwner) => {
+    const { db, owners, service, transport } = fixture();
+    owners.set('instance-1', invalidOwner);
+
+    await expect(service.refreshHealth('instance-1')).rejects.toEqual(
+      expect.objectContaining<Partial<StatusSnapshotError>>({ code: 'INSTANCE_NOT_FOUND' }),
+    );
+    expect(transport.get).not.toHaveBeenCalled();
+    expect(row(db)).toBeUndefined();
+  });
+
+  it('does not persist a probe after its owner origin and revision change', async () => {
+    const { db, instances, owners, service, transport } = fixture();
+    let resolve!: (value: HealthTransportResponse) => void;
+    transport.get.mockImplementationOnce(
+      async () =>
+        new Promise((done) => {
+          resolve = done;
+        }),
+    );
+
+    const running = service.refreshHealth('instance-1');
+    await vi.waitFor(() => expect(transport.get).toHaveBeenCalledTimes(1));
+    owners.set('instance-1', {
+      id: 'instance-1',
+      origin: 'http://192.168.1.21:3000',
+      revision: 2,
+    });
+    resolve({ status: 200, body: '{"status":"old-owner"}' });
+
+    await expect(running).resolves.toMatchObject({
+      state: 'fresh',
+      payload: { status: 'old-owner' },
+    });
+    expect(instances.get).toHaveBeenCalledTimes(2);
+    expect(row(db)).toBeUndefined();
+  });
+
+  it('does not persist a probe after its owner is deleted', async () => {
+    const { db, instances, owners, service, transport } = fixture();
+    let resolve!: (value: HealthTransportResponse) => void;
+    transport.get.mockImplementationOnce(
+      async () =>
+        new Promise((done) => {
+          resolve = done;
+        }),
+    );
+
+    const running = service.refreshHealth('instance-1');
+    await vi.waitFor(() => expect(transport.get).toHaveBeenCalledTimes(1));
+    owners.delete('instance-1');
+    resolve({ status: 200, body: '{"status":"deleted-owner"}' });
+
+    await expect(running).resolves.toMatchObject({ payload: { status: 'deleted-owner' } });
+    expect(instances.get).toHaveBeenCalledTimes(2);
+    expect(row(db)).toBeUndefined();
+  });
+
+  it('atomically fences persistence when the database owner changes after the reread', async () => {
+    const holder: { db?: Database.Database } = {};
+    const result = fixture({
+      idFactory: () => {
+        holder
+          .db!.prepare('UPDATE instances SET base_url=?,config_revision=? WHERE id=?')
+          .run('http://192.168.1.21:3000', 2, 'instance-1');
+        return 'snapshot-id';
+      },
+    });
+    holder.db = result.db;
+
+    await expect(result.service.refreshHealth('instance-1')).resolves.toMatchObject({
+      payload: { status: 'ok' },
+    });
+    expect(result.instances.get).toHaveBeenCalledTimes(2);
+    expect(row(result.db)).toBeUndefined();
+  });
+
  it('does no network/write for missing instances and no write after in-flight deletion', async () => {
    const first = fixture();
    await expect(first.service.refreshHealth('missing')).rejects.toEqual(
@@ -274,6 +402,7 @@ describe('StatusSnapshotService health snapshots', () => {
    );
    const running = first.service.refreshHealth('instance-1');
    await vi.waitFor(() => expect(first.transport.get).toHaveBeenCalledTimes(1));
+    first.owners.delete('instance-1');
    first.db.prepare('DELETE FROM instances WHERE id=?').run('instance-1');
    resolve({ status: 200, body: '{"status":"ok"}' });
    await running;
@@ -19,6 +19,7 @@ export type HealthSnapshotErrorCode =
 export interface HealthSnapshotInstance {
  readonly id: string;
  readonly origin: string;
+  readonly revision: number;
 }

 export interface HealthEnvelope {
@@ -157,7 +158,32 @@ function transportErrorCode(error: unknown): SnapshotErrorCode {
  return 'UPSTREAM_UNAVAILABLE';
 }

-const healthUrl = (origin: string): string => `${origin.replace(/\/$/, '')}/api/health`;
+function isValidOwner(
+  instanceId: string,
+  owner: HealthSnapshotInstance | undefined,
+): owner is HealthSnapshotInstance {
+  if (
+    !owner ||
+    owner.id !== instanceId ||
+    !Number.isSafeInteger(owner.revision) ||
+    owner.revision <= 0
+  ) {
+    return false;
+  }
+  try {
+    const parsed = new URL(owner.origin);
+    return (
+      (parsed.protocol === 'http:' || parsed.protocol === 'https:') &&
+      parsed.origin === owner.origin &&
+      parsed.username === '' &&
+      parsed.password === ''
+    );
+  } catch {
+    return false;
+  }
+}
+
+const healthUrl = (origin: string): string => `${origin}/api/health`;

 export class StatusSnapshotService {
  readonly #db: Database.Database;
@@ -188,7 +214,7 @@ export class StatusSnapshotService {

  async refreshHealth(instanceId: string): Promise<HealthSnapshot> {
    const instance = await this.#instances.get(instanceId);
-    if (!instance) throw new StatusSnapshotError('INSTANCE_NOT_FOUND');
+    if (!isValidOwner(instanceId, instance)) throw new StatusSnapshotError('INSTANCE_NOT_FOUND');

    const generation = ++this.#nextGeneration;
    this.#latestGeneration.set(instanceId, generation);
@@ -201,6 +227,12 @@ export class StatusSnapshotService {
    } catch (error) {
      transportError = error;
    }
+    let currentOwner: HealthSnapshotInstance | undefined;
+    try {
+      currentOwner = await this.#instances.get(instanceId);
+    } catch {
+      // Owner lookup failures fence persistence, but do not discard the computed probe result.
+    }
    const completed = this.#now();
    const observedAt = completed.toISOString();
    const previous = this.#previous(instanceId);
@@ -216,7 +248,13 @@ export class StatusSnapshotService {
    const { snapshot } = classified;

    if (this.#latestGeneration.get(instanceId) === generation) {
-      this.#persist(snapshot, classified.persisted);
+      if (
+        isValidOwner(instanceId, currentOwner) &&
+        currentOwner.origin === instance.origin &&
+        currentOwner.revision === instance.revision
+      ) {
+        this.#persist(instance, snapshot, classified.persisted);
+      }
      this.#latestGeneration.delete(instanceId);
    }
    return Object.freeze({ ...snapshot, payload: Object.freeze({ ...snapshot.payload }) });
@@ -354,18 +392,23 @@ export class StatusSnapshotService {
    };
  }

-  #persist(snapshot: HealthSnapshot, persisted: PersistedHealthEnvelope): void {
+  #persist(
+    owner: HealthSnapshotInstance,
+    snapshot: HealthSnapshot,
+    persisted: PersistedHealthEnvelope,
+  ): void {
    this.#db
      .prepare(
        `INSERT INTO status_snapshots
          (id,instance_id,category,state,payload_json,observed_at,expires_at,created_at)
-         SELECT ?,id,'health',?,?,?,?,? FROM instances WHERE id=?
+         SELECT ?,id,'health',?,?,?,?,? FROM instances
+         WHERE id=? AND base_url=? AND config_revision=?
         ON CONFLICT(instance_id,category) DO UPDATE SET
          state=excluded.state,
          payload_json=excluded.payload_json,
          observed_at=excluded.observed_at,
          expires_at=excluded.expires_at
-         WHERE excluded.observed_at > status_snapshots.observed_at`,
+         WHERE excluded.observed_at >= status_snapshots.observed_at`,
      )
      .run(
        this.#id(),
@@ -375,6 +418,8 @@ export class StatusSnapshotService {
        snapshot.expiresAt,
        snapshot.observedAt,
        snapshot.instanceId,
+        owner.origin,
+        owner.revision,
      );
  }
 }