From 6ecc38d9270db35244715bd72faf6c6b5c2a3b0d Mon Sep 17 00:00:00 2001 From: ttipo Date: Sun, 8 Nov 2026 00:48:37 +0900 Subject: [PATCH] =?UTF-8?q?refactor(backend):=20=EB=8F=99=EC=8B=9C=20?= =?UTF-8?q?=EC=88=98=20=EC=A0=9C=ED=95=9C=EA=B3=BC=20=EC=A7=81=EC=9B=90?= =?UTF-8?q?=EB=A7=88=EB=8B=A4=20=EC=B0=A8=EB=A1=80=EB=A5=BC=20shared/utils?= =?UTF-8?q?=20=EB=A1=9C=20=EC=98=AE=EA=B8=B4=EB=8B=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 앱 새 메일 알림(푸시)도 14단계 검색 색인처럼 서버 전체 동시 수와 직원마다 한 번에 하나를 쓴다. 푸시 모듈이 mail-search 를 가져오지 않게 ConcurrencyGate 를 shared/utils 로 옮기고, 직원마다 차례를 KeyedQueue 로 뽑아 MailSearchRuns 가 상속하게 했다. 14단계 동작은 같다. Co-Authored-By: Claude Opus 5.5 --- .../modules/mail-search/mail-search-runs.ts | 64 +------------------ .../mail-search/mail-search-sync.service.ts | 3 +- .../src/shared/utils/concurrency-gate.spec.ts | 48 ++++++++++++++ backend/src/shared/utils/concurrency-gate.ts | 34 ++++++++++ backend/src/shared/utils/keyed-queue.spec.ts | 54 ++++++++++++++++ backend/src/shared/utils/keyed-queue.ts | 34 ++++++++++ 6 files changed, 175 insertions(+), 62 deletions(-) create mode 100644 backend/src/shared/utils/concurrency-gate.spec.ts create mode 100644 backend/src/shared/utils/concurrency-gate.ts create mode 100644 backend/src/shared/utils/keyed-queue.spec.ts create mode 100644 backend/src/shared/utils/keyed-queue.ts diff --git a/backend/src/modules/mail-search/mail-search-runs.ts b/backend/src/modules/mail-search/mail-search-runs.ts index c4c3166..2b90f54 100644 --- a/backend/src/modules/mail-search/mail-search-runs.ts +++ b/backend/src/modules/mail-search/mail-search-runs.ts @@ -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>(); - - /** 그 직원의 앞선 일이 끝난 뒤 `task` 를 한다 → 그 결과 */ - run(user: string, task: () => Promise): Promise { - 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(task: () => Promise): Promise { - if (this.active >= this.limit) { - await new Promise((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 {} diff --git a/backend/src/modules/mail-search/mail-search-sync.service.ts b/backend/src/modules/mail-search/mail-search-sync.service.ts index bae36c8..3f5f391 100644 --- a/backend/src/modules/mail-search/mail-search-sync.service.ts +++ b/backend/src/modules/mail-search/mail-search-sync.service.ts @@ -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, diff --git a/backend/src/shared/utils/concurrency-gate.spec.ts b/backend/src/shared/utils/concurrency-gate.spec.ts new file mode 100644 index 0000000..d53522a --- /dev/null +++ b/backend/src/shared/utils/concurrency-gate.spec.ts @@ -0,0 +1,48 @@ +import { ConcurrencyGate } from './concurrency-gate'; + +/** 손으로 끝내는 일 */ +function task() { + let finish!: () => void; + let fail!: (error: unknown) => void; + const done = new Promise((resolve, reject) => { + finish = resolve; + fail = reject; + }); + return { done, finish, fail }; +} + +async function flush(): Promise { + 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 }); + }); +}); diff --git a/backend/src/shared/utils/concurrency-gate.ts b/backend/src/shared/utils/concurrency-gate.ts new file mode 100644 index 0000000..b5fbeac --- /dev/null +++ b/backend/src/shared/utils/concurrency-gate.ts @@ -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(task: () => Promise): Promise { + if (this.active >= this.limit) { + await new Promise((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; + } + } +} diff --git a/backend/src/shared/utils/keyed-queue.spec.ts b/backend/src/shared/utils/keyed-queue.spec.ts new file mode 100644 index 0000000..42fe0e8 --- /dev/null +++ b/backend/src/shared/utils/keyed-queue.spec.ts @@ -0,0 +1,54 @@ +import { + currentRequestId, + runWithRequestContext, +} from '@/shared/logger/request-context'; +import { KeyedQueue } from './keyed-queue'; + +async function flush(): Promise { + 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((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(); + }); +}); diff --git a/backend/src/shared/utils/keyed-queue.ts b/backend/src/shared/utils/keyed-queue.ts new file mode 100644 index 0000000..1f89c16 --- /dev/null +++ b/backend/src/shared/utils/keyed-queue.ts @@ -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>(); + + /** 그 열쇠의 앞선 일이 끝난 뒤 `task` 를 한다 → 그 결과 */ + run(key: string, task: () => Promise): Promise { + 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()); + } +}