fix: add size limit to indexer queries on jobs table to avoid oom (#6293)

* Update nix flake

* Update ee repo ref

* update ee-repo-ref

* Move collect stream with limits util to common

* add ee-repo-ref

* update ee-repo-ref

* update eereporef

backend/ee-repo-ref.txt

@@ -1 +1 @@
-7c9d175d7580530189bf4d41ac1edc438be0470a
+e871ff90d272ac171258fe0e1c23020242bba35f

backend/windmill-common/src/lib.rs

@@ -82,6 +82,7 @@ pub mod variables;
 pub mod worker;
 pub mod workspaces;
 pub mod triggers;
+pub mod stream;
 
 pub const DEFAULT_MAX_CONNECTIONS_SERVER: u32 = 50;
 pub const DEFAULT_MAX_CONNECTIONS_WORKER: u32 = 5;

backend/windmill-common/src/stream.rs

@@ -0,0 +1,42 @@
+use futures::{Stream, TryStreamExt};
+
+pub async fn collect_stream_with_limits<T, S>(
+    mut row_stream: S,
+    max_size_megabytes: usize,
+    estimate_row_size: fn(&T) -> anyhow::Result<usize>,
+    cutoff_time: chrono::TimeDelta,
+) -> Result<Vec<T>, anyhow::Error>
+where
+    S: Stream<Item = Result<T, sqlx::Error>> + Unpin,
+{
+    let started_time = chrono::Utc::now();
+    let mut results = Vec::new();
+    let mut total_size = 0;
+    let max_size_bytes = max_size_megabytes * 1024 * 1024;
+
+    while let Some(row) = row_stream.try_next().await? {
+        let row_size = estimate_row_size(&row)?;
+
+        results.push(row);
+        total_size += row_size;
+        if total_size + row_size > max_size_bytes {
+            tracing::info!(
+                "Stream was cutoff at {} elements because the collected size reached the treshold of {} MB",
+                results.len(),
+                max_size_megabytes
+            );
+            break;
+        }
+
+        if chrono::Utc::now() - started_time > cutoff_time {
+            tracing::info!(
+                "Stream was cutoff at {} elements because collecting it was taking longer than {}s",
+                results.len(),
+                cutoff_time.num_seconds()
+            );
+            break;
+        }
+    }
+
+    Ok(results)
+}