feat(workflows): defererred sqlite tasks

Author: MasterPtato

docker/dev-full/grafana/dashboards/chirp-workflow.json

@@ -1662,14 +1662,14 @@
 						"uid": "prometheus"
 					},
 					"editorMode": "code",
-					"expr": "sum(increase(rivet_chirp_workflow_branch_upsert_duration_bucket{workflow_name=~\"[[workflow_name]]\"} [$__rate_interval])) by (le)",
+					"expr": "sum(increase(rivet_chirp_workflow_loop_iteration_duration_bucket{workflow_name=~\"[[workflow_name]]\"} [$__rate_interval])) by (le)",
 					"format": "heatmap",
 					"legendFormat": "{{le}}",
 					"range": true,
 					"refId": "A"
 				}
 			],
-			"title": "Branch Upsert Duration",
+			"title": "Loop Iteration Duration",
 			"tooltip": {
 				"show": true,
 				"showHistogram": true
@@ -1779,13 +1779,13 @@
 						"uid": "prometheus"
 					},
 					"editorMode": "code",
-					"expr": "sum(irate(rivet_chirp_workflow_branch_upsert_duration_count{workflow_name=~\"[[workflow_name]]\"} [$__rate_interval])) by (workflow_name)",
+					"expr": "sum(irate(rivet_chirp_workflow_loop_iteration_duration_count{workflow_name=~\"[[workflow_name]]\"} [$__rate_interval])) by (workflow_name)",
 					"legendFormat": "{{workflow_name}}",
 					"range": true,
 					"refId": "A"
 				}
 			],
-			"title": "Branch Events Per Second",
+			"title": "Loop Events Per Second",
 			"type": "timeseries"
 		},
 		{

examples/system-test/yarn.lock

@@ -277,13 +277,6 @@ __metadata:
   languageName: node
   linkType: hard
 
-"@types/deno@npm:^2.2.0":
-  version: 2.2.0
-  resolution: "@types/deno@npm:2.2.0"
-  checksum: 10c0/cb45bbffe66a3008224a509c6bcb338921cc68b9045363f77ba5d84650d879b8fd4c810db24369a93fbce4a8e2855808bb141c0447feb47d911a7512ba374bde
-  languageName: node
-  linkType: hard
-
 "@types/node@npm:*, @types/node@npm:^22.13.9":
   version: 22.13.9
   resolution: "@types/node@npm:22.13.9"
@@ -1526,7 +1519,6 @@ __metadata:
     "@hono/node-ws": "npm:^1.1.0"
     "@rivet-gg/actor-core": "npm:^5.1.2"
     "@rivet-gg/api": "npm:^24.6.2"
-    "@types/deno": "npm:^2.2.0"
     "@types/node": "npm:^22.13.9"
     "@types/ws": "npm:^8.18.0"
     hono: "npm:^4.6.17"

packages/common/cache/build/src/driver.rs

@@ -370,14 +370,22 @@ impl moka::Expiry<String, ExpiringValue> for ValueExpiry {
 }
 
 /// In-memory cache driver implementation using the moka crate
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 pub struct InMemoryDriver {
 	service_name: String,
 	cache: Cache<String, ExpiringValue>,
 	/// In-memory rate limiting store - maps keys to hit counts with expiration
 	rate_limits: Cache<String, ExpiringValue>,
 }
 
+impl Debug for InMemoryDriver {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.debug_struct("InMemoryDriver")
+			.field("service_name", &self.service_name)
+			.finish()
+	}
+}
+
 impl InMemoryDriver {
 	pub fn new(service_name: String, max_capacity: u64) -> Self {
 		// Create a cache with ValueExpiry implementation for custom expiration times
@@ -391,7 +399,7 @@ impl InMemoryDriver {
 			.build();
 
 		Self {
-			service_name, // Kept for future expansion (e.g., namespacing keys by service)
+			service_name,
 			cache,
 			rate_limits,
 		}

packages/common/chirp-workflow/core/src/builder/workflow/message.rs

@@ -107,32 +107,24 @@ impl<'a, M: Message> MessageBuilder<'a, M> {
 			let tags = serde_json::Value::Object(self.tags);
 			let tags2 = tags.clone();
 
-			let (msg, write) = tokio::join!(
-				async {
-					self.ctx
-						.db()
-						.commit_workflow_message_send_event(
-							self.ctx.workflow_id(),
-							&location,
-							self.version,
-							&tags,
-							M::NAME,
-							&body_val,
-							self.ctx.loop_location(),
-						)
-						.await
-				},
-				async {
-					if self.wait {
-						self.ctx.msg_ctx().message_wait(tags2, self.body).await
-					} else {
-						self.ctx.msg_ctx().message(tags2, self.body).await
-					}
-				},
-			);
-
-			msg.map_err(GlobalError::raw)?;
-			write.map_err(GlobalError::raw)?;
+			self.ctx
+				.db()
+				.commit_workflow_message_send_event(
+					self.ctx.workflow_id(),
+					&location,
+					self.version,
+					&tags,
+					M::NAME,
+					&body_val,
+					self.ctx.loop_location(),
+				)
+				.await?;
+
+			if self.wait {
+				self.ctx.msg_ctx().message_wait(tags2, self.body).await?;
+			} else {
+				self.ctx.msg_ctx().message(tags2, self.body).await?;
+			}
 
 			let dt = start_instant.elapsed().as_secs_f64();
 			metrics::MESSAGE_SEND_DURATION

packages/common/chirp-workflow/core/src/ctx/workflow.rs

@@ -774,8 +774,6 @@ impl WorkflowCtx {
 					.map_err(WorkflowError::SerializeLoopOutput)
 					.map_err(GlobalError::raw)?;
 
-				let start_instant = Instant::now();
-
 				// Insert event before loop is run so the history is consistent
 				self.db
 					.upsert_workflow_loop_event(
@@ -790,11 +788,6 @@ impl WorkflowCtx {
 					)
 					.await?;
 
-				let dt = start_instant.elapsed().as_secs_f64();
-				metrics::BRANCH_UPSERT_DURATION
-					.with_label_values(&[&self.name])
-					.observe(dt);
-
 				(0, state, None)
 			};
 
@@ -875,7 +868,7 @@ impl WorkflowCtx {
 							.map_err(GlobalError::raw)?;
 
 						let dt = start_instant.elapsed().as_secs_f64();
-						metrics::BRANCH_UPSERT_DURATION
+						metrics::LOOP_ITERATION_DURATION
 							.with_label_values(&[&self.name])
 							.observe(dt - dt2);
 					}
@@ -910,7 +903,7 @@ impl WorkflowCtx {
 							.map_err(GlobalError::raw)?;
 
 						let dt = start_instant.elapsed().as_secs_f64();
-						metrics::BRANCH_UPSERT_DURATION
+						metrics::LOOP_ITERATION_DURATION
 							.with_label_values(&[&self.name])
 							.observe(dt - dt2);