//! Durable translation-task state for official TextUnit handoff. //! //! The official TextUnit queue remains an immutable release artifact. This //! module stores the mutable worker state separately so a provider worker can //! retry or complete a task without rewriting the published release. use crate::official_textunit_queue::{ textunit_task_matches, OfficialTextUnitTask, OfficialTextUnitTaskQuery, OfficialTextUnitTaskQueue, OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE, }; use crate::path_security::{ ensure_path_within_root, ensure_safe_file_target, read_file_no_symlink, write_file_atomic, STATE_FILE_MODE, }; use bat_core::Result; use serde::{Deserialize, Serialize}; use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions}; use sqlx::{QueryBuilder, Sqlite, SqlitePool}; use std::collections::{BTreeMap, BTreeSet}; use std::path::Path; use std::str::FromStr; use std::time::{Duration, SystemTime, UNIX_EPOCH}; /// SQLite schema version for durable translation task state. pub const TRANSLATION_TASK_SCHEMA_VERSION: u32 = 2; const TRANSLATION_TASK_SCHEMA_COMPONENT: &str = "translation_tasks"; /// SQLite file name stored under a published official release root. pub const TRANSLATION_TASK_REPOSITORY_FILE: &str = "translation-tasks.sqlite"; /// Translation handoff schema version. pub const TRANSLATION_HANDOFF_SCHEMA_VERSION: u32 = 1; /// Versioned handoff file stored under a published official release root. pub const TRANSLATION_HANDOFF_FILE: &str = "translation-handoff.json"; /// Lifecycle status for one translation job. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TranslationJobStatus { /// No provider work has started. Queued, /// At least one provider run is active. Translating, /// Provider output is waiting for human review. Review, /// All source units have translated output ready for patching. Ready, /// A localized release was published. Published, /// One or more provider runs failed. Failed, } impl TranslationJobStatus { /// Returns the stable handoff label. pub fn as_str(self) -> &'static str { match self { Self::Queued => "queued", Self::Translating => "translating", Self::Review => "review", Self::Ready => "ready", Self::Published => "published", Self::Failed => "failed", } } } /// Lifecycle status for one resource/TextUnit task in a translation job. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TranslationUnitStatus { /// Waiting for a provider. Queued, /// A provider is processing the unit. Translating, /// Provider output exists but is not yet reviewed. Translated, /// Human review accepted the output. Reviewed, /// A localized patch was generated. Patched, /// The localized release contains this unit. Published, /// Provider processing failed. Failed, /// The source task was skipped by parsing or policy. Skipped, } /// Status of one provider execution recorded in a translation handoff. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum ProviderRunStatus { /// Waiting to be scheduled. Queued, /// Provider work is active. Running, /// Provider work completed. Succeeded, /// Provider work failed. Failed, /// Provider work was cancelled. Cancelled, } /// Durable job-level translation state. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TranslationJob { /// Stable job ID derived from the official release. pub job_id: String, /// Official release consumed by this job. pub official_release_id: String, /// Previous official release, when known. pub previous_release_id: Option, /// Localized release produced by a later patch/publish stage. pub localized_release_id: Option, /// Current job lifecycle status. pub status: TranslationJobStatus, /// Number of source units represented by the handoff. pub unit_count: usize, /// Queue generation time. pub created_unix_seconds: u64, /// Last handoff state generation time. pub updated_unix_seconds: u64, /// Job-level diagnostic, if failed. pub failure_reason: Option, } /// One translation unit linked back to an immutable official task. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TranslationUnit { /// Stable unit ID. pub unit_id: String, /// Owning translation job. pub job_id: String, /// Immutable official TextUnit task ID. pub task_id: String, /// Official release containing the source. pub official_release_id: String, /// Resource destination under the official release. pub destination: String, /// ZIP/archive entry, when the source was nested. pub archive_entry: Option, /// Number of extracted TextUnits represented by this task. pub text_unit_count: usize, /// Extracted TextUnit format labels. pub text_unit_formats: Vec, /// Current translation lifecycle state. pub status: TranslationUnitStatus, /// Provider failure diagnostic, when present. pub failure_reason: Option, /// TextUnit-level provider results associated with this resource task. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub translation_results: Vec, } /// One provider result linked to an immutable TextUnit. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TranslationTaskUnitResult { /// Stable TextUnit ID from `official-textunit-index.json`. pub unit_id: String, /// Source text submitted to the provider. pub source_text: String, /// Provider-produced or human-supplied translation. pub translated_text: String, /// Result source kind. #[serde(default)] pub source_kind: TranslationTaskResultSourceKind, /// Trusted Translation Memory record used for this result, when applicable. #[serde(default, skip_serializing_if = "Option::is_none")] pub translation_memory_record_id: Option, /// Provider identifier. pub provider: String, /// Provider run that produced this result. pub provider_run_id: String, /// Result persistence time. pub translated_unix_seconds: u64, /// Deterministic Glossary QA result, when a project Glossary was available. #[serde(default, skip_serializing_if = "Option::is_none")] pub glossary_qa: Option, /// Explicit human confirmation for a blocking Glossary deviation. #[serde(default, skip_serializing_if = "Option::is_none")] pub glossary_override: Option, } /// Source of one persisted TextUnit translation result. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TranslationTaskResultSourceKind { /// Result returned by the configured provider. #[default] Provider, /// Result submitted through the manual task update interface. Manual, /// Result reused from a trusted Translation Memory entry. TranslationMemory, } impl TranslationTaskResultSourceKind { /// Returns the stable JSON label. pub const fn as_str(self) -> &'static str { match self { Self::Provider => "provider", Self::Manual => "manual", Self::TranslationMemory => "translation_memory", } } } /// One provider execution associated with one or more translation units. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct ProviderRun { /// Stable provider run ID. pub provider_run_id: String, /// Owning translation job. pub job_id: String, /// Provider identifier, reserved for future plugin implementations. pub provider: String, /// Current provider run status. pub status: ProviderRunStatus, /// Unit IDs submitted to this run. pub unit_ids: Vec, /// Number of attempts represented by this run. pub attempt_count: u32, /// Provider failure diagnostic, when present. pub failure_reason: Option, } /// Read-only handoff view consumed by translation workers and query clients. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct TranslationHandoff { /// Handoff schema version. pub handoff_version: u32, /// Official queue generation time. pub generated_unix_seconds: u64, /// Stable source queue file name. pub source_queue_file: String, /// Job-level state. pub job: TranslationJob, /// Resource/TextUnit task states. pub units: Vec, /// Provider run states. pub provider_runs: Vec, } /// Builds a translation handoff from immutable queue data and mutable worker /// state. No provider or network call is made. pub fn build_translation_handoff( queue: &OfficialTextUnitTaskQueue, tasks: &[PersistedTranslationTask], ) -> TranslationHandoff { let job_id = format!("official-release:{}", queue.official_release_id); let persisted = tasks .iter() .map(|task| (task.task.task_id.as_str(), task)) .collect::>(); let mut units = Vec::with_capacity(queue.tasks.len()); let mut provider_runs = BTreeMap::::new(); for task in &queue.tasks { let state = persisted .get(task.task_id.as_str()) .copied() .map(|task| { ( task.task_status, task.failure_reason.clone(), task.attempt_count, task.provider_run_id.clone(), task.provider.clone(), task.translation_results.clone(), ) }) .unwrap_or_else(|| { ( initial_task_status(task), if task.status == crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedParseFailed { task.reason.clone() } else { None }, 0, None, None, Vec::new(), ) }); let unit_status = if state.5.is_empty() { match state.0 { TranslationTaskStatus::Queued => TranslationUnitStatus::Queued, TranslationTaskStatus::Running => TranslationUnitStatus::Translating, TranslationTaskStatus::Failed => TranslationUnitStatus::Failed, TranslationTaskStatus::Completed => TranslationUnitStatus::Translated, TranslationTaskStatus::Skipped => TranslationUnitStatus::Skipped, } } else { // A task can retain successful TM hits while the remaining provider // units are failed or waiting for retry. TranslationUnitStatus::Translated }; let unit = TranslationUnit { unit_id: task.task_id.clone(), job_id: job_id.clone(), task_id: task.task_id.clone(), official_release_id: task.official_release_id.clone(), destination: task.destination.clone(), archive_entry: task.archive_entry.clone(), text_unit_count: task.text_unit_count, text_unit_formats: task.text_unit_formats.clone(), status: unit_status, failure_reason: state.1.clone(), translation_results: state.5.clone(), }; if let Some(provider_run_id) = state.3 { let run = provider_runs .entry(provider_run_id.clone()) .or_insert_with(|| ProviderRun { provider_run_id, job_id: job_id.clone(), provider: state.4.unwrap_or_else(|| "worker".to_string()), status: ProviderRunStatus::Queued, unit_ids: Vec::new(), attempt_count: 0, failure_reason: None, }); run.unit_ids.push(unit.unit_id.clone()); run.attempt_count = run.attempt_count.max(state.2); let candidate_status = match state.0 { TranslationTaskStatus::Running => ProviderRunStatus::Running, TranslationTaskStatus::Failed => ProviderRunStatus::Failed, TranslationTaskStatus::Completed => ProviderRunStatus::Succeeded, TranslationTaskStatus::Queued | TranslationTaskStatus::Skipped => run.status, }; if provider_run_status_rank(candidate_status) > provider_run_status_rank(run.status) { run.status = candidate_status; } if state.1.is_some() { run.failure_reason = state.1.clone(); } } units.push(unit); } let job_failure_reason = units.iter().find_map(|unit| unit.failure_reason.clone()); let has_running = units .iter() .any(|unit| unit.status == TranslationUnitStatus::Translating); let candidate_units = units .iter() .filter(|unit| unit.status != TranslationUnitStatus::Skipped) .collect::>(); let all_translated = !candidate_units.is_empty() && candidate_units .iter() .all(|unit| unit.status == TranslationUnitStatus::Translated); let job_status = if job_failure_reason.is_some() { TranslationJobStatus::Failed } else if has_running { TranslationJobStatus::Translating } else if all_translated { TranslationJobStatus::Ready } else { TranslationJobStatus::Queued }; TranslationHandoff { handoff_version: TRANSLATION_HANDOFF_SCHEMA_VERSION, generated_unix_seconds: queue.generated_unix_seconds, source_queue_file: OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE.to_string(), job: TranslationJob { job_id, official_release_id: queue.official_release_id.clone(), previous_release_id: queue.previous_release_id.clone(), localized_release_id: None, status: job_status, unit_count: units.len(), created_unix_seconds: queue.generated_unix_seconds, updated_unix_seconds: unix_seconds_now(), failure_reason: job_failure_reason, }, units, provider_runs: provider_runs.into_values().collect(), } } /// Writes a reviewed handoff view under one release root. pub fn write_translation_handoff_at( resource_root: &Path, handoff: &TranslationHandoff, ) -> std::result::Result<(), String> { let path = resource_root.join(TRANSLATION_HANDOFF_FILE); ensure_path_within_root(resource_root, &path)?; ensure_safe_file_target(resource_root, &path, "翻译 handoff")?; let bytes = serde_json::to_vec_pretty(handoff) .map_err(|error| format!("序列化翻译 handoff 失败 {}：{error}", path.display()))?; write_file_atomic(&path, &bytes, STATE_FILE_MODE, "翻译 handoff") } /// Reads the current translation handoff view from one release root. pub fn read_translation_handoff_at( resource_root: &Path, ) -> std::result::Result, String> { let path = resource_root.join(TRANSLATION_HANDOFF_FILE); let Some(bytes) = read_file_no_symlink(&path, "翻译 handoff")? else { return Ok(None); }; let handoff: TranslationHandoff = serde_json::from_slice(&bytes) .map_err(|error| format!("解析翻译 handoff 失败 {}：{error}", path.display()))?; if handoff.handoff_version != TRANSLATION_HANDOFF_SCHEMA_VERSION { return Err(format!( "不支持的翻译 handoff 版本 {}，文件 {}", handoff.handoff_version, path.display() )); } Ok(Some(handoff)) } /// Mutable state owned by a translation worker. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum TranslationTaskStatus { /// The task is waiting for a provider worker. Queued, /// A provider worker is processing the task. Running, /// The provider worker failed the task. Failed, /// The provider worker completed the task. Completed, /// The task was intentionally excluded from provider processing. Skipped, } impl TranslationTaskStatus { /// Returns the stable RPC and database label. pub fn as_str(self) -> &'static str { match self { Self::Queued => "queued", Self::Running => "running", Self::Failed => "failed", Self::Completed => "completed", Self::Skipped => "skipped", } } /// Parses a stable RPC or database label. pub fn parse(value: &str) -> Option { match value { "queued" => Some(Self::Queued), "running" => Some(Self::Running), "failed" => Some(Self::Failed), "completed" => Some(Self::Completed), "skipped" => Some(Self::Skipped), _ => None, } } } /// One official TextUnit task with mutable provider state. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] pub struct PersistedTranslationTask { /// Immutable task data derived from the published release. #[serde(flatten)] pub task: OfficialTextUnitTask, /// Current worker state. pub task_status: TranslationTaskStatus, /// Provider failure reason, if the worker recorded one. pub failure_reason: Option, /// Number of provider attempts. pub attempt_count: u32, /// First persistence time as Unix seconds. pub created_unix_seconds: u64, /// Last state or metadata update time as Unix seconds. pub updated_unix_seconds: u64, /// Completion time as Unix seconds, if completed. #[serde(skip_serializing_if = "Option::is_none")] pub completed_unix_seconds: Option, /// Provider-side run identifier, if known. #[serde(skip_serializing_if = "Option::is_none")] pub provider_run_id: Option, /// Provider-produced results keyed by TextUnit ID. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub translation_results: Vec, /// Provider identifier used by the latest run. #[serde(skip_serializing_if = "Option::is_none")] pub provider: Option, /// Worker currently holding the lease. #[serde(skip_serializing_if = "Option::is_none")] pub lease_owner: Option, /// Lease expiry as Unix seconds. #[serde(skip_serializing_if = "Option::is_none")] pub lease_expires_unix_seconds: Option, /// Stable failure classification. #[serde(skip_serializing_if = "Option::is_none")] pub failure_class: Option, /// Whether the latest failure can be retried. pub failure_retryable: bool, /// Earliest Unix time at which a failed task may be claimed again. #[serde(skip_serializing_if = "Option::is_none")] pub next_attempt_unix_seconds: Option, } /// Mutable persistence metadata associated with an official TextUnit task. #[derive(Debug, Clone, PartialEq, Eq)] pub struct PersistedTranslationTaskState { /// Current worker state. pub task_status: TranslationTaskStatus, /// Provider failure reason, if the worker recorded one. pub failure_reason: Option, /// Number of provider attempts. pub attempt_count: u32, /// First persistence time as Unix seconds. pub created_unix_seconds: u64, /// Last state or metadata update time as Unix seconds. pub updated_unix_seconds: u64, /// Completion time as Unix seconds, if completed. pub completed_unix_seconds: Option, /// Provider-side run identifier, if known. pub provider_run_id: Option, /// Provider-produced results keyed by TextUnit ID. pub translation_results: Vec, /// Provider identifier used by the latest run. pub provider: Option, /// Worker currently holding the lease. pub lease_owner: Option, /// Lease expiry as Unix seconds. pub lease_expires_unix_seconds: Option, /// Stable failure classification. pub failure_class: Option, /// Whether the latest failure can be retried. pub failure_retryable: bool, /// Earliest Unix time at which a failed task may be claimed again. pub next_attempt_unix_seconds: Option, } /// Provider failure data used to atomically release a task lease. #[derive(Debug, Clone, PartialEq, Eq)] pub struct TranslationTaskFailure { /// Task being failed. pub task_id: String, /// Worker that owns the lease. pub worker_id: String, /// Provider run associated with the lease. pub provider_run_id: String, /// Stable failure classification. pub failure_class: String, /// Redacted diagnostic message. pub failure_reason: String, /// Whether a future attempt may retry the task. pub retryable: bool, /// Earliest retry time, when retryable. pub next_attempt_unix_seconds: Option, } /// Result of synchronizing an immutable release queue into SQLite. #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] pub struct TranslationTaskSyncReport { /// Number of newly inserted task rows. pub inserted_count: usize, /// Number of existing rows whose immutable task data was refreshed. pub refreshed_count: usize, /// Number of existing worker states preserved. pub preserved_state_count: usize, /// Number of stale rows removed because they are absent from the new queue. pub removed_count: usize, } /// SQLite-backed translation task state repository. #[derive(Debug, Clone)] pub struct SqliteTranslationTaskRepository { pool: SqlitePool, } impl SqliteTranslationTaskRepository { /// Opens or creates the database and applies the translation-task schema. pub async fn new(path: impl AsRef) -> Result { Self::open_with(path.as_ref(), true).await } /// Opens an existing database and applies compatible migrations. pub async fn open(path: impl AsRef) -> Result { Self::open_with(path.as_ref(), false).await } async fn open_with(path: &Path, create_if_missing: bool) -> Result { if create_if_missing { if let Some(parent) = path.parent() { tokio::fs::create_dir_all(parent).await?; } } let options = SqliteConnectOptions::from_str(&format!("sqlite://{}", path.display())) .map_err(|error| bat_core::Error::Other(error.into()))? .create_if_missing(create_if_missing) .journal_mode(SqliteJournalMode::Wal) .busy_timeout(Duration::from_secs(30)); let pool = SqlitePoolOptions::new() .max_connections(1) .connect_with(options) .await .map_err(|error| bat_core::Error::Other(error.into()))?; let repository = Self { pool }; repository.init_schema().await?; Ok(repository) } /// Returns the durable translation-task database path under one release root. pub fn repository_path(resource_root: &Path) -> std::path::PathBuf { resource_root.join(TRANSLATION_TASK_REPOSITORY_FILE) } async fn init_schema(&self) -> Result<()> { sqlx::query( r#" CREATE TABLE IF NOT EXISTS schema_migrations ( component TEXT PRIMARY KEY NOT NULL, version INTEGER NOT NULL CHECK(version >= 1) ) "#, ) .execute(&self.pool) .await .map_err(db_error)?; sqlx::query( r#" CREATE TABLE IF NOT EXISTS translation_tasks ( task_id TEXT PRIMARY KEY NOT NULL, official_release_id TEXT NOT NULL, destination TEXT NOT NULL, archive_entry TEXT, queue_status TEXT NOT NULL, queue_reason TEXT, parse_status TEXT, text_unit_formats_json TEXT NOT NULL DEFAULT '[]', task_json TEXT NOT NULL DEFAULT '{}', worker_status TEXT NOT NULL DEFAULT 'queued', failure_reason TEXT, attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0), created_unix_seconds INTEGER NOT NULL, updated_unix_seconds INTEGER NOT NULL, completed_unix_seconds INTEGER, provider_run_id TEXT, translation_results_json TEXT NOT NULL DEFAULT '[]', provider TEXT, lease_owner TEXT, lease_expires_unix_seconds INTEGER, failure_class TEXT, failure_retryable INTEGER NOT NULL DEFAULT 0, next_attempt_unix_seconds INTEGER ) "#, ) .execute(&self.pool) .await .map_err(db_error)?; // These defaults keep old experimental databases readable while the // schema version table records the migration boundary explicitly. ensure_column(&self.pool, "translation_tasks", "queue_reason", "TEXT").await?; ensure_column(&self.pool, "translation_tasks", "parse_status", "TEXT").await?; ensure_column( &self.pool, "translation_tasks", "text_unit_formats_json", "TEXT NOT NULL DEFAULT '[]'", ) .await?; ensure_column( &self.pool, "translation_tasks", "task_json", "TEXT NOT NULL DEFAULT '{}'", ) .await?; ensure_column( &self.pool, "translation_tasks", "worker_status", "TEXT NOT NULL DEFAULT 'queued'", ) .await?; ensure_column(&self.pool, "translation_tasks", "failure_reason", "TEXT").await?; ensure_column( &self.pool, "translation_tasks", "attempt_count", "INTEGER NOT NULL DEFAULT 0", ) .await?; ensure_column( &self.pool, "translation_tasks", "created_unix_seconds", "INTEGER NOT NULL DEFAULT 0", ) .await?; ensure_column( &self.pool, "translation_tasks", "updated_unix_seconds", "INTEGER NOT NULL DEFAULT 0", ) .await?; ensure_column( &self.pool, "translation_tasks", "completed_unix_seconds", "INTEGER", ) .await?; ensure_column(&self.pool, "translation_tasks", "provider_run_id", "TEXT").await?; ensure_column( &self.pool, "translation_tasks", "translation_results_json", "TEXT NOT NULL DEFAULT '[]'", ) .await?; ensure_column(&self.pool, "translation_tasks", "provider", "TEXT").await?; ensure_column(&self.pool, "translation_tasks", "lease_owner", "TEXT").await?; ensure_column( &self.pool, "translation_tasks", "lease_expires_unix_seconds", "INTEGER", ) .await?; ensure_column(&self.pool, "translation_tasks", "failure_class", "TEXT").await?; ensure_column( &self.pool, "translation_tasks", "failure_retryable", "INTEGER NOT NULL DEFAULT 0", ) .await?; ensure_column( &self.pool, "translation_tasks", "next_attempt_unix_seconds", "INTEGER", ) .await?; let current: Option = sqlx::query_scalar("SELECT version FROM schema_migrations WHERE component = ?1") .bind(TRANSLATION_TASK_SCHEMA_COMPONENT) .fetch_optional(&self.pool) .await .map_err(db_error)?; if current.is_some_and(|version| version > i64::from(TRANSLATION_TASK_SCHEMA_VERSION)) { return Err(bat_core::Error::InvalidArgument(format!( "不支持的翻译任务 schema 版本：{}", current.unwrap_or_default() ))); } sqlx::query( r#" INSERT INTO schema_migrations(component, version) VALUES (?1, ?2) ON CONFLICT(component) DO UPDATE SET version = excluded.version "#, ) .bind(TRANSLATION_TASK_SCHEMA_COMPONENT) .bind(i64::from(TRANSLATION_TASK_SCHEMA_VERSION)) .execute(&self.pool) .await .map_err(db_error)?; Ok(()) } /// Synchronizes one immutable release queue without resetting worker state. pub async fn sync_queue( &self, queue: &OfficialTextUnitTaskQueue, ) -> Result { let mut transaction = self.pool.begin().await.map_err(db_error)?; let mut report = TranslationTaskSyncReport::default(); for task in &queue.tasks { let task_json = serde_json::to_string(task) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; let existing: Option = sqlx::query_as( r#" SELECT task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, completed_unix_seconds, provider_run_id, translation_results_json, provider, lease_owner, lease_expires_unix_seconds, failure_class, failure_retryable, next_attempt_unix_seconds FROM translation_tasks WHERE task_id = ?1 "#, ) .bind(&task.task_id) .fetch_optional(&mut *transaction) .await .map_err(db_error)?; let now = unix_seconds_now_i64(); let initial_status = initial_task_status(task); let initial_failure_reason = (initial_status == TranslationTaskStatus::Skipped) .then(|| task.reason.clone()) .flatten(); let formats = serde_json::to_string(&task.text_unit_formats) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; let parse_status = task.parse_status.map(parse_status_label); if let Some(( previous_task_json, worker_status, failure_reason, attempt_count, created, completed, provider_run_id, translation_results_json, provider, lease_owner, lease_expires, failure_class, failure_retryable, next_attempt, )) = existing { let immutable_unchanged = previous_task_json == task_json; let ( status, failure_reason, attempt_count, created, completed, provider_run_id, translation_results_json, provider, lease_owner, lease_expires, failure_class, failure_retryable, next_attempt, ) = if immutable_unchanged { ( worker_status, failure_reason, attempt_count, created, completed, provider_run_id, translation_results_json, provider, lease_owner, lease_expires, failure_class, failure_retryable, next_attempt, ) } else { report.refreshed_count += 1; ( initial_status.as_str().to_string(), initial_failure_reason.clone(), 0_i64, now, None, None, "[]".to_string(), None, None, None, None, 0_i64, None, ) }; if immutable_unchanged { report.preserved_state_count += 1; } sqlx::query( r#" UPDATE translation_tasks SET official_release_id = ?2, destination = ?3, archive_entry = ?4, queue_status = ?5, queue_reason = ?6, parse_status = ?7, text_unit_formats_json = ?8, task_json = ?9, worker_status = ?10, failure_reason = ?11, attempt_count = ?12, created_unix_seconds = ?13, updated_unix_seconds = ?14, completed_unix_seconds = ?15, provider_run_id = ?16, translation_results_json = ?17, provider = ?18, lease_owner = ?19, lease_expires_unix_seconds = ?20, failure_class = ?21, failure_retryable = ?22, next_attempt_unix_seconds = ?23 WHERE task_id = ?1 "#, ) .bind(&task.task_id) .bind(&task.official_release_id) .bind(&task.destination) .bind(&task.archive_entry) .bind(task.status.as_str()) .bind(&task.reason) .bind(parse_status) .bind(formats) .bind(task_json) .bind(status) .bind(failure_reason) .bind(attempt_count) .bind(created) .bind(now) .bind(completed) .bind(provider_run_id) .bind(translation_results_json) .bind(provider) .bind(lease_owner) .bind(lease_expires) .bind(failure_class) .bind(failure_retryable) .bind(next_attempt) .execute(&mut *transaction) .await .map_err(db_error)?; } else { report.inserted_count += 1; sqlx::query( r#" INSERT INTO translation_tasks ( task_id, official_release_id, destination, archive_entry, queue_status, queue_reason, parse_status, text_unit_formats_json, task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, updated_unix_seconds ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, 0, ?12, ?12) "#, ) .bind(&task.task_id) .bind(&task.official_release_id) .bind(&task.destination) .bind(&task.archive_entry) .bind(task.status.as_str()) .bind(&task.reason) .bind(parse_status) .bind(formats) .bind(task_json) .bind(initial_status.as_str()) .bind(initial_failure_reason) .bind(now) .execute(&mut *transaction) .await .map_err(db_error)?; } } let task_ids = queue .tasks .iter() .map(|task| task.task_id.as_str()) .collect::>(); let removed_count = if task_ids.is_empty() { sqlx::query("DELETE FROM translation_tasks") .execute(&mut *transaction) .await .map_err(db_error)? .rows_affected() as usize } else { let mut query = QueryBuilder::::new("DELETE FROM translation_tasks WHERE task_id NOT IN ("); let mut separated = query.separated(", "); for task_id in task_ids { separated.push_bind(task_id); } separated.push_unseparated(")"); query .build() .execute(&mut *transaction) .await .map_err(db_error)? .rows_affected() as usize }; report.removed_count = removed_count; transaction.commit().await.map_err(db_error)?; Ok(report) } /// Returns all persisted tasks matching the queue and worker filters. pub async fn list( &self, query: &OfficialTextUnitTaskQuery, ) -> Result> { let rows: Vec = sqlx::query_as( r#" SELECT task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, updated_unix_seconds, completed_unix_seconds, provider_run_id, translation_results_json, provider, lease_owner, lease_expires_unix_seconds, failure_class, failure_retryable, next_attempt_unix_seconds FROM translation_tasks ORDER BY task_id "#, ) .fetch_all(&self.pool) .await .map_err(db_error)?; rows.into_iter() .map(PersistedTranslationTask::from_row) .collect::>>() .map(|tasks| { tasks .into_iter() .filter(|task| matches_query(task, query)) .collect() }) } /// Returns the number of persisted tasks matching a query. pub async fn count(&self, query: &OfficialTextUnitTaskQuery) -> Result { Ok(self.list(query).await?.len() as u64) } /// Requeues tasks whose worker lease expired. pub async fn recover_expired_leases(&self, now_unix_seconds: u64) -> Result { let now = i64::try_from(now_unix_seconds).unwrap_or(i64::MAX); let result = sqlx::query( r#" UPDATE translation_tasks SET worker_status = CASE WHEN attempt_count > 0 THEN 'failed' ELSE 'queued' END, failure_reason = 'worker lease expired; task recovered', failure_class = 'lease_expired', failure_retryable = 1, next_attempt_unix_seconds = ?1, updated_unix_seconds = ?1, lease_owner = NULL, lease_expires_unix_seconds = NULL WHERE worker_status = 'running' AND lease_expires_unix_seconds IS NOT NULL AND lease_expires_unix_seconds <= ?1 "#, ) .bind(now) .execute(&self.pool) .await .map_err(db_error)?; Ok(result.rows_affected()) } /// Claims one queued or retryable failed task and assigns an exclusive /// lease to one worker. pub async fn claim_next( &self, worker_id: &str, provider: &str, lease_seconds: u64, max_attempts: u32, ) -> Result> { if worker_id.trim().is_empty() || provider.trim().is_empty() { return Err(bat_core::Error::InvalidArgument( "translation worker_id/provider 不能为空".to_string(), )); } if max_attempts == 0 { return Err(bat_core::Error::InvalidArgument( "translation worker max_attempts 必须大于 0".to_string(), )); } if lease_seconds == 0 { return Err(bat_core::Error::InvalidArgument( "translation worker lease_seconds 必须大于 0".to_string(), )); } let now = unix_seconds_now_i64(); let lease_expires = now.saturating_add(i64::try_from(lease_seconds).unwrap_or(i64::MAX)); let mut transaction = self.pool.begin().await.map_err(db_error)?; let row: Option = sqlx::query_as( r#" SELECT task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, updated_unix_seconds, completed_unix_seconds, provider_run_id, translation_results_json, provider, lease_owner, lease_expires_unix_seconds, failure_class, failure_retryable, next_attempt_unix_seconds FROM translation_tasks WHERE queue_status = 'queued_offline' AND ( worker_status = 'queued' OR ( worker_status = 'failed' AND failure_retryable = 1 AND attempt_count < ?1 AND (next_attempt_unix_seconds IS NULL OR next_attempt_unix_seconds <= ?2) ) ) AND (lease_expires_unix_seconds IS NULL OR lease_expires_unix_seconds <= ?2) ORDER BY task_id LIMIT 1 "#, ) .bind(i64::from(max_attempts)) .bind(now) .fetch_optional(&mut *transaction) .await .map_err(db_error)?; let Some(row) = row else { transaction.commit().await.map_err(db_error)?; return Ok(None); }; let current = PersistedTranslationTask::from_row(row)?; let attempt_count = current.attempt_count.saturating_add(1); let provider_run_id = format!( "{provider}:{}:attempt-{attempt_count}", current.task.task_id ); let result = sqlx::query( r#" UPDATE translation_tasks SET worker_status = 'running', failure_reason = NULL, attempt_count = ?2, updated_unix_seconds = ?3, completed_unix_seconds = NULL, provider_run_id = ?4, provider = ?5, lease_owner = ?6, lease_expires_unix_seconds = ?7, failure_class = NULL, failure_retryable = 0, next_attempt_unix_seconds = NULL WHERE task_id = ?1 AND ( worker_status = 'queued' OR ( worker_status = 'failed' AND failure_retryable = 1 AND attempt_count < ?8 AND (next_attempt_unix_seconds IS NULL OR next_attempt_unix_seconds <= ?3) ) ) AND queue_status = 'queued_offline' AND (lease_expires_unix_seconds IS NULL OR lease_expires_unix_seconds <= ?3) "#, ) .bind(¤t.task.task_id) .bind(i64::from(attempt_count)) .bind(now) .bind(&provider_run_id) .bind(provider) .bind(worker_id) .bind(lease_expires) .bind(i64::from(max_attempts)) .execute(&mut *transaction) .await .map_err(db_error)?; if result.rows_affected() != 1 { transaction.rollback().await.map_err(db_error)?; return Ok(None); } transaction.commit().await.map_err(db_error)?; self.find(¤t.task.task_id).await.map(Some) } /// Stores an idempotent provider result while the worker still owns its lease. pub async fn complete_claim( &self, task_id: &str, worker_id: &str, provider_run_id: &str, provider: &str, translation_results: &[TranslationTaskUnitResult], ) -> Result { let results_json = serde_json::to_string(translation_results) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; let now = unix_seconds_now_i64(); let result = sqlx::query( r#" UPDATE translation_tasks SET worker_status = 'completed', failure_reason = NULL, updated_unix_seconds = ?2, completed_unix_seconds = ?2, provider_run_id = ?3, provider = ?4, translation_results_json = ?5, lease_owner = NULL, lease_expires_unix_seconds = NULL, failure_class = NULL, failure_retryable = 0, next_attempt_unix_seconds = NULL WHERE task_id = ?1 AND worker_status = 'running' AND lease_owner = ?6 AND provider_run_id = ?3 "#, ) .bind(task_id) .bind(now) .bind(provider_run_id) .bind(provider) .bind(results_json) .bind(worker_id) .execute(&self.pool) .await .map_err(db_error)?; if result.rows_affected() != 1 { return Err(bat_core::Error::InvalidArgument(format!( "翻译任务 {} 的 lease 已失效，拒绝写入 provider 结果", task_id ))); } self.find(task_id).await } /// Records one provider failure and releases the worker lease. pub async fn fail_claim( &self, failure: TranslationTaskFailure, ) -> Result { self.fail_claim_with_results(failure, &[]).await } /// Records a provider failure while retaining any already-resolved TextUnit /// results, such as trusted Translation Memory hits. pub async fn fail_claim_with_results( &self, failure: TranslationTaskFailure, translation_results: &[TranslationTaskUnitResult], ) -> Result { let now = unix_seconds_now_i64(); let next_attempt = failure .next_attempt_unix_seconds .map(|value| i64::try_from(value).unwrap_or(i64::MAX)); let translation_results_json = if translation_results.is_empty() { None } else { Some( serde_json::to_string(translation_results) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?, ) }; let result = sqlx::query( r#" UPDATE translation_tasks SET worker_status = 'failed', failure_reason = ?2, updated_unix_seconds = ?3, lease_owner = NULL, lease_expires_unix_seconds = NULL, failure_class = ?4, failure_retryable = ?5, next_attempt_unix_seconds = ?6, translation_results_json = COALESCE(?7, translation_results_json) WHERE task_id = ?1 AND worker_status = 'running' AND lease_owner = ?8 AND provider_run_id = ?9 "#, ) .bind(&failure.task_id) .bind(&failure.failure_reason) .bind(now) .bind(&failure.failure_class) .bind(if failure.retryable { 1_i64 } else { 0_i64 }) .bind(next_attempt) .bind(translation_results_json) .bind(&failure.worker_id) .bind(&failure.provider_run_id) .execute(&self.pool) .await .map_err(db_error)?; if result.rows_affected() != 1 { return Err(bat_core::Error::InvalidArgument(format!( "翻译任务 {} 的 lease 已失效，拒绝写入 provider 失败状态", failure.task_id ))); } self.find(&failure.task_id).await } /// Updates provider state and returns the durable task record. pub async fn update_status( &self, task_id: &str, status: TranslationTaskStatus, failure_reason: Option, provider_run_id: Option, ) -> Result { self.update_status_with_results( task_id, status, failure_reason, provider_run_id, None, None, ) .await } /// Updates provider state and optionally replaces durable TextUnit results. pub async fn update_status_with_results( &self, task_id: &str, status: TranslationTaskStatus, failure_reason: Option, provider_run_id: Option, provider: Option, translation_results: Option<&[TranslationTaskUnitResult]>, ) -> Result { let current = self.find(task_id).await?; let now = unix_seconds_now_i64(); let attempt_count = if status == TranslationTaskStatus::Running && current.task_status != TranslationTaskStatus::Running { current.attempt_count.saturating_add(1) } else { current.attempt_count }; let normalized_reason = failure_reason.filter(|reason| !reason.trim().is_empty()); let provider_run_id = provider_run_id.filter(|value| !value.trim().is_empty()); let provider = provider.filter(|value| !value.trim().is_empty()); let translation_results_json = translation_results .map(serde_json::to_string) .transpose() .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; let completed = (status == TranslationTaskStatus::Completed).then_some(now); sqlx::query( r#" UPDATE translation_tasks SET worker_status = ?2, failure_reason = ?3, attempt_count = ?4, updated_unix_seconds = ?5, completed_unix_seconds = ?6, provider_run_id = COALESCE(?7, provider_run_id), provider = COALESCE(?8, provider), translation_results_json = COALESCE(?9, translation_results_json), lease_owner = NULL, lease_expires_unix_seconds = NULL, failure_class = NULL, failure_retryable = 0, next_attempt_unix_seconds = NULL WHERE task_id = ?1 "#, ) .bind(task_id) .bind(status.as_str()) .bind(normalized_reason) .bind(i64::from(attempt_count)) .bind(now) .bind(completed) .bind(provider_run_id) .bind(provider) .bind(translation_results_json) .execute(&self.pool) .await .map_err(db_error)?; self.find(task_id).await } /// Finds one task by its stable ID. pub async fn find(&self, task_id: &str) -> Result { let row: Option = sqlx::query_as( r#" SELECT task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, updated_unix_seconds, completed_unix_seconds, provider_run_id, translation_results_json, provider, lease_owner, lease_expires_unix_seconds, failure_class, failure_retryable, next_attempt_unix_seconds FROM translation_tasks WHERE task_id = ?1 "#, ) .bind(task_id) .fetch_optional(&self.pool) .await .map_err(db_error)?; row.map(PersistedTranslationTask::from_row) .transpose()? .ok_or_else(|| bat_core::Error::NotFound(task_id.to_string())) } } type ExistingTaskRow = ( String, String, Option, i64, i64, Option, Option, String, Option, Option, Option, Option, i64, Option, ); type TranslationTaskRow = ( String, String, Option, i64, i64, i64, Option, Option, String, Option, Option, Option, Option, i64, Option, ); impl PersistedTranslationTask { /// Builds a persisted translation task from an immutable queue task. pub fn from_task(task: OfficialTextUnitTask, state: PersistedTranslationTaskState) -> Self { Self { task, task_status: state.task_status, failure_reason: state.failure_reason, attempt_count: state.attempt_count, created_unix_seconds: state.created_unix_seconds, updated_unix_seconds: state.updated_unix_seconds, completed_unix_seconds: state.completed_unix_seconds, provider_run_id: state.provider_run_id, translation_results: state.translation_results, provider: state.provider, lease_owner: state.lease_owner, lease_expires_unix_seconds: state.lease_expires_unix_seconds, failure_class: state.failure_class, failure_retryable: state.failure_retryable, next_attempt_unix_seconds: state.next_attempt_unix_seconds, } } /// Builds a synthetic persisted task for older release roots without SQLite state. pub fn from_queued_task(task: OfficialTextUnitTask, generated_unix_seconds: u64) -> Self { let task_status = initial_task_status(&task); let failure_reason = if task_status == TranslationTaskStatus::Skipped { task.reason.clone() } else { None }; Self::from_task( task, PersistedTranslationTaskState { task_status, failure_reason, attempt_count: 0, created_unix_seconds: generated_unix_seconds, updated_unix_seconds: generated_unix_seconds, completed_unix_seconds: None, provider_run_id: None, translation_results: Vec::new(), provider: None, lease_owner: None, lease_expires_unix_seconds: None, failure_class: None, failure_retryable: false, next_attempt_unix_seconds: None, }, ) } fn from_row(row: TranslationTaskRow) -> Result { let ( task_json, worker_status, failure_reason, attempt_count, created_unix_seconds, updated_unix_seconds, completed_unix_seconds, provider_run_id, translation_results_json, provider, lease_owner, lease_expires_unix_seconds, failure_class, failure_retryable, next_attempt_unix_seconds, ) = row; let task = serde_json::from_str(&task_json) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; let task_status = TranslationTaskStatus::parse(&worker_status).ok_or_else(|| { bat_core::Error::Serialization(format!("未知翻译任务 worker 状态：{worker_status}")) })?; let translation_results = serde_json::from_str(&translation_results_json) .map_err(|error| bat_core::Error::Serialization(error.to_string()))?; Ok(Self { task, task_status, failure_reason, attempt_count: u32::try_from(attempt_count).map_err(|_| { bat_core::Error::Serialization("翻译任务 attempt_count 超出范围".to_string()) })?, created_unix_seconds: u64::try_from(created_unix_seconds).map_err(|_| { bat_core::Error::Serialization("翻译任务 created 时间无效".to_string()) })?, updated_unix_seconds: u64::try_from(updated_unix_seconds).map_err(|_| { bat_core::Error::Serialization("翻译任务 updated 时间无效".to_string()) })?, completed_unix_seconds: completed_unix_seconds .map(|value| { u64::try_from(value).map_err(|_| { bat_core::Error::Serialization("翻译任务 completed 时间无效".to_string()) }) }) .transpose()?, provider_run_id, translation_results, provider, lease_owner, lease_expires_unix_seconds: lease_expires_unix_seconds .map(|value| { u64::try_from(value).map_err(|_| { bat_core::Error::Serialization("翻译任务 lease 时间无效".to_string()) }) }) .transpose()?, failure_class, failure_retryable: failure_retryable != 0, next_attempt_unix_seconds: next_attempt_unix_seconds .map(|value| { u64::try_from(value).map_err(|_| { bat_core::Error::Serialization("翻译任务 next_attempt 时间无效".to_string()) }) }) .transpose()?, }) } } /// Synchronizes a release queue into the durable SQLite repository for that release. pub async fn sync_translation_task_repository_at( resource_root: &Path, queue: &OfficialTextUnitTaskQueue, ) -> Result { let path = SqliteTranslationTaskRepository::repository_path(resource_root); let repository = SqliteTranslationTaskRepository::new(&path).await?; repository.sync_queue(queue).await } fn matches_query(task: &PersistedTranslationTask, query: &OfficialTextUnitTaskQuery) -> bool { let mut queue_query = query.clone(); queue_query.task_status = None; if !textunit_task_matches(&task.task, &queue_query) { return false; } if query .task_status .as_ref() .is_some_and(|status| task.task_status.as_str() != status) { return false; } if let Some(has_failure_reason) = query.has_failure_reason { if task.failure_reason.is_some() != has_failure_reason { return false; } } true } fn initial_task_status(task: &OfficialTextUnitTask) -> TranslationTaskStatus { match task.status { crate::official_textunit_queue::OfficialTextUnitTaskStatus::QueuedOffline => { TranslationTaskStatus::Queued } crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedNoParseEntry | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedNoTextUnit | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedParseFailed | crate::official_textunit_queue::OfficialTextUnitTaskStatus::SkippedUnsupported => { TranslationTaskStatus::Skipped } } } fn parse_status_label(status: crate::official_parse::OfficialParseStatus) -> &'static str { match status { crate::official_parse::OfficialParseStatus::Parsed => "parsed", crate::official_parse::OfficialParseStatus::SkippedUnsupported => "skipped_unsupported", crate::official_parse::OfficialParseStatus::Failed => "failed", } } async fn ensure_column( pool: &SqlitePool, table: &str, column: &str, column_type: &str, ) -> Result<()> { let exists: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info(?1) WHERE name = ?2") .bind(table) .bind(column) .fetch_one(pool) .await .map_err(db_error)?; if exists == 0 { let mut query = QueryBuilder::::new("ALTER TABLE "); query .push(table) .push(" ADD COLUMN ") .push(column) .push(" "); query.push(column_type); query.build().execute(pool).await.map_err(db_error)?; } Ok(()) } fn db_error(error: sqlx::Error) -> bat_core::Error { bat_core::Error::Other(error.into()) } fn unix_seconds_now_i64() -> i64 { SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default() .as_secs() .try_into() .unwrap_or(i64::MAX) } fn unix_seconds_now() -> u64 { unix_seconds_now_i64().max(0) as u64 } fn provider_run_status_rank(status: ProviderRunStatus) -> u8 { match status { ProviderRunStatus::Queued => 0, ProviderRunStatus::Succeeded => 1, ProviderRunStatus::Running => 2, ProviderRunStatus::Cancelled => 3, ProviderRunStatus::Failed => 4, } } #[cfg(test)] mod tests { use super::*; use crate::official_changes::OfficialResourceChangeKind; use crate::official_parse::{OfficialParseSourceKind, OfficialParseStatus}; use crate::official_textunit_queue::{ OfficialTextUnitTaskStatus, OfficialTextUnitTaskSummary, OFFICIAL_TEXTUNIT_TASK_QUEUE_VERSION, }; fn task( task_id: &str, destination: &str, status: OfficialTextUnitTaskStatus, parse_status: Option, reason: Option<&str>, ) -> OfficialTextUnitTask { OfficialTextUnitTask { task_id: task_id.to_string(), official_release_id: "release-current".to_string(), destination: destination.to_string(), change_kind: OfficialResourceChangeKind::Added, url: format!("https://example.invalid/{destination}"), bytes: 10, blake3: format!("{task_id}-hash"), parse_entry_key: Some(format!("direct:{destination}")), archive_entry: None, source_kind: Some(OfficialParseSourceKind::DirectBundle), parse_status, text_asset_count: usize::from(status == OfficialTextUnitTaskStatus::QueuedOffline), text_assets: if status == OfficialTextUnitTaskStatus::QueuedOffline { vec!["Scenario".to_string()] } else { Vec::new() }, text_unit_count: if status == OfficialTextUnitTaskStatus::QueuedOffline { 3 } else { 0 }, text_unit_formats: if status == OfficialTextUnitTaskStatus::QueuedOffline { vec!["plain".to_string()] } else { Vec::new() }, text_unit_error_count: 0, status, reason: reason.map(str::to_string), } } fn queue(tasks: Vec) -> OfficialTextUnitTaskQueue { OfficialTextUnitTaskQueue { queue_version: OFFICIAL_TEXTUNIT_TASK_QUEUE_VERSION, official_release_id: "release-current".to_string(), previous_release_id: Some("release-previous".to_string()), generated_unix_seconds: 123, current_resource_root: std::path::PathBuf::from("/tmp/release-current"), summary: OfficialTextUnitTaskSummary { resource_candidate_count: tasks.len(), parse_entry_count: tasks.len(), queued_task_count: tasks .iter() .filter(|task| task.status == OfficialTextUnitTaskStatus::QueuedOffline) .count(), skipped_parse_failed_count: tasks .iter() .filter(|task| task.status == OfficialTextUnitTaskStatus::SkippedParseFailed) .count(), text_unit_count: tasks.iter().map(|task| task.text_unit_count).sum(), ..OfficialTextUnitTaskSummary::default() }, tasks, } } #[tokio::test] async fn sqlite_translation_tasks_sync_and_preserve_worker_state() { let temp = tempfile::TempDir::new().unwrap(); let repository = SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite")) .await .unwrap(); let queue = queue(vec![ task( "task-a", "Bundles/a.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, ), task( "task-b", "Bundles/b.bundle", OfficialTextUnitTaskStatus::SkippedParseFailed, Some(OfficialParseStatus::Failed), Some("parser failed"), ), ]); let first = repository.sync_queue(&queue).await.unwrap(); assert_eq!(first.inserted_count, 2); assert_eq!(first.preserved_state_count, 0); let failed = repository .list(&OfficialTextUnitTaskQuery { task_status: Some("skipped".to_string()), has_reason: Some(true), ..OfficialTextUnitTaskQuery::default() }) .await .unwrap(); assert_eq!(failed.len(), 1); assert_eq!(failed[0].task.task_id, "task-b"); assert_eq!(failed[0].failure_reason.as_deref(), Some("parser failed")); let running = repository .update_status( "task-a", TranslationTaskStatus::Running, None, Some("run-1".to_string()), ) .await .unwrap(); assert_eq!(running.task_status, TranslationTaskStatus::Running); assert_eq!(running.attempt_count, 1); let failed = repository .update_status( "task-a", TranslationTaskStatus::Failed, Some("remote provider rejected payload".to_string()), None, ) .await .unwrap(); assert_eq!(failed.task_status, TranslationTaskStatus::Failed); assert_eq!( failed.failure_reason.as_deref(), Some("remote provider rejected payload") ); assert_eq!(failed.provider_run_id.as_deref(), Some("run-1")); let second = repository.sync_queue(&queue).await.unwrap(); assert_eq!(second.inserted_count, 0); assert_eq!(second.preserved_state_count, 2); let retrievable = repository .list(&OfficialTextUnitTaskQuery { task_status: Some("failed".to_string()), has_failure_reason: Some(true), ..OfficialTextUnitTaskQuery::default() }) .await .unwrap(); assert_eq!(retrievable.len(), 1); assert_eq!(retrievable[0].task.task_id, "task-a"); assert_eq!(retrievable[0].attempt_count, 1); } #[tokio::test] async fn sqlite_translation_tasks_persist_manual_results_without_worker_lease() { let temp = tempfile::TempDir::new().unwrap(); let repository = SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite")) .await .unwrap(); let queue = queue(vec![task( "task-a", "Bundles/a.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, )]); repository.sync_queue(&queue).await.unwrap(); let result = TranslationTaskUnitResult { unit_id: "unit-a".to_string(), source_text: "source".to_string(), translated_text: "manual translation".to_string(), source_kind: TranslationTaskResultSourceKind::Manual, translation_memory_record_id: None, provider: "manual".to_string(), provider_run_id: "manual-run-1".to_string(), translated_unix_seconds: 321, glossary_qa: None, glossary_override: None, }; let updated = repository .update_status_with_results( "task-a", TranslationTaskStatus::Completed, None, Some("manual-run-1".to_string()), Some("manual".to_string()), Some(std::slice::from_ref(&result)), ) .await .unwrap(); assert_eq!(updated.task_status, TranslationTaskStatus::Completed); assert_eq!(updated.provider.as_deref(), Some("manual")); assert_eq!(updated.provider_run_id.as_deref(), Some("manual-run-1")); assert_eq!(updated.translation_results, vec![result.clone()]); assert_eq!( repository.find("task-a").await.unwrap().translation_results, vec![result] ); } #[tokio::test] async fn sqlite_translation_tasks_recover_expired_leases_for_retry() { let temp = tempfile::TempDir::new().unwrap(); let repository = SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite")) .await .unwrap(); let queue = queue(vec![task( "task-a", "Bundles/a.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, )]); repository.sync_queue(&queue).await.unwrap(); let claimed = repository .claim_next("worker-a", "mock", 1, 3) .await .unwrap() .unwrap(); assert_eq!(claimed.task_status, TranslationTaskStatus::Running); assert_eq!(claimed.lease_owner.as_deref(), Some("worker-a")); force_expire_lease(&repository, "task-a").await; let recovered = repository .recover_expired_leases(unix_seconds_now()) .await .unwrap(); assert_eq!(recovered, 1); let recovered_task = repository.find("task-a").await.unwrap(); assert_eq!(recovered_task.task_status, TranslationTaskStatus::Failed); assert_eq!( recovered_task.failure_reason.as_deref(), Some("worker lease expired; task recovered") ); assert_eq!( recovered_task.failure_class.as_deref(), Some("lease_expired") ); assert!(recovered_task.failure_retryable); assert_eq!(recovered_task.lease_owner, None); let reclaimed = repository .claim_next("worker-b", "mock", 30, 3) .await .unwrap() .unwrap(); assert_eq!(reclaimed.task_status, TranslationTaskStatus::Running); assert_eq!(reclaimed.attempt_count, 2); assert_eq!(reclaimed.lease_owner.as_deref(), Some("worker-b")); } #[tokio::test] async fn sqlite_translation_tasks_reject_stale_worker_writes() { let temp = tempfile::TempDir::new().unwrap(); let repository = SqliteTranslationTaskRepository::new(temp.path().join("translation-tasks.sqlite")) .await .unwrap(); let queue = queue(vec![task( "task-a", "Bundles/a.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, )]); repository.sync_queue(&queue).await.unwrap(); let first_claim = repository .claim_next("worker-a", "mock", 1, 3) .await .unwrap() .unwrap(); let first_run = first_claim.provider_run_id.clone().unwrap(); force_expire_lease(&repository, "task-a").await; repository .recover_expired_leases(unix_seconds_now()) .await .unwrap(); let second_claim = repository .claim_next("worker-b", "mock", 30, 3) .await .unwrap() .unwrap(); let second_run = second_claim.provider_run_id.clone().unwrap(); let result = TranslationTaskUnitResult { unit_id: "unit-a".to_string(), source_text: "source".to_string(), translated_text: "translated".to_string(), source_kind: TranslationTaskResultSourceKind::Provider, translation_memory_record_id: None, provider: "mock".to_string(), provider_run_id: second_run.clone(), translated_unix_seconds: 1, glossary_qa: None, glossary_override: None, }; assert!(repository .complete_claim( "task-a", "worker-a", &first_run, "mock", std::slice::from_ref(&result) ) .await .is_err()); let completed = repository .complete_claim( "task-a", "worker-b", &second_run, "mock", std::slice::from_ref(&result), ) .await .unwrap(); assert_eq!(completed.task_status, TranslationTaskStatus::Completed); assert_eq!(completed.translation_results, vec![result.clone()]); assert!(repository .fail_claim(TranslationTaskFailure { task_id: "task-a".to_string(), worker_id: "worker-a".to_string(), provider_run_id: first_run, failure_class: "network".to_string(), failure_reason: "late failure".to_string(), retryable: true, next_attempt_unix_seconds: Some(2), }) .await .is_err()); let task = repository.find("task-a").await.unwrap(); assert_eq!(task.task_status, TranslationTaskStatus::Completed); assert_eq!(task.translation_results, vec![result]); } #[test] fn translation_handoff_maps_worker_state_and_preserves_provider_progress() { let queue = queue(vec![ task( "task-a", "Bundles/a.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, ), task( "task-b", "Bundles/b.bundle", OfficialTextUnitTaskStatus::QueuedOffline, Some(OfficialParseStatus::Parsed), None, ), task( "task-c", "Bundles/c.bundle", OfficialTextUnitTaskStatus::SkippedParseFailed, Some(OfficialParseStatus::Failed), Some("parser failed"), ), ]); let persisted = vec![ PersistedTranslationTask { task: queue.tasks[0].clone(), task_status: TranslationTaskStatus::Running, failure_reason: None, attempt_count: 2, created_unix_seconds: 123, updated_unix_seconds: 124, completed_unix_seconds: None, provider_run_id: Some("provider-run-1".to_string()), translation_results: Vec::new(), provider: Some("fixture".to_string()), lease_owner: None, lease_expires_unix_seconds: None, failure_class: None, failure_retryable: false, next_attempt_unix_seconds: None, }, PersistedTranslationTask { task: queue.tasks[1].clone(), task_status: TranslationTaskStatus::Completed, failure_reason: None, attempt_count: 1, created_unix_seconds: 123, updated_unix_seconds: 125, completed_unix_seconds: Some(125), provider_run_id: Some("provider-run-1".to_string()), translation_results: Vec::new(), provider: Some("fixture".to_string()), lease_owner: None, lease_expires_unix_seconds: None, failure_class: None, failure_retryable: false, next_attempt_unix_seconds: None, }, ]; let handoff = build_translation_handoff(&queue, &persisted); assert_eq!(handoff.handoff_version, TRANSLATION_HANDOFF_SCHEMA_VERSION); assert_eq!(handoff.source_queue_file, OFFICIAL_TEXTUNIT_TASK_QUEUE_FILE); assert_eq!(handoff.job.status, TranslationJobStatus::Failed); assert_eq!(handoff.job.unit_count, 3); assert_eq!(handoff.units[0].status, TranslationUnitStatus::Translating); assert_eq!(handoff.units[1].status, TranslationUnitStatus::Translated); assert_eq!(handoff.units[2].status, TranslationUnitStatus::Skipped); assert_eq!( handoff.units[2].failure_reason.as_deref(), Some("parser failed") ); assert_eq!(handoff.provider_runs.len(), 1); assert_eq!(handoff.provider_runs[0].status, ProviderRunStatus::Running); assert_eq!(handoff.provider_runs[0].unit_ids, ["task-a", "task-b"]); assert_eq!(handoff.provider_runs[0].attempt_count, 2); } async fn force_expire_lease(repository: &SqliteTranslationTaskRepository, task_id: &str) { sqlx::query( r#" UPDATE translation_tasks SET lease_expires_unix_seconds = 0 WHERE task_id = ?1 "#, ) .bind(task_id) .execute(&repository.pool) .await .unwrap(); } #[test] fn translation_handoff_file_round_trips_with_version_check() { let temp = tempfile::TempDir::new().unwrap(); let handoff = build_translation_handoff(&queue(Vec::new()), &[]); write_translation_handoff_at(temp.path(), &handoff).unwrap(); let loaded = read_translation_handoff_at(temp.path()).unwrap().unwrap(); assert_eq!(loaded, handoff); assert!(temp.path().join(TRANSLATION_HANDOFF_FILE).is_file()); } }