kdletters
951caac32d
新增外部生成队列概览和单任务状态契约 将跳一跳、拼消消、敲木鱼图片生成动作接入worker队列 前端生成等待页展示当前任务和队列数量 更新外部生成worker运维文档和团队决策记录
937 lines
32 KiB
Rust
937 lines
32 KiB
Rust
use crate::*;
|
|
|
|
const EXTERNAL_GENERATION_STATUS_PENDING: &str = "pending";
|
|
const EXTERNAL_GENERATION_STATUS_RUNNING: &str = "running";
|
|
const EXTERNAL_GENERATION_STATUS_COMPLETED: &str = "completed";
|
|
const EXTERNAL_GENERATION_STATUS_FAILED: &str = "failed";
|
|
const EXTERNAL_GENERATION_STATUS_CANCELLED: &str = "cancelled";
|
|
|
|
#[spacetimedb::table(
|
|
accessor = external_generation_job,
|
|
index(
|
|
accessor = by_external_generation_job_status_available,
|
|
btree(columns = [status, available_at])
|
|
),
|
|
index(
|
|
accessor = by_external_generation_job_worker_id,
|
|
btree(columns = [worker_id])
|
|
),
|
|
index(
|
|
accessor = by_external_generation_job_source,
|
|
btree(columns = [source_module, source_entity_id])
|
|
),
|
|
index(
|
|
accessor = by_external_generation_job_owner_user_id,
|
|
btree(columns = [owner_user_id])
|
|
)
|
|
)]
|
|
#[derive(Clone)]
|
|
pub struct ExternalGenerationJob {
|
|
#[primary_key]
|
|
pub(crate) job_id: String,
|
|
#[unique]
|
|
pub(crate) dedupe_key: String,
|
|
pub(crate) job_kind: String,
|
|
pub(crate) owner_user_id: String,
|
|
pub(crate) source_module: String,
|
|
pub(crate) source_entity_id: String,
|
|
pub(crate) request_label: String,
|
|
pub(crate) request_payload_json: String,
|
|
pub(crate) status: String,
|
|
pub(crate) attempt: u32,
|
|
pub(crate) max_attempts: u32,
|
|
pub(crate) last_error_message: Option<String>,
|
|
pub(crate) worker_id: Option<String>,
|
|
pub(crate) lease_expires_at: Option<Timestamp>,
|
|
pub(crate) available_at: Timestamp,
|
|
pub(crate) result_payload_json: Option<String>,
|
|
pub(crate) created_at: Timestamp,
|
|
pub(crate) started_at: Option<Timestamp>,
|
|
pub(crate) completed_at: Option<Timestamp>,
|
|
pub(crate) updated_at: Timestamp,
|
|
#[default(None::<String>)]
|
|
pub(crate) lease_token: Option<String>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobEnqueueInput {
|
|
pub job_id: String,
|
|
pub dedupe_key: String,
|
|
pub job_kind: String,
|
|
pub owner_user_id: String,
|
|
pub source_module: String,
|
|
pub source_entity_id: String,
|
|
pub request_label: String,
|
|
pub request_payload_json: String,
|
|
pub max_attempts: u32,
|
|
pub available_at_micros: i64,
|
|
pub created_at_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobClaimInput {
|
|
pub worker_id: String,
|
|
pub limit: u32,
|
|
pub lease_expires_at_micros: i64,
|
|
pub claimed_at_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobRenewLeaseInput {
|
|
pub job_id: String,
|
|
pub worker_id: String,
|
|
pub lease_token: String,
|
|
pub lease_expires_at_micros: i64,
|
|
pub renewed_at_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobCompleteInput {
|
|
pub job_id: String,
|
|
pub worker_id: String,
|
|
pub lease_token: String,
|
|
pub result_payload_json: Option<String>,
|
|
pub completed_at_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobFailInput {
|
|
pub job_id: String,
|
|
pub worker_id: String,
|
|
pub lease_token: String,
|
|
pub error_message: String,
|
|
pub retry_after_micros: i64,
|
|
pub failed_at_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobGetInput {
|
|
pub job_id: String,
|
|
pub owner_user_id: String,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobSnapshot {
|
|
pub job_id: String,
|
|
pub dedupe_key: String,
|
|
pub job_kind: String,
|
|
pub owner_user_id: String,
|
|
pub source_module: String,
|
|
pub source_entity_id: String,
|
|
pub request_label: String,
|
|
pub request_payload_json: String,
|
|
pub status: String,
|
|
pub attempt: u32,
|
|
pub max_attempts: u32,
|
|
pub last_error_message: Option<String>,
|
|
pub worker_id: Option<String>,
|
|
pub lease_expires_at_micros: Option<i64>,
|
|
pub available_at_micros: i64,
|
|
pub result_payload_json: Option<String>,
|
|
pub created_at_micros: i64,
|
|
pub started_at_micros: Option<i64>,
|
|
pub completed_at_micros: Option<i64>,
|
|
pub updated_at_micros: i64,
|
|
pub lease_token: Option<String>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationJobProcedureResult {
|
|
pub ok: bool,
|
|
pub job: Option<ExternalGenerationJobSnapshot>,
|
|
pub jobs: Vec<ExternalGenerationJobSnapshot>,
|
|
pub error_message: Option<String>,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationQueueStatsSnapshot {
|
|
pub pending_count: u32,
|
|
pub delayed_pending_count: u32,
|
|
pub claimable_pending_count: u32,
|
|
pub running_active_count: u32,
|
|
pub expired_running_count: u32,
|
|
// 中文注释：保留字段兼容已生成 bindings；controller 只按非终态队列压力扩缩容，不每轮扫描历史终态任务。
|
|
pub terminal_count: u32,
|
|
pub claimable_count: u32,
|
|
pub oldest_claimable_age_micros: Option<i64>,
|
|
pub now_micros: i64,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
|
|
pub struct ExternalGenerationQueueStatsProcedureResult {
|
|
pub ok: bool,
|
|
pub stats: Option<ExternalGenerationQueueStatsSnapshot>,
|
|
pub error_message: Option<String>,
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn enqueue_external_generation_job_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobEnqueueInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| enqueue_external_generation_job_tx(tx, input.clone())) {
|
|
Ok(job) => single_external_generation_job_result(job),
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn claim_external_generation_jobs_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobClaimInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| claim_external_generation_jobs_tx(tx, input.clone())) {
|
|
Ok(jobs) => ExternalGenerationJobProcedureResult {
|
|
ok: true,
|
|
job: None,
|
|
jobs,
|
|
error_message: None,
|
|
},
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn complete_external_generation_job_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobCompleteInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| complete_external_generation_job_tx(tx, input.clone())) {
|
|
Ok(job) => single_external_generation_job_result(job),
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn renew_external_generation_job_lease_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobRenewLeaseInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| renew_external_generation_job_lease_tx(tx, input.clone())) {
|
|
Ok(job) => single_external_generation_job_result(job),
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn fail_external_generation_job_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobFailInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| fail_external_generation_job_tx(tx, input.clone())) {
|
|
Ok(job) => single_external_generation_job_result(job),
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn get_external_generation_job_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
input: ExternalGenerationJobGetInput,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
match ctx.try_with_tx(|tx| get_external_generation_job_tx(tx, input.clone())) {
|
|
Ok(job) => single_external_generation_job_result(job),
|
|
Err(message) => failed_external_generation_job_result(message),
|
|
}
|
|
}
|
|
|
|
#[spacetimedb::procedure]
|
|
pub fn get_external_generation_queue_stats_and_return(
|
|
ctx: &mut ProcedureContext,
|
|
) -> ExternalGenerationQueueStatsProcedureResult {
|
|
match ctx.try_with_tx(|tx| get_external_generation_queue_stats_tx(tx)) {
|
|
Ok(stats) => ExternalGenerationQueueStatsProcedureResult {
|
|
ok: true,
|
|
stats: Some(stats),
|
|
error_message: None,
|
|
},
|
|
Err(message) => ExternalGenerationQueueStatsProcedureResult {
|
|
ok: false,
|
|
stats: None,
|
|
error_message: Some(message),
|
|
},
|
|
}
|
|
}
|
|
|
|
fn enqueue_external_generation_job_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobEnqueueInput,
|
|
) -> Result<ExternalGenerationJobSnapshot, String> {
|
|
validate_required("external_generation_job.job_id", &input.job_id)?;
|
|
validate_required("external_generation_job.dedupe_key", &input.dedupe_key)?;
|
|
validate_required("external_generation_job.job_kind", &input.job_kind)?;
|
|
validate_required(
|
|
"external_generation_job.owner_user_id",
|
|
&input.owner_user_id,
|
|
)?;
|
|
validate_required(
|
|
"external_generation_job.source_module",
|
|
&input.source_module,
|
|
)?;
|
|
validate_required(
|
|
"external_generation_job.source_entity_id",
|
|
&input.source_entity_id,
|
|
)?;
|
|
validate_required(
|
|
"external_generation_job.request_label",
|
|
&input.request_label,
|
|
)?;
|
|
validate_required(
|
|
"external_generation_job.request_payload_json",
|
|
&input.request_payload_json,
|
|
)?;
|
|
|
|
if let Some(row) = ctx
|
|
.db
|
|
.external_generation_job()
|
|
.dedupe_key()
|
|
.find(&input.dedupe_key)
|
|
{
|
|
return Ok(map_external_generation_job_row(row));
|
|
}
|
|
if ctx
|
|
.db
|
|
.external_generation_job()
|
|
.job_id()
|
|
.find(&input.job_id)
|
|
.is_some()
|
|
{
|
|
return Err("external_generation_job.job_id 已存在".to_string());
|
|
}
|
|
|
|
let now = Timestamp::from_micros_since_unix_epoch(input.created_at_micros);
|
|
let available_at = Timestamp::from_micros_since_unix_epoch(input.available_at_micros);
|
|
let row = ExternalGenerationJob {
|
|
job_id: input.job_id.trim().to_string(),
|
|
dedupe_key: input.dedupe_key.trim().to_string(),
|
|
job_kind: input.job_kind.trim().to_string(),
|
|
owner_user_id: input.owner_user_id.trim().to_string(),
|
|
source_module: input.source_module.trim().to_string(),
|
|
source_entity_id: input.source_entity_id.trim().to_string(),
|
|
request_label: input.request_label.trim().to_string(),
|
|
request_payload_json: input.request_payload_json.trim().to_string(),
|
|
status: EXTERNAL_GENERATION_STATUS_PENDING.to_string(),
|
|
attempt: 0,
|
|
max_attempts: input.max_attempts.max(1),
|
|
last_error_message: None,
|
|
worker_id: None,
|
|
lease_expires_at: None,
|
|
available_at,
|
|
result_payload_json: None,
|
|
created_at: now,
|
|
started_at: None,
|
|
completed_at: None,
|
|
updated_at: now,
|
|
lease_token: None,
|
|
};
|
|
ctx.db.external_generation_job().insert(row.clone());
|
|
Ok(map_external_generation_job_row(row))
|
|
}
|
|
|
|
fn claim_external_generation_jobs_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobClaimInput,
|
|
) -> Result<Vec<ExternalGenerationJobSnapshot>, String> {
|
|
validate_required("external_generation_job.worker_id", &input.worker_id)?;
|
|
if input.limit == 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
|
|
let claim_time = ctx.timestamp;
|
|
let lease_duration_micros = duration_between_micros(
|
|
input.lease_expires_at_micros,
|
|
input.claimed_at_micros,
|
|
"external_generation_job.lease_duration",
|
|
)?;
|
|
let lease_expires_at = timestamp_after_micros(claim_time, lease_duration_micros);
|
|
let worker_id = input.worker_id.trim().to_string();
|
|
let limit = input.limit.min(64) as usize;
|
|
let mut candidates = Vec::new();
|
|
|
|
candidates.extend(
|
|
ctx.db
|
|
.external_generation_job()
|
|
.by_external_generation_job_status_available()
|
|
.filter(&EXTERNAL_GENERATION_STATUS_PENDING.to_string())
|
|
.filter(|row| is_external_generation_job_claimable(row, claim_time)),
|
|
);
|
|
candidates.extend(
|
|
ctx.db
|
|
.external_generation_job()
|
|
.by_external_generation_job_status_available()
|
|
.filter(&EXTERNAL_GENERATION_STATUS_RUNNING.to_string())
|
|
.filter(|row| is_external_generation_job_claimable(row, claim_time)),
|
|
);
|
|
|
|
candidates.sort_by(|left, right| {
|
|
left.available_at
|
|
.to_micros_since_unix_epoch()
|
|
.cmp(&right.available_at.to_micros_since_unix_epoch())
|
|
.then_with(|| {
|
|
left.created_at
|
|
.to_micros_since_unix_epoch()
|
|
.cmp(&right.created_at.to_micros_since_unix_epoch())
|
|
})
|
|
.then_with(|| left.job_id.cmp(&right.job_id))
|
|
});
|
|
|
|
let mut claimed = Vec::new();
|
|
for mut row in candidates.into_iter().take(limit) {
|
|
let next_attempt = row.attempt.saturating_add(1);
|
|
let lease_token = build_external_generation_lease_token(
|
|
&row.job_id,
|
|
&worker_id,
|
|
next_attempt,
|
|
claim_time,
|
|
);
|
|
row.status = EXTERNAL_GENERATION_STATUS_RUNNING.to_string();
|
|
row.worker_id = Some(worker_id.clone());
|
|
row.lease_expires_at = Some(lease_expires_at);
|
|
row.lease_token = Some(lease_token);
|
|
row.attempt = next_attempt;
|
|
if row.started_at.is_none() {
|
|
row.started_at = Some(claim_time);
|
|
}
|
|
row.updated_at = claim_time;
|
|
persist_external_generation_job_row(ctx, row.clone());
|
|
claimed.push(map_external_generation_job_row(row));
|
|
}
|
|
|
|
Ok(claimed)
|
|
}
|
|
|
|
fn complete_external_generation_job_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobCompleteInput,
|
|
) -> Result<ExternalGenerationJobSnapshot, String> {
|
|
let mut row = get_worker_owned_external_generation_job(
|
|
ctx,
|
|
&input.job_id,
|
|
&input.worker_id,
|
|
&input.lease_token,
|
|
)?;
|
|
let completed_at = ctx.timestamp;
|
|
row.status = EXTERNAL_GENERATION_STATUS_COMPLETED.to_string();
|
|
row.result_payload_json = input
|
|
.result_payload_json
|
|
.and_then(|value| normalize_optional_text(value.as_str()));
|
|
row.lease_expires_at = None;
|
|
row.completed_at = Some(completed_at);
|
|
row.updated_at = completed_at;
|
|
persist_external_generation_job_row(ctx, row.clone());
|
|
Ok(map_external_generation_job_row(row))
|
|
}
|
|
|
|
fn get_external_generation_job_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobGetInput,
|
|
) -> Result<ExternalGenerationJobSnapshot, String> {
|
|
validate_required("external_generation_job.job_id", &input.job_id)?;
|
|
validate_required(
|
|
"external_generation_job.owner_user_id",
|
|
&input.owner_user_id,
|
|
)?;
|
|
let row = ctx
|
|
.db
|
|
.external_generation_job()
|
|
.job_id()
|
|
.find(&input.job_id.trim().to_string())
|
|
.ok_or_else(|| "external_generation_job 不存在".to_string())?;
|
|
if row.owner_user_id.trim() != input.owner_user_id.trim() {
|
|
return Err("external_generation_job 不存在".to_string());
|
|
}
|
|
|
|
Ok(map_external_generation_job_row(row))
|
|
}
|
|
|
|
fn renew_external_generation_job_lease_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobRenewLeaseInput,
|
|
) -> Result<ExternalGenerationJobSnapshot, String> {
|
|
let mut row = get_worker_owned_external_generation_job(
|
|
ctx,
|
|
&input.job_id,
|
|
&input.worker_id,
|
|
&input.lease_token,
|
|
)?;
|
|
let renewed_at = ctx.timestamp;
|
|
let lease_duration_micros = duration_between_micros(
|
|
input.lease_expires_at_micros,
|
|
input.renewed_at_micros,
|
|
"external_generation_job.lease_duration",
|
|
)?;
|
|
row.lease_expires_at = Some(timestamp_after_micros(renewed_at, lease_duration_micros));
|
|
row.updated_at = renewed_at;
|
|
persist_external_generation_job_row(ctx, row.clone());
|
|
Ok(map_external_generation_job_row(row))
|
|
}
|
|
|
|
fn fail_external_generation_job_tx(
|
|
ctx: &ReducerContext,
|
|
input: ExternalGenerationJobFailInput,
|
|
) -> Result<ExternalGenerationJobSnapshot, String> {
|
|
let error_message = input.error_message.trim();
|
|
if error_message.is_empty() {
|
|
return Err("external_generation_job.error_message 不能为空".to_string());
|
|
}
|
|
|
|
let mut row = get_worker_owned_external_generation_job(
|
|
ctx,
|
|
&input.job_id,
|
|
&input.worker_id,
|
|
&input.lease_token,
|
|
)?;
|
|
let failed_at = ctx.timestamp;
|
|
let retry_delay_micros = duration_between_micros(
|
|
input.retry_after_micros,
|
|
input.failed_at_micros,
|
|
"external_generation_job.retry_delay",
|
|
)?;
|
|
row.last_error_message = Some(error_message.to_string());
|
|
row.lease_expires_at = None;
|
|
row.worker_id = None;
|
|
row.lease_token = None;
|
|
row.updated_at = failed_at;
|
|
|
|
if row.attempt < row.max_attempts {
|
|
row.status = EXTERNAL_GENERATION_STATUS_PENDING.to_string();
|
|
row.available_at = timestamp_after_micros(failed_at, retry_delay_micros);
|
|
} else {
|
|
row.status = EXTERNAL_GENERATION_STATUS_FAILED.to_string();
|
|
row.completed_at = Some(failed_at);
|
|
}
|
|
|
|
persist_external_generation_job_row(ctx, row.clone());
|
|
Ok(map_external_generation_job_row(row))
|
|
}
|
|
|
|
fn get_external_generation_queue_stats_tx(
|
|
ctx: &ReducerContext,
|
|
) -> Result<ExternalGenerationQueueStatsSnapshot, String> {
|
|
let now = ctx.timestamp;
|
|
let now_micros = now.to_micros_since_unix_epoch();
|
|
let mut stats = ExternalGenerationQueueStatsSnapshot {
|
|
pending_count: 0,
|
|
delayed_pending_count: 0,
|
|
claimable_pending_count: 0,
|
|
running_active_count: 0,
|
|
expired_running_count: 0,
|
|
terminal_count: 0,
|
|
claimable_count: 0,
|
|
oldest_claimable_age_micros: None,
|
|
now_micros,
|
|
};
|
|
|
|
for row in ctx
|
|
.db
|
|
.external_generation_job()
|
|
.by_external_generation_job_status_available()
|
|
.filter(&EXTERNAL_GENERATION_STATUS_PENDING.to_string())
|
|
{
|
|
stats.pending_count = stats.pending_count.saturating_add(1);
|
|
if is_external_generation_job_claimable(&row, now) {
|
|
stats.claimable_pending_count = stats.claimable_pending_count.saturating_add(1);
|
|
record_external_generation_claimable_age(&mut stats, &row, now_micros);
|
|
} else {
|
|
stats.delayed_pending_count = stats.delayed_pending_count.saturating_add(1);
|
|
}
|
|
}
|
|
|
|
for row in ctx
|
|
.db
|
|
.external_generation_job()
|
|
.by_external_generation_job_status_available()
|
|
.filter(&EXTERNAL_GENERATION_STATUS_RUNNING.to_string())
|
|
{
|
|
if is_external_generation_job_claimable(&row, now) {
|
|
stats.expired_running_count = stats.expired_running_count.saturating_add(1);
|
|
record_external_generation_claimable_age(&mut stats, &row, now_micros);
|
|
} else {
|
|
stats.running_active_count = stats.running_active_count.saturating_add(1);
|
|
}
|
|
}
|
|
|
|
stats.claimable_count = stats
|
|
.claimable_pending_count
|
|
.saturating_add(stats.expired_running_count);
|
|
Ok(stats)
|
|
}
|
|
|
|
pub(crate) fn validate_external_generation_job_lease_for_tx(
|
|
ctx: &ReducerContext,
|
|
job_id: &str,
|
|
worker_id: &str,
|
|
lease_token: &str,
|
|
expected_job_kinds: &[&str],
|
|
expected_owner_user_id: &str,
|
|
expected_source_module: &str,
|
|
expected_source_entity_ids: &[String],
|
|
) -> Result<(), String> {
|
|
let row = get_worker_owned_external_generation_job(ctx, job_id, worker_id, lease_token)?;
|
|
if !expected_job_kinds.is_empty()
|
|
&& !expected_job_kinds
|
|
.iter()
|
|
.any(|expected| row.job_kind.trim() == expected.trim())
|
|
{
|
|
return Err("external_generation_job job_kind 与业务写回不匹配".to_string());
|
|
}
|
|
if row.owner_user_id.trim() != expected_owner_user_id.trim() {
|
|
return Err("external_generation_job owner_user_id 与业务写回不匹配".to_string());
|
|
}
|
|
if row.source_module.trim() != expected_source_module.trim() {
|
|
return Err("external_generation_job source_module 与业务写回不匹配".to_string());
|
|
}
|
|
if !expected_source_entity_ids
|
|
.iter()
|
|
.any(|expected| row.source_entity_id.trim() == expected.trim())
|
|
{
|
|
return Err("external_generation_job source_entity_id 与业务写回不匹配".to_string());
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn get_worker_owned_external_generation_job(
|
|
ctx: &ReducerContext,
|
|
job_id: &str,
|
|
worker_id: &str,
|
|
lease_token: &str,
|
|
) -> Result<ExternalGenerationJob, String> {
|
|
validate_required("external_generation_job.job_id", job_id)?;
|
|
validate_required("external_generation_job.worker_id", worker_id)?;
|
|
validate_required("external_generation_job.lease_token", lease_token)?;
|
|
let row = ctx
|
|
.db
|
|
.external_generation_job()
|
|
.job_id()
|
|
.find(&job_id.trim().to_string())
|
|
.ok_or_else(|| "external_generation_job 不存在".to_string())?;
|
|
if row.status != EXTERNAL_GENERATION_STATUS_RUNNING {
|
|
return Err("external_generation_job 当前不是 running 状态".to_string());
|
|
}
|
|
if !is_external_generation_job_owned_by_worker(&row, worker_id) {
|
|
return Err("external_generation_job worker lease 不匹配".to_string());
|
|
}
|
|
if !is_external_generation_job_owned_by_lease_token(&row, lease_token) {
|
|
return Err("external_generation_job lease token 不匹配".to_string());
|
|
}
|
|
if !is_external_generation_job_lease_active(&row, ctx.timestamp) {
|
|
return Err("external_generation_job lease 已过期".to_string());
|
|
}
|
|
Ok(row)
|
|
}
|
|
|
|
fn is_external_generation_job_owned_by_worker(
|
|
row: &ExternalGenerationJob,
|
|
worker_id: &str,
|
|
) -> bool {
|
|
row.worker_id.as_deref() == Some(worker_id.trim())
|
|
}
|
|
|
|
fn is_external_generation_job_owned_by_lease_token(
|
|
row: &ExternalGenerationJob,
|
|
lease_token: &str,
|
|
) -> bool {
|
|
row.lease_token.as_deref() == Some(lease_token.trim())
|
|
}
|
|
|
|
fn is_external_generation_job_lease_active(row: &ExternalGenerationJob, now: Timestamp) -> bool {
|
|
row.lease_expires_at
|
|
.map(|lease_expires_at| lease_expires_at > now)
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
fn is_external_generation_job_claimable(row: &ExternalGenerationJob, now: Timestamp) -> bool {
|
|
match row.status.as_str() {
|
|
EXTERNAL_GENERATION_STATUS_PENDING => row.available_at <= now,
|
|
EXTERNAL_GENERATION_STATUS_RUNNING => row
|
|
.lease_expires_at
|
|
.map(|lease_expires_at| lease_expires_at <= now)
|
|
.unwrap_or(true),
|
|
EXTERNAL_GENERATION_STATUS_COMPLETED
|
|
| EXTERNAL_GENERATION_STATUS_FAILED
|
|
| EXTERNAL_GENERATION_STATUS_CANCELLED => false,
|
|
_ => false,
|
|
}
|
|
}
|
|
|
|
fn record_external_generation_claimable_age(
|
|
stats: &mut ExternalGenerationQueueStatsSnapshot,
|
|
row: &ExternalGenerationJob,
|
|
now_micros: i64,
|
|
) {
|
|
let age = now_micros
|
|
.saturating_sub(row.available_at.to_micros_since_unix_epoch())
|
|
.max(0);
|
|
stats.oldest_claimable_age_micros = Some(
|
|
stats
|
|
.oldest_claimable_age_micros
|
|
.map(|current| current.max(age))
|
|
.unwrap_or(age),
|
|
);
|
|
}
|
|
|
|
fn persist_external_generation_job_row(ctx: &ReducerContext, row: ExternalGenerationJob) {
|
|
ctx.db
|
|
.external_generation_job()
|
|
.job_id()
|
|
.delete(&row.job_id);
|
|
ctx.db.external_generation_job().insert(row);
|
|
}
|
|
|
|
fn map_external_generation_job_row(row: ExternalGenerationJob) -> ExternalGenerationJobSnapshot {
|
|
ExternalGenerationJobSnapshot {
|
|
job_id: row.job_id,
|
|
dedupe_key: row.dedupe_key,
|
|
job_kind: row.job_kind,
|
|
owner_user_id: row.owner_user_id,
|
|
source_module: row.source_module,
|
|
source_entity_id: row.source_entity_id,
|
|
request_label: row.request_label,
|
|
request_payload_json: row.request_payload_json,
|
|
status: row.status,
|
|
attempt: row.attempt,
|
|
max_attempts: row.max_attempts,
|
|
last_error_message: row.last_error_message,
|
|
worker_id: row.worker_id,
|
|
lease_expires_at_micros: row
|
|
.lease_expires_at
|
|
.map(|value| value.to_micros_since_unix_epoch()),
|
|
available_at_micros: row.available_at.to_micros_since_unix_epoch(),
|
|
result_payload_json: row.result_payload_json,
|
|
created_at_micros: row.created_at.to_micros_since_unix_epoch(),
|
|
started_at_micros: row
|
|
.started_at
|
|
.map(|value| value.to_micros_since_unix_epoch()),
|
|
completed_at_micros: row
|
|
.completed_at
|
|
.map(|value| value.to_micros_since_unix_epoch()),
|
|
updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
|
|
lease_token: row.lease_token,
|
|
}
|
|
}
|
|
|
|
fn single_external_generation_job_result(
|
|
job: ExternalGenerationJobSnapshot,
|
|
) -> ExternalGenerationJobProcedureResult {
|
|
ExternalGenerationJobProcedureResult {
|
|
ok: true,
|
|
job: Some(job),
|
|
jobs: Vec::new(),
|
|
error_message: None,
|
|
}
|
|
}
|
|
|
|
fn failed_external_generation_job_result(message: String) -> ExternalGenerationJobProcedureResult {
|
|
ExternalGenerationJobProcedureResult {
|
|
ok: false,
|
|
job: None,
|
|
jobs: Vec::new(),
|
|
error_message: Some(message),
|
|
}
|
|
}
|
|
|
|
fn validate_required(field: &str, value: &str) -> Result<(), String> {
|
|
if value.trim().is_empty() {
|
|
return Err(format!("{field} 不能为空"));
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn duration_between_micros(later: i64, earlier: i64, field: &str) -> Result<i64, String> {
|
|
let duration = later.saturating_sub(earlier);
|
|
if duration <= 0 {
|
|
return Err(format!("{field} 必须大于 0"));
|
|
}
|
|
Ok(duration)
|
|
}
|
|
|
|
fn timestamp_after_micros(timestamp: Timestamp, duration_micros: i64) -> Timestamp {
|
|
Timestamp::from_micros_since_unix_epoch(
|
|
timestamp
|
|
.to_micros_since_unix_epoch()
|
|
.saturating_add(duration_micros.max(0)),
|
|
)
|
|
}
|
|
|
|
fn build_external_generation_lease_token(
|
|
job_id: &str,
|
|
worker_id: &str,
|
|
attempt: u32,
|
|
claimed_at: Timestamp,
|
|
) -> String {
|
|
format!(
|
|
"{}:{}:{}:{}",
|
|
job_id.trim(),
|
|
worker_id.trim(),
|
|
attempt,
|
|
claimed_at.to_micros_since_unix_epoch()
|
|
)
|
|
}
|
|
|
|
fn normalize_optional_text(value: &str) -> Option<String> {
|
|
let normalized = value.trim();
|
|
if normalized.is_empty() {
|
|
None
|
|
} else {
|
|
Some(normalized.to_string())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn external_generation_job_result_failure_is_structured() {
|
|
let result = failed_external_generation_job_result("失败".to_string());
|
|
assert!(!result.ok);
|
|
assert_eq!(result.error_message.as_deref(), Some("失败"));
|
|
assert!(result.jobs.is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn pending_job_is_claimable_only_after_available_time() {
|
|
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_PENDING);
|
|
row.available_at = micros(1_000);
|
|
|
|
assert!(!is_external_generation_job_claimable(&row, micros(999)));
|
|
assert!(is_external_generation_job_claimable(&row, micros(1_000)));
|
|
}
|
|
|
|
#[test]
|
|
fn running_job_is_claimable_only_after_lease_expires() {
|
|
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
|
|
row.lease_expires_at = Some(micros(2_000));
|
|
|
|
assert!(!is_external_generation_job_claimable(&row, micros(1_999)));
|
|
assert!(is_external_generation_job_claimable(&row, micros(2_000)));
|
|
}
|
|
|
|
#[test]
|
|
fn terminal_job_is_not_claimable() {
|
|
for status in [
|
|
EXTERNAL_GENERATION_STATUS_COMPLETED,
|
|
EXTERNAL_GENERATION_STATUS_FAILED,
|
|
EXTERNAL_GENERATION_STATUS_CANCELLED,
|
|
] {
|
|
let row = external_generation_job_fixture(status);
|
|
|
|
assert!(!is_external_generation_job_claimable(&row, micros(10_000)));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn worker_ownership_requires_matching_trimmed_worker_id() {
|
|
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
|
|
row.worker_id = Some("worker-a".to_string());
|
|
|
|
assert!(is_external_generation_job_owned_by_worker(
|
|
&row,
|
|
" worker-a "
|
|
));
|
|
assert!(!is_external_generation_job_owned_by_worker(
|
|
&row, "worker-b"
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn worker_ownership_requires_matching_trimmed_lease_token() {
|
|
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
|
|
row.lease_token = Some("job-1:worker-a:1:1000".to_string());
|
|
|
|
assert!(is_external_generation_job_owned_by_lease_token(
|
|
&row,
|
|
" job-1:worker-a:1:1000 "
|
|
));
|
|
assert!(!is_external_generation_job_owned_by_lease_token(
|
|
&row,
|
|
"job-1:worker-a:2:2000"
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn worker_lease_is_active_only_before_expiry() {
|
|
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
|
|
row.lease_expires_at = Some(micros(2_000));
|
|
|
|
assert!(is_external_generation_job_lease_active(&row, micros(1_999)));
|
|
assert!(!is_external_generation_job_lease_active(
|
|
&row,
|
|
micros(2_000)
|
|
));
|
|
}
|
|
|
|
#[test]
|
|
fn lease_token_changes_with_claim_attempt() {
|
|
let first =
|
|
build_external_generation_lease_token("extgen-test", "worker-a", 1, micros(1_000));
|
|
let second =
|
|
build_external_generation_lease_token("extgen-test", "worker-a", 2, micros(2_000));
|
|
|
|
assert_ne!(first, second);
|
|
}
|
|
|
|
#[test]
|
|
fn claimable_age_keeps_oldest_available_job() {
|
|
let mut stats = ExternalGenerationQueueStatsSnapshot {
|
|
pending_count: 0,
|
|
delayed_pending_count: 0,
|
|
claimable_pending_count: 0,
|
|
running_active_count: 0,
|
|
expired_running_count: 0,
|
|
terminal_count: 0,
|
|
claimable_count: 0,
|
|
oldest_claimable_age_micros: None,
|
|
now_micros: 10_000,
|
|
};
|
|
let mut old_job = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_PENDING);
|
|
old_job.available_at = micros(1_000);
|
|
let mut newer_job = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
|
|
newer_job.available_at = micros(8_000);
|
|
|
|
record_external_generation_claimable_age(&mut stats, &newer_job, 10_000);
|
|
record_external_generation_claimable_age(&mut stats, &old_job, 10_000);
|
|
|
|
assert_eq!(stats.oldest_claimable_age_micros, Some(9_000));
|
|
}
|
|
|
|
#[test]
|
|
fn positive_duration_between_client_times_is_preserved() {
|
|
assert_eq!(
|
|
duration_between_micros(3_500, 1_000, "external_generation_job.lease_duration"),
|
|
Ok(2_500),
|
|
);
|
|
assert!(duration_between_micros(1_000, 1_000, "duration").is_err());
|
|
}
|
|
|
|
fn external_generation_job_fixture(status: &str) -> ExternalGenerationJob {
|
|
ExternalGenerationJob {
|
|
job_id: "extgen-test".to_string(),
|
|
dedupe_key: "puzzle:compile:test".to_string(),
|
|
job_kind: "puzzle_compile_draft".to_string(),
|
|
owner_user_id: "user-1".to_string(),
|
|
source_module: "puzzle".to_string(),
|
|
source_entity_id: "session-1".to_string(),
|
|
request_label: "拼图首关草稿生成".to_string(),
|
|
request_payload_json: r#"{"sessionId":"session-1"}"#.to_string(),
|
|
status: status.to_string(),
|
|
attempt: 0,
|
|
max_attempts: 1,
|
|
last_error_message: None,
|
|
worker_id: None,
|
|
lease_expires_at: None,
|
|
available_at: micros(0),
|
|
result_payload_json: None,
|
|
created_at: micros(0),
|
|
started_at: None,
|
|
completed_at: None,
|
|
updated_at: micros(0),
|
|
lease_token: None,
|
|
}
|
|
}
|
|
|
|
fn micros(value: i64) -> Timestamp {
|
|
Timestamp::from_micros_since_unix_epoch(value)
|
|
}
|
|
}
|