Something went wrong. Try again.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384//! The sync engine: a single-task scheduler that dispatches per-subject and global jobs under a//! concurrency cap, honors ESI error-limit/rate-limit backoff by pausing all dispatch, and drains the//! optimistic write outbox on a fixed cadence.
use std::{ any::Any, collections::{HashMap, HashSet}, future::Future, panic::AssertUnwindSafe, pin::Pin, sync::Arc, time::Duration,};
use chrono::{DateTime, Duration as ChronoDuration, Utc};use futures_util::FutureExt;use tokio::{sync::mpsc, task::JoinSet, time::Instant};
use super::{ command::Command, drain, event::Event, handle::Handle, job::{self, JobCtx, JobKey, JobKind}, outbox::Registry, outcome::Outcome, schedule::Schedule, subject::Subject, token,};use crate::{ clients::{Error, esi, eve_image, eve_sso}, config::FeatureFlags, store::{ Database, images, model::{Credential, OwnerType, SyncLedger}, repo::{character, infra, org, sync_ledger}, },};
const DONE_RETENTION: Duration = Duration::from_secs(60 * 60);
const DRAIN_INTERVAL: Duration = Duration::from_secs(5);
const EVENT_BUFFER: usize = 64;
const GLOBAL_SUBJECT: Subject = Subject::Character(0);
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
const MARKET_ORDER_LAUNCH_FLOOR: Duration = Duration::from_secs(60);
const MAX_CONCURRENT_JOBS: usize = 4;
const NOT_READY_RETRY: Duration = Duration::from_secs(3);
const PENDING_RETRY: Duration = Duration::from_secs(120);
const RESPAWN_BACKOFF_BASE: Duration = Duration::from_secs(1);
const RESPAWN_BACKOFF_CAP: Duration = Duration::from_secs(30);
const RESPAWN_BREAKER_THRESHOLD: u32 = 5;
const RESPAWN_BREAKER_WINDOW: Duration = Duration::from_secs(60);
type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
#[derive(Default)]struct Breaker { attempts: u32, consecutive: u32,}
impl Breaker { fn record_restart(&mut self, lived: Duration) -> Option<RestartPlan> { if lived >= RESPAWN_BREAKER_WINDOW { self.consecutive = 0; } self.consecutive += 1; if self.consecutive >= RESPAWN_BREAKER_THRESHOLD { return None; } self.attempts += 1; Some(RestartPlan { attempt: self.attempts, backoff: backoff_for(self.consecutive), }) }
fn reset(&mut self) { self.attempts = 0; self.consecutive = 0; }}
enum Disposition { Restart, Stop { deliberate: bool },}
struct Engine { db: Database, drain_at: Instant, esi: Arc<esi::Client>, events: mpsc::Sender<Event>, // A connection pool reserved exclusively for housekeeping (the post-job ledger upsert plus the // outbox drain/prune maintenance), kept separate from `db` so the worker pool can never starve // housekeeping of a connection. Workers hold up to MAX_CONCURRENT_JOBS of `db`'s connections; this // pool guarantees finish() always has a free connection and never queues behind `busy_timeout`. housekeeping: Database, image: Arc<eve_image::Client>, image_store: images::Store, outbox: Registry, paused_until: Instant, schedule: Schedule, sso: Arc<eve_sso::Client>,}
impl Engine { async fn deferred_gathers(&self, subject: Subject) -> HashSet<JobKind> { if self.parent_row_exists(subject).await { return HashSet::new(); } profile_kind(subject) .on_success_triggers() .iter() .copied() .filter(|kind| kind.applies_to(subject)) .collect() }
async fn discover(&mut self) { let now = Instant::now(); let owned_characters = self.discover_credentials(now).await; self.discover_unowned_characters(&owned_characters, now).await; }
async fn discover_credentials(&mut self, now: Instant) -> HashSet<i64> { let mut owned_characters = HashSet::new(); let Ok(credentials) = infra::all(&self.db).await else { return owned_characters; }; for credential in credentials { if credential.owner_type() == OwnerType::Character { owned_characters.insert(credential.owner_id()); } self.enroll_credential(&credential, now).await; } owned_characters }
async fn enroll_credential(&mut self, credential: &Credential, now: Instant) { let subject = subject_from(credential.owner_id(), credential.owner_type()); let granted: HashSet<&str> = credential .scopes() .as_deref() .unwrap_or_default() .split_whitespace() .collect(); let rows = self.ledger_rows(subject).await; let seeds = future_seeds(&rows, now); let deferred = self.deferred_gathers(subject).await; // A flagged entity's token is known-bad, so enroll only its public (no-scope) jobs and keep its // privileged jobs out of the schedule until the TokenAudit clears the flag. let needs_reauth = credential.needs_reauth(); let kinds = if needs_reauth { JobKind::public_for_subject(subject) } else { JobKind::granted_for_subject(subject, &granted) }; self .schedule .enroll_kinds_deferred(subject, kinds.clone(), now, &seeds, &deferred); self.emit_seeds(subject, &kinds, &rows, needs_reauth); }
async fn discover_unowned_characters(&mut self, owned_characters: &HashSet<i64>, now: Instant) { let Ok(characters) = character::all(&self.db).await else { return; }; for character in characters { if owned_characters.contains(&character.id()) { continue; } let subject = Subject::Character(character.id()); let rows = self.ledger_rows(subject).await; let seeds = future_seeds(&rows, now); let kinds = JobKind::public_for_subject(subject); self.schedule.enroll_kinds_seeded(subject, kinds.clone(), now, &seeds); self.emit_seeds(subject, &kinds, &rows, false); } }
async fn ledger_rows(&self, subject: Subject) -> Vec<SyncLedger> { let (subject_id, subject_type) = owner_of(subject); sync_ledger::for_subject(&self.db, subject_type, subject_id) .await .unwrap_or_default() }
fn emit_seeds(&self, subject: Subject, kinds: &[JobKind], rows: &[SyncLedger], needs_reauth: bool) { for &kind in kinds { if !kind.is_feature_enabled(self.schedule.features()) { continue; } let key = JobKey::new(kind, subject); if needs_reauth && kind.gating_scope(subject).is_some() { self.emit(Event::Seeded { key, outcome: Outcome::Blocked { reason: "needs re-authentication".to_string(), }, next_in_secs: None, }); continue; } let Some(row) = rows.iter().find(|row| *row.kind() == format!("{kind:?}")) else { continue; }; let Some(outcome) = seeded_outcome(row) else { continue; }; let next_in_secs = future_seed(row, kind, Instant::now()).map(|at| at.saturating_duration_since(Instant::now()).as_secs()); self.emit(Event::Seeded { key, outcome, next_in_secs, }); } }
fn emit(&self, event: Event) { let _ = self.events.try_send(event); }
async fn enroll_global(&mut self) { let now = Instant::now(); let rows = self.ledger_rows(GLOBAL_SUBJECT).await; let seeds = future_seeds(&rows, now); let kinds = global_kinds(); self .schedule .enroll_kinds_seeded(GLOBAL_SUBJECT, kinds.clone(), now, &seeds); self.emit_seeds(GLOBAL_SUBJECT, &kinds, &rows, false); }
async fn enroll_subject(&mut self, subject: Subject, now: Instant) { let (owner_id, owner_type) = owner_of(subject); match infra::get(&self.db, owner_id, owner_type).await { Ok(Some(credential)) => { let granted: HashSet<&str> = credential .scopes() .as_deref() .unwrap_or_default() .split_whitespace() .collect(); let kinds = if credential.needs_reauth() { JobKind::public_for_subject(subject) } else { JobKind::granted_for_subject(subject, &granted) }; let seeds = self.seeds_for(subject, now).await; let deferred = self.deferred_gathers(subject).await; self .schedule .enroll_kinds_deferred(subject, kinds, now, &seeds, &deferred); } _ => self.schedule.enroll(subject, now), } }
async fn finish(&mut self, key: JobKey, result: Result<Outcome, Error>) { let now = Instant::now(); let attempt_at = Utc::now().to_rfc3339(); let outcome = match result { Ok(outcome) => { if is_pending_outcome(&outcome) { self .schedule .reschedule_throttle(key, now, PENDING_RETRY.min(key.kind.interval())); } else { self.schedule.reschedule_success(key, now); } for triggered in key.kind.on_success_triggers() { if triggered.applies_to(key.subject) { self.schedule.make_due_now_for_subject(*triggered, key.subject, now); } else if triggered.is_global() { self.schedule.make_due_now(*triggered, now); } } tracing::debug!(?key, outcome = outcome.label(), "sync job finished"); self.emit(Event::Finished { key, outcome: outcome.clone(), }); outcome } Err(Error::NotReady) => { self.schedule.reschedule_throttle(key, now, NOT_READY_RETRY); tracing::info!( ?key, retry_secs = NOT_READY_RETRY.as_secs(), "parent record not yet persisted; short-retrying without an ESI call" ); Outcome::NotReady } Err(Error::ErrorLimited { reset_secs, }) => { let delay = Duration::from_secs(reset_secs.max(1)); self.paused_until = now + delay; self.schedule.reschedule_throttle(key, now, delay); tracing::warn!(?key, reset_secs, "ESI error-limited; pausing dispatch"); self.emit(Event::BackingOff { key, retry_secs: reset_secs, }); Outcome::Failed { reason: format!("ESI error-limited; resets in {reset_secs}s"), } } Err(Error::RateLimit { retry_after_secs, }) => { let delay = Duration::from_secs(retry_after_secs.max(1)); self.schedule.reschedule_throttle(key, now, delay); tracing::warn!(?key, retry_after_secs, "rate-limited; backing off"); self.emit(Event::BackingOff { key, retry_secs: retry_after_secs, }); Outcome::Failed { reason: format!("rate-limited; retry after {retry_after_secs}s"), } } Err(error) if is_permanent_failure(&error) => { self.schedule.reschedule_permanent(key, now); let reason = error.to_string(); tracing::warn!(?key, %error, "sync job failed permanently; parking until re-authentication"); self.emit(Event::Failed { key, reason: reason.clone(), }); Outcome::Failed { reason, } } Err(error) => { self.schedule.reschedule_failure(key, now); let reason = error.to_string(); if error.is_foreign_key_violation() { tracing::warn!( ?key, %error, "sync job failed: foreign-key violation; error names the character/corp whose org row was missing" ); } else { tracing::warn!(?key, %error, "sync job failed"); } self.emit(Event::Failed { key, reason: reason.clone(), }); Outcome::Failed { reason, } } }; let next_in = self.schedule.next_in(key, now); let next_eligible_at = next_in.map(|delay| { (Utc::now() + ChronoDuration::from_std(delay).unwrap_or_else(|_| ChronoDuration::zero())).to_rfc3339() }); self .record_ledger(key, &outcome, &attempt_at, next_eligible_at.as_deref()) .await; if let Some(next_in) = next_in { self.emit(Event::Scheduled { key, next_in_secs: next_in.as_secs(), }); } }
async fn handle_command(&mut self, command: Command) { let now = Instant::now(); match command { Command::Discover => self.discover().await, #[cfg(test)] Command::Drain => self.drain_at = now, Command::Enroll(subject) => self.enroll_subject(subject, now).await, Command::RunNow(subject) => self.schedule.run_now(subject, now), Command::SetFeatures(features) => self.schedule.reconcile_features(features), Command::Shutdown => {} // intercepted in the run loop before reaching here; arm exists for exhaustiveness Command::Withdraw(subject) => self.schedule.withdraw(subject), } }
async fn maybe_drain(&mut self, now: Instant) { if now < self.drain_at { return; } self.drain_at = now + DRAIN_INTERVAL; let events = self.events.clone(); let emit = move |event: Event| { let _ = events.try_send(event); }; match drain::drain(&self.housekeeping, &self.esi, &self.sso, &self.outbox, &emit).await { Ok(outcome) => { if let Some(reset_secs) = outcome.error_limit_reset_secs { let until = now + Duration::from_secs(reset_secs.max(1)); self.paused_until = self.paused_until.max(until); tracing::warn!(reset_secs, "outbox drain error-limited; pausing dispatch"); } } Err(error) => tracing::warn!(%error, "outbox drain pass failed"), } self.prune_done_rows().await; }
fn next_wake(&self, now: Instant) -> Instant { let dispatch_at = if now < self.paused_until { Some(self.paused_until) } else { let job_deadline = self.schedule.next_deadline(); let maintenance_at = self.drain_at; Some(job_deadline.map_or(maintenance_at, |job| job.min(maintenance_at))) }; dispatch_at.map_or(now + HEARTBEAT_INTERVAL, |at| at.min(now + HEARTBEAT_INTERVAL)) }
async fn parent_row_exists(&self, subject: Subject) -> bool { match subject { Subject::Character(id) => matches!(character::get(&self.db, id).await, Ok(Some(_))), Subject::Corporation(id) => { matches!(org::get_corporation(&self.db, id).await, Ok(Some(_))) } } }
async fn prune_done_rows(&self) { let before = (Utc::now() - ChronoDuration::from_std(DONE_RETENTION).unwrap_or(ChronoDuration::zero())).to_rfc3339(); match infra::prune_done(&self.housekeeping, &before).await { Ok(pruned) if pruned > 0 => { tracing::debug!(pruned, "pruned done outbox rows past retention") } Ok(_) => {} Err(error) => tracing::warn!(%error, "pruning done outbox rows failed"), } }
async fn record_ledger(&self, key: JobKey, outcome: &Outcome, attempt_at: &str, next_eligible_at: Option<&str>) { let (subject_id, subject_type) = owner_of(key.subject); let success_at = matches!(outcome, Outcome::Synced { .. } | Outcome::Empty).then_some(attempt_at); if let Err(error) = sync_ledger::upsert( &self.housekeeping, subject_type, subject_id, &format!("{:?}", key.kind), outcome.label(), outcome.rows_touched(), outcome.reason(), success_at, next_eligible_at, ) .await { tracing::warn!(?key, %error, "failed to record sync ledger row"); } }
async fn run(mut self, commands: &mut mpsc::UnboundedReceiver<Command>) -> ExitReason { self.enroll_global().await; self.discover().await; let mut in_flight: JoinSet<(JobKey, Result<Outcome, Error>)> = JoinSet::new(); let mut last_dispatched: Option<JobKey> = None; let reason = loop { let now = Instant::now(); if now >= self.paused_until { let free = MAX_CONCURRENT_JOBS.saturating_sub(in_flight.len()); for key in self.schedule.due(now).into_iter().take(free) { self.schedule.mark_in_flight(key); tracing::debug!(?key, "sync job started"); self.emit(Event::Started { key, }); let db = self.db.clone(); let esi = Arc::clone(&self.esi); let sso = Arc::clone(&self.sso); let image = Arc::clone(&self.image); let image_store = self.image_store.clone(); last_dispatched = Some(key); in_flight.spawn(async move { (key, run_job(&db, &esi, &sso, &image, &image_store, key).await) }); } self.maybe_drain(now).await; }
let deadline = self.next_wake(now); tokio::select! { finished = in_flight.join_next(), if !in_flight.is_empty() => match finished { Some(Ok((key, result))) => self.finish(key, result).await, Some(Err(join_error)) => tracing::error!(%join_error, "a sync job task panicked"), None => {} }, _ = tokio::time::sleep_until(deadline) => { self.enroll_global().await; self.discover().await; self.emit(Event::Heartbeat); } command = commands.recv() => match command { Some(Command::Shutdown) => { in_flight.shutdown().await; break ExitReason::Shutdown; } Some(command) => self.handle_command(command).await, None => break ExitReason::ChannelClosed, }, } }; let in_flight_count = in_flight.len(); if reason.is_deliberate() { tracing::debug!( target: "pod::lifecycle", reason = reason.label(), in_flight = in_flight_count, last_dispatched = ?last_dispatched, "sync engine run loop exited", ); } else { tracing::warn!( target: "pod::lifecycle", reason = reason.label(), in_flight = in_flight_count, last_dispatched = ?last_dispatched, "sync engine run loop exited unexpectedly", ); } reason }
async fn seeds_for(&self, subject: Subject, now: Instant) -> HashMap<JobKind, Instant> { future_seeds(&self.ledger_rows(subject).await, now) }}
fn future_seeds(rows: &[SyncLedger], now: Instant) -> HashMap<JobKind, Instant> { let mut seeds = HashMap::new(); for row in rows { let Some(kind) = JobKind::ALL .iter() .copied() .find(|kind| format!("{kind:?}") == *row.kind()) else { continue; }; if let Some(at) = future_seed(row, kind, now) { seeds.insert(kind, at); } } seeds}
fn seeded_outcome(row: &SyncLedger) -> Option<Outcome> { let reason = row.last_reason().clone().unwrap_or_default(); match row.outcome().as_str() { "synced" => Some(Outcome::Synced { rows_touched: row.rows_touched(), }), "empty" => Some(Outcome::Empty), "blocked" => Some(Outcome::Blocked { reason, }), "skipped" => Some(Outcome::Skipped { reason, }), "failed" => Some(Outcome::Failed { reason, }), "not_ready" => Some(Outcome::NotReady), _ => None, }}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]enum ExitReason { ChannelClosed, Shutdown,}
impl ExitReason { fn is_deliberate(self) -> bool { matches!(self, ExitReason::Shutdown) }
fn label(self) -> &'static str { match self { ExitReason::ChannelClosed => "channel-closed", ExitReason::Shutdown => "shutdown", } }}
struct RestartPlan { attempt: u32, backoff: Duration,}
struct Supervisor<O>where O: Fn() -> Registry + Send + Sync,{ command_rx: mpsc::UnboundedReceiver<Command>, db: Database, esi: Arc<esi::Client>, events: mpsc::Sender<Event>, features: FeatureFlags, housekeeping: Database, image: Arc<eve_image::Client>, image_store: images::Store, outbox: O, restart_rx: mpsc::UnboundedReceiver<()>, sso: Arc<eve_sso::Client>,}
impl<O> Supervisor<O>where O: Fn() -> Registry + Send + Sync,{ async fn supervise(self) { let Supervisor { command_rx, db, esi, events, features, housekeeping, image, image_store, outbox, restart_rx, sso, } = self; let build = || Engine { db: db.clone(), drain_at: Instant::now(), esi: Arc::clone(&esi), events: events.clone(), housekeeping: housekeeping.clone(), image: Arc::clone(&image), image_store: image_store.clone(), outbox: outbox(), paused_until: Instant::now(), schedule: Schedule::with_features(features), sso: Arc::clone(&sso), }; // Each attempt rebuilds a fresh Engine but feeds it the same command_rx and event_tx, so the // app's Handle and forward_sync_events stay wired to whichever engine is current. supervise_loop(&events, command_rx, restart_rx, |command_rx| { Box::pin(build().run(command_rx)) }) .await; }}
pub fn spawn( db: Database, housekeeping: Database, esi: Arc<esi::Client>, sso: Arc<eve_sso::Client>, image: Arc<eve_image::Client>, image_store: images::Store, features: FeatureFlags,) -> (Handle, mpsc::Receiver<Event>) { spawn_with_registry(db, housekeeping, esi, sso, image, image_store, features, || { super::mail_handlers::registry() .extend(super::calendar_handlers::registry()) .extend(super::contact_handlers::registry()) })}
#[allow(clippy::too_many_arguments)]fn spawn_with_registry<O>( db: Database, housekeeping: Database, esi: Arc<esi::Client>, sso: Arc<eve_sso::Client>, image: Arc<eve_image::Client>, image_store: images::Store, features: FeatureFlags, outbox: O,) -> (Handle, mpsc::Receiver<Event>)where O: Fn() -> Registry + Send + Sync + 'static,{ let (command_tx, command_rx) = mpsc::unbounded_channel(); let (event_tx, event_rx) = mpsc::channel(EVENT_BUFFER); let (restart_tx, restart_rx) = mpsc::unbounded_channel(); let supervisor = Supervisor { command_rx, db, esi, events: event_tx, features, housekeeping, image, image_store, outbox, restart_rx, sso, }; tokio::spawn(supervisor.supervise()); (Handle::new(command_tx, restart_tx), event_rx)}
fn backoff_for(consecutive: u32) -> Duration { let shift = consecutive.saturating_sub(1).min(u32::BITS - 1); RESPAWN_BACKOFF_BASE .saturating_mul(1u32.checked_shl(shift).unwrap_or(u32::MAX)) .min(RESPAWN_BACKOFF_CAP)}
fn disposition(outcome: Result<ExitReason, Box<dyn Any + Send>>) -> Disposition { match outcome { Ok(ExitReason::Shutdown) => Disposition::Stop { deliberate: true, }, Ok(ExitReason::ChannelClosed) => Disposition::Stop { deliberate: false, }, Err(panic) => { let detail = panic_message(&panic); tracing::error!(target: "pod::lifecycle", detail, "the sync engine loop panicked"); Disposition::Restart } }}
fn give_up_reason(consecutive: u32) -> String { format!("the sync engine died {consecutive} times in quick succession; auto-restart gave up")}
fn panic_message(panic: &(dyn Any + Send)) -> &str { panic .downcast_ref::<&'static str>() .copied() .or_else(|| panic.downcast_ref::<String>().map(String::as_str)) .unwrap_or("unknown panic")}
async fn supervise_loop<F>( events: &mpsc::Sender<Event>, mut command_rx: mpsc::UnboundedReceiver<Command>, mut restart_rx: mpsc::UnboundedReceiver<()>, mut run_engine: F,) where F: for<'a> FnMut(&'a mut mpsc::UnboundedReceiver<Command>) -> BoxFuture<'a, ExitReason>,{ let mut breaker = Breaker::default(); loop { let started = Instant::now(); // catch_unwind isolates a panic in the run loop into an Err here instead of unwinding the // supervisor task itself, which would silently end auto-restart and defeat the whole point. let outcome = AssertUnwindSafe(run_engine(&mut command_rx)).catch_unwind().await;
match disposition(outcome) { Disposition::Stop { deliberate, } => { if deliberate { tracing::info!(target: "pod::lifecycle", "the sync engine task stopped"); } else { tracing::info!( target: "pod::lifecycle", "the sync engine command channel closed; no clients remain, so the supervisor stops" ); } return; } Disposition::Restart => match breaker.record_restart(started.elapsed()) { None => { let consecutive = breaker.consecutive; tracing::error!( target: "pod::lifecycle", consecutive, "the sync engine died repeatedly in quick succession; the circuit breaker tripped, giving up auto-restart" ); let _ = events.try_send(Event::GaveUp { reason: give_up_reason(consecutive), }); // The breaker has tripped, so stop auto-restarting — but stay alive parked on the restart // channel so a user-triggered "Restart sync" can still reach the (otherwise dead) engine // and resume. The supervisor only truly exits when every Handle is dropped (app teardown). match restart_rx.recv().await { Some(()) => { breaker.reset(); tracing::info!(target: "pod::lifecycle", "restarting the sync engine on request after giving up"); let _ = events.try_send(Event::Restarted { #[cfg(test)] attempt: 0, }); } None => return, } } Some(plan) => { tracing::warn!( target: "pod::lifecycle", attempt = plan.attempt, consecutive = breaker.consecutive, backoff_secs = plan.backoff.as_secs(), "the sync engine died unexpectedly; restarting after backoff" ); // A manual restart short-circuits the backoff wait and resets the breaker so retries // resume immediately rather than after the (possibly capped) delay. tokio::select! { _ = tokio::time::sleep(plan.backoff) => {} signal = restart_rx.recv() => { if signal.is_none() { return; } breaker.reset(); } } let _ = events.try_send(Event::Restarted { #[cfg(test)] attempt: plan.attempt, }); } }, } }}
async fn run_job( db: &Database, esi: &Arc<esi::Client>, sso: &Arc<eve_sso::Client>, image: &Arc<eve_image::Client>, image_store: &images::Store, key: JobKey,) -> Result<Outcome, Error> { let grant = if key.kind.required_scope().is_empty() { None } else { let (owner_id, owner_type) = owner_of(key.subject); // Honor the persisted needs-reauth flag at dispatch: a flagged entity's token is known-bad, so // skip the work entirely rather than refresh-and-fail against it. The job reschedules on its // normal interval (a benign no-op), not the tight backoff loop, and stays skipped until the // TokenAudit clears the flag or a re-auth's run_now revives it. if is_flagged_needs_reauth(db, owner_id, owner_type).await { tracing::debug!(?key, "skipping privileged job for an entity flagged needs-reauth"); return Ok(Outcome::Skipped { reason: "entity needs re-authentication".to_string(), }); } let Some(grant) = token::fresh_token(db, sso, owner_id, owner_type).await? else { tracing::warn!( ?key, "privileged job has no usable credential; surfacing as needs-reauth" ); return Err(Error::Auth(format!( "{:?} for {:?} has no usable credential", key.kind, key.subject ))); }; let granted: HashSet<&str> = grant.scopes().iter().map(String::as_str).collect(); if !key.kind.is_scope_granted(key.subject, &granted) { tracing::debug!(?key, "skipping job whose required scope the credential does not grant"); return Ok(Outcome::Skipped { reason: "required scope not granted".to_string(), }); } Some(grant) }; let ctx = JobCtx { db, esi, image, image_store, key, grant: grant.as_ref(), // Only the global TokenAudit job needs raw SSO access (to force-refresh every credential); every // other job has already had its grant resolved above, so it gets None. sso: matches!(key.kind, JobKind::TokenAudit).then_some(sso.as_ref()), }; let outcome = job::run(&ctx).await; // Coalesce this job's HTTP-cache writes into one batched transaction. Each ESI call buffers its // cache upsert in memory (reads still see their own writes); flushing once per job collapses the // hundreds of one-row write transactions a sync used to issue into a few batched ones, dropping the // single SQLite writer's queue depth. Best-effort: a flush failure must not fail an otherwise-good // job — the entries simply remain buffered for the next flush. if let Err(error) = esi.http().flush_cache().await { tracing::warn!(?key, %error, "failed to flush HTTP cache buffer after job"); } outcome}
async fn is_flagged_needs_reauth(db: &Database, owner_id: i64, owner_type: OwnerType) -> bool { matches!(infra::get(db, owner_id, owner_type).await, Ok(Some(credential)) if credential.needs_reauth())}
fn is_pending_outcome(outcome: &Outcome) -> bool { matches!(outcome, Outcome::Blocked { .. } | Outcome::NotReady)}
fn is_permanent_failure(error: &Error) -> bool { match error { Error::Auth(message) | Error::Internal(message) => { message.contains("needs re-authentication") || message.contains("has no authorizing character") || message.contains("has no usable credential") } // A revoked refresh token surfaces as HTTP 400 (invalid_grant) from the SSO refresh path that // every privileged job runs through before its ESI call. Classifying it permanent parks the job // (reschedule_permanent) instead of looping it on the ~300s backoff against a dead token; the // TokenAudit job and a re-auth's run_now are what revive it. _ => token::is_revoked_refresh(error), }}
fn future_seed(row: &SyncLedger, kind: JobKind, now: Instant) -> Option<Instant> { // Market-order jobs bypass ledger eligibility on launch so a prior session's backoff can't stall // the first sync pass; the floor below only suppresses this when a real sync already completed // moments ago, avoiding a duplicate fetch on a rapid restart. if kind.is_market_order() && !synced_within_launch_floor(row) { return Some(now); } let eligible = ledger_eligible_at(row, kind)?; let delay = (eligible - Utc::now()).to_std().ok()?; Some(now + delay.min(kind.interval()))}
fn synced_within_launch_floor(row: &SyncLedger) -> bool { let Some(last_success_at) = row.last_success_at().as_deref().and_then(parse_rfc3339) else { return false; }; Utc::now().signed_duration_since(last_success_at) < ChronoDuration::from_std(MARKET_ORDER_LAUNCH_FLOOR).unwrap_or_else(|_| ChronoDuration::zero())}
fn ledger_eligible_at(row: &SyncLedger, kind: JobKind) -> Option<DateTime<Utc>> { if let Some(next_eligible_at) = row.next_eligible_at().as_deref().and_then(parse_rfc3339) { return Some(next_eligible_at); } let last_success_at = row.last_success_at().as_deref().and_then(parse_rfc3339)?; Some(last_success_at + ChronoDuration::from_std(kind.interval()).unwrap_or_else(|_| ChronoDuration::zero()))}
fn parse_rfc3339(value: &str) -> Option<DateTime<Utc>> { DateTime::parse_from_rfc3339(value) .ok() .map(|at| at.with_timezone(&Utc))}
fn global_kinds() -> Vec<JobKind> { JobKind::ALL.iter().copied().filter(|kind| kind.is_global()).collect()}
fn owner_of(subject: Subject) -> (i64, OwnerType) { match subject { Subject::Character(id) => (id, OwnerType::Character), Subject::Corporation(id) => (id, OwnerType::Corporation), }}
fn profile_kind(subject: Subject) -> JobKind { match subject { Subject::Character(_) => JobKind::CharacterProfile, Subject::Corporation(_) => JobKind::CorporationProfile, }}
fn subject_from(owner_id: i64, owner_type: OwnerType) -> Subject { match owner_type { OwnerType::Character => Subject::Character(owner_id), OwnerType::Corporation => Subject::Corporation(owner_id), }}
#[cfg(test)]mod tests { use wiremock::{ Mock, MockServer, ResponseTemplate, matchers::{method, path}, };
use super::*; use crate::{ clients::http, store::{self, repo::character}, };
async fn spawn_engine(base_url: String) -> (Handle, mpsc::Receiver<Event>, Database, tempfile::TempDir) { let db = store::open_test().await.unwrap(); let (handle, events, images_dir) = spawn_engine_with_db(db.clone(), base_url).await; (handle, events, db, images_dir) }
async fn spawn_engine_with_db(db: Database, base_url: String) -> (Handle, mpsc::Receiver<Event>, tempfile::TempDir) { let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), base_url.clone())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), base_url)); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let store = images::Store::new(images_dir.path().to_path_buf()); let (handle, events) = spawn(db.clone(), db, esi, sso, image, store, FeatureFlags::default()); (handle, events, images_dir) }
async fn spawn_run(base_url: String) -> (mpsc::UnboundedSender<Command>, tokio::task::JoinHandle<ExitReason>) { let db = store::open_test().await.unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), base_url.clone())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), base_url)); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let store = images::Store::new(images_dir.path().to_path_buf()); let (events, _event_rx) = mpsc::channel(EVENT_BUFFER); let (command_tx, command_rx) = mpsc::unbounded_channel(); let engine = Engine { db: db.clone(), drain_at: Instant::now(), esi, events, housekeeping: db, image, image_store: store, outbox: Registry::new(), paused_until: Instant::now(), schedule: Schedule::with_features(FeatureFlags::default()), sso, }; let run = tokio::spawn(async move { let mut command_rx = command_rx; engine.run(&mut command_rx).await }); (command_tx, run) }
async fn wait_for<F: Fn(&Event) -> bool>(events: &mut mpsc::Receiver<Event>, predicate: F) -> Event { loop { let event = tokio::time::timeout(Duration::from_secs(5), events.recv()) .await .expect("timed out waiting for event") .expect("event channel closed"); if predicate(&event) { return event; } } }
async fn mount_json(server: &MockServer, route: &str, body: serde_json::Value) { Mock::given(method("GET")) .and(path(route)) .respond_with(ResponseTemplate::new(200).set_body_json(body)) .mount(server) .await; }
async fn mount_character_profile(server: &MockServer, character_id: i64) { mount_json( server, &format!("/characters/{character_id}/"), serde_json::json!({ "alliance_id": 300, "birthday": "2010-01-01T00:00:00Z", "bloodline_id": 5, "corporation_id": 200, "gender": "male", "name": "Test Pilot", "race_id": 1, }), ) .await; mount_json( server, "/corporations/200/", serde_json::json!({ "alliance_id": 300, "ceo_id": character_id, "creator_id": character_id, "member_count": 42, "name": "Test Corp", "tax_rate": 0.1, "ticker": "TST", }), ) .await; mount_json( server, "/alliances/300/", serde_json::json!({ "creator_corporation_id": 200, "creator_id": character_id, "date_founded": "2005-01-01T00:00:00Z", "name": "Test Alliance", "ticker": "TSTA", }), ) .await; mount_json( server, "/universe/races/", serde_json::json!([{ "alliance_id": 300, "description": "The Caldari.", "name": "Caldari", "race_id": 1 }]), ) .await; mount_json( server, "/universe/bloodlines/", serde_json::json!([ { "bloodline_id": 5, "charisma": 6, "corporation_id": 200, "description": "The Civire.", "intelligence": 7, "memory": 5, "name": "Civire", "perception": 5, "race_id": 1, "ship_type_id": 601, "willpower": 5 }, ]), ) .await; Mock::given(method("GET")) .and(path(format!("/characters/{character_id}/portrait"))) .respond_with(ResponseTemplate::new(200).set_body_raw(vec![1u8, 2, 3], "image/jpeg")) .mount(server) .await; }
async fn seed_ship_type(db: &Database) { sqlx::query("INSERT INTO item_categories (id, name, published) VALUES (6, 'Ship', 1)") .execute(db.writer()) .await .unwrap(); sqlx::query("INSERT INTO item_groups (id, category_id, name, published) VALUES (25, 6, 'Frigate', 1)") .execute(db.writer()) .await .unwrap(); sqlx::query( "INSERT INTO item_types (id, group_id, description, name, published) VALUES (601, 25, 'Merlin', 'Merlin', 1)", ) .execute(db.writer()) .await .unwrap(); }
async fn seed_character(db: &Database, id: i64) { let mut tx = db.writer().begin().await.unwrap(); sqlx::query("PRAGMA defer_foreign_keys = ON") .execute(&mut *tx) .await .unwrap(); sqlx::query("INSERT INTO races (id, alliance_id, description, name) VALUES (1, 0, 'r', 'Race')") .execute(&mut *tx) .await .unwrap(); sqlx::query( "INSERT INTO corporations (id, ceo_id, creator_id, member_count, name, tax_rate, ticker) \ VALUES (1, 0, 0, 1, 'Corp', 0.0, 'CRP')", ) .execute(&mut *tx) .await .unwrap(); sqlx::query( "INSERT INTO bloodlines (id, corporation_id, race_id, charisma, description, intelligence, \ memory, name, perception, willpower) VALUES (1, 1, 1, 1, 'b', 1, 1, 'Bloodline', 1, 1)", ) .execute(&mut *tx) .await .unwrap(); sqlx::query( "INSERT INTO characters (id, bloodline_id, corporation_id, race_id, birthday, gender, name) \ VALUES (?, 1, 1, 1, '2010-01-01T00:00:00Z', 'male', 'NPC Pilot')", ) .bind(id) .execute(&mut *tx) .await .unwrap(); tx.commit().await.unwrap(); }
async fn mount_skill_picture(server: &MockServer, character_id: i64) { mount_json( server, &format!("/characters/{character_id}/skills/"), serde_json::json!({ "skills": [ { "active_skill_level": 5, "skill_id": 3300, "skillpoints_in_skill": 256000, "trained_skill_level": 5 }, ], "total_sp": 256000, "unallocated_sp": 15000, }), ) .await; mount_json( server, &format!("/characters/{character_id}/skillqueue/"), serde_json::json!([ { "finish_date": "2026-06-01T00:00:00Z", "finished_level": 5, "queue_position": 0, "skill_id": 3300 }, ]), ) .await; mount_json( server, &format!("/characters/{character_id}/attributes/"), serde_json::json!({ "charisma": 20, "intelligence": 22, "memory": 21, "perception": 20, "willpower": 20, "bonus_remaps": 2, "last_remap_date": "2023-04-01T12:00:00Z", "accrued_remap_cooldown_date": "2024-04-01T12:00:00Z", }), ) .await; mount_json( server, &format!("/characters/{character_id}/implants/"), serde_json::json!([9899]), ) .await; mount_json( server, "/universe/types/9899/", serde_json::json!({ "description": "A memory implant.", "group_id": 300, "name": "Memory Augmentation - Basic", "published": true, "type_id": 9899, "dogma_attributes": [{ "attribute_id": 177, "value": 3.0 }], }), ) .await; mount_json( server, "/universe/types/3300/", serde_json::json!({ "description": "Gunnery.", "group_id": 255, "market_group_id": 1112, "name": "Gunnery", "published": true, "type_id": 3300, "dogma_attributes": [ { "attribute_id": 275, "value": 1.0 }, { "attribute_id": 180, "value": 167.0 }, { "attribute_id": 181, "value": 168.0 }, ], }), ) .await; mount_json( server, "/universe/groups/255/", serde_json::json!({ "category_id": 16, "group_id": 255, "name": "Gunnery", "published": true, "types": [3300] }), ) .await; mount_json( server, "/universe/categories/16/", serde_json::json!({ "category_id": 16, "groups": [255], "name": "Skill", "published": true }), ) .await; mount_json( server, "/markets/groups/1112/", serde_json::json!({ "description": "Skill books.", "market_group_id": 1112, "name": "Skills", "parent_group_id": 1111, "types": [3300], }), ) .await; mount_json( server, "/markets/groups/1111/", serde_json::json!({ "description": "All market groups.", "market_group_id": 1111, "name": "Market", "types": [], }), ) .await; }
mod backoff_for { use pretty_assertions::assert_eq;
use super::*;
#[test] fn it_clamps_at_the_cap_and_never_overflows() { assert_eq!(super::super::backoff_for(20), RESPAWN_BACKOFF_CAP); assert_eq!(super::super::backoff_for(u32::MAX), RESPAWN_BACKOFF_CAP); }
#[test] fn it_grows_exponentially_from_the_base() { assert_eq!(super::super::backoff_for(1), RESPAWN_BACKOFF_BASE); assert_eq!(super::super::backoff_for(2), RESPAWN_BACKOFF_BASE * 2); assert_eq!(super::super::backoff_for(3), RESPAWN_BACKOFF_BASE * 4); } }
mod breaker { use pretty_assertions::assert_eq;
use super::*;
#[test] fn it_clears_the_streak_and_attempt_count_on_reset() { let mut breaker = Breaker::default(); for _ in 0..RESPAWN_BREAKER_THRESHOLD - 1 { breaker.record_restart(Duration::ZERO); }
breaker.reset(); let after_reset: Vec<_> = (0..RESPAWN_BREAKER_THRESHOLD - 1) .map(|_| breaker.record_restart(Duration::ZERO)) .collect();
assert!( after_reset.iter().all(Option::is_some), "after a reset the breaker grants a full fresh run of restarts before tripping again" ); assert_eq!( after_reset .first() .and_then(|plan| plan.as_ref()) .map(|plan| plan.attempt), Some(1), "a reset restarts the attempt counter from one" ); }
#[test] fn it_plans_a_restart_with_a_rising_attempt_count() { let mut breaker = Breaker::default();
let first = breaker .record_restart(Duration::ZERO) .expect("a first death plans a restart"); let second = breaker .record_restart(Duration::ZERO) .expect("a second death plans a restart");
assert_eq!(first.attempt, 1); assert_eq!(second.attempt, 2); }
#[test] fn it_resets_the_streak_when_an_engine_survives_the_window() { let mut breaker = Breaker::default(); for _ in 0..RESPAWN_BREAKER_THRESHOLD - 1 { assert!(breaker.record_restart(Duration::ZERO).is_some()); }
let after_a_long_run = breaker.record_restart(RESPAWN_BREAKER_WINDOW);
assert!( after_a_long_run.is_some(), "an engine that lived past the window resets the streak, so the next death restarts rather than trips" ); }
#[test] fn it_trips_and_gives_up_after_the_threshold_of_rapid_deaths() { let mut breaker = Breaker::default();
let plans: Vec<_> = (0..RESPAWN_BREAKER_THRESHOLD) .map(|_| breaker.record_restart(Duration::ZERO)) .collect();
let restarts = plans.iter().take_while(|plan| plan.is_some()).count();
assert_eq!( restarts, (RESPAWN_BREAKER_THRESHOLD - 1) as usize, "the supervisor restarts up to one shy of the threshold, then trips" ); assert!( plans.last().expect("at least one plan").is_none(), "the threshold-th rapid death trips the breaker and stops restarting" ); } }
mod disposition { use super::*;
#[test] fn it_restarts_on_a_panic_payload() { let restart = super::super::disposition(Err(Box::new("boom")));
assert!(matches!(restart, Disposition::Restart)); }
#[test] fn it_stops_deliberately_on_shutdown() { let stop = super::super::disposition(Ok(ExitReason::Shutdown));
assert!(matches!( stop, Disposition::Stop { deliberate: true } )); }
#[test] fn it_stops_without_deliberation_on_channel_close() { let stop = super::super::disposition(Ok(ExitReason::ChannelClosed));
assert!(matches!( stop, Disposition::Stop { deliberate: false } )); } }
mod drain { use pretty_assertions::assert_eq;
use super::*; use crate::{ store::{model::OwnerType, repo::infra}, sync::outbox::{OutboxKind, test_support::*}, };
async fn spawn_engine_with_drain(base_url: String) -> (Database, StubCalls, tempfile::TempDir) { let db = store::open_test().await.unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), base_url.clone())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), base_url)); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let calls = StubCalls::default(); let registry_calls = calls.clone(); let (_handle, _events) = spawn_with_registry( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), move || Registry::new().with(Box::new(StubHandler::new(OutboxKind::MailSend, registry_calls.clone()))), ); (db, calls, images_dir) }
async fn spawn_engine_with_drain_and_handle( base_url: String, ) -> (Handle, mpsc::Receiver<Event>, Database, StubCalls, tempfile::TempDir) { let db = store::open_test().await.unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), base_url.clone())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), base_url)); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let calls = StubCalls::default(); let registry_calls = calls.clone(); let (handle, events) = spawn_with_registry( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), move || Registry::new().with(Box::new(StubHandler::new(OutboxKind::MailSend, registry_calls.clone()))), ); (handle, events, db, calls, images_dir) }
async fn spawn_engine_over_db( db: &Database, base_url: String, ) -> (Handle, mpsc::Receiver<Event>, StubCalls, tempfile::TempDir) { let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), base_url.clone())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), base_url)); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let calls = StubCalls::default(); let registry_calls = calls.clone(); let (handle, events) = spawn_with_registry( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), move || Registry::new().with(Box::new(StubHandler::new(OutboxKind::MailSend, registry_calls.clone()))), ); (handle, events, calls, images_dir) }
async fn status_of(db: &Database, id: i64) -> String { sqlx::query_scalar::<_, String>("SELECT status FROM outbox WHERE id = ?") .bind(id) .fetch_one(&db.0) .await .unwrap() }
async fn wait_for_status(db: &Database, id: i64, want: &str) { let deadline = tokio::time::Instant::now() + Duration::from_secs(10); loop { if status_of(db, id).await == want { return; } if tokio::time::Instant::now() >= deadline { panic!( "outbox row {id} never reached status {want:?} (was {:?})", status_of(db, id).await ); } tokio::time::sleep(Duration::from_millis(50)).await; } }
#[tokio::test] async fn it_does_not_re_execute_an_already_done_row_after_restart() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); infra::upsert(&db, 7400, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let row = infra::append( &db, OwnerType::Character, 7400, "mail.send", "{\"body\":\"sent\"}", None, ) .await .unwrap(); sqlx::query("UPDATE outbox SET status = 'done', updated_at = ? WHERE id = ?") .bind(Utc::now().to_rfc3339()) .bind(row.id()) .execute(db.writer()) .await .unwrap();
let (_handle, _events, calls, _images) = spawn_engine_over_db(&db, server.uri()).await; tokio::time::sleep(Duration::from_millis(200)).await;
assert_eq!( calls.executes(), 0, "a terminal done row must never be re-claimed or re-executed by a restarted engine" ); assert_eq!( status_of(&db, row.id()).await, "done", "the done row stays done; replay is a no-op" ); }
#[tokio::test] async fn it_drains_a_durable_row_after_a_kill_and_restart() { let server = MockServer::start().await; let db = store::open_test().await.unwrap();
let row = infra::append( &db, OwnerType::Character, 7300, "mail.send", "{\"body\":\"queued\"}", None, ) .await .unwrap();
{ let (handle, events, first_calls, _images) = spawn_engine_over_db(&db, server.uri()).await; tokio::time::sleep(Duration::from_millis(200)).await; let undrained = status_of(&db, row.id()).await; assert!( undrained == "pending" || undrained == "inflight", "the un-credentialed row must stay drainable while the first engine lives (was {undrained:?})" ); assert_eq!( first_calls.executes(), 0, "the first engine must not perform an ESI write for a subject it cannot authenticate" ); drop(handle); drop(events); }
infra::upsert(&db, 7300, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let (_handle, _events, calls, _images) = spawn_engine_over_db(&db, server.uri()).await;
wait_for_status(&db, row.id(), "done").await; assert_eq!( calls.executes(), 1, "the restarted engine performs the queued ESI write exactly once" ); }
#[tokio::test] async fn it_drains_a_pre_seeded_credentialed_row_to_done_on_tick() { let server = MockServer::start().await; let (db, calls, _images) = spawn_engine_with_drain(server.uri()).await; infra::upsert(&db, 8800, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let owned = infra::append(&db, OwnerType::Character, 8800, "mail.send", "{\"body\":\"hi\"}", None) .await .unwrap(); let unowned = infra::append(&db, OwnerType::Character, 8801, "mail.send", "{\"body\":\"hi\"}", None) .await .unwrap();
wait_for_status(&db, owned.id(), "done").await;
assert_eq!( calls.executes(), 1, "the credentialed row's ESI write should run exactly once" ); let unowned_status = status_of(&db, unowned.id()).await; assert!( unowned_status == "pending" || unowned_status == "inflight", "an un-credentialed subject's row must stay drainable, never done or failed (was {unowned_status:?})" ); }
#[tokio::test] async fn it_drains_a_row_on_a_control_channel_nudge() { let server = MockServer::start().await; let (handle, _events, db, calls, _images) = spawn_engine_with_drain_and_handle(server.uri()).await; infra::upsert(&db, 9900, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let row = infra::append(&db, OwnerType::Character, 9900, "mail.send", "{\"body\":\"hi\"}", None) .await .unwrap();
handle.drain();
wait_for_status(&db, row.id(), "done").await; assert!(calls.executes() >= 1, "the nudged drain ran the row's ESI write"); }
#[tokio::test] async fn it_prunes_done_rows_past_the_retention_window_on_tick() { let server = MockServer::start().await; let (db, _calls, _images) = spawn_engine_with_drain(server.uri()).await; let stale = infra::append(&db, OwnerType::Character, 8900, "mail.send", "{}", None) .await .unwrap(); let fresh = infra::append(&db, OwnerType::Character, 8901, "mail.send", "{}", None) .await .unwrap(); let long_ago = (Utc::now() - ChronoDuration::days(1)).to_rfc3339(); let now = Utc::now().to_rfc3339(); sqlx::query("UPDATE outbox SET status = 'done', updated_at = ? WHERE id = ?") .bind(&long_ago) .bind(stale.id()) .execute(db.writer()) .await .unwrap(); sqlx::query("UPDATE outbox SET status = 'done', updated_at = ? WHERE id = ?") .bind(&now) .bind(fresh.id()) .execute(db.writer()) .await .unwrap();
let deadline = tokio::time::Instant::now() + Duration::from_secs(10); loop { let remaining = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM outbox WHERE id = ?") .bind(stale.id()) .fetch_one(&db.0) .await .unwrap(); if remaining == 0 { break; } if tokio::time::Instant::now() >= deadline { panic!("stale done row was never pruned"); } tokio::time::sleep(Duration::from_millis(50)).await; }
assert_eq!( status_of(&db, fresh.id()).await, "done", "a done row inside the retention window must be retained for the indicator" ); }
#[tokio::test] async fn it_redrains_a_row_stuck_inflight_by_a_crashed_engine() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); infra::upsert(&db, 7500, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let row = infra::append(&db, OwnerType::Character, 7500, "mail.send", "{\"body\":\"mid\"}", None) .await .unwrap(); sqlx::query("UPDATE outbox SET status = 'inflight', next_attempt_at = ?, updated_at = ? WHERE id = ?") .bind(Utc::now().to_rfc3339()) .bind(Utc::now().to_rfc3339()) .bind(row.id()) .execute(db.writer()) .await .unwrap();
let (_handle, _events, calls, _images) = spawn_engine_over_db(&db, server.uri()).await;
wait_for_status(&db, row.id(), "done").await; assert_eq!( calls.executes(), 1, "the fresh engine re-claims the stuck inflight row and performs its ESI write" ); } }
mod exit_reason { use pretty_assertions::assert_eq;
use super::*;
#[tokio::test] async fn it_returns_channel_closed_when_the_command_sender_is_dropped() { let server = MockServer::start().await; let (commands, run) = spawn_run(server.uri()).await;
drop(commands); let reason = tokio::time::timeout(Duration::from_secs(5), run) .await .expect("the run loop should exit promptly once the command channel closes") .expect("the run task should not panic");
assert_eq!( reason, ExitReason::ChannelClosed, "dropping the only command sender closes the channel and yields ExitReason::ChannelClosed" ); assert!( !reason.is_deliberate(), "a channel close is an unexpected exit, not a deliberate shutdown" ); }
#[tokio::test] async fn it_returns_shutdown_when_a_shutdown_command_breaks_the_loop() { let server = MockServer::start().await; let (commands, run) = spawn_run(server.uri()).await;
commands.send(Command::Shutdown).unwrap(); let reason = tokio::time::timeout(Duration::from_secs(5), run) .await .expect("the run loop should exit promptly on shutdown") .expect("the run task should not panic");
assert_eq!( reason, ExitReason::Shutdown, "a deliberate Shutdown command yields the deliberate ExitReason::Shutdown" ); assert!( reason.is_deliberate(), "Shutdown is a deliberate exit, not an unexpected one" ); } }
mod global_kinds { use pretty_assertions::assert_eq;
use super::*;
#[test] fn it_is_the_complement_of_the_per_subject_set() { let global = super::super::global_kinds(); for kind in JobKind::ALL.iter().copied() { let per_subject = kind.applies_to(Subject::Character(1)) || kind.applies_to(Subject::Corporation(1)); assert_eq!( global.contains(&kind), !per_subject, "{kind:?} must be global iff it applies to no real subject" ); } }
#[test] fn it_selects_kinds_that_apply_to_no_real_subject() { let kinds = super::super::global_kinds();
assert!( kinds.contains(&JobKind::NetWorthSnapshot), "the global lane must include the net-worth snapshot, got {kinds:?}" ); for kind in &kinds { assert!( !kind.applies_to(Subject::Character(1)) && !kind.applies_to(Subject::Corporation(1)), "{kind:?} applies to a real subject, so it belongs to per-subject discovery, not the global lane" ); } assert!(!kinds.contains(&JobKind::CharacterProfile)); assert!(!kinds.contains(&JobKind::CorporationWallet)); } }
mod global_lane { use pretty_assertions::assert_eq;
use super::*; use crate::store::{model::OwnerType, repo::finance};
async fn insert_journal(db: &Database, id: i64, character_id: i64, balance: f64) { sqlx::query( "INSERT INTO character_wallet_journal (id, character_id, date, description, ref_type, amount, balance) \ VALUES (?, ?, ?, ?, ?, ?, ?)", ) .bind(id) .bind(character_id) .bind("2026-01-01") .bind("Test") .bind("test") .bind(balance) .bind(balance) .execute(db.writer()) .await .unwrap(); }
#[tokio::test] async fn it_does_not_enroll_the_snapshot_when_wallet_and_assets_are_off() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); seed_character(&db, 7701).await; infra::upsert(&db, 7701, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); insert_journal(&db, 1, 7701, 999.0).await; let flags: FeatureFlags = toml::from_str("wallet = false\nasset_tracking = false").unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), flags, );
let deadline = tokio::time::Instant::now() + Duration::from_secs(2); while let Ok(Some(event)) = tokio::time::timeout_at(deadline, events.recv()).await { if let Event::Started { key, } | Event::Finished { key, .. } = event { assert_ne!( key.kind, JobKind::NetWorthSnapshot, "the snapshot lane must not enroll while both Wallet and Assets are off" ); } }
let rows = finance::for_character_since(&db, 7701, "2000-01-01").await.unwrap(); assert!( rows.is_empty(), "no snapshot should be written while both Wallet and Assets are off" ); }
#[tokio::test] async fn it_forces_a_net_worth_snapshot_immediately_after_a_wallet_sync() { let server = MockServer::start().await; mount_json( &server, "/characters/7702/wallet/journal/", serde_json::json!([ { "amount": 1000.0, "balance": 50_000.0, "date": "2026-05-30T12:00:00Z", "description": "Donation", "id": 123_456_789_i64, "ref_type": "player_donation" }, ]), ) .await; mount_json(&server, "/characters/7702/wallet/transactions/", serde_json::json!([])).await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); seed_character(&db, 7702).await; infra::upsert( &db, 7702, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_WALLET), ) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
wait_for( &mut events, |event| matches!(event, Event::Finished { key, .. } if key.kind == JobKind::CharacterWallet), ) .await; wait_for( &mut events, |event| matches!(event, Event::Finished { key, .. } if key.kind == JobKind::NetWorthSnapshot), ) .await;
let rows = finance::for_character_since(&db, 7702, "2000-01-01").await.unwrap(); assert_eq!( rows.len(), 2, "the forced snapshot writes today's row plus the backfilled journal day" ); assert_eq!(rows[0].liquid(), 50_000.0); }
#[tokio::test] async fn it_runs_the_global_net_worth_snapshot_at_startup() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); seed_character(&db, 7700).await; infra::upsert(&db, 7700, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); insert_journal(&db, 1, 7700, 1_234.0).await; let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
wait_for( &mut events, |event| matches!(event, Event::Finished { key, .. } if key.kind == JobKind::NetWorthSnapshot), ) .await;
let rows = finance::for_character_since(&db, 7700, "2000-01-01").await.unwrap(); assert_eq!( rows.len(), 2, "the global lane writes today's snapshot row plus the backfilled journal day" ); assert_eq!(rows[0].liquid(), 1_234.0); assert_eq!(rows[0].net_worth(), 1_234.0); } }
mod hydrate { use super::*;
async fn seed_ledger(db: &Database, kind: &str, next_eligible_at: Option<&str>, last_success_at: Option<&str>) { sync_ledger::upsert( db, OwnerType::Character, 0, kind, "synced", 0, None, last_success_at, next_eligible_at, ) .await .unwrap(); }
#[tokio::test] async fn it_breaks_the_run_loop_on_a_shutdown_command() { let server = MockServer::start().await; let (handle, mut events, _db, _images) = spawn_engine(server.uri()).await;
handle.shutdown();
let drained = tokio::time::timeout(Duration::from_secs(5), async { while events.recv().await.is_some() {} }).await;
assert!( drained.is_ok(), "shutdown breaks the loop, dropping the engine and closing its event channel" ); }
#[tokio::test] async fn it_does_not_rerun_a_job_whose_ledger_eligibility_is_in_the_future() { let server = MockServer::start().await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); let future = (Utc::now() + ChronoDuration::minutes(30)).to_rfc3339(); seed_ledger(&db, "IndustryCostIndices", Some(&future), None).await; seed_ledger(&db, "KillmailDetailBackfill", Some(&future), None).await; seed_ledger(&db, "KillmailReconcile", Some(&future), None).await; seed_ledger(&db, "MarketPrices", Some(&future), None).await; seed_ledger(&db, "NetWorthSnapshot", Some(&future), None).await; seed_ledger(&db, "TokenAudit", Some(&future), None).await; seed_ledger(&db, "WalletJournalReconcile", Some(&future), None).await;
let (_handle, mut events, _images) = spawn_engine_with_db(db, server.uri()).await;
let started = tokio::time::timeout(Duration::from_millis(300), async { loop { match events.recv().await { Some(Event::Started { key, }) => return Some(key), Some(_) => continue, None => return None, } } }) .await; assert!( started.is_err(), "a job seeded with a future ledger eligibility must not run at startup, got {started:?}" ); }
#[tokio::test] async fn it_falls_back_to_last_success_plus_interval_when_next_eligible_is_absent() { let db = store::open_test().await.unwrap(); let recent_success = (Utc::now() - ChronoDuration::minutes(1)).to_rfc3339(); seed_ledger(&db, "CharacterProfile", None, Some(&recent_success)).await; let row = sync_ledger::get(&db, OwnerType::Character, 0, "CharacterProfile") .await .unwrap() .unwrap();
assert!( future_seed(&row, JobKind::CharacterProfile, Instant::now()).is_some(), "a recent success within one interval defers the next run even without an explicit next-eligible" ); }
#[tokio::test] async fn it_reruns_a_job_whose_ledger_eligibility_has_passed() { let server = MockServer::start().await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); let past = (Utc::now() - ChronoDuration::hours(1)).to_rfc3339(); seed_ledger(&db, "MarketPrices", Some(&past), None).await;
let (_handle, mut events, _images) = spawn_engine_with_db(db, server.uri()).await;
wait_for( &mut events, |event| matches!(event, Event::Started { key } if key.kind == JobKind::MarketPrices), ) .await; }
#[tokio::test] async fn it_seeds_a_future_eligible_row_within_one_interval() { let db = store::open_test().await.unwrap(); let future = (Utc::now() + ChronoDuration::minutes(30)).to_rfc3339(); seed_ledger(&db, "CharacterProfile", Some(&future), None).await; let row = sync_ledger::get(&db, OwnerType::Character, 0, "CharacterProfile") .await .unwrap() .unwrap(); let now = Instant::now();
let seed = future_seed(&row, JobKind::CharacterProfile, now).expect("a future row seeds a deferred next-run");
assert!(seed > now, "the seed defers the next run past now"); assert!( seed <= now + JobKind::CharacterProfile.interval(), "the seed is clamped to at most one interval" ); }
#[tokio::test] async fn it_yields_no_seed_for_a_stale_row() { let db = store::open_test().await.unwrap(); let past = (Utc::now() - ChronoDuration::hours(1)).to_rfc3339(); seed_ledger(&db, "CharacterProfile", Some(&past), None).await; let row = sync_ledger::get(&db, OwnerType::Character, 0, "CharacterProfile") .await .unwrap() .unwrap();
assert!( future_seed(&row, JobKind::CharacterProfile, Instant::now()).is_none(), "a past eligibility leaves the kind due now" ); }
#[tokio::test] async fn it_forces_a_market_order_kind_due_now_at_launch_despite_a_future_eligibility() { let db = store::open_test().await.unwrap(); let future = (Utc::now() + ChronoDuration::minutes(30)).to_rfc3339(); seed_ledger(&db, "CharacterMarketOrders", Some(&future), None).await; let row = sync_ledger::get(&db, OwnerType::Character, 0, "CharacterMarketOrders") .await .unwrap() .unwrap(); let now = Instant::now();
let seed = future_seed(&row, JobKind::CharacterMarketOrders, now);
assert_eq!( seed, Some(now), "a market-order kind ignores a deferred eligibility and runs on the first launch pass" ); }
#[tokio::test] async fn it_defers_a_market_order_kind_synced_within_the_launch_floor() { let db = store::open_test().await.unwrap(); let recent = (Utc::now() - ChronoDuration::seconds(5)).to_rfc3339(); seed_ledger(&db, "CharacterMarketOrders", None, Some(&recent)).await; let row = sync_ledger::get(&db, OwnerType::Character, 0, "CharacterMarketOrders") .await .unwrap() .unwrap(); let now = Instant::now();
let seed = future_seed(&row, JobKind::CharacterMarketOrders, now) .expect("a success inside the launch floor defers within one interval");
assert!( seed > now, "a market-order synced inside the launch floor keeps its deferred next run" ); }
#[tokio::test] async fn it_seeds_market_orders_due_now_while_other_kinds_defer_at_launch() { let db = store::open_test().await.unwrap(); let future = (Utc::now() + ChronoDuration::minutes(30)).to_rfc3339(); seed_ledger(&db, "CharacterMarketOrders", Some(&future), None).await; seed_ledger(&db, "CharacterProfile", Some(&future), None).await; let rows = vec![ sync_ledger::get(&db, OwnerType::Character, 0, "CharacterMarketOrders") .await .unwrap() .unwrap(), sync_ledger::get(&db, OwnerType::Character, 0, "CharacterProfile") .await .unwrap() .unwrap(), ]; let now = Instant::now();
let seeds = future_seeds(&rows, now);
assert_eq!( seeds.get(&JobKind::CharacterMarketOrders), Some(&now), "market orders are due now on the first launch pass" ); assert!( seeds.get(&JobKind::CharacterProfile).is_some_and(|at| *at > now), "non-market kinds keep their last-success-plus-interval seed" ); }
async fn seed_ledger_outcome(db: &Database, kind: &str, outcome: &str, next_eligible_at: Option<&str>) { let success = matches!(outcome, "synced" | "empty").then_some("2026-01-01T00:00:00Z"); sync_ledger::upsert( db, OwnerType::Character, 0, kind, outcome, 0, None, success, next_eligible_at, ) .await .unwrap(); }
#[tokio::test] async fn it_emits_a_seeded_event_for_a_fresh_global_job_at_launch() { let server = MockServer::start().await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); let future = (Utc::now() + ChronoDuration::minutes(30)).to_rfc3339(); seed_ledger_outcome(&db, "MarketPrices", "synced", Some(&future)).await;
let (_handle, mut events, _images) = spawn_engine_with_db(db, server.uri()).await;
let seeded = wait_for( &mut events, |event| matches!(event, Event::Seeded { key, .. } if key.kind == JobKind::MarketPrices), ) .await; let Event::Seeded { outcome, next_in_secs, .. } = seeded else { panic!("expected a Seeded event"); }; assert!( matches!(outcome, Outcome::Synced { .. }), "a synced ledger row seeds Synced" ); assert!( next_in_secs.is_some_and(|secs| secs > 0), "a future-eligible fresh job seeds a countdown rather than a bare Queued" ); }
#[test] fn it_maps_each_ledger_outcome_token_back_to_an_outcome() { let row = |token: &str, reason: Option<&str>| { SyncLedger::new_for_test( "CharacterProfile".to_string(), token.to_string(), reason.map(str::to_string), ) };
assert!(matches!( seeded_outcome(&row("synced", None)), Some(Outcome::Synced { .. }) )); assert!(matches!(seeded_outcome(&row("empty", None)), Some(Outcome::Empty))); assert!(matches!( seeded_outcome(&row("not_ready", None)), Some(Outcome::NotReady) )); assert_eq!( seeded_outcome(&row("blocked", Some("no scope"))), Some(Outcome::Blocked { reason: "no scope".to_string() }) ); assert_eq!( seeded_outcome(&row("failed", Some("token expired"))), Some(Outcome::Failed { reason: "token expired".to_string() }) ); assert!(seeded_outcome(&row("nonsense", None)).is_none()); } }
mod is_pending_outcome { use super::*;
#[test] fn it_does_not_treat_a_benign_empty_or_synced_outcome_as_pending() { assert!( !super::super::is_pending_outcome(&Outcome::Empty), "a benign empty result is a success, not a pending retry" ); assert!(!super::super::is_pending_outcome(&Outcome::Synced { rows_touched: 3 })); }
#[test] fn it_treats_blocked_and_not_ready_as_pending() { assert!(super::super::is_pending_outcome(&Outcome::Blocked { reason: "no scope".to_owned() })); assert!(super::super::is_pending_outcome(&Outcome::NotReady)); } }
mod is_permanent_failure { use super::*;
async fn http_error_with_status(status: u16) -> Error { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/x")) .respond_with(ResponseTemplate::new(status)) .mount(&server) .await; let resp = reqwest::get(format!("{}/x", server.uri())).await.unwrap(); Error::Http(resp.error_for_status().unwrap_err()) }
#[test] fn it_classifies_a_missing_authorizing_character_as_permanent() { let error = Error::Internal("corporation credential for 1 has no authorizing character".to_owned());
assert!(super::super::is_permanent_failure(&error)); }
#[test] fn it_classifies_a_missing_credential_auth_error_as_permanent() { let error = Error::Auth("CharacterWallet for Character(7) has no usable credential".to_owned());
assert!(super::super::is_permanent_failure(&error)); }
#[test] fn it_classifies_a_needs_reauthentication_failure_as_permanent() { let error = Error::Internal( "authorizing character 5 no longer holds an accounting role; needs re-authentication".to_owned(), );
assert!(super::super::is_permanent_failure(&error)); }
#[tokio::test] async fn it_classifies_a_revoked_token_refresh_400_as_permanent() { let error = http_error_with_status(400).await;
assert!( super::super::is_permanent_failure(&error), "a revoked refresh token surfaces as HTTP 400 and must park rather than backoff-hammer" ); }
#[tokio::test] async fn it_does_not_classify_a_transient_500_as_permanent() { let error = http_error_with_status(500).await;
assert!( !super::super::is_permanent_failure(&error), "a server-side 500 is transient and must keep retrying on backoff, not park" ); }
#[test] fn it_does_not_classify_a_transient_failure_as_permanent() { assert!(!super::super::is_permanent_failure(&Error::NotReady)); assert!(!super::super::is_permanent_failure(&Error::Internal( "write logo failed".to_owned() ))); } }
mod run { use super::*;
#[tokio::test] async fn it_backs_off_when_esi_rate_limits() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/9/")) .respond_with(ResponseTemplate::new(429).insert_header("Retry-After", "30")) .mount(&server) .await; let (handle, mut events, db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(9));
let backing_off = wait_for(&mut events, |event| matches!(event, Event::BackingOff { .. })).await; assert!(matches!( backing_off, Event::BackingOff { retry_secs: 30, .. } )); assert!(character::get(&db, 9).await.unwrap().is_none()); }
#[tokio::test] async fn it_defers_a_gather_job_whose_parent_row_is_absent_until_the_profile_commits() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/5000/wallet/journal/")) .respond_with(ResponseTemplate::new(500)) .expect(0) .mount(&server) .await; Mock::given(method("GET")) .and(path("/characters/5000/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let db = store::open_test().await.unwrap(); infra::upsert( &db, 5000, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_WALLET), ) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db, esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
let wallet = JobKey::new(JobKind::CharacterWallet, Subject::Character(5000)); let started = tokio::time::timeout(Duration::from_millis(300), async { while let Some(event) = events.recv().await { if matches!(event, Event::Started { key } if key == wallet) { return true; } } false }) .await; assert!( started.is_err() || !started.unwrap(), "a parentless subject's wallet gather must stay deferred — it must not dispatch before the profile commits" ); }
#[tokio::test] async fn it_discovers_credentialed_subjects_without_an_enroll_command() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); infra::upsert(&db, 55, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db, esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
let started = wait_for( &mut events, |event| matches!(event, Event::Started { key } if key.subject == Subject::Character(55)), ) .await; assert!(matches!(started, Event::Started { key } if key.subject == Subject::Character(55))); }
#[tokio::test] async fn it_enrolls_a_newly_credentialed_subject_on_a_discover_command() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/6161/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let db = store::open_test().await.unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
infra::upsert(&db, 6161, OwnerType::Character, "tok", "rt", 4_102_444_800, None, None) .await .unwrap(); handle.discover();
let started = wait_for( &mut events, |event| matches!(event, Event::Started { key } if key.subject == Subject::Character(6161)), ) .await; assert!( matches!(started, Event::Started { key } if key.subject == Subject::Character(6161)), "a Discover command must enroll a credential added after startup without waiting for the heartbeat" ); }
#[tokio::test] async fn it_globally_pauses_dispatch_on_an_esi_error_limit() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/10/")) .respond_with(ResponseTemplate::new(420).insert_header("X-ESI-Error-Limit-Reset", "30")) .mount(&server) .await; let (handle, mut events, _db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(10));
let backing_off = wait_for(&mut events, |event| { matches!( event, Event::BackingOff { key, retry_secs: 30 } if *key == JobKey::new(JobKind::CharacterProfile, Subject::Character(10)) ) }) .await; let Event::BackingOff { key: limited, .. } = backing_off else { unreachable!("the predicate only matches a BackingOff event"); };
let redispatched = tokio::time::timeout(Duration::from_millis(300), async { while let Some(event) = events.recv().await { if matches!(event, Event::Started { key } if key == limited) { return; } } }) .await; assert!( redispatched.is_err(), "the error-limited job re-dispatched during the governor window" ); }
#[tokio::test] async fn it_records_a_failure_outcome_without_a_success_timestamp() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/7/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let (handle, mut events, db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(7));
wait_for( &mut events, |event| { matches!(event, Event::Scheduled { key, .. } if key == &JobKey::new(JobKind::CharacterProfile, Subject::Character(7))) }, ) .await;
let row = sync_ledger::get(&db, OwnerType::Character, 7, "CharacterProfile") .await .unwrap() .expect("a failed job still records a ledger row"); assert_eq!(row.outcome(), "failed"); assert!(row.last_reason().is_some(), "a failure records why it failed"); assert!( row.last_success_at().is_none(), "a failure must not stamp last_success_at" ); }
#[tokio::test] async fn it_records_a_synced_ledger_row_when_a_job_finishes() { let server = MockServer::start().await; mount_json( &server, "/markets/prices/", serde_json::json!([{ "adjusted_price": 1.0, "average_price": 1.0, "type_id": 34 }]), ) .await; let (_handle, mut events, db, _images) = spawn_engine(server.uri()).await;
wait_for( &mut events, |event| matches!(event, Event::Scheduled { key, .. } if key.kind == JobKind::MarketPrices), ) .await;
let row = sync_ledger::get(&db, OwnerType::Character, 0, "MarketPrices") .await .unwrap() .expect("a finished job records a ledger row"); assert_eq!(row.outcome(), "synced"); assert!(!row.last_attempt_at().is_empty()); assert!( row.last_success_at().is_some(), "a synced terminal stamps last_success_at" ); assert!( row.next_eligible_at().is_some(), "the ledger carries the scheduled next-eligible time" ); }
#[tokio::test] async fn it_reruns_a_subject_on_run_now() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/30/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let (handle, mut events, _db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(30)); wait_for(&mut events, |event| matches!(event, Event::Failed { .. })).await;
handle.run_now(Subject::Character(30));
let rerun = wait_for( &mut events, |event| matches!(event, Event::Started { key } if key.subject == Subject::Character(30)), ) .await; assert!(matches!(rerun, Event::Started { key } if key.subject == Subject::Character(30))); }
#[tokio::test] async fn it_reschedules_an_empty_outcome_at_its_normal_interval() { let server = MockServer::start().await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let (_handle, mut events, _db, _images) = spawn_engine(server.uri()).await;
let scheduled = wait_for( &mut events, |event| matches!(event, Event::Scheduled { key, .. } if key.kind == JobKind::MarketPrices), ) .await; let Event::Scheduled { next_in_secs, .. } = scheduled else { unreachable!() };
assert!( JobKind::MarketPrices.interval().as_secs() > PENDING_RETRY.as_secs(), "the pending retry is meaningfully shorter than the job's normal interval" ); assert!( next_in_secs > PENDING_RETRY.as_secs(), "a benign empty result settles to the normal interval, not the pending window, got {next_in_secs}s" ); }
#[tokio::test] async fn it_runs_a_gather_for_an_existing_subject_without_waiting_for_a_fresh_profile() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/4478/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; mount_json(&server, "/characters/4478/wallet/journal/", serde_json::json!([])).await; mount_json(&server, "/characters/4478/wallet/transactions/", serde_json::json!([])).await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); seed_character(&db, 4478).await; infra::upsert( &db, 4478, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_WALLET), ) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db, esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
let wallet = JobKey::new(JobKind::CharacterWallet, Subject::Character(4478)); let mut profile_succeeded = false; let ran = tokio::time::timeout(Duration::from_secs(5), async { loop { match events.recv().await { Some(Event::Started { key, }) if key == wallet => return true, Some(Event::Finished { key, .. }) if key.kind == JobKind::CharacterProfile && key.subject == Subject::Character(4478) => { profile_succeeded = true; } Some(_) => continue, None => return false, } } }) .await .expect("timed out waiting for the wallet gather to run");
assert!( ran, "an existing subject's wallet gather must run on its own schedule, not wait for a profile trigger" ); assert!( !profile_succeeded, "the profile job fails for this subject, so the gather ran without any fresh profile success" ); }
#[tokio::test] async fn it_runs_a_new_subjects_gather_only_after_the_profile_commits_its_parent_row() { let server = MockServer::start().await; mount_character_profile(&server, 101).await; mount_json(&server, "/characters/101/wallet/journal/", serde_json::json!([])).await; mount_json(&server, "/characters/101/wallet/transactions/", serde_json::json!([])).await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); infra::upsert( &db, 101, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_WALLET), ) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), ); seed_ship_type(&db).await;
handle.enroll(Subject::Character(101));
let wallet = JobKey::new(JobKind::CharacterWallet, Subject::Character(101)); let profile = JobKey::new(JobKind::CharacterProfile, Subject::Character(101)); let mut profile_committed = false; let ordered = tokio::time::timeout(Duration::from_secs(5), async { loop { match events.recv().await { Some(Event::Started { key, }) if key == wallet => return profile_committed, Some(Event::Finished { key, .. }) if key == profile => profile_committed = true, Some(_) => continue, None => return false, } } }) .await .expect("timed out waiting for the wallet gather to start");
assert!( ordered, "the wallet gather for a brand-new subject must not start until CharacterProfile has committed the parent row" ); }
#[tokio::test] async fn it_runs_an_enrolled_job_and_reports_failure_without_persisting() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/7/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let (handle, mut events, db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(7));
let started = wait_for( &mut events, |event| matches!(event, Event::Started { key } if key.subject == Subject::Character(7)), ) .await; assert!(matches!(started, Event::Started { key } if key.subject == Subject::Character(7))); wait_for( &mut events, |event| matches!(event, Event::Failed { key, .. } if key.subject == Subject::Character(7)), ) .await; assert!(character::get(&db, 7).await.unwrap().is_none()); }
#[tokio::test] async fn it_runs_character_skills_and_lands_skills_when_wallet_off_and_skill_monitoring_on() { let server = MockServer::start().await; mount_skill_picture(&server, 4242).await; let db = store::open_test().await.unwrap(); seed_character(&db, 4242).await; infra::upsert( &db, 4242, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_SKILLS), ) .await .unwrap(); let flags: FeatureFlags = toml::from_str("wallet = false\nskill_monitoring = true").unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db.clone(), esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), flags, );
let terminal = wait_for(&mut events, |event| { matches!( event, Event::Finished { key, .. } | Event::Failed { key, .. } if key.kind == JobKind::CharacterSkills ) }) .await;
assert!( matches!(terminal, Event::Finished { .. }), "expected CharacterSkills to finish, got {terminal:?}" ); let skills = crate::store::repo::character::skills(&db, 4242, chrono::Utc::now()) .await .unwrap(); assert_eq!(skills.len(), 1, "the skill sheet should be persisted"); let queue = crate::store::repo::character::skillqueue(&db, 4242).await.unwrap(); assert_eq!(queue.len(), 1, "the skill queue should be persisted"); }
#[tokio::test] async fn it_runs_jobs_for_multiple_characters_concurrently() { let server = MockServer::start().await; for id in [41_i64, 42] { Mock::given(method("GET")) .and(path(format!("/characters/{id}/"))) .respond_with(ResponseTemplate::new(500).set_delay(Duration::from_millis(300))) .mount(&server) .await; } let (handle, mut events, _db, _images) = spawn_engine(server.uri()).await;
handle.enroll(Subject::Character(41)); handle.enroll(Subject::Character(42));
let mut profiles_started = HashSet::new(); loop { let event = tokio::time::timeout(Duration::from_secs(5), events.recv()) .await .expect("timed out waiting for both profile jobs to start") .expect("event channel closed"); match event { Event::Started { key, } if key.kind == JobKind::CharacterProfile => { profiles_started.insert(key.subject); if profiles_started.len() == 2 { break; } } Event::Finished { key, .. } | Event::Failed { key, .. } if key.kind == JobKind::CharacterProfile => { panic!("a profile job reached a terminal before both started — jobs ran serially: {key:?}"); } _ => {} } }
assert_eq!(profiles_started.len(), 2); }
#[tokio::test] async fn it_schedules_only_jobs_whose_scopes_a_partial_grant_covers() { let server = MockServer::start().await; mount_json(&server, "/characters/4477/wallet/journal/", serde_json::json!([])).await; mount_json(&server, "/characters/4477/wallet/transactions/", serde_json::json!([])).await; mount_json(&server, "/markets/prices/", serde_json::json!([])).await; let db = store::open_test().await.unwrap(); seed_character(&db, 4477).await; infra::upsert( &db, 4477, OwnerType::Character, "tok", "rt", 4_102_444_800, None, Some(esi::scopes::CHARACTER_WALLET), ) .await .unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db, esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
let mut kinds = Vec::new(); let deadline = tokio::time::Instant::now() + Duration::from_secs(2); while let Ok(Some(event)) = tokio::time::timeout_at(deadline, events.recv()).await { if let Event::Started { key, } | Event::Failed { key, .. } = event && key.subject == Subject::Character(4477) { kinds.push(key.kind); } }
assert!( kinds.contains(&JobKind::CharacterWallet), "the granted wallet scope schedules CharacterWallet, got {kinds:?}" ); assert!( !kinds.contains(&JobKind::CharacterMarketOrders), "the ungranted market-orders scope must never schedule (no permanent 401 loop), got {kinds:?}" ); }
#[tokio::test] async fn it_schedules_only_public_jobs_for_a_non_owned_character() { let server = MockServer::start().await; Mock::given(method("GET")) .and(path("/characters/3004069/")) .respond_with(ResponseTemplate::new(500)) .mount(&server) .await; let db = store::open_test().await.unwrap(); seed_character(&db, 3004069).await; let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let (_handle, mut events) = spawn( db.clone(), db, esi, sso, image, images::Store::new(images_dir.path().to_path_buf()), FeatureFlags::default(), );
let mut kinds = Vec::new(); let deadline = tokio::time::Instant::now() + Duration::from_secs(2); while let Ok(Some(event)) = tokio::time::timeout_at(deadline, events.recv()).await { if let Event::Started { key, } | Event::Failed { key, .. } = event && key.subject == Subject::Character(3004069) { kinds.push(key.kind); } }
assert!( kinds.contains(&JobKind::CharacterProfile), "the non-owned character's public profile job should still run, got {kinds:?}" ); assert!( !kinds.contains(&JobKind::CharacterTelemetry) && !kinds.contains(&JobKind::CharacterWallet), "no privileged job should ever be scheduled for a non-owned character, got {kinds:?}" ); }
#[tokio::test] async fn it_surfaces_a_privileged_job_without_a_credential_as_needs_reauth() { let server = MockServer::start().await; let db = store::open_test().await.unwrap(); let http = http::Client::builder(http::Cache::new(db.clone())).build(); let esi = Arc::new(esi::Client::with_base_url(http.clone(), server.uri())); let image = Arc::new(eve_image::Client::with_base_url(http.clone(), server.uri())); let sso = Arc::new(eve_sso::Client::new(http, "test-client")); let images_dir = tempfile::tempdir().unwrap(); let store = images::Store::new(images_dir.path().to_path_buf()); let key = JobKey::new(JobKind::CharacterWallet, Subject::Character(7));
let result = run_job(&db, &esi, &sso, &image, &store, key).await;
assert!( matches!(result, Err(Error::Auth(_))), "a privileged job with no usable credential must surface as needs-reauth, not a clean Finished" ); }
#[tokio::test] async fn it_syncs_an_enrolled_character_end_to_end() { let server = MockServer::start().await; mount_json( &server, "/characters/100/", serde_json::json!({ "alliance_id": 300, "birthday": "2010-01-01T00:00:00Z", "bloodline_id": 5, "corporation_id": 200, "gender": "male", "name": "Test Pilot", "race_id": 1, }), ) .await; mount_json( &server, "/corporations/200/", serde_json::json!({ "alliance_id": 300, "ceo_id": 100, "creator_id": 100, "member_count": 42, "name": "Test Corp", "tax_rate": 0.1, "ticker": "TST", }), ) .await; mount_json( &server, "/alliances/300/", serde_json::json!({ "creator_corporation_id": 200, "creator_id": 100, "date_founded": "2005-01-01T00:00:00Z", "name": "Test Alliance", "ticker": "TSTA", }), ) .await; mount_json( &server, "/universe/races/", serde_json::json!([ { "alliance_id": 300, "description": "The Caldari.", "name": "Caldari", "race_id": 1 }, ]), ) .await; mount_json( &server, "/universe/bloodlines/", serde_json::json!([ { "bloodline_id": 5, "charisma": 6, "corporation_id": 200, "description": "The Civire.", "intelligence": 7, "memory": 5, "name": "Civire", "perception": 5, "race_id": 1, "ship_type_id": 601, "willpower": 5 }, ]), ) .await; Mock::given(method("GET")) .and(path("/characters/100/portrait")) .respond_with(ResponseTemplate::new(200).set_body_raw(vec![1u8, 2, 3], "image/jpeg")) .mount(&server) .await; let (handle, mut events, db, images_dir) = spawn_engine(server.uri()).await; seed_ship_type(&db).await;
handle.enroll(Subject::Character(100));
let terminal = wait_for(&mut events, |event| { matches!( event, Event::Finished { key, .. } | Event::Failed { key, .. } if key.kind == JobKind::CharacterProfile ) }) .await; assert!( matches!(terminal, Event::Finished { .. }), "expected Finished, got {terminal:?}" ); let character = character::get(&db, 100).await.unwrap().expect("character persisted"); assert_eq!(character.name(), "Test Pilot"); assert!( crate::store::repo::org::get_corporation(&db, 200) .await .unwrap() .is_some() ); assert!( images_dir.path().join("characters").join("100.jpg").exists(), "the portrait should be written to the image store as part of the dataset" ); } }
mod supervise_loop { use std::sync::{ Arc, atomic::{AtomicU32, Ordering}, };
use super::*;
#[tokio::test(start_paused = true)] async fn it_gives_up_and_parks_after_the_circuit_breaker_trips() { let (events_tx, mut events_rx) = mpsc::channel(EVENT_BUFFER); let (_command_tx, command_rx) = mpsc::unbounded_channel(); let (restart_tx, restart_rx) = mpsc::unbounded_channel(); let runs = Arc::new(AtomicU32::new(0)); let runs_in_loop = Arc::clone(&runs);
let loop_handle = tokio::spawn(async move { super::super::supervise_loop(&events_tx, command_rx, restart_rx, move |_command_rx| { let runs = Arc::clone(&runs_in_loop); Box::pin(async move { runs.fetch_add(1, Ordering::SeqCst); panic!("engine hot-loops"); }) }) .await; });
let gave_up = tokio::time::timeout(Duration::from_secs(30), async { loop { match events_rx.recv().await { Some(Event::GaveUp { reason, }) => return reason, Some(_) => continue, None => panic!("the event channel closed before the supervisor gave up"), } } }) .await .expect("the breaker should trip and emit a give-up signal rather than hot-loop forever");
assert_eq!( runs.load(Ordering::SeqCst), RESPAWN_BREAKER_THRESHOLD, "the engine runs exactly threshold times: each death restarts until the threshold-th trips" ); assert!( gave_up.contains("gave up"), "the give-up signal carries a human-meaningful reason, got {gave_up:?}" );
drop(restart_tx); tokio::time::timeout(Duration::from_secs(5), loop_handle) .await .expect("the supervisor exits once its restart channel closes") .expect("the supervisor task should not panic"); }
#[tokio::test(start_paused = true)] async fn it_respawns_the_engine_after_an_unexpected_death_and_keeps_the_command_path() { let (events_tx, mut events_rx) = mpsc::channel(EVENT_BUFFER); let (command_tx, command_rx) = mpsc::unbounded_channel(); let (_restart_tx, restart_rx) = mpsc::unbounded_channel(); let runs = Arc::new(AtomicU32::new(0)); let runs_in_loop = Arc::clone(&runs); let reached_new_engine = Arc::new(AtomicU32::new(0)); let reached_in_loop = Arc::clone(&reached_new_engine);
command_tx.send(Command::Drain).unwrap();
let loop_handle = tokio::spawn(async move { super::super::supervise_loop(&events_tx, command_rx, restart_rx, move |command_rx| { let runs = Arc::clone(&runs_in_loop); let reached = Arc::clone(&reached_in_loop); Box::pin(async move { if runs.fetch_add(1, Ordering::SeqCst) == 0 { panic!("simulated engine panic"); } if matches!(command_rx.recv().await, Some(Command::Drain)) { reached.fetch_add(1, Ordering::SeqCst); } ExitReason::Shutdown }) }) .await; });
let restarted = tokio::time::timeout(Duration::from_secs(5), events_rx.recv()) .await .expect("the supervisor should emit a restart signal") .expect("the event channel should stay open across a respawn"); tokio::time::timeout(Duration::from_secs(5), loop_handle) .await .expect("the loop should terminate once the second attempt shuts down") .expect("the supervisor task should not panic");
assert!( matches!( restarted, Event::Restarted { attempt: 1 } ), "the first unexpected death emits a Restarted signal for attempt 1, got {restarted:?}" ); assert_eq!( runs.load(Ordering::SeqCst), 2, "the engine ran twice: the panicking attempt and its respawn" ); assert_eq!( reached_new_engine.load(Ordering::SeqCst), 1, "the command queued before the death reached the respawned engine over the surviving receiver" ); }
#[tokio::test(start_paused = true)] async fn it_resumes_from_a_parked_give_up_when_a_manual_restart_arrives() { let (events_tx, mut events_rx) = mpsc::channel(EVENT_BUFFER); let (_command_tx, command_rx) = mpsc::unbounded_channel(); let (restart_tx, restart_rx) = mpsc::unbounded_channel(); let runs = Arc::new(AtomicU32::new(0)); let runs_in_loop = Arc::clone(&runs);
let loop_handle = tokio::spawn(async move { super::super::supervise_loop(&events_tx, command_rx, restart_rx, move |_command_rx| { let runs = Arc::clone(&runs_in_loop); Box::pin(async move { if runs.fetch_add(1, Ordering::SeqCst) < RESPAWN_BREAKER_THRESHOLD { panic!("engine hot-loops"); } ExitReason::Shutdown }) }) .await; });
tokio::time::timeout(Duration::from_secs(30), async { loop { match events_rx.recv().await { Some(Event::GaveUp { .. }) => return, Some(_) => continue, None => panic!("the event channel closed before the supervisor gave up"), } } }) .await .expect("the supervisor should give up and park");
restart_tx.send(()).unwrap(); tokio::time::timeout(Duration::from_secs(5), loop_handle) .await .expect("the manual restart should unpark the supervisor and let the fresh engine settle") .expect("the supervisor task should not panic");
assert_eq!( runs.load(Ordering::SeqCst), RESPAWN_BREAKER_THRESHOLD + 1, "the manual restart resets the breaker and respawns the engine one more time past the trip" ); }
#[tokio::test(start_paused = true)] async fn it_short_circuits_backoff_when_a_manual_restart_arrives() { let (events_tx, mut events_rx) = mpsc::channel(EVENT_BUFFER); let (_command_tx, command_rx) = mpsc::unbounded_channel(); let (restart_tx, restart_rx) = mpsc::unbounded_channel(); let runs = Arc::new(AtomicU32::new(0)); let runs_in_loop = Arc::clone(&runs);
restart_tx.send(()).unwrap();
let loop_handle = tokio::spawn(async move { super::super::supervise_loop(&events_tx, command_rx, restart_rx, move |_command_rx| { let runs = Arc::clone(&runs_in_loop); Box::pin(async move { if runs.fetch_add(1, Ordering::SeqCst) == 0 { panic!("first attempt dies"); } ExitReason::Shutdown }) }) .await; });
tokio::time::timeout(Duration::from_secs(5), async { loop { match events_rx.recv().await { Some(Event::Restarted { .. }) => return, Some(_) => continue, None => panic!("the event channel closed before a restart was emitted"), } } }) .await .expect("a queued manual restart short-circuits the backoff and emits a restart promptly"); tokio::time::timeout(Duration::from_secs(5), loop_handle) .await .expect("the loop should terminate once the respawn shuts down") .expect("the supervisor task should not panic");
assert_eq!( runs.load(Ordering::SeqCst), 2, "the engine ran twice: the dying attempt and the manually-resumed respawn" ); }
#[tokio::test(start_paused = true)] async fn it_stops_without_respawn_on_a_deliberate_shutdown() { let (events_tx, _events_rx) = mpsc::channel(EVENT_BUFFER); let (_command_tx, command_rx) = mpsc::unbounded_channel(); let (_restart_tx, restart_rx) = mpsc::unbounded_channel(); let runs = Arc::new(AtomicU32::new(0)); let runs_in_loop = Arc::clone(&runs);
let loop_handle = tokio::spawn(async move { super::super::supervise_loop(&events_tx, command_rx, restart_rx, move |_command_rx| { let runs = Arc::clone(&runs_in_loop); Box::pin(async move { runs.fetch_add(1, Ordering::SeqCst); ExitReason::Shutdown }) }) .await; });
tokio::time::timeout(Duration::from_secs(5), loop_handle) .await .expect("a deliberate shutdown ends the loop immediately") .expect("the supervisor task should not panic");
assert_eq!( runs.load(Ordering::SeqCst), 1, "a deliberate Shutdown never respawns; the engine runs exactly once" ); } }}