1pub mod aggregator;
2pub mod circuit_breaker;
3pub mod freshness_filter;
4
5#[cfg(any(feature = "borsh", feature = "schemars"))]
6use alloc::{format, string::ToString};
7
8use crate::Price;
9use aggregator::method::Aggregate;
10pub use aggregator::Aggregator;
11use circuit_breaker::{
12 CircuitBreakerOutcome, CircuitBreakerRule, CircuitBreakerSet, PriceAcceptance,
13};
14pub use freshness_filter::FreshnessFilter;
15
16use templar_primitives::time::Nanoseconds;
17
18serialize! {
19 #[derive(Debug, Clone, PartialEq, Eq)]
20 pub struct WeightedSource<S> {
21 pub source: S,
22 pub weight: u32,
23 }
24}
25
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub enum ResolveError {
28 Aggregation(aggregator::method::Error),
29 CircuitBreaker(circuit_breaker::CircuitBreakerError),
30}
31
32impl From<aggregator::method::Error> for ResolveError {
33 fn from(error: aggregator::method::Error) -> Self {
34 Self::Aggregation(error)
35 }
36}
37
38impl From<circuit_breaker::CircuitBreakerError> for ResolveError {
39 fn from(error: circuit_breaker::CircuitBreakerError) -> Self {
40 Self::CircuitBreaker(error)
41 }
42}
43
44impl core::fmt::Display for ResolveError {
45 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
46 match self {
47 Self::Aggregation(error) => write!(f, "aggregation failed: {error}"),
48 Self::CircuitBreaker(error) => write!(f, "circuit breaker failed: {error}"),
49 }
50 }
51}
52
53#[cfg(feature = "std")]
54impl std::error::Error for ResolveError {}
55
56impl<S> WeightedSource<S> {
57 pub fn new(source: impl Into<S>, weight: u32) -> Self {
58 Self {
59 source: source.into(),
60 weight,
61 }
62 }
63}
64
65serialize! {
66 #[derive(Debug, Clone, PartialEq, Eq)]
67 pub struct Proxy<S> {
68 pub aggregator: Aggregator<S>,
69 pub freshness_filter: FreshnessFilter,
70 }
71}
72
73impl<S> Proxy<S> {
74 #[must_use]
75 pub fn new(aggregator: Aggregator<S>, freshness_filter: FreshnessFilter) -> Self {
76 Self {
77 aggregator,
78 freshness_filter,
79 }
80 }
81
82 #[must_use]
83 pub fn median_low(
84 sources: impl IntoIterator<Item = S>,
85 freshness_filter: FreshnessFilter,
86 ) -> Self {
87 Self::new(Aggregator::median_low(sources), freshness_filter)
88 }
89
90 #[must_use]
91 pub fn priority(
92 sources: impl IntoIterator<Item = S>,
93 freshness_filter: FreshnessFilter,
94 ) -> Self {
95 Self::new(Aggregator::priority(sources), freshness_filter)
96 }
97
98 #[must_use]
99 pub fn with_freshness_filter(mut self, freshness_filter: FreshnessFilter) -> Self {
100 self.freshness_filter = freshness_filter;
101 self
102 }
103
104 pub fn sources(&self) -> aggregator::SourceIter<'_, S> {
105 self.aggregator.sources()
106 }
107
108 pub fn resolve<I, R>(
109 &self,
110 circuit_breakers: &mut CircuitBreakerSet<R>,
111 prices: I,
112 now: Nanoseconds,
113 ) -> Result<CircuitBreakerOutcome<PriceAcceptance>, ResolveError>
114 where
115 I: IntoIterator<Item = Option<Price>>,
116 I::IntoIter: ExactSizeIterator<Item = Option<Price>>,
117 R: CircuitBreakerRule,
118 {
119 let price = self.aggregate(prices, now)?;
120 let acceptance = circuit_breakers.try_accept_price(price, now)?;
121 Ok(acceptance)
122 }
123
124 fn aggregate<I>(&self, prices: I, now: Nanoseconds) -> Result<Price, aggregator::method::Error>
125 where
126 I: IntoIterator<Item = Option<Price>>,
127 I::IntoIter: ExactSizeIterator<Item = Option<Price>>,
128 {
129 self.aggregator.aggregate(prices.into_iter().map(|price| {
130 price
131 .filter(Price::has_strictly_positive_confidence_interval)
132 .filter(|price| self.freshness_filter.accepts(price, now))
133 }))
134 }
135}
136
137#[cfg(test)]
138mod tests {
139 use alloc::vec;
140 use alloc::vec::Vec;
141 use rstest::rstest;
142 use templar_primitives::Nanoseconds;
143
144 use crate::{
145 proxy::{
146 aggregator::method::{median::MedianLow, Error},
147 circuit_breaker::{
148 CircuitBreaker, CircuitBreakerSet, CircuitBreakerSetConfig, PriceBlockedReason,
149 StepwiseChange,
150 },
151 Aggregator, FreshnessFilter, Proxy, ResolveError, WeightedSource,
152 },
153 Price,
154 };
155 use templar_primitives::Decimal;
156
157 fn price(value: i64, conf: u64, publish_time_s: u64) -> Price {
158 Price {
159 price: value,
160 conf,
161 expo: -6,
162 publish_time_ns: Nanoseconds::from_secs(publish_time_s),
163 }
164 }
165
166 fn median_proxy(freshness_filter: FreshnessFilter, min_sources: u32) -> Proxy<&'static str> {
167 let mut aggregator = MedianLow::new([
168 WeightedSource::new("source-a", 1),
169 WeightedSource::new("source-b", 1),
170 ]);
171 aggregator.min_sources = min_sources;
172
173 Proxy::new(Aggregator::MedianLow(aggregator), freshness_filter)
174 }
175
176 fn priority_proxy(freshness_filter: FreshnessFilter) -> Proxy<&'static str> {
177 Proxy::priority(["source-a", "source-b"], freshness_filter)
178 }
179
180 #[test]
181 fn resolve_applies_min_sources_after_filtering() {
182 let proxy = median_proxy(
183 FreshnessFilter {
184 max_age_ns: Some(Nanoseconds::from_secs(500)),
185 max_clock_drift_ns: None,
186 },
187 2,
188 );
189 let prices = vec![
190 Some(price(1_000_000, 0, 1_000)),
191 Some(price(2_000_000, 0, 100)),
192 ];
193
194 let error = proxy
195 .resolve(
196 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
197 prices,
198 Nanoseconds::from_secs(1_000),
199 )
200 .unwrap_err();
201
202 assert!(matches!(
203 error,
204 ResolveError::Aggregation(Error::TooFewValidSources {
205 expected: 2,
206 actual: 1,
207 })
208 ));
209 }
210
211 #[test]
212 fn resolve_median_applies_min_sources_after_invalid_price_filtering() {
213 let proxy = median_proxy(FreshnessFilter::new(None, None), 2);
214 let prices = vec![
215 Some(price(1_000_000, 1_000_000, 1_000)),
216 Some(price(2_000_000, 0, 1_000)),
217 ];
218
219 let error = proxy
220 .resolve(
221 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
222 prices,
223 Nanoseconds::from_secs(1_000),
224 )
225 .unwrap_err();
226
227 assert!(matches!(
228 error,
229 ResolveError::Aggregation(Error::TooFewValidSources {
230 expected: 2,
231 actual: 1,
232 })
233 ));
234 }
235
236 #[rstest]
237 #[case::one_under_included(501, 1000, 500, true)]
238 #[case::exactly_at_limit_included(500, 1000, 500, true)]
239 #[case::one_over_excluded(499, 1000, 500, false)]
240 fn resolve_max_age_boundary(
241 #[case] publish_time_s: i64,
242 #[case] now_s: i64,
243 #[case] max_age_s: u64,
244 #[case] included: bool,
245 ) {
246 let proxy = median_proxy(
247 FreshnessFilter {
248 max_age_ns: Some(Nanoseconds::from_secs(max_age_s)),
249 max_clock_drift_ns: None,
250 },
251 1,
252 );
253 let now = Nanoseconds::from_secs(u64::try_from(now_s).unwrap());
254 let prices = vec![
255 Some(price(1_000_000, 0, u64::try_from(publish_time_s).unwrap())),
256 Some(price(9_999_999, 0, u64::try_from(now_s).unwrap())),
257 ];
258
259 let result = proxy
260 .resolve(
261 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
262 prices,
263 now,
264 )
265 .unwrap();
266
267 assert_eq!(
268 result.value.unwrap().price,
269 if included { 1_000_000 } else { 9_999_999 }
270 );
271 }
272
273 #[rstest]
274 #[case::exactly_at_limit_included(1500, 1000, 500, true)]
275 #[case::one_over_excluded(1501, 1000, 500, false)]
276 fn resolve_max_clock_drift_boundary(
277 #[case] publish_time_s: i64,
278 #[case] now_s: i64,
279 #[case] max_clock_drift_s: u64,
280 #[case] included: bool,
281 ) {
282 let proxy = median_proxy(
283 FreshnessFilter::new(None, Some(Nanoseconds::from_secs(max_clock_drift_s))),
284 1,
285 );
286 let now = Nanoseconds::from_secs(u64::try_from(now_s).unwrap());
287 let prices = vec![
288 Some(price(1_000_000, 0, u64::try_from(publish_time_s).unwrap())),
289 Some(price(9_999_999, 0, u64::try_from(now_s).unwrap())),
290 ];
291
292 let result = proxy
293 .resolve(
294 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
295 prices,
296 now,
297 )
298 .unwrap();
299
300 assert_eq!(
301 result.value.unwrap().price,
302 if included { 1_000_000 } else { 9_999_999 }
303 );
304 }
305
306 #[rstest]
307 #[case(
308 FreshnessFilter::new(Some(Nanoseconds::from_secs(500)), None),
309 vec![
310 Some(price(1_000_000, 0, 100)),
311 Some(price(2_000_000, 0, 1_000)),
312 ]
313 )]
314 #[case(
315 FreshnessFilter::new(None, Some(Nanoseconds::from_secs(500))),
316 vec![
317 Some(price(1_000_000, 0, 1_501)),
318 Some(price(2_000_000, 0, 1_000)),
319 ]
320 )]
321 fn resolve_priority_skips_filtered_first_source(
322 #[case] freshness_filter: FreshnessFilter,
323 #[case] prices: Vec<Option<Price>>,
324 ) {
325 let proxy = priority_proxy(freshness_filter);
326 let result = proxy
327 .resolve(
328 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
329 prices,
330 Nanoseconds::from_secs(1_000),
331 )
332 .unwrap();
333
334 assert_eq!(result.value.unwrap().price, 2_000_000);
335 }
336
337 #[test]
338 fn resolve_priority_skips_invalid_first_source() {
339 let proxy = priority_proxy(FreshnessFilter::new(None, None));
340 let result = proxy
341 .resolve(
342 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
343 vec![
344 Some(price(1_000_000, 1_000_000, 1_000)),
345 Some(price(2_000_000, 0, 1_000)),
346 ],
347 Nanoseconds::from_secs(1_000),
348 )
349 .unwrap();
350
351 assert_eq!(result.value.unwrap().price, 2_000_000);
352 }
353
354 #[test]
355 fn resolve_excludes_zero_publish_time_when_stale() {
356 let proxy = median_proxy(
357 FreshnessFilter {
358 max_age_ns: Some(Nanoseconds::from_ms(500)),
359 max_clock_drift_ns: None,
360 },
361 1,
362 );
363 let prices = vec![
364 Some(Price {
365 price: 1_000_000,
366 conf: 0,
367 expo: -6,
368 publish_time_ns: Nanoseconds::zero(),
369 }),
370 Some(Price {
371 price: 9_999_999,
372 conf: 0,
373 expo: -6,
374 publish_time_ns: Nanoseconds::from_ms(1_000),
375 }),
376 ];
377
378 let result = proxy
379 .resolve(
380 &mut CircuitBreakerSet::<CircuitBreaker>::empty(),
381 prices,
382 Nanoseconds::from_ms(1_000),
383 )
384 .unwrap();
385
386 assert_eq!(result.value.unwrap().price, 9_999_999);
387 }
388
389 #[test]
390 fn resolve_applies_tripped_circuit_breaker_while_persisting_history() {
391 let proxy = priority_proxy(FreshnessFilter::new(None, None));
392 let mut circuit_breakers = CircuitBreakerSet::new(CircuitBreakerSetConfig {
393 sample_interval_ns: Nanoseconds::zero(),
394 history_len: 2,
395 });
396 let breaker_id = 0;
397 circuit_breakers
398 .add(
399 breaker_id,
400 CircuitBreaker::StepwiseChange(StepwiseChange {
401 max_relative_change: Decimal::from_u8(1) / 10_u8,
402 }),
403 )
404 .unwrap();
405 let now = Nanoseconds::from_secs(1_000);
406
407 proxy
408 .resolve(
409 &mut circuit_breakers,
410 [Some(price(100, 0, 1_000)), None],
411 now,
412 )
413 .unwrap();
414 let resolution = proxy
415 .resolve(
416 &mut circuit_breakers,
417 [Some(price(120, 0, 1_001)), None],
418 now,
419 )
420 .unwrap();
421 assert_eq!(
422 resolution.value,
423 Err(PriceBlockedReason::BreakerTripped {
424 blocking_breaker_ids: vec![breaker_id]
425 })
426 );
427 assert_eq!(resolution.events.len(), 1);
428
429 let resolution = proxy
430 .resolve(
431 &mut circuit_breakers,
432 [Some(price(130, 0, 1_002)), None],
433 now,
434 )
435 .unwrap();
436 assert_eq!(
437 resolution.value,
438 Err(PriceBlockedReason::BreakerTripped {
439 blocking_breaker_ids: vec![breaker_id]
440 })
441 );
442 assert!(resolution.events.is_empty());
443
444 assert_eq!(circuit_breakers.accepted_history().len(), 1);
445 assert_eq!(
446 circuit_breakers
447 .accepted_history()
448 .get(0)
449 .unwrap()
450 .price
451 .price,
452 100
453 );
454 assert_eq!(circuit_breakers.observed_history().len(), 2);
455 assert_eq!(
456 circuit_breakers
457 .observed_history()
458 .get(0)
459 .unwrap()
460 .price
461 .price,
462 120
463 );
464 assert_eq!(
465 circuit_breakers
466 .observed_history()
467 .get(1)
468 .unwrap()
469 .price
470 .price,
471 130
472 );
473 }
474
475 #[test]
476 fn resolve_length_mismatch_preserves_populated_breaker_state() {
477 let proxy = priority_proxy(FreshnessFilter::empty());
478 let mut circuit_breakers = CircuitBreakerSet::new(CircuitBreakerSetConfig {
479 sample_interval_ns: Nanoseconds::zero(),
480 history_len: 2,
481 });
482 circuit_breakers
483 .add(
484 0,
485 CircuitBreaker::StepwiseChange(StepwiseChange {
486 max_relative_change: Decimal::from_u8(1) / 10_u8,
487 }),
488 )
489 .unwrap();
490 proxy
491 .resolve(
492 &mut circuit_breakers,
493 [Some(price(100, 0, 1_000)), None],
494 Nanoseconds::from_secs(1_000),
495 )
496 .unwrap();
497 let before = circuit_breakers.clone();
498
499 assert_eq!(
500 proxy.resolve(
501 &mut circuit_breakers,
502 [Some(price(120, 0, 1_001))],
503 Nanoseconds::from_secs(1_001),
504 ),
505 Err(ResolveError::Aggregation(Error::LengthMismatch {
506 expected: 2,
507 actual: 1,
508 }))
509 );
510 assert_eq!(circuit_breakers, before);
511 }
512}