vecna
/
lox
mirror of https://gitlab.torproject.org/vecna/lox.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add endpoint for advancing time during testing

This commit is contained in:
Vecna 2024-04-26 22:34:54 -04:00
parent ec5217222e
commit 0644431dee
4 changed files with 76 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
crates/lox-distributor/config.json
View File

@ -4,6 +4,7 @@
}, },
"lox_authority_port": 8001, "lox_authority_port": 8001,
"test_port": 8005,
"troll_patrol_port": 8002, "troll_patrol_port": 8002,
"metrics_port": 5222, "metrics_port": 5222,
"bridge_config": { "bridge_config": {

45
crates/lox-distributor/src/main.rs
View File

@ -3,7 +3,7 @@
use clap::Parser; use clap::Parser;
use curve25519_dalek::ristretto::RistrettoBasepointTable; use curve25519_dalek::ristretto::RistrettoBasepointTable;
use futures::future; use futures::{future, join};
use hyper::{ use hyper::{
server::conn::AddrStream, server::conn::AddrStream,
service::{make_service_fn, service_fn}, service::{make_service_fn, service_fn},
@ -33,6 +33,8 @@ mod request_handler;
use request_handler::handle; use request_handler::handle;
mod resource_parser; mod resource_parser;
use resource_parser::{parse_into_bridgelines, parse_into_buckets}; use resource_parser::{parse_into_bridgelines, parse_into_buckets};
mod test_handler;
use test_handler::handle as test_handle;
mod troll_patrol_handler; mod troll_patrol_handler;
use troll_patrol_handler::handle as tp_handle; use troll_patrol_handler::handle as tp_handle;
@ -71,6 +73,7 @@ struct Config {
db: DbConfig, db: DbConfig,
metrics_port: u16, metrics_port: u16,
lox_authority_port: u16, lox_authority_port: u16,
test_port: u16,
troll_patrol_port: u16, troll_patrol_port: u16,
bridge_config: BridgeConfig, bridge_config: BridgeConfig,
rtype: ResourceInfo, rtype: ResourceInfo,
@ -263,6 +266,14 @@ async fn context_manager(
lox_db.write_context(context.clone()); lox_db.write_context(context.clone());
sleep(Duration::from_millis(1)).await; sleep(Duration::from_millis(1)).await;
} }
TestRequest { req, sender } => {
let response = test_handle(context.clone(), req).await;
if let Err(e) = sender.send(response) {
eprintln!("Server Response Error: {:?}", e);
};
lox_db.write_context(context.clone());
sleep(Duration::from_millis(1)).await;
}
TpRequest { req, sender } => { TpRequest { req, sender } => {
let response = tp_handle(context.clone(), &mut Htables, req).await; let response = tp_handle(context.clone(), &mut Htables, req).await;
if let Err(e) = sender.send(response) { if let Err(e) = sender.send(response) {
@ -291,6 +302,10 @@ enum Command {
req: Request<Body>, req: Request<Body>,
sender: oneshot::Sender<Result<Response<Body>, Infallible>>, sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
}, },
TestRequest {
req: Request<Body>,
sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
},
TpRequest { TpRequest {
req: Request<Body>, req: Request<Body>,
sender: oneshot::Sender<Result<Response<Body>, Infallible>>, sender: oneshot::Sender<Result<Response<Body>, Infallible>>,
@ -311,6 +326,7 @@ async fn main() {
let (rdsys_tx, context_rx) = mpsc::channel(32); let (rdsys_tx, context_rx) = mpsc::channel(32);
let request_tx = rdsys_tx.clone(); let request_tx = rdsys_tx.clone();
let test_request_tx = rdsys_tx.clone();
let tp_request_tx = rdsys_tx.clone(); let tp_request_tx = rdsys_tx.clone();
let shutdown_cmd_tx = rdsys_tx.clone(); let shutdown_cmd_tx = rdsys_tx.clone();
@ -385,6 +401,23 @@ async fn main() {
async move { Ok::<_, Infallible>(service) } async move { Ok::<_, Infallible>(service) }
}); });
let test_make_service = make_service_fn(move |_conn: &AddrStream| {
let request_tx = test_request_tx.clone();
let service = service_fn(move |req| {
let request_tx = request_tx.clone();
let (response_tx, response_rx) = oneshot::channel();
let cmd = Command::Request {
req,
sender: response_tx,
};
async move {
request_tx.send(cmd).await.unwrap();
response_rx.await.unwrap()
}
});
async move { Ok::<_, Infallible>(service) }
});
let tp_make_service = make_service_fn(move |_conn: &AddrStream| { let tp_make_service = make_service_fn(move |_conn: &AddrStream| {
let request_tx = tp_request_tx.clone(); let request_tx = tp_request_tx.clone();
let service = service_fn(move |req| { let service = service_fn(move |req| {
@ -406,19 +439,27 @@ async fn main() {
let pub_addr = SocketAddr::from(([127, 0, 0, 1], config.lox_authority_port)); let pub_addr = SocketAddr::from(([127, 0, 0, 1], config.lox_authority_port));
let server = Server::bind(&pub_addr).serve(make_service); let server = Server::bind(&pub_addr).serve(make_service);
let graceful = server.with_graceful_shutdown(shutdown_signal()); let graceful = server.with_graceful_shutdown(shutdown_signal());
// Address for test commands
let test_addr = SocketAddr::from(([127, 0, 0, 1], config.test_port));
let test_server = Server::bind(&test_addr).serve(test_make_service);
let test_graceful = test_server.with_graceful_shutdown(shutdown_signal());
// Address for connections from Troll Patrol // Address for connections from Troll Patrol
let tp_addr = SocketAddr::from(([127, 0, 0, 1], config.troll_patrol_port)); let tp_addr = SocketAddr::from(([127, 0, 0, 1], config.troll_patrol_port));
let tp_server = Server::bind(&tp_addr).serve(tp_make_service); let tp_server = Server::bind(&tp_addr).serve(tp_make_service);
let tp_graceful = tp_server.with_graceful_shutdown(shutdown_signal()); let tp_graceful = tp_server.with_graceful_shutdown(shutdown_signal());
println!("Listening on {}", pub_addr); println!("Listening on {}", pub_addr);
println!("Listening on {}", test_addr);
println!("Listening on {}", tp_addr); println!("Listening on {}", tp_addr);
let (a, b) = future::join(graceful, tp_graceful).await; let (a, b, c) = join!(graceful, test_graceful, tp_graceful);
if a.is_err() { if a.is_err() {
eprintln!("server error: {}", a.unwrap_err()); eprintln!("server error: {}", a.unwrap_err());
} }
if b.is_err() { if b.is_err() {
eprintln!("server error: {}", b.unwrap_err()); eprintln!("server error: {}", b.unwrap_err());
} }
if c.is_err() {
eprintln!("server error: {}", c.unwrap_err());
}
future::join_all([ future::join_all([
metrics_handler, metrics_handler,
rdsys_request_handler, rdsys_request_handler,

5
crates/lox-distributor/src/request_handler.rs
View File

@ -56,11 +56,6 @@ pub async fn handle(
let bytes = body::to_bytes(req.into_body()).await.unwrap(); let bytes = body::to_bytes(req.into_body()).await.unwrap();
cloned_context.verify_and_send_blockage_migration(bytes) cloned_context.verify_and_send_blockage_migration(bytes)
}), }),
//#[cfg(test)]
(&Method::POST, "/advancedays") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap();
cloned_context.advance_days_with_response_test(bytes)
}),
_ => { _ => {
// Return 404 not found response. // Return 404 not found response.
Ok(Response::builder() Ok(Response::builder()

32
crates/lox-distributor/src/test_handler.rs Normal file
View File

@ -0,0 +1,32 @@
use crate::lox_context::LoxServerContext;
use hyper::{body, header::HeaderValue, Body, Method, Request, Response, StatusCode};
use std::convert::Infallible;
// Handle requests used in testing
pub async fn handle(
cloned_context: LoxServerContext,
req: Request<Body>,
) -> Result<Response<Body>, Infallible> {
match req.method() {
&Method::OPTIONS => Ok(Response::builder()
.header("Access-Control-Allow-Origin", HeaderValue::from_static("*"))
.header("Access-Control-Allow-Headers", "accept, content-type")
.header("Access-Control-Allow-Methods", "POST")
.status(200)
.body(Body::from("Allow POST"))
.unwrap()),
_ => match (req.method(), req.uri().path()) {
(&Method::POST, "/advancedays") => Ok::<_, Infallible>({
let bytes = body::to_bytes(req.into_body()).await.unwrap();
cloned_context.advance_days_with_response_test(bytes)
}),
_ => {
// Return 404 not found response.
Ok(Response::builder()
.status(StatusCode::NOT_FOUND)
.body(Body::from("Not found"))
.unwrap())
}
},
}
}