[slopfix] fix(pegboard): decrement kv_storage_used metric on actor delete

Author: MasterPtato

engine/packages/pegboard/src/workflows/actor/metrics.rs

@@ -120,6 +120,10 @@ pub(crate) async fn pegboard_actor_metrics(ctx: &mut WorkflowCtx, input: &Input)
 	})
 	.await?;
 
+	// The actor's kv is deleted on destroy, so remove its contribution from the running kv
+	// storage total.
+	ctx.v(2).activity(RemoveKvMetricsInput {}).await?;
+
 	Ok(())
 }
 
@@ -323,6 +327,37 @@ async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> R
 	Ok(())
 }
 
+#[derive(Debug, Serialize, Deserialize, Hash)]
+struct RemoveKvMetricsInput {}
+
+#[activity(RemoveKvMetrics)]
+async fn remove_kv_metrics(ctx: &ActivityCtx, input: &RemoveKvMetricsInput) -> Result<()> {
+	let mut state = ctx.state::<State>()?;
+
+	let namespace_id = state.namespace_id;
+	let name = &state.name;
+	let last_kv_storage_size = state.last_kv_storage_size;
+
+	if last_kv_storage_size != 0 {
+		ctx.udb()?
+			.txn("pegboard_actor_remove_kv_metrics", |tx| async move {
+				namespace::keys::metric::inc(
+					&tx.with_subspace(namespace::keys::subspace()),
+					namespace_id,
+					namespace::keys::metric::Metric::KvStorageUsed(name.to_string()),
+					-last_kv_storage_size,
+				);
+
+				Ok(())
+			})
+			.await?;
+	}
+
+	state.last_kv_storage_size = 0;
+
+	Ok(())
+}
+
 #[signal("pegboard_actor_metrics_pause")]
 pub struct Pause {
 	pub ts: i64,

engine/packages/pegboard/src/workflows/actor2/metrics.rs

@@ -119,6 +119,10 @@ pub(crate) async fn pegboard_actor2_metrics(ctx: &mut WorkflowCtx, input: &Input
 	})
 	.await?;
 
+	// The actor's kv is deleted on destroy, so remove its contribution from the running kv
+	// storage total.
+	ctx.v(2).activity(RemoveKvMetricsInput {}).await?;
+
 	Ok(())
 }
 
@@ -169,6 +173,8 @@ async fn record_metrics(ctx: &ActivityCtx, input: &RecordMetricsInput) -> Result
 	let name = &state.name;
 	ctx.udb()?
 		.txn("pegboard_actor2_record_metrics", |tx| async move {
+			tx.priority(Priority::Low)?;
+
 			namespace::keys::metric::inc(
 				&tx.with_subspace(namespace::keys::subspace()),
 				namespace_id,
@@ -212,6 +218,8 @@ async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> R
 			.txn("pegboard_actor2_kv_size_scan", |tx| {
 				let last_key = &last_key;
 				async move {
+					tx.priority(Priority::Low)?;
+
 					let start = Instant::now();
 					let tx = tx.with_subspace(keys::subspace());
 
@@ -296,6 +304,8 @@ async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> R
 
 	ctx.udb()?
 		.txn("pegboard_actor2_record_kv_metrics", |tx| async move {
+			tx.priority(Priority::Low)?;
+
 			namespace::keys::metric::inc(
 				&tx.with_subspace(namespace::keys::subspace()),
 				namespace_id,
@@ -312,6 +322,39 @@ async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> R
 	Ok(())
 }
 
+#[derive(Debug, Serialize, Deserialize, Hash)]
+struct RemoveKvMetricsInput {}
+
+#[activity(RemoveKvMetrics)]
+async fn remove_kv_metrics(ctx: &ActivityCtx, input: &RemoveKvMetricsInput) -> Result<()> {
+	let mut state = ctx.state::<State>()?;
+
+	let namespace_id = state.namespace_id;
+	let name = &state.name;
+	let last_kv_storage_size = state.last_kv_storage_size;
+
+	if last_kv_storage_size != 0 {
+		ctx.udb()?
+			.txn("pegboard_actor2_remove_kv_metrics", |tx| async move {
+				tx.priority(Priority::Low)?;
+
+				namespace::keys::metric::inc(
+					&tx.with_subspace(namespace::keys::subspace()),
+					namespace_id,
+					namespace::keys::metric::Metric::KvStorageUsed(name.to_string()),
+					-last_kv_storage_size,
+				);
+
+				Ok(())
+			})
+			.await?;
+	}
+
+	state.last_kv_storage_size = 0;
+
+	Ok(())
+}
+
 #[signal("pegboard_actor_metrics_pause")]
 pub struct Pause {
 	pub ts: i64,