nits: add abel suggestion

Author: dieriba

backend/Cargo.lock

@@ -28,7 +28,7 @@ zip = ["dep:async_zip"]
 oauth2 = ["dep:async-oauth2"]
 http_trigger = ["dep:matchit"]
 static_frontend = ["dep:rust-embed"]
-database = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:memchr", "dep:thiserror", "dep:rust_decimal"]
+database = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal"]
 
 [dependencies]
 windmill-queue.workspace = true
@@ -115,6 +115,5 @@ ulid.workspace = true
 rust-postgres = { workspace = true, optional = true }
 pg_escape = { workspace = true, optional = true }
 byteorder = { workspace = true, optional = true }
-memchr = { workspace = true, optional = true }
 thiserror = { workspace = true, optional = true }
 rust_decimal = { workspace = true, optional = true }

backend/windmill-api/Cargo.toml

@@ -3,13 +3,12 @@
 use core::str;
 use std::{
     cmp,
-    io::{self, Read},
+    io::{self, Cursor, Read},
     str::Utf8Error,
 };
 
 use byteorder::{BigEndian, ReadBytesExt};
 use bytes::Bytes;
-use memchr::memchr;
 use rust_postgres::types::{Oid, Type};
 use thiserror::Error;
 
@@ -20,13 +19,12 @@ const X_LOG_DATA_BYTE: u8 = b'w';
 /**
 * This implementation is inspired by Postgres replication functionality
 * from https://github.com/supabase/pg_replicate
-* 
-* Original implementation: 
+*
+* Original implementation:
 * - https://github.com/supabase/pg_replicate/blob/main/pg_replicate/src/conversions/cdc_event.rs
-* 
+*
 */
 
-
 #[derive(Debug)]
 pub struct PrimaryKeepAliveBody {
     pub wal_end: u64,
@@ -255,7 +253,7 @@ impl Buffer {
     }
 
     fn read_cstr(&mut self) -> Result<String, ConversionError> {
-        match memchr(0, self.slice()) {
+        match self.slice().iter().position(|&x| x == 0) {
             Some(pos) => {
                 let start = self.idx;
                 let end = start + pos;

backend/windmill-api/src/postgres_triggers/replication_message.rs

@@ -1,6 +1,7 @@
 use std::{collections::HashMap, pin::Pin};
 
 use crate::{
+    db::DB,
     postgres_triggers::{
         get_database_resource,
         relation::RelationConverter,
@@ -10,7 +11,6 @@ use crate::{
         },
         run_job,
     },
-    db::DB,
     users::fetch_api_authed,
 };
 use bytes::{BufMut, Bytes, BytesMut};
@@ -141,8 +141,8 @@ async fn update_ping(
     db: &DB,
     postgres_trigger: &DatabaseTrigger,
     error: Option<&str>,
-) -> Option<()> {
-    match sqlx::query_scalar!(
+) -> Result<bool, sqlx::Error> {
+    let updated = sqlx::query_scalar!(
         r#"
         UPDATE 
             postgres_trigger
@@ -162,8 +162,9 @@ async fn update_ping(
         *INSTANCE_NAME
     )
     .fetch_optional(db)
-    .await
-    {
+    .await;
+
+    match updated {
         Ok(updated) => {
             if updated.flatten().is_none() {
                 // allow faster restart of database trigger
@@ -187,27 +188,32 @@ async fn update_ping(
                     "Database {} changed, disabled, or deleted, stopping...",
                     postgres_trigger.path
                 );
-                return None;
+                return Ok(false);
             }
         }
-        Err(err) => {
-            tracing::warn!(
-                "Error updating ping of database {}: {:?}",
-                postgres_trigger.path,
-                err
-            );
-            return None;
-        }
+        Err(err) => return Err(err),
     };
 
-    Some(())
+    Ok(true)
 }
 
 async fn loop_ping(db: &DB, postgres_trigger: &DatabaseTrigger, error: Option<&str>) {
     loop {
-        if update_ping(db, postgres_trigger, error).await.is_none() {
-            return;
+        let result = update_ping(db, postgres_trigger, error).await;
+
+        match result {
+            Ok(false) => return,
+            Err(err) => {
+                tracing::warn!(
+                    "Error updating ping of database {}: {:?}",
+                    postgres_trigger.path,
+                    err
+                );
+                return;
+            }
+            _ => {}
         }
+
         tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
     }
 }
@@ -271,7 +277,7 @@ async fn listen_to_transactions(
 
                     if let Err(err) = &message {
                         tracing::debug!("{}", err.to_string());
-                        update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
+                        let _ = update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
                         return;
                     }
 
@@ -281,7 +287,7 @@ async fn listen_to_transactions(
                         Ok(logical_message) => logical_message,
                         Err(err) => {
                             tracing::debug!("{}", err.to_string());
-                            update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
+                            let _ = update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
                             return;
                         }
                     };
@@ -297,7 +303,7 @@ async fn listen_to_transactions(
                             let logical_replication_message = match x_log_data.parse(&logicail_replication_settings) {
                                 Ok(logical_replication_message) => logical_replication_message,
                                 Err(err) => {
-                                    update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
+                                    let _ = update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
                                     return;
                                 }
                             };
@@ -325,7 +331,7 @@ async fn listen_to_transactions(
                                 let relation = match relations.get_relation(o_id) {
                                     Ok(relation) => relation,
                                     Err(err) => {
-                                        update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
+                                        let _ = update_ping(&db, postgres_trigger, Some(&err.to_string())).await;
                                         return;
                                     }
                                 };
@@ -386,7 +392,7 @@ async fn try_to_listen_to_database_transactions(
                 tokio::spawn(async move {
                     let result = listen_to_transactions(&pg_trigger, db.clone(), killpill_rx).await;
                     if let Err(e) = result {
-                        update_ping(&db, &pg_trigger, Some(e.to_string().as_str())).await;
+                        let _ = update_ping(&db, &pg_trigger, Some(e.to_string().as_str())).await;
                     };
                 });
             } else {

backend/windmill-api/src/postgres_triggers/trigger.rs

@@ -691,29 +691,27 @@ pub fn decrypt(mc: &MagicCrypt256, value: String) -> Result<String> {
     })
 }
 
-struct Variable {
-    value: String,
-    is_secret: bool,
-}
-
 pub async fn get_variable_or_self(path: String, db: &DB, w_id: &str) -> Result<String> {
     if !path.starts_with("$var:") {
         return Ok(path);
     }
     let path = path.strip_prefix("$var:").unwrap().to_string();
-    let mut variable = sqlx::query_as!(
-        Variable,
-        "SELECT value, is_secret
-        FROM variable
-        WHERE path = $1 AND workspace_id = $2",
+
+    let record = sqlx::query!(
+        "SELECT value, is_secret 
+         FROM variable 
+         WHERE path = $1 AND workspace_id = $2",
         &path,
         &w_id
     )
     .fetch_one(db)
     .await?;
-    if variable.is_secret {
+
+    let mut value = record.value;
+    if record.is_secret {
         let mc = build_crypt(db, w_id).await?;
-        variable.value = decrypt(&mc, variable.value)?;
+        value = decrypt(&mc, value)?;
     }
-    Ok(variable.value)
+
+    Ok(value)
 }