FakeYou rendering *almost* works

This commit is contained in:
Brandon Thomas committed 2023-02-14 01:37:26 -05:00
1 parent 5be10ae87a
commit 6b937b7cd0
22 files changed
+515 -10

No files matched your search

Generated
+1
View File
@@ -756,6 +756,7 @@ dependencies = [
"egui",
"enums",
"errors",
"fakeyou_client",
"files",
"glium",
"hashing",
+6 -2
View File
@@ -1,6 +1,10 @@
// ===== MySql =====
pub mod media_uploads;
pub mod model_categories;
pub mod trending_model_analytics;
pub mod user_ratings;
pub mod web_rendition_targets; // NB: SqLite
pub mod web_scraping_targets; // NB: Sqlite
// ===== Sqlite =====
pub mod tts_render_targets;
pub mod web_rendition_targets;
pub mod web_scraping_targets;
@@ -0,0 +1 @@
pub mod tts_render_status;
@@ -0,0 +1,97 @@
use serde::Deserialize;
use serde::Serialize;
#[cfg(test)]
use strum::EnumCount;
#[cfg(test)]
use strum::EnumIter;
/// Used in the SqLite `web_scraping_targets` table in a `TEXT` field named `scraping_status`.
#[cfg_attr(test, derive(EnumIter, EnumCount))]
#[derive(Clone, Copy, Eq, PartialEq, Hash, Deserialize, Serialize)]
pub enum TtsRenderStatus {
#[serde(rename = "new")]
New,
#[serde(rename = "skipped")]
Skipped,
#[serde(rename = "failed")]
Failed,
#[serde(rename = "permanently_failed")]
PermanentlyFailed,
#[serde(rename = "success")]
Success,
}
// TODO(bt, 2023-01-17): This desperately needs MySQL integration tests!
impl_enum_display_and_debug_using_to_str!(TtsRenderStatus);
impl_sqlite_enum_coders!(TtsRenderStatus);
/// NB: Legacy API for older code.
impl TtsRenderStatus {
pub fn to_str(&self) -> &'static str {
match self {
Self::New => "new",
Self::Skipped => "skipped",
Self::Failed => "failed",
Self::PermanentlyFailed => "permanently_failed",
Self::Success => "success",
}
}
pub fn from_str(value: &str) -> Result<Self, String> {
match value {
"new" => Ok(Self::New),
"skipped" => Ok(Self::Skipped),
"failed" => Ok(Self::Failed),
"permanently_failed" => Ok(Self::PermanentlyFailed),
"success" => Ok(Self::Success),
_ => Err(format!("invalid value: {:?}", value)),
}
}
}
#[cfg(test)]
mod tests {
use crate::by_table::tts_render_targets::tts_render_status::TtsRenderStatus;
use crate::test_helpers::assert_serialization;
mod serde {
use super::*;
#[test]
fn test_serialization() {
assert_serialization(TtsRenderStatus::New, "new");
assert_serialization(TtsRenderStatus::Skipped, "skipped");
assert_serialization(TtsRenderStatus::Failed, "failed");
assert_serialization(TtsRenderStatus::PermanentlyFailed, "permanently_failed");
assert_serialization(TtsRenderStatus::Success, "success");
}
}
mod impl_methods {
use super::*;
#[test]
fn test_to_str() {
assert_eq!(TtsRenderStatus::New.to_str(), "new");
assert_eq!(TtsRenderStatus::Skipped.to_str(), "skipped");
assert_eq!(TtsRenderStatus::Failed.to_str(), "failed");
assert_eq!(TtsRenderStatus::PermanentlyFailed.to_str(), "permanently_failed");
assert_eq!(TtsRenderStatus::Success.to_str(), "success");
}
#[test]
fn test_from_str() {
assert_eq!(TtsRenderStatus::from_str("new").unwrap(), TtsRenderStatus::New);
assert_eq!(TtsRenderStatus::from_str("skipped").unwrap(), TtsRenderStatus::Skipped);
assert_eq!(TtsRenderStatus::from_str("failed").unwrap(), TtsRenderStatus::Failed);
assert_eq!(TtsRenderStatus::from_str("permanently_failed").unwrap(), TtsRenderStatus::PermanentlyFailed);
assert_eq!(TtsRenderStatus::from_str("success").unwrap(), TtsRenderStatus::Success);
assert!(TtsRenderStatus::from_str("foo").is_err());
}
}
}
+2
View File
@@ -19,6 +19,7 @@ pub(crate) enum EntityType {
ModelCategory,
NewsStory, // NB: aichatbot / sqlite
TtsModel,
TtsRenderTask, // NB: aichatbot / sqlite
User,
VoiceConversionModel,
W2lTemplate,
@@ -34,6 +35,7 @@ impl EntityType {
Self::ModelCategory => "CAT:", // NB: Old-style prefix, do not use for future tokens.
Self::NewsStory => "news_story_",
Self::TtsModel => "TM:", // NB: Old-style prefix, do not use for future tokens.
Self::TtsRenderTask => "tts_task_",
Self::User => "U:", // NB: Old-style prefix, do not use for future tokens.
Self::VoiceConversionModel => "voco_",
Self::W2lTemplate => "WT:", // NB: Old-style prefix, do not use for future tokens.
+4
View File
@@ -1,4 +1,8 @@
// ===== For MySQL =====
pub mod model_categories;
pub mod tts_models;
pub mod w2l_templates;
// ===== For Sqlite =====
pub mod news_stories;
pub mod tts_render_tasks;
@@ -3,7 +3,7 @@ use serde::Deserialize;
use serde::Serialize;
use std::fmt::Debug;
/// The primary key for "generic" inference jobs.
/// The primary key for news stories (Sqlite)
#[derive(Clone, PartialEq, Eq, sqlx::Type, Debug, Serialize, Deserialize)]
#[sqlx(transparent)]
pub struct NewsStoryToken(pub String);
@@ -0,0 +1,12 @@
use crate::prefixes::EntityType;
use serde::Deserialize;
use serde::Serialize;
use std::fmt::Debug;
/// The primary key for TTS render tasks (Sqlite / AiChatBotSidecar)
#[derive(Clone, PartialEq, Eq, sqlx::Type, Debug, Serialize, Deserialize)]
#[sqlx(transparent)]
pub struct TtsRenderTaskToken(pub String);
impl_string_token!(TtsRenderTaskToken);
impl_crockford_generator!(TtsRenderTaskToken, 32usize, EntityType::TtsRenderTask, CrockfordMixed);
@@ -22,18 +22,19 @@ impl FakeYouApiClient {
}
// TODO: need to yield better, more "library"-appropriate errors.
pub async fn post_inference(&self, request: CreateTtsInferenceRequest<'_>) -> AnyhowResult<CreateTtsInferenceResponse> {
pub async fn create_tts_inference(&self, request: CreateTtsInferenceRequest<'_>) -> AnyhowResult<CreateTtsInferenceResponse> {
let url = format!("https://{}/tts/inference", self.api_domain);
let response = self.client
.post(url)
.header(AUTHORIZATION_HEADER, &self.api_token)
.json(&request)
.send()
.await?
.bytes()
.text()
.await?;
let response = serde_json::from_str(response)?;
let response = serde_json::from_str(&response)?;
Ok(response)
}
@@ -45,11 +46,12 @@ impl FakeYouApiClient {
let response = self.client
.get(url)
.header(AUTHORIZATION_HEADER, &self.api_token)
.send()
.await?
.bytes()
.text()
.await?;
let response = serde_json::from_str(response)?;
let response = serde_json::from_str(&response)?;
Ok(response)
}
@@ -5,6 +5,7 @@ use sqlx::SqlitePool;
use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus;
use tokens::tokens::news_stories::NewsStoryToken;
use tokens::tokens::tts_models::TtsModelToken;
use tokens::tokens::tts_render_tasks::TtsRenderTaskToken;
pub struct Args <'a> {
// TODO: This will be multiple types in the future
@@ -21,9 +22,12 @@ pub async fn insert_tts_render_target(args: Args<'_>) -> AnyhowResult<()> {
let news_story_token = args.news_story_token.to_string();
let tts_voice_identifier= args.tts_voice_identifier.to_string();
let tts_render_task_token = TtsRenderTaskToken::generate().to_string();
let query = sqlx::query!(
r#"
INSERT INTO tts_render_targets(
token,
story_type,
story_token,
tts_service,
@@ -31,6 +35,7 @@ INSERT INTO tts_render_targets(
full_text
)
VALUES (
?,
"news_story",
?,
"fakeyou",
@@ -38,6 +43,7 @@ VALUES (
?
)
"#,
tts_render_task_token,
news_story_token,
tts_voice_identifier,
args.full_text
@@ -0,0 +1,105 @@
use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus;
use errors::{anyhow, AnyhowResult};
use sqlx::SqlitePool;
use tokens::tokens::tts_models::TtsModelToken;
use tokens::tokens::tts_render_tasks::TtsRenderTaskToken;
pub struct TtsRenderTarget {
pub id: i64,
pub token: TtsRenderTaskToken,
// Composite foreign key
pub story_type: String,
pub story_token: String,
pub tts_service: String,
pub tts_voice_identifier: String,
pub full_text: String,
pub maybe_result_url: Option<String>,
pub maybe_result_relative_filesystem_location: Option<String>,
pub tts_render_status: TtsRenderStatus,
pub tts_render_attempts: i64,
}
pub async fn list_tts_render_targets(
tts_render_status: TtsRenderStatus,
last_id: i64,
limit: i64,
sqlite_pool: &SqlitePool,
) -> AnyhowResult<Vec<TtsRenderTarget>> {
// NB: Sqlx doesn't support `WHERE ... IN (...)` "yet". :(
// https://github.com/launchbadge/sqlx/blob/6d0d7402c8a9cbea2676a1795e9fb50b0cf60c03/FAQ.md?plain=1#L73
let tts_render_status = tts_render_status.to_str().to_string();
let query = sqlx::query_as!(
RawInternalTtsRenderTarget,
r#"
SELECT
id,
token as `token: tokens::tokens::tts_render_tasks::TtsRenderTaskToken`,
story_type,
story_token,
tts_service,
tts_voice_identifier,
full_text,
maybe_result_url,
maybe_result_relative_filesystem_location,
tts_render_status as `tts_render_status: enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus`,
tts_render_attempts
FROM tts_render_targets
WHERE
tts_render_status = ?
AND id > ?
ORDER BY id ASC
LIMIT ?
"#,
tts_render_status,
last_id,
limit,
);
let records = query.fetch_all(sqlite_pool)
.await?;
let records = records.into_iter()
.map(|record : RawInternalTtsRenderTarget| {
TtsRenderTarget {
id: record.id,
token: record.token,
story_type: record.story_type,
story_token: record.story_token,
tts_service: record.tts_service,
tts_voice_identifier: record.tts_voice_identifier,
full_text: record.full_text,
maybe_result_url: record.maybe_result_url,
maybe_result_relative_filesystem_location: record.maybe_result_relative_filesystem_location,
tts_render_status: record.tts_render_status,
tts_render_attempts: record.tts_render_attempts,
}
})
.collect::<Vec<TtsRenderTarget>>();
Ok(records)
}
struct RawInternalTtsRenderTarget {
pub id: i64,
pub token: TtsRenderTaskToken,
// Composite foreign key
pub story_type: String,
pub story_token: String,
pub tts_service: String,
pub tts_voice_identifier: String,
pub full_text: String,
pub maybe_result_url: Option<String>,
pub maybe_result_relative_filesystem_location: Option<String>,
pub tts_render_status: TtsRenderStatus,
pub tts_render_attempts: i64,
}
@@ -1 +1,4 @@
pub mod insert_tts_render_target;
pub mod insert_tts_render_target;
pub mod list_tts_render_targets;
pub mod update_tts_render_target_unsuccessfully_submitted;
pub mod update_tts_render_target_successfully_submitted;
@@ -0,0 +1,33 @@
use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
use errors::{anyhow, AnyhowResult};
use sqlx::SqlitePool;
use tokens::tokens::news_stories::NewsStoryToken;
use tokens::tokens::tts_render_tasks::TtsRenderTaskToken;
pub struct Args <'a> {
pub tts_render_task_token: &'a TtsRenderTaskToken,
pub sqlite_pool: &'a SqlitePool,
}
pub async fn update_tts_render_target_successfully_submitted(args: Args<'_>) -> AnyhowResult<()> {
let query = sqlx::query!(
r#"
UPDATE tts_render_targets
SET
tts_render_status = "processing",
tts_render_attempts = tts_render_attempts + 1,
version = version + 1
WHERE
token = ?
"#,
args.tts_render_task_token,
);
let query_result = query.execute(args.sqlite_pool)
.await;
match query_result {
Ok(_) => Ok(()),
Err(err) => Err(anyhow!("error updating: {:?}", err)),
}
}
@@ -0,0 +1,48 @@
use enums::common::sqlite::awaitable_job_status::AwaitableJobStatus;
use errors::{anyhow, AnyhowResult};
use sqlx::SqlitePool;
use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus;
use tokens::tokens::news_stories::NewsStoryToken;
use tokens::tokens::tts_render_tasks::TtsRenderTaskToken;
pub struct Args <'a> {
pub tts_render_task_token: &'a TtsRenderTaskToken,
pub tts_render_attempts: i64,
pub sqlite_pool: &'a SqlitePool,
}
pub async fn update_tts_render_target_unsuccessfully_submitted(args: Args<'_>) -> AnyhowResult<()> {
let mut tts_render_status = next_status(args.tts_render_attempts);
let query = sqlx::query!(
r#"
UPDATE tts_render_targets
SET
tts_render_status = ?,
tts_render_attempts = ?,
version = version + 1
WHERE
token = ?
"#,
tts_render_status,
args.tts_render_attempts,
args.tts_render_task_token,
);
let query_result = query.execute(args.sqlite_pool)
.await;
match query_result {
Ok(_) => Ok(()),
Err(err) => Err(anyhow!("error updating: {:?}", err)),
}
}
fn next_status(attempts: i64) -> TtsRenderStatus {
if attempts >= 2 {
TtsRenderStatus::PermanentlyFailed
} else {
TtsRenderStatus::Failed
}
}
@@ -19,6 +19,7 @@ actix_helpers = { path = "../../lib/actix_helpers" }
concurrency = { path = "../../lib/concurrency" }
enums = { path = "../../_types/enums" }
errors = { path = "../../lib/errors" }
fakeyou_client = { path = "../../lib/fakeyou_client" }
files = { path = "../../lib/files" }
hashing = { path = "../../lib/hashing" }
http_server_common = { path = "../../lib/http_server_common" }
@@ -7,6 +7,10 @@
CREATE TABLE tts_render_targets (
id INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
-- Effective primary key.
-- Follows a single TTS audio through the phases of rendering
token TEXT NOT NULL UNIQUE,
-- Composite primary key for what this TTS render job belongs to (eg. news story)
story_type TEXT NOT NULL,
story_token TEXT NOT NULL,
@@ -30,7 +34,7 @@ CREATE TABLE tts_render_targets (
-- For audio results that get downloaded, they live on the filesystem here
-- This is relative to the runtime data directory.
maybe_filesystem_relative_location TEXT,
maybe_result_relative_filesystem_location TEXT,
-- TTS render status: "new", "processing", "failed", "permanently_failed", "success", "skipped"
tts_render_status TEXT DEFAULT "new" NOT NULL,
@@ -35,9 +35,11 @@ use std::sync::Arc;
use std::thread;
use std::time::Duration;
use tokio::runtime::{Builder, Runtime};
use fakeyou_client::fakeyou_api_client::FakeYouApiClient;
use web_scrapers::sites::cnn::cnn_article_scraper::cnn_article_scraper;
use web_scrapers::sites::techcrunch::techcrunch_article_scraper::techcrunch_article_scraper;
use web_scrapers::sites::theguardian::theguardian_scraper::theguardian_scraper_test;
use crate::workers::audio::fakeyou_audio_create::main_loop::fakeyou_audio_create_main_loop;
//#[tokio::main]
//pub async fn main() -> AnyhowResult<()> {
@@ -87,15 +89,20 @@ pub async fn main() -> AnyhowResult<()> {
let openai_client = Arc::new(Client::new()
.with_api_key(startup_args.openai_secret_key.clone()));
let fakeyou_client = Arc::new(FakeYouApiClient::make_production_client(
&startup_args.fakeyou_api_token));
let save_directory = SaveDirectory::new(&startup_args.save_directory);
let database_url = easyenv::get_env_string_required("DATABASE_URL")?;
let pool = SqlitePoolOptions::new()
.max_connections(5)
.connect(&database_url).await?;
let job_state = Arc::new(JobState {
openai_client: openai_client.clone(),
fakeyou_client: fakeyou_client.clone(),
save_directory: save_directory.clone(),
sqlite_pool: pool,
app_control_state: app_control_state_inner.clone(),
@@ -124,6 +131,7 @@ pub async fn main() -> AnyhowResult<()> {
let job_state4 = job_state.clone();
let job_state5 = job_state.clone();
let job_state6 = job_state.clone();
let job_state7 = job_state.clone();
tokio_runtime.spawn(async {
let _r = web_index_ingestion_main_loop(job_state2).await;
@@ -145,6 +153,10 @@ pub async fn main() -> AnyhowResult<()> {
let _r = news_story_audio_preprocessing_main_loop(job_state6).await;
});
tokio_runtime.spawn(async {
let _r = fakeyou_audio_create_main_loop(job_state7).await;
});
// TODO: Final scheduling thread
//tokio_runtime.spawn(async {
// // TODO...
@@ -1,12 +1,15 @@
use async_openai::Client;
use crate::persistence::save_directory::SaveDirectory;
use crate::shared_state::app_control_state::AppControlState;
use fakeyou_client::fakeyou_api_client::FakeYouApiClient;
use sqlx::{Pool, Sqlite};
use std::sync::Arc;
#[derive(Clone)]
pub struct JobState {
pub openai_client: Arc<Client>,
pub fakeyou_client: Arc<FakeYouApiClient>,
pub save_directory: SaveDirectory,
pub sqlite_pool: Pool<Sqlite>,
@@ -12,6 +12,7 @@ pub struct StartupArgs {
pub save_directory: String,
pub openai_secret_key: String,
pub fakeyou_api_token: String,
}
/// Read in environment and CLI args and output the consolidated startup state
@@ -28,11 +29,14 @@ pub fn get_startup_args() -> AnyhowResult<StartupArgs> {
let openai_secret_key = easyenv::get_env_string_required("OPENAI_SECRET_KEY")?;
let fakeyou_api_token = easyenv::get_env_string_required("FAKEYOU_API_TOKEN")?;
let save_directory = get_save_directory(&matches)?;
Ok(StartupArgs {
save_directory,
openai_secret_key,
fakeyou_api_token,
})
}
@@ -0,0 +1,103 @@
use crate::shared_state::job_state::JobState;
use enums::by_table::tts_render_targets::tts_render_status::TtsRenderStatus;
use enums::by_table::web_scraping_targets::scraping_status::ScrapingStatus;
use errors::AnyhowResult;
use log::{debug, error, info};
use sqlite_queries::queries::by_table::tts_render_targets::list_tts_render_targets::list_tts_render_targets;
use sqlite_queries::queries::by_table::web_scraping_targets::insert_web_scraping_target::{Args, insert_web_scraping_target};
use sqlite_queries::queries::by_table::web_scraping_targets::list_web_scraping_targets::WebScrapingTarget as WebScrapingTargetRecord;
use sqlite_queries::queries::by_table::web_scraping_targets::list_web_scraping_targets::list_web_scraping_targets;
use sqlx::sqlite::SqlitePoolOptions;
use std::future::Future;
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use strum::IntoEnumIterator;
use web_scrapers::payloads::web_scraping_result::ScrapedWebArticle;
use web_scrapers::payloads::web_scraping_target::WebScrapingTarget;
use web_scrapers::sites::cnn::cnn_indexer::{cnn_indexer, CnnFeed};
use web_scrapers::sites::techcrunch::techcrunch_indexer::{techcrunch_indexer, TechcrunchFeed};
use crate::workers::audio::fakeyou_audio_create::process_single_record::process_single_record;
/// Get audio jobs and kick off FakeYou requests
pub async fn fakeyou_audio_create_main_loop(job_state: Arc<JobState>) {
loop {
debug!("fakeyou_audio_create main loop");
while job_state.app_control_state.is_fakeyou_paused() {
thread::sleep(Duration::from_secs(5));
}
single_job_loop_iteration(&job_state).await;
info!("fakeyou_audio_create loop finished; waiting...");
thread::sleep(Duration::from_secs(60));
}
}
// NB: No failures at this level, because we don't want to prevent progress on a stuck feed.
async fn single_job_loop_iteration(job_state: &Arc<JobState>) {
let statuses = vec![TtsRenderStatus::New, TtsRenderStatus::Failed];
for status in statuses {
scrape_jobs_of_status(status, job_state).await;
}
}
async fn scrape_jobs_of_status(status: TtsRenderStatus, job_state: &Arc<JobState>) {
const BATCH_SIZE : i64 = 10;
let mut last_id = 0;
let mut failure_count = 0;
loop {
// NB: Protect sqlite from contention.
thread::sleep(Duration::from_millis(500));
debug!("fakeyou_audio_create querying {:?} targets from id > {} ...", &status, last_id);
let query_result = list_tts_render_targets(
status, last_id, BATCH_SIZE, &job_state.sqlite_pool).await;
let targets = match query_result {
Ok(targets) => targets,
Err(err) => {
error!("failure querying batch: {:?}", err);
failure_count += 1;
// NB: Don't starve progress.
if failure_count > 2 {
failure_count = 0;
last_id += 1;
} else if failure_count > 3 {
return;
}
continue;
}
};
failure_count = 0;
if targets.is_empty() {
return; // Done with batches.
}
for target in targets {
debug!("Render audio: {}/{}", target.story_type, target.story_token);
while job_state.app_control_state.is_fakeyou_paused() {
thread::sleep(Duration::from_secs(5));
}
let result = process_single_record(&target, job_state).await;
if let Err(err) = result {
error!("Error processing target: {:?}", err);
}
last_id = target.id;
thread::sleep(Duration::from_secs(3));
}
}
}
@@ -0,0 +1,2 @@
pub mod main_loop;
pub mod process_single_record;
@@ -0,0 +1,58 @@
use crate::shared_state::job_state::JobState;
use errors::AnyhowResult;
use fakeyou_client::api::tts_inference::CreateTtsInferenceRequest;
use log::{error, info};
use sqlite_queries::queries::by_table::tts_render_targets::list_tts_render_targets::TtsRenderTarget;
use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_successfully_submitted::Args as SuccessArgs;
use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_successfully_submitted::update_tts_render_target_successfully_submitted;
use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_unsuccessfully_submitted::Args as UnsuccessfulArgs;
use sqlite_queries::queries::by_table::tts_render_targets::update_tts_render_target_unsuccessfully_submitted::update_tts_render_target_unsuccessfully_submitted;
use std::sync::Arc;
use tokens::tokens::tts_models::TtsModelToken;
pub async fn process_single_record(target: &TtsRenderTarget, job_state: &Arc<JobState>) -> AnyhowResult<()> {
info!("Posting FakeYou create TTS request...");
let tts_model_token = TtsModelToken::new_from_str(&target.tts_voice_identifier);
let tts_render_attempts = target.tts_render_attempts + 1;
let result = job_state
.fakeyou_client
.create_tts_inference(CreateTtsInferenceRequest {
uuid_idempotency_token: "",
tts_model_token: &tts_model_token,
inference_text: &target.full_text,
}).await;
match result {
Err(err) => {
error!("problem submitting to FakeYou: {:}", err);
update_tts_render_target_unsuccessfully_submitted(UnsuccessfulArgs {
tts_render_task_token: &target.token,
tts_render_attempts,
sqlite_pool: &job_state.sqlite_pool,
}).await?;
},
Ok(res) => {
if !res.success {
error!("unknown problem in submitting to FakeYou");
update_tts_render_target_unsuccessfully_submitted(UnsuccessfulArgs {
tts_render_task_token: &target.token,
tts_render_attempts,
sqlite_pool: &job_state.sqlite_pool,
}).await?;
} else {
update_tts_render_target_successfully_submitted(SuccessArgs {
tts_render_task_token: &target.token,
sqlite_pool: &job_state.sqlite_pool,
}).await?;
}
},
}
Ok(())
}