8.5 분산 쿼리(Query) 및 응답(Reply) 메커니즘

“세상에 존재하는 모든 로봇에게 물어본다. 현재 배터리가 20% 미만인 놈들만 즉시 응답하라!”

이런 명령을 기존의 REST API나 gRPC로 구현하려면, 중앙 서버(Node.js)가 DB를 뒤져 전 세계 모든 로봇의 IP 목록을 가져온 뒤, for 문을 돌리며 수백 번의 HTTP/RPC 요청을 병렬로 쏘아야 한다. 여기서 한 대라도 네트워크가 끊겨 타임아웃이 나면? 예외 처리(Try-Catch) 지옥이 열린다.

Zenoh의 Query / Reply 시스템은 TypeScript 개발자에게 ’IP 주소’라는 개념 자체를 망각(Oblivion) 하게 해준다. 브라우저에서 날아간 단 한 줄의 session.get("fleet/*/status") 쿼리는 그 즉시 Zenoh 라우터에 의해 우주망 단위의 멀티캐스트 파동으로 변해 로봇들에게 흩뿌려지며, 대답을 한 놈들의 응답만 주워 담아 안전하게 비동기 이터레이터(AsyncIterator)로 예쁘게 반환한다.
이 챕터에서는 서버리스(Serverless) 분산 검색망을 구축하는 마법 같은 무기를 다룬다.

1. 기존 RPC(Remote Procedure Call) 및 REST API와의 차별성

웹 개발자가 Zenoh의 쿼리(Query)를 처음 접할 때 가장 범하기 쉬운 오류는 이것을 “그냥 가벼운 HTTP GET 요청” 비스무리한 것으로 치부하는 것이다. 근본적인 아키텍처 결이 다르다.

1.0.1 [인스펙션] 분산 쿼리가 기존 HTTP/RPC를 박살 내는 3가지 이유

1. [위치 투명성] 내가 물어보는 대상이 ’서버(Server)’가 아니다
REST API로 GET 192.168.0.5/api/weather 를 때린다면, 그 IP에 기계가 살아있어야만 대답을 얻는다.
Zenoh 쿼리는 데이터의 이름을 부른다 get("weather/seoul"). 특정 기계가 죽었더라도, 옆에 있던 다른 기계(DB 백업, 다른 캐시 노드)가 이 이름을 들었다면 알아서 대답해 준다.

2. [다대일(1:N) 응답] 하나의 질문에 100명이 동시에 대답한다
gRPC나 HTTP는 본질적으로 ’1:1 양방향 통신’이다. 브라우저가 질문 하나를 던지면 응답(Response) 본문도 1개다.
Zenoh 쿼리는 fleet/*/location 이라고 와일드카드를 섞어 던지면, 100대의 로봇이 “나여깄소!” 하고 100개의 독립적인 Reply 패킷을 우르르 던진다. TS 런타임은 이걸 for await 루프로 하나씩 주워 먹기만 하면 된다.

3. [지연 네트워크 방어] 타임아웃(Timeout)의 우아함
HTTP에서 로봇 1대가 타임아웃에 걸려 30초 내내 돌아오지 않으면, Promise.all 에 묶인 다른 99대 로봇의 정상적인 응답까지 클라이언트 화면에 출력되지 못하는 ‘아키텍처 스턴(Stun)’ 현상을 겪는다.
Zenoh는 타임아웃(예: 3초)을 주면, 3초 동안 도착한 대답들만 살려두고 나머지는 조용히 버려버린다. 이로 인해 대시보드의 렌더링(프레임율) 방어가 100% 보장된다.

2. Queryable 노드 등록 및 원격 요청 수신 파이프라인

“질문을 던지는 자가 있다면, 질문에 대답하는 자치구(Zone)가 있어야 한다.”

이것이 TypeScript로 마이크로서비스(Microservice)를 짜는 순간이다. Express.js로 라우터(Router)를 짜듯, 특정 경로의 질문이 날아왔을 때 대답을 즉조해서(Generate) 날려주는 Queryable 을 띄운다.

2.0.1 [Runbook] 마이크로 응답기(Queryable) 전진 배치

import * as zenoh from "@eclipse-zenoh/zenoh";
import * as os from "os";

async function bootServerlessNode() {
    const session = await zenoh.open();

    // 1. "server/status"를 물어보면 내가 대답하겠다고 선언!
    // -> 이 순간 이 TS 앱은 분산 라우팅망에 '응답 기지국'으로 정식 등재된다.
    const queryable = await session.declareQueryable(
        "server/status", 
        (query: zenoh.Query) => {
            console.log(`[접수] 누군가 상태를 묻는다! 타겟: ${query.keyExpr}`);
            
            // 2. 가벼운 OS 매트릭스 정보를 가공한다.
            const stats = {
                cpuUsage: os.loadavg()[0],
                freeMem: os.freemem(),
                upTime: os.uptime()
            };
            
            // 3. 질문자에게 그 즉시 대답(Reply)을 강제 송신한다!
            // query.reply() 는 HTTP res.send() 와 100% 같은 역할을 한다.
            const payload = Buffer.from(JSON.stringify(stats));
            query.reply(query.keyExpr, payload);
            
            console.log(">> 대답 정상 발송 완료.");
        }
    );

    console.log(">> 서버 상태 응답기(Queryable) 온라인.");
    
    // 계속 켜둔다
    // queryable.undeclare() 하기 전까지 영원히 동작
}

이 코드를 AWS 람다(Lambda)나 수십 대의 PM2 워커(Worker)에 복사해서 띄우면, 클라이언트가 server/status 로 질문 한 번 던졌을 때 수십 개의 서버 스탯(Stats)들이 우르르 돌아오는 완벽한 로드밸런싱/분산 집계(Aggregation) 팜(Farm)이 무코드(No-code) 환경설정으로 완성된다.

3. Get 요청을 통한 분산 데이터 수집(Consolidation)

앞 장에서 수십 대의 서버 스탯 응답기를 띄워 뒀으니, 이제 중앙 대시보드(React 등) 쪽에서 그 정보들을 “한 번의 마우스 클릭“으로 싹쓸이해 오는 스위핑(Sweeping) 전술이다.

3.0.1 [Runbook] 다중 타겟 지향성 전파 (Sweeping)

React 의 버튼 onClick 핸들러나 Node.js의 스케줄러(Cron) 안에서 실행될 코드다.

import * as zenoh from "@eclipse-zenoh/zenoh";

async function fetchAllStats(session: zenoh.Session) {
    console.log("📡 스캔 개시...");
    
    const statsList: any[] = [];
    const decoder = new TextDecoder("utf-8");

    // [전술 타격] session.get() 은 와일드카드를 섞을 때 가장 무서운 무기가 된다.
    // get 메서드는 즉시 AsyncIterable(비동기 제너레이터) 를 반환한다!
    const replies = await session.get("server/**", zenoh.Queueing.default());
    
    // 뼛속까지 파이썬/JS 다운 'for await' 순회 구문!
    // 네트워크 너머의 100대 서버 중 먼저 답장 온 놈부터 순차적으로 이 루프 안쪽을 돈다.
    for await (const reply of replies) {
        try {
            // reply는 성공(Ok)과 에러(Err)의 이중 타입을 갖는다.
            let sample = reply.ok; // 성공했다고 강단
            
            const rawText = decoder.decode(sample.payload);
            const data = JSON.parse(rawText);
            
            console.log(`[건져올림: ${sample.keyExpr}] CPU: ${data.cpuUsage}`);
            statsList.push({ source: sample.keyExpr, data });
            
        } catch (e) {
            console.warn("에러를 보낸 노드 존재 혹은 파싱 실패", e);
        }
    }

    // [중요!] 이 영역에 도달했다는 것은, 
    // "망에 흩어진 모든(All) 서버가 대답을 완전히 끝마쳤다" 
    // 혹은 "설정한 Timeout 시간이 다 흘러서 타임오버 되었다"라는 뜻이다.
    console.log(`📡 스캔 완료. 총 ${statsList.length}대 기계 확인됨.`);
    return statsList;
}

오직 이 우아한 for await 패턴 덕분에, UI 렌더링 스레드는 수십 개의 데이터를 기다리느라 동결창(Freeze)에 빠지지 않으며, 대답이 오는 족족 즉시 React 상태(setState)를 점진적으로 업데이트 하는 화려한 애니메이션 구동이 가능해진다.

4. 다중 응답(Multiple Replies)의 비동기 순회(Async Iteration) 처리

왜 굳이 for await 구문(Async Iteration)을 써야 할까? replies.toArray() 같은 걸로 배열에 한 방에 모아서 주면 안 되나?

4.0.1 [아키텍처 인스펙션] 메모리 OOM 방어를 위한 Async Iteration

비동기 이터레이터 아키텍처는 TS가 선택할 수 있는 가장 가볍고 파괴적인 로드 밸런싱(Load Balancing) 기술이다.

만약 지구상에 10만 대의 온도 센서가 있고 일제히 대답을 날렸을 때, get() 이 배열을 리턴했다면 프론트엔드의 V8 자바스크립트 엔진 힙 메모리는 배열에 데이터를 쌓다가 터져버린다(OOM, Out of Memory).

하지만 AsyncIterator는 데이터가 한 개 도착할 때마다 V8 메모리에 반짝 생겼다가, for 문이 한 바퀴 도는 순간 가비지 컬렉터(GC)에 의해 파괴된다.

// --- 가장 치명적인 안티 패턴 (OOM의 주범) ---
async function badAggregation(session) {
    const allData = [];
    const replies = await session.get("sensors/**", zenoh.Queueing.default());
    
    // 메모리에 10만 개를 무식하게 다 처넣음.
    for await (const reply of replies) {
        allData.push(reply.ok.payload);
    }
    
    // 이 시점에 브라우저는 이미 뻗어서 하얗게 화면이 날아감.
    renderToTable(allData); 
}

// --- 베스트 프랙티스: 스트리밍 렌더링 ---
async function goodStreaming(session) {
    const replies = await session.get("sensors/**", zenoh.Queueing.default());
    
    let sum = 0;
    let count = 0;
    
    for await (const reply of replies) {
        // [핵심] 배열에 저장(Store)하지 않고, 즉시 값만 계산(Reduce)하고 버려버린다!
        const val = parseInt(decoder.decode(reply.ok.payload));
        sum += val;
        count++;
        
        // 1,000개 모일때마다 화면 한 번씩 갱신 (점진적 렌더링)
        if (count % 1000 === 0) {
            updateReactUI(sum / count); 
        }
    }
}

Zenoh는 TypeScript의 문법 중 가장 고도화된 스펙인 ES2018 Async Generators 의 속성을 100% 이해하고 C++ 메모리 파이프에서 한 땀씩 한 땀씩 바이트를 짜내어 준다. 당신이 이 철학을 이해하면 무한(Infinity)의 데이터도 1MB 램으로 처리할 수 있다.

5. 선택자(Selector)를 이용한 타겟 쿼리 및 응답 필터링

1,000대의 로봇 중 1대만 살짝 응답을 얻고 싶거나, 백엔드 데이터베이스 캐시(Cache)에만 살짝 물어봐서 대역폭을 아끼고 싶다면? Zenoh의 쿼리 타겟(Target) 옵션을 튜닝한다.

5.0.1 [Runbook] 데이터 검색 스코프 통제 전술

zenoh.QueryTarget Enum 객체를 응용한다.

import * as zenoh from "@eclipse-zenoh/zenoh";

async function smartQuery(session: zenoh.Session) {
    
    // 전술 1. 가장 가까운 놈 1명만 대답하라 (BEST_MATCHING)
    // -> 내가 한국에 있다면 서울 지사의 캐시 서버가 대답하는 순간, 
    // 미국에 있는 로봇에게까지 이 질문 패킷이 건너가지 않고 즉각 차단된다. (트래픽 최적화)
    const optsBest = {
        target: zenoh.QueryTarget.BEST_MATCHING
    };
    
    // 전술 2. 파라미터(Query Params) 부착 타격
    // HTTP 의 /api/robot?cmd=stop&auth=token 과 100% 일치하는 문법이다.
    const optsWithParam = {
        target: zenoh.QueryTarget.ALL,
        parameters: "cmd=reboot&force=true" // Queryable 노드쪽 로직 제어용
    };
    
    console.log("-> 가장 가까운 1기 타겟팅 쿼리 전송");
    const replies = await session.get("robot/*/status", zenoh.Queueing.default(), optsBest);
    
    for await (const reply of replies) {
        // 단 1개만 찍히고 for문은 바로 탈출하게 된다.
        console.log(`응답 수신: ${reply.ok.keyExpr}`);
    }
}

// ---- Queryable 측 (수신부) 에서의 파라미터 파싱 ----
async function smartQueryable(session: zenoh.Session) {
    await session.declareQueryable("robot/*/status", (query: zenoh.Query) => {
        // 클라이언트가 보낸 parameters: "cmd=reboot&force=true" 가 읽힌다!
        const params = query.parameters;
        
        if (params.includes("cmd=reboot")) {
            console.log(">> 앗! 조종자가 리부팅을 지시했다!");
        }
        
        query.reply(query.keyExpr, Buffer.from("ok"));
    });
}

와일드카드와 QueryTarget.BEST_MATCHING, 그리고 로직 분기용 parameters를 버무리면, TypeScript 하나로 사실상 ‘글로벌 분산 GraphQL’ 엔진이나 ‘NoSQL 문서 라우팅 시스템’ 구조를 흉내 낼 수 있다는 엄청난 아키텍처적 가능성을 확보한 것이다.

6. 쿼리 타임아웃 설정 및 지연(Latency) 보상

프론트엔드 React 개발자라면 axios 에 타임아웃을 걸어놓고 1초 만에 답이 없으면 사용자에게 스켈레톤 로딩(Skeleton UI)을 그리거나 에러 팝업을 띄우는 게 일상일 것이다. Zenoh 도 본질은 네트워크 통신이므로 똑같은 타임오버 관리가 필수다.

6.0.1 [Runbook] V8 루프 동결 방어 전술

Zenoh의 타임아웃은 “Exception(에러)” 을 강제로 던지지 않는다!
대신 묵묵히 타임아웃 시간까지만 기다리다가, “지금까지 모은 대답만 넘겨주고 for 반복문을 조용히 폭파(종료)” 시켜버리는 매우 세련된 기법을 취한다.

import * as zenoh from "@eclipse-zenoh/zenoh";

async function timeoutDefensiveQuery(session: zenoh.Session) {
    console.log("⏱️ 로봇 상태 조회 개시 (타임아웃 한도: 1.5초)");
    
    // [방어막 1] 쿼리 옵션 객체에 timeout 프로퍼티를 준다. (단위: 문자열로 줄 수도 있다)
    // 혹은 Config의 기본 connect/timeout 을 따른다.
    // 주의: TypeScript API에서는 옵션 전달이 버전마다 조금 다르다. (숫자형 ms 등)
    // 통상 conf.insertJson5("connect/timeout", "1500") 세팅이 가장 지배적이다.

    const tStart = Date.now();
    let replyCount = 0;
    
    // 만약 라우터망에 10대의 로봇 중 2대가 죽어있다면?
    const replies = await session.get("robot/*/battery", zenoh.Queueing.default(), {
        timeout: 1500 // 1.5초 뒤 컷오프!
    });
    
    for await (const reply of replies) {
        replyCount++;
    }
    
    const elapsed = Date.now() - tStart;
    
    // [지연 분석]
    // 10대가 다 살아서 0.1초만에 대답이 다 도달했다면: elapsed 는 약 100ms 다.
    // 2대가 죽었다면: Zenoh는 그 2대를 1.5초(타임아웃) 꽉 채워 기다린 후 포기한다. 
    // 따라서 elapsed 는 무조건 1500ms(1.5초) 이상이 찍히게 된다!
    console.log(`수집 종료. 모은 대답: ${replyCount}개 / 소요시간: ${elapsed}ms`);
    
    if (elapsed >= 1500 && replyCount < 10) {
        // 이 판정 트리(Decision Tree)를 통해 프론트엔드는
        // "아! 현재 전체 10대 중 누군가의 전원이 꺼져있구나" 라고 유추하고 
        // 화면에 노란색 Warning UI를 띄워준다.
        console.warn("⚠️ 네트워크 지연 혹은 결손(Loss) 단말 감지!");
    }
}

이 패턴이 바로 “응답하는 자만 대답한다“는 분산 네트워크의 불확실성(Entropy)을 프론트엔드의 확고한 렌더링 UX로 찍어누르는 1티어 팩토리(Factory) 데브옵스 프로그래머의 스킬이다.