Consolidate Shopify patches onto v0.9.30-shopify-patches

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?;

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;
       }

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;
       }

libsql-server/src/query_analysis.rs

@@ -233,7 +233,7 @@ impl StmtKind {
     }
 }
 
-fn to_ascii_lower(s: &str) -> Cow<str> {
+fn to_ascii_lower(s: &str) -> Cow<'_, str> {
     if s.chars().all(|c| char::is_ascii_lowercase(&c)) {
         Cow::Borrowed(s)
     } else {

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;
       }

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

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)
     }

libsql/src/hrana/cursor.rs

@@ -147,7 +147,7 @@ where
         })
     }
 
-    pub async fn next_step(&mut self) -> Result<CursorStep<S>> {
+    pub async fn next_step(&mut self) -> Result<CursorStep<'_, S>> {
         CursorStep::new(self).await
     }
 

libsql/src/hrana/hyper.rs

@@ -276,7 +276,7 @@ impl crate::statement::Stmt for crate::hrana::Statement<HttpSender> {
         self.cols.len()
     }
 
-    fn columns(&self) -> Vec<crate::Column> {
+    fn columns(&self) -> Vec<crate::Column<'_>> {
         //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.

libsql/src/local/impls.rs

@@ -160,7 +160,7 @@ impl Stmt for LibsqlStmt {
         self.0.column_count()
     }
 
-    fn columns(&self) -> Vec<Column> {
+    fn columns(&self) -> Vec<Column<'_>> {
         self.0.columns()
     }
 }

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<Column> {
+    pub fn columns(&self) -> Vec<Column<'_>> {
         let n = self.column_count();
         let mut cols = Vec::with_capacity(n);
         for i in 0..n {

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<http::Request<BoxBody>> 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<S> {
+    inner: S,
+}
+
+impl<S> DiagnosticResponseService<S> {
+    fn new(inner: S) -> Self {
+        Self { inner }
+    }
+}
+
+impl<S, ReqBody> Service<http::Request<ReqBody>> for DiagnosticResponseService<S>
+where
+    S: Service<http::Request<ReqBody>, Response = http::Response<hyper::Body>, Error = hyper::Error>
+        + Clone
+        + Send
+        + 'static,
+    S::Future: Send + 'static,
+    ReqBody: Send + 'static,
+{
+    type Response = http::Response<hyper::Body>;
+    type Error = hyper::Error;
+    type Future =
+        Pin<Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>> + Send>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: http::Request<ReqBody>) -> 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("<none>")
+                .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 {

libsql/src/replication/connection.rs

@@ -797,7 +797,7 @@ impl Stmt for RemoteStatement {
         }
     }
 
-    fn columns(&self) -> Vec<Column> {
+    fn columns(&self) -> Vec<Column<'_>> {
         if let Some(stmt) = self.local_statement.as_ref() {
             return stmt.columns();
         }

libsql/src/statement.rs

@@ -24,7 +24,7 @@ pub(crate) trait Stmt {
 
     fn column_count(&self) -> usize;
 
-    fn columns(&self) -> Vec<Column>;
+    fn columns(&self) -> Vec<Column<'_>>;
 }
 
 /// A cached prepared statement.
@@ -103,7 +103,7 @@ impl Statement {
     }
 
     /// Fetch the list of columns for the prepared statement.
-    pub fn columns(&self) -> Vec<Column> {
+    pub fn columns(&self) -> Vec<Column<'_>> {
         self.inner.columns()
     }
 }

libsql/src/sync/statement.rs

@@ -67,7 +67,7 @@ impl Stmt for SyncedStatement {
         self.inner.column_count()
     }
 
-    fn columns(&self) -> Vec<Column> {
+    fn columns(&self) -> Vec<Column<'_>> {
         self.inner.columns()
     }
 }

rust-toolchain.toml

@@ -1,3 +1,3 @@
 [toolchain]
 profile = "default"
-channel = "1.85.0"
+channel = "1.98.1"

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<Column> {
+    pub fn columns(&self) -> Vec<Column<'_>> {
         let n = self.column_count();
         let mut cols = Vec::with_capacity(n);
         for i in 0..n {