From f7bf057888b53adb22bf9eba272444ebcab886ff Mon Sep 17 00:00:00 2001
From: wendrul <53628737+wendrul@users.noreply.github.com>
Date: Wed, 30 Jul 2025 03:27:59 +0900
Subject: [PATCH] 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               |  2 +-
 backend/windmill-common/src/lib.rs    |  1 +
 backend/windmill-common/src/stream.rs | 42 +++++++++++++++++++++++++++
 3 files changed, 44 insertions(+), 1 deletion(-)
 create mode 100644 backend/windmill-common/src/stream.rs

diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt
index cd2c81c45a..bf94637847 100644
--- a/backend/ee-repo-ref.txt
+++ b/backend/ee-repo-ref.txt
@@ -1 +1 @@
-7c9d175d7580530189bf4d41ac1edc438be0470a
\ No newline at end of file
+e871ff90d272ac171258fe0e1c23020242bba35f
diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs
index c639d33b0b..299e99ebd4 100644
--- a/backend/windmill-common/src/lib.rs
+++ b/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;
diff --git a/backend/windmill-common/src/stream.rs b/backend/windmill-common/src/stream.rs
new file mode 100644
index 0000000000..e07e0a91f0
--- /dev/null
+++ b/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)
+}