From 9eac2e0612c269469cb70e3769f530bb8c5475b6 Mon Sep 17 00:00:00 2001
From: Alex Chihaia <chihaiaalex@gmail.com>
Date: Fri, 6 Nov 2026 09:50:28 +0200
Subject: [PATCH] Add source-timestamp propagation helpers for pipeline latency

---
 pom.xml                                       |  2 +-
 .../hsl/common/metrics/PipelineMetrics.java   | 54 +++++++++++++++++++
 2 files changed, 55 insertions(+), 1 deletion(-)

diff --git a/pom.xml b/pom.xml
index 153f992..3b0d53a 100644
--- a/pom.xml
+++ b/pom.xml
@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>fi.hsl</groupId>
     <artifactId>transitdata-common</artifactId>
-    <version>3.1.0</version>
+    <version>3.2.0</version>
     <packaging>jar</packaging>
     <name>Common utilities for Transitdata projects</name>
 
diff --git a/src/main/java/fi/hsl/common/metrics/PipelineMetrics.java b/src/main/java/fi/hsl/common/metrics/PipelineMetrics.java
index 989bace..8dbf2d5 100644
--- a/src/main/java/fi/hsl/common/metrics/PipelineMetrics.java
+++ b/src/main/java/fi/hsl/common/metrics/PipelineMetrics.java
@@ -1,6 +1,9 @@
 package fi.hsl.common.metrics;
 
+import fi.hsl.common.transitdata.TransitdataProperties;
 import io.prometheus.client.Histogram;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.TypedMessageBuilder;
 
 public final class PipelineMetrics {
 
@@ -25,4 +28,55 @@ public static void observeLatency(final String pipeline, final double seconds) {
     public static void observePublishDuration(final String pipeline, final double seconds) {
         PUBLISH_DURATION.labels(pipeline).observe(seconds);
     }
+
+    /**
+     * Resolve the pipeline-ingress timestamp (epoch ms) for a message being produced: forwarded from the incoming
+     * message's source timestamp when present, otherwise the current time. Pass null for a source component that has
+     * no incoming Pulsar message.
+     */
+    public static long resolveSourceTimestampMs(final Message<?> incoming) {
+        if (incoming != null) {
+            final String raw = incoming.getProperty(TransitdataProperties.KEY_SOURCE_MESSAGE_TIMESTAMP_MS);
+            if (raw != null && !raw.isEmpty()) {
+                try {
+                    return Long.parseLong(raw);
+                } catch (final NumberFormatException ignored) {
+                    // malformed value, fall through to current time
+                }
+            }
+        }
+        return System.currentTimeMillis();
+    }
+
+    /**
+     * Stamp the pipeline-ingress (source) timestamp on an outgoing message so it propagates through the pipeline. The
+     * first producer in a chain stamps the current time; downstream stages forward the value unchanged. Pass the
+     * incoming message to forward, or null at a source component.
+     */
+    public static void stampSourceTimestamp(final TypedMessageBuilder<?> out, final Message<?> incoming) {
+        out.property(TransitdataProperties.KEY_SOURCE_MESSAGE_TIMESTAMP_MS,
+                String.valueOf(resolveSourceTimestampMs(incoming)));
+    }
+
+    /**
+     * Observe pipeline latency as now minus the message's source (pipeline-ingress) timestamp. No observation is
+     * recorded when the message carries no valid source timestamp.
+     */
+    public static void observeLatencyFromSource(final String pipeline, final Message<?> message) {
+        if (message == null) {
+            return;
+        }
+        final String raw = message.getProperty(TransitdataProperties.KEY_SOURCE_MESSAGE_TIMESTAMP_MS);
+        if (raw == null || raw.isEmpty()) {
+            return;
+        }
+        try {
+            final long sourceTs = Long.parseLong(raw);
+            if (sourceTs > 0) {
+                observeLatency(pipeline, (System.currentTimeMillis() - sourceTs) / 1000.0);
+            }
+        } catch (final NumberFormatException ignored) {
+            // no observation for malformed timestamp
+        }
+    }
 }