Commit d1c30148 by luoqi

Merge independent Meyer clinic progress and shared polling quotas into test

parents 051fcab8 120db745
Pipeline #3843 failed in 0 seconds
...@@ -4,10 +4,11 @@ import { runMeyerTick } from '../integrations/meyer/scheduler'; ...@@ -4,10 +4,11 @@ import { runMeyerTick } from '../integrations/meyer/scheduler';
import { readFile } from 'node:fs/promises'; import { readFile } from 'node:fs/promises';
import { resolve } from 'node:path'; import { resolve } from 'node:path';
import { NestFactory } from '@nestjs/core'; import { NestFactory } from '@nestjs/core';
import { loadMeyerScopes } from '../integrations/meyer/org-mapping'; import { loadMeyerScopes, loadMeyerScopeResults } from '../integrations/meyer/org-mapping';
import { ConfigSchema, InputConfigSchema, hash, imageKey } from '../integrations/meyer/contracts'; import { ConfigSchema, InputConfigSchema, hash, imageKey } from '../integrations/meyer/contracts';
import { MeyerClient } from '../integrations/meyer/client'; import { MeyerClient } from '../integrations/meyer/client';
import { collectMeyer } from '../integrations/meyer/collector'; import { collectMeyer } from '../integrations/meyer/collector';
import { prepareClinicStates } from '../integrations/meyer/clinic-state';
import { backfillMeyerFileRefs } from '../integrations/meyer/backfill-file-refs'; import { backfillMeyerFileRefs } from '../integrations/meyer/backfill-file-refs';
import { PrismaClient } from '@prisma/client'; import { PrismaClient } from '@prisma/client';
import { loadDiagnosisCodes } from '../integrations/meyer/mapper'; import { loadDiagnosisCodes } from '../integrations/meyer/mapper';
...@@ -34,7 +35,7 @@ export function parseMeyerArgs(argv: string[]) { ...@@ -34,7 +35,7 @@ export function parseMeyerArgs(argv: string[]) {
if (args.tick && !args.apply) throw new Error('--tick/--watch requires --apply; preview cannot advance cursor'); if (args.tick && !args.apply) throw new Error('--tick/--watch requires --apply; preview cannot advance cursor');
return args; return args;
} }
export async function runMeyerCli(argv: string[]): Promise<number> { export async function runMeyerCli(argv: string[], isolateWatchFailures = false): Promise<number> {
const args = parseMeyerArgs(argv); const args = parseMeyerArgs(argv);
if (args.help) { if (args.help) {
console.log('meyer-imaging --config=<json> --state-dir=<private-directory> [--apply] [--tick | --watch] [--image-files=<imageId>] [--backfill-file-refs]\n' + console.log('meyer-imaging --config=<json> --state-dir=<private-directory> [--apply] [--tick | --watch] [--image-files=<imageId>] [--backfill-file-refs]\n' +
...@@ -46,15 +47,18 @@ export async function runMeyerCli(argv: string[]): Promise<number> { ...@@ -46,15 +47,18 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const parsed = InputConfigSchema.safeParse(JSON.parse(await readFile(resolve(args.config), 'utf8'))); const parsed = InputConfigSchema.safeParse(JSON.parse(await readFile(resolve(args.config), 'utf8')));
if (!parsed.success) throw new Error('invalid Meyer config; check required fields, scope, dates and HTTPS base URL'); if (!parsed.success) throw new Error('invalid Meyer config; check required fields, scope, dates and HTTPS base URL');
const input = parsed.data; const input = parsed.data;
// watch 的每轮重新读取组织数据;映射变化会被既有 state identity 拦截。 // 每轮核对组织映射;单诊所异常隔离,已登记诊所的归属变化仍由私有状态拒绝。
const { clinicIds, ...fields } = input; const { clinicIds, ...fields } = input;
const config = ConfigSchema.parse({ ...fields, scopes: await loadMeyerScopes(clinicIds) }); const resolved = args.tick ? await loadMeyerScopeResults(clinicIds)
: { scopes: await loadMeyerScopes(clinicIds), failures: [] };
if (!resolved.scopes.length) throw new Error('Meyer clinic mapping missing, conflicting or invalid');
const config = ConfigSchema.parse({ ...fields, scopes: resolved.scopes });
if (args.watch) { if (args.watch) {
const controller = new AbortController(); const controller = new AbortController();
const stop = () => controller.abort(); const stop = () => controller.abort();
process.once('SIGINT', stop); process.once('SIGTERM', stop); process.once('SIGINT', stop); process.once('SIGTERM', stop);
try { try {
await watchMeyer(config, () => withRoundDeadline(() => runMeyerCli([...argv.filter(a => a !== '--watch'), '--tick']), config.roundTimeoutMs), controller.signal); await watchMeyer(config, () => withRoundDeadline(() => runMeyerCli([...argv.filter(a => a !== '--watch'), '--tick'], true), config.roundTimeoutMs), controller.signal);
return 0; return 0;
} finally { process.removeListener('SIGINT', stop); process.removeListener('SIGTERM', stop); } } finally { process.removeListener('SIGINT', stop); process.removeListener('SIGTERM', stop); }
} }
...@@ -63,16 +67,22 @@ export async function runMeyerCli(argv: string[]): Promise<number> { ...@@ -63,16 +67,22 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
const client = new MeyerClient(config, { appKey, secret: process.env.MEYER_APP_SECRET ?? '' }); const client = new MeyerClient(config, { appKey, secret: process.env.MEYER_APP_SECRET ?? '' });
const codes = loadDiagnosisCodes(resolve(__dirname, '../../data/_shared/dict/diagnosis-kcode.yaml')); const codes = loadDiagnosisCodes(resolve(__dirname, '../../data/_shared/dict/diagnosis-kcode.yaml'));
const store = new MeyerStore(resolve(args.state)); const store = new MeyerStore(resolve(args.state));
const release = await store.acquire({ baseUrl: config.baseUrl.replace(/\/$/, ''), appKey, const identity = { baseUrl: config.baseUrl.replace(/\/$/, ''), appKey,
namespace: config.namespace, host: config.host, timezone: config.timezone, namespace: config.namespace, host: config.host, timezone: config.timezone };
scopes: [...config.scopes].sort((a, b) => a.clinicId.localeCompare(b.clinicId)) }); const dbTarget = new URL(process.env.DATABASE_URL!);
const target = hash([dbTarget.hostname, dbTarget.port, dbTarget.pathname, dbTarget.searchParams.get('schema') ?? 'public']);
const validateTarget = async () => {
const previous = await store.read<{ target: string }>('database');
if (previous && previous.target !== target) throw new Error('state directory belongs to another PAC database');
};
const release = await store.acquire({ version: 2, ...identity }, [{ ...identity,
scopes: [...config.scopes].sort((a, b) => a.clinicId.localeCompare(b.clinicId)) }],
async () => { await validateTarget(); await prepareClinicStates(store, config); });
let app: Awaited<ReturnType<typeof NestFactory.createApplicationContext>> | undefined; let app: Awaited<ReturnType<typeof NestFactory.createApplicationContext>> | undefined;
try { try {
if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL required for organization mapping'); if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL required for organization mapping');
const db = new URL(process.env.DATABASE_URL); await validateTarget();
const target = hash([db.hostname, db.port, db.pathname, db.searchParams.get('schema') ?? 'public']);
const previous = await store.read<{ target: string }>('database'); const previous = await store.read<{ target: string }>('database');
if (previous && previous.target !== target) throw new Error('state directory belongs to another PAC database');
if (!previous) await store.write('database', { target }); if (!previous) await store.write('database', { target });
if (args.backfillFileRefs) { if (args.backfillFileRefs) {
const db = new PrismaClient(); const db = new PrismaClient();
...@@ -110,12 +120,17 @@ export async function runMeyerCli(argv: string[]): Promise<number> { ...@@ -110,12 +120,17 @@ export async function runMeyerCli(argv: string[]): Promise<number> {
} }
if (args.tick) { if (args.tick) {
const status = await runMeyerTick(config, c => client.forWindow(c), store, codes, sink!); const status = await runMeyerTick(config, c => client.forWindow(c), store, codes, sink!);
console.log(JSON.stringify({ mode: 'tick', pending: status.pending, autoPending: status.autoPending, await store.write('scheduler-status', { ...status, scopeFailures: resolved.failures });
console.log(JSON.stringify({ mode: 'tick', clinicFailures: status.clinicFailures,
mappingFailures: resolved.failures.length, sourceRequests: status.sourceRequests, clinicsVisited: status.clinics.length, pending: status.pending, autoPending: status.autoPending,
manualReview: status.manualReview, retiredUnsupported: status.retiredUnsupported, retryAttempted: status.retryAttempted, manualReview: status.manualReview, retiredUnsupported: status.retiredUnsupported, retryAttempted: status.retryAttempted,
retryFailed: status.retryFailed, scanned: status.scan?.examined ?? 0, retryFailed: status.retryFailed, scanned: status.scan?.examined ?? 0,
scanComplete: status.scan?.complete ?? false, scanFailed: status.scan?.failed ?? 0, scanComplete: status.scan?.complete ?? false, scanFailed: status.scan?.failed ?? 0,
imagesApplied: status.scan?.imagesApplied ?? 0, analysisFailed: status.scan?.analysisFailed ?? 0 })); imagesApplied: status.scan?.imagesApplied ?? 0, analysisFailed: status.scan?.analysisFailed ?? 0 }));
return status.retryFailed || status.scan?.failed || status.scan?.analysisFailed ? 1 : status.pending ? 2 : 0; const failed = resolved.failures.length || status.clinicFailures || status.queueErrors || status.retryFailed ||
status.scan?.failed || status.scan?.analysisFailed;
// 诊所故障已有独立退避;watch 不再因此降低正常诊所的调度频率。共享故障仍抛出。
return failed && !isolateWatchFailures ? 1 : status.pending ? 2 : 0;
} }
const result = await collectMeyer(config, client, store, codes, sink); const result = await collectMeyer(config, client, store, codes, sink);
// 控制台只输出数量;原文、患者标识和逐条错误留在私有状态目录。 // 控制台只输出数量;原文、患者标识和逐条错误留在私有状态目录。
......
import { cp } from 'node:fs/promises';
import { join } from 'node:path';
import { randomUUID } from 'node:crypto';
import { hash, sourceTime, type MeyerConfig, type MeyerScope } from './contracts';
import type { ScanProgress } from './scan';
import { MeyerStore } from './store';
interface Layout { version: 2; legacyScopes: MeyerScope[]; migrated: boolean }
export const clinicStateStore = (store: MeyerStore, scope: MeyerScope) =>
new MeyerStore(join(store.directory, 'clinics', hash(scope.clinicId)));
/** 迁移只复制调度断点;根目录原文、快照、影像键和已有重试次数保持不动。 */
export async function prepareClinicStates(store: MeyerStore, config: MeyerConfig) {
let layout = await store.read<Layout>('scheduler-layout');
if (!layout) {
layout = { version: 2, legacyScopes: config.scopes, migrated: false };
await store.write('scheduler-layout', layout);
}
if (layout.version !== 2) throw new Error('unsupported Meyer scheduler layout');
if (layout.migrated) return;
const cursor = await store.read<{ through: string }>('cursor');
const progress = await store.read<ScanProgress>('scan-progress');
if (progress && (!Number.isInteger(progress.scopeIndex) || progress.scopeIndex < 0 ||
progress.scopeIndex > layout.legacyScopes.length || !/^[a-f0-9-]{36}$/.test(progress.generation))) {
throw new Error('invalid legacy scan checkpoint; migration stopped');
}
for (const [index, scope] of layout.legacyScopes.entries()) {
const target = clinicStateStore(store, scope);
await target.write('scope', scope);
if (cursor) await target.write('cursor', cursor);
if (progress) {
if (index < progress.scopeIndex) await target.write('cursor', { through:
cursor && sourceTime(cursor.through) >= sourceTime(progress.end) ? cursor.through : progress.end });
else if (index === progress.scopeIndex) {
// ledger 内的旧 scopeIndex 改挂到单诊所的 0;缺失的必要 ledger 继续由扫描器拒绝。
try { await cp(join(store.directory, 'scan-ledger', progress.generation, String(index)),
join(target.directory, 'scan-ledger', progress.generation, '0'), { recursive: true }); }
catch (e) { if ((e as NodeJS.ErrnoException).code !== 'ENOENT') throw e; }
await target.write('scan-progress', { ...progress, scopeIndex: 0 });
} else await target.write('scan-progress', { version: 1, generation: randomUUID(),
start: progress.start, end: progress.end, scopeIndex: 0, page: 1, row: 0, pageSize: progress.pageSize });
}
}
// 标记完成后才允许任何诊所运行;迁移中断时可按冻结的 legacyScopes 幂等复制。
await store.write('scheduler-layout', { ...layout, migrated: true });
}
...@@ -115,6 +115,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto ...@@ -115,6 +115,10 @@ export async function collectMeyer(config: MeyerConfig, source: MeyerSource, sto
await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image }); await store.write(`deferred-${key}`, { reason: 'unbound patient', observedAt, image });
result.deferred++; continue; result.deferred++; continue;
} }
if (source.canRequestAnalysis?.() === false) {
retry.reason = 'analysis_pending';
result.deferred++; continue;
}
retry.analysisAttempts!++; retry.analysisAttempts!++;
retry.reason = 'analysis_pending'; retry.reason = 'analysis_pending';
retry.nextRetryAt = nextAnalysisRetry(config, image, now); retry.nextRetryAt = nextAnalysisRetry(config, image, now);
......
...@@ -17,5 +17,7 @@ ...@@ -17,5 +17,7 @@
"errorBackoffMaxMs": 3600000, "errorBackoffMaxMs": 3600000,
"clinicIds": [ "clinicIds": [
"replace-clinic-id" "replace-clinic-id"
] ],
"maxClinicsPerTick": 5,
"maxSourceRequestsPerTick": 100
} }
...@@ -38,6 +38,8 @@ const ConfigFields = z.object({ ...@@ -38,6 +38,8 @@ const ConfigFields = z.object({
roundTimeoutMs: z.number().int().min(1000).max(86400000).default(600000), roundTimeoutMs: z.number().int().min(1000).max(86400000).default(600000),
pollIntervalMs: z.number().int().min(1000).max(86400000).default(900000), pollIntervalMs: z.number().int().min(1000).max(86400000).default(900000),
maxNewImagesPerTick: z.number().int().min(1).max(10000).default(100), maxNewImagesPerTick: z.number().int().min(1).max(10000).default(100),
maxClinicsPerTick: z.number().int().min(1).max(100).default(5),
maxSourceRequestsPerTick: z.number().int().min(2).max(10000).default(100),
maxRetriesPerTick: z.number().int().min(0).max(1000).default(10), maxRetriesPerTick: z.number().int().min(0).max(1000).default(10),
analysisSlowAfterMs: z.number().int().min(1000).max(86400000).default(900000), analysisSlowAfterMs: z.number().int().min(1000).max(86400000).default(900000),
analysisSlowRetryMs: z.number().int().min(1000).max(86400000).default(3600000), analysisSlowRetryMs: z.number().int().min(1000).max(86400000).default(3600000),
...@@ -96,6 +98,8 @@ export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> } ...@@ -96,6 +98,8 @@ export interface MeyerSink { apply(bundle: MeyerBundle): Promise<void> }
export interface MeyerSource { export interface MeyerSource {
list(scope: MeyerScope, page: number): Promise<{ total: number; page: number; size: number; images: unknown[] }>; list(scope: MeyerScope, page: number): Promise<{ total: number; page: number; size: number; images: unknown[] }>;
symptom(image: MeyerImage): Promise<unknown>; symptom(image: MeyerImage): Promise<unknown>;
/** 调度预算不足时延后分析,不能虚增实际分析请求次数。 */
canRequestAnalysis?(): boolean;
} }
export function hash(value: unknown): string { export function hash(value: unknown): string {
const stable = (v: unknown): unknown => Array.isArray(v) ? v.map(stable) : const stable = (v: unknown): unknown => Array.isArray(v) ? v.map(stable) :
......
...@@ -44,3 +44,20 @@ export async function loadMeyerScopes(clinicIds: string[]): Promise<MeyerScope[] ...@@ -44,3 +44,20 @@ export async function loadMeyerScopes(clinicIds: string[]): Promise<MeyerScope[]
throw error; throw error;
} finally { await db.$disconnect(); } } finally { await db.$disconnect(); }
} }
/** 定时采集隔离单诊所映射异常;数据库等共享依赖故障仍终止本轮。 */
export async function loadMeyerScopeResults(clinicIds: string[]) {
if (!process.env.DATABASE_URL) throw new Error('DATABASE_URL required for automatic clinic mapping');
const db = new PrismaClient();
const scopes: MeyerScope[] = [], failures: Array<{ clinicId: string; reason: string }> = [];
try {
for (const clinicId of clinicIds) {
try { scopes.push(...await resolveMeyerScopes(db, [clinicId])); }
catch (e) {
if (!(e instanceof MappingError)) throw new Error('PAC organization lookup failed');
failures.push({ clinicId, reason: 'organization mapping missing, conflicting or invalid' });
}
}
return { scopes, failures };
} finally { await db.$disconnect(); }
}
...@@ -3,6 +3,7 @@ import { join } from 'node:path'; ...@@ -3,6 +3,7 @@ import { join } from 'node:path';
import { collectMeyer, type CollectResult } from './collector'; import { collectMeyer, type CollectResult } from './collector';
import { hash, type MeyerConfig, type MeyerScope, type MeyerSource, type MeyerSink } from './contracts'; import { hash, type MeyerConfig, type MeyerScope, type MeyerSource, type MeyerSink } from './contracts';
import { MeyerStore } from './store'; import { MeyerStore } from './store';
import { MeyerBudgetExhausted } from './tick-budget';
type Page = Awaited<ReturnType<MeyerSource['list']>>; type Page = Awaited<ReturnType<MeyerSource['list']>>;
export interface ScanProgress { export interface ScanProgress {
...@@ -17,7 +18,7 @@ function checkPage(batch: Page, page: number, size: number) { ...@@ -17,7 +18,7 @@ function checkPage(batch: Page, page: number, size: number) {
} }
/** 已读取页先落盘;逐行成功/已持久化待处理后才移动页内位置。 */ /** 已读取页先落盘;逐行成功/已持久化待处理后才移动页内位置。 */
export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource, export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig) => MeyerSource,
store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string, now = Date.now()) { store: MeyerStore, codes: Record<string, string>, sink: MeyerSink, start: string, end: string, now = Date.now(), evidenceStore = store) {
let progress = await store.read<ScanProgress>('scan-progress'); let progress = await store.read<ScanProgress>('scan-progress');
if (!progress) { if (!progress) {
progress = { version: 1, generation: randomUUID(), start, end, scopeIndex: 0, page: 1, row: 0, pageSize: config.pageSize }; progress = { version: 1, generation: randomUUID(), start, end, scopeIndex: 0, page: 1, row: 0, pageSize: config.pageSize };
...@@ -97,8 +98,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig ...@@ -97,8 +98,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
seen.add(ids(image)); seen.add(ids(image));
const one = await collectMeyer({ ...window, scopes: [scope] }, { const one = await collectMeyer({ ...window, scopes: [scope] }, {
list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }), list: async () => ({ total: 1, page: 1, size: window.pageSize, images: [image] }),
canRequestAnalysis: () => source.canRequestAnalysis?.() ?? true,
symptom: im => source.symptom(im), symptom: im => source.symptom(im),
}, store, codes, sink, true, now); }, evidenceStore, codes, sink, true, now);
for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed', 'imagesApplied', 'analysisFailed'] as const) result[key] += one[key]; for (const key of ['seen', 'collected', 'applied', 'alreadyApplied', 'deferred', 'failed', 'imagesApplied', 'analysisFailed'] as const) result[key] += one[key];
result.errors.push(...one.errors); result.errors.push(...one.errors);
if (one.failed) return result; // 不跳过未知失败;重试从本行恢复。 if (one.failed) return result; // 不跳过未知失败;重试从本行恢复。
...@@ -116,7 +118,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig ...@@ -116,7 +118,9 @@ export async function scanWindow(config: MeyerConfig, sourceFor: (c: MeyerConfig
} }
result.complete = progress.scopeIndex === config.scopes.length; result.complete = progress.scopeIndex === config.scopes.length;
} catch (e) { } catch (e) {
if (!(e instanceof MeyerBudgetExhausted)) {
result.failed++; result.errors.push({ key: 'page-scan', reason: e instanceof Error ? e.message : 'scan failed' }); result.failed++; result.errors.push({ key: 'page-scan', reason: e instanceof Error ? e.message : 'scan failed' });
} }
}
return result; return result;
} }
...@@ -7,7 +7,7 @@ import { hash, type MeyerBundle } from './contracts'; ...@@ -7,7 +7,7 @@ import { hash, type MeyerBundle } from './contracts';
/** 独立磁盘留底,不写主摄入游标。目录只能被单个 CLI 占用,文件默认仅当前用户可读。 */ /** 独立磁盘留底,不写主摄入游标。目录只能被单个 CLI 占用,文件默认仅当前用户可读。 */
export class MeyerStore { export class MeyerStore {
constructor(readonly directory: string) {} constructor(readonly directory: string) {}
async acquire(identity: unknown): Promise<() => Promise<void>> { async acquire(identity: unknown, legacyIdentities: unknown[] = [], migrate?: () => Promise<void>): Promise<() => Promise<void>> {
await mkdir(this.directory, { recursive: true, mode: 0o700 }); await mkdir(this.directory, { recursive: true, mode: 0o700 });
const lock = join(this.directory, '.lock'); const lock = join(this.directory, '.lock');
try { await mkdir(lock, { mode: 0o700 }); } catch { try { await mkdir(lock, { mode: 0o700 }); } catch {
...@@ -33,8 +33,13 @@ export class MeyerStore { ...@@ -33,8 +33,13 @@ export class MeyerStore {
await writeFile(join(lock, 'pid'), String(process.pid), { mode: 0o600 }); await writeFile(join(lock, 'pid'), String(process.pid), { mode: 0o600 });
const expected = hash(identity); const expected = hash(identity);
const existing = await this.read<{ identity: string }>('identity'); const existing = await this.read<{ identity: string }>('identity');
if (existing && existing.identity !== expected) throw new Error('Meyer state scope/config mismatch; use its original configuration'); if (existing && existing.identity !== expected && !legacyIdentities.some(v => hash(v) === existing.identity)) {
if (!existing) await this.write('identity', { identity: expected }); throw new Error('Meyer state scope/config mismatch; use its original configuration');
}
if (!existing || existing.identity !== expected) {
await migrate?.();
await this.write('identity', { identity: expected });
}
return () => rm(lock, { recursive: true }); return () => rm(lock, { recursive: true });
} catch (error) { await rm(lock, { recursive: true }); throw error; } } catch (error) { await rm(lock, { recursive: true }); throw error; }
} }
......
import type { MeyerConfig, MeyerSource } from './contracts';
export class MeyerBudgetExhausted extends Error {}
/** 整轮与诊所配额同时扣减;只统计列表/分析业务请求,不含令牌申请。 */
export class MeyerTickBudget {
requests = 0;
readonly deadline: number;
constructor(readonly limit: number, durationMs: number) { this.deadline = Date.now() + durationMs; }
available() { return this.requests < this.limit && Date.now() < this.deadline; }
source(factory: (config: MeyerConfig) => MeyerSource, allowance: number) {
let used = 0;
const available = () => used < allowance && this.available();
const spend = () => {
if (!available()) throw new MeyerBudgetExhausted('Meyer tick budget exhausted');
used++; this.requests++;
};
return { available, factory: (config: MeyerConfig): MeyerSource => {
const source = factory(config);
return {
list: async (scope, page) => { spend(); return source.list(scope, page); },
canRequestAnalysis: available,
symptom: async image => { spend(); return source.symptom(image); },
};
} };
}
}
...@@ -5,6 +5,7 @@ import { collectMeyer, type PendingImage } from '../src/integrations/meyer/colle ...@@ -5,6 +5,7 @@ import { collectMeyer, type PendingImage } from '../src/integrations/meyer/colle
import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts'; import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts';
import { analysisCapability } from '../src/integrations/meyer/retry-policy'; import { analysisCapability } from '../src/integrations/meyer/retry-policy';
import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler'; import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store'; import { MeyerStore } from '../src/integrations/meyer/store';
import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures'; import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures';
...@@ -130,7 +131,7 @@ describe('Meyer analysis retry policy', () => { ...@@ -130,7 +131,7 @@ describe('Meyer analysis retry policy', () => {
expect(result.retryFailed).toBe(1); expect(result.retryFailed).toBe(1);
expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1, autoRetry: false, reason: 'analysis_exhausted' }); expect(await s.pending.read(s.key)).toMatchObject({ analysisAttempts: 1, autoRetry: false, reason: 'analysis_exhausted' });
expect(s.source.symptom).toHaveBeenCalledTimes(1); expect(s.source.symptom).toHaveBeenCalledTimes(1);
await s.store.write('cursor', { through: chinaTime(created + 4 * 3600000) }); await clinicStateStore(s.store, s.config.scopes[0]!).write('cursor', { through: chinaTime(created + 4 * 3600000) });
const factory = jest.fn(() => s.source); const factory = jest.fn(() => s.source);
await runMeyerTick(s.config, factory, s.store, codes, s.sink, created + 4 * 3600000); await runMeyerTick(s.config, factory, s.store, codes, s.sink, created + 4 * 3600000);
expect(factory).not.toHaveBeenCalled(); expect(factory).not.toHaveBeenCalled();
......
import { mkdtemp, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { tmpdir } from 'node:os';
import { randomUUID } from 'node:crypto';
import { ConfigSchema, hash, imageKey, type MeyerConfig, type MeyerScope, type MeyerSource } from '../src/integrations/meyer/contracts';
import { MeyerStore } from '../src/integrations/meyer/store';
import { clinicStateStore, prepareClinicStates } from '../src/integrations/meyer/clinic-state';
import { runMeyerTick } from '../src/integrations/meyer/scheduler';
import { meyerConfig, meyerImage, meyerSymptom, codes } from './meyer-imaging-fixtures';
describe('Meyer clinic isolation and shared quotas', () => {
let dir: string, store: MeyerStore;
const now = Date.parse('2026-10-10T03:00:00Z');
const scopes = (count: number): MeyerScope[] => Array.from({ length: count }, (_, i) => ({
clinicId: `clinic-${String(i).padStart(3, '0')}`, tenantId: 'friday-market', sourceUnit: 'brand-a' }));
const config = (count = 2) => ConfigSchema.parse({ ...meyerConfig(), scopes: scopes(count), timeStart: '2026-10-10 09:00:00' });
const empty: MeyerSource = { list: async (_, page) => ({ total: 0, page, size: 100, images: [] }), symptom: async () => meyerSymptom() };
beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'meyer-clinic-')); store = new MeyerStore(dir); });
afterEach(async () => { await rm(dir, { recursive: true, force: true }); });
const sink = { apply: async () => {} };
const state = (c: MeyerConfig, index = 0) => clinicStateStore(store, c.scopes[index]!);
test('failing clinic keeps its cursor and backs off; another clinic advances in the same tick', async () => {
const c = config(), seen: string[] = [];
const factory = (): MeyerSource => ({ ...empty, list: async (scope, page) => {
seen.push(scope.clinicId); if (scope.clinicId === c.scopes[0]!.clinicId) throw Error('clinic unavailable');
return empty.list(scope, page);
} });
const result = await runMeyerTick(c, factory, store, codes, sink, now);
expect(result.clinicFailures).toBe(1);
expect(await state(c).read('cursor')).toBeNull();
expect(await state(c, 1).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
expect(seen).toEqual(c.scopes.map(s => s.clinicId));
seen.length = 0;
const later = await runMeyerTick(c, factory, store, codes, sink, now + 1000);
expect(later.clinics.find(x => x.clinicId === c.scopes[0]!.clinicId)?.skipped).toBe('backoff');
expect(seen).toEqual([c.scopes[1]!.clinicId]);
});
test('analysis outage uses item retry times and does not back off new image scans', async () => {
const c = config(1); let pages = 0;
const factory = (window: MeyerConfig): MeyerSource => ({
list: async scope => { pages++; const images = window.timeStart <= meyerImage().imageCreateTime!
? [{ ...meyerImage(), clinicId: scope.clinicId }] : [];
return { total: images.length, page: 1, size: 100, images }; },
symptom: async () => { throw Error('analysis unavailable'); },
});
const first = await runMeyerTick(c, factory, store, codes, sink, now);
expect(first.scan.analysisFailed).toBe(1);
expect(await state(c).read('failure')).toBeNull();
const second = await runMeyerTick(c, factory, store, codes, sink, now + c.pollIntervalMs);
expect(second.clinics[0]?.skipped).toBeUndefined(); expect(pages).toBe(2);
});
test('100 clinics share ten retries per round and rotate without starving later clinics', async () => {
const c = config(100), pending = new MeyerStore(join(dir, 'pending'));
for (const scope of c.scopes) {
const im = { ...meyerImage(), clinicId: scope.clinicId };
await pending.write(imageKey(scope, im.imageId), { scope, image: im, attempts: 1,
analysisAttempts: 2, nextRetryAt: '2000-01-01T00:00:00Z' });
}
const seen = new Set<string>();
for (let i = 0; i < 20; i++) {
const result = await runMeyerTick(c, () => empty, store, codes, sink, now + i * c.pollIntervalMs);
expect(result.retryAttempted).toBeLessThanOrEqual(10);
expect(result.sourceRequests).toBeLessThanOrEqual(100);
expect(result.scan.pages).toBeLessThanOrEqual(10);
expect(result.clinics.length).toBeLessThanOrEqual(5);
for (const x of result.clinics) seen.add(x.clinicId);
}
expect(seen.size).toBe(100);
expect((await pending.keys()).length).toBe(100);
}, 60000);
test('tiny image budget still rotates across clinics; long first-clinic history cannot monopolize scans', async () => {
const c = { ...config(), maxNewImagesPerTick: 1 }, written: string[] = [];
const factory = (): MeyerSource => ({ list: async scope => ({ total: 3, page: 1, size: 100,
images: [0, 1, 2].map(n => ({ ...meyerImage(), imageId: `im-${n}`, clinicId: scope.clinicId, aiType: 32 })) }),
symptom: async () => { throw Error('unsupported'); } });
for (let i = 0; i < 2; i++) await runMeyerTick(c, factory, store, codes,
{ apply: async b => { written.push(b.scope.clinicId); } }, now);
expect(written).toEqual(c.scopes.map(s => s.clinicId));
for (let i = 0; i < 2; i++) expect(await state(c, i).read('scan-progress')).toMatchObject({ row: 1 });
});
test('request quota defers analysis without consuming an actual analysis attempt or dropping metadata', async () => {
const c = { ...config(1), maxSourceRequestsPerTick: 2 }, written: string[] = [];
const images = [1, 2].map(n => ({ ...meyerImage(), imageId: `im-${n}`, clinicId: c.scopes[0]!.clinicId }));
const analysis = jest.fn(async () => ({ teeth: [], diagnosis: '', description: '' }));
const result = await runMeyerTick(c, () => ({ list: async () => ({ total: 2, page: 1, size: 100, images }),
symptom: analysis }), store, codes, { apply: async b => { written.push(b.raw.image.imageId); } }, now);
expect(result.sourceRequests).toBe(2); expect(result.clinicFailures).toBe(0);
expect(written).toHaveLength(2); expect(analysis).toHaveBeenCalledTimes(1);
const pending = new MeyerStore(join(dir, 'pending'));
expect(await pending.read(imageKey(c.scopes[0]!, images[1]!.imageId))).toMatchObject({ analysisAttempts: 0, autoRetry: true });
});
test('migration retains earlier completed clinics, current page ledger and later unscanned window', async () => {
const c = config(3), generation = randomUUID();
await store.write('cursor', { through: '2026-10-10 09:00:00' });
await store.write('scan-progress', { version: 1, generation, start: '2026-10-10 09:00:00', end: '2026-10-10 10:00:00',
scopeIndex: 1, page: 2, row: 1, pageSize: 100 });
const ledger = new MeyerStore(join(dir, 'scan-ledger', generation, '1'));
await ledger.write('page-1', ['legacy-image']);
await prepareClinicStates(store, c);
expect(await state(c).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
expect(await state(c, 1).read('scan-progress')).toMatchObject({ scopeIndex: 0, page: 2, row: 1, generation });
expect(await new MeyerStore(join(state(c, 1).directory, 'scan-ledger', generation, '0')).read('page-1')).toEqual(['legacy-image']);
expect(await state(c, 2).read('scan-progress')).toMatchObject({ page: 1, row: 0, end: '2026-10-10 10:00:00' });
await state(c).write('cursor', { through: '2026-10-11 00:00:00' });
await prepareClinicStates(store, c);
expect(await state(c).read('cursor')).toEqual({ through: '2026-10-11 00:00:00' });
});
test('added clinics start at configured history, not another clinic cursor; removed queues are retained', async () => {
const c = config(1); await store.write('cursor', { through: '2026-10-10 10:00:00' });
await prepareClinicStates(store, c);
const extra = config(2), windows: Record<string, string> = {};
await runMeyerTick(extra, window => ({ ...empty, list: async (s, page) => {
windows[s.clinicId] = window.timeStart; return empty.list(s, page);
} }), store, codes, sink, now);
expect(windows[c.scopes[0]!.clinicId]).toBe('2026-10-10 09:59:00');
expect(windows[extra.scopes[1]!.clinicId]).toBe('2026-10-10 09:00:00');
const pending = new MeyerStore(join(dir, 'pending'));
await pending.write('retired-clinic', { scope: extra.scopes[1], image: meyerImage(), attempts: 2, nextRetryAt: '2000-01-01T00:00:00Z' });
const result = await runMeyerTick(c, () => empty, store, codes, sink, now);
expect(result.retryAttempted).toBe(0); expect(await pending.read('retired-clinic')).not.toBeNull();
});
test('ownership changes and corrupt state fail only the affected clinic', async () => {
const c = config(); await prepareClinicStates(store, c);
await state(c).write('scope', { ...c.scopes[0], sourceUnit: 'another-brand' });
const result = await runMeyerTick(c, () => empty, store, codes, sink, now);
expect(result.clinicFailures).toBe(1);
expect(await state(c).read('cursor')).toBeNull();
expect(await state(c, 1).read('cursor')).not.toBeNull();
});
test('identity upgrade is locked and accepts only an exact legacy identity', async () => {
const old = { account: 'a', scopes: config().scopes }, next = { version: 2, account: 'a' };
const release = await store.acquire(old); await release();
const migrated = jest.fn(async () => { expect(await store.read('identity')).toEqual({ identity: hash(old) }); });
await expect(store.acquire(next, [{ ...old, account: 'b' }], migrated)).rejects.toThrow('scope/config');
expect(migrated).not.toHaveBeenCalled();
const done = await store.acquire(next, [old], migrated); await done();
expect(migrated).toHaveBeenCalledTimes(1);
const again = await store.acquire(next, [], migrated); await again();
expect(migrated).toHaveBeenCalledTimes(1);
});
});
import { createMeyerOpenHttp } from './helpers/meyer-open-http'; import { createMeyerOpenHttp } from './helpers/meyer-open-http';
import { hashSecret } from '@pac/utils'; import { hashSecret } from '@pac/utils';
import { PatientFactsLookupResponseSchema } from '../src/modules/open/patient-facts-lookup.schema'; import { PatientFactsLookupResponseSchema } from '../src/modules/open/patient-facts-lookup.schema';
import { resolveMeyerScopes } from '../src/integrations/meyer/org-mapping'; import { resolveMeyerScopes, loadMeyerScopeResults } from '../src/integrations/meyer/org-mapping';
import { InputConfigSchema } from '../src/integrations/meyer/contracts'; import { InputConfigSchema } from '../src/integrations/meyer/contracts';
import { NestFactory } from '@nestjs/core'; import { NestFactory } from '@nestjs/core';
import type { INestApplicationContext } from '@nestjs/common'; import type { INestApplicationContext } from '@nestjs/common';
...@@ -68,6 +68,19 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => { ...@@ -68,6 +68,19 @@ describePg('Meyer | isolated PostgreSQL + actual minimal CLI module', () => {
clinicId: 'clinic-a', action: 'visit_registered', subjectType: 'visit', subjectId: 'host-visit', clinicId: 'clinic-a', action: 'visit_registered', subjectType: 'visit', subjectId: 'host-visit',
rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } }); rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
} }
test('automatic clinic mapping isolates missing and conflicting clinics without losing valid scopes', async () => {
await hostRecord();
const result = await loadMeyerScopeResults(['missing-clinic', 'clinic-a']);
expect(result.scopes).toEqual(meyerConfig().scopes);
expect(result.failures).toEqual([{ clinicId: 'missing-clinic', reason: expect.any(String) }]);
await prisma.patientTransaction.create({ data: { hostId, tenantId: 'friday-market', patientId: otherPatientId,
clinicId: 'conflicting-clinic', action: 'visit_registered', subjectType: 'visit', subjectId: 'other', rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
await prisma.patientTransaction.create({ data: { hostId, tenantId: 'friday-market', patientId,
clinicId: 'conflicting-clinic', action: 'visit_registered', subjectType: 'visit', subjectId: 'another', rawPayload: {}, payloadHash: 'test', occurredAt: new Date() } });
const next = await loadMeyerScopeResults(['conflicting-clinic', 'clinic-a']);
expect(next.scopes).toEqual(meyerConfig().scopes);
expect(next.failures.map(x => x.clinicId)).toEqual(['conflicting-clinic']);
});
test('file refs backfill preserves versions, watermarks, frozen evidence and analysis; subsequent replay is unchanged', async () => { test('file refs backfill preserves versions, watermarks, frozen evidence and analysis; subsequent replay is unchanged', async () => {
const bundle = meyerBundle(); const bundle = meyerBundle();
await sink.apply(bundle); await sink.apply(bundle);
......
...@@ -2,6 +2,7 @@ import { mkdtemp, rm, mkdir, writeFile } from 'node:fs/promises'; ...@@ -2,6 +2,7 @@ import { mkdtemp, rm, mkdir, writeFile } from 'node:fs/promises';
import { hostname, tmpdir } from 'node:os'; import { hostname, tmpdir } from 'node:os';
import { spawn } from 'node:child_process'; import { spawn } from 'node:child_process';
import { join } from 'node:path'; import { join } from 'node:path';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store'; import { MeyerStore } from '../src/integrations/meyer/store';
import { runMeyerTick } from '../src/integrations/meyer/scheduler'; import { runMeyerTick } from '../src/integrations/meyer/scheduler';
import { watchMeyer, withRoundDeadline } from '../src/integrations/meyer/watch'; import { watchMeyer, withRoundDeadline } from '../src/integrations/meyer/watch';
...@@ -24,7 +25,7 @@ describe('Meyer conservative limits and recovery', () => { ...@@ -24,7 +25,7 @@ describe('Meyer conservative limits and recovery', () => {
const sink = { apply: async (b: MeyerBundle) => { if (b.phase === 'analysis') applied.push(b.raw.image.imageId); } }; const sink = { apply: async (b: MeyerBundle) => { if (b.phase === 'analysis') applied.push(b.raw.image.imageId); } };
const tick = () => runMeyerTick(c, () => source(rows), new MeyerStore(dir), codes, sink, now); const tick = () => runMeyerTick(c, () => source(rows), new MeyerStore(dir), codes, sink, now);
const first = await tick(); expect(first.scan?.examined).toBe(2); expect(first.cursor).toBeNull(); const first = await tick(); expect(first.scan?.examined).toBe(2); expect(first.cursor).toBeNull();
expect(await new MeyerStore(dir).read('scan-progress')).toMatchObject({ page: 1, row: 2 }); expect(await clinicStateStore(new MeyerStore(dir), c.scopes[0]!).read('scan-progress')).toMatchObject({ page: 1, row: 2 });
rows.forEach(r => { Object.assign(r, { imageUrl: 'https://example.com/renewed-signature' }); }); rows.forEach(r => { Object.assign(r, { imageUrl: 'https://example.com/renewed-signature' }); });
const second = await tick(); expect(second.scan?.examined).toBe(2); expect(second.cursor).toBeNull(); const second = await tick(); expect(second.scan?.examined).toBe(2); expect(second.cursor).toBeNull();
const third = await tick(); expect(third.scan?.examined).toBe(1); expect(third.scan?.complete).toBe(true); const third = await tick(); expect(third.scan?.examined).toBe(1); expect(third.scan?.complete).toBe(true);
...@@ -52,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => { ...@@ -52,7 +53,7 @@ describe('Meyer conservative limits and recovery', () => {
const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now); const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull(); expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
down = false; down = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now); const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now + c.pollIntervalMs);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10); expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
}); });
test('duplicate caused by offset drift blocks completion and can recover by re-scan', async () => { test('duplicate caused by offset drift blocks completion and can recover by re-scan', async () => {
...@@ -65,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => { ...@@ -65,7 +66,7 @@ describe('Meyer conservative limits and recovery', () => {
const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now); const first = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now);
expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull(); expect(first.scan?.failed).toBe(1); expect(first.cursor).toBeNull();
drift = false; drift = false;
const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now); const next = await runMeyerTick(c, factory, new MeyerStore(dir), codes, sink, now + c.pollIntervalMs);
expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10); expect(next.scan?.complete).toBe(true); expect(sink.apply).toHaveBeenCalledTimes(10);
}); });
test('watch backs off for thrown outages and error exits, recovers, then stops cleanly', async () => { test('watch backs off for thrown outages and error exits, recovers, then stops cleanly', async () => {
......
...@@ -2,6 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises'; ...@@ -2,6 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises';
import { join } from 'node:path'; import { join } from 'node:path';
import { tmpdir } from 'node:os'; import { tmpdir } from 'node:os';
import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler'; import { runMeyerTick, chinaTime } from '../src/integrations/meyer/scheduler';
import { clinicStateStore } from '../src/integrations/meyer/clinic-state';
import { MeyerStore } from '../src/integrations/meyer/store'; import { MeyerStore } from '../src/integrations/meyer/store';
import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts'; import { imageKey, type MeyerSource } from '../src/integrations/meyer/contracts';
import { parseMeyerArgs } from '../src/cli/meyer-imaging.cli'; import { parseMeyerArgs } from '../src/cli/meyer-imaging.cli';
...@@ -25,7 +26,7 @@ describe('independent Meyer scheduling', () => { ...@@ -25,7 +26,7 @@ describe('independent Meyer scheduling', () => {
const first = await runMeyerTick(c, factory, store, codes, sink, now); const first = await runMeyerTick(c, factory, store, codes, sink, now);
expect(first.pending).toBe(1); expect(first.scan?.failed).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(1); expect(first.pending).toBe(1); expect(first.scan?.failed).toBe(0); expect(sink.apply).toHaveBeenCalledTimes(1);
expect(first.scan?.imagesApplied).toBe(1); expect(first.scan?.imagesApplied).toBe(1);
expect(await store.read('cursor')).toEqual({ through: '2026-10-10 10:00:00' }); expect(await clinicStateStore(store, c.scopes[0]!).read('cursor')).toEqual({ through: '2026-10-10 10:00:00' });
const key = imageKey(c.scopes[0]!, im.imageId); const key = imageKey(c.scopes[0]!, im.imageId);
const item = await pending.read<any>(key); await pending.write(key, { ...item, nextRetryAt: '2000-01-01T00:00:00Z' }); const item = await pending.read<any>(key); await pending.write(key, { ...item, nextRetryAt: '2000-01-01T00:00:00Z' });
ready = true; ready = true;
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment