fix(pb): fix metrics wf inconsistent calculation (#4103)

# 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

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

@@ -93,7 +93,7 @@ pub(crate) async fn pegboard_actor_metrics(ctx: &mut WorkflowCtx, input: &Input)
 
 			// Timeout was reached, record duration up till now
 			if sigs.is_empty() {
-				let now = util::timestamp::now();
+				let now = ctx.v(2).activity(GetTsInput {}).await?;
 				if let Some(last_recorded_awake_ts) = state.last_recorded_awake_ts {
 					new_awake_duration += now - last_recorded_awake_ts;
 				}
@@ -124,14 +124,14 @@ pub(crate) async fn pegboard_actor_metrics(ctx: &mut WorkflowCtx, input: &Input)
 }
 
 #[derive(Debug, Serialize, Deserialize, Hash)]
-pub struct InitStateInput {
+struct InitStateInput {
 	actor_id: Id,
 	namespace_id: Id,
 	name: String,
 }
 
 #[activity(InitState)]
-pub async fn init_state(ctx: &ActivityCtx, input: &InitStateInput) -> Result<bool> {
+async fn init_state(ctx: &ActivityCtx, input: &InitStateInput) -> Result<bool> {
 	let mut state = ctx.state::<Option<State>>()?;
 
 	*state = Some(State {
@@ -155,13 +155,21 @@ pub async fn init_state(ctx: &ActivityCtx, input: &InitStateInput) -> Result<boo
 }
 
 #[derive(Debug, Serialize, Deserialize, Hash)]
-pub struct RecordMetricsInput {
+struct GetTsInput {}
+
+#[activity(GetTs)]
+async fn get_ts(ctx: &ActivityCtx, input: &GetTsInput) -> Result<i64> {
+	Ok(util::timestamp::now())
+}
+
+#[derive(Debug, Serialize, Deserialize, Hash)]
+struct RecordMetricsInput {
 	/// Milliseconds.
 	awake_duration: i64,
 }
 
 #[activity(RecordMetrics)]
-pub async fn record_metrics(ctx: &ActivityCtx, input: &RecordMetricsInput) -> Result<()> {
+async fn record_metrics(ctx: &ActivityCtx, input: &RecordMetricsInput) -> Result<()> {
 	let state = ctx.state::<State>()?;
 
 	// Seconds (rounded up)
@@ -193,10 +201,10 @@ enum KvStorageQueryResult {
 }
 
 #[derive(Debug, Serialize, Deserialize, Hash)]
-pub struct RecordKvMetricsInput {}
+struct RecordKvMetricsInput {}
 
 #[activity(RecordKvMetrics)]
-pub async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> Result<()> {
+async fn record_kv_metrics(ctx: &ActivityCtx, input: &RecordKvMetricsInput) -> Result<()> {
 	let mut state = ctx.state::<State>()?;
 
 	let actor_id = state.actor_id;

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

@@ -291,6 +291,7 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> Result<()>
 			namespace_id: input.namespace_id,
 			name: input.name.clone(),
 		})
+		.tag("actor_id", input.actor_id)
 		.dispatch()
 		.await?;