이슈 파악하기

Test unsupported listen() in @fedify/cfworkersClosed

#881·dahlia·6일 전에 종료

Fedify에서 가장 처음 맡은 이슈는 #881입니다.

이슈의 내용은 다음과 같습니다.

출처

@fedify/cfworkers cannot implement listen() because Cloudflare Queues are consumed through Worker queue handlers. The adapter should keep failing clearly when that method is called. Add a test in packages/cfworkers/src/mod.test.ts that verifies WorkersMessageQueue.listen() rejects with a clear TypeError. Suggested check: mise run check-each cfworkers.

우선 이 이슈를 전체적인 이해를 하기 전에 Fedify의 메시지 큐 동작 방식을 알면 좋습니다.

메시지 큐 동작 방식

Fedify는 서버간 데이터를 주고 받을 때, 비동기 처리와 요청 재시도를 위해서 메시지 큐(message queue) 대기열 시스템을 사용하는데, 이 때 메시지 큐를 구현하기 위한 두 가지 메서드가 있습니다.

• enqueue() : 대기열에 메시지를 순차적으로 넣는 역할
• listen() : 대기열에서 메시지가 들어올 때까지 지켜보다가(listen) 메시지가 들어오면 꺼내서 처리하는 역할

일반적인 Node.js 서버 환경에서 사용하는 메시지 큐는 서버가 항상 켜져있는 상태로 동작합니다.
이 환경에서는 ‘폴링(Polling)‘이나 ‘스트리밍(Streaming)’ 방식으로 listen()을 구현합니다.

반면, cloudflare workers는 서버리스 환경 이기 때문에 항상 켜져있는 상태로 큐를 지켜보는 listen() 프로세스를 내부로 둘 수 없습니다.

cloudflare는 어떻게 listen()을 대체할까?

Cloudflare 플랫폼 인프라 자체가 그 역할을 대신합니다.

How Queues Works

Learn about Queues architecture including producers, consumers, and message lifecycle.

developers.cloudflare.com

동작 방식

1. 큐 감시 : cloudflare 시스템 자체가 내부적으로 큐를 감시합니다.
2. Worker 깨우기 : 큐에 메시지가 들어오면, Cloudflare가 잠들어있던 Worker를 강제로 깨웁니다.
3. 핸들러 호출 : 이때 Worker 내부의 queue라는 약속된 핸들러로 메시지를 던져줍니다.
4. 종료 : Worker는 받은 메시지를 처리하는 함수를 딱 한 번 실행하고, 일이 끝나면 다시 즉시 종료됩니다.

Fedify 문서에서도 동일한 내용을 언급하고 있습니다.

Cloudflare Queues API는 큐에서 메시지를 폴링하는 방법을 제공하지 않으므로, WorkersMessageQueue.listen() 메서드는 호출되면 항상 TypeError를 던집니다. 대신 Cloudflare worker에 queue() 메서드를 정의해야 하며, 큐에 새 메시지가 있을 때 Cloudflare Queues API가 이를 호출합니다.(번역된 내용)

원문 출처 :

Message queue | Fedify

Fedify docs

fedify.dev

이슈 내용 하나씩 뜯어보기

이제 이슈 내용을 하나씩 풀어서 이해해보겠습니다.

“@fedify/cfworkers cannot implement listen() because Cloudflare Queues are consumed through Worker queue handlers.”

→ cloudflare 환경의 특성상 시스템이 알아서 메시지를 주기 때문에, Fedify가 Cloudflare 용으로 만든 어댑터 패키지(@fedify/cfworkers) 안에는 개발자가 직접 호출하는 listen() 기능을 구현할 수 없습니다.

The adapter should keep failing clearly when that method is called.

→ 따라서 어답터는 해당 메서드가 호출될 때 명확하게 계속 실패해야 합니다.

Add a test in packages/cfworkers/src/mod.test.ts that verifies WorkersMessageQueue.listen() rejects with a clear TypeError.

→ 실제로 listen()을 실행했을 때 TypeError가 올바르게 발생하는지 확인하는 유닛 테스트 코드를 작성하라는 지시입니다.

Suggested check: mise run check-each cfworkers.

→ 테스트 코드를 전부 작성한 뒤, 형식·린트·타입이 깨지지 않았는지 검사하라는 가이드라인입니다. (이 명령이 실제로 무엇을 하는지는 뒤에서 다룹니다.)

기여 준비부터 테스트 코드 작성까지

사전 준비

기본적으로 기여를 하기 위한 과정을 익혀야 합니다.

우선 원본 레포에서 본인의 Github으로 fork하는 것부터 시작합니다.
그 이후에는 각 프로젝트의 CONTRIBUTING.md 파일을 참고하여 규칙에 따라 기여를 하면 됩니다.

Fedify 기여 전체 문서 가이드 :

fedify/CONTRIBUTING.md at main · fedify-dev/fedify

ActivityPub server framework in TypeScript. Contribute to fedify-dev/fedify development by creating an account on GitHub.

github.com

개발 환경

개발 환경 설정 관련 가이드 :

fedify/CONTRIBUTING.md at main · fedify-dev/fedify

ActivityPub server framework in TypeScript. Contribute to fedify-dev/fedify development by creating an account on GitHub.

github.com

개발 환경은 말 그대로 현재 소프트웨어를 개발, 테스트, 디버깅하기 위해 필요한 시스템적 환경을 의미합니다.
오픈소스같이 여러 개발자가 참여하는 프로젝트에서는 모든 팀원이 동일한 버전의 언어와 도구를 사용하는 표준화된 개발 환경을 유지하는 것이 매우 중요합니다.

Fedify의 개발 환경 관리 도구는 mise를 사용중입니다.

mise를 간단히 소개하자면, 기존의 nvm, pyenv, rbenv, asdf 등 여러 개로 쪼개져 있던 언어별 버전 관리 도구들을 하나로 통합한 일종의 올인원(All-in-one) 개발 환경 관리 도구입니다.
Rust로 작성되어 실행 속도가 매우 빠르고, 프로젝트 root에 mise.toml 설정 파일만 있으면 Node.js, Python, Go 등 다양한 프로그래밍 언어와 CLI 도구의 버전을 자동으로 스위칭하고, 환경 변수까지 한 번에 관리할 수 있습니다.

mise는 자동으로 설치되지는 않기 때문에 따로 설치 명령어를 실행해야 합니다.

[macOS 기준]

brew install mise    # mise가 없으면 설치
mise --version       # mise 버전 확인

이제 초기 설정을 해보겠습니다.

1. 초기 환경 설정

mise trust
mise install

• mise trust : 안전을 위해 이 프로젝트의 mise.toml 설정을 로컬 컴퓨터에서 신뢰하겠다고 승인하는 단계입니다.
• mise install : 이 명령어를 실행하면 post-install 훅이 작동하여 내부적으로 다음 작업들이 자동으로 수행됩니다.
  • mise deps : 코드 생성, 의존성 패키지 설치, 패키지 빌드 자동 진행
  • Sacho의 Git 머지 드라이버(Merge Drivers) 등록
  • Sacho의 커밋 훅(Commit Hooks) 설치
  • 기존 프리커밋 훅이 없는 경우, Git 프리커밋 훅(Pre-commit Hook) 자동 설치
• mise run hooks:install : 이미 프리커밋 훅이 있어서 자동 설치가 건너뛰어졌다면, 이 명령으로 직접 설치하거나 갱신할 수 있습니다.

2. 개발 워크플로우 실행 및 작업(Task) 관리

이 프로젝트의 모든 작업은 mise를 통해 제어가 되기 때문에 저장소 관리 작업을 수행할 때 npm이나 pnpm을 절대 직접 호출하면 안됩니다.

하단의 명령어를 통해 현재 프로젝트에서 사용할 수 있는 명령어나 특정 작업의 상세 내용을 확인할 수 있습니다.

mise tasks
mise tasks <task>

원하는 작업을 실행할 때는 mise run 뒤에 작업 이름을 붙여 사용합니다.

mise run check

Deno 테스트 실행 예시

mise run test:deno

제가 맡은 작업에서는 mise run check-each cfworkers 로 체크하면 됩니다.

테스트 코드 작성

테스트에서 가장 중요한 것은 테스트의 대상은 무엇이고, 그것의 의존성은 무엇인지 생각해야 합니다.

위에서 언급했듯이 WorkersMessageQueue.listen() 에서 TypeError가 발생하는 지에 대해 확인해야 합니다.

우선 /packages/cfworkers/src/mod.ts 로 가보겠습니다.

export class WorkersMessageQueue implements MessageQueue {
  #queue: Queue;
  #orderingKv?: WorkersKvNamespaceLike<string>;
  #orderingKeyPrefix: string;
  #orderingLockTtl: number;

  //...
  listen(
    // deno-lint-ignore no-explicit-any
    _handler: (message: any) => Promise<void> | void,
    _options?: MessageQueueListenOptions,
  ): Promise<void> {
    throw new TypeError("WorkersMessageQueue does not support listen().  " + "Use Federation.processQueuedTask() method instead.");
  }
}

여기서는 앞서 본대로 TypeError를 throw 해야하기 때문에 이를 검증해보도록 하겠습니다.

검증 방법

검증하는 방법은 mise run check-each cfworkers 가 있지만, 이는 실질적인 테스트 코드 검증 방식은 아닙니다.

이 명령어는 다음과 같이 명시되어 있습니다.

"check": "deno fmt --check && deno lint && deno check src/*.ts"

• deno fmt --check : 들여쓰기, 줄바꿈이 규칙에 맞는가?
• deno lint : 금지된 패턴을 사용하였는가?
• deno check : 타입이 맞는가?

즉, 셋 다 코드를 실행하지 않고, 파일을 읽고 분석하는 역할입니다.

반면, vitest는 실제로 실행하기 때문에 queue.listen()을 진짜 호출하고, 예외 케이스를 잡아서 instanceof TypeError인지 확인합니다.

check-each : 맞춤법 검사기 → ‘지구는 평평하다’도 맞춤법은 완벽합니다.
vitest : 직접 실험해보기

따라서 아래 명령어를 사용하여 이번 테스트 코드를 검증해보도록 하겠습니다.

cd packages/cfworkers
mise exec -- pnpm exec vitest run src/mod.test.ts -t "listen"

검증하기

아무것도 구현하지 않은 상태에서 검증해보겠습니다.

테스트가 하나도 실행되지 않은 채 통과로 끝납니다. 검증할 대상 자체가 없으니 당연한 결과입니다.

하지만 아래 처럼 TypeError가 아니라 일반 ‘Error’를 throw 한다면 어떨까요?

listen(
  // deno-lint-ignore no-explicit-any
  _handler: (message: any) => Promise<void> | void,
  _options?: MessageQueueListenOptions,
): Promise<void> {
  throw new Error(
    "WorkersMessageQueue does not support listen().  " +
      "Use Federation.processQueuedTask() method instead.",
  );
}

우선 mise run check-each cfworkers 로 mise의 문법 검사를 해주었습니다.

위 케이스와 마찬가지로 동일하게 정상적으로 체크가 완료됩니다.

여기서 의문점이 들 수 있습니다. ‘아까 deno check 에서 타입이 맞지 않은 경우를 검사한다고 했는데, TypeError → Error로 변경한 것은 왜 오류로 받아들여지지 않을까요?’

이는 TypeScript에는 ‘함수가 어떤 예외를 던지는 지’에 대한 개념이 없기 때문입니다.

Java에는 throws IOException 같은 체크 예외들이 있어서 컴파일러가 예외를 추적합니다. 하지만 TypeScript에는 그런 문법이 없습니다. 함수의 타입은 (): Promise<void> 하나뿐이고, 안에서 뭘 던지든 타입은 바뀌지 않습니다.

이어서 vitest도 진행해보았습니다.

여기서도 실행된 테스트가 없어서 그냥 통과로 끝납니다. 검증할 코드가 없으니 TypeError든 Error든 결과가 같습니다.
즉, 이걸 구분해내는 게 이번 작업의 목표입니다.

테스트 코드

// packages/cfworkers/src/mod.test.ts

describe("WorkersMessageQueue", () => {
  // listen() throws before touching the queue, so no methods are needed.
  const mockQueue = {} as unknown as Queue;

  it("listen() throws TypeError", () => {
    const queue = new WorkersMessageQueue(mockQueue);
    expect(() => queue.listen(() => {})).toThrow(TypeError);
    expect(() => queue.listen(() => {})).toThrow("WorkersMessageQueue does not support listen().  " + "Use Federation.processQueuedTask() method instead.");
  });
});

테스트 목적과 동작

• 목적 : WorkersMessageQueue.listen() 메서드가 지원되지 않음을 보장합니다.
• 동작 : .listen() 메서드를 호출하면 다른 대기열(Queue) 작업이 시작되기 전에 즉시 에러가 발생합니다.

코드 세부 분석

• Mock 객체 생성 : mockQueue를 비어있는 객체로 선언하여 실제 큐의 복잡한 메서드들을 구현하지 않고 Mock 데이터로 전달합니다.
• 인스턴스 생성 : new WorkersMessageQueue(mockQueue)를 통해 테스트할 객체를 만듭니다.
• 에러 검증 : expect(…).toThrow()를 사용하여 메서드 실행 시 TypeError가 발생하는지 확인합니다.
• 메시지 검증 : 예외가 발생할 때 “WorkersMessageQueue does not support listen(). Use Federation.processQueuedTask() method instead.”가 포함되는지 확인합니다. toThrow(문자열) 은 부분 일치라, 뒤쪽 안내 문구까지 넣어두면 누군가 그 안내를 지웠을 때 테스트가 잡아냅니다.

마찬가지로 TypeError를 Error로 변경하고, vitest로 직접 검증해보았습니다.

이번에는 테스트가 실패합니다. toThrow(TypeError) 단언이 걸리고, vitest가 실제로 던져진 건 Error라고 알려줍니다. 테스트 코드가 없을 때는 아무 일도 없었지만, 이제는 구현이 계약을 어기는 순간 바로 잡아내게 됩니다.

멘토 리뷰와 회고

멘토분들의 리뷰

이제 실질적인 테스트 코드 구현을 마치고, Matrix(OSSCA Fedify 멘토/멘티용 커뮤니티)에 리뷰 요청을 하였습니다.

가장 첫 번째 리뷰는 이재열 멘토님의 리뷰였습니다.

리뷰를 받고, 출처 명시를 위해 cloudflare 공식 문서를 활용하여 해당 자료를 조사하였습니다.

자료를 조사하면서 오해의 소지가 있을 만한 사실을 확인하였습니다.

PR 본문에는 ‘Cloudflare Queues는 폴링(polling) API 방식이 아닌, Worker Queue 핸들러를 통해 처리된다.‘라고 명시를 했는데, ‘rather than a polling API’라는 표현이 자칫 Cloudflare Queues는 polling 방식을 사용하지 않는다고 오해할 여지가 있습니다.

Pull consumers

Pull messages from a Cloudflare Queue over HTTP from any environment or language.

developers.cloudflare.com

공식 문서에 따르면,

A pull-based consumer allows you to pull from a queue over HTTP from any environment and/or programming language outside of Cloudflare Workers.

Cloudflare에는 푸시(push) 방식의 consumer worker(queue() 핸들러)도 있지만, 풀(pull) 방식의 HTTP 풀 consumer도 존재합니다. 즉, 폴링(polling) API가 아예 없는 것은 아닙니다.

다만 여기서 사용할 수 없는 것인데, 폴링 API는 워커(Workers) 외부에서 실행되며, 동일한 큐에서 푸시 방식 consumer도 병행할 수 없기 때문입니다. 결국 PR 본문의 스코프를 워커 내부로 한정하도록 수정하였습니다.

수정된 PR 본문

WorkersMessageQueue cannot implement listen() because the queue() handler is the only way to consume a queue inside a Worker, and Cloudflare invokes it rather than exposing anything the adapter can poll:

A consumer Worker, which is push-based: the Worker is invoked when the queue has messages to deliver.

A pull-based HTTP pull consumer does exist, but it runs “outside of Cloudflare Workers” and cannot coexist with a push-based consumer on the same queue, so it is not available here. The adapter throws a TypeError instead, which is the behavior documented in docs/manual/mq.md.

마침내 머지에 성공하여 fedify #1001에 첫 기여에 성공하였습니다.

Add a unit test for WorkersMessageQueue.listen()Merged

PR 1001·seongmin36·6일 전에 머지

후기

이슈 문구와 실제 구현이 어긋난 지점

이슈에는 listen() 이 “rejects with a clear TypeError”라고 적혀 있습니다. rejects는 Promise가 거부된다는 뜻이니, 처음엔 이렇게 쓰려고 했습니다.

await expect(queue.listen(() => {})).rejects.toThrow(TypeError);

그런데 이 코드는 동작하지 않습니다. 구현을 다시 보면 listen() 은 async 함수가 아닙니다.

listen(
    _handler: (message: any) => Promise<void> | void,
    _options?: MessageQueueListenOptions
): Promise<void> {
    throw new TypeError(...);
}

반환 타입이 Promise<void>일 뿐, 본문은 동기적으로 throw합니다. queue.listen(...)은 거부된 Promise를 돌려주는 게 아니라 호출하는 그 자리에서 예외를 던집니다. 그래서 expect(queue.listen(…)) 처럼 쓰면 ‘expect’에 인자를 넘기기도 전에 예외가 터져서 테스트 자체가 깨집니다 .

해결은 함수로 감싸는 것 입니다.

expect(() => queue.listen(() => {})).toThrow(TypeError);

toThrow()는 전달받은 함수를 자기가 직접 호출하고 그 과정에서 나온 예외를 잡습니다. 예외를 잡는 방식이 동기와 비동기에서 완전히 다르다는 걸 이번에 몸으로 알았습니다.

타입만 보면 Promise<void>라 비동기처럼 보이는데, 실제 동작은 동기였다는 점이 앞에서 다룬 “TypeScript는 함수가 어떤 예외를 던지는지 모른다”와도 이어집니다.

즉, 타입은 많은 걸 알려주지만, 전부를 알려주지는 않습니다.

#879로 확장

같은 파일에서 이어갈 이슈를 하나 더 맡고 싶습니다.

Test locked queue messages in @fedify/cfworkersOpen

#879·dahlia·5일 전에 업데이트

WorkersMessageQueue는 Workers Queues에서 순서를 보장하기 위해 KV 락을 쓰는데, 락이 이미 걸려 있으면 메시지를 지금 처리하지 않고 나중에 재시도해야 합니다. processMessage()가 그 상황에서 shouldProcess: false를 돌려주는지 검증하는 이슈입니다.

이번에 src/mod.test.ts에 WorkersMessageQueue describe 블록을 직접 만들었으니, #879는 그 블록을 확장하는 작업이 됩니다. 마침 같은 파일에 MockKvNamespace가 이미 있어서 락 저장소를 흉내낼 도구도 갖춰져 있습니다.

첫 기여에서는 환경 설정과 기여 규칙을 익히는 데 시간을 많이 썼는데, 이제 그 비용이 없으니 코드 자체에 더 집중할 수 있을 것 같습니다.

다음 글에서는 #879 이슈로 KV 락 재시도 로직(processMessage())을 검증하는 과정을 정리할 예정입니다.