Rust iterator·fold로 상태없는 집계 리팩터링

서버 로그, 이벤트 스트림, 배치 결과처럼 “한 번 훑으면서 집계”하는 코드는 처음엔 for 루프와 mut 변수 몇 개로 시작합니다. 문제는 요구사항이 늘면서 상태가 여기저기 퍼지고, 분기마다 continue 나 break 가 섞이며, 누락된 케이스가 생기기 쉽다는 점입니다. Rust는 소유권과 불변성을 기본값으로 두는 언어라서, 이런 집계 로직을 iterator 와 fold 로 상태없는(stateless) 형태로 정리하면 코드 품질이 눈에 띄게 좋아집니다.

이 글에서는

for + mut 기반 집계를 fold 로 옮기는 사고방식
filter_map / try_fold / scan 등 실전 조합
오버플로, 에러, 조기 종료, 윈도우 집계 같은 “현업에서 꼭 나오는” 케이스

를 예제로 다룹니다.

왜 “상태없는 집계”가 중요한가

집계 로직은 대부분 다음과 같은 문제를 겪습니다.

숨은 상태: 루프 밖에 선언된 mut 변수들이 여러 분기에서 갱신되어, 어떤 조건에서 값이 바뀌는지 추적이 어렵습니다.
테스트 어려움: 중간 상태를 관찰하려면 로깅을 심거나, 루프를 쪼개야 합니다.
에러 처리의 복잡도: 파싱 실패, I/O 오류, overflow 같은 예외 흐름이 루프에 섞이며 가독성이 급격히 떨어집니다.
성능 튜닝의 난이도: 병렬 처리나 스트리밍 처리로 확장하려면 상태 공유가 발목을 잡습니다.

fold 는 “누적기(accumulator)를 입력 스트림에 대해 순수하게 갱신한다”는 모델을 강제합니다. 누적기는 값으로 전달되고 반환되므로, 상태 변경 지점이 명확해지고(함수형 스타일), Rust의 타입 시스템이 누락 케이스를 더 잘 잡아줍니다.

문제 코드: `for` 루프에 흩어진 상태

예를 들어, 결제 이벤트를 읽어 통계를 만든다고 해봅시다.

성공 결제 합계
실패 건수
통화별 합계
이상치(금액이 너무 큼) 감지

use std::collections::HashMap;

#[derive(Debug, Clone)]
struct Event {
    user_id: String,
    amount_cents: i64,
    currency: String,
    ok: bool,
}

#[derive(Debug, Default)]
struct Stats {
    ok_sum_cents: i64,
    fail_count: u64,
    per_currency_sum: HashMap<String, i64>,
    has_outlier: bool,
}

fn aggregate_bad(events: &[Event]) -> Stats {
    let mut stats = Stats::default();

    for e in events {
        if e.amount_cents > 1_000_000_00 {
            stats.has_outlier = true;
        }

        if e.ok {
            stats.ok_sum_cents += e.amount_cents;
            *stats
                .per_currency_sum
                .entry(e.currency.clone())
                .or_insert(0) += e.amount_cents;
        } else {
            stats.fail_count += 1;
        }
    }

    stats
}

작아 보이지만, 여기에 “파싱 실패는 제외”, “금액 음수는 별도 카운트”, “특정 사용자 제외” 같은 요구가 붙으면 if 가 중첩되고 상태 갱신이 분기마다 갈라집니다.

1단계: 누적기를 명시하고 `fold` 로 옮기기

fold 의 핵심은

누적기 타입을 명시한다
입력 하나를 처리해 누적기를 반환한다

입니다.

use std::collections::HashMap;

#[derive(Debug, Default)]
struct Stats {
    ok_sum_cents: i64,
    fail_count: u64,
    per_currency_sum: HashMap<String, i64>,
    has_outlier: bool,
}

fn aggregate_fold(events: &[Event]) -> Stats {
    events.iter().fold(Stats::default(), |mut acc, e| {
        if e.amount_cents > 1_000_000_00 {
            acc.has_outlier = true;
        }

        if e.ok {
            acc.ok_sum_cents += e.amount_cents;
            *acc.per_currency_sum.entry(e.currency.clone()).or_insert(0) += e.amount_cents;
        } else {
            acc.fail_count += 1;
        }

        acc
    })
}

여전히 mut acc 는 쓰지만, 상태의 범위가 클로저 내부로 좁혀졌고 “입력 하나 처리 후 누적기를 반환”하는 형태가 강제됩니다. 이 구조가 되면 다음 리팩터링이 쉬워집니다.

분기별 업데이트를 작은 함수로 추출
overflow/에러를 Result 로 끌어올리기
필터링/매핑을 iterator 체인으로 분리

2단계: 전처리 로직을 `filter` / `filter_map` 으로 분리

현업에서는 이벤트가 String 한 줄로 들어오고 파싱이 필요합니다. 파싱 실패를 집계에서 분리하면, 집계 함수는 “정상 이벤트만” 다루게 되어 단순해집니다.

fn parse_event(line: &str) -> Option<Event> {
    // 예시: "user,amount,currency,ok" 형태라고 가정
    let mut it = line.split(',');
    let user_id = it.next()?.to_string();
    let amount_cents: i64 = it.next()?.parse().ok()?;
    let currency = it.next()?.to_string();
    let ok: bool = it.next()?.parse().ok()?;

    Some(Event { user_id, amount_cents, currency, ok })
}

fn aggregate_lines(lines: &[String]) -> Stats {
    lines
        .iter()
        .filter_map(|s| parse_event(s))
        .fold(Stats::default(), |mut acc, e| {
            if e.amount_cents > 1_000_000_00 {
                acc.has_outlier = true;
            }
            if e.ok {
                acc.ok_sum_cents += e.amount_cents;
                *acc.per_currency_sum.entry(e.currency).or_insert(0) += e.amount_cents;
            } else {
                acc.fail_count += 1;
            }
            acc
        })
}

여기서 중요한 포인트는 filter_map 이후에는 Event 가 소유권으로 이동한다는 점입니다. 그래서 e.currency.clone() 없이 e.currency 를 그대로 HashMap 키로 넣을 수 있습니다. 작은 차이지만, 대량 데이터에서는 할당/복사 비용을 줄이는 데 도움이 됩니다.

3단계: 에러를 삼키지 말고 `try_fold` 로 끌어올리기

Option 으로 파싱 실패를 무시하는 방식은 “실패가 정상인 경우”엔 괜찮지만, 운영에서는 실패율이 올라갔을 때 원인 추적이 어렵습니다. Rust에는 Iterator::try_fold 가 있어, 누적 중 에러가 발생하면 즉시 중단하고 Result 를 반환할 수 있습니다.

use std::collections::HashMap;

#[derive(Debug, Default)]
struct Stats {
    ok_sum_cents: i64,
    fail_count: u64,
    per_currency_sum: HashMap<String, i64>,
    has_outlier: bool,
}

#[derive(Debug)]
enum ParseError {
    BadFormat,
    BadAmount,
    BadOk,
}

fn parse_event_result(line: &str) -> Result<Event, ParseError> {
    let mut it = line.split(',');
    let user_id = it.next().ok_or(ParseError::BadFormat)?.to_string();
    let amount_raw = it.next().ok_or(ParseError::BadFormat)?;
    let amount_cents: i64 = amount_raw.parse().map_err(|_| ParseError::BadAmount)?;
    let currency = it.next().ok_or(ParseError::BadFormat)?.to_string();
    let ok_raw = it.next().ok_or(ParseError::BadFormat)?;
    let ok: bool = ok_raw.parse().map_err(|_| ParseError::BadOk)?;

    Ok(Event { user_id, amount_cents, currency, ok })
}

fn aggregate_lines_strict(lines: &[String]) -> Result<Stats, ParseError> {
    lines.iter().try_fold(Stats::default(), |mut acc, s| {
        let e = parse_event_result(s)?;

        if e.amount_cents > 1_000_000_00 {
            acc.has_outlier = true;
        }
        if e.ok {
            acc.ok_sum_cents += e.amount_cents;
            *acc.per_currency_sum.entry(e.currency).or_insert(0) += e.amount_cents;
        } else {
            acc.fail_count += 1;
        }

        Ok(acc)
    })
}

이 패턴은 “파이프라인 중간에 실패가 있으면 즉시 중단”이 자연스럽고, 호출자는 에러를 로그로 남기거나 재처리를 트리거할 수 있습니다. 운영 관점에서 이런 식의 실패 전파는 배치/스트리밍 모두에서 중요합니다.

대규모 데이터 파이프라인에서 재색인, 재처리 같은 후속 조치가 필요한 경우가 많은데, 이런 운영 이슈는 벡터 인덱싱 계열에서도 자주 등장합니다. 예를 들어 임베딩 드리프트를 감지하고 재색인을 설계할 때도 “정상/비정상 이벤트를 어떻게 정의하고 집계할지”가 핵심입니다. 관련 주제는 Milvus·Pinecone 임베딩 드리프트 탐지와 재색인 글이 참고가 됩니다.

4단계: 오버플로와 도메인 제약을 타입으로 고정하기

금액 합계는 생각보다 쉽게 overflow 납니다. Rust의 기본 산술은 release 빌드에서 overflow 시 wrap 될 수 있어(디버그에서는 panic), “조용한 데이터 오염”이 생길 수 있습니다. 집계에서는 다음 중 하나를 선택하는 게 안전합니다.

checked_add 로 overflow 시 에러 반환
saturating_add 로 상한 고정
더 큰 타입(i128)으로 승격

try_fold 와 checked_add 를 결합하면 overflow를 즉시 감지할 수 있습니다.

#[derive(Debug)]
enum AggError {
    Parse(ParseError),
    Overflow,
}

fn aggregate_lines_checked(lines: &[String]) -> Result<Stats, AggError> {
    lines.iter().try_fold(Stats::default(), |mut acc, s| {
        let e = parse_event_result(s).map_err(AggError::Parse)?;

        if e.ok {
            acc.ok_sum_cents = acc.ok_sum_cents.checked_add(e.amount_cents).ok_or(AggError::Overflow)?;
            let entry = acc.per_currency_sum.entry(e.currency).or_insert(0);
            *entry = entry.checked_add(e.amount_cents).ok_or(AggError::Overflow)?;
        } else {
            acc.fail_count += 1;
        }

        Ok(acc)
    })
}

이렇게 하면 “데이터가 커지면서 언젠가 터질 문제”를 조기에 잡아낼 수 있고, 장애가 나도 원인이 명확해집니다.

5단계: `scan` 으로 “중간 상태 스트림”을 만들기

fold 는 최종 결과만 주지만, 운영에서는 “시간순으로 누적이 어떻게 변했는지”를 보고 싶을 때가 많습니다. 이때 scan 이 유용합니다. scan 은 내부 상태를 유지하면서 각 단계의 산출물을 iterator로 흘려보냅니다.

예를 들어, 누적 합계가 임계치를 넘는 첫 시점을 찾는다면 다음처럼 작성할 수 있습니다.

fn first_crossing_index(events: &[Event], threshold: i64) -> Option<usize> {
    events
        .iter()
        .scan(0_i64, |sum, e| {
            if e.ok {
                *sum += e.amount_cents;
            }
            Some(*sum)
        })
        .position(|sum| sum >= threshold)
}

for 루프로도 가능하지만, scan 을 쓰면 “중간 상태를 스트림으로 만든 뒤, 그 스트림에 대해 position 같은 표준 연산을 적용”할 수 있어 조합성이 좋아집니다.

6단계: 윈도우 집계는 `fold` 보다 “자료구조 + iterator” 조합이 깔끔하다

상태없는 집계를 지향하더라도, 이동 평균이나 최근 N 개 기반 집계는 본질적으로 “제한된 상태”가 필요합니다. 이때도 상태를 루프 밖으로 흩뿌리기보다, 상태를 담는 구조체를 만들고 iterator로 결합하는 편이 유지보수에 유리합니다.

use std::collections::VecDeque;

#[derive(Default)]
struct MovingSum {
    window: VecDeque<i64>,
    sum: i64,
    n: usize,
}

impl MovingSum {
    fn new(n: usize) -> Self {
        Self { window: VecDeque::with_capacity(n), sum: 0, n }
    }

    fn push(&mut self, x: i64) -> i64 {
        self.window.push_back(x);
        self.sum += x;
        if self.window.len() > self.n {
            if let Some(front) = self.window.pop_front() {
                self.sum -= front;
            }
        }
        self.sum
    }
}

fn moving_sum(events: &[Event], n: usize) -> Vec<i64> {
    let mut ms = MovingSum::new(n);
    events
        .iter()
        .map(|e| if e.ok { e.amount_cents } else { 0 })
        .map(|x| ms.push(x))
        .collect()
}

여기서는 엄밀히 말해 클로저가 외부의 ms 를 변경하지만, 상태는 MovingSum 하나로 캡슐화되어 있고 테스트 단위도 명확합니다. “상태를 없애는 것”이 목표가 아니라, “상태를 통제 가능한 형태로 가두는 것”이 목표라고 보는 편이 현실적입니다.

성능 관점: iterator 체인은 느리지 않은가

Rust iterator는 대부분 인라이닝과 최적화가 잘 되어, 단순 루프와 거의 동일한 성능을 내는 경우가 많습니다. 오히려 다음 이점이 있습니다.

불필요한 임시 컬렉션을 만들지 않고 스트리밍 처리 가능
filter_map 등으로 early discard가 쉬움
try_fold 로 조기 종료가 자연스러움

다만 병목이 되는 지점은 보통 iterator 자체가 아니라

문자열 파싱/할당
HashMap 키 복사
캐시 미스

같은 곳입니다. 집계 리팩터링을 하다가 HashMap 업데이트가 병목이 되면, 키를 String 으로 매번 만들지 말고 입력에서 &str 를 빌려 쓰거나(수명 관리 필요), 사전 매핑을 두거나, 더 나은 인덱스 구조를 고려해야 합니다.

이런 “메타데이터 필터링/인덱스 설계가 성능을 좌우”하는 패턴은 벡터 DB에서도 유사하게 나타납니다. 예를 들어 Pinecone 메타데이터 필터 느림, 인덱스 설계로 5배 개선 글에서처럼, 필터 조건을 어떤 구조로 유지하느냐가 전체 처리량에 큰 영향을 줍니다.

리팩터링 체크리스트

집계 코드를 iterator·fold 로 옮길 때 아래 순서로 진행하면 시행착오가 줄어듭니다.

누적기 타입을 먼저 만든다: Stats 같은 구조체에 “최종적으로 필요한 값”을 다 넣는다.
전처리와 집계를 분리한다: 파싱/정규화는 map·filter_map 로, 집계는 fold 로.
실패를 정책으로 명시한다: 무시할 건 filter_map, 중단할 건 try_fold.
도메인 제약을 코드로 박는다: overflow는 checked_add, 입력 범위는 clamp 또는 검증.
상태가 필요한 집계는 캡슐화한다: 윈도우/중복 제거는 자료구조로 감싼다.

마무리

Rust에서 iterator 와 fold 는 단순한 문법 취향이 아니라, 집계 로직을 “입력 스트림과 누적기의 순수한 변환”으로 모델링하게 만들어 줍니다. 그 결과

상태 변경 지점이 명확해지고
에러/overflow 같은 비정상 흐름이 코드 구조로 드러나며
테스트와 확장이 쉬워집니다.

집계는 데이터 파이프라인, 관측(Observability), 검색/추천 인덱싱 등 거의 모든 백엔드 도메인의 기본기입니다. for 루프가 복잡해지기 시작했다면, filter_map 과 try_fold 를 먼저 도입해 “상태를 좁히고 실패를 드러내는” 방향으로 리팩터링해 보세요.

왜 “상태없는 집계”가 중요한가

문제 코드: for 루프에 흩어진 상태

1단계: 누적기를 명시하고 fold 로 옮기기

2단계: 전처리 로직을 filter / filter_map 으로 분리

3단계: 에러를 삼키지 말고 try_fold 로 끌어올리기