conproxy 0.2.1

Cache proxy server for heterogeneous RAG/vector search backends with cascade, scope, and MCP
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

//! gRPC compression tests — validates gzip and zstd negotiation on the proxy.
//!
//! Connects directly to the gRPC port (HTTP port - 1) and verifies that
//! queries succeed with each compression encoding. The proxy must have
//! `accept_compressed` + `send_compressed` enabled for Gzip and Zstd.

use crate::helpers::constants::{category_enabled, proxy_url};
use crate::helpers::report::TestReport;
use crate::run_test;

use conproxy::proxy::grpc::proto::{self, search_service_client::SearchServiceClient};
use tonic::codec::CompressionEncoding;
use tonic::transport::Channel;

/// Derive the gRPC URL from the HTTP base URL (port - 1).
fn grpc_url() -> String {
    // proxy_url() is "http://127.0.0.1:8081", gRPC is on 8080
    let http_port: u16 = proxy_url()
        .rsplit(':')
        .next()
        .and_then(|p| p.parse().ok())
        .unwrap_or(8081);
    format!("http://127.0.0.1:{}", http_port - 1)
}

fn make_query_request() -> proto::QueryRequest {
    proto::QueryRequest {
        query: "gzip compression test".into(),
        top_k: 3,
        ..Default::default()
    }
}

pub fn run(report: &mut TestReport) {
    if !category_enabled("compression") {
        return;
    }

    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("Failed to build tokio runtime");

    let url = grpc_url();

    // Test 1: gRPC query with gzip compression
    run_test!(report, "compression", "gRPC query with gzip", {
        let url = url.clone();
        rt.block_on(async {
            let channel = Channel::from_shared(url)
                .expect("invalid URL")
                .connect()
                .await
                .expect("Failed to connect to gRPC");

            let mut client = SearchServiceClient::new(channel)
                .send_compressed(CompressionEncoding::Gzip)
                .accept_compressed(CompressionEncoding::Gzip);

            let resp = client
                .query(make_query_request())
                .await
                .expect("gzip query failed");

            let inner = resp.into_inner();
            assert!(
                !inner.results.is_empty() || inner.cache_status >= 0,
                "Expected a valid response with gzip compression"
            );
        });
    });

    // Test 2: gRPC query with zstd compression
    run_test!(report, "compression", "gRPC query with zstd", {
        let url = url.clone();
        rt.block_on(async {
            let channel = Channel::from_shared(url)
                .expect("invalid URL")
                .connect()
                .await
                .expect("Failed to connect to gRPC");

            let mut client = SearchServiceClient::new(channel)
                .send_compressed(CompressionEncoding::Zstd)
                .accept_compressed(CompressionEncoding::Zstd);

            let resp = client
                .query(make_query_request())
                .await
                .expect("zstd query failed");

            let inner = resp.into_inner();
            assert!(
                !inner.results.is_empty() || inner.cache_status >= 0,
                "Expected a valid response with zstd compression"
            );
        });
    });

    // Test 3: gRPC query without compression (backward compat)
    run_test!(report, "compression", "gRPC query without compression", {
        let url = url.clone();
        rt.block_on(async {
            let channel = Channel::from_shared(url)
                .expect("invalid URL")
                .connect()
                .await
                .expect("Failed to connect to gRPC");

            let mut client = SearchServiceClient::new(channel);

            let resp = client
                .query(make_query_request())
                .await
                .expect("uncompressed query failed");

            let inner = resp.into_inner();
            assert!(
                !inner.results.is_empty() || inner.cache_status >= 0,
                "Expected a valid response without compression"
            );
        });
    });
}