fix: improve flow performance at high-throughput

backend/.sqlx/query-11ae5e1aade2bc9a78af2d956ece73766ae3330319619a68383b508a76f2d19c.json

@@ -42,6 +42,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json

@@ -67,6 +67,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-502781c4e2fc692db7a66b850450a1934b2403a5cff0421fe9609f9b99d8ee95.json

@@ -1,22 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM completed_job WHERE started_at + ((duration_ms/1000 + $1) || ' s')::interval <= now() RETURNING id",
-  "describe": {
-    "columns": [
-      {
-        "ordinal": 0,
-        "name": "id",
-        "type_info": "Uuid"
-      }
-    ],
-    "parameters": {
-      "Left": [
-        "Int8"
-      ]
-    },
-    "nullable": [
-      false
-    ]
-  },
-  "hash": "502781c4e2fc692db7a66b850450a1934b2403a5cff0421fe9609f9b99d8ee95"
-}

backend/.sqlx/query-52ad1773a9f5a1b6be1bb16af408f051f27d3f21b615c468b85aa75b64d943db.json

@@ -28,6 +28,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json

@@ -60,6 +60,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-6b9ff3fbca9e825c95d14705082a10de88172c0c748a45aba4d2d03c3b58f54d.json

@@ -46,6 +46,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json

@@ -40,6 +40,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-b6074eb9369de071f79165bf506fbc44f1688035e1928b4c3649d875187b1e1d.json

@@ -42,6 +42,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-bb0ee03198bcad69a1777447cf5c42fc22700e3cbf8cd44621a8380536d77552.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval  AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "id",
+        "type_info": "Uuid"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "bb0ee03198bcad69a1777447cf5c42fc22700e3cbf8cd44621a8380536d77552"
+}

backend/.sqlx/query-c7febee7146a28481ce36f3570bceb38751a6527827525a582d9f8804981f552.json

@@ -42,6 +42,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/.sqlx/query-d48070ef889723f408ed6c1a97b5ab9f0e98340cd33c6e29824f6030f0089071.json

@@ -37,6 +37,7 @@
                 "bash",
                 "postgresql",
                 "nativets",
+                "Nativets",
                 "bun",
                 "mysql",
                 "bigquery",

backend/src/monitor.rs

@@ -194,7 +194,7 @@ pub async fn delete_expired_items(db: &DB) -> () {
     let job_retention_secs = *JOB_RETENTION_SECS.read().await;
     if job_retention_secs > 0 {
         let deleted_jobs = sqlx::query_scalar!(
-                "DELETE FROM completed_job WHERE started_at + ((duration_ms/1000 + $1) || ' s')::interval <= now() RETURNING id",
+                "DELETE FROM completed_job WHERE created_at <= now() - ($1::bigint::text || ' s')::interval  AND started_at + ((duration_ms/1000 + $1::bigint) || ' s')::interval <= now() RETURNING id",
                 job_retention_secs
             )
             .fetch_all(db)

backend/windmill-queue/src/jobs.rs

@@ -415,7 +415,7 @@ pub async fn add_completed_job<
     tx.commit().await?;
 
     #[cfg(feature = "enterprise")]
-    if !is_flow && _duration > 1000 {
+    if *CLOUD_HOSTED && !is_flow && _duration > 1000 {
         let additional_usage = _duration / 1000;
         let w_id = &queued_job.workspace_id;
         let premium_workspace = *windmill_common::worker::CLOUD_HOSTED
@@ -1716,7 +1716,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
     priority_override: Option<i16>,
 ) -> Result<(Uuid, QueueTransaction<'c, R>), Error> {
     #[cfg(feature = "enterprise")]
-    {
+    if *CLOUD_HOSTED {
         let premium_workspace = *CLOUD_HOSTED
             && sqlx::query_scalar!("SELECT premium FROM workspace WHERE id = $1", workspace_id)
                 .fetch_one(_db)

benchmarks/pulumi/index.ts

@@ -273,7 +273,59 @@ db.address.apply((address) => {
         ],
         environment: [
           { name: "MODE", value: "worker" },
+          { name: "NUM_WORKERS", value: "1" },
+          // { name: "METRICS_ADDR", value: "true" },
+          { name: "RUST_LOG", value: "info" },
+          {
+            name: "DATABASE_URL",
+            value: `postgres://postgres:postgres@${address}/windmill?sslmode=disable`,
+          },
+        ],
+
+        logConfiguration: {
+          logDriver: "awslogs",
+          options: {
+            "awslogs-group": "windmill-worker",
+            "awslogs-region": "us-east-2",
+            "awslogs-create-group": "true",
+            "awslogs-stream-prefix": "windmill-worker",
+          },
+        },
+        dockerLabels: {
+          PROMETHEUS_EXPORTER_PORT: "8001",
+        },
+      },
+    ]),
+    volumes: [
+      {
+        name: "dependency_cache",
+        dockerVolumeConfiguration: {
+          scope: "shared",
+          autoprovision: true,
+        },
+      },
+    ],
+  });
+
+  const worker_td2 = new aws.ecs.TaskDefinition("worker-3", {
+    family: "windmill-worker-2",
+    containerDefinitions: JSON.stringify([
+      {
+        name: "windmill-worker",
+        image: "ghcr.io/windmill-labs/windmill-ee:main",
+        cpu: 1024,
+        memory: 1800,
+        essential: true,
+        mountPaths: [
+          {
+            containerPath: "/tmp/windmill/cache",
+            sourceVolume: "dependency_cache",
+          },
+        ],
+        environment: [
+          { name: "WORKER_GROUP", value: "dedicated" },
           { name: "NUM_WORKERS", value: "10" },
+          { name: "MODE", value: "worker" },
           // { name: "METRICS_ADDR", value: "true" },
           { name: "RUST_LOG", value: "info" },
           {
@@ -381,6 +433,19 @@ db.address.apply((address) => {
       },
     ],
   });
+
+  const service_worker2 = new aws.ecs.Service("service-worker-2", {
+    cluster: cluster.id,
+    taskDefinition: worker_td2.arn,
+    desiredCount: 5,
+    forceNewDeployment: true,
+    orderedPlacementStrategies: [
+      {
+        type: "binpack",
+        field: "cpu",
+      },
+    ],
+  });
 });
 
 module.exports = {