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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140

#![allow(
    clippy::arithmetic_side_effects,
    clippy::indexing_slicing,
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::panic
)]

use super::*;
use std::sync::atomic::{AtomicUsize, Ordering};

#[test]
fn test_scheduler_creation() {
    let cancel = CancellationToken::new();
    let scheduler = WorkerScheduler::new(cancel);
    assert_eq!(scheduler.task_count(), 0);
    assert_eq!(scheduler.enabled_count(), 0);
    assert!(scheduler.task_names().is_empty());
}

#[test]
fn test_scheduler_register_tasks() {
    let cancel = CancellationToken::new();
    let mut scheduler = WorkerScheduler::new(cancel);

    scheduler.register("health_check", Duration::from_secs(30), || {
        Box::pin(async {})
    });
    scheduler.register("cleanup", Duration::from_secs(300), || Box::pin(async {}));
    scheduler.register("version_check", Duration::from_secs(60), || {
        Box::pin(async {})
    });

    assert_eq!(scheduler.task_count(), 3);
    assert_eq!(scheduler.enabled_count(), 3);
}

#[test]
fn test_scheduler_task_names() {
    let cancel = CancellationToken::new();
    let mut scheduler = WorkerScheduler::new(cancel);

    scheduler.register("alpha", Duration::from_secs(1), || Box::pin(async {}));
    scheduler.register("beta", Duration::from_secs(1), || Box::pin(async {}));
    scheduler.register_optional(
        "gamma",
        Duration::from_secs(1),
        false,
        || Box::pin(async {}),
    );

    let names = scheduler.task_names();
    assert_eq!(names, vec!["alpha", "beta", "gamma"]);
}

#[test]
fn test_scheduler_disabled_tasks() {
    let cancel = CancellationToken::new();
    let mut scheduler = WorkerScheduler::new(cancel);

    scheduler.register("enabled_1", Duration::from_secs(1), || Box::pin(async {}));
    scheduler.register_optional("disabled_1", Duration::from_secs(1), false, || {
        Box::pin(async {})
    });
    scheduler.register("enabled_2", Duration::from_secs(1), || Box::pin(async {}));
    scheduler.register_optional("disabled_2", Duration::from_secs(1), false, || {
        Box::pin(async {})
    });

    assert_eq!(scheduler.task_count(), 4);
    assert_eq!(scheduler.enabled_count(), 2);
}

#[tokio::test]
async fn test_scheduler_run_and_shutdown() {
    let cancel = CancellationToken::new();
    let mut scheduler = WorkerScheduler::new(cancel.clone());

    let counter_a = Arc::new(AtomicUsize::new(0));
    let counter_b = Arc::new(AtomicUsize::new(0));

    let ca = counter_a.clone();
    scheduler.register("task_a", Duration::from_millis(10), move || {
        let ca = ca.clone();
        Box::pin(async move {
            ca.fetch_add(1, Ordering::Relaxed);
        })
    });

    let cb = counter_b.clone();
    scheduler.register("task_b", Duration::from_millis(20), move || {
        let cb = cb.clone();
        Box::pin(async move {
            cb.fetch_add(1, Ordering::Relaxed);
        })
    });

    // Spawn the scheduler and let it run briefly.
    let handle = tokio::spawn(async move {
        scheduler.run().await;
    });

    // Let tasks tick a few times.
    tokio::time::sleep(Duration::from_millis(100)).await;
    cancel.cancel();

    // Wait for scheduler to finish.
    handle.await.unwrap();

    // Both counters should have been incremented at least once.
    let a = counter_a.load(Ordering::Relaxed);
    let b = counter_b.load(Ordering::Relaxed);
    assert!(a >= 1, "task_a should have run at least once, got {a}");
    assert!(b >= 1, "task_b should have run at least once, got {b}");

    // task_a (10ms interval) should have run at least as often as task_b (20ms interval).
    assert!(
        a >= b,
        "task_a ({a}) should have run at least as often as task_b ({b})"
    );
}

#[tokio::test]
async fn test_scheduler_empty_run() {
    let cancel = CancellationToken::new();
    let scheduler = WorkerScheduler::new(cancel.clone());

    assert_eq!(scheduler.task_count(), 0);

    let handle = tokio::spawn(async move {
        scheduler.run().await;
    });

    // Give a moment then cancel.
    tokio::time::sleep(Duration::from_millis(20)).await;
    cancel.cancel();

    // Should complete without panic.
    handle.await.unwrap();
}