feat(gas): add automatic workflow pruning (#4169)

# Description

Please include a summary of the changes and the related issue. Please also include relevant motivation and context.

## Type of change

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes.

## Checklist:

- [ ] My code follows the style guidelines of this project
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my feature works
- [ ] New and existing unit tests pass locally with my changes

Author: MasterPtato

Cargo.lock

@@ -20,11 +20,11 @@ members = [
   "engine/packages/error-macros",
   "engine/packages/gasoline",
   "engine/packages/gasoline-macros",
+  "engine/packages/gasoline-runtime",
   "engine/packages/guard",
   "engine/packages/guard-core",
   "engine/packages/logs",
   "engine/packages/metrics",
-  "engine/packages/metrics-aggregator",
   "engine/packages/namespace",
   "engine/packages/pegboard",
   "engine/packages/pegboard-gateway",
@@ -395,6 +395,9 @@ members = [
     [workspace.dependencies.gasoline-macros]
     path = "engine/packages/gasoline-macros"
 
+    [workspace.dependencies.gasoline-runtime]
+    path = "engine/packages/gasoline-runtime"
+
     [workspace.dependencies.rivet-guard]
     path = "engine/packages/guard"
 
@@ -407,9 +410,6 @@ members = [
     [workspace.dependencies.rivet-metrics]
     path = "engine/packages/metrics"
 
-    [workspace.dependencies.rivet-metrics-aggregator]
-    path = "engine/packages/metrics-aggregator"
-
     [workspace.dependencies.namespace]
     path = "engine/packages/namespace"
 

Cargo.toml

@@ -0,0 +1,3 @@
+# Workflow Pruning
+
+TODO

docs/engine/GASOLINE/WORKFLOW_PRUNING.md

@@ -8,6 +8,7 @@ license.workspace = true
 [dependencies]
 epoxy.workspace = true
 gas.workspace = true
+gasoline-runtime.workspace = true
 namespace.workspace = true
 pegboard.workspace = true
 rivet-config.workspace = true
@@ -18,4 +19,4 @@ serde.workspace = true
 tokio-tungstenite.workspace = true
 tracing.workspace = true
 universaldb.workspace = true
-url.workspace = true
+url.workspace = true

engine/artifacts/config-schema.json

@@ -22,6 +22,8 @@ pub async fn start(config: rivet_config::Config, pools: rivet_pools::Pools) -> R
 		},
 		create_default_namespace(&ctx),
 		backfill::run(&ctx),
+		setup_pegboard_metrics_aggregator(&ctx),
+		setup_gas_pruner(&ctx),
 	)?;
 
 	Ok(())
@@ -103,3 +105,27 @@ async fn create_default_namespace(ctx: &StandaloneCtx) -> Result<()> {
 
 	Ok(())
 }
+
+async fn setup_pegboard_metrics_aggregator(ctx: &StandaloneCtx) -> Result<()> {
+	// Create metrics aggregator if does not exist
+	let workflow_id = ctx
+		.workflow(pegboard::workflows::metrics_aggregator::Input {})
+		.unique()
+		.dispatch()
+		.await?;
+	tracing::info!(%workflow_id, "created pegboard metrics aggregator");
+
+	Ok(())
+}
+
+async fn setup_gas_pruner(ctx: &StandaloneCtx) -> Result<()> {
+	// Create gas pruner if does not exist
+	let workflow_id = ctx
+		.workflow(gasoline_runtime::workflows::pruner::Input {})
+		.unique()
+		.dispatch()
+		.await?;
+	tracing::info!(%workflow_id, "created gasoline pruner");
+
+	Ok(())
+}

engine/packages/bootstrap/Cargo.toml

@@ -8,6 +8,8 @@ use serde::{Deserialize, Serialize};
 pub struct Runtime {
 	#[serde(default)]
 	pub worker: Worker,
+	#[serde(default)]
+	pub gasoline: Gasoline,
 	/// Time (in seconds) to allow for guard to wait for pending requests after receiving SIGTERM. Defaults
 	/// to 1 hour.
 	guard_shutdown_duration: Option<u32>,
@@ -71,3 +73,31 @@ impl Worker {
 		Duration::from_secs(self.shutdown_duration.unwrap_or(30) as u64)
 	}
 }
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct Gasoline {
+	/// Time (in seconds) after completion before considering a workflow eligible for pruning. Defaults to 7
+	/// days. Set to 0 to never prune workflow data.
+	prune_eligibility_duration: Option<u64>,
+	/// Time (in seconds) to periodically check for workflows to prune. Defaults to 12 hours.
+	prune_interval_duration: Option<u64>,
+}
+
+impl Gasoline {
+	pub fn prune_eligibility_duration(&self) -> Option<Duration> {
+		if let Some(prune_eligibility_duration) = self.prune_eligibility_duration {
+			if prune_eligibility_duration == 0 {
+				None
+			} else {
+				Some(Duration::from_secs(prune_eligibility_duration))
+			}
+		} else {
+			Some(Duration::from_secs(60 * 60 * 24 * 7))
+		}
+	}
+
+	pub fn prune_interval_duration(&self) -> Duration {
+		Duration::from_secs(self.prune_interval_duration.unwrap_or(60 * 60 * 12))
+	}
+}

engine/packages/bootstrap/src/lib.rs

@@ -32,7 +32,6 @@ rivet-cache.workspace = true
 rivet-config.workspace = true
 rivet-guard.workspace = true
 rivet-logs.workspace = true
-rivet-metrics-aggregator.workspace = true
 rivet-pools.workspace = true
 rivet-runtime.workspace = true
 rivet-service-manager.workspace = true

engine/packages/config/src/config/runtime.rs

@@ -21,12 +21,6 @@ pub fn config(_rivet_config: rivet_config::Config) -> Result<RunConfigData> {
 			|config, pools| Box::pin(rivet_workflow_worker::start(config, pools)),
 			true,
 		),
-		Service::new(
-			"metrics_aggregator",
-			ServiceKind::Standalone,
-			|config, pools| Box::pin(rivet_metrics_aggregator::start(config, pools)),
-			false,
-		),
 		Service::new(
 			"bootstrap",
 			ServiceKind::Oneshot,