fix delete job route, roll back terminate job (for now)

This commit is contained in:
Brandon Thomas committed 2024-04-26 11:00:45 -04:00
1 parent 502f46aabc
commit bc16d5b048
5 files changed
+36 -45

No files matched your search

@@ -7,6 +7,7 @@ use config::shared_constants::{DEFAULT_MYSQL_CONNECTION_STRING, DEFAULT_RUST_LOG
use errors::AnyhowResult;
use crate::cli_args::parse_cli_args;
use crate::seeding::model_weights::seed_weights;
use crate::seeding::user_roles::seed_user_roles;
use crate::seeding::users::seed_user_accounts;
@@ -53,7 +54,7 @@ pub async fn main() -> AnyhowResult<()> {
// seed_media_files(&pool, maybe_bucket_clients.as_ref()).await?;
//seed_zero_shot_tts(&pool, maybe_bucket_clients.as_ref()).await?;
// seed_voice_conversion(&pool).await?;
//seed_weights(&pool).await?;
seed_weights(&pool).await?;
// seed_media_seedtool(&pool).await?;
// seed_tts_tacotron2(&pool, maybe_bucket_clients.as_ref()).await?;
@@ -1128,9 +1128,9 @@ pub async fn seed_weights(mysql_pool: &Pool<MySql>) -> AnyhowResult<()> {
};
// original_seed_weights(mysql_pool,user_token).await?;
// seed_weights_for_user_token(mysql_pool, user_token).await?;
seed_weights_for_user_token(mysql_pool, user_token).await?;
// seed_weights_for_paging(mysql_pool,user_token).await?;
seed_weights_for_testing_inference(mysql_pool,user_token.clone()).await?;
//seed_weights_for_testing_inference(mysql_pool,user_token.clone()).await?;
// println!("TESTING DOWLOAD");
//test_seed_weights_files().await?;
// seed_workflows_for_testing_inference(mysql_pool,user_token.clone()).await?;
@@ -122,38 +122,11 @@ pub async fn list_session_jobs_from_connection(
{
let mut query_builder: QueryBuilder<MySql> = make_query_builder();
// NB: Inner query is meant to run fast via query planner. This is
// meant to save us from users that might automate 10k+ jobs within
// a 36-hour period.
// The inner query is limited to 100 records, which will spare the
// outer query. Furthermore, even though `created_at` lacks an index,
// sort by id will accomplish this quickly.
{
query_builder.push(r#"
WHERE jobs.id IN (
SELECT j.id
FROM generic_inference_jobs as j
WHERE j.maybe_creator_user_token =
"#);
query_builder.push(" WHERE jobs.maybe_creator_user_token = ");
query_builder.push_bind(args.user_token.to_string());
query_builder.push_bind(args.user_token.to_string());
query_builder.push(r#"
AND j.created_at > DATE_SUB(NOW(), INTERVAL 36 HOUR)
ORDER BY j.id DESC
LIMIT 100
)
"#);
}
// 'AS x\n AND jobs.is_dismissed_by_user = FALSE AND jobs.status NOT IN (?, ?)' at line 62" })
query_builder.push(" AND jobs.is_dismissed_by_user = FALSE ");
// query_builder.push(" WHERE jobs.maybe_creator_user_token = ");
// query_builder.push_bind(args.user_token.to_string());
// // NB: `created_at` doesn't have an index, but `maybe_creator_user_token` does.
// query_builder.push(" AND jobs.created_at > DATE_SUB(NOW(), INTERVAL 36 HOUR) ");
// NB: `created_at` doesn't have an index, but `maybe_creator_user_token` does.
query_builder.push(" AND jobs.created_at > DATE_SUB(NOW(), INTERVAL 36 HOUR) ");
if let Some(statuses) = args.maybe_include_job_statuses {
if !statuses.is_empty() {
@@ -179,8 +152,8 @@ pub async fn list_session_jobs_from_connection(
}
}
// query_builder.push(" ORDER BY jobs.id DESC ");
// query_builder.push(" LIMIT 100 ");
query_builder.push(" ORDER BY jobs.id DESC ");
query_builder.push(" LIMIT 100 ");
let query = query_builder.build_query_as::<RawGenericInferenceJobStatus>();
@@ -5,7 +5,7 @@ use actix_web::{HttpMessage, HttpRequest, HttpResponse, web};
use actix_web::error::ResponseError;
use actix_web::http::StatusCode;
use actix_web::web::Path;
use log::{error, warn};
use log::{error, info, warn};
use utoipa::ToSchema;
use http_server_common::request::get_request_ip::get_request_ip;
@@ -79,6 +79,8 @@ pub async fn terminate_inference_job_handler(
path: Path<TerminateInferenceJobPathInfo>,
server_state: web::Data<Arc<ServerState>>) -> Result<HttpResponse, TerminateInferenceJobError>
{
info!("terminate_inference_job_handler: {:?}", path.token);
let mut mysql_connection = server_state.mysql_pool.acquire()
.await
.map_err(|e| {
@@ -1,8 +1,10 @@
use actix_http::body::MessageBody;
use actix_service::ServiceFactory;
use actix_web::{App, Error};
use actix_web::{App, Error, HttpResponse, web};
use actix_web::dev::{ServiceRequest, ServiceResponse};
use actix_helpers::route_builder::RouteBuilder;
use crate::http_server::endpoints::inference_job::batch_get_inference_job_status_handler::batch_get_inference_job_status_handler;
use crate::http_server::endpoints::inference_job::get_inference_job_status_handler::get_inference_job_status_handler;
use crate::http_server::endpoints::inference_job::get_pending_inference_job_count_handler::get_pending_inference_job_count_handler;
@@ -20,13 +22,26 @@ pub fn add_job_routes<T, B> (app: App<T>) -> App<T>
InitError = (),
>,
{
// NB(bt): New routes
let mut app = RouteBuilder::from_app(app)
.add_get("/v1/jobs/batch", batch_get_inference_job_status_handler)
.add_get("/v1/jobs/session", list_session_jobs_handler)
.add_get("/v1/jobs/job/{token}", get_inference_job_status_handler)
.add_delete("/v1/jobs/job/{token}", terminate_inference_job_handler, true)
.into_app();
let mut app =
app.service(
web::scope("/v1/jobs")
.service(
web::resource("/job/{token}")
.route(web::get().to(get_inference_job_status_handler))
.route(web::delete().to(terminate_inference_job_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/jobs/batch")
.route(web::get().to(batch_get_inference_job_status_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
.service(
web::resource("/v1/jobs/session")
.route(web::get().to(list_session_jobs_handler))
.route(web::head().to(|| HttpResponse::Ok()))
)
);
// NB(bt): Old routes (these must be supported for AI streamers).
let mut app = RouteBuilder::from_app(app)