feat(metrics): record workflow and processing lifecycle counters (#44)

src/processing/handler.zig

@@ -541,6 +541,8 @@ pub const ProcessingHandler = struct {
         // replay path so a RUNNING job resumes after restart (FLO-104).
         self.startPipelines(owned_id, &def);
 
+        if (shard.metrics_registry) |m| m.processing.recordSubmitted();
+
         // Return the job ID
         shard.sendOkResponse(conn, req.header.request_id, job_id);
     }
@@ -859,6 +861,13 @@ pub const ProcessingHandler = struct {
         };
         const value = &[_]u8{@intFromEnum(status)};
         _ = persistence_mod.persistEntry(shard, entry_type, Flags.NONE, namespace, job_id, value) catch {};
+
+        // Both terminal transitions route through here.
+        if (shard.metrics_registry) |m| switch (status) {
+            .cancelled => m.processing.recordCancelled(),
+            .stopped => m.processing.recordCompleted(),
+            else => {},
+        };
     }
 
     /// Persist a processing_savepoint entry. Key = savepoint_id.

src/workflow/handler.zig

@@ -795,6 +795,8 @@ pub const WorkflowHandler = struct {
             .history = .empty,
         };
 
+        if (shard.metrics_registry) |m| m.workflow.recordStarted();
+
         // Add initial history event
         const evt_type = self.allocator.dupe(u8, "workflow_started") catch {
             self.freeRunRecord(&run);
@@ -907,6 +909,7 @@ pub const WorkflowHandler = struct {
 
         // Add history event
         self.addHistoryEvent(run, "signal_received", signal_type, now_ms);
+        if (shard.metrics_registry) |m| m.workflow.recordSignalDelivered();
 
         // If the run is waiting for this signal type, resume execution
         if (run.status == .waiting) {
@@ -1565,6 +1568,7 @@ pub const WorkflowHandler = struct {
                     }
 
                     self.addHistoryEvent(run, "step_completed", step_label, now_ms);
+                    if (shard.metrics_registry) |m| m.workflow.recordStepExecuted();
 
                     // Reset retry + poll counters on step transition
                     run.retry_count = 0;
@@ -1604,6 +1608,7 @@ pub const WorkflowHandler = struct {
                     if (signal_found) {
                         // Signal already received — follow "success" transition
                         self.addHistoryEvent(run, "step_completed", step_label, now_ms);
+                    if (shard.metrics_registry) |m| m.workflow.recordStepExecuted();
                         const transition = wait_step.getTransition(definition.StepOutcome.success) orelse {
                             self.completeRun(shard, run_ns_key, run, .failed, "no success transition for wait step", now_ms);
                             return;
@@ -2372,6 +2377,7 @@ pub const WorkflowHandler = struct {
         run.status = .running;
         self.addHistoryEvent(run, "child_completed", outcome, now_ms);
         self.addHistoryEvent(run, "step_completed", step_label, now_ms);
+                    if (shard.metrics_registry) |m| m.workflow.recordStepExecuted();
 
         // Clear pending child state.
         if (run.pending_child_run_id_owned) |a| self.allocator.free(a);
@@ -2444,6 +2450,7 @@ pub const WorkflowHandler = struct {
         run.status = .running;
         self.addHistoryEvent(run, "action_completed", outcome, now_ms);
         self.addHistoryEvent(run, "step_completed", step_label, now_ms);
+                    if (shard.metrics_registry) |m| m.workflow.recordStepExecuted();
 
         // Clear pending action state
         if (run.pending_action_run_id_owned) |a| self.allocator.free(a);
@@ -2663,6 +2670,16 @@ pub const WorkflowHandler = struct {
         run.status = status;
         run.completed_at_ms = now_ms;
 
+        // Every terminal transition passes through here, so this is the one
+        // place the outcome counters need to be recorded.
+        if (shard.metrics_registry) |m| switch (status) {
+            .completed => m.workflow.recordCompleted(),
+            .failed => m.workflow.recordFailed(),
+            .cancelled => m.workflow.recordCancelled(),
+            .timed_out => m.workflow.recordTimedOut(),
+            else => {},
+        };
+
         // Resolve explicit output mapping from definition (if declared)
         if (status == .completed) {
             self.resolveWorkflowOutput(run, run_ns_key);
@@ -4086,6 +4103,10 @@ fn createTestShard(actions: *ActionsHandler) !Shard {
     shard.actions_handler = actions;
     shard.peer_shards = null;
     shard.peer_inboxes = null;
+    // `Shard = undefined` means field defaults do not apply — anything the code
+    // under test reads must be assigned here or it holds garbage.
+    shard.metrics_registry = null;
+    shard.shard_metrics = null;
     const raft_node = try std.testing.allocator.create(RaftNode);
     raft_node.* = try RaftNode.init(std.testing.allocator, 1, 0, 4096, .{});
     try raft_node.bootstrap();

tests/e2e/metrics_values_test.zig

@@ -23,6 +23,17 @@ fn seriesValue(body: []const u8, name: []const u8, needle: []const u8) ?u64 {
     return null;
 }
 
+/// Value of an unlabeled `name <value>` series.
+fn scalarValue(body: []const u8, name: []const u8) ?u64 {
+    var lines = std.mem.splitScalar(u8, body, '\n');
+    while (lines.next()) |line| {
+        if (!std.mem.startsWith(u8, line, name)) continue;
+        if (line.len <= name.len or line[name.len] != ' ') continue;
+        return std.fmt.parseInt(u64, std.mem.trim(u8, line[name.len + 1 ..], " \r"), 10) catch continue;
+    }
+    return null;
+}
+
 test "e2e/metrics: stream append counters carry real values" {
     var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
         .server = .{ .metrics_enabled = true },
@@ -105,3 +116,63 @@ test "e2e/metrics: per-shard counters are attributed, not left at zero" {
     const cmds = seriesValue(resp.body, "flo_shard_commands_total", "shard") orelse 0;
     try testing.expect(cmds > 0);
 }
+
+test "e2e/metrics: workflow lifecycle counters carry real values" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "action", "register", "echo" });
+
+    const workflow_def =
+        \\kind: Workflow
+        \\name: metrics-wf
+        \\version: 1.0.0
+        \\start.run: @actions/echo
+        \\start.transition.success: flo.Completed
+        \\start.transition.failure: flo.Failed
+    ;
+    const path = try stdx.testing.writeDottedToTempYaml(testing.allocator, workflow_def, "metrics-wf.yaml");
+    defer stdx.testing.cleanupTempFile(testing.allocator, path);
+
+    try ctx.exec(&.{ "workflow", "create", "-f", path });
+    try ctx.exec(&.{ "workflow", "start", "metrics-wf", "{\"a\":1}" });
+    try ctx.exec(&.{ "workflow", "start", "metrics-wf", "{\"a\":2}" });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    // Two runs started — the family exported 0 regardless of traffic before,
+    // because the workflow handler never touched the registry.
+    try testing.expectEqual(@as(?u64, 2), scalarValue(resp.body, "flo_workflow_started_total"));
+}
+
+test "e2e/metrics: processing submitted counter carries a real value" {
+    var ctx = try stdx.testing.TestContext.initWithConfig(testing.allocator, .{
+        .server = .{ .metrics_enabled = true },
+    });
+    defer ctx.deinit();
+
+    try ctx.exec(&.{ "stream", "append", "proc_src", "seed" });
+
+    const job =
+        \\kind: Processing
+        \\name: metrics-proc
+        \\sources.[0].stream.name: proc_src
+        \\sinks.[0].stream.name: proc_dst
+    ;
+    const path = try stdx.testing.writeDottedToTempYaml(testing.allocator, job, "metrics-proc.yaml");
+    defer stdx.testing.cleanupTempFile(testing.allocator, path);
+
+    try ctx.exec(&.{ "processing", "submit", path });
+
+    var http = try ctx.createMetricsHttp();
+    defer http.deinit();
+    var resp = try http.get("/metrics");
+    defer resp.deinit();
+
+    try testing.expectEqual(@as(?u64, 1), scalarValue(resp.body, "flo_processing_jobs_submitted_total"));
+}