GenarrativeAI/Genarrative
10
0
Fork 0
Code Issues 23 Pull Requests 8 Actions Packages Projects 1 Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into editor-agent-refactored

Browse Source 
# Conflicts:
#	docs/technical/【前端架构】图片画布编辑器MVP接入方案-2026-06-11.md
#	docs/technical/【后端架构】外部生成Worker化方案-2026-06-03.md
#	docs/【后端架构】server-rs与SpacetimeDB数据契约-2026-05-15.md
#	server-rs/crates/api-server/src/editor_agent.rs
#	server-rs/crates/api-server/src/external_generation_worker.rs
#	server-rs/crates/spacetime-client/src/lib.rs
#	server-rs/crates/spacetime-client/src/mapper.rs
#	server-rs/crates/spacetime-client/src/mapper/external_generation.rs
This commit is contained in:
王德宇
2026-07-16 19:32:14 +08:00
parent 621022ea4f aecabdacfd
commit c99769736c
57 changed files with 6201 additions and 853 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server-rs/crates/api-server/src/character_animation_assets.rs
+265 -180
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/config.rs
+1 -73
View File
@@ -17,8 +17,6 @@ const DEFAULT_EXTERNAL_GENERATION_WORKER_LEASE_SECONDS: u64 = 600;
const DEFAULT_EXTERNAL_GENERATION_WORKER_JOB_TIMEOUT_SECONDS: u64 = 900;
const DEFAULT_EXTERNAL_GENERATION_WORKER_LONG_JOB_TIMEOUT_SECONDS: u64 = 1_800;
pub(crate) const DEFAULT_VECTOR_ENGINE_IMAGE_REQUEST_TIMEOUT_MS: u64 = 1_000_000;
const DEFAULT_EDITOR_BACKGROUND_REMOVAL_BASE_URL: &str = "http://58.87.105.82";
const DEFAULT_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS: u64 = 120_000;
const DEFAULT_EDITOR_BGFILTER_BASE_URL: &str = "http://58.87.105.82/bgfilter";
const DEFAULT_EDITOR_BGFILTER_REQUEST_TIMEOUT_MS: u64 = 180_000;
const DEFAULT_EDITOR_BGFILTER_CIRCUIT_FAILURE_THRESHOLD: u32 = 3;
@@ -64,9 +62,6 @@ pub struct AppConfig {
pub wallet_refund_outbox_flush_interval: Duration,
pub wallet_refund_outbox_max_bytes: u64,
pub editor_generation_pricing_override_path: PathBuf,
pub editor_background_removal_base_url: String,
pub editor_background_removal_token: Option<String>,
pub editor_background_removal_request_timeout_ms: u64,
pub editor_bgfilter_base_url: String,
pub editor_bgfilter_token: Option<String>,
pub editor_bgfilter_request_timeout_ms: u64,
@@ -312,11 +307,6 @@ impl Default for AppConfig {
wallet_refund_outbox_max_bytes: 64 * 1024 * 1024,
editor_generation_pricing_override_path:
crate::editor_generation_config::default_editor_generation_pricing_override_path(),
editor_background_removal_base_url: DEFAULT_EDITOR_BACKGROUND_REMOVAL_BASE_URL
.to_string(),
editor_background_removal_token: None,
editor_background_removal_request_timeout_ms:
DEFAULT_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS,
editor_bgfilter_base_url: DEFAULT_EDITOR_BGFILTER_BASE_URL.to_string(),
editor_bgfilter_token: None,
editor_bgfilter_request_timeout_ms: DEFAULT_EDITOR_BGFILTER_REQUEST_TIMEOUT_MS,
@@ -515,18 +505,6 @@ impl AppConfig {
{
config.editor_generation_pricing_override_path = PathBuf::from(pricing_override_path);
}
if let Some(base_url) =
read_first_non_empty_env(&["GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_BASE_URL"])
{
config.editor_background_removal_base_url = base_url;
}
config.editor_background_removal_token =
read_first_non_empty_env(&["GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_TOKEN"]);
if let Some(timeout_ms) = read_first_positive_u64_env(&[
"GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS",
]) {
config.editor_background_removal_request_timeout_ms = timeout_ms;
}
if let Some(base_url) = read_first_non_empty_env(&["GENARRATIVE_EDITOR_BGFILTER_BASE_URL"])
{
config.editor_bgfilter_base_url = base_url;
@@ -1612,8 +1590,7 @@ fn parse_positive_u16(raw: &str) -> Option<u16> {
#[cfg(test)]
mod tests {
use super::{
AppConfig, DEFAULT_EDITOR_BACKGROUND_REMOVAL_BASE_URL,
DEFAULT_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS, DEFAULT_EDITOR_BGFILTER_BASE_URL,
AppConfig, DEFAULT_EDITOR_BGFILTER_BASE_URL,
DEFAULT_EDITOR_BGFILTER_CIRCUIT_COOLDOWN_SECONDS,
DEFAULT_EDITOR_BGFILTER_CIRCUIT_FAILURE_THRESHOLD,
DEFAULT_EDITOR_BGFILTER_REQUEST_TIMEOUT_MS,
@@ -1638,15 +1615,6 @@ mod tests {
assert!(config.llm_base_url.is_empty());
// assert!(config.apimart_base_url.is_empty());
assert!(config.vector_engine_base_url.is_empty());
assert_eq!(
config.editor_background_removal_base_url,
DEFAULT_EDITOR_BACKGROUND_REMOVAL_BASE_URL
);
assert_eq!(
config.editor_background_removal_request_timeout_ms,
DEFAULT_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS
);
assert!(config.editor_background_removal_token.is_none());
assert_eq!(
config.editor_bgfilter_base_url,
DEFAULT_EDITOR_BGFILTER_BASE_URL
@@ -2387,46 +2355,6 @@ mod tests {
}
}
#[test]
fn from_env_reads_editor_background_removal_settings() {
let _guard = ENV_LOCK
.get_or_init(|| Mutex::new(()))
.lock()
.expect("env lock should not poison");
unsafe {
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_BASE_URL");
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_TOKEN");
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS");
std::env::set_var(
"GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_BASE_URL",
"http://10.0.0.12:8090",
);
std::env::set_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_TOKEN", "token-1");
std::env::set_var(
"GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS",
"90000",
);
}
let config = AppConfig::from_env();
assert_eq!(
config.editor_background_removal_base_url,
"http://10.0.0.12:8090"
);
assert_eq!(
config.editor_background_removal_token.as_deref(),
Some("token-1")
);
assert_eq!(config.editor_background_removal_request_timeout_ms, 90_000);
unsafe {
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_BASE_URL");
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_TOKEN");
std::env::remove_var("GENARRATIVE_EDITOR_BACKGROUND_REMOVAL_REQUEST_TIMEOUT_MS");
}
}
#[test]
fn from_env_reads_editor_bgfilter_settings_and_reuses_background_token() {
let _guard = ENV_LOCK
server-rs/crates/api-server/src/editor_project.rs
+1092 -134
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/external_editor_api.rs
+16
View File
@@ -684,6 +684,7 @@ pub async fn generate_external_editor_character_animation(
request_context,
principal.owner_user_id().to_string(),
payload,
None,
)
.await
}
@@ -750,6 +751,7 @@ fn editor_generation_caller(
owner_user_id: principal.owner_user_id().to_string(),
audit_subject_user_id: Some(principal.owner_user_id().to_string()),
audit_project_id: normalize_optional_string(project_id),
phase_reporter: None,
}
}
@@ -882,6 +884,20 @@ mod tests {
["sliceWarning"]["anyOf"][0]["$ref"],
"#/components/schemas/EditorIconSpritesheetSliceWarning"
);
assert_eq!(
parsed["components"]["schemas"]["EditorImageGenerationResponse"]["properties"]
["warning"]["anyOf"][0]["$ref"],
"#/components/schemas/EditorGenerationWarning"
);
assert_eq!(
parsed["components"]["schemas"]["EditorIconSpritesheetGenerationResponse"]["properties"]
["warning"]["anyOf"][0]["$ref"],
"#/components/schemas/EditorGenerationWarning"
);
assert_eq!(
parsed["components"]["schemas"]["EditorGenerationWarning"]["required"],
json!(["code", "reason"])
);
assert!(
parsed["paths"]
.get("/api/external/v1/editor/ui-designs/assets/extractions")
server-rs/crates/api-server/src/external_generation.rs
+66
View File
@@ -172,6 +172,9 @@ fn map_external_generation_job_status(
100,
None,
),
"running" if job.phase.as_deref() == Some("processing") => {
(ExternalGenerationJobStatus::Running, "正在处理。", 70, None)
}
"running" => (ExternalGenerationJobStatus::Running, "正在生成。", 35, None),
"failed" => (
ExternalGenerationJobStatus::Failed,
@@ -314,6 +317,7 @@ mod tests {
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: None,
warning_message: Some("连通域数量不足".to_string()),
};
let status = map_external_generation_job_status_detail(summary.clone());
@@ -326,4 +330,66 @@ mod tests {
assert_eq!(task.warning.as_deref(), Some("连通域数量不足"));
assert!(task.error.is_none());
}
#[test]
fn maps_running_processing_phase_from_backend_projection() {
let status = map_external_generation_job_status(ExternalGenerationJobSummaryRecord {
job_id: "task-processing".to_string(),
job_kind: "editor_character_animation_generation".to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "角色动作生成".to_string(),
request_prompt: None,
status: "running".to_string(),
last_error_message: None,
created_at: "2026-07-13T08:00:00Z".to_string(),
started_at: Some("2026-07-13T08:00:01Z".to_string()),
completed_at: None,
updated_at: "2026-07-13T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 4,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: Some("processing".to_string()),
warning_message: None,
});
assert_eq!(status.status, ExternalGenerationJobStatus::Running);
assert_eq!(status.phase_detail, "正在处理。");
assert_eq!(status.progress, 70);
}
#[test]
fn maps_legacy_running_job_without_phase_as_generating() {
let mut job = ExternalGenerationJobSummaryRecord {
job_id: "task-legacy".to_string(),
job_kind: "editor_image_generation".to_string(),
owner_user_id: "user-1".to_string(),
source_module: "editor-canvas".to_string(),
source_entity_id: "project-1".to_string(),
request_label: "图片生成".to_string(),
request_prompt: None,
status: "running".to_string(),
last_error_message: None,
created_at: "2026-07-13T08:00:00Z".to_string(),
started_at: Some("2026-07-13T08:00:01Z".to_string()),
completed_at: None,
updated_at: "2026-07-13T08:00:10Z".to_string(),
updated_at_micros: 1_000,
price_mud_points: 4,
refund_ledger_id: None,
notification_acknowledged_at: None,
notification_acknowledged_at_micros: None,
phase: None,
warning_message: None,
};
let legacy = map_external_generation_job_status(job.clone());
assert_eq!(legacy.phase_detail, "正在生成。");
job.phase = Some("generating".to_string());
let generating = map_external_generation_job_status(job);
assert_eq!(generating.phase_detail, "正在生成。");
}
}
server-rs/crates/api-server/src/external_generation_worker.rs
+370 -40
View File
File diff suppressed because it is too large Load Diff
server-rs/crates/api-server/src/state.rs
+26
View File
@@ -276,6 +276,7 @@ pub struct AppStateInner {
llm_client: Option<LlmClient>,
creative_agent_gpt5_client: Option<LlmClient>,
matting_client: Option<MattingClient>,
editor_bgfilter_http_client: reqwest::Client,
creative_agent_executor: Arc<MockLangChainRustAgentExecutor>,
// Phase 1 任务 E 的 creative session facade 暂存在 api-server。
// creative_agent_* 表由任务 D 收口后，这里只保留读写 facade。
@@ -515,6 +516,7 @@ impl AppState {
let llm_client = build_llm_client(&config)?;
let creative_agent_gpt5_client = build_creative_agent_gpt5_client(&config)?;
let matting_client = build_matting_client(&config)?;
let editor_bgfilter_http_client = build_editor_bgfilter_http_client(&config)?;
let http_request_permit_pools = HttpRequestPermitPools::from_config(&config);
let (profile_recharge_order_updates, _) = broadcast::channel(128);
@@ -555,6 +557,7 @@ impl AppState {
llm_client,
creative_agent_gpt5_client,
matting_client,
editor_bgfilter_http_client,
creative_agent_executor: Arc::new(MockLangChainRustAgentExecutor),
creative_agent_sessions: Arc::new(Mutex::new(HashMap::new())),
profile_recharge_order_updates,
@@ -1255,6 +1258,10 @@ impl AppState {
self.matting_client.as_ref()
}
pub fn editor_bgfilter_http_client(&self) -> &reqwest::Client {
&self.editor_bgfilter_http_client
}
pub fn creative_agent_executor(&self) -> Arc<MockLangChainRustAgentExecutor> {
self.creative_agent_executor.clone()
}
@@ -1954,6 +1961,25 @@ fn build_matting_client(config: &AppConfig) -> Result<Option<MattingClient>, App
.map_err(|error| AppStateInitError::DependencyUnavailable(error.to_string()))
}
fn build_editor_bgfilter_http_client(
config: &AppConfig,
) -> Result<reqwest::Client, AppStateInitError> {
reqwest::Client::builder()
.timeout(std::time::Duration::from_millis(
config.editor_bgfilter_request_timeout_ms.max(1),
))
.connect_timeout(std::time::Duration::from_secs(30))
.pool_idle_timeout(std::time::Duration::from_secs(300))
.pool_max_idle_per_host(64)
.tcp_keepalive(std::time::Duration::from_secs(60))
.build()
.map_err(|error| {
AppStateInitError::DependencyUnavailable(format!(
"构建共享 BgFilter HTTP 客户端失败：{error}"
))
})
}
fn build_wechat_client(config: &AppConfig) -> WechatClient {
WechatClient::new(WechatConfig {
app_id: config.wechat_mini_program_app_id.clone(),
server-rs/crates/module-assets/src/asset_object_core.rs
+20 -3
View File
@@ -1,5 +1,5 @@
use shared_kernel::{
build_prefixed_seed_id, format_timestamp_micros, normalize_optional_string,
build_prefixed_seed_entropy_id, format_timestamp_micros, normalize_optional_string,
normalize_required_string,
};
@@ -203,12 +203,14 @@ pub fn build_asset_entity_binding_record(
}
}
// 资产确认可能在同一微秒内并发发生（例如角色动作帧 48 路并发确认），
// 主键必须带随机熵；时间前缀仅保留调试可读性，不参与唯一性保证。
pub fn generate_asset_object_id(seed_micros: i64) -> String {
build_prefixed_seed_id(ASSET_OBJECT_ID_PREFIX, seed_micros)
build_prefixed_seed_entropy_id(ASSET_OBJECT_ID_PREFIX, seed_micros)
}
pub fn generate_asset_binding_id(seed_micros: i64) -> String {
build_prefixed_seed_id(ASSET_BINDING_ID_PREFIX, seed_micros)
build_prefixed_seed_entropy_id(ASSET_BINDING_ID_PREFIX, seed_micros)
}
pub fn normalize_optional_value(value: Option<String>) -> Option<String> {
@@ -219,6 +221,21 @@ pub fn normalize_optional_value(value: Option<String>) -> Option<String> {
mod tests {
use super::*;
#[test]
fn generated_ids_with_same_seed_micros_never_collide() {
let seed_micros = 1_713_686_400_000_000;
let first_object_id = generate_asset_object_id(seed_micros);
let second_object_id = generate_asset_object_id(seed_micros);
assert!(first_object_id.starts_with(ASSET_OBJECT_ID_PREFIX));
assert_ne!(first_object_id, second_object_id);
let first_binding_id = generate_asset_binding_id(seed_micros);
let second_binding_id = generate_asset_binding_id(seed_micros);
assert!(first_binding_id.starts_with(ASSET_BINDING_ID_PREFIX));
assert_ne!(first_binding_id, second_binding_id);
}
#[test]
fn validate_asset_object_fields_accepts_minimal_private_object_contract() {
let result = validate_asset_object_fields(
server-rs/crates/shared-kernel/src/lib.rs
+25
View File
@@ -30,6 +30,21 @@ pub fn build_prefixed_seed_id(prefix: &str, seed_micros: i64) -> String {
format!("{prefix}{seed_micros:x}")
}
/// 统一生成“前缀 + 十六进制微秒种子 + UUID simple 随机后缀”的唯一 ID。
/// 纯微秒种子在并发写入时会撞主键，带熵版本保留时间前缀的可读排序，同时保证全局唯一。
#[cfg(not(target_arch = "wasm32"))]
pub fn build_prefixed_seed_entropy_id(prefix: &str, seed_micros: i64) -> String {
format!("{prefix}{seed_micros:x}_{}", Uuid::new_v4().simple())
}
/// SpacetimeDB 的 wasm32 模块不应走浏览器/本地随机 UUID 生成。
#[cfg(target_arch = "wasm32")]
pub fn build_prefixed_seed_entropy_id(_prefix: &str, _seed_micros: i64) -> String {
panic!(
"shared-kernel::build_prefixed_seed_entropy_id 不支持 wasm32，请改用显式 ID 或 SpacetimeDB 上下文生成能力"
)
}
/// 统一生成“前缀 + UUID simple”随机 ID，适合会话态或一次性票据主键。
#[cfg(not(target_arch = "wasm32"))]
pub fn build_prefixed_uuid_id(prefix: &str) -> String {
@@ -124,6 +139,16 @@ mod tests {
assert_eq!(build_prefixed_seed_id("assetobj_", 255), "assetobj_ff");
}
#[test]
fn build_prefixed_seed_entropy_id_keeps_seed_prefix_and_never_collides_on_same_seed() {
let first = build_prefixed_seed_entropy_id("assetobj_", 255);
let second = build_prefixed_seed_entropy_id("assetobj_", 255);
assert!(first.starts_with("assetobj_ff_"));
assert!(second.starts_with("assetobj_ff_"));
assert_ne!(first, second);
}
#[test]
fn format_timestamp_micros_is_stable() {
assert_eq!(
server-rs/crates/spacetime-client/src/external_generation.rs
+121
View File
@@ -8,6 +8,43 @@ const EXTERNAL_GENERATION_QUEUE_WAKE_SUBSCRIPTION_QUERIES: [&str; 2] = [
"SELECT * FROM external_generation_job WHERE status = 'running'",
];
#[derive(Debug)]
pub enum ExternalGenerationJobPhaseUpdateError {
LeaseFencingRejected(String),
Rejected(String),
Rpc(SpacetimeClientError),
}
impl ExternalGenerationJobPhaseUpdateError {
pub fn is_lease_fencing_rejected(&self) -> bool {
matches!(self, Self::LeaseFencingRejected(_))
}
pub fn is_retryable_transport(&self) -> bool {
matches!(
self,
Self::Rpc(
SpacetimeClientError::Build(_)
| SpacetimeClientError::ConnectDropped
| SpacetimeClientError::Timeout(_)
)
)
}
}
impl std::fmt::Display for ExternalGenerationJobPhaseUpdateError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::LeaseFencingRejected(message) | Self::Rejected(message) => {
formatter.write_str(message)
}
Self::Rpc(error) => std::fmt::Display::fmt(error, formatter),
}
}
}
impl std::error::Error for ExternalGenerationJobPhaseUpdateError {}
pub struct ExternalGenerationQueueWakeSubscription {
connection: DbConnection,
_subscriptions: Vec<SubscriptionHandle>,
@@ -257,6 +294,47 @@ impl SpacetimeClient {
.await
}
pub async fn update_external_generation_job_phase(
&self,
input: ExternalGenerationJobPhaseUpdateRecordInput,
) -> Result<ExternalGenerationJobRecord, ExternalGenerationJobPhaseUpdateError> {
let procedure_input = input.into();
let outcome = self
.call_after_connect(
"update_external_generation_job_phase_and_return",
move |connection, sender| {
connection
.procedures()
.update_external_generation_job_phase_and_return_then(
procedure_input,
move |_, result| {
let mapped = result
.map_err(SpacetimeClientError::from_sdk_error)
.map(map_external_generation_job_phase_update_procedure_result);
send_once(&sender, mapped);
},
);
},
)
.await
.map_err(ExternalGenerationJobPhaseUpdateError::Rpc)?;
match outcome {
ExternalGenerationJobPhaseUpdateProcedureOutcome::Updated(job) => Ok(job),
ExternalGenerationJobPhaseUpdateProcedureOutcome::Rejected { kind, message } => {
match kind {
ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected => Err(
ExternalGenerationJobPhaseUpdateError::LeaseFencingRejected(message),
),
ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected => {
Err(ExternalGenerationJobPhaseUpdateError::Rejected(message))
}
}
}
}
}
pub async fn fail_external_generation_job(
&self,
input: ExternalGenerationJobFailRecordInput,
@@ -486,3 +564,46 @@ fn send_external_generation_queue_wake(sender: &watch::Sender<u64>, counter: &At
let next = counter.fetch_add(1, Ordering::Relaxed).saturating_add(1);
let _ = sender.send(next);
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn phase_update_only_retries_transport_errors() {
let build = ExternalGenerationJobPhaseUpdateError::Rpc(SpacetimeClientError::Build(
"initial websocket connect failed".to_string(),
));
assert!(build.is_retryable_transport());
let timeout = ExternalGenerationJobPhaseUpdateError::Rpc(SpacetimeClientError::Timeout(
SpacetimeClientStage::ProcedureResult,
));
assert!(timeout.is_retryable_transport());
assert!(!timeout.is_lease_fencing_rejected());
let disconnected =
ExternalGenerationJobPhaseUpdateError::Rpc(SpacetimeClientError::ConnectDropped);
assert!(disconnected.is_retryable_transport());
let lease = ExternalGenerationJobPhaseUpdateError::LeaseFencingRejected(
"lease token 不匹配".to_string(),
);
assert!(!lease.is_retryable_transport());
assert!(lease.is_lease_fencing_rejected());
let rejected = ExternalGenerationJobPhaseUpdateError::Rejected("phase 非法".to_string());
assert!(!rejected.is_retryable_transport());
assert!(!rejected.is_lease_fencing_rejected());
let procedure = ExternalGenerationJobPhaseUpdateError::Rpc(
SpacetimeClientError::Procedure("procedure rejected".to_string()),
);
assert!(!procedure.is_retryable_transport());
let runtime = ExternalGenerationJobPhaseUpdateError::Rpc(SpacetimeClientError::Runtime(
"runtime invariant failed".to_string(),
));
assert!(!runtime.is_retryable_transport());
}
}
server-rs/crates/spacetime-client/src/lib.rs
+18 -20
View File
@@ -60,20 +60,19 @@ pub use mapper::{
ExternalGenerationJobClaimRecordInput, ExternalGenerationJobCompleteRecordInput,
ExternalGenerationJobEnqueueRecordInput, ExternalGenerationJobFailRecordInput,
ExternalGenerationJobGetRecordInput, ExternalGenerationJobListRecord,
ExternalGenerationJobListRecordInput, ExternalGenerationJobRecord,
ExternalGenerationJobResultRecord,
ExternalGenerationJobRenewLeaseRecordInput, ExternalGenerationJobSummaryListRecord,
ExternalGenerationJobSummaryRecord, ExternalGenerationQueueStatsRecord,
FeatureGateConfigRecord, JumpHopActionRequest, JumpHopActionResponse, JumpHopActionType,
JumpHopCharacterAsset, JumpHopDifficulty, JumpHopDraftResponse, JumpHopGalleryCardResponse,
JumpHopGalleryDetailResponse, JumpHopGalleryResponse, JumpHopGenerationStatus,
JumpHopJumpRequest, JumpHopJumpResponse, JumpHopJumpResult, JumpHopLastJump, JumpHopPath,
JumpHopPlatform, JumpHopRestartRunRequest, JumpHopRunResponse, JumpHopRunStatus,
JumpHopRuntimeRunSnapshotResponse, JumpHopScoring, JumpHopSessionResponse,
JumpHopSessionSnapshotResponse, JumpHopStartRunRequest, JumpHopStylePreset, JumpHopTileAsset,
JumpHopTileType, JumpHopWorkDetailResponse, JumpHopWorkMutationResponse,
JumpHopWorkProfileResponse, JumpHopWorkSummaryResponse, JumpHopWorksResponse,
JumpHopWorkspaceCreateRequest, Match3DAgentMessageFinalizeRecordInput,
ExternalGenerationJobListRecordInput, ExternalGenerationJobPhaseUpdateRecordInput,
ExternalGenerationJobRecord, ExternalGenerationJobRenewLeaseRecordInput,
ExternalGenerationJobSummaryListRecord, ExternalGenerationJobSummaryRecord,
ExternalGenerationQueueStatsRecord, FeatureGateConfigRecord, JumpHopActionRequest,
JumpHopActionResponse, JumpHopActionType, JumpHopCharacterAsset, JumpHopDifficulty,
JumpHopDraftResponse, JumpHopGalleryCardResponse, JumpHopGalleryDetailResponse,
JumpHopGalleryResponse, JumpHopGenerationStatus, JumpHopJumpRequest, JumpHopJumpResponse,
JumpHopJumpResult, JumpHopLastJump, JumpHopPath, JumpHopPlatform, JumpHopRestartRunRequest,
JumpHopRunResponse, JumpHopRunStatus, JumpHopRuntimeRunSnapshotResponse, JumpHopScoring,
JumpHopSessionResponse, JumpHopSessionSnapshotResponse, JumpHopStartRunRequest,
JumpHopStylePreset, JumpHopTileAsset, JumpHopTileType, JumpHopWorkDetailResponse,
JumpHopWorkMutationResponse, JumpHopWorkProfileResponse, JumpHopWorkSummaryResponse,
JumpHopWorksResponse, JumpHopWorkspaceCreateRequest, Match3DAgentMessageFinalizeRecordInput,
Match3DAgentMessageRecord, Match3DAgentMessageSubmitRecordInput,
Match3DAgentSessionCreateRecordInput, Match3DAgentSessionRecord, Match3DAnchorItemRecord,
Match3DAnchorPackRecord, Match3DClickConfirmationRecord, Match3DCompileDraftRecordInput,
@@ -153,7 +152,9 @@ pub mod editor_agent;
pub mod editor_project;
pub mod external_api_key;
pub mod external_generation;
pub use external_generation::ExternalGenerationQueueWakeSubscription;
pub use external_generation::{
ExternalGenerationJobPhaseUpdateError, ExternalGenerationQueueWakeSubscription,
};
pub mod profile_recharge_expiration;
pub use profile_recharge_expiration::ProfileRechargeExpirationSubscription;
@@ -932,14 +933,11 @@ impl SpacetimeClient {
.on_connect(move |_, _, _| {
send_connect_once(&connect_sender, Ok(()));
})
.on_disconnect(move |_, error| {
.on_disconnect(move |_, _error| {
broken_flag.store(true, Ordering::SeqCst);
let message = error
.map(|error| error.to_string())
.unwrap_or_else(|| "SpacetimeDB 连接已断开".to_string());
send_connect_once(
&disconnect_sender,
Err(SpacetimeClientError::Procedure(message)),
Err(SpacetimeClientError::ConnectDropped),
);
})
.build()
server-rs/crates/spacetime-client/src/mapper.rs
+7 -4
View File
@@ -118,10 +118,13 @@ pub use self::external_generation::{
ExternalGenerationJobCompleteRecordInput, ExternalGenerationJobEnqueueRecordInput,
ExternalGenerationJobFailRecordInput, ExternalGenerationJobGetRecordInput,
ExternalGenerationJobListRecord, ExternalGenerationJobListRecordInput,
ExternalGenerationJobRecord, ExternalGenerationJobResultRecord,
ExternalGenerationJobRenewLeaseRecordInput,
ExternalGenerationJobSummaryListRecord, ExternalGenerationJobSummaryRecord,
ExternalGenerationQueueStatsRecord,
ExternalGenerationJobPhaseUpdateRecordInput, ExternalGenerationJobRecord,
ExternalGenerationJobRenewLeaseRecordInput, ExternalGenerationJobSummaryListRecord,
ExternalGenerationJobSummaryRecord, ExternalGenerationQueueStatsRecord,
};
pub(crate) use self::external_generation::{
ExternalGenerationJobPhaseUpdateProcedureOutcome,
map_external_generation_job_phase_update_procedure_result,
};
pub use self::jump_hop::{
JumpHopActionRequest, JumpHopActionResponse, JumpHopActionType, JumpHopCharacterAsset,
server-rs/crates/spacetime-client/src/mapper/external_generation.rs
+93 -13
View File
@@ -54,6 +54,17 @@ impl From<ExternalGenerationJobRenewLeaseRecordInput> for ExternalGenerationJobR
}
}
impl From<ExternalGenerationJobPhaseUpdateRecordInput> for ExternalGenerationJobPhaseUpdateInput {
fn from(input: ExternalGenerationJobPhaseUpdateRecordInput) -> Self {
Self {
job_id: input.job_id,
worker_id: input.worker_id,
lease_token: input.lease_token,
phase: input.phase,
}
}
}
impl From<ExternalGenerationJobFailRecordInput> for ExternalGenerationJobFailInput {
fn from(input: ExternalGenerationJobFailRecordInput) -> Self {
Self {
@@ -112,22 +123,37 @@ pub(crate) fn map_external_generation_job_procedure_result(
Ok(map_external_generation_job_snapshot(job))
}
pub(crate) fn map_external_generation_job_result_procedure_result(
result: ExternalGenerationJobResultProcedureResult,
) -> Result<ExternalGenerationJobResultRecord, SpacetimeClientError> {
pub(crate) enum ExternalGenerationJobPhaseUpdateProcedureOutcome {
Updated(ExternalGenerationJobRecord),
Rejected {
kind: ExternalGenerationJobPhaseUpdateFailureKind,
message: String,
},
}
pub(crate) fn map_external_generation_job_phase_update_procedure_result(
result: ExternalGenerationJobPhaseUpdateProcedureResult,
) -> ExternalGenerationJobPhaseUpdateProcedureOutcome {
if !result.ok {
return Err(SpacetimeClientError::procedure_failed(result.error_message));
return ExternalGenerationJobPhaseUpdateProcedureOutcome::Rejected {
kind: result
.failure_kind
.unwrap_or(ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected),
message: result
.error_message
.unwrap_or_else(|| "SpacetimeDB phase update procedure 返回未知拒绝".to_string()),
};
}
let result = result.result.ok_or_else(|| {
SpacetimeClientError::missing_snapshot("external_generation_job 结果快照")
})?;
Ok(ExternalGenerationJobResultRecord {
job_id: result.job_id,
status: result.status,
last_error_message: result.last_error_message,
result_payload_json: result.result_payload_json,
})
match result.job {
Some(job) => ExternalGenerationJobPhaseUpdateProcedureOutcome::Updated(
map_external_generation_job_snapshot(job),
),
None => ExternalGenerationJobPhaseUpdateProcedureOutcome::Rejected {
kind: ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected,
message: "SpacetimeDB phase update procedure 未返回任务快照".to_string(),
},
}
}
pub(crate) fn map_external_generation_job_claim_result(
@@ -255,6 +281,7 @@ fn map_external_generation_job_snapshot(
.notification_acknowledged_at_micros
.map(format_timestamp_micros),
notification_acknowledged_at_micros: snapshot.notification_acknowledged_at_micros,
phase: snapshot.phase,
}
}
@@ -282,6 +309,7 @@ fn map_external_generation_job_summary_snapshot(
.notification_acknowledged_at_micros
.map(format_timestamp_micros),
notification_acknowledged_at_micros: snapshot.notification_acknowledged_at_micros,
phase: snapshot.phase,
warning_message: snapshot.warning_message,
}
}
@@ -328,6 +356,14 @@ pub struct ExternalGenerationJobRenewLeaseRecordInput {
pub renewed_at_micros: i64,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExternalGenerationJobPhaseUpdateRecordInput {
pub job_id: String,
pub worker_id: String,
pub lease_token: String,
pub phase: String,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct ExternalGenerationJobFailRecordInput {
pub job_id: String,
@@ -388,6 +424,7 @@ pub struct ExternalGenerationJobRecord {
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub phase: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
@@ -427,6 +464,7 @@ pub struct ExternalGenerationJobSummaryRecord {
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub phase: Option<String>,
pub warning_message: Option<String>,
}
@@ -451,3 +489,45 @@ pub struct ExternalGenerationQueueStatsRecord {
pub oldest_claimable_age_micros: Option<i64>,
pub now_micros: i64,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn phase_update_mapper_keeps_structured_rejection_kind() {
let lease_rejection = map_external_generation_job_phase_update_procedure_result(
ExternalGenerationJobPhaseUpdateProcedureResult {
ok: false,
job: None,
failure_kind: Some(
ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected,
),
error_message: Some("lease 已过期".to_string()),
},
);
assert!(matches!(
lease_rejection,
ExternalGenerationJobPhaseUpdateProcedureOutcome::Rejected {
kind: ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected,
..
}
));
let other_rejection = map_external_generation_job_phase_update_procedure_result(
ExternalGenerationJobPhaseUpdateProcedureResult {
ok: false,
job: None,
failure_kind: Some(ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected),
error_message: Some("phase 非法".to_string()),
},
);
assert!(matches!(
other_rejection,
ExternalGenerationJobPhaseUpdateProcedureOutcome::Rejected {
kind: ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected,
..
}
));
}
}
server-rs/crates/spacetime-client/src/module_bindings.rs
+8
View File
@@ -516,6 +516,9 @@ pub mod external_generation_job_get_input_type;
pub mod external_generation_job_list_input_type;
pub mod external_generation_job_payload_compaction_input_type;
pub mod external_generation_job_payload_compaction_procedure_result_type;
pub mod external_generation_job_phase_update_failure_kind_type;
pub mod external_generation_job_phase_update_input_type;
pub mod external_generation_job_phase_update_procedure_result_type;
pub mod external_generation_job_procedure_result_type;
pub mod external_generation_job_result_procedure_result_type;
pub mod external_generation_job_renew_lease_input_type;
@@ -1329,6 +1332,7 @@ pub mod update_editor_asset_and_return_procedure;
pub mod update_editor_asset_folder_and_return_procedure;
pub mod update_editor_project_resource_showcase_and_return_procedure;
pub mod update_editor_showcase_asset_display_and_return_procedure;
pub mod update_external_generation_job_phase_and_return_procedure;
pub mod update_jump_hop_work_procedure;
pub mod update_match_3_d_work_procedure;
pub mod update_puzzle_clear_work_procedure;
@@ -1951,6 +1955,9 @@ pub use external_generation_job_get_input_type::ExternalGenerationJobGetInput;
pub use external_generation_job_list_input_type::ExternalGenerationJobListInput;
pub use external_generation_job_payload_compaction_input_type::ExternalGenerationJobPayloadCompactionInput;
pub use external_generation_job_payload_compaction_procedure_result_type::ExternalGenerationJobPayloadCompactionProcedureResult;
pub use external_generation_job_phase_update_failure_kind_type::ExternalGenerationJobPhaseUpdateFailureKind;
pub use external_generation_job_phase_update_input_type::ExternalGenerationJobPhaseUpdateInput;
pub use external_generation_job_phase_update_procedure_result_type::ExternalGenerationJobPhaseUpdateProcedureResult;
pub use external_generation_job_procedure_result_type::ExternalGenerationJobProcedureResult;
pub use external_generation_job_result_procedure_result_type::ExternalGenerationJobResultProcedureResult;
pub use external_generation_job_renew_lease_input_type::ExternalGenerationJobRenewLeaseInput;
@@ -2764,6 +2771,7 @@ pub use update_editor_asset_and_return_procedure::update_editor_asset_and_return
pub use update_editor_asset_folder_and_return_procedure::update_editor_asset_folder_and_return;
pub use update_editor_project_resource_showcase_and_return_procedure::update_editor_project_resource_showcase_and_return;
pub use update_editor_showcase_asset_display_and_return_procedure::update_editor_showcase_asset_display_and_return;
pub use update_external_generation_job_phase_and_return_procedure::update_external_generation_job_phase_and_return;
pub use update_jump_hop_work_procedure::update_jump_hop_work;
pub use update_match_3_d_work_procedure::update_match_3_d_work;
pub use update_puzzle_clear_work_procedure::update_puzzle_clear_work;
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_phase_update_failure_kind_type.rs
+18
View File
@@ -0,0 +1,18 @@
// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE
// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD.
#![allow(unused, clippy::all)]
use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws};
#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)]
#[sats(crate = __lib)]
#[derive(Copy, Eq, Hash)]
pub enum ExternalGenerationJobPhaseUpdateFailureKind {
LeaseFencingRejected,
OtherRejected,
}
impl __sdk::InModule for ExternalGenerationJobPhaseUpdateFailureKind {
type Module = super::RemoteModule;
}
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_phase_update_input_type.rs
+18
View File
@@ -0,0 +1,18 @@
// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE
// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD.
#![allow(unused, clippy::all)]
use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws};
#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)]
#[sats(crate = __lib)]
pub struct ExternalGenerationJobPhaseUpdateInput {
pub job_id: String,
pub worker_id: String,
pub lease_token: String,
pub phase: String,
}
impl __sdk::InModule for ExternalGenerationJobPhaseUpdateInput {
type Module = super::RemoteModule;
}
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_phase_update_procedure_result_type.rs
+21
View File
@@ -0,0 +1,21 @@
// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE
// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD.
#![allow(unused, clippy::all)]
use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws};
use super::external_generation_job_phase_update_failure_kind_type::ExternalGenerationJobPhaseUpdateFailureKind;
use super::external_generation_job_snapshot_type::ExternalGenerationJobSnapshot;
#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)]
#[sats(crate = __lib)]
pub struct ExternalGenerationJobPhaseUpdateProcedureResult {
pub ok: bool,
pub job: Option<ExternalGenerationJobSnapshot>,
pub failure_kind: Option<ExternalGenerationJobPhaseUpdateFailureKind>,
pub error_message: Option<String>,
}
impl __sdk::InModule for ExternalGenerationJobPhaseUpdateProcedureResult {
type Module = super::RemoteModule;
}
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_snapshot_type.rs
+1
View File
@@ -31,6 +31,7 @@ pub struct ExternalGenerationJobSnapshot {
pub price_mud_points: u64,
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub phase: Option<String>,
}
impl __sdk::InModule for ExternalGenerationJobSnapshot {
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_summary_snapshot_type.rs
+1
View File
@@ -24,6 +24,7 @@ pub struct ExternalGenerationJobSummarySnapshot {
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub warning_message: Option<String>,
pub phase: Option<String>,
}
impl __sdk::InModule for ExternalGenerationJobSummarySnapshot {
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_summary_type.rs
+3
View File
@@ -24,6 +24,7 @@ pub struct ExternalGenerationJobSummary {
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at: Option<__sdk::Timestamp>,
pub warning_message: Option<String>,
pub phase: Option<String>,
}
impl __sdk::InModule for ExternalGenerationJobSummary {
@@ -55,6 +56,7 @@ pub struct ExternalGenerationJobSummaryCols {
pub notification_acknowledged_at:
__sdk::__query_builder::Col<ExternalGenerationJobSummary, Option<__sdk::Timestamp>>,
pub warning_message: __sdk::__query_builder::Col<ExternalGenerationJobSummary, Option<String>>,
pub phase: __sdk::__query_builder::Col<ExternalGenerationJobSummary, Option<String>>,
}
impl __sdk::__query_builder::HasCols for ExternalGenerationJobSummary {
@@ -81,6 +83,7 @@ impl __sdk::__query_builder::HasCols for ExternalGenerationJobSummary {
"notification_acknowledged_at",
),
warning_message: __sdk::__query_builder::Col::new(table_name, "warning_message"),
phase: __sdk::__query_builder::Col::new(table_name, "phase"),
}
}
}
server-rs/crates/spacetime-client/src/module_bindings/external_generation_job_type.rs
+3
View File
@@ -31,6 +31,7 @@ pub struct ExternalGenerationJob {
pub price_mud_points: u64,
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at: Option<__sdk::Timestamp>,
pub phase: Option<String>,
}
impl __sdk::InModule for ExternalGenerationJob {
@@ -67,6 +68,7 @@ pub struct ExternalGenerationJobCols {
pub refund_ledger_id: __sdk::__query_builder::Col<ExternalGenerationJob, Option<String>>,
pub notification_acknowledged_at:
__sdk::__query_builder::Col<ExternalGenerationJob, Option<__sdk::Timestamp>>,
pub phase: __sdk::__query_builder::Col<ExternalGenerationJob, Option<String>>,
}
impl __sdk::__query_builder::HasCols for ExternalGenerationJob {
@@ -106,6 +108,7 @@ impl __sdk::__query_builder::HasCols for ExternalGenerationJob {
table_name,
"notification_acknowledged_at",
),
phase: __sdk::__query_builder::Col::new(table_name, "phase"),
}
}
}
server-rs/crates/spacetime-client/src/module_bindings/update_external_generation_job_phase_and_return_procedure.rs
+62
View File
@@ -0,0 +1,62 @@
// THIS FILE IS AUTOMATICALLY GENERATED BY SPACETIMEDB. EDITS TO THIS FILE
// WILL NOT BE SAVED. MODIFY TABLES IN YOUR MODULE SOURCE CODE INSTEAD.
#![allow(unused, clippy::all)]
use spacetimedb_sdk::__codegen::{self as __sdk, __lib, __sats, __ws};
use super::external_generation_job_phase_update_input_type::ExternalGenerationJobPhaseUpdateInput;
use super::external_generation_job_phase_update_procedure_result_type::ExternalGenerationJobPhaseUpdateProcedureResult;
#[derive(__lib::ser::Serialize, __lib::de::Deserialize, Clone, PartialEq, Debug)]
#[sats(crate = __lib)]
struct UpdateExternalGenerationJobPhaseAndReturnArgs {
pub input: ExternalGenerationJobPhaseUpdateInput,
}
impl __sdk::InModule for UpdateExternalGenerationJobPhaseAndReturnArgs {
type Module = super::RemoteModule;
}
#[allow(non_camel_case_types)]
/// Extension trait for access to the procedure `update_external_generation_job_phase_and_return`.
///
/// Implemented for [`super::RemoteProcedures`].
pub trait update_external_generation_job_phase_and_return {
fn update_external_generation_job_phase_and_return(
&self,
input: ExternalGenerationJobPhaseUpdateInput,
) {
self.update_external_generation_job_phase_and_return_then(input, |_, _| {});
}
fn update_external_generation_job_phase_and_return_then(
&self,
input: ExternalGenerationJobPhaseUpdateInput,
__callback: impl FnOnce(
&super::ProcedureEventContext,
Result<ExternalGenerationJobPhaseUpdateProcedureResult, __sdk::InternalError>,
) + Send
+ 'static,
);
}
impl update_external_generation_job_phase_and_return for super::RemoteProcedures {
fn update_external_generation_job_phase_and_return_then(
&self,
input: ExternalGenerationJobPhaseUpdateInput,
__callback: impl FnOnce(
&super::ProcedureEventContext,
Result<ExternalGenerationJobPhaseUpdateProcedureResult, __sdk::InternalError>,
) + Send
+ 'static,
) {
self.imp
.invoke_procedure_with_callback::<_, ExternalGenerationJobPhaseUpdateProcedureResult>(
"update_external_generation_job_phase_and_return",
UpdateExternalGenerationJobPhaseAndReturnArgs { input },
__callback,
);
}
}
server-rs/crates/spacetime-module/src/external_generation.rs
+254 -10
View File
@@ -7,6 +7,8 @@ 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";
const EXTERNAL_GENERATION_PHASE_GENERATING: &str = "generating";
const EXTERNAL_GENERATION_PHASE_PROCESSING: &str = "processing";
const EXTERNAL_GENERATION_EVENT_ENQUEUED: &str = "enqueued";
const EXTERNAL_GENERATION_EVENT_CLAIMED: &str = "claimed";
const EXTERNAL_GENERATION_EVENT_LEASE_RENEWED: &str = "lease_renewed";
@@ -85,6 +87,8 @@ pub struct ExternalGenerationJob {
pub(crate) refund_ledger_id: Option<String>,
#[default(None::<Timestamp>)]
pub(crate) notification_acknowledged_at: Option<Timestamp>,
#[default(None::<String>)]
pub(crate) phase: Option<String>,
}
#[spacetimedb::table(
@@ -139,6 +143,8 @@ pub struct ExternalGenerationJobSummary {
pub(crate) notification_acknowledged_at: Option<Timestamp>,
#[default(None::<String>)]
pub(crate) warning_message: Option<String>,
#[default(None::<String>)]
pub(crate) phase: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
@@ -174,6 +180,20 @@ pub struct ExternalGenerationJobRenewLeaseInput {
pub renewed_at_micros: i64,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobPhaseUpdateInput {
pub job_id: String,
pub worker_id: String,
pub lease_token: String,
pub phase: String,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, SpacetimeType)]
pub enum ExternalGenerationJobPhaseUpdateFailureKind {
LeaseFencingRejected,
OtherRejected,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobCompleteInput {
pub job_id: String,
@@ -257,6 +277,7 @@ pub struct ExternalGenerationJobSnapshot {
pub price_mud_points: u64,
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub phase: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
@@ -287,6 +308,14 @@ pub struct ExternalGenerationJobResultProcedureResult {
pub error_message: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobPhaseUpdateProcedureResult {
pub ok: bool,
pub job: Option<ExternalGenerationJobSnapshot>,
pub failure_kind: Option<ExternalGenerationJobPhaseUpdateFailureKind>,
pub error_message: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
pub struct ExternalGenerationJobSummarySnapshot {
pub job_id: String,
@@ -306,6 +335,7 @@ pub struct ExternalGenerationJobSummarySnapshot {
pub refund_ledger_id: Option<String>,
pub notification_acknowledged_at_micros: Option<i64>,
pub warning_message: Option<String>,
pub phase: Option<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, SpacetimeType)]
@@ -446,6 +476,34 @@ pub fn renew_external_generation_job_lease_and_return(
}
}
#[spacetimedb::procedure]
pub fn update_external_generation_job_phase_and_return(
ctx: &mut ProcedureContext,
input: ExternalGenerationJobPhaseUpdateInput,
) -> ExternalGenerationJobPhaseUpdateProcedureResult {
let caller = ctx.sender();
match ctx.try_with_tx(|tx| {
crate::editor_project_storage::require_editor_generation_runtime_service_identity(
tx, caller,
)
.map_err(ExternalGenerationJobPhaseUpdateError::other)?;
update_external_generation_job_phase_tx(tx, input.clone())
}) {
Ok(job) => ExternalGenerationJobPhaseUpdateProcedureResult {
ok: true,
job: Some(job),
failure_kind: None,
error_message: None,
},
Err(error) => ExternalGenerationJobPhaseUpdateProcedureResult {
ok: false,
job: None,
failure_kind: Some(error.kind),
error_message: Some(error.message),
},
}
}
#[spacetimedb::procedure]
pub fn fail_external_generation_job_and_return(
ctx: &mut ProcedureContext,
@@ -717,6 +775,7 @@ fn enqueue_external_generation_job_tx(
price_mud_points: input.price_mud_points,
refund_ledger_id: None,
notification_acknowledged_at: None,
phase: None,
};
persist_external_generation_job_row(ctx, row.clone());
insert_external_generation_job_event(
@@ -791,6 +850,7 @@ fn claim_external_generation_jobs_tx(
claim_time,
);
row.status = EXTERNAL_GENERATION_STATUS_RUNNING.to_string();
row.phase = Some(EXTERNAL_GENERATION_PHASE_GENERATING.to_string());
row.worker_id = Some(worker_id.clone());
row.lease_expires_at = Some(lease_expires_at);
row.lease_token = Some(lease_token);
@@ -1336,6 +1396,24 @@ fn renew_external_generation_job_lease_tx(
Ok(map_external_generation_job_row(row))
}
fn update_external_generation_job_phase_tx(
ctx: &ReducerContext,
input: ExternalGenerationJobPhaseUpdateInput,
) -> Result<ExternalGenerationJobSnapshot, ExternalGenerationJobPhaseUpdateError> {
let phase = normalize_external_generation_job_phase(&input.phase)
.map_err(ExternalGenerationJobPhaseUpdateError::other)?;
let mut row = get_worker_owned_external_generation_job_for_phase_update(
ctx,
&input.job_id,
&input.worker_id,
&input.lease_token,
)?;
row.phase = Some(phase);
row.updated_at = ctx.timestamp;
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,
@@ -1475,28 +1553,96 @@ fn get_worker_owned_external_generation_job(
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)?;
get_worker_owned_external_generation_job_for_phase_update(ctx, job_id, worker_id, lease_token)
.map_err(|error| error.message)
}
#[derive(Debug)]
struct ExternalGenerationJobPhaseUpdateError {
kind: ExternalGenerationJobPhaseUpdateFailureKind,
message: String,
}
impl ExternalGenerationJobPhaseUpdateError {
fn lease_fencing(message: impl Into<String>) -> Self {
Self {
kind: ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected,
message: message.into(),
}
}
fn other(message: impl Into<String>) -> Self {
Self {
kind: ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected,
message: message.into(),
}
}
}
impl std::fmt::Display for ExternalGenerationJobPhaseUpdateError {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
formatter.write_str(self.message.as_str())
}
}
impl std::error::Error for ExternalGenerationJobPhaseUpdateError {}
fn get_worker_owned_external_generation_job_for_phase_update(
ctx: &ReducerContext,
job_id: &str,
worker_id: &str,
lease_token: &str,
) -> Result<ExternalGenerationJob, ExternalGenerationJobPhaseUpdateError> {
validate_required("external_generation_job.job_id", job_id)
.map_err(ExternalGenerationJobPhaseUpdateError::other)?;
validate_required("external_generation_job.worker_id", worker_id)
.map_err(ExternalGenerationJobPhaseUpdateError::other)?;
validate_required("external_generation_job.lease_token", lease_token)
.map_err(ExternalGenerationJobPhaseUpdateError::other)?;
let row = ctx
.db
.external_generation_job()
.job_id()
.find(&job_id.trim().to_string())
.ok_or_else(|| "external_generation_job 不存在".to_string())?;
.ok_or_else(|| {
ExternalGenerationJobPhaseUpdateError::lease_fencing("external_generation_job 不存在")
})?;
validate_external_generation_job_phase_update_lease(
&row,
worker_id,
lease_token,
ctx.timestamp,
)?;
Ok(row)
}
fn validate_external_generation_job_phase_update_lease(
row: &ExternalGenerationJob,
worker_id: &str,
lease_token: &str,
now: Timestamp,
) -> Result<(), ExternalGenerationJobPhaseUpdateError> {
if row.status != EXTERNAL_GENERATION_STATUS_RUNNING {
return Err("external_generation_job 当前不是 running 状态".to_string());
return Err(ExternalGenerationJobPhaseUpdateError::lease_fencing(
"external_generation_job 当前不是 running 状态",
));
}
if !is_external_generation_job_owned_by_worker(&row, worker_id) {
return Err("external_generation_job worker lease 不匹配".to_string());
return Err(ExternalGenerationJobPhaseUpdateError::lease_fencing(
"external_generation_job worker lease 不匹配",
));
}
if !is_external_generation_job_owned_by_lease_token(&row, lease_token) {
return Err("external_generation_job lease token 不匹配".to_string());
return Err(ExternalGenerationJobPhaseUpdateError::lease_fencing(
"external_generation_job lease token 不匹配",
));
}
if !is_external_generation_job_lease_active(&row, ctx.timestamp) {
return Err("external_generation_job lease 已过期".to_string());
if !is_external_generation_job_lease_active(row, now) {
return Err(ExternalGenerationJobPhaseUpdateError::lease_fencing(
"external_generation_job lease 已过期",
));
}
Ok(row)
Ok(())
}
fn is_external_generation_job_owned_by_worker(
@@ -1740,6 +1886,18 @@ fn normalize_external_generation_job_status_filter(statuses: &[String]) -> Vec<&
.collect()
}
fn normalize_external_generation_job_phase(phase: &str) -> Result<String, String> {
match phase.trim() {
EXTERNAL_GENERATION_PHASE_GENERATING => {
Ok(EXTERNAL_GENERATION_PHASE_GENERATING.to_string())
}
EXTERNAL_GENERATION_PHASE_PROCESSING => {
Ok(EXTERNAL_GENERATION_PHASE_PROCESSING.to_string())
}
_ => Err("external_generation_job.phase 只支持 generating 或 processing".to_string()),
}
}
fn record_external_generation_claimable_age(
stats: &mut ExternalGenerationQueueStatsSnapshot,
row: &ExternalGenerationJob,
@@ -1831,6 +1989,7 @@ fn build_external_generation_job_summary_row(
warning_message: extract_external_generation_warning_message(
row.result_payload_json.as_deref(),
),
phase: row.phase.clone(),
}
}
@@ -2077,6 +2236,7 @@ fn map_external_generation_job_row(row: ExternalGenerationJob) -> ExternalGenera
price_mud_points: row.price_mud_points,
refund_ledger_id: row.refund_ledger_id,
notification_acknowledged_at_micros,
phase: row.phase,
}
}
@@ -2107,6 +2267,7 @@ fn map_external_generation_job_summary_row(
.notification_acknowledged_at
.map(|value| value.to_micros_since_unix_epoch()),
warning_message: row.warning_message,
phase: row.phase,
}
}
@@ -2143,6 +2304,7 @@ fn map_external_generation_job_summary_to_compat_snapshot(
price_mud_points: summary.price_mud_points,
refund_ledger_id: summary.refund_ledger_id,
notification_acknowledged_at_micros: summary.notification_acknowledged_at_micros,
phase: summary.phase,
}
}
@@ -2490,6 +2652,87 @@ fn normalize_optional_text(value: &str) -> Option<String> {
mod tests {
use super::*;
#[test]
fn external_generation_phase_only_accepts_known_execution_phases() {
assert_eq!(
normalize_external_generation_job_phase(" processing ").as_deref(),
Ok(EXTERNAL_GENERATION_PHASE_PROCESSING)
);
assert!(normalize_external_generation_job_phase("uploading").is_err());
}
#[test]
fn external_generation_phase_rejection_kind_is_machine_readable() {
let lease = ExternalGenerationJobPhaseUpdateError::lease_fencing("lease 已过期");
assert_eq!(
lease.kind,
ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected
);
assert_eq!(lease.message, "lease 已过期");
let other = ExternalGenerationJobPhaseUpdateError::other("phase 非法");
assert_eq!(
other.kind,
ExternalGenerationJobPhaseUpdateFailureKind::OtherRejected
);
assert_eq!(other.message, "phase 非法");
}
#[test]
fn external_generation_phase_lease_guard_classifies_every_fencing_rejection() {
let mut row = external_generation_job_fixture(EXTERNAL_GENERATION_STATUS_RUNNING);
row.worker_id = Some("worker-a".to_string());
row.lease_token = Some("lease-1".to_string());
row.lease_expires_at = Some(micros(2_000));
assert!(
validate_external_generation_job_phase_update_lease(
&row,
"worker-a",
"lease-1",
micros(1_999),
)
.is_ok()
);
let mut terminal = row.clone();
terminal.status = EXTERNAL_GENERATION_STATUS_COMPLETED.to_string();
let cases = [
validate_external_generation_job_phase_update_lease(
&terminal,
"worker-a",
"lease-1",
micros(1_999),
),
validate_external_generation_job_phase_update_lease(
&row,
"worker-b",
"lease-1",
micros(1_999),
),
validate_external_generation_job_phase_update_lease(
&row,
"worker-a",
"lease-2",
micros(1_999),
),
validate_external_generation_job_phase_update_lease(
&row,
"worker-a",
"lease-1",
micros(2_000),
),
];
for result in cases {
let error = result.expect_err("stale worker 必须被 fencing 拒绝");
assert_eq!(
error.kind,
ExternalGenerationJobPhaseUpdateFailureKind::LeaseFencingRejected
);
}
}
#[test]
fn external_generation_job_result_failure_is_structured() {
let result = failed_external_generation_job_result("失败".to_string());
@@ -3029,6 +3272,7 @@ mod tests {
price_mud_points: 10,
refund_ledger_id: None,
notification_acknowledged_at: None,
phase: None,
}
}
server-rs/crates/spacetime-module/src/migration.rs
+9
View File
@@ -1376,6 +1376,15 @@ fn normalize_migration_row(table_name: &str, value: &serde_json::Value) -> serde
.or_insert(serde_json::Value::Null);
}
}
if table_name == "external_generation_job" || table_name == "external_generation_job_summary" {
if let Some(object) = next_value.as_object_mut() {
// 中文注释：执行阶段晚于外部生成主表和摘要投影加入，旧迁移包按未知阶段兼容；
// BFF 会把 running + phase=null 视为 generating。
object
.entry("phase".to_string())
.or_insert(serde_json::Value::Null);
}
}
if table_name == "big_fish_creation_session" {
if let Some(object) = next_value.as_object_mut() {
// 中文注释：旧迁移包没有公开游玩次数字段，导入时按新建作品默认 0 兼容。