generated from ttipo/project-base
refactor(backend): 동시 수 제한과 직원마다 차례를 shared/utils 로 옮긴다
앱 새 메일 알림(푸시)도 14단계 검색 색인처럼 서버 전체 동시 수와 직원마다 한 번에 하나를 쓴다. 푸시 모듈이 mail-search 를 가져오지 않게 ConcurrencyGate 를 shared/utils 로 옮기고, 직원마다 차례를 KeyedQueue 로 뽑아 MailSearchRuns 가 상속하게 했다. 14단계 동작은 같다. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,71 +1,13 @@
|
||||
import { Injectable } from '@nestjs/common';
|
||||
|
||||
import { runOutsideRequestContext } from '@/shared/logger/request-context';
|
||||
import { KeyedQueue } from '@/shared/utils/keyed-queue';
|
||||
|
||||
/**
|
||||
* 직원마다 색인 일을 한 번에 하나씩 (14단계 설계 B2 — "같은 직원은 한 번에 하나") — 맞추기 · 채우기 한 묶음 · 전체 대조가 같은 줄에 선다.
|
||||
* 앞의 일이 실패해도 다음 일은 한다. 일은 요청 컨텍스트 밖에서 돈다 — 처음 부른 요청의 추적 ID 를 물려받지 않게.
|
||||
* 앞의 일이 실패해도 다음 일은 한다. 일은 요청 컨텍스트 밖에서 돈다 — 처음 부른 요청의 추적 ID 를 물려받지 않게 (차례는 shared/utils/keyed-queue.ts).
|
||||
*
|
||||
* ⚠️ 프로세스 안의 차례다 — 인스턴스를 늘리면 두 인스턴스가 같은 직원을 함께 맞춘다(쓰기는 상태 줄을 잠가 차례로 되지만 같은 일을 두 번 한다).
|
||||
* 그때는 잠금(Redis `SET NX PX` 또는 `pg_advisory_lock`)으로 옮긴다 (nestjs 스킬의 Redis 목록)
|
||||
*/
|
||||
@Injectable()
|
||||
export class MailSearchRuns {
|
||||
private readonly tails = new Map<string, Promise<void>>();
|
||||
|
||||
/** 그 직원의 앞선 일이 끝난 뒤 `task` 를 한다 → 그 결과 */
|
||||
run<T>(user: string, task: () => Promise<T>): Promise<T> {
|
||||
const key = user.toLowerCase();
|
||||
const previous = this.tails.get(key) ?? Promise.resolve();
|
||||
const start = () => runOutsideRequestContext(task);
|
||||
const result = previous.then(start, start);
|
||||
const tail = result.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
this.tails.set(key, tail);
|
||||
void tail.then(() => {
|
||||
if (this.tails.get(key) === tail) this.tails.delete(key);
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 그 직원의 일이 줄에 있거나 도는 중인가 */
|
||||
isBusy(user: string): boolean {
|
||||
return this.tails.has(user.toLowerCase());
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 함께 도는 일의 수 — 넘는 일은 먼저 온 차례로 기다린다 (14단계 — 실시간 소식이 직원 수만큼 한꺼번에 와도 맞추기는 이만큼만 함께).
|
||||
* 앞의 일이 실패해도 자리는 돌려준다.
|
||||
*
|
||||
* ⚠️ 프로세스 안의 셈이다 — 인스턴스를 늘리면 인스턴스마다 따로 센다 (nestjs 스킬의 Redis 목록)
|
||||
*/
|
||||
export class ConcurrencyGate {
|
||||
private active = 0;
|
||||
private readonly waiting: (() => void)[] = [];
|
||||
|
||||
constructor(private readonly limit: number) {}
|
||||
|
||||
/** 지금 도는 일 · 기다리는 일의 수 (시험용) */
|
||||
get load(): { readonly active: number; readonly waiting: number } {
|
||||
return { active: this.active, waiting: this.waiting.length };
|
||||
}
|
||||
|
||||
async run<T>(task: () => Promise<T>): Promise<T> {
|
||||
if (this.active >= this.limit) {
|
||||
await new Promise<void>((resolve) => this.waiting.push(resolve));
|
||||
} else {
|
||||
this.active += 1;
|
||||
}
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
// 기다리는 일에 자리를 그대로 넘긴다 (자리 수는 그대로) — 없으면 돌려준다
|
||||
const next = this.waiting.shift();
|
||||
if (next) next();
|
||||
else this.active -= 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
export class MailSearchRuns extends KeyedQueue {}
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
PasswordExpiredException,
|
||||
} from '@/modules/mail/mail.provider';
|
||||
import { MAIL_SETTINGS, type MailSettings } from '@/modules/mail/mail.settings';
|
||||
import { ConcurrencyGate } from '@/shared/utils/concurrency-gate';
|
||||
import { maskEmail } from '@/shared/utils/mask';
|
||||
import { settledWithin } from '@/shared/utils/settle';
|
||||
import {
|
||||
@@ -31,7 +32,7 @@ import {
|
||||
MailSearchIndexRepository,
|
||||
type SearchStatePatch,
|
||||
} from './mail-search-index.repository';
|
||||
import { ConcurrencyGate, MailSearchRuns } from './mail-search-runs';
|
||||
import { MailSearchRuns } from './mail-search-runs';
|
||||
import {
|
||||
MAIL_SEARCH_TIMING,
|
||||
SYSTEM_SEARCH_TIMING,
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import { ConcurrencyGate } from './concurrency-gate';
|
||||
|
||||
/** 손으로 끝내는 일 */
|
||||
function task() {
|
||||
let finish!: () => void;
|
||||
let fail!: (error: unknown) => void;
|
||||
const done = new Promise<void>((resolve, reject) => {
|
||||
finish = resolve;
|
||||
fail = reject;
|
||||
});
|
||||
return { done, finish, fail };
|
||||
}
|
||||
|
||||
async function flush(): Promise<void> {
|
||||
for (let round = 0; round < 5; round += 1) {
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
}
|
||||
}
|
||||
|
||||
describe('ConcurrencyGate (함께 도는 일의 수)', () => {
|
||||
it('한도까지만 함께 돌고, 넘는 일은 먼저 온 차례로 기다린다 — 앞의 일이 실패해도 자리를 넘긴다', async () => {
|
||||
const gate = new ConcurrencyGate(2);
|
||||
const started: number[] = [];
|
||||
const tasks = [task(), task(), task(), task()];
|
||||
const runs = tasks.map((one, index) =>
|
||||
gate.run(() => {
|
||||
started.push(index);
|
||||
return one.done;
|
||||
}),
|
||||
);
|
||||
await flush();
|
||||
expect(started).toEqual([0, 1]);
|
||||
expect(gate.load).toEqual({ active: 2, waiting: 2 });
|
||||
|
||||
tasks[1].fail(new Error('실패'));
|
||||
await expect(runs[1]).rejects.toThrow('실패');
|
||||
await flush();
|
||||
expect(started).toEqual([0, 1, 2]);
|
||||
|
||||
tasks[0].finish();
|
||||
tasks[2].finish();
|
||||
await flush();
|
||||
expect(started).toEqual([0, 1, 2, 3]);
|
||||
tasks[3].finish();
|
||||
await Promise.all([runs[0], runs[2], runs[3]]);
|
||||
expect(gate.load).toEqual({ active: 0, waiting: 0 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
/**
|
||||
* 함께 도는 일의 수 — 넘는 일은 먼저 온 차례로 기다린다. 앞의 일이 실패해도 자리는 돌려준다.
|
||||
* 쓰는 곳: 14단계 검색 색인 맞추기(실시간 소식이 직원 수만큼 한꺼번에 와도 맞추기는 이만큼만 함께 — mail-search-sync.service.ts),
|
||||
* 앱 새 메일 알림의 FCM 보내기 · 엔진 연결 붙잡기(modules/push/fcm/fcm-sender.ts · modules/mail/new-mail-push.service.ts).
|
||||
*
|
||||
* ⚠️ 프로세스 안의 셈이다 — 인스턴스를 늘리면 인스턴스마다 따로 센다 (nestjs 스킬의 Redis 목록)
|
||||
*/
|
||||
export class ConcurrencyGate {
|
||||
private active = 0;
|
||||
private readonly waiting: (() => void)[] = [];
|
||||
|
||||
constructor(private readonly limit: number) {}
|
||||
|
||||
/** 지금 도는 일 · 기다리는 일의 수 (시험용) */
|
||||
get load(): { readonly active: number; readonly waiting: number } {
|
||||
return { active: this.active, waiting: this.waiting.length };
|
||||
}
|
||||
|
||||
async run<T>(task: () => Promise<T>): Promise<T> {
|
||||
if (this.active >= this.limit) {
|
||||
await new Promise<void>((resolve) => this.waiting.push(resolve));
|
||||
} else {
|
||||
this.active += 1;
|
||||
}
|
||||
try {
|
||||
return await task();
|
||||
} finally {
|
||||
// 기다리는 일에 자리를 그대로 넘긴다 (자리 수는 그대로) — 없으면 돌려준다
|
||||
const next = this.waiting.shift();
|
||||
if (next) next();
|
||||
else this.active -= 1;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
import {
|
||||
currentRequestId,
|
||||
runWithRequestContext,
|
||||
} from '@/shared/logger/request-context';
|
||||
import { KeyedQueue } from './keyed-queue';
|
||||
|
||||
async function flush(): Promise<void> {
|
||||
for (let round = 0; round < 5; round += 1) {
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
}
|
||||
}
|
||||
|
||||
describe('KeyedQueue (열쇠마다 한 번에 하나)', () => {
|
||||
it('같은 열쇠(대소문자 무시)의 일은 차례로 — 다른 열쇠는 함께, 앞의 일이 실패해도 다음 일은 한다', async () => {
|
||||
const queue = new KeyedQueue();
|
||||
const order: string[] = [];
|
||||
let release!: () => void;
|
||||
const first = queue.run('Ara@x.test', async () => {
|
||||
order.push('a1 시작');
|
||||
await new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
order.push('a1 끝');
|
||||
throw new Error('실패');
|
||||
});
|
||||
const second = queue.run('ara@x.test', () => {
|
||||
order.push('a2');
|
||||
return Promise.resolve('둘째');
|
||||
});
|
||||
const other = queue.run('doyun@x.test', () => {
|
||||
order.push('d1');
|
||||
return Promise.resolve();
|
||||
});
|
||||
await flush();
|
||||
expect(order).toEqual(['a1 시작', 'd1']);
|
||||
expect(queue.isBusy('ARA@x.test')).toBe(true);
|
||||
|
||||
release();
|
||||
await expect(first).rejects.toThrow('실패');
|
||||
await expect(second).resolves.toBe('둘째');
|
||||
await other;
|
||||
expect(order).toEqual(['a1 시작', 'd1', 'a1 끝', 'a2']);
|
||||
await flush();
|
||||
expect(queue.isBusy('ara@x.test')).toBe(false);
|
||||
});
|
||||
|
||||
it('일은 요청 컨텍스트 밖에서 돈다 — 처음 부른 요청의 추적 ID 를 물려받지 않는다', async () => {
|
||||
const queue = new KeyedQueue();
|
||||
const seen = await runWithRequestContext({ requestId: 'trace-1' }, () =>
|
||||
queue.run('ara@x.test', () => Promise.resolve(currentRequestId())),
|
||||
);
|
||||
expect(seen).toBeUndefined();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
import { runOutsideRequestContext } from '@/shared/logger/request-context';
|
||||
|
||||
/**
|
||||
* 열쇠(직원 주소 등)마다 일을 한 번에 하나씩 — 같은 열쇠의 일은 줄을 서고, 앞의 일이 실패해도 다음 일은 한다.
|
||||
* 열쇠는 소문자로 맞춰 견준다. 일은 요청 컨텍스트 밖에서 돈다 — 처음 부른 요청의 추적 ID 를 물려받지 않게.
|
||||
* 쓰는 곳: 14단계 검색 색인의 직원마다 차례(MailSearchRuns), 앱 새 메일 알림의 직원마다 보내기 차례(new-mail-push.service.ts).
|
||||
*
|
||||
* ⚠️ 프로세스 안의 차례다 — 인스턴스를 늘리면 인스턴스마다 따로 선다 (nestjs 스킬의 Redis 목록)
|
||||
*/
|
||||
export class KeyedQueue {
|
||||
private readonly tails = new Map<string, Promise<void>>();
|
||||
|
||||
/** 그 열쇠의 앞선 일이 끝난 뒤 `task` 를 한다 → 그 결과 */
|
||||
run<T>(key: string, task: () => Promise<T>): Promise<T> {
|
||||
const normalized = key.toLowerCase();
|
||||
const previous = this.tails.get(normalized) ?? Promise.resolve();
|
||||
const start = () => runOutsideRequestContext(task);
|
||||
const result = previous.then(start, start);
|
||||
const tail = result.then(
|
||||
() => undefined,
|
||||
() => undefined,
|
||||
);
|
||||
this.tails.set(normalized, tail);
|
||||
void tail.then(() => {
|
||||
if (this.tails.get(normalized) === tail) this.tails.delete(normalized);
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
/** 그 열쇠의 일이 줄에 있거나 도는 중인가 */
|
||||
isBusy(key: string): boolean {
|
||||
return this.tails.has(key.toLowerCase());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user