From 3d47f50a5b746321d67aa21f86ab4059a3d01a86 Mon Sep 17 00:00:00 2001 From: Tom Tang Date: Wed, 13 May 2026 11:58:32 +0800 Subject: [PATCH] feat(ingestion): surface schema-service 409 as a typed conflict error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When the schema service refuses a proposed schema because its `descriptive_name` already maps to a different Approved canonical (the new 409 introduced upstream in schema_service PR #145 — kanban #49603), the message needs to reach the user as "There's already a schema called 'Photography' on this node. Rename your new schema or reuse the existing one." — not as "Schema creation error: Failed to create schema via schema service: schema service refused duplicate descriptive_name …" wrapped twice. Adds `IngestionError::SchemaDescriptiveNameConflict { descriptive_name, existing_canonical, reason }` with a `user_message()` arm that produces the rename / reuse copy. `schema_creation::create_new_schema_with_node` detects the conflict via the message sentinel emitted by `schema_service_client::add_schema` and converts it to the typed variant before propagating. The detection is string-based on purpose: the fold_db_node pinned rev of schema_service doesn't have the new error message yet (the bump cascade lands it on its 2h schedule). On the current rev, the detection is a no-op — the legacy "Schema service returned unexpected CONFLICT (409)" message doesn't match — and the existing `IngestionError::SchemaCreationError` path stays. Once the cascade bumps the rev, the new message text starts arriving and the typed variant fires automatically. A follow-up PR will swap to the typed `AddSchemaOutcome::Conflict` enum once that's available. Tests: - `parse_schema_name_conflict_recognises_the_canonical_message` pins the sentinel + parsing against the exact message shape `add_schema` produces. - `parse_schema_name_conflict_ignores_unrelated_errors` keeps transport / 5xx errors flowing through the existing path. - `schema_descriptive_name_conflict_user_message_is_actionable` pins the UI-facing copy. Filed kanban follow-ups: prod-cleanup (separate scope; needs deliberate data preservation plan) and concurrent-Lambda race prevention (needs backend-level atomicity — DynamoDB conditional writes / per-name lease). Co-Authored-By: Claude Opus 4.7 (1M context) --- src/ingestion/error.rs | 133 ++++++++++++++++++ .../ingestion_service/schema_creation.rs | 24 +++- 2 files changed, 151 insertions(+), 6 deletions(-) diff --git a/src/ingestion/error.rs b/src/ingestion/error.rs index 2bd7d915..cf144987 100644 --- a/src/ingestion/error.rs +++ b/src/ingestion/error.rs @@ -60,6 +60,21 @@ pub enum IngestionError { /// Connection errors (cannot reach the AI service) #[error("{provider} connection error: {message}")] ConnectionError { provider: String, message: String }, + + /// The schema service refused the proposal because another Approved + /// schema with the same `descriptive_name` already exists and the two + /// can't be cleanly merged (e.g. cross-`schema_type`). Carries the + /// existing canonical hash so the UI / CLI can prompt the user to + /// rename or reuse the existing schema instead of growing a duplicate. + #[error( + "schema service refused duplicate descriptive_name '{descriptive_name}' \ + (existing canonical: {existing_canonical}): {reason}" + )] + SchemaDescriptiveNameConflict { + descriptive_name: String, + existing_canonical: String, + reason: String, + }, } impl IngestionError { @@ -113,11 +128,71 @@ impl IngestionError { Self::InvalidInput(msg) => { format!("Invalid input: {}", msg) } + Self::SchemaDescriptiveNameConflict { + descriptive_name, .. + } => { + format!( + "There's already a schema called '{}' on this node. \ + Rename your new schema or reuse the existing one — \ + two schemas can't share a descriptive name.", + descriptive_name, + ) + } _ => self.to_string(), } } } +/// Sentinel prefix the schema service emits in its 409 message body. Stable +/// across versions — `schema_service_client::add_schema` lifts the +/// `DescriptiveNameConflict` fields into a message string starting with this +/// phrase. We string-match on it so this code keeps working through the +/// schema_service rev bump that introduces the typed `add_schema_typed` +/// path; a follow-up will swap to the typed enum once the rev cascades. +pub(crate) const SCHEMA_NAME_CONFLICT_SENTINEL: &str = + "schema service refused duplicate descriptive_name"; + +/// Inspect the `Display` form of a `FoldDbError` returned by +/// `SchemaServiceClient::add_schema` and, if it matches the +/// schema-service 409 message, parse out the typed conflict fields. +/// Returns `None` if the error is something else (transport failure, +/// 5xx, etc). +pub(crate) fn parse_schema_name_conflict(err_display: &str) -> Option { + if !err_display.contains(SCHEMA_NAME_CONFLICT_SENTINEL) { + return None; + } + // Message shape (see schema_service_client::add_schema): + // "schema service refused duplicate descriptive_name '': \ + // an Approved schema with identity_hash '' is already registered. \ + // Reason: . Rename the new schema or reuse the existing one." + let descriptive_name = single_quoted_after(err_display, SCHEMA_NAME_CONFLICT_SENTINEL)?; + let existing_canonical = single_quoted_after(err_display, "identity_hash")?; + let reason = err_display + .find("Reason: ") + .map(|i| { + let tail = &err_display[i + "Reason: ".len()..]; + tail.split_once(". Rename") + .map(|(r, _)| r.to_string()) + .unwrap_or_else(|| tail.to_string()) + }) + .unwrap_or_else(|| "(reason missing)".to_string()); + Some(IngestionError::SchemaDescriptiveNameConflict { + descriptive_name, + existing_canonical, + reason, + }) +} + +/// Pull the first `'...'`-quoted substring that appears after `marker` in +/// `haystack`. Returns `None` if either the marker or a single-quoted +/// region after it is missing. +fn single_quoted_after(haystack: &str, marker: &str) -> Option { + let after = haystack.split_once(marker).map(|(_, rest)| rest)?; + let start = after.find('\'')? + 1; + let end = after[start..].find('\'')?; + Some(after[start..start + end].to_string()) +} + /// Classify an HTTP error response from an LLM provider into a specific error variant. pub fn classify_llm_error(provider: &str, status_code: u16, body: &str) -> IngestionError { match status_code { @@ -269,6 +344,64 @@ mod tests { assert!(schema.user_message().contains("parse fail")); } + #[test] + fn parse_schema_name_conflict_recognises_the_canonical_message() { + // Exact shape emitted by schema_service_client::add_schema. Keep in + // sync with that crate's wording — `SCHEMA_NAME_CONFLICT_SENTINEL` + // pins the prefix. + let msg = "Configuration error: schema service refused duplicate \ + descriptive_name 'Photography': an Approved schema with \ + identity_hash 'abc123def456' is already registered. \ + Reason: incoming schema_type HashRange differs from \ + existing Hash; cross-schema_type expansion would corrupt \ + molecule reads. Rename the new schema or reuse the \ + existing one."; + let parsed = parse_schema_name_conflict(msg).expect("should parse"); + match parsed { + IngestionError::SchemaDescriptiveNameConflict { + descriptive_name, + existing_canonical, + reason, + } => { + assert_eq!(descriptive_name, "Photography"); + assert_eq!(existing_canonical, "abc123def456"); + assert!(reason.contains("schema_type")); + assert!(!reason.contains("Rename")); + } + other => panic!("expected SchemaDescriptiveNameConflict, got {:?}", other), + } + } + + #[test] + fn parse_schema_name_conflict_ignores_unrelated_errors() { + assert!(parse_schema_name_conflict("transport timeout").is_none()); + assert!(parse_schema_name_conflict( + "Schema service add schema failed with status 500: oops" + ) + .is_none()); + assert!(parse_schema_name_conflict("").is_none()); + } + + #[test] + fn schema_descriptive_name_conflict_user_message_is_actionable() { + let err = IngestionError::SchemaDescriptiveNameConflict { + descriptive_name: "Photography".to_string(), + existing_canonical: "abc123".to_string(), + reason: "schema_type mismatch".to_string(), + }; + let msg = err.user_message(); + assert!( + msg.contains("Photography"), + "user_message must name the conflicting schema; got {:?}", + msg + ); + assert!( + msg.to_lowercase().contains("rename") || msg.to_lowercase().contains("reuse"), + "user_message must tell the user what to do; got {:?}", + msg + ); + } + #[test] fn test_truncate_body_short() { let short = "short body"; diff --git a/src/ingestion/ingestion_service/schema_creation.rs b/src/ingestion/ingestion_service/schema_creation.rs index 83bcbca6..c46cbe77 100644 --- a/src/ingestion/ingestion_service/schema_creation.rs +++ b/src/ingestion/ingestion_service/schema_creation.rs @@ -290,14 +290,26 @@ impl IngestionService { // don't race on creating/expanding the same schema. let _lock = self.schema_creation_lock.lock().await; - // Add schema to the schema service via the node - let add_response = { - node.add_schema_to_service(&schema).await.map_err(|error| { - IngestionError::SchemaCreationError(format!( + // Add schema to the schema service via the node. A 409 from the + // service (descriptive_name already bound to a different active + // canonical) bubbles up as a `FoldDbError::Config` whose message + // begins with the sentinel `schema service refused duplicate + // descriptive_name '...'` — parse it out into the typed + // `SchemaDescriptiveNameConflict` so the UI/CLI sees a clean + // "Rename your schema" prompt instead of a wrapped "Failed to + // create schema via schema service: ". + let add_response = match node.add_schema_to_service(&schema).await { + Ok(response) => response, + Err(error) => { + let display = error.to_string(); + if let Some(typed) = crate::ingestion::error::parse_schema_name_conflict(&display) { + return Err(typed); + } + return Err(IngestionError::SchemaCreationError(format!( "Failed to create schema via schema service: {}", error - )) - })? + ))); + } }; let schema_response = &add_response.schema;