diff --git a/bottomless-cli/src/replicator_extras.rs b/bottomless-cli/src/replicator_extras.rs index 94670bd07f..9586020c54 100644 --- a/bottomless-cli/src/replicator_extras.rs +++ b/bottomless-cli/src/replicator_extras.rs @@ -382,9 +382,8 @@ impl Replicator { let frame = tokio::fs::File::open(&obj).await?; let frame_buf_reader = BufReader::new(frame); - let mut frameno = first_frame_no; let mut reader = bottomless::read::BatchReader::new( - frameno, + first_frame_no, frame_buf_reader, page_size as usize, compression_kind, @@ -411,7 +410,6 @@ impl Replicator { ); pending_pages.flush(db).await?; } - frameno += 1; last_received_frame_no += 1; } db.flush().await?; diff --git a/libsql-ffi/bundled/SQLite3MultipleCiphers/src/sqlite3.c b/libsql-ffi/bundled/SQLite3MultipleCiphers/src/sqlite3.c index 8dc3ca8e72..4147f83b82 100644 --- a/libsql-ffi/bundled/SQLite3MultipleCiphers/src/sqlite3.c +++ b/libsql-ffi/bundled/SQLite3MultipleCiphers/src/sqlite3.c @@ -211731,7 +211731,7 @@ static int vectorParseSqliteText( continue; } if( this != ',' && this != ']' ){ - if( iBuf > MAX_FLOAT_CHAR_SZ ){ + if( iBuf >= MAX_FLOAT_CHAR_SZ ){ *pzErrMsg = sqlite3_mprintf("vector: float string length exceeded %d characters: '%s'", MAX_FLOAT_CHAR_SZ, valueBuf); goto error; } diff --git a/libsql-ffi/bundled/src/sqlite3.c b/libsql-ffi/bundled/src/sqlite3.c index 8dc3ca8e72..4147f83b82 100644 --- a/libsql-ffi/bundled/src/sqlite3.c +++ b/libsql-ffi/bundled/src/sqlite3.c @@ -211731,7 +211731,7 @@ static int vectorParseSqliteText( continue; } if( this != ',' && this != ']' ){ - if( iBuf > MAX_FLOAT_CHAR_SZ ){ + if( iBuf >= MAX_FLOAT_CHAR_SZ ){ *pzErrMsg = sqlite3_mprintf("vector: float string length exceeded %d characters: '%s'", MAX_FLOAT_CHAR_SZ, valueBuf); goto error; } diff --git a/libsql-server/src/query_analysis.rs b/libsql-server/src/query_analysis.rs index 5762b86d4c..5478047eba 100644 --- a/libsql-server/src/query_analysis.rs +++ b/libsql-server/src/query_analysis.rs @@ -233,7 +233,7 @@ impl StmtKind { } } -fn to_ascii_lower(s: &str) -> Cow { +fn to_ascii_lower(s: &str) -> Cow<'_, str> { if s.chars().all(|c| char::is_ascii_lowercase(&c)) { Cow::Borrowed(s) } else { diff --git a/libsql-sqlite3/src/vector.c b/libsql-sqlite3/src/vector.c index 51f8af5d05..ee1c025528 100644 --- a/libsql-sqlite3/src/vector.c +++ b/libsql-sqlite3/src/vector.c @@ -217,7 +217,7 @@ static int vectorParseSqliteText( continue; } if( this != ',' && this != ']' ){ - if( iBuf > MAX_FLOAT_CHAR_SZ ){ + if( iBuf >= MAX_FLOAT_CHAR_SZ ){ *pzErrMsg = sqlite3_mprintf("vector: float string length exceeded %d characters: '%s'", MAX_FLOAT_CHAR_SZ, valueBuf); goto error; } diff --git a/libsql-sqlite3/test/libsql_vector.test b/libsql-sqlite3/test/libsql_vector.test index 793358e068..3cfe69fedc 100644 --- a/libsql-sqlite3/test/libsql_vector.test +++ b/libsql-sqlite3/test/libsql_vector.test @@ -201,6 +201,56 @@ do_execsql_test vector-1-conversion-f8 { {[-20,-35.25,1.0625,1.63281,2.20313,2.76563,10.1875,99.5,104.5,110]} A0C10DC2883FD13F0D4031402341C742D142DC4206 } +foreach {name sql} { + vector {SELECT vector_extract(vector($input)) = vector_extract(vector('[1]'))} + vector32 {SELECT vector_extract(vector32($input)) = vector_extract(vector32('[1]'))} + vector64 {SELECT vector_extract(vector64($input)) = vector_extract(vector64('[1]'))} + vector8 {SELECT vector_extract(vector8($input)) = vector_extract(vector8('[1]'))} + vector16 {SELECT vector_extract(vector16($input)) = vector_extract(vector16('[1]'))} + vectorb16 {SELECT vector_extract(vectorb16($input)) = vector_extract(vectorb16('[1]'))} + vector1bit {SELECT vector_extract(vector1bit($input)) = vector_extract(vector1bit('[1]'))} + extract {SELECT vector_extract($input) = '[1]'} + cos-left {SELECT vector_distance_cos($input, '[1]') = 0} + cos-right {SELECT vector_distance_cos('[1]', $input) = 0} + l2-left {SELECT vector_distance_l2($input, '[1]') = 0} + l2-right {SELECT vector_distance_l2('[1]', $input) = 0} +} { + foreach length {1023 1024} { + set element "[string repeat 0 [expr {$length - 1}]]1" + set input [format {[%s]} $element] + do_execsql_test vector-1-text-length-$name-$length $sql {1} + } + + foreach length {1025 1026 5000} { + set input [format {[%s]} [string repeat 1 $length]] + do_catchsql_test vector-1-text-length-$name-$length $sql [list 1 \ + "vector: float string length exceeded 1024 characters: '[string repeat 1 1024]'" + ] + } + + set element "1[string repeat x 1023]" + set input [format {[%s]} $element] + do_catchsql_test vector-1-text-length-$name-invalid-1024 $sql [list 1 \ + "vector: invalid float at position 0: '$element'" + ] + + set input [format {[%sx]} $element] + do_catchsql_test vector-1-text-length-$name-invalid-1025 $sql [list 1 \ + "vector: float string length exceeded 1024 characters: '$element'" + ] +} + +set element "[string repeat 0 1023]1" +set input [format {[%s,%s]} $element $element] +do_execsql_test vector-1-text-length-multiple-elements { + SELECT vector_extract(vector($input)); +} {{[1,1]}} + +set input [format {[1,%sx]} $element] +do_catchsql_test vector-1-text-length-invalid-second-element { + SELECT vector($input); +} [list 1 "vector: float string length exceeded 1024 characters: '$element'"] + proc error_messages {sql} { set ret "" set stmt [sqlite3_prepare db $sql -1 dummy] @@ -239,3 +289,5 @@ do_test vector-1-func-errors { {vector_distance: vectors must have the same type: 1 != 2} {vector_distance: l2 distance is not supported for float1bit vectors} }] + +finish_test diff --git a/libsql-sys/src/wal/mod.rs b/libsql-sys/src/wal/mod.rs index 71e0c21ee3..1a35d7cc73 100644 --- a/libsql-sys/src/wal/mod.rs +++ b/libsql-sys/src/wal/mod.rs @@ -131,7 +131,7 @@ impl PageHeaders { Self { inner } } - pub fn iter(&self) -> PageHdrIter { + pub fn iter(&self) -> PageHdrIter<'_> { // TODO: move LIBSQL_PAGE_SIZE PageHdrIter::new(self.as_ptr(), 4096) } diff --git a/libsql/src/hrana/cursor.rs b/libsql/src/hrana/cursor.rs index aa0b1191b1..4aa62c9832 100644 --- a/libsql/src/hrana/cursor.rs +++ b/libsql/src/hrana/cursor.rs @@ -147,7 +147,7 @@ where }) } - pub async fn next_step(&mut self) -> Result> { + pub async fn next_step(&mut self) -> Result> { CursorStep::new(self).await } diff --git a/libsql/src/hrana/hyper.rs b/libsql/src/hrana/hyper.rs index 300602c27e..3ea1ad631e 100644 --- a/libsql/src/hrana/hyper.rs +++ b/libsql/src/hrana/hyper.rs @@ -276,7 +276,7 @@ impl crate::statement::Stmt for crate::hrana::Statement { self.cols.len() } - fn columns(&self) -> Vec { + fn columns(&self) -> Vec> { //FIXME: there are several blockers here: // 1. We cannot know the column types before sending a query, so this method will never return results right // away. diff --git a/libsql/src/local/impls.rs b/libsql/src/local/impls.rs index b86405610e..5714d4761a 100644 --- a/libsql/src/local/impls.rs +++ b/libsql/src/local/impls.rs @@ -160,7 +160,7 @@ impl Stmt for LibsqlStmt { self.0.column_count() } - fn columns(&self) -> Vec { + fn columns(&self) -> Vec> { self.0.columns() } } diff --git a/libsql/src/local/statement.rs b/libsql/src/local/statement.rs index c31e751734..e8e0b415e1 100644 --- a/libsql/src/local/statement.rs +++ b/libsql/src/local/statement.rs @@ -339,7 +339,7 @@ impl Statement { /// If associated DB schema can be altered concurrently, you should make /// sure that current statement has already been stepped once before /// calling this method. - pub fn columns(&self) -> Vec { + pub fn columns(&self) -> Vec> { let n = self.column_count(); let mut cols = Vec::with_capacity(n); for i in 0..n { diff --git a/libsql/src/replication/client.rs b/libsql/src/replication/client.rs index 90a7ab4565..330306a526 100644 --- a/libsql/src/replication/client.rs +++ b/libsql/src/replication/client.rs @@ -140,6 +140,7 @@ impl GrpcChannel { .pool_idle_timeout(None) .pool_max_idle_per_host(3) .build(connector); + let client = DiagnosticResponseService::new(client); let client = GrpcWebClientService::new(client); let classifier = GrpcErrorsAsFailures::new().with_success(GrpcCode::FailedPrecondition); @@ -183,6 +184,76 @@ impl Service> for GrpcChannel { } } +/// Diagnostic middleware that intercepts raw HTTP responses before gRPC-web +/// framing. When the response status is not 200 or the content-type is not +/// grpc-web, it buffers and logs the response body so we can diagnose +/// "Invalid header bit N" errors from tonic-web. +#[derive(Clone)] +struct DiagnosticResponseService { + inner: S, +} + +impl DiagnosticResponseService { + fn new(inner: S) -> Self { + Self { inner } + } +} + +impl Service> for DiagnosticResponseService +where + S: Service, Response = http::Response, Error = hyper::Error> + + Clone + + Send + + 'static, + S::Future: Send + 'static, + ReqBody: Send + 'static, +{ + type Response = http::Response; + type Error = hyper::Error; + type Future = + Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: http::Request) -> Self::Future { + let uri = req.uri().clone(); + let fut = self.inner.call(req); + Box::pin(async move { + let resp = fut.await?; + let status = resp.status(); + let content_type = resp + .headers() + .get(http::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let is_grpc = content_type.contains("grpc"); + + if status != http::StatusCode::OK || !is_grpc { + // Buffer the body to log it, then re-create the response + let (parts, body) = resp.into_parts(); + let body_bytes = hyper::body::to_bytes(body).await.unwrap_or_default(); + let preview_len = std::cmp::min(body_bytes.len(), 1024); + let body_preview = String::from_utf8_lossy(&body_bytes[..preview_len]); + tracing::warn!( + status = %status, + uri = %uri, + content_type = %content_type, + body_len = body_bytes.len(), + body_preview = %body_preview, + "[libsql diagnostic] non-gRPC HTTP response — will cause 'Invalid header bit' error" + ); + Ok(http::Response::from_parts(parts, hyper::Body::from(body_bytes))) + } else { + tracing::trace!(status = %status, uri = %uri, "[libsql diagnostic] gRPC response OK"); + Ok(resp) + } + }) + } +} + #[derive(Clone)] /// Contains token and namespace headers to append to every request. pub struct GrpcInterceptor { diff --git a/libsql/src/replication/connection.rs b/libsql/src/replication/connection.rs index 418ae03465..e21d01511e 100644 --- a/libsql/src/replication/connection.rs +++ b/libsql/src/replication/connection.rs @@ -797,7 +797,7 @@ impl Stmt for RemoteStatement { } } - fn columns(&self) -> Vec { + fn columns(&self) -> Vec> { if let Some(stmt) = self.local_statement.as_ref() { return stmt.columns(); } diff --git a/libsql/src/statement.rs b/libsql/src/statement.rs index 861fdf8023..e5704f42fe 100644 --- a/libsql/src/statement.rs +++ b/libsql/src/statement.rs @@ -24,7 +24,7 @@ pub(crate) trait Stmt { fn column_count(&self) -> usize; - fn columns(&self) -> Vec; + fn columns(&self) -> Vec>; } /// A cached prepared statement. @@ -103,7 +103,7 @@ impl Statement { } /// Fetch the list of columns for the prepared statement. - pub fn columns(&self) -> Vec { + pub fn columns(&self) -> Vec> { self.inner.columns() } } diff --git a/libsql/src/sync/statement.rs b/libsql/src/sync/statement.rs index b3de1338ef..7b31577672 100644 --- a/libsql/src/sync/statement.rs +++ b/libsql/src/sync/statement.rs @@ -67,7 +67,7 @@ impl Stmt for SyncedStatement { self.inner.column_count() } - fn columns(&self) -> Vec { + fn columns(&self) -> Vec> { self.inner.columns() } } diff --git a/rust-toolchain.toml b/rust-toolchain.toml index c2324b9b48..2449619493 100644 --- a/rust-toolchain.toml +++ b/rust-toolchain.toml @@ -1,3 +1,3 @@ [toolchain] profile = "default" -channel = "1.85.0" +channel = "1.98.1" diff --git a/vendored/rusqlite/src/column.rs b/vendored/rusqlite/src/column.rs index 4413a62bcb..64c1cac9c0 100644 --- a/vendored/rusqlite/src/column.rs +++ b/vendored/rusqlite/src/column.rs @@ -136,7 +136,7 @@ impl Statement<'_> { /// calling this method. #[cfg(feature = "column_decltype")] #[cfg_attr(docsrs, doc(cfg(feature = "column_decltype")))] - pub fn columns(&self) -> Vec { + pub fn columns(&self) -> Vec> { let n = self.column_count(); let mut cols = Vec::with_capacity(n); for i in 0..n {