diff --git a/.gitignore b/.gitignore index 4d05e972..61c77eea 100755 --- a/.gitignore +++ b/.gitignore @@ -25,3 +25,6 @@ extenddb.toml.bak extenddb-*.toml extenddb-*.toml.bak extenddb-*.toml.keep + +# Local wire-test instance +.gw-instance/ diff --git a/crates/core/src/types/capacity.rs b/crates/core/src/types/capacity.rs index d16a8843..cf4a170b 100755 --- a/crates/core/src/types/capacity.rs +++ b/crates/core/src/types/capacity.rs @@ -53,10 +53,16 @@ pub struct Capacity { /// Capacity consumed by a vector index. /// /// Vector indexes meter in their own units, separate from table read and write -/// capacity: `VectorSearchRequestBytes` for `SearchVectors`, and -/// `VectorWriteRequestBytes` for writes replicated into the index. Both are -/// byte figures, not unit figures, and each is omitted rather than reported as -/// zero when the operation does not consume it. +/// capacity. A `SearchVectors` charge is reported twice, as +/// `VectorSearchRequestBytes` and `VectorSearchUnits` with the same value; a +/// write replicated into the index is reported as `VectorWriteRequestBytes`. +/// Each member is omitted rather than reported as zero when the operation does +/// not consume it. +/// +/// The duplicated search member is measured, not a guess: probe P8 against real +/// Amazon DynamoDB on 2026-08-19 captured both members on every search, always +/// equal, under both `INDEXES` and `TOTAL`. The write side has no such capture, +/// so it carries the bytes member alone until one exists. #[derive(Debug, Clone, Default, Serialize)] pub struct VectorCapacity { /// Bytes consumed by a `SearchVectors` operation. @@ -65,6 +71,10 @@ pub struct VectorCapacity { skip_serializing_if = "Option::is_none" )] pub vector_search_request_bytes: Option, + /// The same figure as `VectorSearchRequestBytes`, under the name a client + /// reading units expects. + #[serde(rename = "VectorSearchUnits", skip_serializing_if = "Option::is_none")] + pub vector_search_units: Option, /// Bytes consumed replicating a write into the index. #[serde( rename = "VectorWriteRequestBytes", @@ -73,6 +83,33 @@ pub struct VectorCapacity { pub vector_write_request_bytes: Option, } +impl VectorCapacity { + /// A search charge, which the service reports under both search member names + /// with the same value. + /// + /// A constructor rather than a struct literal at the call site, so the two + /// members cannot drift apart: there is no way to set one and forget the + /// other. + #[must_use] + pub const fn search(bytes: f64) -> Self { + Self { + vector_search_request_bytes: Some(bytes), + vector_search_units: Some(bytes), + vector_write_request_bytes: None, + } + } + + /// A write-replication charge. + #[must_use] + pub const fn write(bytes: f64) -> Self { + Self { + vector_search_request_bytes: None, + vector_search_units: None, + vector_write_request_bytes: Some(bytes), + } + } +} + /// Consumed capacity information returned when requested. #[derive(Debug, Clone, Serialize)] pub struct ConsumedCapacity { @@ -201,15 +238,7 @@ impl ConsumedCapacity { } let map: HashMap = charges .into_iter() - .map(|(name, bytes)| { - ( - name, - VectorCapacity { - vector_search_request_bytes: None, - vector_write_request_bytes: Some(bytes), - }, - ) - }) + .map(|(name, bytes)| (name, VectorCapacity::write(bytes))) .collect(); if !map.is_empty() { self.vector_indexes = Some(map); @@ -490,4 +519,47 @@ mod tests { .is_none() ); } + + /// A search charge serialises both measured members and nothing else. + /// + /// Probe P8 (2026-08-19, real Amazon DynamoDB) captured the whole + /// `SearchVectors` shape as + /// `{"VectorSearchRequestBytes": 1024.0, "VectorSearchUnits": 1024.0}` under + /// both `INDEXES` and `TOTAL`. A client reading the units member got `null` + /// from ExtendDB and a number from the service. + #[test] + fn a_search_charge_carries_both_measured_members() { + let capacity = VectorCapacity::search(2048.0); + let Ok(value) = serde_json::to_value(capacity) else { + panic!("vector capacity should serialize"); + }; + let members: Vec<&str> = value + .as_object() + .expect("object") + .keys() + .map(String::as_str) + .collect(); + assert_eq!(members, ["VectorSearchRequestBytes", "VectorSearchUnits"]); + assert_eq!(value["VectorSearchRequestBytes"], 2048.0); + assert_eq!(value["VectorSearchUnits"], 2048.0); + } + + /// The write charge is untouched by the search-side addition. The service's + /// write-side units member is NOT measured, so nothing may be invented for + /// it: a write charge still carries exactly one member. + #[test] + fn a_write_charge_carries_only_the_measured_bytes_member() { + let capacity = ConsumedCapacity::write("table", 1.0, true) + .with_vector_writes([("vidx".to_owned(), 512.0)], true); + let Ok(value) = serde_json::to_value(capacity) else { + panic!("indexed capacity should serialize"); + }; + let members: Vec<&str> = value["VectorIndexes"]["vidx"] + .as_object() + .expect("object") + .keys() + .map(String::as_str) + .collect(); + assert_eq!(members, ["VectorWriteRequestBytes"]); + } } diff --git a/crates/core/src/types/mod.rs b/crates/core/src/types/mod.rs index bb393f46..fb4e009b 100755 --- a/crates/core/src/types/mod.rs +++ b/crates/core/src/types/mod.rs @@ -66,6 +66,7 @@ pub use table::{ VECTOR_SEARCH_SCHEMA_UNDECLARED, VectorAttribute, VectorIndexDescription, VectorIndexSpecification, VectorIndexUpdate, vector_attribute_conflicting_definition, vector_attribute_redefines_key, vector_attribute_redefines_vector, + vector_index_delete_in_allocation_phase, }; pub use transaction::{ CancellationReason, ItemResponse, TransactConditionCheck, TransactDelete, TransactGet, diff --git a/crates/core/src/types/table.rs b/crates/core/src/types/table.rs index 50bb0dba..d5f1d854 100755 --- a/crates/core/src/types/table.rs +++ b/crates/core/src/types/table.rs @@ -589,6 +589,27 @@ pub const VECTOR_INDEX_ALREADY_EXISTS: &str = "Attempting to create an index whi pub const VECTOR_INDEX_CREATE_IN_USE_PREFIX: &str = "Attempt to change a resource which is still in use: Index is being created."; +/// Message the service returns when a vector index is deleted while its creation +/// is still in the resource-allocation phase. +/// +/// Measured byte-exact on 2026-08-19 (probe P2), carried as a +/// `ResourceInUseException` with HTTP 400. Deleting a `CREATING` vector index is +/// phase-dependent: refused while the index reports `Backfilling: false`, +/// accepted once it reports `Backfilling: true`, which is why the text tells the +/// caller to retry rather than that the request was wrong. +/// +/// A function rather than a bare constant because the service names both +/// resources, separated by a single space and with no comma: +/// "... is active. Table: t Index: i". +#[must_use] +pub fn vector_index_delete_in_allocation_phase(table_name: &str, index_name: &str) -> String { + format!( + "Attempt to change a resource which is still in use: Index creation is in resource \ + allocation phase. Retry deletion during backfilling phase or when the index is active. \ + Table: {table_name} Index: {index_name}" + ) +} + impl VectorIndexDescription { /// Reject a description whose reported state the service would never produce. /// diff --git a/crates/core/src/validation/mod.rs b/crates/core/src/validation/mod.rs index 55eda5a9..87077676 100755 --- a/crates/core/src/validation/mod.rs +++ b/crates/core/src/validation/mod.rs @@ -13,7 +13,8 @@ use crate::limits::LimitsConfig; use crate::types::{ AttributeDefinition, AttributeValue, BillingMode, CreateTableInput, DeleteItemInput, GetItemInput, Item, KeySchemaElement, KeyType, MAX_VECTOR_INDEXES_PER_TABLE, PutItemInput, - ReturnValues, ScalarAttributeType, Select, UpdateItemInput, item_size_bytes, + ReturnValues, ScalarAttributeType, Select, UpdateItemInput, VECTOR_INDEX_COUNT_LIMIT_CREATE, + VECTOR_INDEX_REQUIRES_PAY_PER_REQUEST, item_size_bytes, }; /// Validate a table name per Virtual `DynamoDB` rules. @@ -389,15 +390,14 @@ fn validate_vector_indexes(input: &CreateTableInput) -> Result<(), DynamoDbError // vector index capacity mode as on-demand only. `BillingMode` defaults to // PROVISIONED when absent, so an omitted BillingMode is a rejection too. // - // Wording measured against the service on 2026-08-11; the earlier text was a - // reasonable paraphrase but not what the service says. + // Wording measured against the service on 2026-08-11 and held in the + // constant, which the UpdateTable paths share: the service returns one + // string for every direction of this rule. if !vis.is_empty() && input.billing_mode.unwrap_or(BillingMode::Provisioned) != BillingMode::PayPerRequest { return Err(DynamoDbError::ValidationException( - "One or more parameter values were invalid: Vector indexes are only supported \ - for PAY_PER_REQUEST tables" - .to_owned(), + VECTOR_INDEX_REQUIRES_PAY_PER_REQUEST.to_owned(), )); } @@ -405,10 +405,9 @@ fn validate_vector_indexes(input: &CreateTableInput) -> Result<(), DynamoDbError // echo the offending count here, unlike its SearchSchema messages, so neither // does this. if vis.len() > MAX_VECTOR_INDEXES_PER_TABLE { - return Err(DynamoDbError::ValidationException(format!( - "One or more parameter values were invalid: VectorIndex count exceeds the \ - per-table limit of {MAX_VECTOR_INDEXES_PER_TABLE}" - ))); + return Err(DynamoDbError::ValidationException( + VECTOR_INDEX_COUNT_LIMIT_CREATE.to_owned(), + )); } // Several indexes may share one vector attribute, but they must agree on its @@ -2341,6 +2340,18 @@ mod tests { per-table limit of {MAX_VECTOR_INDEXES_PER_TABLE}" ) ); + + // Both messages are now taken from the shared constants rather than + // inlined here, and the count constant spells its limit out as a literal. + // Assert the two agree, so raising the limit cannot leave the message + // stating the old one. + assert_eq!( + VECTOR_INDEX_COUNT_LIMIT_CREATE, + format!( + "One or more parameter values were invalid: VectorIndex count exceeds the \ + per-table limit of {MAX_VECTOR_INDEXES_PER_TABLE}" + ) + ); } #[test] diff --git a/crates/core/src/validation/vector_item.rs b/crates/core/src/validation/vector_item.rs index f955878a..dc48c01b 100644 --- a/crates/core/src/validation/vector_item.rs +++ b/crates/core/src/validation/vector_item.rs @@ -26,6 +26,19 @@ fn invalid(msg: impl Into) -> DynamoDbError { DynamoDbError::ValidationException(msg.into()) } +/// Envelope the service uses for every vector-attribute write rejection. +/// +/// Measured byte-exact against real Amazon DynamoDB on 2026-08-27, all four +/// kinds (wrong dimension count, wrong attribute type, a component out of f32 +/// range, and the element-level type error) in us-west-2 and us-east-1: one +/// sentence, ordinary full stop. The envelope is region-uniform, not per kind: +/// eu-west-1 measured the SAME DAY states the sentence twice with a space +/// before the second full stop, for all four kinds alike. An earlier capture +/// (2026-08-19, us-west-2) recorded the doubled envelope for three kinds and +/// this single one for the fourth, so neither region tracks the other's +/// history cleanly; these strings pin the us shape as measured 2026-08-27. +const INVALID_PARAMETER_VALUES_ONCE: &str = "One or more parameter values were invalid."; + /// Validate the vector-relevant attributes of an item being written against the /// table's vector indexes. /// @@ -192,24 +205,22 @@ fn validate_vector_attribute( let dimensions = index.dimensions as usize; let AttributeValue::L(elements) = value else { - // Measured 2026-08-07 against N, S and NS in the vector position, all three - // of which produce this exact text. Note "32-bit floating point number - // list" rather than any phrasing of "list of numbers", and no full stop - // before IndexName, matching the size message below. + // Measured 2026-08-27 with a String in the vector position: a prose type + // with no actual-type token and no full stop before IndexName. The + // 2026-08-19 capture recorded "Expected: L, Actual: S" here; the body + // changed with the envelope, and eu-west-1 still returns the old body. return Err(invalid(format!( - "One or more parameter values were invalid. Invalid type for parameter {attr}, \ + "{INVALID_PARAMETER_VALUES_ONCE} Invalid type for parameter {attr}, \ Expected: 32-bit floating point number list IndexName: {index_name}" ))); }; if elements.len() != dimensions { - // Punctuation matches the service exactly, verified 2026-08-05: - // "...were invalid. Invalid size for parameter emb, Expected: 4, - // Actual: 3 IndexName: vidx" - // Note the full stop after "invalid", the comma after the parameter - // name, and the absence of one after the actual count. + // Punctuation matches the service exactly, verified 2026-08-27: a comma + // after the parameter name and NO full stop after the actual count. The + // 2026-08-19 capture had a full stop there; it left with the envelope. return Err(invalid(format!( - "One or more parameter values were invalid. Invalid size for parameter {attr}, \ + "{INVALID_PARAMETER_VALUES_ONCE} Invalid size for parameter {attr}, \ Expected: {dimensions}, Actual: {} IndexName: {index_name}", elements.len() ))); @@ -230,15 +241,18 @@ fn validate_vector_attribute( .map(format_scientific) .unwrap_or_else(|_| number.clone()); return Err(invalid(format!( - "One or more parameter values were invalid. Invalid value for parameter \ + "{INVALID_PARAMETER_VALUES_ONCE} Invalid value for parameter \ {attr}[{position}], Value: {display} is outside valid range \ [-3.4028235E38, 3.4028235E38]. IndexName: {index_name}" ))); } } other => { + // Measured 2026-08-27 for a String element; the 2026-08-19 + // capture agrees on this body. Same envelope as every other kind + // in this family since the envelopes went region-uniform. return Err(invalid(format!( - "One or more parameter values were invalid. Invalid type for parameter \ + "{INVALID_PARAMETER_VALUES_ONCE} Invalid type for parameter \ {attr}[{position}], Expected: 32-bit floating point number, Actual: {}. \ IndexName: {index_name}", attribute_type_token(other) @@ -448,14 +462,15 @@ mod tests { /// Exact wording, not fragments. /// - /// Verified against real DynamoDB on 2026-08-05: a `PutItem` carrying three - /// values against a four-dimension index returned - /// "One or more parameter values were invalid. Invalid size for parameter - /// emb, Expected: 4, Actual: 3 IndexName: vidx". - /// - /// The punctuation is load-bearing: a full stop after "invalid", a comma - /// after the parameter name, and no separator after the actual count. The - /// fragment assertions above cannot catch a regression in any of those, so + /// Re-measured against real Amazon DynamoDB on 2026-08-27 in us-west-2 and + /// us-east-1: a `PutItem` carrying three values against a four-dimension + /// index returns "One or more parameter values were invalid. Invalid size + /// for parameter emb, Expected: 4, Actual: 3 IndexName: vidx". The + /// punctuation is load-bearing: one envelope sentence, and no full stop + /// between the actual count and IndexName. The 2026-08-19 capture had a + /// doubled envelope and a full stop there; eu-west-1 still answers that + /// older shape, so these strings pin the us regions as measured 2026-08-27. + /// The fragment assertions above cannot catch an envelope regression, so /// this asserts the whole string. #[test] fn dimension_mismatch_message_matches_the_service_exactly() { @@ -473,18 +488,21 @@ mod tests { assert!(message.contains("Expected: 5, Actual: 0")); } - /// Asserted whole rather than by fragment. Every message below was measured - /// against the live service on 2026-08-07, and the previous fragment assertions - /// excluded the "One or more parameter values were invalid" prefix, which is - /// precisely where three of the four had drifted: two used a colon where the - /// service uses a full stop, and one omitted the prefix entirely. + /// Asserted whole rather than by fragment. + /// + /// Re-measured 2026-08-27 (a String in the vector position, us-west-2 and + /// us-east-1): a prose expected type with no actual-type token and no full + /// stop before IndexName. The 2026-08-19 capture recorded "Expected: L, + /// Actual: S" with the doubled envelope, which eu-west-1 still answers; the + /// body travelled with the envelope, so both pin to the 2026-08-27 us shape. #[test] fn rejects_non_list_vector() { let message = err(&item_with_vector(AttributeValue::N("0.1".to_owned()))); assert_eq!( message, "One or more parameter values were invalid. Invalid type for parameter \ - ProductEmbedding, Expected: 32-bit floating point number list IndexName: ProductIndex" + ProductEmbedding, Expected: 32-bit floating point number list \ + IndexName: ProductIndex" ); } @@ -573,6 +591,94 @@ mod tests { validate_vector_write(&item, &[index()], &defs()).unwrap(); } + /// The probe fixture: attribute `emb`, index `vidx`, four dimensions, which is + /// exactly what probes P4, P5, P9 and P10 ran against real Amazon DynamoDB on + /// 2026-08-19. Lets the three assertions below compare against the captured + /// wire strings byte for byte rather than against a re-templated form of them. + fn probe_index() -> VectorIndexKeyInfo { + VectorIndexKeyInfo { + index_name: "vidx".to_owned(), + dimensions: 4, + vector_attribute_name: "emb".to_owned(), + search_schema: Vec::new(), + projection: crate::types::Projection { + projection_type: crate::types::ProjectionType::All, + non_key_attributes: None, + }, + } + } + + fn probe_err(vector: AttributeValue) -> String { + let mut item = Item::new(); + item.insert("pk".to_owned(), AttributeValue::S("a".to_owned())); + item.insert("emb".to_owned(), vector); + match validate_vector_write(&item, &[probe_index()], &[]).unwrap_err() { + DynamoDbError::ValidationException(m) => m, + other => panic!("expected ValidationException, got {other:?}"), + } + } + + /// Byte-for-byte against the 2026-08-27 capture (us-west-2 and us-east-1, + /// `wrong-dims` in `/tmp/ddbprobe/envelope-2026-08-27.jsonl`). + #[test] + fn wrong_dimension_count_is_byte_identical_to_the_service() { + assert_eq!( + probe_err(num_vec(&["0.1", "0.2", "0.3"])), + "One or more parameter values were invalid. Invalid size for parameter emb, \ + Expected: 4, Actual: 3 IndexName: vidx" + ); + } + + /// Byte-for-byte against the 2026-08-27 capture (`wrong-type`). + #[test] + fn wrong_attribute_type_is_byte_identical_to_the_service() { + assert_eq!( + probe_err(AttributeValue::S("not-a-vector".to_owned())), + "One or more parameter values were invalid. Invalid type for parameter emb, \ + Expected: 32-bit floating point number list IndexName: vidx" + ); + } + + /// Byte-for-byte against the 2026-08-27 capture (`element-type`), the one + /// kind whose body did not change between the 2026-08-19 and 2026-08-27 + /// measurements. + /// + /// Both measured actual types are asserted, because `BOOL` is the one that + /// shows the type token is not restricted to the scalar key types. + #[test] + fn wrong_element_type_is_byte_identical_to_the_service() { + let with = |element: AttributeValue| { + probe_err(AttributeValue::L(vec![ + AttributeValue::N("0.1".to_owned()), + element, + AttributeValue::N("0".to_owned()), + AttributeValue::N("0".to_owned()), + ])) + }; + assert_eq!( + with(AttributeValue::S("x".to_owned())), + "One or more parameter values were invalid. Invalid type for parameter emb[1], \ + Expected: 32-bit floating point number, Actual: S. IndexName: vidx" + ); + assert_eq!( + with(AttributeValue::Bool(true)), + "One or more parameter values were invalid. Invalid type for parameter emb[1], \ + Expected: 32-bit floating point number, Actual: BOOL. IndexName: vidx" + ); + } + + /// Byte-for-byte against the 2026-08-27 capture (`out-of-range`), including + /// the service's `3.5E+38` normalisation of the submitted `3.5E38`. + #[test] + fn f32_overflow_is_byte_identical_to_the_service() { + assert_eq!( + probe_err(num_vec(&["0", "3.5E38", "0", "0"])), + "One or more parameter values were invalid. Invalid value for parameter emb[1], \ + Value: 3.5E+38 is outside valid range [-3.4028235E38, 3.4028235E38]. \ + IndexName: vidx" + ); + } + #[test] fn format_scientific_adds_exponent_sign() { assert_eq!(format_scientific(1.3e40), "1.3E+40"); diff --git a/crates/engine/src/create_table.rs b/crates/engine/src/create_table.rs index 7df6da3b..3463fa2b 100755 --- a/crates/engine/src/create_table.rs +++ b/crates/engine/src/create_table.rs @@ -112,14 +112,12 @@ pub(crate) fn storage_err_to_dynamo(e: extenddb_storage::error::StorageError) -> StorageError::IndexAlreadyExists(name) => DynamoDbError::ValidationException(format!( "One or more parameter values were invalid: Index already exists: {name}" )), - // The sentence AWS documents for this refusal, quoted in the vector - // search tutorial's readiness callout and pinned by the ground-truth - // runs of 2026-08-24 (us-east-1 and eu-west-2). - StorageError::IndexesInUse(_) => DynamoDbError::ResourceInUseException( - "Cannot delete table while indexes are being created, updated, or deleted.".to_owned(), - ), - StorageError::LimitExceeded(msg) => DynamoDbError::LimitExceededException(msg), - // Retryable by definition, so it maps like Connection: a 503 the SDKs + // The resource exists and the request is well formed; its current state + // forbids the change. The backend owns the wording because it owns the + // state: the documented delete-table sentence and the measured + // phase-dependent vector refusal both arrive through this one arm. + StorageError::IndexesInUse(msg) => DynamoDbError::ResourceInUseException(msg), + StorageError::LimitExceeded(msg) => DynamoDbError::LimitExceededException(msg), // Retryable by definition, so it maps like Connection: a 503 the SDKs // retry, rather than a 500 they surface. StorageError::Transient(msg) => { tracing::warn!(transient_error = %msg, "transient storage error"); diff --git a/crates/engine/src/search_vectors.rs b/crates/engine/src/search_vectors.rs index ab4b66ae..731c011c 100644 --- a/crates/engine/src/search_vectors.rs +++ b/crates/engine/src/search_vectors.rs @@ -277,9 +277,12 @@ pub async fn handle_search_vectors( // a ProjectionExpression, and a TopK larger than the item count). let count = search_results.len() as i64; - // The service reports `ConsumedCapacity.VectorSearchRequestBytes`, a byte - // figure, not a unit figure. See `search_request_bytes` for the measured - // model and why exact parity is not achievable. + // The service reports the same figure twice, as + // `ConsumedCapacity.VectorSearchRequestBytes` and + // `ConsumedCapacity.VectorSearchUnits` (probe P8, 2026-08-19: both members on + // every search, always equal, under INDEXES and TOTAL alike). Both are byte + // figures. See `search_request_bytes` for the measured model and why exact + // parity is not achievable. // // `vectors_returned` is counted from the projected results rather than inferred // from whether the expression mentions the attribute, so an alias, a nested @@ -300,10 +303,7 @@ pub async fn handle_search_vectors( ); let consumed_capacity = match input.return_consumed_capacity { ReturnConsumedCapacity::None => None, - _ => Some(extenddb_core::types::VectorCapacity { - vector_search_request_bytes: Some(request_bytes), - vector_write_request_bytes: None, - }), + _ => Some(extenddb_core::types::VectorCapacity::search(request_bytes)), }; let output = SearchVectorsOutput { diff --git a/crates/engine/src/update_table.rs b/crates/engine/src/update_table.rs index 8a8e6580..61a25824 100755 --- a/crates/engine/src/update_table.rs +++ b/crates/engine/src/update_table.rs @@ -175,39 +175,7 @@ pub async fn handle_update_table( .storage .update_table(&ctx.account_id, input) .await - .map_err(|e| match e { - extenddb_storage::error::StorageError::TableNotFound(_name) => { - DynamoDbError::ResourceNotFoundException("Requested resource not found".to_string()) - } - extenddb_storage::error::StorageError::TableNotActive(name) => { - DynamoDbError::ResourceInUseException(format!( - "Table {name} is not in ACTIVE state" - )) - } - extenddb_storage::error::StorageError::IndexAlreadyExists(name) => { - DynamoDbError::ValidationException(format!( - "One or more parameter values were invalid: Index already exists: {name}" - )) - } - extenddb_storage::error::StorageError::IndexNotFound(name) => { - DynamoDbError::ResourceNotFoundException(format!( - "Requested resource not found: Index {name} for table {table_name}" - )) - } - extenddb_storage::error::StorageError::NoOpUpdate(msg) => { - DynamoDbError::ValidationException(msg) - } - extenddb_storage::error::StorageError::Validation(msg) => { - DynamoDbError::ValidationException(msg) - } - extenddb_storage::error::StorageError::LimitExceeded(msg) => { - DynamoDbError::LimitExceededException(msg) - } - other => { - tracing::error!(internal_error = %other, "storage internal error"); - DynamoDbError::InternalServerError("Internal server error".to_owned()) - } - })?; + .map_err(|e| update_table_err_to_dynamo(e, &table_name))?; // A declared-capable backend must not silently drop the vector index change. // Checked against the description it returned, so this cannot be opted out of. @@ -230,3 +198,108 @@ pub async fn handle_update_table( }; serialize_output(&output) } + +/// Map a backend failure from `update_table` onto the wire error. +/// +/// A named function rather than an inline closure so every arm is reachable from +/// a unit test. Two of them are not reachable any other way: no in-tree backend +/// yet returns `Unsupported` or `IndexesInUse` from `update_table`, and the +/// vector capability gate refuses vector requests before the backend is called +/// at all, so a wire test cannot provoke either one today. +fn update_table_err_to_dynamo( + e: extenddb_storage::error::StorageError, + table_name: &str, +) -> DynamoDbError { + use extenddb_storage::error::StorageError; + match e { + StorageError::TableNotFound(_name) => { + DynamoDbError::ResourceNotFoundException("Requested resource not found".to_string()) + } + StorageError::TableNotActive(name) => { + DynamoDbError::ResourceInUseException(format!("Table {name} is not in ACTIVE state")) + } + StorageError::IndexAlreadyExists(name) => DynamoDbError::ValidationException(format!( + "One or more parameter values were invalid: Index already exists: {name}" + )), + StorageError::IndexNotFound(name) => DynamoDbError::ResourceNotFoundException(format!( + "Requested resource not found: Index {name} for table {table_name}" + )), + StorageError::NoOpUpdate(msg) => DynamoDbError::ValidationException(msg), + StorageError::Validation(msg) => DynamoDbError::ValidationException(msg), + StorageError::LimitExceeded(msg) => DynamoDbError::LimitExceededException(msg), + // Not a fault, so deliberately not logged at error level: the backend + // never claimed the feature. Same mapping CreateTable uses; without this + // arm a capability refusal fell through to a 500, which tells a caller + // the server is broken when the request simply cannot be served here. + StorageError::Unsupported(msg) => DynamoDbError::ValidationException(msg), + // The change is refused by the resource's current state, not by the + // request. The backend supplies the whole message because only it knows + // the state. + StorageError::IndexesInUse(msg) => DynamoDbError::ResourceInUseException(msg), + other => { + tracing::error!(internal_error = %other, "storage internal error"); + DynamoDbError::InternalServerError("Internal server error".to_owned()) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use extenddb_storage::error::StorageError; + + /// A backend that cannot serve the request reports a 400, not a 500. + /// + /// This arm was missing while CreateTable had it, so the same refusal + /// answered differently depending on which operation carried it. It matters + /// for a backend whose vector capability is decided at runtime rather than at + /// compile time: its refusal arrives through this path. + #[test] + fn an_unsupported_feature_is_a_validation_exception() { + let err = update_table_err_to_dynamo( + StorageError::Unsupported("Vector indexes are not supported here".to_owned()), + "t", + ); + match err { + DynamoDbError::ValidationException(msg) => { + assert_eq!(msg, "Vector indexes are not supported here"); + } + other => panic!("expected ValidationException, got {other:?}"), + } + } + + /// The backend's whole message survives, unwrapped and unprefixed, because + /// the service's own wording for this case names both the table and the index + /// and no layer above the backend knows either. + #[test] + fn a_resource_in_use_refusal_keeps_the_backend_message() { + let measured = extenddb_core::types::vector_index_delete_in_allocation_phase("t", "vidx"); + let err = update_table_err_to_dynamo(StorageError::IndexesInUse(measured.clone()), "t"); + match err { + DynamoDbError::ResourceInUseException(msg) => assert_eq!(msg, measured), + other => panic!("expected ResourceInUseException, got {other:?}"), + } + } + + /// The measured whole string, byte for byte, from probe P2 on 2026-08-19. + #[test] + fn the_allocation_phase_refusal_is_the_measured_whole_string() { + assert_eq!( + extenddb_core::types::vector_index_delete_in_allocation_phase( + "eddbprobe-backfill", + "vidx2" + ), + "Attempt to change a resource which is still in use: Index creation is in resource \ + allocation phase. Retry deletion during backfilling phase or when the index is \ + active. Table: eddbprobe-backfill Index: vidx2" + ); + } + + /// A genuine fault still reports a 500 and is still logged as one. + #[test] + fn an_internal_failure_is_still_an_internal_server_error() { + let err = + update_table_err_to_dynamo(StorageError::Internal("disk on fire".to_owned()), "t"); + assert!(matches!(err, DynamoDbError::InternalServerError(_))); + } +} diff --git a/crates/storage-sqlite/src/data/index.rs b/crates/storage-sqlite/src/data/index.rs index 1d8066e0..21e89c94 100644 --- a/crates/storage-sqlite/src/data/index.rs +++ b/crates/storage-sqlite/src/data/index.rs @@ -103,7 +103,7 @@ pub(crate) struct GsiApplyContext { #[serde(untagged)] pub(crate) enum PendingApplyContext { Gsi(GsiApplyContext), - Vector(super::vector_index::VectorApplyContext), + Vector(extenddb_storage::vector_lifecycle::VectorApplyContext), } impl PendingApplyContext { @@ -560,11 +560,11 @@ pub(crate) async fn apply_claimed_row( #[cfg(test)] mod pending_context_tests { use super::{GsiApplyContext, GsiIndexDef, PendingApplyContext}; - use crate::data::vector_index::{VectorApplyContext, VectorIndexMeta}; use extenddb_core::types::{ AttributeDefinition, KeySchemaElement, KeyType, Projection, ProjectionType, ScalarAttributeType, }; + use extenddb_storage::vector_lifecycle::{VectorApplyContext, VectorIndexMeta}; /// A context written by a build that predates vector rows, verbatim. It must /// still deserialize. diff --git a/crates/storage-sqlite/src/data/vector_index.rs b/crates/storage-sqlite/src/data/vector_index.rs index 7ccc1c58..042c3dbc 100644 --- a/crates/storage-sqlite/src/data/vector_index.rs +++ b/crates/storage-sqlite/src/data/vector_index.rs @@ -33,45 +33,16 @@ //! the removal is the point: an item that loses its vector attribute must leave the //! index, and skipping the enqueue would leave the stale row in place forever. -use serde::{Deserialize, Serialize}; - use extenddb_core::types::{AttributeDefinition, Item, KeySchemaElement, SearchSchemaElementType}; use extenddb_core::validation::{vector_components, vector_norm}; use extenddb_storage::error::StorageError; use extenddb_storage::util::pk_to_text; +use extenddb_storage::vector_lifecycle::{ + BackfillRow, BatchOutcome, VectorApplyContext, VectorIndexBuild, VectorIndexMeta, + classify_backfill_row, item_is_indexable, item_partition, projected_payload, +}; use super::{BoundValue, all_sort_key_info, sk_bound, vector_table_name}; -use crate::vector_search::partition_value; - -/// A vector index as the write path needs it. -/// -/// Serializable because the asynchronous path snapshots it verbatim into the -/// pending row's [`VectorApplyContext`]. The write path and the worker therefore -/// apply from the *same* description of the index, which is the property that stops -/// a queued write from being reinterpreted under a later definition. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct VectorIndexMeta { - pub index_id: String, - pub dimensions: usize, - pub vector_attribute_name: String, - /// The index's projection, applied to the stored row exactly as the GSI path - /// applies its own. Not applying it was an unexplained divergence from the - /// sibling, and it made a search return attributes the index does not project. - pub projection: extenddb_core::types::Projection, - /// The single HASH element's attribute name, when the index declares one. - /// `None` means the index is unscoped and every row shares one partition. - pub hash_attribute_name: Option, - /// Every attribute named by the SearchSchema, HASH and INLINE_FILTER alike. - /// - /// These are projected regardless of `ProjectionType`, which is the documented - /// rule for a vector index and is NOT GSI `KEYS_ONLY` semantics: `KEYS_ONLY` - /// on a vector index projects the base primary key, the vector attribute and - /// the inline filter attributes. Withholding them is not merely a reporting - /// difference, it breaks search: the filter is evaluated against the stored - /// payload, so a missing filter attribute makes every row fail the predicate - /// and a filtered search match nothing. - pub search_schema_attribute_names: Vec, -} /// Load the vector indexes of a table. /// @@ -129,38 +100,6 @@ pub(crate) async fn fetch_vector_indexes_for_table( Ok(out) } -/// Whether an item belongs in a vector index. -/// -/// It must carry the vector attribute, and the HASH attribute when the index -/// declares one: without the latter the row could not be placed in a partition, -/// and putting it in the unscoped partition would make it visible to searches of -/// every other partition. Not an error, exactly as a GSI silently omits an item -/// missing its index key. -fn item_is_indexable(item: &Item, meta: &VectorIndexMeta) -> bool { - if !item.contains_key(&meta.vector_attribute_name) { - return false; - } - match &meta.hash_attribute_name { - Some(name) => item.contains_key(name), - None => true, - } -} - -/// The partition column value for an item under one index. -fn item_partition(item: &Item, meta: &VectorIndexMeta) -> Result { - match &meta.hash_attribute_name { - Some(name) => { - let value = item.get(name).ok_or_else(|| { - StorageError::Internal( - "indexable check passed but the hash attribute is absent".to_owned(), - ) - })?; - partition_value(Some((name.as_str(), value))) - } - None => partition_value(None), - } -} - /// Base-key bind values for a row, in key-schema order. fn base_key_binds( item: &Item, @@ -202,27 +141,6 @@ fn base_key_columns( cols } -/// Everything the propagation worker needs to apply one vector index update, -/// serialized into `gsi_pending.index_context`. -/// -/// `table_id` is carried here even though the queue row has a `table_id` column of -/// its own, because a vector data table is named from the table id *and* the index -/// id. Reading it from the context preserves the invariant that the context alone -/// is sufficient, rather than splitting one apply's inputs across a column and a -/// JSON blob. Both are written from the same variable in the same statement, so -/// they cannot disagree. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct VectorApplyContext { - pub(crate) base_key_schema: Vec, - pub(crate) attribute_definitions: Vec, - pub(crate) table_id: String, - /// Deliberately named `vector` rather than `index`: it is the field whose - /// presence lets the untagged `PendingApplyContext` tell a vector row from a - /// GSI row by shape alone. See that type for why the discriminant is a shape - /// and not a tag. - pub(crate) vector: VectorIndexMeta, -} - /// Maintain every vector index on a table for one item write. /// /// The single entry point for the write path, and the one place that decides @@ -455,28 +373,10 @@ pub(crate) async fn insert_vector_row( } let norm = vector_norm(&components); let part = item_partition(item, meta)?; - // Projected exactly as the GSI sibling projects, so a search returns what - // the index declares and no more. - let mut projected = - super::index::project_item_for_index(item, &[], base_key_schema, &meta.projection); - // The SearchSchema attributes are always projected, whatever the - // ProjectionType. See `search_schema_attribute_names` for why: the inline - // filter is evaluated against this payload, so dropping the attribute would - // silently turn every filtered search into a zero-result search. - for name in &meta.search_schema_attribute_names { - if !projected.contains_key(name) - && let Some(v) = item.get(name) - { - projected.insert(name.clone(), v.clone()); - } - } - // The vector itself is not kept in the payload: it is already in the `vec` - // column as `f32`, which is the width the service validates against, and the - // search path rebuilds the attribute from those bits. Keeping a verbatim - // decimal copy here duplicated 10 to 15 KB per row at 1024 dimensions and - // would have returned the client's original precision where the service - // returns the narrowed value. - projected.remove(&meta.vector_attribute_name); + // Projection, the always-projected SearchSchema attributes, and the stripped + // vector attribute are the shared payload rules: see `projected_payload` for + // why each holds. + let projected = projected_payload(item, base_key_schema, meta); let item_json = serde_json::to_string(&projected) .map_err(|e| StorageError::Internal(format!("serialize item: {e}")))?; @@ -550,8 +450,8 @@ impl<'a> BackfillPlan<'a> { /// Backfill one batch of existing rows into the vector index. /// -/// Returns `(written, fetched, last_rowid)`. `fetched` distinguishes a short read -/// (the end) from a full one, and `last_rowid` is the cursor to resume from. +/// Returns a [`BatchOutcome`] whose `fetched` distinguishes a short read (the end) +/// from a full one, and whose `cursor` is the rowid to resume from. /// /// Pagination is by KEY, not by `OFFSET`. Offset anchors on a position, so removing any /// already-scanned row shifts every later position by one and the next batch skips a @@ -575,24 +475,16 @@ impl<'a> BackfillPlan<'a> { /// /// It was unreachable while the whole backfill ran in one transaction, and became /// reachable the moment batches started committing independently. -/// One batch's outcome: rows indexed, rows skipped as poison, rows fetched -/// (for termination), and the cursor for the next batch. -struct BatchOutcome { - written: usize, - skipped: usize, - fetched: i64, - last_rowid: i64, -} - async fn backfill_vector_batch( tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, plan: &BackfillPlan<'_>, limit: i64, - after_rowid: i64, -) -> Result { + after_rowid: Option, +) -> Result, StorageError> { let base_table = super::data_table_name(plan.table_id); // `rowid > ?` with 0 as the initial cursor: every real rowid is positive, so // the first batch needs no separate query shape. + let after_rowid = after_rowid.unwrap_or(0); let sql = format!("SELECT rowid, item_data FROM {base_table} WHERE rowid > ? ORDER BY rowid LIMIT ?"); let rows: Vec<(i64, String)> = sqlx::query_as(&sql) @@ -602,135 +494,177 @@ async fn backfill_vector_batch( .await .map_err(crate::sqlite_util::map_sqlx_err)?; let fetched = i64::try_from(rows.len()).unwrap_or(limit); - let last_rowid = rows.last().map_or(after_rowid, |(rid, _)| *rid); + let cursor = rows.last().map(|(rid, _)| *rid); let mut written = 0usize; let mut skipped = 0usize; for (rowid, item_json) in rows { - // Poison classification. The live write path treats a malformed vector - // as an invariant violation and errors loudly, because core validation - // ran before storage was reached. That reasoning is FALSE here: rows - // written before the index existed never passed vector validation, so - // a malformed or wrong-dimension vector in the base table is expected - // input for a backfill, not a bug. Propagating it wedged the build in - // an infinite recovery loop: the error left the index CREATING, the - // watchdog re-ran the rebuild, and the same row failed again, forever, - // while the CREATING hold also froze every queued index write for the - // table. A row whose stored bytes cannot enter the index is skipped - // and counted instead, exactly as a GSI omits an item whose key - // attribute has the wrong type. Transient failures (the INSERT itself - // erroring) still propagate: those are retryable and must not drop - // rows. - let Ok(item) = serde_json::from_str::(&item_json) else { - tracing::warn!( - rowid, - index = %plan.meta.index_id, - "backfill: stored item is unparseable; skipping row" - ); - skipped += 1; - continue; - }; - if !item_is_indexable(&item, plan.meta) { - continue; - } - let vector_ok = item - .get(&plan.meta.vector_attribute_name) - .and_then(vector_components) - .is_some_and(|c| c.len() == plan.meta.dimensions); - if !vector_ok { - tracing::warn!( - rowid, - index = %plan.meta.index_id, - "backfill: vector attribute malformed or wrong dimension; skipping row" - ); - skipped += 1; - continue; + // Poison classification lives in the shared lifecycle + // (`classify_backfill_row`), so the two producers of a vector row and + // every backend skip and count the same rows. Transient failures (the + // INSERT below erroring) still propagate: those are retryable and must + // not drop rows. + match classify_backfill_row(&item_json, plan.meta, &rowid) { + BackfillRow::Poison => skipped += 1, + BackfillRow::Omit => {} + BackfillRow::Index(item) => { + insert_vector_row( + tx, + plan.table_id, + plan.meta, + &item, + plan.base_key_schema, + plan.attr_defs, + &plan.key_cols, + VectorRowConflict::KeepExisting, + ) + .await?; + written += 1; + } } - insert_vector_row( - tx, - plan.table_id, - plan.meta, - &item, - plan.base_key_schema, - plan.attr_defs, - &plan.key_cols, - VectorRowConflict::KeepExisting, - ) - .await?; - written += 1; } Ok(BatchOutcome { written, skipped, fetched, - last_rowid, + cursor, }) } -/// A completed backfill: rows indexed and rows skipped as poison. `skipped` -/// is recorded on the catalog row so an ACTIVE index that deliberately omits -/// rows says so, rather than the omission being indistinguishable from a bug. -pub(crate) struct BackfillOutcome { - pub(crate) written: usize, - pub(crate) skipped: usize, -} - -/// Backfill the index in independently committed batches, releasing SQLite's write -/// lock between them. -/// -/// This is what lets the base table stay writable while an index builds, which is how -/// the service behaves: the table remains ACTIVE and accepts writes throughout, and -/// only the index reports CREATING. Holding one transaction for the whole backfill -/// would block every write until it finished. +/// The SQLite implementation of the shared build-lifecycle primitives +/// (`extenddb_storage::vector_lifecycle`). /// -/// Releasing the lock is also what creates the ordering hazard this design has to -/// answer. A write landing mid-backfill is enqueued, and if it were applied before the -/// backfill wrote its older snapshot of the same item, the index would converge on the -/// stale generation. The queue worker therefore refuses to claim any row for a table -/// whose vector index is still CREATING, so those writes accumulate and are applied -/// only after this returns and the index flips to ACTIVE. +/// One value describes one index under construction. The shared drivers +/// (`run_backfill`, `complete_build`, `rebuild_index`) own the ordering rules; +/// this type owns the SQL: each batch takes the engine write lock and a +/// `BEGIN IMMEDIATE` transaction and releases both before it returns, which is +/// what lets the base table stay writable while the index builds. /// -/// A crash part-way leaves the index in CREATING with some rows written, which -/// `reconcile_incomplete_vector_indexes` repairs at startup by rebuilding it. -pub(crate) async fn backfill_vector_index_in_batches( - pool: &sqlx::SqlitePool, - write_lock: &tokio::sync::Mutex<()>, - table_id: &str, - meta: &VectorIndexMeta, - base_key_schema: &[KeySchemaElement], - attr_defs: &[AttributeDefinition], - batch_delay: std::time::Duration, -) -> Result { - const BATCH: i64 = 500; - let plan = BackfillPlan::new(table_id, meta, base_key_schema, attr_defs); - let mut cursor: i64 = 0; - let mut written = 0usize; - let mut skipped = 0usize; - loop { - let outcome = { - let _writer = write_lock.lock().await; - let mut tx = pool - .begin_with("BEGIN IMMEDIATE") - .await - .map_err(crate::sqlite_util::map_sqlx_err)?; - let result = backfill_vector_batch(&mut tx, &plan, BATCH, cursor).await?; - tx.commit() - .await - .map_err(crate::sqlite_util::map_sqlx_err)?; - result - }; - written += outcome.written; - skipped += outcome.skipped; - if outcome.fetched < BATCH { - break; - } - cursor = outcome.last_rowid; - // Outside the lock, so a write can actually proceed during the pause. Zero in - // production; a test sets it so a write is guaranteed to land mid-backfill. - if !batch_delay.is_zero() { - tokio::time::sleep(batch_delay).await; - } +/// `meta` starts `None` on the recovery path: the definition is read from the +/// catalog inside `reset_data_table`'s transaction, because the request that +/// created the index is long gone. The create path loads it up front. +pub(crate) struct SqliteVectorBuild { + pub(crate) pool: sqlx::SqlitePool, + pub(crate) write_lock: std::sync::Arc>, + pub(crate) gsi_notify: std::sync::Arc, + pub(crate) table_id: String, + pub(crate) index_id: String, + pub(crate) base_key_schema: Vec, + pub(crate) attribute_definitions: Vec, + pub(crate) meta: Option, +} + +impl VectorIndexBuild for SqliteVectorBuild { + /// `rowid`, not the primary key: unique whatever the key layout, and valid + /// because every write is `INSERT ... ON CONFLICT DO UPDATE`, which never + /// reassigns one. See `backfill_vector_batch` for the full argument. + type Cursor = i64; + + async fn backfill_batch( + &mut self, + cursor: Option, + limit: i64, + ) -> Result, StorageError> { + let meta = self.meta.as_ref().ok_or_else(|| { + StorageError::Internal( + "vector backfill started before the index definition was loaded".to_owned(), + ) + })?; + let plan = BackfillPlan::new( + &self.table_id, + meta, + &self.base_key_schema, + &self.attribute_definitions, + ); + // Lock and transaction are scoped to the batch and released before this + // returns, so the driver's inter-batch pause really lets a write through. + let _writer = self.write_lock.lock().await; + let mut tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(crate::sqlite_util::map_sqlx_err)?; + let outcome = backfill_vector_batch(&mut tx, &plan, limit, cursor).await?; + tx.commit() + .await + .map_err(crate::sqlite_util::map_sqlx_err)?; + Ok(outcome) + } + + async fn set_backfilling(&mut self) -> Result<(), StorageError> { + sqlx::query( + "UPDATE vector_indexes SET backfilling = 1 WHERE table_id = ? AND index_id = ?", + ) + .bind(&self.table_id) + .bind(&self.index_id) + .execute(&self.pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + Ok(()) + } + + async fn mark_active(&mut self, skipped: usize) -> Result<(), StorageError> { + // `backfilling` is cleared to NULL rather than set to 0, because the + // service removes the member once ACTIVE and the catalog CHECK + // constraint enforces that pairing. + sqlx::query( + "UPDATE vector_indexes SET index_status = 'ACTIVE', backfilling = NULL, \ + skipped_item_count = ? \ + WHERE table_id = ? AND index_id = ?", + ) + .bind(i64::try_from(skipped).unwrap_or(i64::MAX)) + .bind(&self.table_id) + .bind(&self.index_id) + .execute(&self.pool) + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + Ok(()) + } + + async fn reset_data_table(&mut self) -> Result<(), StorageError> { + // Drop and recreate under the lock; the definition is read from the + // catalog rather than reconstructed, because the request that created + // it is long gone. + let _writer = self.write_lock.lock().await; + crate::SqliteEngine::drop_vector_data_table_by_id( + &self.pool, + &self.table_id, + &self.index_id, + ) + .await?; + let mut data_tx = self + .pool + .begin_with("BEGIN IMMEDIATE") + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + crate::SqliteEngine::create_vector_data_table( + &mut data_tx, + &self.table_id, + &self.index_id, + &self.base_key_schema, + &self.attribute_definitions, + ) + .await?; + let meta = fetch_vector_indexes_for_table(&mut data_tx, &self.table_id) + .await? + .into_iter() + .find(|m| m.index_id == self.index_id) + .ok_or_else(|| { + StorageError::Internal(format!( + "vector index {} was selected as CREATING but has no catalog row", + self.index_id + )) + })?; + data_tx + .commit() + .await + .map_err(|e| StorageError::Internal(e.to_string()))?; + self.meta = Some(meta); + Ok(()) + } + + fn notify_active(&mut self) { + self.gsi_notify.notify_waiters(); } - Ok(BackfillOutcome { written, skipped }) } #[cfg(test)] @@ -828,7 +762,7 @@ mod tests { }; let plan = BackfillPlan::new(table_id, &meta, &ks, &ad); - let mut cursor: i64 = 0; + let mut cursor: Option = None; let mut written = 0usize; loop { let outcome = backfill_vector_batch(&mut tx, &plan, 3, cursor) @@ -838,7 +772,7 @@ mod tests { if outcome.fetched < 3 { break; } - cursor = outcome.last_rowid; + cursor = outcome.cursor; } tx.commit().await.expect("commit"); @@ -943,7 +877,7 @@ mod tests { }; let plan = BackfillPlan::new(table_id, &meta, &ks, &ad); - let outcome = backfill_vector_batch(&mut tx, &plan, 100, 0) + let outcome = backfill_vector_batch(&mut tx, &plan, 100, None) .await .expect("a batch containing poison rows must still complete"); tx.commit().await.expect("commit"); @@ -1048,7 +982,6 @@ mod tests { // The backfill now reaches the same base key and must complete rather // than error on the conflict, leaving exactly one row for the key. - let write_lock = tokio::sync::Mutex::new(()); let meta = fetch_vector_indexes_for_table( &mut engine.pool.begin_with("BEGIN IMMEDIATE").await.expect("tx"), table_id, @@ -1058,13 +991,21 @@ mod tests { .into_iter() .find(|m| m.index_id == "vidx-c") .expect("meta"); - let outcome = backfill_vector_index_in_batches( - &engine.pool, - &write_lock, - table_id, - &meta, - &ks, - &ad, + // Driven through the shared driver over the backend hooks, which is the + // production path since the lifecycle extraction. + let mut ops = SqliteVectorBuild { + pool: engine.pool.clone(), + write_lock: std::sync::Arc::clone(&engine.write_lock), + gsi_notify: engine.gsi_notify(), + table_id: table_id.to_owned(), + index_id: "vidx-c".to_owned(), + base_key_schema: ks.clone(), + attribute_definitions: ad.clone(), + meta: Some(meta), + }; + let outcome = extenddb_storage::vector_lifecycle::run_backfill( + &mut ops, + extenddb_storage::vector_lifecycle::BACKFILL_BATCH, std::time::Duration::ZERO, ) .await diff --git a/crates/storage-sqlite/src/delete_table.rs b/crates/storage-sqlite/src/delete_table.rs index 741b6d03..4621b387 100644 --- a/crates/storage-sqlite/src/delete_table.rs +++ b/crates/storage-sqlite/src/delete_table.rs @@ -73,8 +73,7 @@ impl SqliteEngine { // so a build whose catalog row commits while this call is waiting // on the write lock is still caught: the check and the cascade // delete are atomic against a concurrent UpdateTable. - Self::refuse_if_index_build_in_flight(&mut tx, &row.table_id, &input.table_name) - .await?; + Self::refuse_if_index_build_in_flight(&mut tx, &row.table_id).await?; sqlx::query("DELETE FROM tags WHERE resource_arn = ?") .bind(&table_arn) .execute(&mut *tx) @@ -115,8 +114,7 @@ impl SqliteEngine { // the transaction that flips the table to DELETING for the same // reason: without it, a build committing between an earlier check // and this flip would have its table deleted underneath it. - Self::refuse_if_index_build_in_flight(&mut tx, &row.table_id, &input.table_name) - .await?; + Self::refuse_if_index_build_in_flight(&mut tx, &row.table_id).await?; sqlx::query( "UPDATE tables SET table_status = 'DELETING', status_transition_at = ? \ WHERE account_id = ? AND table_name = ?", @@ -147,7 +145,6 @@ impl SqliteEngine { async fn refuse_if_index_build_in_flight( tx: &mut sqlx::Transaction<'_, sqlx::Sqlite>, table_id: &str, - table_name: &str, ) -> Result<(), StorageError> { let in_flight: Option<(i64,)> = sqlx::query_as( "SELECT 1 FROM vector_indexes WHERE table_id = ? AND backfilling IS NOT NULL LIMIT 1", @@ -157,7 +154,16 @@ impl SqliteEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; if in_flight.is_some() { - return Err(StorageError::IndexesInUse(table_name.to_owned())); + // The sentence AWS documents for this refusal, quoted in the vector + // search tutorial's readiness callout and pinned by the ground-truth + // runs of 2026-08-24 (us-east-1 and eu-west-2). Composed here rather + // than in the engine map because the variant carries the whole + // client-facing message: only the backend knows which in-use state + // it refused for. + return Err(StorageError::IndexesInUse( + "Cannot delete table while indexes are being created, updated, or deleted." + .to_owned(), + )); } Ok(()) } diff --git a/crates/storage-sqlite/src/update_table.rs b/crates/storage-sqlite/src/update_table.rs index 4ab4d300..7d04b444 100644 --- a/crates/storage-sqlite/src/update_table.rs +++ b/crates/storage-sqlite/src/update_table.rs @@ -19,6 +19,7 @@ use extenddb_core::types::{ }; use extenddb_storage::error::StorageError; use extenddb_storage::util::effective_attribute_definitions; +use extenddb_storage::vector_lifecycle::VectorIndexBuild; use crate::store::SqliteEngine; @@ -37,8 +38,8 @@ impl SqliteEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let row: Option<(String, String, String, String)> = sqlx::query_as( - "SELECT table_status, table_id, key_schema, attribute_definitions \ + let row: Option<(String, String, String, String, Option)> = sqlx::query_as( + "SELECT table_status, table_id, key_schema, attribute_definitions, billing_mode \ FROM tables WHERE account_id = ? AND table_name = ?", ) .bind(account_id) @@ -47,7 +48,10 @@ impl SqliteEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; - let (status, table_id, ks_json, ad_json) = + // `stored_billing_mode` is read once here and threaded through every + // billing check below, so the several readers of one row cannot + // disagree. A NULL stored value means PROVISIONED. + let (status, table_id, ks_json, ad_json, stored_billing_mode) = row.ok_or_else(|| StorageError::TableNotFound(input.table_name.clone()))?; if status != "ACTIVE" { return Err(StorageError::TableNotActive(input.table_name.clone())); @@ -70,6 +74,32 @@ impl SqliteEngine { } } + // The other direction of the same rule: a vector index cannot be ADDED to + // a table that is provisioned. Measured 2026-08-19 against a live + // PROVISIONED table, which returned the identical string, so the two + // directions share one constant. + // + // The check is on the request's NET billing mode, not the table's stored + // mode: an UpdateTable that switches to PAY_PER_REQUEST and creates the + // index in the same call was measured to succeed. That is the same + // net-effect evaluation the index-count limit below uses, and it is why + // the request's own billing_mode wins when present. + let creates_vector_index = input + .vector_index_updates + .as_ref() + .is_some_and(|updates| updates.iter().any(|u| u.create.is_some())); + if creates_vector_index { + let net_pay_per_request = match input.billing_mode { + Some(mode) => mode == BillingMode::PayPerRequest, + None => stored_billing_mode.as_deref() == Some("PAY_PER_REQUEST"), + }; + if !net_pay_per_request { + return Err(StorageError::Validation( + extenddb_core::types::VECTOR_INDEX_REQUIRES_PAY_PER_REQUEST.to_owned(), + )); + } + } + // Reject ProvisionedThroughput when the effective billing mode is // PAY_PER_REQUEST. The effective mode is the requested billing_mode when // the request changes it, otherwise the table's current mode. Real @@ -81,17 +111,7 @@ impl SqliteEngine { let effective_ppr = match input.billing_mode { Some(BillingMode::PayPerRequest) => true, Some(BillingMode::Provisioned) => false, - None => { - let current_bm: Option> = sqlx::query_scalar( - "SELECT billing_mode FROM tables WHERE account_id = ? AND table_name = ?", - ) - .bind(account_id) - .bind(&input.table_name) - .fetch_optional(&mut *tx) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - current_bm.flatten().as_deref() == Some("PAY_PER_REQUEST") - } + None => stored_billing_mode.as_deref() == Some("PAY_PER_REQUEST"), }; if effective_ppr { return Err(StorageError::Validation( @@ -104,8 +124,8 @@ impl SqliteEngine { if matches!(input.billing_mode, Some(BillingMode::Provisioned)) && let Some(ref pt) = input.provisioned_throughput { - let current: Option<(Option, Option)> = sqlx::query_as( - "SELECT billing_mode, provisioned_throughput FROM tables \ + let current: Option> = sqlx::query_scalar( + "SELECT provisioned_throughput FROM tables \ WHERE account_id = ? AND table_name = ?", ) .bind(account_id) @@ -113,8 +133,9 @@ impl SqliteEngine { .fetch_optional(&mut *tx) .await .map_err(|e| StorageError::Internal(e.to_string()))?; - if let Some((bm, cur_pt)) = current { - let is_prov = bm.as_deref() == Some("PROVISIONED") || bm.is_none(); + if let Some(cur_pt) = current { + let is_prov = stored_billing_mode.as_deref() == Some("PROVISIONED") + || stored_billing_mode.is_none(); let cur: serde_json::Value = cur_pt .as_deref() .and_then(|s| serde_json::from_str(s).ok()) @@ -729,17 +750,25 @@ impl SqliteEngine { .await .map_err(|e| StorageError::Internal(e.to_string()))?; + // The build's storage primitives, owned, so the detached task can + // outlive this call. The shared drivers in + // `extenddb_storage::vector_lifecycle` own the ordering rules; this + // value owns the SQL. + let mut ops = crate::data::vector_index::SqliteVectorBuild { + pool: self.pool.clone(), + write_lock: std::sync::Arc::clone(&self.write_lock), + gsi_notify: self.gsi_notify(), + table_id: table_id.to_owned(), + index_id: index_id.to_owned(), + base_key_schema: base_ks.to_vec(), + attribute_definitions: effective_ad.to_vec(), + meta: None, + }; + // The scan is about to start, so the member becomes true. Set outside the // backfill transaction, otherwise no observer could see it: the whole point // of the flag is to be readable while the scan is in progress. - sqlx::query( - "UPDATE vector_indexes SET backfilling = 1 WHERE table_id = ? AND index_id = ?", - ) - .bind(table_id) - .bind(index_id) - .execute(&self.pool) - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; + ops.set_backfilling().await?; let mut meta_tx = self .pool @@ -761,31 +790,30 @@ impl SqliteEngine { .commit() .await .map_err(|e| StorageError::Internal(e.to_string()))?; + ops.meta = Some(metas); - // Everything the scan needs, owned, so it can outlive this call. - let pool = self.pool.clone(); - let write_lock = std::sync::Arc::clone(&self.write_lock); let batch_delay = std::time::Duration::from_millis(self.vector_backfill_batch_delay().await); // The floor on how long the index stays CREATING. Captured with the // creation instant here rather than inside the task, so the hold // measures from when the caller could first observe the index, and a // live settings change mid-build does not move an in-flight deadline. + // The wait itself lives in the shared driver, beside the flip it delays. let min_creating = std::time::Duration::from_millis(self.vector_index_min_creating_ms().await); let created_at = tokio::time::Instant::now(); - let owned_table_id = table_id.to_owned(); let owned_index_id = index_id.to_owned(); let owned_index_name = create.index_name.clone(); - let owned_base_ks = base_ks.to_vec(); - let owned_ad = effective_ad.to_vec(); - let gsi_notify = self.gsi_notify(); // Registered BEFORE the spawn, so there is no instant where the catalog // says CREATING and the registry disagrees while the task is viable. The // guard deregisters on every exit path including a panic, which is the // whole point: a CREATING index with no registry entry is provably // orphaned, and the worker's recovery sweep may rebuild it. + // + // This registry is build OWNERSHIP, which the shared lifecycle leaves to + // the backend by design: a single process can prove a build's liveness + // in memory, where a multi-process backend needs a cross-process claim. self.vector_builds_running .lock() .expect("registry poisoned") @@ -797,12 +825,10 @@ impl SqliteEngine { // and writable throughout, taking over eight minutes on an empty table when // measured, and searches against the index are refused until it is ACTIVE. // - // Not awaited, so failures cannot be returned to the caller. They are logged - // and the index is deliberately LEFT in CREATING, which is the state the - // worker's recovery sweep repairs at runtime (and - // `reconcile_incomplete_vector_indexes` at startup). Flipping it to - // ACTIVE on error would publish a partially populated index, and there is no - // failure state on the wire for an index to sit in. + // Not awaited, so failures cannot be returned to the caller. They are + // logged by `complete_build`, which deliberately leaves the index in + // CREATING: that is the state the worker's recovery sweep repairs at + // runtime (and `reconcile_incomplete_vector_indexes` at startup). tokio::spawn(async move { struct Deregister( std::sync::Arc>>, @@ -815,75 +841,15 @@ impl SqliteEngine { } } } - let _deregister = Deregister(registry, owned_index_id.clone()); - let result = crate::data::vector_index::backfill_vector_index_in_batches( - &pool, - &write_lock, - &owned_table_id, - &metas, - &owned_base_ks, - &owned_ad, + let _deregister = Deregister(registry, owned_index_id); + extenddb_storage::vector_lifecycle::complete_build( + ops, + &owned_index_name, + extenddb_storage::vector_lifecycle::BACKFILL_BATCH, batch_delay, + Some(created_at + min_creating), ) .await; - match result { - Ok(outcome) => { - // The service's online-index machinery never finishes an - // added index instantly, so the CREATING walk (`Backfilling` - // false, then true, then ACTIVE with the member absent) is - // always observable there. A local backfill over a small - // table completes in milliseconds, which would collapse that - // walk into an instant no DescribeTable poll can catch, so - // the flip waits out the remainder of the configured floor. - // Searches keep answering the documented not-ready rejection - // and writes keep queuing behind the CREATING hold for the - // duration, exactly as during a real backfill. - tokio::time::sleep_until(created_at + min_creating).await; - // Populated, so the index can serve. `backfilling` is cleared to - // NULL rather than set to 0, because the service removes the - // member once ACTIVE and the catalog CHECK constraint enforces - // that pairing. The skipped count is recorded in the same flip, - // but only in the catalog (and the log line below): DescribeTable - // parity forbids inventing a response field for it, so an - // operator diagnosing missing search results finds it by - // querying the catalog, not through the API. - let flip = sqlx::query( - "UPDATE vector_indexes SET index_status = 'ACTIVE', backfilling = NULL, \ - skipped_item_count = ? \ - WHERE table_id = ? AND index_id = ?", - ) - .bind(i64::try_from(outcome.skipped).unwrap_or(i64::MAX)) - .bind(&owned_table_id) - .bind(&owned_index_id) - .execute(&pool) - .await; - match flip { - Ok(_) => { - tracing::info!( - index_name = %owned_index_name, - vectors_indexed = outcome.written, - vectors_skipped = outcome.skipped, - "vector index backfill complete" - ); - // Writes that landed during the backfill were held by the - // worker because this index was CREATING. It is ACTIVE - // now, so wake the worker rather than leaving them to sit - // until its next idle timeout. - gsi_notify.notify_waiters(); - } - Err(e) => tracing::error!( - index_name = %owned_index_name, - "vector index backfill finished but the ACTIVE flip failed, \ - leaving it CREATING for startup reconciliation: {e}" - ), - } - } - Err(e) => tracing::error!( - index_name = %owned_index_name, - "vector index backfill failed, leaving it CREATING for startup \ - reconciliation: {e}" - ), - } }); Ok(()) } @@ -1105,10 +1071,16 @@ impl SqliteEngine { /// Drop, recreate, backfill, and flip one vector index to `ACTIVE`. /// /// The shared body of startup reconciliation and the worker's runtime - /// recovery, factored so the two repairs cannot drift. The backfill runs on - /// the batched path, releasing the write lock between batches, so a recovery + /// recovery lives in `extenddb_storage::vector_lifecycle::rebuild_index`, + /// factored so the two repairs cannot drift. The backfill runs on the + /// batched path, releasing the write lock between batches, so a recovery /// on a large table cannot become a write-availability outage for every - /// other table. + /// other table. An earlier version ran the whole backfill in one lock-held + /// transaction, which on a large table blocked writes to EVERY table for + /// the full rebuild. The batched path is safe here for the same reasons it + /// is safe on create: the index is CREATING throughout, so the worker + /// holds this table's queue rows and searches are refused, and the rowid + /// cursor tolerates concurrent base-table writes. async fn rebuild_one_vector_index( &self, index_id: &str, @@ -1121,78 +1093,24 @@ impl SqliteEngine { let attr_defs: Vec = serde_json::from_str(base_ad_json) .map_err(|e| StorageError::Internal(e.to_string()))?; - // Drop and recreate under the lock, then backfill BATCHED, releasing the - // lock between batches exactly as the normal create path does. An earlier - // version ran the whole backfill in one lock-held transaction, which on a - // large table blocked writes to EVERY table for the full rebuild: a - // write-availability outage as the price of recovering one index. The - // batched path is safe here for the same reasons it is safe on create: - // the index is CREATING throughout, so the worker holds this table's - // queue rows and searches are refused, and the rowid cursor tolerates - // concurrent base-table writes. Recovery uses no batch delay: the lever - // exists for tests, and recovery should finish as fast as batching - // allows. - let meta; - { - let _writer = self.write_lock.lock().await; - Self::drop_vector_data_table_by_id(&self.pool, table_id, index_id).await?; - - let mut data_tx = self - .pool - .begin_with("BEGIN IMMEDIATE") - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - Self::create_vector_data_table( - &mut data_tx, - table_id, - index_id, - &base_key_schema, - &attr_defs, - ) - .await?; - // The definition is read from the catalog rather than reconstructed, - // because the request that created it is long gone. - meta = crate::data::vector_index::fetch_vector_indexes_for_table( - &mut data_tx, - table_id, - ) - .await? - .into_iter() - .find(|m| m.index_id == index_id) - .ok_or_else(|| { - StorageError::Internal(format!( - "vector index {index_id} was selected as CREATING but has no catalog row" - )) - })?; - data_tx - .commit() - .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - } - - let outcome = crate::data::vector_index::backfill_vector_index_in_batches( - &self.pool, - &self.write_lock, - table_id, - &meta, - &base_key_schema, - &attr_defs, - std::time::Duration::ZERO, - ) - .await?; - - sqlx::query( - "UPDATE vector_indexes SET index_status = 'ACTIVE', backfilling = NULL, \ - skipped_item_count = ? \ - WHERE table_id = ? AND index_id = ?", + // `meta` starts empty: the shared driver's reset step reloads the + // definition from the catalog inside its own transaction, because the + // request that created the index is long gone. + let mut ops = crate::data::vector_index::SqliteVectorBuild { + pool: self.pool.clone(), + write_lock: std::sync::Arc::clone(&self.write_lock), + gsi_notify: self.gsi_notify(), + table_id: table_id.to_owned(), + index_id: index_id.to_owned(), + base_key_schema, + attribute_definitions: attr_defs, + meta: None, + }; + extenddb_storage::vector_lifecycle::rebuild_index( + &mut ops, + extenddb_storage::vector_lifecycle::BACKFILL_BATCH, ) - .bind(i64::try_from(outcome.skipped).unwrap_or(i64::MAX)) - .bind(table_id) - .bind(index_id) - .execute(&self.pool) .await - .map_err(|e| StorageError::Internal(e.to_string()))?; - Ok(outcome.written) } } diff --git a/crates/storage-sqlite/src/vector_search.rs b/crates/storage-sqlite/src/vector_search.rs index 3ddf2031..1c144967 100644 --- a/crates/storage-sqlite/src/vector_search.rs +++ b/crates/storage-sqlite/src/vector_search.rs @@ -21,7 +21,7 @@ use extenddb_core::types::{AttributeValue, DistanceFunction, Item}; use extenddb_storage::error::StorageError; -use extenddb_storage::util::pk_to_text; +use extenddb_storage::vector_lifecycle::partition_value; use extenddb_storage::{ BoxedFuture, VectorHit, VectorSearch, VectorSearchEngine, VectorSearchOutput, VectorSearchResult, @@ -30,39 +30,6 @@ use extenddb_storage::{ use crate::data::vector_table_name; use crate::store::SqliteEngine; -/// Partition value for an index that declares no HASH element. -/// -/// Such an index searches the whole table, so every row shares one partition -/// rather than the scan needing a second code path. -/// -/// What makes this safe is **not** that the value is unguessable. `pk_to_text` -/// stores an `S` attribute verbatim, so a caller could supply this exact string as -/// a partition key. The guarantee is structural instead: the partition is chosen -/// from the *index's* schema, not from the item, and each index has its own data -/// table. So within one table either every row is keyed by a real hash value (the -/// index declares a HASH element) or every row uses this sentinel (it does not). -/// The two never coexist, so there is nothing for a collision to leak into. -/// -/// The leading NUL is defence in depth for the day that invariant changes, for -/// instance if several indexes ever shared one table. It is deliberately not what -/// correctness rests on, because a reader who believed it was would then feel free -/// to weaken it. -pub(crate) const UNSCOPED_PARTITION: &str = "\u{0}all"; - -/// The partition column value for a vector row. -/// -/// Uses the same `pk_to_text` encoding as item partition keys, so a value written -/// by the write path and a value derived from a search request are byte-identical. -/// Getting this wrong would not fail loudly: it would silently return no hits. -pub(crate) fn partition_value( - hash_key: Option<(&str, &AttributeValue)>, -) -> Result { - match hash_key { - Some((_, value)) => Ok(pk_to_text(value)?.into_owned()), - None => Ok(UNSCOPED_PARTITION.to_owned()), - } -} - /// Decode a stored vector blob into `f32`s. /// /// Rejects a truncated blob rather than reading a short vector, because a @@ -483,36 +450,4 @@ mod tests { "unexpected: {err:?}" ); } - - /// The partition is chosen from the index's schema, never from the item, which - /// is the invariant that makes the sentinel safe. Asserted as the property - /// rather than against one example: the previous version compared the sentinel - /// to `pk_to_text(S("all"))` only, and passed unchanged when the sentinel was - /// weakened to the ordinary string `"unscoped"`. - #[test] - fn the_partition_comes_from_the_index_schema_not_the_item() { - // No HASH element declared: the sentinel, whatever the item holds. - assert_eq!(partition_value(None).unwrap(), UNSCOPED_PARTITION); - - // A HASH element declared: the item's value, verbatim for S. - for value in ["all", "unscoped", UNSCOPED_PARTITION, ""] { - let scoped = - partition_value(Some(("pk", &AttributeValue::S(value.to_owned())))).unwrap(); - assert_eq!( - scoped, value, - "a scoped partition must be the attribute value itself" - ); - } - } - - /// The sentinel keeps a leading NUL as defence in depth. Not what correctness - /// rests on (see the constant's documentation), but weakening it to an ordinary - /// string should break a test rather than pass silently. - #[test] - fn the_unscoped_sentinel_keeps_its_unusual_prefix() { - assert!( - UNSCOPED_PARTITION.starts_with('\0'), - "sentinel must keep its NUL prefix: {UNSCOPED_PARTITION:?}" - ); - } } diff --git a/crates/storage/src/error.rs b/crates/storage/src/error.rs index a628a602..b98593c0 100755 --- a/crates/storage/src/error.rs +++ b/crates/storage/src/error.rs @@ -14,12 +14,14 @@ pub enum StorageError { IndexNotFound(String), #[error("Index already exists: {0}")] IndexAlreadyExists(String), - /// `DeleteTable` arrived while an online index operation (an - /// UpdateTable-created vector index still backfilling) was in progress on - /// the table. Maps to `ResourceInUseException` with the sentence AWS - /// documents in the vector search tutorial's readiness callout: "Cannot - /// delete table while indexes are being created, updated, or deleted." - #[error("Indexes are being created, updated, or deleted on table: {0}")] + /// A change refused because an index on the resource is mid-transition. + /// Carries the whole client-facing message, because the state, and + /// therefore the wording, is known only to the backend that holds it: a + /// DeleteTable refused while indexes build carries the sentence AWS + /// documents for that case, and a vector index deleted while its creation + /// is still allocating resources carries the measured phase-dependent + /// refusal (2026-08-19). Maps to `ResourceInUseException` verbatim. + #[error("{0}")] IndexesInUse(String), #[error("Deletion protection enabled: {0}")] DeletionProtected(String), diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index 46115f41..fa730dbc 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -20,6 +20,7 @@ pub mod operations; pub mod server_components; pub mod settings_store; pub mod transact; +pub mod vector_lifecycle; pub use backend::{Backend, BackendAlreadySet, backend_name, set_backend, try_backend}; diff --git a/crates/storage/src/vector_lifecycle/backfill.rs b/crates/storage/src/vector_lifecycle/backfill.rs new file mode 100644 index 00000000..71ef2c10 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/backfill.rs @@ -0,0 +1,100 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Row classification and batch accounting for the vector index backfill. + +use std::fmt; + +use extenddb_core::types::Item; +use extenddb_core::validation::vector_components; + +use super::meta::{VectorIndexMeta, item_is_indexable}; + +/// Rows scanned per backfill batch, shared by the create path and recovery. +pub const BACKFILL_BATCH: i64 = 500; + +/// One batch's outcome: rows indexed, rows skipped as poison, rows fetched +/// (for termination), and the cursor for the next batch. +/// +/// The cursor type is backend-owned: SQLite scans by `rowid`; a backend whose +/// base tables have no rowid pages by keyset over the full primary key. +/// `cursor` is `None` when the batch fetched no rows. +#[derive(Debug)] +pub struct BatchOutcome { + pub written: usize, + pub skipped: usize, + pub fetched: i64, + pub cursor: Option, +} + +/// A completed backfill: rows indexed and rows skipped as poison. `skipped` +/// is recorded on the catalog row so an ACTIVE index that deliberately omits +/// rows says so, rather than the omission being indistinguishable from a bug. +#[derive(Debug)] +pub struct BackfillOutcome { + pub written: usize, + pub skipped: usize, +} + +/// What a backfill does with one scanned base-table row. +#[derive(Debug)] +pub enum BackfillRow { + /// Parsed, indexable, and its stored vector is well formed: write the row. + Index(Item), + /// Not indexable (no vector attribute, or no HASH attribute when the index + /// declares one): omitted without counting, exactly as a GSI omits an item + /// missing its index key. + Omit, + /// Poison: the stored bytes cannot enter the index. Skipped and counted. + Poison, +} + +/// Classify one scanned base-table row for the backfill. +/// +/// Poison classification. The live write path treats a malformed vector +/// as an invariant violation and errors loudly, because core validation +/// ran before storage was reached. That reasoning is FALSE here: rows +/// written before the index existed never passed vector validation, so +/// a malformed or wrong-dimension vector in the base table is expected +/// input for a backfill, not a bug. Propagating it wedged the build in +/// an infinite recovery loop: the error left the index CREATING, the +/// watchdog re-ran the rebuild, and the same row failed again, forever, +/// while the CREATING hold also froze every queued index write for the +/// table. A row whose stored bytes cannot enter the index is skipped +/// and counted instead, exactly as a GSI omits an item whose key +/// attribute has the wrong type. Transient failures (the row INSERT itself +/// erroring) still propagate from the backend's batch and must not drop rows; +/// classification never sees them. +/// +/// `cursor` names the row in the skip warnings, in whatever form the backend's +/// scan uses. +pub fn classify_backfill_row( + item_json: &str, + meta: &VectorIndexMeta, + cursor: &dyn fmt::Display, +) -> BackfillRow { + let Ok(item) = serde_json::from_str::(item_json) else { + tracing::warn!( + cursor = %cursor, + index = %meta.index_id, + "backfill: stored item is unparseable; skipping row" + ); + return BackfillRow::Poison; + }; + if !item_is_indexable(&item, meta) { + return BackfillRow::Omit; + } + let vector_ok = item + .get(&meta.vector_attribute_name) + .and_then(vector_components) + .is_some_and(|c| c.len() == meta.dimensions); + if !vector_ok { + tracing::warn!( + cursor = %cursor, + index = %meta.index_id, + "backfill: vector attribute malformed or wrong dimension; skipping row" + ); + return BackfillRow::Poison; + } + BackfillRow::Index(item) +} diff --git a/crates/storage/src/vector_lifecycle/build.rs b/crates/storage/src/vector_lifecycle/build.rs new file mode 100644 index 00000000..8d050217 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/build.rs @@ -0,0 +1,452 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The build state machine: backfill orchestration, the `ACTIVE` flip, and the +//! rebuild used by crash recovery. + +use std::future::Future; +use std::time::Duration; + +use super::backfill::{BackfillOutcome, BatchOutcome}; +use crate::error::StorageError; + +/// Storage primitives one backend supplies for one vector index's build. +/// +/// One value describes one index under construction: the implementor carries +/// the table id, index id, base key schema, and whatever connection handles its +/// batches need. The shared drivers ([`run_backfill`], [`complete_build`], +/// [`rebuild_index`]) own the ordering rules; the primitives own SQL, +/// transactions, and locking. +/// +/// Build **ownership** is acquired by the backend before it spawns +/// [`complete_build`] (an in-process registry entry, or a cross-process +/// advisory lock) and released when the task ends, so it is not a primitive +/// here: the shared drivers never decide who owns a build, only what a build +/// does. Liveness renewal during a long backfill goes through +/// [`Self::heartbeat`]. +pub trait VectorIndexBuild: Send { + /// The backfill scan cursor. SQLite uses `rowid`; a backend without rowids + /// uses keyset pagination over the full primary key. Opaque to the driver: + /// it is threaded from one batch into the next and never inspected. + type Cursor: Send; + + /// Run one transactional backfill batch: scan up to `limit` base rows after + /// `cursor` (`None` means from the start), classify each with + /// [`super::classify_backfill_row`], write the indexable ones, and commit. + /// + /// Batches commit independently so the base table stays writable while the + /// index builds. Any locking a batch needs is acquired inside this call and + /// released before it returns, so the inter-batch pause in [`run_backfill`] + /// really does let a concurrent write proceed. + fn backfill_batch( + &mut self, + cursor: Option, + limit: i64, + ) -> impl Future, StorageError>> + Send; + + /// Record that the scan is about to start: `CREATING` with + /// `Backfilling: true`. + /// + /// Called by the backend's create path after the data table exists and + /// before [`complete_build`] is spawned, and set outside the backfill + /// transaction, otherwise no observer could see it: the whole point of the + /// flag is to be readable while the scan is in progress. + fn set_backfilling(&mut self) -> impl Future> + Send; + + /// Publish the index: `ACTIVE`, the `Backfilling` member cleared to absent + /// (not `false`), and the poison-skip count recorded on the catalog row. + /// + /// The count lives only in the catalog (and the completion log line): + /// DescribeTable parity forbids inventing a response field for it, so an + /// operator diagnosing missing search results finds it by querying the + /// catalog, not through the API. + fn mark_active( + &mut self, + skipped: usize, + ) -> impl Future> + Send; + + /// Drop and recreate the index's data table, and reload the index + /// definition from the catalog (the request that created it is long gone). + /// + /// The recovery reset. Idempotent by construction: without the drop, a + /// retry would duplicate every row it had already written before the + /// crash, and a search would return the same item several times. + fn reset_data_table(&mut self) -> impl Future> + Send; + + /// Wake whatever replays writes held while the index was `CREATING`. + /// + /// Writes that landed during the backfill were held by the propagation + /// worker's claim gate. The index is `ACTIVE` now, so the worker must be + /// woken rather than leaving them to sit until its next idle timeout. + fn notify_active(&mut self); + + /// Renew build liveness, called between batches by [`run_backfill`]. + /// + /// A single-process backend proves liveness by its in-process registry and + /// needs nothing here (the default). A multi-process backend renews its + /// heartbeat so peers can tell a slow build from a dead one. + fn heartbeat(&mut self) -> impl Future> + Send { + async { Ok(()) } + } +} + +/// Backfill the index in independently committed batches. +/// +/// This is what lets the base table stay writable while an index builds, which is +/// how the service behaves: the table remains ACTIVE and accepts writes +/// throughout, and only the index reports CREATING. Holding one transaction for +/// the whole backfill would block every write until it finished. +/// +/// Releasing the write path between batches is also what creates the ordering +/// hazard this design has to answer. A write landing mid-backfill is enqueued, +/// and if it were applied before the backfill wrote its older snapshot of the +/// same item, the index would converge on the stale generation. The propagation +/// worker therefore refuses to claim any row for a table whose vector index is +/// still CREATING, so those writes accumulate and are applied only after the +/// index flips to ACTIVE. +/// +/// A crash part-way leaves the index in CREATING with some rows written, which +/// the backend's startup reconciler repairs by rebuilding ([`rebuild_index`]). +/// +/// `batch_delay` pauses between batches, outside any lock, so a write can +/// actually proceed during the pause. Zero in production; a test sets it so a +/// write is guaranteed to land mid-backfill. +/// +/// Exported deliberately, not by accident: it is the building block the two +/// drivers compose, and a backend orchestrating its own build task may call it +/// directly. A caller that does so owns the failure contract [`complete_build`] +/// otherwise provides: on error, leave the index CREATING and let recovery +/// rebuild it. +/// +/// # Errors +/// Propagates the first batch or heartbeat failure; rows already committed by +/// earlier batches stay in place for the recovery rebuild to supersede. +pub async fn run_backfill( + ops: &mut B, + batch_size: i64, + batch_delay: Duration, +) -> Result { + let mut cursor: Option = None; + let mut written = 0usize; + let mut skipped = 0usize; + loop { + ops.heartbeat().await?; + let outcome = ops.backfill_batch(cursor.take(), batch_size).await?; + written += outcome.written; + skipped += outcome.skipped; + if outcome.fetched < batch_size { + break; + } + let Some(next) = outcome.cursor else { + break; + }; + cursor = Some(next); + if !batch_delay.is_zero() { + tokio::time::sleep(batch_delay).await; + } + } + Ok(BackfillOutcome { written, skipped }) +} + +/// The detached build task's body: backfill, then publish or leave `CREATING`. +/// +/// Runs detached from the `UpdateTable` call, so the caller returns while the +/// index is still CREATING. The service behaves this way, and it is the whole +/// point: a table stays ACTIVE and writable throughout, and searches against +/// the index are refused until it is ACTIVE. +/// +/// Not awaited by anyone, so failures cannot be returned. They are logged and +/// the index is deliberately LEFT in CREATING, which is the state the backend's +/// recovery repairs (its stuck-build sweep at runtime, its reconciler at +/// startup). Flipping to ACTIVE on error would publish a partially populated +/// index, and there is no failure state on the wire for an index to sit in. +pub async fn complete_build( + mut ops: B, + index_name: &str, + batch_size: i64, + batch_delay: Duration, + earliest_active: Option, +) { + match run_backfill(&mut ops, batch_size, batch_delay).await { + Ok(outcome) => { + // The service's online-index machinery never finishes an added index + // instantly, so the CREATING walk (`Backfilling` false, then true, + // then ACTIVE with the member absent) is always observable there. A + // local backfill over a small table completes in milliseconds, which + // would collapse that walk into an instant no DescribeTable poll can + // catch, so the flip waits out the remainder of the caller's floor. + // Searches keep answering the documented not-ready rejection and + // writes keep queuing behind the CREATING hold for the duration, + // exactly as during a real backfill. `None` on the repair paths: + // recovery is not a client-observable creation and should publish as + // fast as batching allows. + if let Some(deadline) = earliest_active { + tokio::time::sleep_until(deadline).await; + } + match ops.mark_active(outcome.skipped).await { + Ok(()) => { + tracing::info!( + index_name = %index_name, + vectors_indexed = outcome.written, + vectors_skipped = outcome.skipped, + "vector index backfill complete" + ); + // Writes that landed during the backfill were held by the + // worker because this index was CREATING. It is ACTIVE + // now, so wake the worker rather than leaving them to sit + // until its next idle timeout. + ops.notify_active(); + } + Err(e) => tracing::error!( + index_name = %index_name, + "vector index backfill finished but the ACTIVE flip failed, \ + leaving it CREATING for startup reconciliation: {e}" + ), + } + } + Err(e) => tracing::error!( + index_name = %index_name, + "vector index backfill failed, leaving it CREATING for startup \ + reconciliation: {e}" + ), + } +} + +/// Drop, recreate, backfill, and flip one vector index to `ACTIVE`. +/// +/// The shared body of startup reconciliation and runtime stuck-build recovery, +/// factored so the two repairs cannot drift. The backfill runs on the batched +/// path, releasing the write path between batches, so a recovery on a large +/// table cannot become a write-availability outage. Recovery uses no batch +/// delay: the lever exists for tests, and recovery should finish as fast as +/// batching allows. +/// +/// Returns the number of rows written, which is what distinguishes "backfilled +/// nothing because no item carries the vector" from "backfilled nothing because +/// the scan is broken". +/// +/// Deliberately does NOT notify: the backend's recovery paths own when held +/// queue rows become claimable (startup reconciliation runs before the workers +/// exist; runtime recovery notifies once after a whole sweep). +/// +/// One deliberate exception to the status sequence in the module docs: a +/// rebuild does not re-assert `Backfilling: true`, so an index whose build died +/// before its own `set_backfilling` call is rebuilt while DescribeTable still +/// reports `false`. This matches the pre-extraction behavior; the flip to +/// `ACTIVE` clears the member either way. +/// +/// # Errors +/// Unlike [`complete_build`], every failure propagates, including the terminal +/// flip: the caller is a repair loop with its own retry story, not a detached +/// task. +pub async fn rebuild_index( + ops: &mut B, + batch_size: i64, +) -> Result { + ops.reset_data_table().await?; + let outcome = run_backfill(ops, batch_size, Duration::ZERO).await?; + ops.mark_active(outcome.skipped).await?; + Ok(outcome.written) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::{Arc, Mutex}; + + /// A scripted backend: each entry is one batch's outcome, and every + /// primitive call is recorded so the tests assert the driver's ordering + /// decisions rather than its side effects. + #[derive(Default)] + struct Script { + batches: Vec, StorageError>>, + flip_fails: bool, + log: Vec, + } + + #[derive(Clone, Default)] + struct MockBuild(Arc>); + + impl MockBuild { + fn log(&self) -> Vec { + self.0.lock().unwrap().log.clone() + } + } + + impl VectorIndexBuild for MockBuild { + type Cursor = u32; + + async fn backfill_batch( + &mut self, + cursor: Option, + limit: i64, + ) -> Result, StorageError> { + let mut s = self.0.lock().unwrap(); + s.log + .push(format!("batch(cursor={cursor:?}, limit={limit})")); + if s.batches.is_empty() { + return Err(StorageError::Internal("script exhausted".to_owned())); + } + s.batches.remove(0) + } + + async fn set_backfilling(&mut self) -> Result<(), StorageError> { + self.0 + .lock() + .unwrap() + .log + .push("set_backfilling".to_owned()); + Ok(()) + } + + async fn mark_active(&mut self, skipped: usize) -> Result<(), StorageError> { + let mut s = self.0.lock().unwrap(); + s.log.push(format!("mark_active(skipped={skipped})")); + if s.flip_fails { + return Err(StorageError::Internal("flip failed".to_owned())); + } + Ok(()) + } + + async fn reset_data_table(&mut self) -> Result<(), StorageError> { + self.0.lock().unwrap().log.push("reset".to_owned()); + Ok(()) + } + + fn notify_active(&mut self) { + self.0.lock().unwrap().log.push("notify".to_owned()); + } + } + + fn batch( + written: usize, + skipped: usize, + fetched: i64, + cursor: Option, + ) -> BatchOutcome { + BatchOutcome { + written, + skipped, + fetched, + cursor, + } + } + + /// The loop threads each batch's cursor into the next call, terminates on a + /// short batch, and sums both counters across batches. + #[tokio::test] + async fn run_backfill_threads_the_cursor_and_stops_on_a_short_batch() { + let mock = MockBuild::default(); + mock.0.lock().unwrap().batches = vec![ + Ok(batch(3, 0, 3, Some(30))), + Ok(batch(2, 1, 3, Some(60))), + Ok(batch(1, 0, 1, Some(70))), + ]; + let mut ops = mock.clone(); + let outcome = run_backfill(&mut ops, 3, Duration::ZERO) + .await + .expect("backfill"); + assert_eq!(outcome.written, 6); + assert_eq!(outcome.skipped, 1); + assert_eq!( + mock.log(), + vec![ + "batch(cursor=None, limit=3)", + "batch(cursor=Some(30), limit=3)", + "batch(cursor=Some(60), limit=3)", + ], + "each batch resumes from the previous batch's cursor, and the short \ + third batch ends the scan" + ); + } + + /// A successful build publishes with the summed skip count and only then + /// wakes the held queue rows. + #[tokio::test] + async fn complete_build_marks_active_then_notifies() { + let mock = MockBuild::default(); + mock.0.lock().unwrap().batches = + vec![Ok(batch(2, 1, 2, Some(2))), Ok(batch(0, 1, 0, None))]; + complete_build(mock.clone(), "vidx", 2, Duration::ZERO, None).await; + assert_eq!( + mock.log(), + vec![ + "batch(cursor=None, limit=2)", + "batch(cursor=Some(2), limit=2)", + "mark_active(skipped=2)", + "notify", + ], + "publish carries the total skip count, and the wake follows the flip" + ); + } + + /// A failed backfill leaves the index CREATING: no flip, no wake. That is + /// the repair contract the reconciler and the stuck-build sweep rely on. + #[tokio::test] + async fn complete_build_leaves_a_failed_build_in_creating() { + let mock = MockBuild::default(); + mock.0.lock().unwrap().batches = vec![Err(StorageError::Internal("scan died".to_owned()))]; + complete_build(mock.clone(), "vidx", 2, Duration::ZERO, None).await; + assert_eq!( + mock.log(), + vec!["batch(cursor=None, limit=2)"], + "neither mark_active nor notify may run after a failed backfill" + ); + } + + /// A failed ACTIVE flip must not wake the queue: the index is still + /// CREATING on disk, so a woken worker would find the hold still in place, + /// and notifying would misrepresent the build as published. + #[tokio::test] + async fn complete_build_does_not_notify_when_the_flip_fails() { + let mock = MockBuild::default(); + { + let mut s = mock.0.lock().unwrap(); + s.batches = vec![Ok(batch(1, 0, 0, None))]; + s.flip_fails = true; + } + complete_build(mock.clone(), "vidx", 2, Duration::ZERO, None).await; + assert_eq!( + mock.log(), + vec!["batch(cursor=None, limit=2)", "mark_active(skipped=0)"], + "no notify after a failed flip" + ); + } + + /// Recovery resets the data table BEFORE scanning (rebuild, not resume), + /// flips at the end, and reports rows written. The flip error propagates, + /// unlike the detached path. + #[tokio::test] + async fn rebuild_index_resets_first_and_propagates_a_flip_failure() { + let mock = MockBuild::default(); + mock.0.lock().unwrap().batches = vec![Ok(batch(4, 1, 0, None))]; + let mut ops = mock.clone(); + let written = rebuild_index(&mut ops, 500).await.expect("rebuild"); + assert_eq!(written, 4); + assert_eq!( + mock.log(), + vec![ + "reset", + "batch(cursor=None, limit=500)", + "mark_active(skipped=1)", + ], + "the reset precedes the scan, or already-written rows collide with \ + the backfill's plain INSERT" + ); + + let failing = MockBuild::default(); + { + let mut s = failing.0.lock().unwrap(); + s.batches = vec![Ok(batch(1, 0, 0, None))]; + s.flip_fails = true; + } + let mut ops = failing.clone(); + let err = rebuild_index(&mut ops, 500) + .await + .expect_err("flip failure"); + assert!( + matches!(err, StorageError::Internal(_)), + "the repair loop must see the flip failure: {err:?}" + ); + } +} diff --git a/crates/storage/src/vector_lifecycle/meta.rs b/crates/storage/src/vector_lifecycle/meta.rs new file mode 100644 index 00000000..5d7fe093 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/meta.rs @@ -0,0 +1,105 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The vector index description shared by the write path, the backfill, and +//! the propagation queue. + +use serde::{Deserialize, Serialize}; + +use extenddb_core::types::{AttributeDefinition, Item, KeySchemaElement}; + +use super::partition::partition_value; +use crate::error::StorageError; + +/// A vector index as the write path needs it. +/// +/// Serializable because the asynchronous path snapshots it verbatim into the +/// pending row's [`VectorApplyContext`]. The write path and the worker therefore +/// apply from the *same* description of the index, which is the property that stops +/// a queued write from being reinterpreted under a later definition. +/// +/// Lives in shared code because both the snapshot bytes and the projection rules +/// hanging off it must be identical across backends: a queue row written by one +/// build of one backend must stay readable, and a row shaped differently between +/// backends would search correctly right up until the difference mattered. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct VectorIndexMeta { + pub index_id: String, + pub dimensions: usize, + pub vector_attribute_name: String, + /// The index's projection, applied to the stored row exactly as the GSI path + /// applies its own. Not applying it was an unexplained divergence from the + /// sibling, and it made a search return attributes the index does not project. + pub projection: extenddb_core::types::Projection, + /// The single HASH element's attribute name, when the index declares one. + /// `None` means the index is unscoped and every row shares one partition. + pub hash_attribute_name: Option, + /// Every attribute named by the SearchSchema, HASH and INLINE_FILTER alike. + /// + /// These are projected regardless of `ProjectionType`, which is the documented + /// rule for a vector index and is NOT GSI `KEYS_ONLY` semantics: `KEYS_ONLY` + /// on a vector index projects the base primary key, the vector attribute and + /// the inline filter attributes. Withholding them is not merely a reporting + /// difference, it breaks search: the filter is evaluated against the stored + /// payload, so a missing filter attribute makes every row fail the predicate + /// and a filtered search match nothing. + pub search_schema_attribute_names: Vec, +} + +/// Everything the propagation worker needs to apply one vector index update, +/// serialized into the pending queue row's context column. +/// +/// `table_id` is carried here even though the queue row has a `table_id` column of +/// its own, because a vector data table is named from the table id *and* the index +/// id. Reading it from the context preserves the invariant that the context alone +/// is sufficient, rather than splitting one apply's inputs across a column and a +/// JSON blob. Both are written from the same variable in the same statement, so +/// they cannot disagree. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct VectorApplyContext { + pub base_key_schema: Vec, + pub attribute_definitions: Vec, + pub table_id: String, + /// Deliberately named `vector` rather than `index`: it is the field whose + /// presence lets an untagged pending-context enum tell a vector row from a + /// GSI row by shape alone. Backends discriminate queue rows by this shape, + /// so the field name is part of the on-disk format and must not change. + pub vector: VectorIndexMeta, +} + +/// Whether an item belongs in a vector index. +/// +/// It must carry the vector attribute, and the HASH attribute when the index +/// declares one: without the latter the row could not be placed in a partition, +/// and putting it in the unscoped partition would make it visible to searches of +/// every other partition. Not an error, exactly as a GSI silently omits an item +/// missing its index key. +#[must_use] +pub fn item_is_indexable(item: &Item, meta: &VectorIndexMeta) -> bool { + if !item.contains_key(&meta.vector_attribute_name) { + return false; + } + match &meta.hash_attribute_name { + Some(name) => item.contains_key(name), + None => true, + } +} + +/// The partition column value for an item under one index. +/// +/// # Errors +/// Fails when the indexable check was bypassed (the HASH attribute is absent) or +/// the attribute value cannot be encoded. +pub fn item_partition(item: &Item, meta: &VectorIndexMeta) -> Result { + match &meta.hash_attribute_name { + Some(name) => { + let value = item.get(name).ok_or_else(|| { + StorageError::Internal( + "indexable check passed but the hash attribute is absent".to_owned(), + ) + })?; + partition_value(Some((name.as_str(), value))) + } + None => partition_value(None), + } +} diff --git a/crates/storage/src/vector_lifecycle/mod.rs b/crates/storage/src/vector_lifecycle/mod.rs new file mode 100644 index 00000000..63f310f3 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/mod.rs @@ -0,0 +1,82 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Shared vector index-build lifecycle, owned here so every backend runs the +//! same state machine. +//! +//! `docs/adr/0005-index-build-lifecycle-ownership.md` committed to this +//! extraction: the lifecycle was allowed to live in `storage-sqlite` only until +//! a second backend implemented `VectorSearchEngine`, and that backend MUST NOT +//! re-implement it. This module is the extraction. A backend supplies storage +//! primitives through [`VectorIndexBuild`]; the ordering rules, the poison-row +//! semantics, and the crash-recovery contract live here and cannot drift +//! per backend. +//! +//! # The lifecycle contract +//! +//! The observable behaviour was measured against Amazon DynamoDB (2026-08-06) +//! and is pinned by the SQLite backend's wire tests: +//! +//! 1. **Status sequence.** An index created by `UpdateTable` appears as +//! `CREATING` with `Backfilling: false`, flips to `Backfilling: true` when +//! the scan starts ([`VectorIndexBuild::set_backfilling`]), and becomes +//! `ACTIVE` with the `Backfilling` member absent in a single transition +//! ([`VectorIndexBuild::mark_active`]). An index created by `CreateTable` +//! skips the sequence: the table is empty, so it is `ACTIVE` from birth. +//! 2. **The table stays writable throughout.** The backfill commits in +//! independent batches ([`run_backfill`]) rather than holding one +//! transaction, and the build task is detached from the `UpdateTable` +//! call ([`complete_build`] is the task body). +//! 3. **Write ordering during the backfill.** A write that lands while the +//! index is `CREATING` must not reach the index's data table before the +//! backfill's (older) snapshot of the same item does, or the index converges +//! on the stale generation. Each backend enforces this with a claim-time +//! hold on its propagation queue: rows for a table whose vector index is +//! `CREATING` are not claimed, and [`VectorIndexBuild::notify_active`] wakes +//! the worker to replay them once the index is published. The hold is per +//! TABLE, not per index, so a GSI row and a vector row for one item keep +//! their relative order. +//! 4. **Poison rows are skipped and counted, never fatal.** Rows written +//! before the index existed never passed vector validation, so a malformed +//! or wrong-dimension stored vector is expected backfill input, not a bug. +//! [`classify_backfill_row`] owns that classification; the count is +//! recorded on the catalog row at the `ACTIVE` flip. Transient errors still +//! propagate and abort the batch pre-commit. +//! 5. **Failure leaves the index `CREATING`.** There is no failure state on +//! the wire, and flipping to `ACTIVE` would publish a partially populated +//! index. A build that dies is repaired by rebuilding: drop the data table, +//! recreate it, backfill, flip ([`rebuild_index`]). Rebuilding rather than +//! resuming, because rows already written would collide with the backfill's +//! deliberately plain INSERT. +//! 6. **Build ownership is backend-defined.** Recovery must not rebuild an +//! index whose build is still making progress. A single-process backend can +//! prove liveness with an in-process registry; a multi-process backend +//! needs cross-process ownership (an advisory lock and a heartbeat column +//! renewed via [`VectorIndexBuild::heartbeat`]). The candidate-selection +//! policy therefore stays in the backend; the repair it triggers is the +//! shared [`rebuild_index`]. +//! +//! # What stays in the backend +//! +//! SQL and transactions (the primitives behind [`VectorIndexBuild`]), the +//! backfill cursor type (SQLite scans by `rowid`; a backend without rowids +//! uses keyset pagination over the full primary key), the queue hold's claim +//! predicate, and the stuck-build detection policy. The write-path maintenance +//! entry point also stays per backend, built on the shared helpers here +//! ([`item_is_indexable`], [`item_partition`], [`projected_payload`], +//! [`VectorApplyContext`]) so the row shape cannot drift between a live write +//! and a backfill. + +mod backfill; +mod build; +mod meta; +mod partition; +mod payload; + +pub use backfill::{ + BACKFILL_BATCH, BackfillOutcome, BackfillRow, BatchOutcome, classify_backfill_row, +}; +pub use build::{VectorIndexBuild, complete_build, rebuild_index, run_backfill}; +pub use meta::{VectorApplyContext, VectorIndexMeta, item_is_indexable, item_partition}; +pub use partition::{UNSCOPED_PARTITION, partition_value}; +pub use payload::projected_payload; diff --git a/crates/storage/src/vector_lifecycle/partition.rs b/crates/storage/src/vector_lifecycle/partition.rs new file mode 100644 index 00000000..80a44779 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/partition.rs @@ -0,0 +1,85 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Partition scoping for vector index rows. +//! +//! Moved verbatim from the SQLite backend when the lifecycle was extracted +//! (ADR-0005): the encoding must stay byte-identical across backends, because a +//! value written by one backend's write path and a value derived from a search +//! request must compare equal. + +use extenddb_core::types::AttributeValue; + +use crate::error::StorageError; +use crate::util::pk_to_text; + +/// Partition value for an index that declares no HASH element. +/// +/// Such an index searches the whole table, so every row shares one partition +/// rather than the scan needing a second code path. +/// +/// What makes this safe is **not** that the value is unguessable. `pk_to_text` +/// stores an `S` attribute verbatim, so a caller could supply this exact string as +/// a partition key. The guarantee is structural instead: the partition is chosen +/// from the *index's* schema, not from the item, and each index has its own data +/// table. So within one table either every row is keyed by a real hash value (the +/// index declares a HASH element) or every row uses this sentinel (it does not). +/// The two never coexist, so there is nothing for a collision to leak into. +/// +/// The leading NUL is defence in depth for the day that invariant changes, for +/// instance if several indexes ever shared one table. It is deliberately not what +/// correctness rests on, because a reader who believed it was would then feel free +/// to weaken it. +pub const UNSCOPED_PARTITION: &str = "\u{0}all"; + +/// The partition column value for a vector row. +/// +/// Uses the same `pk_to_text` encoding as item partition keys, so a value written +/// by the write path and a value derived from a search request are byte-identical. +/// Getting this wrong would not fail loudly: it would silently return no hits. +/// +/// # Errors +/// Fails when the attribute value cannot be encoded as a partition key text. +pub fn partition_value(hash_key: Option<(&str, &AttributeValue)>) -> Result { + match hash_key { + Some((_, value)) => Ok(pk_to_text(value)?.into_owned()), + None => Ok(UNSCOPED_PARTITION.to_owned()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The partition is chosen from the index's schema, never from the item, which + /// is the invariant that makes the sentinel safe. Asserted as the property + /// rather than against one example: the previous version compared the sentinel + /// to `pk_to_text(S("all"))` only, and passed unchanged when the sentinel was + /// weakened to the ordinary string `"unscoped"`. + #[test] + fn the_partition_comes_from_the_index_schema_not_the_item() { + // No HASH element declared: the sentinel, whatever the item holds. + assert_eq!(partition_value(None).unwrap(), UNSCOPED_PARTITION); + + // A HASH element declared: the item's value, verbatim for S. + for value in ["all", "unscoped", UNSCOPED_PARTITION, ""] { + let scoped = + partition_value(Some(("pk", &AttributeValue::S(value.to_owned())))).unwrap(); + assert_eq!( + scoped, value, + "a scoped partition must be the attribute value itself" + ); + } + } + + /// The sentinel keeps a leading NUL as defence in depth. Not what correctness + /// rests on (see the constant's documentation), but weakening it to an ordinary + /// string should break a test rather than pass silently. + #[test] + fn the_unscoped_sentinel_keeps_its_unusual_prefix() { + assert!( + UNSCOPED_PARTITION.starts_with('\0'), + "sentinel must keep its NUL prefix: {UNSCOPED_PARTITION:?}" + ); + } +} diff --git a/crates/storage/src/vector_lifecycle/payload.rs b/crates/storage/src/vector_lifecycle/payload.rs new file mode 100644 index 00000000..3e65d031 --- /dev/null +++ b/crates/storage/src/vector_lifecycle/payload.rs @@ -0,0 +1,70 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The stored payload of one vector index row. + +use extenddb_core::types::{Item, KeySchemaElement, ProjectionType}; + +use super::meta::VectorIndexMeta; + +/// Build the payload stored alongside one indexed vector. +/// +/// Shared by the write path and by backfill deliberately, across backends. These +/// are the only producers of a vector row, and a second copy of this logic would +/// be free to drift: a backfilled row shaped differently from a live-written one +/// would search correctly right up until the difference mattered, with nothing to +/// catch it. +/// +/// Three rules, in order: +/// +/// 1. The item is projected per the index's projection, exactly as the GSI path +/// projects its own (a vector index has no key schema of its own, so +/// `KEYS_ONLY` and `INCLUDE` start from the base primary key). +/// 2. The SearchSchema attributes are always projected, whatever the +/// `ProjectionType`. See [`VectorIndexMeta::search_schema_attribute_names`] +/// for why: the inline filter is evaluated against this payload, so dropping +/// the attribute would silently turn every filtered search into a zero-result +/// search. +/// 3. The vector itself is not kept in the payload: it is already stored as +/// `f32`, which is the width the service validates against, and the search +/// path rebuilds the attribute from those bits. Keeping a verbatim decimal +/// copy here duplicated 10 to 15 KB per row at 1024 dimensions and would have +/// returned the client's original precision where the service returns the +/// narrowed value. +#[must_use] +pub fn projected_payload( + item: &Item, + base_key_schema: &[KeySchemaElement], + meta: &VectorIndexMeta, +) -> Item { + let mut projected = match meta.projection.projection_type { + ProjectionType::All => item.clone(), + ProjectionType::KeysOnly | ProjectionType::Include => { + let mut projected = Item::new(); + for ks in base_key_schema { + if let Some(v) = item.get(&ks.attribute_name) { + projected.insert(ks.attribute_name.clone(), v.clone()); + } + } + if meta.projection.projection_type == ProjectionType::Include + && let Some(attrs) = &meta.projection.non_key_attributes + { + for attr in attrs { + if let Some(v) = item.get(attr) { + projected.insert(attr.clone(), v.clone()); + } + } + } + projected + } + }; + for name in &meta.search_schema_attribute_names { + if !projected.contains_key(name) + && let Some(v) = item.get(name) + { + projected.insert(name.clone(), v.clone()); + } + } + projected.remove(&meta.vector_attribute_name); + projected +} diff --git a/docs/adr/0005-index-build-lifecycle-ownership.md b/docs/adr/0005-index-build-lifecycle-ownership.md index 87dfc6e8..f8795eda 100644 --- a/docs/adr/0005-index-build-lifecycle-ownership.md +++ b/docs/adr/0005-index-build-lifecycle-ownership.md @@ -1,6 +1,6 @@ # ADR-0005: Index-build lifecycle stays in the backend until a second backend needs it -- Status: Proposed +- Status: Accepted - Date: 2026-08-14 - Deciders: @LeeroyHannigan @@ -116,6 +116,18 @@ the next open is the layer below the property this pins. all-0xFF bound defect in particular) are pre-existing defects independent of this decision and need their own issues. +## Outcome (2026-08-19) + +The condition this decision waited on arrived: the PostgreSQL backend is the +second implementor of `VectorSearchEngine`, and the extraction landed as the +first step of that port, exactly as the Decision section prescribes. The +lifecycle now lives in `crates/storage/src/vector_lifecycle/` (the +`VectorIndexBuild` primitives trait plus the shared backfill, publish, and +rebuild drivers), with the SQLite backend rewired as the first implementor and +its pre-existing vector tests, unchanged, as the acceptance gate. The +procedural commitment above stands: reviewers reject any second copy of the +lifecycle, which now means any backend logic that bypasses the shared module. + --- ## License diff --git a/docs/technical-debt.md b/docs/technical-debt.md index 70a74f47..3cb055de 100755 --- a/docs/technical-debt.md +++ b/docs/technical-debt.md @@ -1,6 +1,6 @@ # Technical Debt Tracker -Last updated: 2026-05-04 (P112) +Last updated: 2026-08-19 ## Categories @@ -30,6 +30,7 @@ Last updated: 2026-05-04 (P112) | F-15 | ~~TTL worker bypasses stream capture — expired item deletions don't generate REMOVE stream records~~ | `bin/cmd_serve.rs:ttl_cleanup_worker` | ~~High~~ | P26 | | F-16 | `transact_write_items.rs` passes `None` for `old_item` in stream capture — `OldImage` always `None` for transaction-originated stream records | `engine/transact_write_items.rs` | Medium | P27 | | F-17 | `validate_attribute_name_sizes` only checks top-level attribute names — nested map keys not validated | `core/validation/mod.rs` | Low | P30 | +| F-18 | UpdateTable Delete of a vector index in the resource-allocation phase (`CREATING`, `Backfilling: false`) is accepted; Amazon DynamoDB refuses it with `ResourceInUseException` until backfilling starts. The message constant, `StorageError::ResourceInUse`, and the engine mapping arms are in place with no runtime producer; enforcement belongs in the shared lifecycle, deferred so the extraction stayed behavior-preserving | `storage-sqlite/update_table.rs` (vector delete branch), `core/types/table.rs` (`vector_index_delete_in_allocation_phase`) | Medium | vector probe P2 | ## Cleanup diff --git a/tests/rust/src/vector_index_search.rs b/tests/rust/src/vector_index_search.rs index d031b28b..4d0e376d 100644 --- a/tests/rust/src/vector_index_search.rs +++ b/tests/rust/src/vector_index_search.rs @@ -1918,3 +1918,339 @@ async fn if_not_exists_cannot_smuggle_a_malformed_vector() { let _ = call("DeleteTable", &format!(r#"{{"TableName": "{name}"}}"#)).await; } + +/// Assert a 400 `ValidationException` whose message is the whole measured string. +/// +/// Whole-string, because every wording quirk in this family is load-bearing and a +/// `contains` assertion is exactly what let three of them drift. +fn assert_validation_message(status: u16, body: &str, expected: &str) { + assert_eq!(status, 400, "expected HTTP 400, body: {body}"); + let json: serde_json::Value = serde_json::from_str(body).expect("body is JSON"); + let type_field = json + .get("__type") + .and_then(|t| t.as_str()) + .unwrap_or_else(|| panic!("no __type in body: {body}")); + assert!( + type_field.ends_with("ValidationException"), + "expected ValidationException, got {type_field} (body: {body})" + ); + let message = json + .get("message") + .or_else(|| json.get("Message")) + .and_then(|m| m.as_str()) + .unwrap_or_else(|| panic!("no message in body: {body}")); + assert_eq!(message, expected); +} + +/// `SearchVectors` reports its charge under both measured member names. +/// +/// Measured on 2026-08-19 (probe P8) against real Amazon DynamoDB: the response +/// carries `{"VectorSearchRequestBytes": N, "VectorSearchUnits": N}` with the two +/// always equal, under `INDEXES` and `TOTAL` alike, and carries neither +/// `TableName` nor `CapacityUnits`. ExtendDB emitted the bytes member alone, so a +/// client reading the units member got `null` where the service gives a number. +/// +/// The absent members are asserted as well as the present ones: a `ConsumedCapacity` +/// built from the ordinary table-capacity shape would satisfy a present-members-only +/// check while returning two members no vector search ever returns. +#[tokio::test] +async fn the_search_charge_reports_both_measured_capacity_members() { + if skip_unless_supported().await { + return; + } + let name = table_name("vcap_shape"); + create_vector_table(&name, 4, "COSINE", false).await; + put_vector(&name, "a", None, &[1.0, 0.0, 0.0, 0.0]).await; + put_vector(&name, "b", None, &[0.0, 1.0, 0.0, 0.0]).await; + // Converge first, so the charge is measured against a populated result set. + search_until_count(&name, &[1.0, 0.0, 0.0, 0.0], 2, None, 2).await; + + for granularity in ["INDEXES", "TOTAL"] { + let body = format!( + r#"{{ + "TableName": "{name}", + "IndexName": "vidx", + "SearchVector": [{{"N": "1"}}, {{"N": "0"}}, {{"N": "0"}}, {{"N": "0"}}], + "TopK": 2, + "ReturnConsumedCapacity": "{granularity}" + }}"# + ); + let (status, text) = call("SearchVectors", &body).await; + assert_eq!(status, 200, "SearchVectors failed: {text}"); + let json: serde_json::Value = serde_json::from_str(&text).expect("JSON"); + let capacity = json + .get("ConsumedCapacity") + .unwrap_or_else(|| panic!("no ConsumedCapacity at {granularity}: {text}")); + let mut members: Vec<&str> = capacity + .as_object() + .unwrap_or_else(|| panic!("ConsumedCapacity is not an object: {text}")) + .keys() + .map(String::as_str) + .collect(); + members.sort_unstable(); + assert_eq!( + members, + ["VectorSearchRequestBytes", "VectorSearchUnits"], + "at {granularity}, the search charge must carry exactly the two \ + measured members: {text}" + ); + let bytes = capacity["VectorSearchRequestBytes"] + .as_f64() + .unwrap_or_else(|| panic!("VectorSearchRequestBytes is not a number: {text}")); + let units = capacity["VectorSearchUnits"] + .as_f64() + .unwrap_or_else(|| panic!("VectorSearchUnits is not a number: {text}")); + assert!( + (bytes - units).abs() < f64::EPSILON, + "the two members must be equal, got {bytes} and {units}: {text}" + ); + assert!( + bytes >= 1024.0, + "the measured floor is 1024, got {bytes}: {text}" + ); + } + + let _ = call("DeleteTable", &format!(r#"{{"TableName": "{name}"}}"#)).await; +} + +/// Every invalid search-vector component returns one measured whole string. +/// +/// Probe P4, 2026-08-19: `NaN`, `Infinity` and a value outside the f32 range all +/// produce the identical message on the search path, so the caller cannot tell +/// the three causes apart. ExtendDB returned its first sentence only. +/// +/// All three inputs are exercised rather than one, because they take different +/// code paths (parse failure, non-finite parse, out-of-range parse) and the +/// service collapses them deliberately. +#[tokio::test] +async fn an_invalid_search_vector_reports_the_measured_whole_string() { + if skip_unless_supported().await { + return; + } + let name = table_name("vsearch_invalid"); + create_vector_table(&name, 2, "COSINE", false).await; + put_vector(&name, "a", None, &[1.0, 0.0]).await; + + for component in ["NaN", "Infinity", "3.5E38"] { + let body = format!( + r#"{{ + "TableName": "{name}", + "IndexName": "vidx", + "SearchVector": [{{"N": "{component}"}}, {{"N": "0"}}], + "TopK": 1 + }}"# + ); + let (status, text) = call("SearchVectors", &body).await; + assert_validation_message( + status, + &text, + "Search vector contains invalid values. All values in the search vector must be a \ + 32-bit floating-point number attribute", + ); + } + + let _ = call("DeleteTable", &format!(r#"{{"TableName": "{name}"}}"#)).await; +} + +/// The write path's three vector rejections, each as a whole measured string. +/// +/// Probes P4, P5, P9 and P10 on 2026-08-19, all against index `vidx` on +/// attribute `emb`, which is what this suite's fixture builds, so these compare +/// byte for byte against the captured wire responses rather than against a +/// re-templated form of them. PR 244 recorded wrong-dimension wire coverage as a +/// known gap; this closes it. +/// +/// Both directions of the size error are asserted (too short and too long) and +/// UpdateItem as well as PutItem, because the service was measured to use one +/// template for all four and a per-path template is the natural way to get it +/// wrong. +#[tokio::test] +async fn an_invalid_written_vector_reports_the_measured_whole_strings() { + if skip_unless_supported().await { + return; + } + let name = table_name("vwrite_invalid"); + create_vector_table(&name, 4, "COSINE", false).await; + put_vector(&name, "indexed1", None, &[0.6, 0.8, 0.0, 0.0]).await; + + let put = |pk: &str, value: &str| { + let body = format!( + r#"{{ + "TableName": "{name}", + "Item": {{"pk": {{"S": "{pk}"}}, "emb": {value}}} + }}"# + ); + async move { call("PutItem", &body).await } + }; + + // Too short. + let (status, text) = put( + "short", + r#"{"L": [{"N": "0.1"}, {"N": "0.2"}, {"N": "0.3"}]}"#, + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. \ + Invalid size for parameter emb, Expected: 4, Actual: 3 IndexName: vidx", + ); + + // Too long: same template, different count. + let (status, text) = put( + "long", + r#"{"L": [{"N": "0.1"}, {"N": "0.2"}, {"N": "0.3"}, {"N": "0.4"}, {"N": "0.5"}]}"#, + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. \ + Invalid size for parameter emb, Expected: 4, Actual: 5 IndexName: vidx", + ); + + // Wrong attribute type: a String where the index expects a list. Sparse + // semantics cover a MISSING attribute only, so this is a refused write. + let (status, text) = put("strattr", r#"{"S": "not-a-vector"}"#).await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. \ + Invalid type for parameter emb, Expected: 32-bit floating point number list \ + IndexName: vidx", + ); + + // A valid DynamoDB number that no f32 can hold. The service echoes it in its + // own normalised form, 3.5E+38 for the submitted 3.5E38. + let (status, text) = put( + "overflow", + r#"{"L": [{"N": "0"}, {"N": "3.5E38"}, {"N": "0"}, {"N": "0"}]}"#, + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. \ + Invalid value for parameter emb[1], Value: 3.5E+38 is outside valid range \ + [-3.4028235E38, 3.4028235E38]. IndexName: vidx", + ); + + // A wrong-typed element INSIDE the list. Same single-sentence envelope as + // every kind above since the 2026-08-27 measurement: envelopes are + // region-uniform (us single, eu doubled the same day) and these strings pin + // the us shape. + let (status, text) = put( + "badelem", + r#"{"L": [{"N": "0.1"}, {"S": "x"}, {"N": "0"}, {"N": "0"}]}"#, + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. Invalid type for parameter emb[1], \ + Expected: 32-bit floating point number, Actual: S. IndexName: vidx", + ); + + // BOOL, the measured case that shows the token is not limited to key types. + let (status, text) = put( + "boolelem", + r#"{"L": [{"N": "0.1"}, {"BOOL": true}, {"N": "0"}, {"N": "0"}]}"#, + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. Invalid type for parameter emb[1], \ + Expected: 32-bit floating point number, Actual: BOOL. IndexName: vidx", + ); + + // UpdateItem shares PutItem's wording exactly, on an item already indexed. + let (status, text) = call( + "UpdateItem", + &format!( + r#"{{"TableName": "{name}", "Key": {{"pk": {{"S": "indexed1"}}}}, + "UpdateExpression": "SET emb = :v", + "ExpressionAttributeValues": {{":v": {{"L": [{{"N": "0.1"}}, {{"N": "0.2"}}, {{"N": "0.3"}}]}}}}}}"# + ), + ) + .await; + assert_validation_message( + status, + &text, + "One or more parameter values were invalid. \ + Invalid size for parameter emb, Expected: 4, Actual: 3 IndexName: vidx", + ); + + let _ = call("DeleteTable", &format!(r#"{{"TableName": "{name}"}}"#)).await; +} + +/// A transaction carrying one invalid vector cancels per item, with the message +/// PutItem would have given. +/// +/// Probe P5 measured the split by validation LAYER, not by API: a +/// vector-semantic failure (wrong size, wrong type, out of f32 range) comes back +/// as `TransactionCanceledException` with `CancellationReasons` in item order, +/// `[None, ValidationError]`, and the failing item's `Message` byte-identical to +/// the standalone PutItem refusal. A request-deserialization failure such as +/// `NaN` fails the whole request top-level instead, which is a different layer +/// and is not asserted here. +/// +/// The valid sibling operation is asserted as `None` rather than ignored: a +/// per-request refusal would report no reasons at all, and a wrongly ordered +/// reasons list is the other way this goes wrong. +#[tokio::test] +async fn a_transaction_reports_an_invalid_vector_per_item() { + if skip_unless_supported().await { + return; + } + let name = table_name("vtx_invalid"); + create_vector_table(&name, 4, "COSINE", false).await; + + let (status, text) = call( + "TransactWriteItems", + &format!( + r#"{{"TransactItems": [ + {{"Put": {{"TableName": "{name}", "Item": {{"pk": {{"S": "valid"}}, + "emb": {{"L": [{{"N": "0.6"}}, {{"N": "0.8"}}, {{"N": "0"}}, {{"N": "0"}}]}}}}}}}}, + {{"Put": {{"TableName": "{name}", "Item": {{"pk": {{"S": "short"}}, + "emb": {{"L": [{{"N": "0.1"}}, {{"N": "0.2"}}, {{"N": "0.3"}}]}}}}}}}} + ]}}"# + ), + ) + .await; + assert_eq!(status, 400, "expected the transaction to cancel: {text}"); + let json: serde_json::Value = serde_json::from_str(&text).expect("body is JSON"); + let type_field = json + .get("__type") + .and_then(|t| t.as_str()) + .unwrap_or_else(|| panic!("no __type in body: {text}")); + assert!( + type_field.ends_with("TransactionCanceledException"), + "expected TransactionCanceledException, got {type_field}: {text}" + ); + let reasons = json + .get("CancellationReasons") + .and_then(|r| r.as_array()) + .unwrap_or_else(|| panic!("no CancellationReasons: {text}")); + assert_eq!(reasons.len(), 2, "one reason per item: {text}"); + assert_eq!( + reasons[0].get("Code").and_then(|c| c.as_str()), + Some("None"), + "the valid item must report None: {text}" + ); + assert_eq!( + reasons[1].get("Code").and_then(|c| c.as_str()), + Some("ValidationError"), + "the invalid item must report ValidationError: {text}" + ); + assert_eq!( + reasons[1].get("Message").and_then(|m| m.as_str()), + Some( + "One or more parameter values were invalid. Invalid size for parameter emb, \ + Expected: 4, Actual: 3 IndexName: vidx" + ), + "the per-item message must equal the PutItem refusal: {text}" + ); + + let _ = call("DeleteTable", &format!(r#"{{"TableName": "{name}"}}"#)).await; +} diff --git a/tests/rust/src/vector_index_update_rules.rs b/tests/rust/src/vector_index_update_rules.rs index 400a454f..9bdef4c6 100644 --- a/tests/rust/src/vector_index_update_rules.rs +++ b/tests/rust/src/vector_index_update_rules.rs @@ -281,3 +281,71 @@ async fn swap_in_one_call_is_refused() { delete_table(&name).await; } + +/// Adding a vector index to a table that is already PROVISIONED is refused, and +/// the refusal is evaluated on the request's NET billing mode. +/// +/// Measured 2026-08-19 (probe P12) against a live PROVISIONED table. Two facts, +/// both asserted here: +/// +/// * `UpdateTable` with a vector-index Create and no `BillingMode` member is +/// refused with the same whole string `CreateTable` returns, so one constant +/// serves both paths and both directions of the rule. +/// * The same request plus `BillingMode: PAY_PER_REQUEST` is accepted. The check +/// reads the request's net state, not the table's stored mode, which is the +/// same net-effect evaluation the index-count limit uses. +/// +/// The accepted case is the discriminating one: a guard written against the +/// stored mode passes the refusal half and wrongly refuses this half. +#[tokio::test] +async fn adding_a_vector_index_to_a_provisioned_table_is_rejected() { + if skip_unless_supported().await { + return; + } + let name = table_name("vupd-prov-add"); + let create = format!( + r#"{{ + "TableName": "{name}", + "AttributeDefinitions": [{{"AttributeName": "pk", "AttributeType": "S"}}], + "KeySchema": [{{"AttributeName": "pk", "KeyType": "HASH"}}], + "BillingMode": "PROVISIONED", + "ProvisionedThroughput": {{"ReadCapacityUnits": 1, "WriteCapacityUnits": 1}} + }}"# + ); + let (status, text) = call("CreateTable", &create).await; + assert_eq!(status, 200, "provisioned setup table failed: {text}"); + wait_for_active(&name).await; + + let (status, text) = call("UpdateTable", &create_update_body(&name, "vidx", "emb")).await; + assert_error( + status, + &text, + "ValidationException", + "One or more parameter values were invalid: Vector indexes are only supported for \ + PAY_PER_REQUEST tables", + ); + + // The net-state case: switch to on-demand and create the index in one call. + let combined = format!( + r#"{{ + "TableName": "{name}", + "BillingMode": "PAY_PER_REQUEST", + "VectorIndexUpdates": [{{"Create": {{ + "IndexName": "vidx", + "VectorAttribute": {{"AttributeName": "emb"}}, + "Dimensions": 4, + "DistanceFunction": "COSINE", + "Projection": {{"ProjectionType": "ALL"}} + }}}}] + }}"# + ); + let (status, text) = call("UpdateTable", &combined).await; + assert_eq!( + status, 200, + "a switch to PAY_PER_REQUEST and a create in one request must be accepted: {text}" + ); + + // Wait the index out before deleting, so the table is not left mid-build. + wait_for_vector_index_active(&name, "vidx").await; + delete_table(&name).await; +}