[Rust 실전 로드맵 30] Axum과 Tokio 백그라운드 작업자 통합하기

Axum handler와 Tokio 작업자를 따로 만들기는 어렵지 않습니다. 까다로운 부분은 둘을 하나의 수명 주기로 묶는 일입니다. 요청이 공유 상태를 읽는 동안 작업자는 HTTP 검사를 실행하고 결과를 저장합니다. 종료 신호가 오면 새 작업을 막되 이미 받은 작업은 잃지 않아야 합니다.

이 fixture는 Rust 2024, rustc·Cargo 1.98.1, Axum 0.8.9, Tokio 1.53.1, tokio-util 0.7.16, SQLx 0.8.6을 사용합니다. 목표는 기능을 많이 넣는 것이 아니라 시작 순서와 종료 계약을 코드로 드러내는 것입니다.

1. 경계를 먼저 고정합니다

HTTP 요청과 저장소를 구체 구현에 바로 묶으면 실패와 종료를 재현하기 어렵습니다. HttpChecker는 검사 대상과 CancellationToken을 받고 도메인 결과를 반환합니다. Repository는 endpoint 조회와 결과 저장만 담당합니다.

#[async_trait]
pub trait HttpChecker: Send + Sync {
    async fn check(&self, endpoint: &Endpoint, cancel: CancellationToken) -> CheckOutcome;
}

#[async_trait]
pub trait Repository: Send + Sync {
    async fn endpoints(&self) -> Result<Vec<Endpoint>, AppError>;
    async fn save_result(&self, endpoint_id: &str, outcome: &CheckOutcome) -> Result<(), AppError>;
    async fn results(&self) -> Result<Vec<(String, String, Option<i64>, Option<String>)>, AppError>;
}

검사기는 성공 상태 코드나 실패 문자열을 반환합니다. 작업자는 두 결과를 모두 저장하므로 DNS 실패나 취소가 조용히 사라지지 않습니다. 테스트용 검사기는 네트워크를 열지 않고 성공, 실패, 대기, 취소를 정확한 시점에 재현합니다.

2. 제한된 채널이 작업 수락 정책을 만듭니다

scheduler 앞의 admission channel과 worker 앞의 work channel은 같은 제한 용량을 사용합니다. try_enqueue는 기다리지 않습니다. 여유가 없으면 QueueError::Full을 반환하므로 API 계층이 즉시 재시도 응답이나 거절 정책을 선택할 수 있습니다.

pub fn try_enqueue(&self, endpoint_id: &str) -> Result<(), QueueError> {
    let endpoint = self
        .endpoints
        .get(endpoint_id)
        .cloned()
        .ok_or(QueueError::UnknownEndpoint)?;
    let guard = self.admission.lock().expect("admission lock poisoned");
    let sender = guard.as_ref().ok_or(QueueError::Closed)?;
    sender.try_send(endpoint).map_err(|error| match error {
        mpsc::error::TrySendError::Full(_) => QueueError::Full,
        mpsc::error::TrySendError::Closed(_) => QueueError::Closed,
    })
}

용량 제한은 단순한 성능 설정이 아닙니다. 생산 속도가 소비 속도를 넘을 때 메모리 대신 호출자에게 압력을 돌려보내는 계약입니다. fixture의 blocking checker 테스트는 첫 작업이 실행 중이고 두 번째 작업이 buffer에 있을 때 세 번째 요청이 Full인지 확인합니다.

3. 시작 순서를 한곳에서 소유합니다

Application::start는 migration이 적용된 repository에서 endpoint를 읽습니다. 공유 상태를 만든 다음 worker와 scheduler를 차례로 띄웁니다. handler가 참조하는 상태는 task spawn 전에 완성됩니다.

startup_steps: vec![
    StartupStep::MigrationsApplied,
    StartupStep::EndpointsLoaded,
    StartupStep::SharedStateBuilt,
    StartupStep::WorkerSpawned,
    StartupStep::SchedulerSpawned,
],

이 순서는 문서용 장식이 아닙니다. 통합 테스트가 그대로 검사합니다. migration이나 초기 조회에 실패하면 background task가 하나만 떠 있는 반쪽짜리 application을 반환하지 않습니다.

4. Axum state에는 API가 읽을 값만 둡니다

/status handler는 endpoint 수와 완료된 검사 수를 반환합니다. Arc<SharedState>를 Router::with_state로 주입하고 State extractor로 꺼냅니다. queue sender와 join handle은 Application이 소유하며 handler state로 흘리지 않습니다.

async fn status(State(state): State<Arc<SharedState>>) -> Json<StatusBody> {
    Json(StatusBody {
        endpoint_count: state.endpoint_count,
        completed_checks: state.completed.load(Ordering::Acquire),
    })
}

pub fn router(&self) -> Router {
    Router::new()
        .route("/status", get(status))
        .with_state(Arc::clone(&self.shared))
}

상태를 작게 유지하면 요청 처리와 process lifecycle의 책임이 섞이지 않습니다. API는 상태를 관찰합니다. 시작·수락·종료의 조정은 Application의 몫입니다.

5. 취소는 검사 seam을 통과합니다

worker는 검사마다 child token을 전달합니다. cancel_checks를 호출하면 현재 검사와 이후 검사기가 취소를 관찰할 수 있습니다. 다만 token은 작업을 강제로 중단하지 않습니다. 검사 구현이 cancelled()를 기다리거나 select! 분기를 두어야 합니다.

fixture의 cancellable checker는 취소를 받은 뒤 Failure { error: "cancelled" }를 반환합니다. worker가 이 결과를 저장한 후 완료 counter를 올리는지 테스트합니다. 취소와 결과 유실을 같은 것으로 취급하지 않는 셈입니다.

6. 종료는 수락 차단, drain, join 순서입니다

정상 종료에서는 admission sender를 먼저 제거합니다. scheduler에는 취소를 알리되 channel에 이미 들어온 항목은 work queue로 모두 넘깁니다. scheduler가 sender를 놓으면 worker는 남은 작업을 처리하고 recv()에서 None을 받아 끝납니다. 마지막으로 두 join handle을 정해진 순서로 기다립니다.

pub async fn shutdown(self) -> Result<ShutdownReport, AppError> {
    self.admission
        .lock()
        .expect("admission lock poisoned")
        .take();
    self.scheduler_cancel.cancel();
    self.scheduler.await??;
    let mut join_order = vec!["scheduler"];
    self.worker.await??;
    join_order.push("worker");
    Ok(ShutdownReport {
        drained_jobs: self.shared.completed.load(Ordering::Acquire),
        scheduler_joined: true,
        worker_joined: true,
        join_order,
    })
}

이 경로에서는 HTTP 검사 token을 자동으로 취소하지 않습니다. 이미 수락한 작업을 drain한다는 계약을 지키기 위해서입니다. 운영 서비스는 drain deadline이 끝난 뒤 cancel_checks를 호출하는 두 단계 정책을 추가할 수 있습니다.

7. 명령으로 계약을 다시 확인합니다

repository root에서 실행합니다.

cd examples/article-30-axum-tokio-background-worker
cargo fmt --all -- --check
cargo check --all-targets --all-features
cargo clippy --all-targets --all-features -- -D warnings
cargo test --all-features
cargo test --release --all-features

통합 테스트 6개는 startup 순서와 Axum state, 성공 저장, 실패 저장, 검사 취소, queue backpressure, 종료 중 drain과 join 순서를 검사합니다. 외부 HTTP server나 공유 database file은 필요하지 않습니다.

이 구조가 모든 종료 정책의 답은 아닙니다. 긴 HTTP 요청에 허용할 drain 시간, persistence 오류가 발생했을 때 재시도할 주체, 여러 worker 사이의 동시성 제한은 service 요구사항으로 남습니다. 먼저 새 작업을 언제 막고 어떤 작업까지 완료할지 정해야 합니다. 그 계약이 정해지면 Axum과 background worker의 경계도 선명해집니다.

전체 소스 코드

이 글의 전체 실행 가능한 소스는 GitHub의 Chapter 30 프로젝트에서 확인할 수 있습니다.