Skip to content

fix(collector): bound explicit historical discovery #172

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
proerror77 merged 1 commit into from
Jul 20, 2026
Merged

fix(collector): bound explicit historical discovery #172

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 186 additions & 0 deletions rust_hft/tools/collector/src/polymarket_raw.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,6 +235,13 @@ impl ReferenceConfig {
}) {
bail!("market IDs must be non-empty, whitespace-free identifiers");
}
if self.market_ids.len() > self.max_markets {
bail!(
"requested market IDs={} exceed max_markets={}",
self.market_ids.len(),
self.max_markets
);
}
if self.max_concurrent_trade_polls > DEFAULT_MAX_CONCURRENT_TRADE_POLLS {
bail!(
"concurrent trade polls cannot exceed the hard limit of {DEFAULT_MAX_CONCURRENT_TRADE_POLLS}"
Expand Down Expand Up @@ -2065,6 +2072,26 @@ impl ReferenceCollector {
}

async fn discover_markets(&self, now: DateTime<Utc>) -> Result<Vec<Value>> {
if !self.config.market_ids.is_empty() {
let mut markets = Vec::with_capacity(self.config.market_ids.len());
for market_id in &self.config.market_ids {
let url = format!(
"{}/{}",
self.endpoints.gamma_market,
urlencoding::encode(market_id)
);
let market = self
.get_json(&url, &[])
.await
.with_context(|| format!("Gamma market detail request failed for {market_id}"))?;
if market.get("id").and_then(Value::as_str) != Some(market_id.as_str()) {
bail!("Gamma market detail did not return requested market ID {market_id}");
}
markets.push(market);
}
return Ok(markets);
}

let mut discovery = GammaDiscovery::default();
for lane in [GammaLane::Open, GammaLane::Closed] {
let base = gamma_discovery_params(&self.config, now, lane.is_closed());
Expand Down Expand Up @@ -3589,6 +3616,14 @@ mod tests {
let error = malformed_market_id.validate().unwrap_err();
assert!(error.to_string().contains("market IDs"));

let excessive_requested_market_ids = ReferenceConfig {
market_ids: BTreeSet::from(["market-1".to_owned(), "market-2".to_owned()]),
max_markets: 1,
..default.clone()
};
let error = excessive_requested_market_ids.validate().unwrap_err();
assert!(error.to_string().contains("requested market IDs"));

let exact_capacity = ReferenceConfig {
max_markets: 2_688,
..default.clone()
Expand Down Expand Up @@ -3920,6 +3955,157 @@ mod tests {
assert_eq!(closed["end_date_min"], "2026-07-14T02:00:00.000000Z");
}

#[tokio::test]
async fn requested_market_ids_bypass_broad_gamma_discovery() {
use std::io::Read as _;

let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(3);
let mut requested = Vec::new();
while requested.len() < 2 && Instant::now() < deadline {
let (mut connection, _) = match listener.accept() {
Ok(accepted) => accepted,
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(5));
continue;
}
Err(error) => panic!("test server accept failed: {error}"),
};
connection.set_nonblocking(false).unwrap();
connection
.set_read_timeout(Some(Duration::from_secs(1)))
.unwrap();
let mut request = Vec::new();
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let mut chunk = [0_u8; 1_024];
let bytes = connection.read(&mut chunk).unwrap();
assert!(bytes > 0, "HTTP request ended before its headers");
request.extend_from_slice(&chunk[..bytes]);
}
let request = std::str::from_utf8(&request).unwrap();
let request_line = request.lines().next().unwrap_or_default();
let market_id = request_line
.strip_prefix("GET /markets/")
.and_then(|path| path.split_once(" HTTP/"))
.map(|(market_id, _)| market_id)
.expect("explicit market IDs must not request the Gamma keyset");
requested.push(market_id.to_owned());
let body = serde_json::to_vec(&json!({
"id": market_id,
"conditionId": format!("condition-{market_id}"),
"question": "Bitcoin Up or Down - 5 minutes",
"startDate": "2026-07-20T08:00:00Z",
"endDate": "2026-07-20T08:05:00Z",
"clobTokenIds": ["up-token", "down-token"],
}))
.unwrap();
write!(
connection,
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.unwrap();
connection.write_all(&body).unwrap();
connection.flush().unwrap();
}
requested
});

let temp = TestDir::new();
let config = ReferenceConfig {
spool_dir: temp.path().to_owned(),
symbols: vec!["BTCUSDT".to_owned()],
market_ids: BTreeSet::from(["bounded-1".to_owned(), "bounded-2".to_owned()]),
..ReferenceConfig::default()
};
let mut collector = ReferenceCollector::new(config).unwrap();
let base_url = format!("http://{address}");
collector.endpoints = ReferenceEndpoints {
gamma_markets: format!("{base_url}/markets/keyset"),
gamma_market: format!("{base_url}/markets"),
data_trades: format!("{base_url}/trades"),
};

let markets = collector
.discover_markets(fixed_time("2026-07-20T09:00:00Z"))
.await
.unwrap();

assert_eq!(
markets
.iter()
.filter_map(|market| market.get("id").and_then(Value::as_str))
.map(str::to_owned)
.collect::<BTreeSet<_>>(),
BTreeSet::from(["bounded-1".to_owned(), "bounded-2".to_owned()])
);
assert_eq!(server.join().unwrap(), ["bounded-1", "bounded-2"]);
}

#[tokio::test]
async fn requested_non_target_market_fails_before_completion() {
use std::io::Read as _;

let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let server = std::thread::spawn(move || {
let (mut connection, _) = listener.accept().unwrap();
let mut request = Vec::new();
while !request.windows(4).any(|window| window == b"\r\n\r\n") {
let mut chunk = [0_u8; 1_024];
let bytes = connection.read(&mut chunk).unwrap();
assert!(bytes > 0, "HTTP request ended before its headers");
request.extend_from_slice(&chunk[..bytes]);
}
assert!(std::str::from_utf8(&request)
.unwrap()
.starts_with("GET /markets/unrelated-market HTTP/"));
let body = serde_json::to_vec(&json!({
"id": "unrelated-market",
"conditionId": "unrelated-condition",
"question": "Will it rain tomorrow?",
"startDate": "2026-07-20T08:00:00Z",
"endDate": "2026-07-20T08:05:00Z",
"clobTokenIds": ["up-token", "down-token"],
}))
.unwrap();
write!(
connection,
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
body.len()
)
.unwrap();
connection.write_all(&body).unwrap();
connection.flush().unwrap();
});

let temp = TestDir::new();
let config = ReferenceConfig {
spool_dir: temp.path().to_owned(),
symbols: vec!["BTCUSDT".to_owned()],
market_ids: BTreeSet::from(["unrelated-market".to_owned()]),
..ReferenceConfig::default()
};
let mut collector = ReferenceCollector::new(config).unwrap();
let base_url = format!("http://{address}");
collector.endpoints = ReferenceEndpoints {
gamma_markets: format!("{base_url}/markets/keyset"),
gamma_market: format!("{base_url}/markets"),
data_trades: format!("{base_url}/trades"),
};

let error = collector.collect_once().await.unwrap_err();

assert!(error.to_string().contains("unrelated-market"));
assert!(!fs::read_to_string(temp.path().join(ACTIVE_TAPE))
.unwrap()
.contains(TRADE_COMPLETION_KIND));
server.join().unwrap();
}

#[test]
fn requested_market_ids_fail_closed_when_discovery_is_incomplete() {
let requested = BTreeSet::from(["2959141".to_owned(), "2959146".to_owned()]);
Expand Down
Loading