백그라운드 워커

이 챕터에서 다루는 것

잡 큐를 소비하는 워커 풀을 만든다. 이 챕터의 중심은 성능이 아니라 한 잡의 실패가 어디까지 번지는가다.

12-1은 HTTP 핸들러에 패닉 복구 미들웨어를 두었고, 그것으로 충분해 보였다. 충분하지 않다. 그 사실을 먼저 확인한다.

복구 미들웨어가 잡지 못하는 것

recover는 자기 고루틴에서만 동작한다. 다른 고루틴이 패닉하면 그쪽의 스택이 풀리고, 아무도 잡지 않으면 런타임이 프로세스 전체를 죽인다.

func handler(c *gin.Context) {
	go func() {
		process(job) // ← 여기서 패닉하면 서버 전체가 내려간다
	}()
	c.Status(http.StatusAccepted)
}

바깥 핸들러의 defer recover()는 실행조차 되지 않는다. 그 defer는 요청 고루틴에 등록되어 있고, 패닉은 다른 고루틴에서 났기 때문이다.

12-1의 복구 미들웨어 주석에도 이 사실을 적어 뒀다.

examples/13-capstone/internal/httpapi/httpapi.go
// recovery는 패닉 복구 경계다. 요청 하나마다 하나씩만 둔다.
//
// 여기서 복구하는 것은 "이 요청 하나를 포기하고 나머지 서버는 살린다"는
// 뜻이지, 패닉을 정상 흐름으로 쓰라는 뜻이 아니다. 복구된 패닉은 항상
// 500이고 항상 버그다.
//
// 이 경계가 잡지 못하는 것이 하나 있다. 핸들러가 띄운 고루틴 안의
// 패닉이다. defer는 자기 고루틴에서만 도므로, 다른 고루틴이 패닉하면
// 프로세스 전체가 죽는다. 13-4가 그 문제를 다룬다.

워커에서 이것이 왜 치명적인가. 잡 하나의 페이로드가 잘못돼서 핸들러가 nil 맵에 쓰기를 한다고 하자. 프로세스가 죽으면

처리 중이던 다른 잡들이 함께 죽는다.
처리 중이던 HTTP 요청 수백 개가 끊긴다.
재시작한 프로세스가 같은 잡을 다시 집어 또 죽는다. 크래시 루프다.

세 번째가 가장 나쁘다. 잡 하나가 서비스를 무한히 내린다.

규칙: `go` 키워드를 직접 쓰지 않는다

해결책은 간단하지만, 사람의 기억에 맡기면 반드시 새어 나간다. 고루틴을 띄우는 자리를 하나로 모은다.

examples/13-capstone/internal/safego/safego.go
// Package safego는 고루틴 경계에서 패닉을 잡는다.
//
// 12-1은 HTTP 핸들러에 패닉 복구 미들웨어를 두었고, 그것으로 충분해
// 보였다. 충분하지 않다. recover는 자기 고루틴에서만 동작하므로,
// 핸들러가 띄운 고루틴이 패닉하면 미들웨어의 defer는 실행조차 되지 않고
// 프로세스 전체가 죽는다.
//
//	go func() {
//	    process(job) // ← 여기서 패닉하면 서버 전체가 내려간다
//	}()
//
// 처리 중이던 다른 요청 수백 개가 함께 끊긴다. 잡 하나가 잘못된 JSON을
// 담고 있었다는 이유로.
//
// 규칙은 하나다. **go 키워드를 직접 쓰지 않는다.** 고루틴을 띄우는
// 모든 자리에 복구 경계가 있어야 하고, 그것을 사람의 기억에 맡기지 않는
// 방법이 이 패키지다.
package safego

패닉을 에러로 바꾸는 것이 핵심이다. 그러면 호출자가 평소처럼 다룰 수 있다.

examples/13-capstone/internal/safego/safego.go
// PanicError는 복구된 패닉이다.
//
// 패닉 값을 에러로 바꿔 돌려주면 호출자가 평소처럼 다룰 수 있다.
// 워커는 이것을 "잡 하나가 실패했다"로 처리하고 재시도로 넘긴다.
type PanicError struct {
	// Name은 패닉한 자리의 이름이다. 스택만으로도 알 수 있지만
	// 스택은 길고 이름은 검색하기 좋다.
	Name string
	// Value는 panic()에 넘겨진 값이다.
	Value any
	// Stack은 패닉 시점의 스택이다. 복구한 뒤에 debug.Stack()을 부르면
	// 이미 스택이 풀린 뒤라 아무 쓸모가 없으므로, defer 안에서 바로 뜬다.
	Stack []byte
}

Stack을 defer 안에서 뜨는 것이 중요하다. Do가 돌아온 뒤에 debug.Stack()을 부르면 이미 스택이 풀려 있어 패닉 지점이 없다. 테스트가 그것을 확인한다.

examples/13-capstone/internal/safego/safego_test.go
	// 스택은 패닉 지점을 담아야 한다. 복구한 뒤에 debug.Stack()을
	// 부르면 이미 풀린 스택이 잡혀서 아무 쓸모가 없다.
	if !strings.Contains(string(pe.Stack), "safego_test.go") {
		t.Errorf("스택에 패닉 지점이 없다:\n%s", pe.Stack)
	}

변환 자체는 열 줄이다. 2-7에서 본 "명명된 반환값에 defer가 대입하는" 형태다.

examples/13-capstone/internal/safego/safego.go
// Do는 fn을 돌리고, 패닉하면 *PanicError로 바꿔 돌려준다.
//
// 명명된 반환값 err에 defer가 대입하는 형태다(2-7). 이것이 패닉을
// 정상적인 에러 반환으로 바꾸는 유일한 방법이다.
func Do(name string, fn func() error) (err error) {
	defer func() {
		if rec := recover(); rec != nil {
			err = &PanicError{Name: name, Value: rec, Stack: debug.Stack()}
		}
	}()
	return fn()
}

패닉 값이 에러였다면 사슬을 잇는다. errors.Is가 뚫고 갈 수 있어야 한다.

examples/13-capstone/internal/safego/safego.go
// Unwrap은 패닉 값이 에러였다면 그것을 노출한다.
// errors.Is로 원래 에러를 찾아 들어갈 수 있어야 한다.
func (e *PanicError) Unwrap() error {
	if err, ok := e.Value.(error); ok {
		return err
	}
	return nil
}

고루틴을 띄우는 쪽은 WaitGroup을 반드시 함께 받는다.

examples/13-capstone/internal/safego/safego.go
// Go는 고루틴을 띄우고 패닉을 로그로 남긴다.
//
// WaitGroup을 함께 받는 것이 의도다. 고루틴을 띄우면서 "이것이 언제
// 끝나는가"에 답하지 못하면 그것은 누수다(7-1). WaitGroup을 인자로
// 요구하면 답을 미룰 수 없다.
func Go(wg *sync.WaitGroup, l *slog.Logger, name string, fn func()) {
	wg.Go(func() {
		if err := Do(name, func() error { fn(); return nil }); err != nil {
			logPanic(l, name, err)
		}
	})
}

wg.Go는 파트 7에서 본 Go 1.25의 형태다. Add(1)과 defer Done()을 짝지을 필요가 없어졌다.

워커 풀의 구조

examples/13-capstone/internal/worker/worker.go
// Package worker는 잡 큐를 소비하는 고루틴 워커 풀이다.
//
// 구조는 셋으로 나뉜다.
//
//   - **디스패처** 하나. 주기적으로 Claim을 불러 잡을 집어 채널에 넣는다.
//     DB를 두드리는 것은 이 고루틴 하나뿐이다.
//   - **워커** N개. 채널에서 잡을 받아 핸들러를 돌린다.
//   - **결과 처리**. 성공이면 Complete, 실패면 백오프를 계산해 Retry,
//     시도 횟수를 다 쓰면 Bury.
//
// 왜 워커마다 각자 폴링하지 않는가. 그러면 DB 쿼리가 워커 수만큼 늘고,
// 워커가 서로 같은 행을 두고 경쟁하며 대부분의 Claim이 빈손으로 돌아온다.
// 디스패처 하나가 한 번에 여러 건을 집어 나눠 주는 편이 쿼리도 적고
// 경쟁도 없다.
package worker

핸들러의 계약은 짧다.

examples/13-capstone/internal/worker/worker.go
// Handler는 잡 하나를 처리한다.
//
// nil을 돌려주면 성공, 에러면 재시도다. 패닉해도 프로세스는 죽지 않고
// 재시도로 처리된다 — 그것이 safego의 역할이다.
//
// **핸들러는 멱등해야 한다.** 워커가 처리를 끝내고 Complete를 부르기
// 직전에 죽으면 같은 잡이 다시 온다. 최소 한 번 실행이라는 성질은
// 12-8의 아웃박스와 같고, 대가도 같다.
type Handler func(ctx context.Context, j task.Job) error

재시도해도 소용없는 실패는 따로 표시한다.

examples/13-capstone/internal/worker/worker.go
// ErrPermanent를 감싸 돌려주면 재시도하지 않고 즉시 dead로 보낸다.
//
// 재시도가 무의미한 실패가 있다. 페이로드 JSON이 깨졌거나, 대상
// 사용자가 이미 탈퇴했거나. 그런 것을 다섯 번 재시도하는 것은 시간과
// 로그만 낭비한다.
var ErrPermanent = errors.New("worker: 재시도할 수 없는 실패")

Run은 고루틴을 직접 띄우지 않는다.

examples/13-capstone/internal/worker/worker.go
	var workers sync.WaitGroup
	for i := range p.cfg.Concurrency {
		name := p.cfg.Name + "-" + strconv.Itoa(i)
		// go 키워드를 직접 쓰지 않는다. 이 고루틴 안에서 패닉이 나면
		// 프로세스가 죽고, 그것이 13-4의 주제다.
		safego.Go(&workers, p.cfg.Logger, name, func() {
			for j := range jobs {
				p.process(name, j)
			}
		})
	}

	p.dispatch(ctx, jobs)
	close(jobs)
	workers.Wait()

그리고 잡 처리 지점에서 한 번 더 감싼다.

examples/13-capstone/internal/worker/worker.go
	start := p.cfg.Now()
	// 여기가 이 패키지의 존재 이유다. 핸들러가 패닉해도 Do가 에러로
	// 바꿔 돌려주므로, 프로세스는 살아 있고 잡은 재시도된다.
	err := safego.Do("job:"+j.Topic, func() error { return h(ctx, j) })
	elapsed := p.cfg.Now().Sub(start)

두 겹인 것이 의도다. safego.Do가 잡 하나를 잡고, safego.Go가 워커 루프 자체의 패닉(예를 들어 우리 코드의 버그)을 잡는다. 안쪽이 뚫려도 바깥이 프로세스를 살린다.

프로세스가 살아남는다는 것을 증명한다

말로만 하면 안 된다. 테스트가 못 박는다.

examples/13-capstone/internal/worker/worker_test.go
// TestPanicInJobDoesNotKillProcess가 이 패키지의 존재 이유다.
//
// 12-1의 HTTP 복구 미들웨어는 이 패닉을 잡지 못한다. 워커 고루틴은
// 요청 고루틴이 아니고, recover는 자기 고루틴에서만 동작하기 때문이다.
// safego 없이 이 테스트를 돌리면 테스트 바이너리 전체가 죽는다 —
// 실패 보고가 아니라 "panic: assignment to entry in nil map" 한 줄과
// 종료 코드 2만 남는다.
//
// 통과한다는 것은 프로세스가 살아남았다는 뜻이고, 살아남았을 뿐 아니라
// 다음 잡도 정상 처리했다는 뜻이다.
func TestPanicInJobDoesNotKillProcess(t *testing.T) {
	t.Parallel()
	st := memstore.New()
	enqueue(t, st, "boom", `{}`)
	enqueue(t, st, "ok", `{}`)

	p := newPool(t, st, map[string]worker.Handler{
		"boom": func(context.Context, task.Job) error {
			var m map[string]int
			m["터진다"] = 1 // nil 맵에 쓰기 — 3-4에서 본 그 패닉이다.
			return nil
		},
		"ok": func(context.Context, task.Job) error { return nil },
	})

단언이 셋이다. 프로세스가 살아 있고(테스트가 끝까지 돈다), 다른 잡이 처리되고, 패닉한 잡은 재시도를 다 쓴 뒤 dead가 된다.

examples/13-capstone/internal/worker/worker_test.go
	s := p.Stats()
	if s.Panicked < 3 {
		t.Errorf("Panicked = %d, 3 이상을 기대 (MaxAttempts만큼 패닉해야 한다)", s.Panicked)
	}
	if s.Succeeded != 1 {
		t.Errorf("Succeeded = %d, 1을 기대 — 패닉이 다른 잡을 막았다", s.Succeeded)
	}
	if counts := countByStatus(t, st); counts[task.JobDead] != 1 {
		t.Errorf("dead = %d, 1을 기대 (%v)", counts[task.JobDead], counts)
	}

:::tip 이 테스트를 실제로 깨뜨려 보기 worker.go의 safego.Do(...)를 h(ctx, j) 직접 호출로 바꾸고 돌려 보면 테스트 실패 메시지가 아니라 런타임 패닉과 종료 코드 2가 나온다. --- FAIL: 줄이 아예 없다 — 테스트 프레임워크가 결과를 보고할 기회조차 없이 프로세스가 죽기 때문이다. 운영에서는 이것이 컨테이너 재시작이다. :::

패닉은 스택과 함께 별도로 센다.

examples/13-capstone/internal/worker/worker.go
	default:
		var pe *safego.PanicError
		if errors.As(err, &pe) {
			p.stats.panicked.Add(1)
			l.LogAttrs(ctx, slog.LevelError, "job panicked",
				slog.String("err", pe.Error()),
				slog.String("stack", string(pe.Stack)))
		}
		p.failed(ctx, l, j, err)

Panicked를 따로 세는 이유는, 이 숫자가 0이 아니면 버그이기 때문이다. 재시도 횟수와 달리 정상 운영에서 올라갈 값이 아니다.

재시도와 지수 백오프

examples/13-capstone/internal/worker/backoff.go
// Backoff는 attempt번째 시도 뒤에 얼마나 기다릴지 정한다.
//
// 세 가지가 다 필요하다.
//
//  1. **지수** — 실패가 이어지면 간격을 늘린다. 고정 간격으로 재시도하면
//     장애 난 대상에 같은 부하를 계속 던져 회복을 방해한다.
//  2. **상한** — 2의 거듭제곱은 금방 몇 시간이 된다. 상한이 없으면
//     일시적 장애 뒤에 잡이 반나절 동안 잠들어 있다.
//  3. **지터** — 이것이 가장 자주 빠진다. 지터가 없으면 같은 순간에
//     실패한 잡 천 개가 같은 순간에 재시도한다. 장애가 회복되는 순간
//     동시에 몰려가 다시 죽인다. 이것이 재시도 폭풍이다.
//
// 여기서는 equal jitter를 쓴다. 절반은 고정, 나머지 절반은 무작위다.
// full jitter(0에서 상한까지 균등)는 분산이 더 좋지만 아주 짧은 대기가
// 나올 수 있어서, 최소 간격을 보장하고 싶을 때는 이쪽이 낫다.
//
// rnd는 [0, 1)을 돌려주는 함수다. 주입받는 이유는 테스트다 —
// math/rand를 직접 부르면 기대값을 적을 수 없다.
func Backoff(attempt int, base, max time.Duration, rnd func() float64) time.Duration {

오버플로도 막아야 한다.

examples/13-capstone/internal/worker/backoff.go
	// 1 << (attempt-1)을 그대로 쓰면 attempt가 63을 넘는 순간 오버플로한다.
	// 지수를 먼저 자른다.
	shift := min(attempt-1, 32)
	d := float64(base) * math.Pow(2, float64(shift))
	if d > float64(max) {
		d = float64(max)
	}

	half := d / 2
	return time.Duration(half + half*rnd())

난수를 주입받으므로 테스트가 결정적이다.

examples/13-capstone/internal/worker/worker_test.go
	base, maxWait := time.Second, 8*time.Second
	// rnd가 0이면 equal jitter의 아래쪽 절반, 1에 가까우면 위쪽 끝이다.
	tests := []struct {
		attempt  int
		rnd      float64
		wantLow  time.Duration
		wantHigh time.Duration
	}{
		{1, 0, 500 * time.Millisecond, 500 * time.Millisecond},
		{2, 0, time.Second, time.Second},
		{3, 0, 2 * time.Second, 2 * time.Second},
		{9, 0, 4 * time.Second, 4 * time.Second},   // 상한에 걸린다
		{9, 0.5, 6 * time.Second, 6 * time.Second}, // 절반 + 절반의 절반
		{100, 0, 4 * time.Second, 4 * time.Second}, // 오버플로하지 않는다
		{1, 0.999, 999 * time.Millisecond, time.Second},
	}

실패 처리 분기는 두 줄이다.

examples/13-capstone/internal/worker/worker.go
// failed는 실패한 잡을 재시도로 되돌리거나 묻는다.
func (p *Pool) failed(ctx context.Context, l *slog.Logger, j task.Job, err error) {
	if errors.Is(err, ErrPermanent) || j.Attempts >= p.cfg.MaxAttempts {
		p.bury(ctx, l, j, err)
		return
	}

포기한 잡은 지우지 않는다.

examples/13-capstone/internal/worker/worker.go
// bury는 잡을 포기한다.
//
// 지우지 않고 dead로 남긴다. 무엇이 왜 실패했는지 나중에 볼 수 있어야
// 하고, 고친 뒤 다시 pending으로 되돌리는 것도 그래야 가능하다.

context를 어디에 쓰고 어디에 쓰지 않는가

이 부분이 정상 종료의 핵심이고, 가장 자주 틀리는 곳이다.

examples/13-capstone/internal/worker/worker.go
	// 처리에는 부모 ctx를 쓰지 않는다. 종료 신호가 오면 부모는 이미
	// 취소되어 있고, 그러면 진행 중인 잡이 전부 즉시 실패한다.
	// 대신 가시성 타임아웃을 마감으로 삼는다 — 그 시간을 넘기면
	// 어차피 다른 워커가 같은 잡을 집기 때문에 계속할 이유가 없다.
	ctx, cancel := context.WithTimeout(context.Background(), p.cfg.Visibility)
	defer cancel()

"context는 위에서 아래로 전파한다"가 기본 규칙이지만, 여기서는 어긴다. 이유는 취소의 의미가 다르기 때문이다. 루트 context의 취소는 "새 일을 시작하지 마라"는 뜻이지 "하던 일을 버려라"가 아니다. 그것을 그대로 전파하면 SIGTERM 순간에 진행 중인 잡이 전부 실패하고, 재시도 카운트만 소모한다.

대신 종료 신호는 디스패처만 받는다.

examples/13-capstone/internal/worker/worker.go
// Run은 ctx가 끝날 때까지 잡을 처리한다.
//
// 종료 순서가 이 함수의 핵심이다.
//
//  1. ctx가 취소되면 디스패처가 폴링을 멈추고 채널을 닫는다.
//  2. 워커는 채널이 비고 닫힐 때까지 남은 잡을 마저 처리한다.
//     range 루프가 그것을 공짜로 해 준다.
//  3. 전부 끝나면 Run이 돌아온다.
//
// 2번이 중요하다. 이미 Claim해서 채널에 들어 있는 잡을 버리면, 그 잡은
// 가시성 타임아웃이 지나야 다시 잡히므로 30초 동안 처리되지 않는다.
//
// 처리 중인 잡에는 ctx를 그대로 넘기지 않는다. 배수 유예 안에서 끝내야
// 하므로 별도의 마감을 건다 — 아래 process를 보라.
func (p *Pool) Run(ctx context.Context) {

for j := range jobs가 이 구조를 공짜로 만들어 준다. 채널이 닫히면 남은 것을 다 꺼낸 뒤 루프가 끝난다. select로 ctx.Done()을 함께 보는 형태로 쓰면 남은 잡을 버리게 된다.

테스트가 확인한다.

examples/13-capstone/internal/worker/worker_test.go
// TestShutdownDrainsInFlight는 종료 시 진행 중인 잡을 마저 끝내는지 본다.
//
// 핸들러가 시작되면 신호를 보내고, 테스트는 그 신호를 받은 뒤 종료를
// 지시한다. Run이 돌아온 시점에 그 잡은 done이어야 한다.
func TestShutdownDrainsInFlight(t *testing.T) {

examples/13-capstone/internal/worker/worker_test.go
	<-started
	cancel() // 종료 지시
	close(release)
	<-done

	if counts := countByStatus(t, st); counts[task.JobDone] != 1 {
		t.Errorf("종료 후 done = %d, 1을 기대. 진행 중이던 잡을 버렸다 (%v)",
			counts[task.JobDone], counts)
	}

고루틴이 남지 않는 것도 확인한다.

examples/13-capstone/internal/worker/worker_test.go
	// Run이 돌아왔다는 것은 워커 WaitGroup이 끝났다는 뜻이다. 그래도
	// 스케줄러가 정리하기까지 시간이 걸리므로 확인하며 기다린다.
	for range 1000 {
		if runtime.NumGoroutine() <= before {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Errorf("고루틴이 남았다: 이전 %d, 이후 %d", before, runtime.NumGoroutine())

서버와 워커의 수명

main에서 둘이 같은 루트 context를 공유한다.

examples/13-capstone/cmd/taskd/main.go
	var workers sync.WaitGroup
	safego.Go(&workers, logger, "worker-pool", func() { pool.Run(ctx) })

종료 순서는 12-8이 정한 그대로다.

examples/13-capstone/cmd/taskd/main.go
	// 4. 워커를 정리한다. ctx는 이미 취소되어 있으므로 디스패처는 멈췄고,
	//    진행 중인 잡만 마저 끝나기를 기다린다.
	workerWG.Wait()

	s := pool.Stats()
	logger.LogAttrs(context.Background(), slog.LevelInfo, "worker stopped",
		slog.Int64("claimed", s.Claimed),
		slog.Int64("succeeded", s.Succeeded),
		slog.Int64("retried", s.Retried),
		slog.Int64("buried", s.Buried),
		slog.Int64("panicked", s.Panicked),
	)

SIGTERM을 보내면 실제로 이렇게 나온다.

level=INFO msg="shutting down" cause="terminated signal received" by_signal=true
level=INFO msg="worker stopped" claimed=1 succeeded=1 retried=0 buried=0 panicked=0
level=INFO msg=stopped

첫 줄과 둘째 줄 사이가 3초다. drain_delay가 3초이기 때문이고, 그동안 레디니스는 이미 내려가 있다(13-5에서 확인한다).

cause가 문자열 에러인 것은 12-8에서 확인한 그대로다. os.Signal을 꺼낼 수 없고, 시그널인지 아닌지는 값 비교로 가른다.

examples/13-capstone/cmd/taskd/main.go
	cause := context.Cause(ctx)
	logger.LogAttrs(context.Background(), slog.LevelInfo, "shutting down",
		// 12-8에서 확인한 대로, 원인은 os.Signal이 아니라 문자열 에러다.
		// 시그널 때문인지는 값 비교로 가른다.
		slog.String("cause", cause.Error()),
		slog.Bool("by_signal", cause != context.Canceled),
	)

처리량은 배치 크기가 정한다

처음 만들었을 때 Claim의 limit을 Concurrency로 뒀다. 워커가 넷이니 네 개씩 집으면 되겠다는 생각이었다. 부하를 걸어 보니 아니었다.

잡을 6000개 쌓아 두고 10초마다 잡 통계를 찍어 봤다.

{"dead":0,"done":468,"pending":5532,"running":0} t=+10s
{"dead":0,"done":548,"pending":5452,"running":0} t=+20s
{"dead":0,"done":628,"pending":5372,"running":0} t=+30s
{"dead":0,"done":708,"pending":5292,"running":0} t=+40s

정확히 10초에 80건. 초당 8건이다. 핸들러는 로그 한 줄 찍는 것이 전부라 마이크로초 단위인데도 그렇다.

이유는 산수다. 폴링 주기가 500ms이고 한 번에 4건을 집으면, 아무리 빨라도 4 ÷ 0.5초 = 8건/초가 상한이다. 핸들러 속도가 아니라 폴링 구조가 처리량을 정하고 있었다.

배치를 분리했다.

examples/13-capstone/internal/worker/worker.go
	Concurrency int
	// Batch는 한 번의 Claim으로 집어 오는 잡 수다.
	//
	// 0이면 Concurrency의 10배를 쓴다. 이 값을 Concurrency와 같게 두면
	// 처리량이 Concurrency/Poll로 고정된다 — 핸들러가 아무리 빨라도
	// 폴링 주기마다 워커 수만큼만 들어오기 때문이다. 실제로 그렇게
	// 짜 놓고 부하를 걸었더니 초당 8건에서 더 오르지 않았다.
	Batch       int
	Poll        time.Duration

같은 조건에서 다시 재면 이렇다. Batch의 기본값은 64다.

{"dead":0,"done":4216,"pending":1784,"running":0} t=+10s
{"dead":0,"done":5496,"pending":504,"running":0} t=+20s
{"dead":0,"done":6000,"pending":0,"running":0} t=+30s

10초에 1280건, 초당 128건이다. 64 ÷ 0.5초 = 128. 예측한 값과 정확히 같다. 그리고 30초 안에 큐가 완전히 비었다.

:::note 이 숫자들에 대해 done의 절대값은 실행마다 다르다(앞선 부하 시험에서 이미 처리된 잡이 있었다). 결정적인 것은 증가율이고, 그 값이 Batch ÷ Poll과 정확히 맞는다는 사실이 측정의 요점이다.

핸들러가 무거우면 이 계산이 달라진다. 그때는 Concurrency가 상한을 정하고 배치는 남아돈다. 어느 쪽이 상한인지를 아는 유일한 방법이 재 보는 것이다. :::

디스패처가 채널로 보내는 부분이 자연스러운 배압이 된다.

examples/13-capstone/internal/worker/worker.go
	for _, j := range claimed {
		select {
		case jobs <- j:
		case <-ctx.Done():
			// 아직 못 보낸 잡은 running으로 남는다. 가시성이 지나면
			// 다음 프로세스가 회수한다. 여기서 Retry로 되돌려 놓으면
			// 더 빨리 재개되지만, 종료 중에 쓰기를 늘리는 것이므로
			// 어느 쪽이 나은지는 상황에 달렸다.
			return
		}
	}

채널에 버퍼가 없는 것이 의도다. 64건을 집어도 워커가 받아 갈 때까지 디스패처가 기다리므로, 다음 폴링은 앞 배치를 다 나눠 준 뒤에야 일어난다. 핸들러가 느리면 폴링이 자동으로 느려진다.

설정 검증에도 이 관계를 넣는다.

examples/13-capstone/internal/config/config.go
	if c.Worker.Batch < c.Worker.Concurrency {
		errs = append(errs, fmt.Errorf("config: worker.batch(%d)가 concurrency(%d)보다 작다. "+
			"폴링 한 번에 워커 수만큼도 못 집으면 처리량이 concurrency/poll로 고정된다",
			c.Worker.Batch, c.Worker.Concurrency))
	}

잡 핸들러

핸들러가 하는 일이 셋뿐인 것이 좋은 모습이다.

examples/13-capstone/internal/worker/handlers.go
// NotifyHandler는 task.created / task.completed 잡을 처리한다.
//
// 하는 일이 셋뿐인 것이 좋은 잡 핸들러의 모습이다.
// 페이로드를 푼다 → 부수효과를 낸다 → 에러를 돌려준다.
// 재시도도 백오프도 로깅도 워커가 한다.
func NotifyHandler(n Notifier) Handler {
	return func(ctx context.Context, j task.Job) error {
		var ev task.Event
		if err := json.Unmarshal([]byte(j.Payload), &ev); err != nil {
			// 깨진 JSON은 다섯 번 재시도해도 계속 깨져 있다.
			// ErrPermanent를 감싸면 워커가 바로 dead로 보낸다.
			return fmt.Errorf("%w: 페이로드 파싱: %v", ErrPermanent, err)
		}
		return n.Notify(ctx, j.Topic, ev)
	}
}

웹훅 알림기에서 재시도할 실패와 아닌 실패를 가른다.

examples/13-capstone/internal/worker/handlers.go
	switch {
	case resp.StatusCode < 300:
		return nil
	case resp.StatusCode == http.StatusTooManyRequests, resp.StatusCode >= 500:
		// 상대가 과부하거나 고장 난 것이다. 재시도한다.
		return fmt.Errorf("웹훅 %d", resp.StatusCode)
	default:
		// 4xx는 우리 요청이 잘못된 것이다. 같은 요청을 다시 보내도
		// 같은 답이 온다.
		return fmt.Errorf("%w: 웹훅 %d", ErrPermanent, resp.StatusCode)
	}

최소 한 번 실행이므로 받는 쪽이 중복을 걸러야 하고, 그러려면 무엇이 같은 이벤트인지 알려 줘야 한다.

examples/13-capstone/internal/worker/handlers.go
	// 멱등 키를 함께 보낸다. 최소 한 번 실행이므로 같은 이벤트가 두 번
	// 갈 수 있고, 받는 쪽이 그것을 걸러 내려면 무엇이 같은 것인지
	// 알려 줘야 한다.
	req.Header.Set("X-Idempotency-Key", fmt.Sprintf("%s:%d", topic, ev.TaskID))

타임아웃 없는 HTTP 클라이언트를 쓰지 않는 것도 중요하다.

examples/13-capstone/internal/worker/handlers.go
// NewWebhook은 웹훅 알림기를 만든다.
//
// 타임아웃 없는 http.Client를 쓰지 않는다. 기본 클라이언트는 타임아웃이
// 없어서, 응답하지 않는 상대 하나가 워커를 영원히 붙잡는다. 12-8의
// 타임아웃 계층에서 가장 안쪽에 해당하는 값이다.
func NewWebhook(url string, timeout time.Duration) *Webhook {

`-race`로 검증한다

워커 풀은 공유 상태가 많다. 통계 카운터, 잡 채널, 저장소. 레이스 검출기 없이는 검증했다고 할 수 없다.

go test -race -count=1 ./...

ok  	example.com/capstone/internal/arch	5.590s
ok  	example.com/capstone/internal/config	1.757s
ok  	example.com/capstone/internal/gormstore	1.803s
ok  	example.com/capstone/internal/httpapi	2.240s
ok  	example.com/capstone/internal/memstore	2.510s
ok  	example.com/capstone/internal/safego	3.270s
ok  	example.com/capstone/internal/worker	2.898s

통계는 원자적 연산으로 센다. 뮤텍스로도 되지만, 잡마다 락을 잡는 것은 낭비다.

examples/13-capstone/internal/worker/worker.go
// Stats는 워커가 무엇을 했는지 센 값이다. 지표로 내보낸다.
type Stats struct {
	Claimed   int64
	Succeeded int64
	Retried   int64
	Buried    int64
	Panicked  int64
}

:::warning 테스트가 측정 대상을 굶기지 않게 이 챕터의 테스트를 처음 쓸 때 조건 확인 루프를 runtime.Gosched()로 돌렸다. t.Parallel 테스트 다섯 개가 동시에 그 루프를 도니 워커 고루틴이 CPU를 얻지 못했고, 전부 Claimed=0으로 10초 타임아웃했다.

--- FAIL: TestHandlesJob (10.00s)
    worker_test.go:103: 조건이 10초 안에 만족되지 않았다. stats={Claimed:0 Succeeded:0 Retried:0 Buried:0 Panicked:0}

time.Sleep(time.Millisecond)으로 바꾸니 전부 1.5초 안에 끝났다. runtime.Gosched()는 "다른 고루틴에 양보한다"이지 "쉰다"가 아니다. 바쁘게 도는 루프는 여전히 바쁘다. :::

examples/13-capstone/internal/worker/worker_test.go
// runUntil은 cond가 참이 될 때까지 풀을 돌리고, 되면 정상 종료시킨다.
//
// "충분히 긴" 시간을 자고 결과를 확인하는 방식 대신 조건을 반복 확인한다.
// 12-4의 leaky.WaitGoroutines와 같은 발상이다 — 느린 머신에서도 깨지지
// 않고, 조건이 만족되는 즉시 진행한다.
//
// 확인 사이에 짧게 자는 것이 중요하다. runtime.Gosched()로 바쁘게 도는
// 루프를 t.Parallel 테스트 여럿이 동시에 돌리면 워커 고루틴이 CPU를
// 얻지 못한다. 실제로 그렇게 짜 봤더니 Claimed가 0인 채로 전부
// 타임아웃했다 — 테스트가 측정 대상을 굶긴 것이다.
func runUntil(t *testing.T, p *worker.Pool, cond func() bool) {

또 하나 실제로 밟은 함정이 있다. 테스트의 고정 시각을 미래로 두면 워커가 잡을 영원히 집지 않는다.

examples/13-capstone/internal/worker/worker_test.go
// base는 과거의 고정 시각이다. 미래로 잡으면 available_at이 아직
// 오지 않은 것이 되어 워커가 잡을 영원히 집지 않는다 — 실제로 그렇게
// 짰다가 모든 테스트가 Claimed=0으로 타임아웃했다.
var base = time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)

증상이 위와 똑같아서(전부 Claimed=0) 원인을 두 번 찾아야 했다. 시각을 고정하는 테스트에서는 그 시각이 실행 시각보다 앞인지를 항상 확인한다.

흔한 실수

go 키워드를 그냥 쓴다. 그 고루틴의 패닉이 프로세스를 죽인다. 이 챕터의 전부다.

recover() 뒤에 debug.Stack()을 부른다. 스택이 이미 풀려 있어 쓸모없다. defer 안에서 뜬다.

고루틴을 띄우면서 "언제 끝나는가"에 답하지 않는다. 누수다(7-1). WaitGroup을 인자로 요구하면 답을 미룰 수 없다.

진행 중인 잡에 루트 context를 그대로 넘긴다. SIGTERM 순간 전부 실패하고 재시도 카운트만 소모한다.

워커 루프를 select로 ctx.Done()과 함께 본다. 채널에 남은 잡을 버린다. for range면 공짜로 배수된다.

백오프에 지터를 안 넣는다. 재시도 폭풍이 난다.

백오프에 상한을 안 둔다. 잡이 반나절 잠든다.

attempts를 처리 후에 올린다. 프로세스가 죽으면 카운트가 안 오르고 무한 재시도가 된다.

배치 크기를 워커 수와 같게 둔다. 처리량이 Concurrency ÷ Poll로 고정된다. 재 보기 전에는 모른다.

재시도해도 소용없는 실패를 재시도한다. ErrPermanent로 가른다.

타임아웃 없는 http.Client를 쓴다. 상대 하나가 워커를 영원히 붙잡는다.

dead 잡을 지운다. 무엇이 왜 실패했는지 볼 수 없게 된다.

정리

recover는 자기 고루틴에서만 동작한다. HTTP 복구 미들웨어는 워커 고루틴의 패닉을 잡지 못하고, 그 패닉은 프로세스를 죽인다.
go 키워드를 직접 쓰지 않는다. safego.Go가 복구 경계와 WaitGroup을 강제한다. 패닉은 에러가 되고, 잡은 재시도된다.
패닉 스택은 defer 안에서 뜬다. 나중에 뜨면 이미 풀려 있다.
디스패처 하나 + 워커 N개. DB를 두드리는 것은 하나뿐이다.
처리량은 Batch ÷ Poll이다. 재서 확인했고, 8건/초가 128건/초가 됐다.
백오프는 지수 + 상한 + 지터. 셋 다 없으면 안 된다.
진행 중인 잡에는 루트 context를 넘기지 않는다. 가시성 타임아웃을 마감으로 쓴다.
for range jobs가 배수를 공짜로 해 준다. 채널을 닫으면 남은 것을 마저 처리한다.
Panicked는 0이어야 하는 숫자다. 올라가면 버그다.

연습문제

Poll을 고정 주기 대신 적응형으로 바꿔 보자. 배치가 가득 차서 돌아오면 즉시 다시 집고, 빈손이면 주기를 두 배로 늘린다(상한까지). 큐가 밀릴 때는 지연이 줄고 한가할 때는 DB 쿼리가 줄어든다. 상한과 하한을 무엇으로 정하겠는가?
지금은 모든 토픽이 한 풀을 공유한다. 느린 토픽 하나가 워커를 다 차지하면 빠른 토픽이 굶는다. 토픽별로 풀을 나누거나 우선순위를 두는 방법을 설계해 보자. Claim의 시그니처는 어떻게 바뀌는가? 인덱스는?
dead 잡을 다시 pending으로 되돌리는 관리 엔드포인트를 만들어 보자. POST /v1/admin/jobs/{id}/retry. 어떤 검증이 필요한가? 되돌릴 때 attempts를 0으로 초기화해야 하는가, 그대로 둬야 하는가? 둘의 차이가 운영에서 어떤 결과를 낳는지 생각해 보자.

이 챕터에서 다루는 것​

복구 미들웨어가 잡지 못하는 것​

규칙: go 키워드를 직접 쓰지 않는다​

워커 풀의 구조​

프로세스가 살아남는다는 것을 증명한다​

재시도와 지수 백오프​

context를 어디에 쓰고 어디에 쓰지 않는가​

서버와 워커의 수명​

처리량은 배치 크기가 정한다​

잡 핸들러​

-race로 검증한다​

흔한 실수​

정리​

연습문제​