중복 실행과 부분 실패 뒤에도 자동화 결과를 안전하게 재개하려면 어떻게 하나요?
같은 입력이 다시 들어와도 결과가 중복되지 않게 하는 성질이 멱등성입니다. 작업 결과는 생겼지만 완료 기록 전에 중단될 수 있으므로, 재개할 때 상태뿐 아니라 실제 결과도 확인해야 합니다. 메모리 안의 작은 모의 예제로 중복 처리와 중간 재개의 차이를 비교합니다.
적용 조건·범위: completed와 failed를 같은 건너뛰기 조건으로 묶지 않습니다. stale processing은 결과파일과 내용·식별자를 먼저 확인합니다. 상태파일과 외부 쓰기를 묶는 transaction이 없으므로 임의 상태 삭제 후 재실행으로 해결하지 않습니다.
공식 문서 참고: AWS Builders’ Library: 재시도·멱등성 키·원자적 기록
결과 생성 뒤 중단된 작업을 재개하기
Node.js 24의 Map 두 개로 상태 저장소와 결과 저장소를 모의합니다. 외부 작업 없이 crash window에서 결과 조회가 필요한 이유만 확인합니다. 실제 파일·DB의 영속성이나 동시 실행 안전성을 보장하는 예제는 아닙니다.
resume-demo.mjs 전체 코드
const states = new Map();
const results = new Map();
function apply(key, crash = false) {
if (states.get(key) === 'completed') return 'skipped';
if (results.has(key)) {
states.set(key, 'completed');
return 'reconciled';
}
states.set(key, 'processing');
results.set(key, { resultId: `result-${key}` });
if (crash) throw new Error('결과 저장 뒤 중단');
states.set(key, 'completed');
return 'completed';
}
try { apply('task-a:create:v1', true); }
catch (error) { console.log(error.message); }
console.log(apply('task-a:create:v1'));
console.log(apply('task-a:create:v1'));
console.log(`결과 수 ${results.size}`);
저장한 폴더에서 node resume-demo.mjs로 실행합니다. 예상 출력은 다음과 같습니다.
결과 저장 뒤 중단
reconciled
skipped
결과 수 1
직접 바꿔 보기: results.has 분기를 지우면 복구 때 결과 생성을 다시 시도한다는 점을 추적해 보세요. 외부 서비스에서는 같은 결과 키의 존재뿐 아니라 요청 내용·revision이 같은지도 확인해야 하며, 조회가 불가능하면 자동 재시도 대신 수동 검토가 필요할 수 있습니다.
핵심 원리
멱등성은 같은 작업을 여러 번 요청해도 최종 결과가 한 번 처리한 것과 같도록 만드는 성질입니다. checkpoint와 resume을 함께 사용하면 중간 실패 이후 처음부터 반복하지 않아도 됩니다.
이번 단계부터는 정상 경로만으로 자동화를 평가하지 않습니다. 중복 이벤트, 네트워크 실패, 늦은 응답, 부분 성공, 재실행이 실제 운영에서 더 중요한 조건이 됩니다.
독립 실습: crash window 그리기
완료 ID만 저장하는 것으로 모든 중복을 막을 수는 없습니다. 외부 작업 성공 뒤 완료 상태 저장 전에 프로세스가 종료되는 구간이 crash window입니다.
처리 의도 기록
→ 업무 결과 생성
→ 완료 상태 기록
↑
결과는 있지만 상태를 못 남기고 중단될 수 있음
재개 시 processing을 실패로 바꿔 무조건 실행하면 이미 만들어진 결과가 중복될 수 있습니다. 결과 ID·내용을 확인한 뒤 완료 상태를 보정하거나 수동 검토로 넘깁니다. 반대로 실제 작업 전에 completed를 기록하면 실패한 작업이 영구히 건너뛰어질 수 있습니다.
키는 sourceId 또는 eventId와 operation을 연결하되, 같은 원본의 수정본을 별도 작업으로 취급해야 한다면 revision도 고려합니다. runId는 매 실행 달라지므로 업무 멱등성 키를 대신하지 않습니다. 파일 기반 예제는 단일 프로세스 학습용이며 분산 lock·DB transaction·외부 API의 exactly-once 보장이 아닙니다.
자주 발생하는 오류와 해결
completed와 failed를 같은 건너뛰기 조건으로 묶지 않습니다. stale processing은 결과파일과 내용·식별자를 먼저 확인합니다. 상태파일과 외부 쓰기를 묶는 transaction이 없으므로 임의 상태 삭제 후 재실행으로 해결하지 않습니다.
학습 완료 체크리스트
- 같은 업무 키 재실행 시 결과를 대조했습니다.
- 부분 실패 후 완료 항목과 실패 항목을 분리해 재개합니다.
- 파일 상태만으로 분산 exactly-once를 보장하지 않는다는 한계를 설명합니다.
선택 실습: 같은 프로젝트로 연습하기
TaskFlow Ops Automation은 작업 처리 상태와 실패·재시도를 관리하는 자동화 예제입니다. 이 선택 실습에서는 처리 흐름을 이어서 만듭니다.
이번 주제: 멱등성 체크포인트 재개 자동화 중복 실행과 부분 실패 복구.
선택: 연재 프로젝트와 전체 자료
연재 프로젝트를 이어 만드는 분을 위한 자료입니다. 이 글의 개념과 예제만 찾으셨다면 필요한 파일과 실행 방법만 펼쳐보세요.
TaskFlow Ops Automation 누적 적용
이어서 시작하기: API 재시도 설계 exponential backoff와 rate limit 대응의 완료 상태를 사용합니다. 중간 합류용 시작 ZIP과 이번 완료 ZIP은 아래 기존 다운로드 안내에서 구분해 선택합니다.
시작 체크포인트: automation-step-08-retry. 이전 회차 결과를 확인하고 새 ZIP은 별도 폴더에서 비교합니다.
현재 파일 구조는 아래 전체 코드의 경로 목록과 같습니다. 원본 input을 덮어쓰지 않으며 output·logs는 실행 결과입니다.
이번 추가 파일: automation/input/resume-tasks.json, src/automation/resume.mjs, tests/resume.test.mjs.
이번 변경 파일: README.md.
resume.mjs와 비식별 입력을 추가합니다. sourceId와 normalize-store-v1으로 만든 키를 파일 상태·불변 결과와 연결합니다. 기존 run()의 실행별 결과 생성과 별도의 재개 경로이며 Webhook이나 scheduler 전체가 자동으로 영속 멱등성을 얻지는 않습니다. 같은 output 폴더를 여러 프로세스가 동시에 쓰지 않는 교육용 구현입니다.
Before는 이전 체크포인트의 입력과 결과를 확인한 상태입니다. 적용은 아래 파일을 상대 경로에 맞춰 저장하고 README 순서로 실행하는 단계입니다. After는 이번 단계의 결과와 실패 처리 한계를 직접 설명할 수 있는 상태입니다.
전체 파일·코드 보기
실습 프로젝트 전체 코드
이 체크포인트는 Node.js 24 이상의 로컬 비식별 실습입니다. 외부 전송과 실제 예약 실행은 포함하지 않습니다.
automation/input/external-tasks.json
[
{ "id": "external-1", "name": " API 작업 ", "due_at": "2026-09-21T00:00:00Z" },
{ "id": "external-2", "name": "" },
{ "id": "external-3", "name": "날짜 오류", "due_at": "2026-02-30T00:00:00Z" },
{ "id": "external-4", "name": "상태 오류", "status": "unknown" }
]
automation/input/resume-tasks.json
[
{ "sourceId": "sample-1", "title": "작업 1" },
{ "sourceId": "sample-2", "title": "작업 2" },
{ "sourceId": "sample-3", "title": "작업 3" }
]
automation/input/task.json
{
"sourceId": "sample-1",
"title": " 주간 작업 점검 ",
"status": "todo",
"dueAt": null
}
automation/input/tasks.csv
sourceId,title,status,dueAt
sample-1, 주간 점검 ,todo,
sample-2,결과 검토,doing,2026-09-20T00:00:00Z
sample-3,,todo,
sample-1, 주간 점검 ,todo,
automation/input/tasks.json
[
{ "sourceId": "sample-1", "title": " 주간 점검 ", "status": "todo", "dueAt": null },
{ "sourceId": "sample-2", "title": "결과 검토", "status": "doing", "dueAt": "2026-09-20T00:00:00Z" },
{ "sourceId": "sample-3", "title": "", "status": "todo", "dueAt": null },
{ "sourceId": "sample-1", "title": " 주간 점검 ", "status": "todo", "dueAt": null }
]
docs/automation-decision.md
# TaskFlow Ops Automation 도구 선택
입력은 개인정보가 없는 Task JSON입니다. sourceId로 원본을 연결하고 제목을 검증·정규화한 후 로컬 결과와 실행 기록을 저장합니다.
| 방식 | 선택 기준 | 운영 책임 |
| --- | --- | --- |
| Zapier / Make | 제공 커넥터와 시각적 연결이 업무에 맞음 | 권한·비용·실패 이력·재실행 확인 |
| n8n | 시각적 흐름과 코드 조합이 필요함 | 배포 방식별 업데이트·보안·상태 저장 책임 확인 |
| Node.js | 파일과 오류 처리를 직접 제어하며 학습함 | 테스트·배포·상태·복구를 직접 설계 |
이 과정은 Node.js 24 기본 모듈을 선택합니다. npm install은 필요하지 않습니다. 실제 외부 계정·토큰·개인정보를 사용하지 않습니다. API와 Webhook은 127.0.0.1 모의 서버, 예약은 수동 시뮬레이션만 사용합니다.
실패 시 입력 형식과 실행 결과를 먼저 확인합니다. 외부 전송·업로드·알림은 사람이 범위와 권한을 검토한 별도 작업입니다. 2편의 재실행은 별도 결과를 만드므로 멱등성 완성이 아닙니다. 9편에서 파일 상태와 결과를 확인하며 재개합니다.
package.json
{"name":"taskflow-ops-automation","private":true,"type":"module","engines":{"node":">=24"},"scripts":{"test":"node --test","start":"node src/automation/run.mjs"}}
README.md
# TaskFlow Ops Automation · 09 멱등성·재개
Node.js 24 이상, 설치할 패키지 없음. ZIP을 푼 폴더에서 실행합니다.
node --test
node src/automation/run.mjs
node src/automation/run.mjs
정상 결과: 각 실행 successCount=1, failedCount=0. automation/output/실행ID/valid.json과 rejected.json, automation/logs/실행ID.json이 생성됩니다. 두 번 실행하면 서로 다른 실행ID 폴더가 생깁니다. 이 단계는 실행 내 sourceId 중복만 거르며 영속 멱등성은 9편에서 구현합니다.
입력: automation/input/task.json. sourceId는 1~80자 영문·숫자·밑줄·하이픈, title은 공백 제거 후 비어 있지 않은 200자 이하 문자열입니다. status는 todo/doing/done(기본 todo), dueAt은 null(기본) 또는 UTC ISO 날짜입니다. 예: 2026-09-21T00:00:00Z. 원본을 덮어쓰지 않습니다.
검증 실패는 rejected.json에 row/sourceId/field/reason/input으로 분리하며 CLI exit 1입니다. 읽기·파싱·저장 자체의 실패도 exit 1입니다. 로그는 집계만 남기며 실제 개인정보는 입력하지 않습니다. 테스트는 임시 폴더에 작은 교육 결과를 남깁니다.
## 이번 편 추가
node src/automation/batch.mjs automation/input/tasks.json
node src/automation/batch.mjs automation/input/tasks.csv
두 형식 모두 successCount=2, failedCount=1, skippedCount=1, partial_failure와 exit 1이 정상 관찰 결과입니다. 오류 행을 의도적으로 포함했습니다. rejected의 row는 헤더를 제외한 입력 레코드 번호(1부터)입니다. 출력 경로는 각 명령의 runId를 확인합니다.
CSV는 sourceId,title,status,dueAt의 단순 UTF-8 예제만 허용합니다. 따옴표가 있는 CSV, 필드 내부 쉼표·줄바꿈은 지원하지 않고 명시적으로 거부합니다. 실제 복잡한 CSV에는 별도 검증된 파서가 필요합니다. 파일 파싱 실패는 전체 중단, 레코드 검증 실패는 나머지 행 처리를 계속합니다.
## 04 로컬 API
node src/automation/demo-api.mjs
모의 서버를 127.0.0.1 임의 포트에 열고 GET 2페이지 3건을 같은 run()으로 저장한 뒤 로컬 POST를 확인하고 종료합니다. successCount=3, 로컬 POST 응답 sample-new가 예상 결과입니다. API_BASE_URL/API_TOKEN은 실제 서비스 연결 때 필요한 설정 경계이며 이 실습은 서버 주소를 자동 생성하고 공개 교육 문자열 local-training-only만 사용합니다. .env나 실제 키가 필요하지 않습니다.
API client는 외부 호스트와 redirect를 거부하고 HTTP 상태를 error.status에 보존합니다. 401/403은 권한, 404는 경로, 429는 제한, 5xx는 서버 오류로 구분합니다. 04에서는 자동 재시도하지 않습니다. 잘못된 페이지 계약, 반복 cursor, 100페이지 초과도 중단합니다. POST는 메모리의 모의 응답만 반환하며 실제 저장 서비스와 연결하지 않습니다.
## 05 외부 필드 매핑
node src/automation/batch.mjs automation/input/external-tasks.json
예상: successCount=1, failedCount=3, exit 1. id→sourceId, name→title, due_at→dueAt을 매핑합니다. 내부/외부 이름이 동시에 있으며 값이 다르면 추측하지 않고 거부합니다. title 숫자를 문자열로 바꾸지 않으며 잘못된 달력 날짜와 미정의 status도 거부합니다. rejected에 원본 비식별 입력과 field/reason을 남깁니다. API·파일·다음 Webhook 모두 동일 normalizeTask를 사용합니다.
## 06 로컬 Webhook
node src/automation/demo-webhook.mjs
같은 이벤트 2회 전달 결과 completed → duplicate. 처리 실패는 completed에 기록하지 않으므로 재시도가 가능하고, 진행 중 동일 eventId는 409입니다. 동일 eventId에 다른 내용도 409입니다. 테스트에서 실제 로컬 POST·인증오류·잘못된 JSON 계약·실패 후 재시도·동시 수신을 확인합니다.
local-webhook-only는 공개 교육 문자열이며 제공자 서명이 아닙니다. 실제 서비스 연결 때는 제공자 공식 raw body·서명 검증과 Secret 관리가 필요합니다. 현재 Map/Set은 프로세스 종료 시 사라집니다. 긴 작업은 수신 응답 전에 모두 처리하면 timeout 위험이 있으므로 운영에서는 영속 큐에 안전히 접수한 뒤 응답하는 별도 설계가 필요합니다. 이 예제는 짧은 로컬 처리 후 200을 반환합니다.
## 07 예약 시뮬레이션
node src/automation/demo-scheduler.mjs
고정 시각 2026-09-21T00:00:00Z는 Asia/Seoul 09시입니다. 동시에 수동 호출한 두 작업은 completed와 skipped_overlap이 됩니다. 기한 경과 Task를 같은 run()으로 처리하고 state/last-schedule-success.json에 마지막 성공을 기록합니다. isMissed는 마지막 성공 이후 27시간을 넘겼는지 점검하는 예시 기준입니다.
OS cron·작업 스케줄러·서비스 예약을 등록하지 않습니다. 0 9 * * *는 Asia/Seoul 시간대로 동작하는 스케줄러를 선택했을 때의 설계 예시입니다. 메모리 lock은 단일 프로세스 내부만 막으며 여러 서버·재시작을 보호하지 않습니다. 실행 완료 후 같은 시각을 다시 수동 호출하는 중복은 09의 영속 멱등성에서 다룹니다. DST 지역은 해당 스케줄러 정책을 별도로 확인합니다.
## 08 제한된 재시도
node src/automation/demo-retry.mjs
모의 HTTP 429 → 503 → 200, attempt 1/2의 retrying과 3의 completed, 조회 2건이 예상됩니다. 기본 최대3회, 총 대기5초, 지수 backoff+jitter를 적용합니다. Retry-After 초 또는 HTTP 날짜가 있으면 더 짧게 기다리지 않으며 예산을 초과하면 중단합니다. 요청별 timeout 2초와 총 대기 예산은 다른 제한입니다.
401/403/400은 자동 재시도하지 않습니다. GET만 getWithRetry를 사용하며 POST를 넣으면 거부합니다. 9편의 파일 멱등성이 외부 POST의 안전한 재시도를 보장하지 않습니다. 테스트의 가짜 sleep은 긴 지연을 없애지만 demo는 실제 짧은 지연을 사용합니다.
## 09 영속 상태와 재개
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo --fail-third
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo
처음 사용하는 resume-demo 폴더 기준 결과: 2성공/1실패(exit1) → 1성공/2skip(exit0) → 3skip(exit0). 각 명령은 별도 Node 프로세스이며 output/results에 최종3파일만 남습니다. 이미 실습한 폴더라면 첫 명령부터 skip이므로 새 실습 폴더명을 지정합니다. 삭제나 상태 초기화는 필요하지 않습니다.
key는 sourceId + normalize-store-v1의 해시입니다. 같은 key에 다른 정규화 결과가 오면 manual_review로 분리합니다. state는 processing/completed/failed, 결과파일은 수정 없이 생성합니다. 결과 저장 후 상태 완료 전 실패해도 기존 결과가 기대 바이트와 같으면 복구하고 다르면 중단합니다. 손상된 상태를 빈 상태로 취급하지 않습니다. completed인데 결과가 없으면 사람이 확인합니다.
이것은 단일 프로세스·로컬 파일 교육 예제이며 여러 프로세스가 같은 폴더에서 동시에 실행하면 안 됩니다. 파일 상태 기록 자체는 DB 트랜잭션이 아니므로 손상 시 자동 복구를 보장하지 않습니다. 운영 외부 API에 exactly-once를 보장하지 않으며 업로드 응답 유실은 실제 결과 조회 후 판정해야 합니다. 테스트는 결과 저장 뒤의 실패와 새 프로세스 재개를 검증합니다.
src/automation/api-client.mjs
export async function request(url, { token, method = 'GET', body, timeoutMs = 2000 } = {}) {
const target = new URL(url);
if (target.protocol !== 'http:' || target.hostname !== '127.0.0.1' || target.username || target.password) throw new Error('교육용 API는 http://127.0.0.1 주소만 허용합니다.');
const response = await fetch(target, {
method, redirect: 'error', signal: AbortSignal.timeout(timeoutMs),
headers: { authorization: `Bearer ${token ?? ''}`, 'content-type': 'application/json' },
...(body === undefined ? {} : { body: JSON.stringify(body) }),
});
if (!response.ok) throw Object.assign(new Error(`HTTP ${response.status}`), { status: response.status, retryAfter: response.headers.get('retry-after') });
return response.json();
}
export async function listTasks(baseURL, token, { get = request } = {}) {
const items = [], seen = new Set();
let cursor = null;
for (let page = 0; page < 100; page += 1) {
const url = new URL('/tasks', baseURL);
if (cursor !== null) url.searchParams.set('cursor', cursor);
const data = await get(url, { token });
if (!data || !Array.isArray(data.items) || !(data.nextCursor === null || typeof data.nextCursor === 'string')) throw new Error('페이지 응답 계약 오류');
items.push(...data.items);
if (data.nextCursor === null) return items;
if (!data.nextCursor || seen.has(data.nextCursor)) throw new Error('반복 cursor 중단');
seen.add(data.nextCursor);
cursor = data.nextCursor;
}
throw new Error('최대 페이지 수 초과');
}
src/automation/batch.mjs
import { extname } from 'node:path';
import { readJSON } from './read-json.mjs';
import { readCSV } from './read-csv.mjs';
import { run } from './run.mjs';
import { isMain } from './paths.mjs';
export async function batch(file, options) {
const ext = extname(String(file)).toLowerCase();
if (!['.json', '.csv'].includes(ext)) throw new Error('JSON 또는 CSV 경로가 필요합니다.');
return run(await (ext === '.csv' ? readCSV(file) : readJSON(file)), options);
}
if (isMain(import.meta.url)) {
try {
const result = await batch(process.argv[2] ?? new URL('../../automation/input/tasks.json', import.meta.url));
console.log(JSON.stringify(result, null, 2));
if (result.failedCount) process.exitCode = 1;
} catch { console.error('batch 읽기·파싱·저장 실패'); process.exitCode = 1; }
}
src/automation/demo-api.mjs
import { startMock } from './mock-api.mjs';
import { listTasks, request } from './api-client.mjs';
import { run } from './run.mjs';
const mock = await startMock();
try {
// 실제 credential이 아닌 로컬 모의 서버의 공개 교육 문자열입니다.
const token = 'local-training-only';
const tasks = await listTasks(mock.url, token);
console.log(JSON.stringify(await run(tasks, { trigger: 'mock-api' }), null, 2));
const created = await request(mock.url + '/tasks', { token, method: 'POST', body: { sourceId: 'sample-new', title: '모의 POST' } });
console.log('로컬 POST 응답:', created.sourceId);
} catch { console.error('모의 API 실습 실패'); process.exitCode = 1; }
finally { await mock.close(); }
src/automation/demo-retry.mjs
import { startMock } from './mock-api.mjs';
import { getWithRetry } from './retry.mjs';
const mock = await startMock({ statuses: [429, 503, 200] });
try {
const result = await getWithRetry(mock.url + '/tasks', { token: 'local-training-only' }, { onAttempt: (entry) => console.log(JSON.stringify(entry)) });
console.log('로컬 조회 결과:', result.items.length);
} catch { console.error('재시도 소진 또는 영구 오류'); process.exitCode = 1; }
finally { await mock.close(); }
src/automation/demo-scheduler.mjs
import { mkdir, writeFile } from 'node:fs/promises';
import { join } from 'node:path';
import { automationRoot } from './paths.mjs';
import { createScheduler, isKstNine } from './scheduler.mjs';
import { run } from './run.mjs';
const simulatedAt = '2026-09-21T00:00:00Z';
const tasks = [{ sourceId: 'overdue-1', title: '기한 경과 모의 작업', status: 'todo', dueAt: '2026-09-20T00:00:00Z' }];
const schedule = createScheduler((at) => run(tasks.filter((t) => t.status !== 'done' && t.dueAt && Date.parse(t.dueAt) < Date.parse(at)), { trigger: 'scheduled-simulation' }), async (log) => {
await mkdir(join(automationRoot, 'state'), { recursive: true });
await writeFile(join(automationRoot, 'state', 'last-schedule-success.json'), JSON.stringify(log, null, 2) + '\n');
});
try {
console.log('KST 09시:', isKstNine(simulatedAt));
console.log(await Promise.all([schedule(simulatedAt), schedule(simulatedAt)]));
} catch { console.error('예약 시뮬레이션 실패'); process.exitCode = 1; }
src/automation/demo-webhook.mjs
import { startWebhook } from './webhook.mjs';
const hook = await startWebhook();
try {
const event = { eventId: 'event-demo-1', eventType: 'task.created', data: { sourceId: 'sample-webhook', title: '모의 이벤트' } };
for (let i = 0; i < 2; i += 1) {
const response = await fetch(hook.url + '/automation/webhook', { method: 'POST', headers: { 'content-type': 'application/json', 'x-training-key': 'local-webhook-only' }, body: JSON.stringify(event) });
if (!response.ok) throw new Error('Webhook 실패');
console.log(await response.json());
}
} catch { console.error('로컬 Webhook 실습 실패'); process.exitCode = 1; }
finally { await hook.close(); }
src/automation/local-http.mjs
export function send(res, status, value, headers = {}) {
res.writeHead(status, { 'content-type': 'application/json; charset=utf-8', ...headers });
res.end(JSON.stringify(value));
}
export async function readBody(req) {
let size = 0;
const parts = [];
for await (const part of req) {
size += part.length;
if (size > 65536) throw new Error('body too large');
parts.push(part);
}
return JSON.parse(Buffer.concat(parts).toString('utf8'));
}
export async function listenLocal(server) {
await new Promise((resolve, reject) => {
server.once('error', reject);
server.listen(0, '127.0.0.1', resolve);
});
return {
url: `http://127.0.0.1:${server.address().port}`,
close: () => new Promise((resolve, reject) => {
server.close((error) => error ? reject(error) : resolve());
server.closeAllConnections();
}),
};
}
src/automation/mock-api.mjs
import { createServer } from 'node:http';
import { setTimeout as sleep } from 'node:timers/promises';
import { send, readBody, listenLocal } from './local-http.mjs';
import { normalizeTask } from './normalize-task.mjs';
export async function startMock({ repeatCursor = false, delay = 0, statuses = [] } = {}) {
let calls = 0;
const tasks = [1, 2, 3].map((n) => ({ sourceId: `sample-${n}`, title: `모의 작업 ${n}`, status: 'todo', dueAt: '2026-09-20T00:00:00Z' }));
const server = createServer(async (req, res) => {
try {
if (delay) await sleep(delay);
if (req.headers.authorization !== 'Bearer local-training-only') return send(res, 401, { error: 'unauthorized' });
const status = statuses[calls++] ?? 200;
if (status !== 200) return send(res, status, { error: 'simulated' }, { 'retry-after': '0' });
const url = new URL(req.url, 'http://127.0.0.1');
if (url.pathname !== '/tasks') return send(res, 404, { error: 'not_found' });
if (req.method === 'GET') {
const cursor = url.searchParams.get('cursor');
if (cursor !== null && cursor !== 'page-2') return send(res, 400, { error: 'cursor' });
return send(res, 200, { items: cursor ? tasks.slice(2) : tasks.slice(0, 2), nextCursor: repeatCursor || !cursor ? 'page-2' : null });
}
if (req.method === 'POST') return send(res, 201, normalizeTask(await readBody(req)));
return send(res, 405, { error: 'method' });
} catch { if (!res.headersSent) send(res, 400, { error: 'invalid_payload' }); }
});
return { ...await listenLocal(server), calls: () => calls };
}
src/automation/normalize-task.mjs
export function invalid(field, reason) {
return Object.assign(new Error(reason), { field });
}
export function normalizeTask(input) {
if (!input || typeof input !== 'object' || Array.isArray(input)) throw invalid('record', '객체가 필요합니다.');
const mapped = { ...input };
for (const [external, internal] of [['id', 'sourceId'], ['name', 'title'], ['due_at', 'dueAt']]) {
if (external in input) {
if (internal in input && input[internal] !== input[external]) throw invalid(internal, '서로 충돌하는 필드가 있습니다.');
mapped[internal] = input[external];
}
}
const { sourceId, title, status = 'todo', dueAt = null } = mapped;
if (typeof sourceId !== 'string' || !/^[a-zA-Z0-9_-]{1,80}$/.test(sourceId)) throw invalid('sourceId', '영문·숫자·밑줄·하이픈 식별자가 필요합니다.');
if (typeof title !== 'string' || !title.trim() || title.length > 200) throw invalid('title', '1~200자 제목이 필요합니다.');
if (!['todo', 'doing', 'done'].includes(status)) throw invalid('status', 'todo, doing, done 중 하나여야 합니다.');
if (dueAt !== null && (typeof dueAt !== 'string' || !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/.test(dueAt) || !Number.isFinite(Date.parse(dueAt)) || new Date(dueAt).toISOString() !== dueAt.replace(/Z$/, dueAt.includes('.') ? 'Z' : '.000Z'))) {
throw invalid('dueAt', '유효한 UTC ISO 날짜 또는 null이 필요합니다.');
}
return { sourceId, title: title.trim(), status, dueAt };
}
src/automation/paths.mjs
import { fileURLToPath } from 'node:url';
import { resolve } from 'node:path';
export const automationRoot = fileURLToPath(new URL('../../automation/', import.meta.url));
export function isMain(url) {
return Boolean(process.argv[1]) && resolve(process.argv[1]) === fileURLToPath(url);
}
src/automation/read-csv.mjs
import { readFile } from 'node:fs/promises';
export function parseCSV(raw) {
if (raw.includes('"')) throw new Error('이 예제는 따옴표·quoted comma·줄바꿈 필드를 지원하지 않습니다.');
const lines = raw.replace(/^\uFEFF/, '').replace(/\r\n/g, '\n').trimEnd().split('\n');
const headers = lines.shift().split(',').map((v) => v.trim());
const allowed = ['sourceId', 'title', 'status', 'dueAt'];
if (!headers.includes('sourceId') || !headers.includes('title') || new Set(headers).size !== headers.length || headers.some((h) => !allowed.includes(h))) throw new Error('CSV 헤더 오류');
return lines.map((line, index) => {
const values = line.split(',');
if (values.length !== headers.length) throw new Error(`CSV ${index + 2}행 필드 수 오류`);
return Object.fromEntries(headers.map((key, i) => [key, key === 'dueAt' && !values[i] ? null : values[i]]));
});
}
export async function readCSV(file) { return parseCSV(await readFile(file, 'utf8')); }
src/automation/read-json.mjs
import { readFile } from 'node:fs/promises';
export async function readJSON(file) {
let value;
try { value = JSON.parse((await readFile(file, 'utf8')).replace(/^\uFEFF/, '')); }
catch { throw new Error('JSON 읽기 또는 파싱 실패'); }
if (!Array.isArray(value)) throw new Error('JSON 배열이 필요합니다.');
return value;
}
src/automation/ready.mjs
const major = Number(process.versions.node.split('.')[0]);
if (major < 24) {
console.error('Node.js 24 이상이 필요합니다.');
process.exitCode = 1;
} else {
console.log('TaskFlow Ops Automation 준비 완료: 로컬 비식별 실습만 실행합니다.');
}
src/automation/resume.mjs
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { createHash, randomUUID } from 'node:crypto';
import { join } from 'node:path';
import { normalizeTask } from './normalize-task.mjs';
import { readJSON } from './read-json.mjs';
import { automationRoot, isMain } from './paths.mjs';
const json = (value) => JSON.stringify(value, null, 2) + '\n';
const hash = (value) => createHash('sha256').update(value).digest('hex');
async function existing(file) {
try { return await readFile(file, 'utf8'); }
catch (error) { if (error.code === 'ENOENT') return null; throw error; }
}
export async function resumeBatch(inputs, { root = automationRoot, failBefore, failAfter, trigger = 'manual-resume' } = {}) {
if (!Array.isArray(inputs)) throw new Error('배열이 필요합니다.');
for (const dir of ['state', 'output/results', 'logs']) await mkdir(join(root, dir), { recursive: true });
const log = { runId: randomUUID(), trigger, startedAt: new Date().toISOString(), inputCount: inputs.length, successCount: 0, skippedCount: 0, recoveredCount: 0, failedCount: 0, items: [] };
for (const [index, input] of inputs.entries()) {
let stateFile, state, task, key;
let canSaveState = false;
try {
task = normalizeTask(input);
key = hash(JSON.stringify([task.sourceId, 'normalize-store-v1']));
stateFile = join(root, 'state', key + '.json');
const resultFile = join(root, 'output', 'results', key + '.json');
const expected = json(task), inputHash = hash(expected);
const rawState = await existing(stateFile);
state = rawState === null ? null : JSON.parse(rawState);
if (rawState !== null && (!state || typeof state !== 'object' || Array.isArray(state) || state.inputHash !== inputHash || state.key !== key || !['processing', 'completed', 'failed'].includes(state.status) || !Number.isInteger(state.attempt) || state.attempt < 1)) throw new Error('state_conflict');
const result = await existing(resultFile);
if (result !== null && result !== expected) throw new Error('result_conflict');
if (state?.status === 'completed') {
if (result === null) throw new Error('completed_result_missing');
log.skippedCount += 1;
log.items.push({ sourceId: task.sourceId, key, status: 'skipped', attempt: state.attempt });
continue;
}
state = { key, sourceId: task.sourceId, inputHash, status: 'processing', attempt: (state?.attempt ?? 0) + 1, runId: log.runId, startedAt: new Date().toISOString() };
canSaveState = true;
await writeFile(stateFile, json(state));
if (failBefore === task.sourceId) throw new Error('simulated_before_result');
if (result === null) await writeFile(resultFile, expected, { flag: 'wx' });
else log.recoveredCount += 1;
if (failAfter === task.sourceId) throw new Error('simulated_after_result');
state.status = 'completed';
state.completedAt = new Date().toISOString();
state.checkpoint = 'local-result-saved';
await writeFile(stateFile, json(state));
log.successCount += 1;
log.items.push({ sourceId: task.sourceId, key, status: 'completed', attempt: state.attempt });
} catch (error) {
// 손상되거나 충돌하는 state는 덮어쓰지 않고 수동 확인 대상으로 남깁니다.
if (canSaveState) {
state.status = 'failed';
state.errorClass = error.message.startsWith('simulated_') ? error.message : 'storage_failure';
await writeFile(stateFile, json(state));
}
log.failedCount += 1;
log.items.push({ row: index + 1, sourceId: task?.sourceId ?? input?.sourceId ?? input?.id ?? null, key: key ?? null, status: 'manual_review', attempt: state?.attempt ?? 0, errorClass: error.field ? 'validation' : canSaveState ? 'processing_or_storage' : 'state_or_result_conflict', nextAction: '입력과 기존 결과를 확인한 뒤 같은 명령으로 재개' });
}
}
log.finishedAt = new Date().toISOString();
log.status = log.failedCount ? (log.successCount + log.skippedCount ? 'partial_failure' : 'failed') : 'completed';
await writeFile(join(root, 'logs', log.runId + '.json'), json(log), { flag: 'wx' });
return log;
}
if (isMain(import.meta.url)) {
try {
const file = process.argv[2] ?? new URL('../../automation/input/resume-tasks.json', import.meta.url);
const inputs = await readJSON(file);
const result = await resumeBatch(inputs, { root: process.argv[3] ?? automationRoot, ...(process.argv[4] === '--fail-third' ? { failBefore: inputs[2]?.sourceId } : {}) });
console.log(JSON.stringify(result, null, 2));
if (result.failedCount) process.exitCode = 1;
} catch { console.error('입력 또는 상태 저장 실패. 기존 파일을 보존하고 수동 확인하세요.'); process.exitCode = 1; }
}
src/automation/retry.mjs
import { setTimeout as defaultSleep } from 'node:timers/promises';
import { request } from './api-client.mjs';
export async function retry(fn, { maxAttempts = 3, baseMs = 100, maxWaitMs = 5000, sleep = defaultSleep, random = Math.random, onAttempt = () => {} } = {}) {
if (!Number.isInteger(maxAttempts) || maxAttempts < 1 || maxAttempts > 10 || !Number.isFinite(baseMs) || baseMs < 0 || !Number.isFinite(maxWaitMs) || maxWaitMs < 0) throw new Error('재시도 설정 오류');
let waited = 0;
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
try {
const value = await fn(attempt);
onAttempt({ attempt, status: 'completed', delayMs: 0 });
return value;
} catch (error) {
const transient = [408, 429, 500, 502, 503, 504].includes(error.status) || error.name === 'TimeoutError' || (error instanceof TypeError && error.message === 'fetch failed');
if (!transient || attempt === maxAttempts) {
onAttempt({ attempt, status: 'failed', httpStatus: error.status ?? null, delayMs: 0 });
throw error;
}
let delay = baseMs * 2 ** (attempt - 1) + Math.floor(random() * baseMs);
if (error.retryAfter !== null && error.retryAfter !== undefined) {
const seconds = Number(error.retryAfter);
const serverDelay = Number.isFinite(seconds) && seconds >= 0 ? seconds * 1000 : Date.parse(error.retryAfter) - Date.now();
if (Number.isFinite(serverDelay)) delay = Math.max(delay, serverDelay, 0);
}
if (waited + delay > maxWaitMs) throw Object.assign(new Error('총 재시도 대기 예산 초과'), { status: error.status });
onAttempt({ attempt, status: 'retrying', httpStatus: error.status ?? null, delayMs: delay });
await sleep(delay);
waited += delay;
}
}
}
export async function getWithRetry(url, options = {}, policy = {}) {
if (options.method && options.method !== 'GET') throw new Error('자동 재시도는 GET만 허용합니다.');
return retry(() => request(url, { ...options, method: 'GET' }), policy);
}
src/automation/run.mjs
import { mkdir, readFile, writeFile } from 'node:fs/promises';
import { randomUUID } from 'node:crypto';
import { join } from 'node:path';
import { normalizeTask } from './normalize-task.mjs';
import { automationRoot, isMain } from './paths.mjs';
export async function run(inputs, { root = automationRoot, trigger = 'manual' } = {}) {
if (!Array.isArray(inputs)) throw new Error('입력 배열이 필요합니다.');
const runId = randomUUID();
const startedAt = new Date().toISOString();
const valid = [], rejected = [], seen = new Set();
let skippedCount = 0;
inputs.forEach((input, index) => {
try {
const task = normalizeTask(input);
if (seen.has(task.sourceId)) { skippedCount += 1; return; }
seen.add(task.sourceId);
valid.push(task);
} catch (error) {
// 원본은 비식별 교육 fixture만 저장합니다. 운영 개인정보는 별도 보호가 필요합니다.
rejected.push({ row: index + 1, sourceId: input?.sourceId ?? input?.id ?? null, field: error.field ?? 'record', reason: error.message, input });
}
});
const output = join(root, 'output', runId);
await mkdir(output, { recursive: true });
await mkdir(join(root, 'logs'), { recursive: true });
const json = (value) => JSON.stringify(value, null, 2) + '\n';
await writeFile(join(output, 'valid.json'), json(valid), { flag: 'wx' });
await writeFile(join(output, 'rejected.json'), json(rejected), { flag: 'wx' });
const log = { runId, trigger, startedAt, finishedAt: new Date().toISOString(), inputCount: inputs.length, successCount: valid.length, failedCount: rejected.length, skippedCount, status: rejected.length ? (valid.length ? 'partial_failure' : 'failed') : 'completed', output };
await writeFile(join(root, 'logs', runId + '.json'), json(log), { flag: 'wx' });
return log;
}
if (isMain(import.meta.url)) {
try {
const file = process.argv[2] ?? join(automationRoot, 'input', 'task.json');
const value = JSON.parse(await readFile(file, 'utf8'));
const result = await run(Array.isArray(value) ? value : [value]);
console.log(JSON.stringify(result, null, 2));
if (result.failedCount) process.exitCode = 1;
} catch { console.error('입력 읽기·파싱 또는 결과 저장 실패. 경로와 형식을 확인하세요.'); process.exitCode = 1; }
}
src/automation/scheduler.mjs
import { randomUUID } from 'node:crypto';
export function isKstNine(iso) {
const parts = new Intl.DateTimeFormat('en-GB', { timeZone: 'Asia/Seoul', hour: '2-digit', minute: '2-digit', hourCycle: 'h23' }).formatToParts(new Date(iso));
return parts.find((p) => p.type === 'hour').value === '09' && parts.find((p) => p.type === 'minute').value === '00';
}
export function isMissed(lastSuccess, now) {
return !lastSuccess || Date.parse(now) - Date.parse(lastSuccess) > 27 * 60 * 60 * 1000;
}
export function createScheduler(job, recordSuccess = async () => {}) {
let running = false;
return async (scheduledAt) => {
if (!Number.isFinite(Date.parse(scheduledAt))) throw new Error('scheduledAt 날짜 오류');
if (running) return { status: 'skipped_overlap', scheduledAt };
running = true;
const runId = randomUUID(), startedAt = new Date().toISOString();
try {
const result = await job(scheduledAt);
if (result?.failedCount) throw new Error('예약 트리거는 실행됐지만 업무 처리 일부가 실패했습니다.');
const log = { runId, scheduledAt, startedAt, finishedAt: new Date().toISOString(), timezone: 'Asia/Seoul', status: 'completed' };
await recordSuccess(log);
return log;
} finally { running = false; }
};
}
src/automation/webhook.mjs
import { createServer } from 'node:http';
import { normalizeTask } from './normalize-task.mjs';
import { run } from './run.mjs';
import { listenLocal, readBody, send } from './local-http.mjs';
export function createEventHandler(processTask) {
const completed = new Map(), inflight = new Set();
return async (event) => {
if (!event || typeof event.eventId !== 'string' || !/^[a-zA-Z0-9_-]{1,80}$/.test(event.eventId) || event.eventType !== 'task.created') throw Object.assign(new Error('이벤트 계약 오류'), { status: 400 });
const task = normalizeTask(event.data);
const fingerprint = JSON.stringify(task);
if (completed.has(event.eventId)) {
if (completed.get(event.eventId) !== fingerprint) throw Object.assign(new Error('eventId 내용 충돌'), { status: 409 });
return { eventId: event.eventId, status: 'duplicate' };
}
if (inflight.has(event.eventId)) throw Object.assign(new Error('처리 중'), { status: 409 });
inflight.add(event.eventId);
try {
await processTask(task, event.eventId);
completed.set(event.eventId, fingerprint);
return { eventId: event.eventId, status: 'completed' };
} finally { inflight.delete(event.eventId); }
};
}
export async function startWebhook({ processTask = (task) => run([task], { trigger: 'webhook' }) } = {}) {
const handle = createEventHandler(processTask);
const server = createServer(async (req, res) => {
if (req.method !== 'POST' || req.url !== '/automation/webhook') return send(res, 404, { error: 'not_found' });
if (req.headers['x-training-key'] !== 'local-webhook-only') return send(res, 401, { error: 'unauthorized' });
let event;
try { event = await readBody(req); }
catch { return send(res, 400, { error: 'invalid_json_or_size' }); }
try { send(res, 200, await handle(event)); }
catch (error) { send(res, error.status ?? (error.field ? 400 : 500), { error: error.status === 409 ? 'conflict_or_inflight' : 'processing_failed' }); }
});
return listenLocal(server);
}
tests/api.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { startMock } from '../src/automation/mock-api.mjs';
import { request, listTasks } from '../src/automation/api-client.mjs';
test('localhost 인증·페이지 조회·POST를 검증한다', async () => {
const mock = await startMock();
try {
const tasks = await listTasks(mock.url, 'local-training-only');
assert.equal(tasks.length, 3);
const created = await request(mock.url + '/tasks', { token: 'local-training-only', method: 'POST', body: { sourceId: 'sample-new', title: '모의 생성' } });
assert.equal(created.sourceId, 'sample-new');
await assert.rejects(request(mock.url + '/tasks', { token: 'wrong' }), (e) => e.status === 401);
await assert.rejects(request(mock.url + '/missing', { token: 'local-training-only' }), (e) => e.status === 404);
} finally { await mock.close(); }
});
test('외부 주소·반복 cursor·timeout을 차단한다', async () => {
await assert.rejects(request('https://example.com/tasks', {}), /127\.0\.0\.1/);
const mock = await startMock({ repeatCursor: true });
try { await assert.rejects(listTasks(mock.url, 'local-training-only'), /cursor/); }
finally { await mock.close(); }
const slow = await startMock({ delay: 80 });
try { await assert.rejects(request(slow.url + '/tasks', { token: 'local-training-only', timeoutMs: 5 })); }
finally { await slow.close(); }
});
tests/core.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtemp, readFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { normalizeTask } from '../src/automation/normalize-task.mjs';
import { run } from '../src/automation/run.mjs';
test('sourceId와 제목을 보존하고 공백을 정규화한다', () => {
assert.deepEqual(normalizeTask({ sourceId: 'sample-1', title: ' 점검 ' }),
{ sourceId: 'sample-1', title: '점검', status: 'todo', dueAt: null });
});
test('빈 제목과 잘못된 식별자를 거부한다', () => {
assert.throws(() => normalizeTask({ sourceId: 'sample-1', title: ' ' }));
assert.throws(() => normalizeTask({ sourceId: '../unsafe', title: '점검' }));
});
test('한 건 처리와 동일 입력의 별도 실행 기록을 확인한다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-core-'));
const input = { sourceId: 'sample-1', title: ' 점검 ' };
const first = await run([input], { root });
const second = await run([input], { root });
assert.equal(first.successCount, 1);
assert.notEqual(first.runId, second.runId);
const output = JSON.parse(await readFile(join(root, 'output', first.runId, 'valid.json'), 'utf8'));
assert.equal(output[0].sourceId, 'sample-1');
});
test('검증 실패는 rejected와 실패 상태로 남는다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-invalid-'));
const result = await run([{ sourceId: 'sample-2', title: '' }], { root });
assert.equal(result.status, 'failed');
assert.equal(result.failedCount, 1);
const rejected = JSON.parse(await readFile(join(root, 'output', result.runId, 'rejected.json'), 'utf8'));
assert.equal(rejected[0].row, 1);
assert.equal(rejected[0].field, 'title');
});
tests/files.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { parseCSV } from '../src/automation/read-csv.mjs';
import { readJSON } from '../src/automation/read-json.mjs';
import { run } from '../src/automation/run.mjs';
import { mkdtemp } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
test('통제된 UTF-8 CSV를 읽는다', () => {
assert.deepEqual(parseCSV('sourceId,title,status,dueAt\r\nsample-1,한글,todo,\r\n'), [{ sourceId: 'sample-1', title: '한글', status: 'todo', dueAt: null }]);
});
test('따옴표 CSV와 필드 수·중복 헤더 오류를 명시 거부한다', () => {
for (const raw of ['sourceId,title\na,"b,c"', 'sourceId,title\na,b,c', 'sourceId,sourceId\na,b']) assert.throws(() => parseCSV(raw));
});
test('JSON과 CSV의 동일 모델, 잘못된 행과 중복 분리', async () => {
const input = await readJSON(new URL('../automation/input/tasks.json', import.meta.url));
const root = await mkdtemp(join(tmpdir(), 'taskflow-batch-'));
const result = await run(input, { root });
assert.equal(result.successCount, 2);
assert.equal(result.failedCount, 1);
assert.equal(result.skippedCount, 1);
assert.equal(result.status, 'partial_failure');
});
tests/mapping.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { normalizeTask } from '../src/automation/normalize-task.mjs';
test('외부 id/name/due_at를 내부 계약으로 매핑한다', () => {
assert.deepEqual(normalizeTask({ id: 'external-1', name: ' 점검 ', due_at: '2026-09-21T00:00:00Z' }), { sourceId: 'external-1', title: '점검', status: 'todo', dueAt: '2026-09-21T00:00:00Z' });
});
test('모호한 중복 필드, 잘못된 날짜·상태·타입을 거부한다', () => {
for (const patch of [{ id: 'other' }, { dueAt: '2026-02-30T00:00:00Z' }, { status: 'unknown' }, { title: 12 }]) {
assert.throws(() => normalizeTask({ sourceId: 'sample-1', title: '점검', ...patch }));
}
});
tests/resume.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { mkdtemp, readdir, readFile, writeFile } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { fileURLToPath } from 'node:url';
import { resumeBatch } from '../src/automation/resume.mjs';
const inputs = [1, 2, 3].map((n) => ({ sourceId: `sample-${n}`, title: `작업 ${n}` }));
test('부분 실패 후 재개·중복·동일 키 내용 충돌', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-resume-'));
const first = await resumeBatch(inputs, { root, failBefore: 'sample-3' });
assert.equal(first.successCount, 2);
assert.equal(first.failedCount, 1);
const next = await resumeBatch(inputs, { root });
assert.equal(next.successCount, 1);
assert.equal(next.skippedCount, 2);
assert.equal((await resumeBatch(inputs, { root })).skippedCount, 3);
const changed = await resumeBatch([{ sourceId: 'sample-1', title: '다른 내용' }], { root });
assert.equal(changed.failedCount, 1);
assert.equal((await readdir(join(root, 'output', 'results'))).length, 3);
});
test('결과 저장 이후 체크포인트 실패를 결과 대조로 복구한다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-window-'));
assert.equal((await resumeBatch([inputs[0]], { root, failAfter: 'sample-1' })).failedCount, 1);
const next = await resumeBatch([inputs[0]], { root });
assert.equal(next.successCount, 1);
assert.equal(next.recoveredCount, 1);
assert.equal((await readdir(join(root, 'output', 'results'))).length, 1);
});
test('손상된 체크포인트는 조용히 초기화하지 않는다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-corrupt-'));
await resumeBatch([inputs[0]], { root });
const state = join(root, 'state', (await readdir(join(root, 'state')))[0]);
await writeFile(state, '{broken');
assert.equal((await resumeBatch([inputs[0]], { root })).failedCount, 1);
assert.equal(await readFile(state, 'utf8'), '{broken');
});
test('새 Node 프로세스가 파일 상태를 읽어 재개한다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-process-'));
const input = join(root, 'input.json');
await writeFile(input, JSON.stringify(inputs));
const cli = fileURLToPath(new URL('../src/automation/resume.mjs', import.meta.url));
const exec = promisify(execFile);
await assert.rejects(exec(process.execPath, [cli, input, root, '--fail-third']), (e) => e.code === 1);
const result = JSON.parse((await exec(process.execPath, [cli, input, root])).stdout);
assert.equal(result.skippedCount, 2);
assert.equal(result.successCount, 1);
assert.equal(result.failedCount, 0);
});
test('JSON 원시값·배열 상태는 신규 상태로 덮어쓰지 않는다', async () => {
const root = await mkdtemp(join(tmpdir(), 'taskflow-state-type-'));
await resumeBatch([inputs[0]], { root });
const file = join(root, 'state', (await readdir(join(root, 'state')))[0]);
for (const value of [null, false, 0, '', [], {}]) {
const raw = JSON.stringify(value);
await writeFile(file, raw);
assert.equal((await resumeBatch([inputs[0]], { root })).failedCount, 1);
assert.equal(await readFile(file, 'utf8'), raw);
}
});
tests/retry.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { retry, getWithRetry } from '../src/automation/retry.mjs';
import { startMock } from '../src/automation/mock-api.mjs';
test('429·503은 제한된 지수 지연으로 재시도한다', async () => {
let attempts = 0;
const waits = [];
const value = await retry(async () => { if (++attempts < 3) throw Object.assign(new Error('temporary'), { status: 503 }); return 'ok'; }, { baseMs: 100, random: () => 0, sleep: async (ms) => waits.push(ms) });
assert.equal(value, 'ok');
assert.deepEqual(waits, [100, 200]);
});
test('401은 즉시 중단하며 attempts와 대기 예산을 넘지 않는다', async () => {
let count = 0;
await assert.rejects(retry(async () => { count += 1; throw Object.assign(new Error('auth'), { status: 401 }); }));
assert.equal(count, 1);
count = 0;
await assert.rejects(retry(async () => { count += 1; throw Object.assign(new Error('limit'), { status: 429, retryAfter: '100' }); }, { maxWaitMs: 1000 }));
assert.equal(count, 1);
count = 0;
await assert.rejects(retry(async () => { count += 1; throw Object.assign(new Error('limit'), { status: 503 }); }, { sleep: async () => {} }));
assert.equal(count, 3);
});
test('실제 모의 HTTP 429 다음 GET 성공, POST 자동재시도 금지', async () => {
const mock = await startMock({ statuses: [429, 200] });
try {
const data = await getWithRetry(mock.url + '/tasks', { token: 'local-training-only' }, { sleep: async () => {} });
assert.equal(data.items.length, 2);
assert.equal(mock.calls(), 2);
await assert.rejects(getWithRetry(mock.url + '/tasks', { method: 'POST' }), /GET/);
} finally { await mock.close(); }
});
tests/scheduler.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { createScheduler, isKstNine, isMissed } from '../src/automation/scheduler.mjs';
test('UTC와 KST 09시·누락 실행을 판정한다', () => {
assert.equal(isKstNine('2026-09-21T00:00:00Z'), true);
assert.equal(isKstNine('2026-09-21T09:00:00Z'), false);
assert.equal(isMissed('2026-09-19T00:00:00Z', '2026-09-21T00:00:00Z'), true);
});
test('중복 실행 skip과 실패 후 lock 해제를 확인한다', async () => {
let release;
const gate = new Promise((resolve) => { release = resolve; });
const schedule = createScheduler(() => gate);
const first = schedule('2026-09-21T00:00:00Z');
assert.equal((await schedule('2026-09-21T00:00:00Z')).status, 'skipped_overlap');
release({ failedCount: 0 });
assert.equal((await first).status, 'completed');
let calls = 0;
const fail = createScheduler(async () => { if (++calls === 1) throw new Error('simulated'); return { failedCount: 0 }; });
await assert.rejects(fail('2026-09-21T00:00:00Z'));
assert.equal((await fail('2026-09-21T00:00:00Z')).status, 'completed');
});
tests/webhook.test.mjs
import test from 'node:test';
import assert from 'node:assert/strict';
import { startWebhook, createEventHandler } from '../src/automation/webhook.mjs';
const event = { eventId: 'event-1', eventType: 'task.created', data: { sourceId: 'sample-1', title: '이벤트 점검' } };
test('실제 로컬 HTTP 수신·중복·인증·잘못된 payload', async () => {
let count = 0;
const hook = await startWebhook({ processTask: async () => { count += 1; } });
const post = (body, token = 'local-webhook-only') => fetch(hook.url + '/automation/webhook', { method: 'POST', headers: { 'content-type': 'application/json', 'x-training-key': token }, body: JSON.stringify(body) });
try {
assert.equal((await post(event)).status, 200);
assert.equal((await (await post(event)).json()).status, 'duplicate');
assert.equal(count, 1);
assert.equal((await post(event, 'wrong')).status, 401);
assert.equal((await post({ eventId: 'bad' })).status, 400);
} finally { await hook.close(); }
});
test('실패한 이벤트는 다시 처리하고 동시에 들어온 이벤트는 막는다', async () => {
let attempts = 0;
const handler = createEventHandler(async () => { if (++attempts === 1) throw new Error('simulated'); });
await assert.rejects(handler(event));
assert.equal((await handler(event)).status, 'completed');
assert.equal(attempts, 2);
await assert.rejects(handler({ ...event, data: { ...event.data, title: '충돌' } }), (e) => e.status === 409);
let release;
const gate = new Promise((resolve) => { release = resolve; });
const slow = createEventHandler(() => gate);
const first = slow(event);
await assert.rejects(slow(event), (e) => e.status === 409);
release();
await first;
});
독자 확인: 세 번째 항목에서 실패하면 어디서 재개할까요?
아래는 비식별 로컬 예제 또는 설계표를 직접 점검하기 위한 절차입니다. 이전에 기록된 로컬 테스트 결과와 구분해 읽으세요.
| 확인 항목 | 실습과 판단 기준 |
|---|---|
| 입력 | 새 빈 automation/resume-demo 폴더와 3개 비식별 Task를 사용합니다. 같은 sourceId·operation으로 재시도하며 내용이 바뀐 동일 키도 준비합니다. |
| 처리 | 문서의 --fail-third 실행 → 정상 재개 → 한 번 더 재실행 순서로 각 runId, item 상태, 결과 ID와 checkpoint를 비교합니다. 처음부터 같은 실습 폴더를 재사용하지 않습니다. |
| 예상 결과 | 새 폴더의 예상 집계는 2성공·1실패(exit 1) → 1성공·2skip → 3skip입니다. 기존 두 결과가 중복 생성되지 않았는지 출력 폴더와 상태 파일을 함께 비교합니다. |
| 같은 입력 재실행 | 완료 키는 skip하고 같은 키에 다른 입력이면 충돌로 처리합니다. 외부 부작용 성공과 완료 기록 사이에 중단된 crash window는 로컬 파일만으로 완전히 증명되지 않으므로 실제 결과 조회·서버 멱등성 키·원자적 저장 중 적합한 수단을 정해야 합니다. |
| 실패와 복구 | 손상된 checkpoint를 빈 상태로 덮어 처음부터 재실행하지 않습니다. 원본 상태를 보존하고 어떤 항목까지 성공했는지 확인한 후 누락된 항목만 재개합니다. |
| 다음 학습 | 10회차에서 같은 상태 계약을 n8n·Make의 단계별 워크플로 설계에 적용합니다. |
이번 편집의 검증 범위: 문서와 공개 본문 반영을 확인합니다. 이 회차의 자동화·외부 서비스·예약 작업을 이번 편집에서 새로 실행하지 않았으므로 실행 결과는 미확인입니다. 위 수치는 기존 예제의 예상값이며 실제 환경에서 직접 확인해야 합니다.
실행과 검증 결과
프로젝트 루트에서 각각 실행합니다. 설치 준비는 1회차를 참고하며 새 의존성은 없습니다.
node --test
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo --fail-third
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo
node src/automation/resume.mjs automation/input/resume-tasks.json automation/resume-demo
누적 테스트 23개가 통과했습니다. 새 Node 프로세스의 재개, 부분 실패, 결과 저장 뒤 상태 기록 실패, 동일 키 내용 충돌과 손상된 상태 보존을 검증했습니다. 처음 사용하는 resume-demo 폴더의 CLI 기대 결과는 2성공·1실패(exit 1) → 1성공·2skip → 3skip입니다. 이미 실행한 폴더라면 처음부터 skip될 수 있으므로 새 실습 폴더명을 사용합니다.
검증 결과는 Node.js 24와 비식별 로컬 모의 환경에 한정됩니다. 실제 외부 API·제공자 서명·운영 인증·예약 실행은 미검증입니다.
완료 체크포인트
automation-step-09-idempotency-resume
다음 학습
10회차 n8n과 Make.com으로 쇼츠 자동화하기 전 반드시 준비할 것들
실습 파일 다운로드
automation-step-09-idempotency-resume 전체 파일 ZIP
합성 데이터와 로컬 파일·localhost 모의 서버로 실행합니다. 외부 서비스 쓰기·실제 알림·운영 예약 작업은 실행하지 않습니다.
글 소개·학습 준비·전체 목차

업무·자동화 · 9 / 11
멱등성 체크포인트 재개 자동화 중복 실행과 부분 실패 복구
Webhook 재전송, retry, scheduler overlap처럼 같은 입력이 다시 들어와도 결과가 중복 생성되지 않도록 멱등성 기준을 설계합니다. 또한 batch 작업이 중간에 실패했을 때 처음부터 다시 실행하지 않고 checkpoint를 기준으로 안전하게 재개하는 방법을 학습합니다.
상단 실습은 이전 TaskFlow 파일 없이 시작합니다. 아래 누적 프로젝트에서는 같은 원리를 파일 입력·API·복구 기록에 차례로 적용합니다.
이번 글에서 배우는 내용
- 멱등성의 의미
- 중복 입력과 중복 결과의 차이
- idempotency key 설계 시 너무 넓거나 좁은 key의 문제
- “처리 시작 기록”과 “처리 완료 기록” 사이의 crash window
- checkpoint와 transaction이 같은 개념이 아닌 이유
- at-least-once delivery 환경에서 중복 방지가 필요한 이유
- exactly-once를 쉽게 보장할 수 있다고 표현하지 않음
선수 지식
입력 식별자, 결과 저장, 실패 재시도의 기본 개념이 필요합니다.
목차
이 글이 도움이 되었나요?
읽음 기록
읽음 — / 전체 11개 · 필수 11개
새 글 받아보기
RSS 리더에서 BlogFlow의 새 글을 확인할 수 있습니다.