클라우드 Mac에서 AsyncStream 버퍼와 종료 동작 검증하기

다운로드 진행률, 기기 상태, 알림 콜백을 AsyncStream에 연결했다고 해서 화면이 정상적으로 갱신되는 것만으로 이벤트 스트림이 올바르게 동작한다고 볼 수는 없습니다. 소비자가 조금만 느려져도 버퍼에서 값이 유실될 수 있고, 화면을 벗어난 뒤에도 원본 구독이 계속 유지될 수 있습니다. VpsKeep 클라우드 Mac에서는 독립적인 Swift 테스트로 이러한 동작을 먼저 검증한 뒤 앱에 연결할 수 있습니다. 경쟁 조건을 찾으려고 화면을 반복해서 클릭할 필요가 없습니다.

버려도 되는 이벤트부터 정의하기

진행률은 대개 최신 값만 있으면 되므로 bufferingNewest가 적합합니다. 먼저 도착한 이벤트를 보존해야 한다면 bufferingOldest를 선택할 수 있습니다. 하지만 어느 쪽도 무손실 큐는 아닙니다. 주문 변경이나 파일 쓰기 완료 확인처럼 유실되면 안 되는 데이터에는 별도의 확인 또는 영속화 메커니즘이 필요합니다.

기본값인 무제한 버퍼도 만능 해결책은 아닙니다. 생산 속도가 소비 속도보다 계속 빠르면 대기 중인 요소가 메모리를 지속적으로 점유합니다. 먼저 버퍼 용량, 버퍼가 찼을 때 보존할 값, 구독을 해제할 시점이라는 세 가지 제약을 명시하세요. 용량은 업무상 허용할 수 있는 적체량을 기준으로 정해야 합니다. 시스템의 남은 메모리에 맞춰 임의로 늘려서는 안 됩니다.

AsyncStream의 yield는 소비자가 버퍼에 빈자리를 만들 때까지 기다리지 않습니다. 크기가 제한된 버퍼가 제공하는 것은 값 폐기 정책이지, 자동 백프레셔가 아닙니다.

이 글에서는 용량이 2인 정수 스트림으로 동작 규칙을 검증합니다. 정수는 이벤트의 순번을 나타내며, 성능 측정값이 아닙니다.

독립적인 테스트 작업 공간 만들기

쓰기 가능한 디렉터리에서 아래 명령을 실행해 현재 선택된 개발 툴체인으로 테스트 패키지를 만드세요. Swift 버전을 기록하고, 이후 재실행할 때도 같은 툴체인을 사용하세요. 패키지에는 서비스용 비밀 키가 필요하지 않으며, 실제 서비스에 연결할 필요도 없습니다.

mkdir StreamContract
cd StreamContract
swift --version
swift package init --type library
mkdir -p Tests/StreamContractTests

다음 절의 코드를 Tests/StreamContractTests/StreamContractTests.swift에 저장하세요. 초기화 과정에서 같은 이름의 파일이 생성되었다면 그대로 덮어쓰면 됩니다. 템플릿이 생성한 다른 테스트는 남겨 두어도 됩니다. 아래 실행 명령은 이 클래스만 선택합니다.

버퍼가 넘쳤을 때의 결과를 확정하기 위해 테스트에서는 먼저 세 요소를 동기적으로 생산한 다음 소비를 시작합니다. 생산자와 소비자가 동시에 스케줄링을 두고 경쟁하게 만들거나, “1초 기다리기”로 버퍼가 찼다고 추측하지 마세요. 여기서 검증하는 대상은 컨테이너의 동작 계약이지, 스레드의 실행 속도가 아닙니다.

폐기된 값과 보존된 순서를 함께 확인하기

최종 배열만 검증해서는 충분하지 않습니다. 세 번째 yield의 반환값도 확인해야 합니다. 최신 요소를 보존할 때는 이전 요소인 1이 폐기되고, 가장 먼저 도착한 요소를 보존할 때는 새 요소인 3이 폐기됩니다. finish를 호출한 뒤에도 이미 버퍼에 들어간 값은 끝까지 읽을 수 있습니다.

import XCTest

final class StreamContractTests: XCTestCase {
    func testNewestKeepsLastTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingNewest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 1)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [2, 3])

        switch pair.continuation.yield(4) {
        case .terminated:
            break
        default:
            XCTFail("Expected a terminated stream")
        }
    }

    func testOldestKeepsFirstTwo() async {
        let pair = AsyncStream<Int>.makeStream(
            bufferingPolicy: .bufferingOldest(2)
        )
        pair.continuation.yield(1)
        pair.continuation.yield(2)

        switch pair.continuation.yield(3) {
        case .dropped(let value):
            XCTAssertEqual(value, 3)
        default:
            XCTFail("Expected a dropped value")
        }

        pair.continuation.finish()
        var received: [Int] = []
        for await value in pair.stream {
            received.append(value)
        }
        XCTAssertEqual(received, [1, 2])
    }
}

실제 운영 코드에서 .dropped를 처리할 때는 폐기 횟수를 누적하거나 업무상 보상 처리를 실행해야 합니다. 진행률 이벤트마다 동기적으로 로그를 기록하지 마세요. .terminated를 받으면 같은 continuation으로 계속 재시도하지 말고, 더 이상 값을 전달하지 않아야 합니다.

취소 시 구독 해제 진입점이 호출되는지 검증하기

같은 파일에 아래 확장을 추가하세요. 스트림에 요소가 없으므로 소비자는 대기하거나 취소에 반응할 수밖에 없습니다. 종료 알림은 테스트 expectation으로 검증하며, 고정된 시간 동안 잠드는 방식에 의존하지 않습니다. 2초는 실패 판정을 위한 타임아웃일 뿐, 실행 단계 사이의 대기 시간이 아닙니다.

extension StreamContractTests {
    func testCancellationReportsTermination() async {
        let ended = expectation(description: "Stream cancelled")
        let pair = AsyncStream<Int>.makeStream()

        pair.continuation.onTermination = { reason in
            switch reason {
            case .cancelled:
                ended.fulfill()
            case .finished:
                XCTFail("Unexpected normal finish")
            @unknown default:
                XCTFail("Unknown termination reason")
            }
        }

        let consumer = Task {
            for await _ in pair.stream {}
        }

        consumer.cancel()
        await consumer.value
        await fulfillment(of: [ended], timeout: 2)
    }
}

실제 어댑터에서는 expectation을 충족시키는 콜백 대신 구독 토큰의 해제 메서드를 호출해야 합니다. 종료 처리가 메인 스레드에서 실행된다는 보장은 없으므로 여기서 UI를 직접 갱신해서는 안 됩니다. 구독 해제는 스레드 안전성과 멱등성을 갖춰야 합니다. 또한 어댑터 내부 잠금을 잡은 상태에서 콜백을 다시 유발할 수 있는 구독 해제 함수를 호출하지 마세요.

구독을 설정하는 과정의 경쟁 조건도 고려해야 합니다. 종료가 먼저 발생하고 토큰이 나중에 반환될 수 있습니다. 어댑터는 “이미 닫힘” 상태를 저장해야 하며, 늦게 도착한 토큰은 닫힌 상태임을 확인하는 즉시 해제해야 합니다. onTermination만 설정하고 이 경로를 놓쳐서는 안 됩니다.

파이프라인에 통합하고 인계 전 확인하기

테스트를 실행하고 출력을 파이프라인의 일반 테스트 기록에 저장하세요. 입력이 고정되어 있으므로 반복 실행해도 같은 배열과 폐기된 요소가 나와야 합니다. 간헐적으로만 통과한다면 대개 테스트에 실제 타이머나 공유 이벤트 소스가 섞여 있다는 뜻입니다.

swift test --filter StreamContractTests

앱에 다시 연결하기 전에 네 가지를 확인하세요. 정상적인 finish에서도 구독이 해제되는지, 태스크 취소 시에도 구독이 해제되는지, 구독 해제 도중 늦게 도착한 콜백이 무시되는지, 종료 이후에 도착한 토큰이 즉시 해제되는지 확인해야 합니다. 먼저 이벤트를 수동으로 전달할 수 있는 가짜 이벤트 소스로 이 경로들을 테스트한 다음 실제 소스를 테스트하세요.

브로드캐스트를 흉내 내려고 여러 태스크가 같은 스트림을 경쟁해서 읽게 하지 마세요. 구독자가 여러 명이어야 한다면 명시적인 분배 계층을 설계하고, 구독자별 버퍼 정책도 각각 정해야 합니다. 마지막으로 테스트 코드, 툴체인 버전, 업무상 동작 계약 설명을 함께 남겨 두세요. 그래야 나중에 이벤트 소스를 연결하는 사람이 어떤 값 폐기가 설계된 동작이고 어떤 것이 장애인지 알 수 있습니다.

자주 묻는 질문