fix: conditions reset honors the latest timestamp of all unacked msgs (…

…#1523) Signed-off-by: Derek Wang <[email protected]>
argoproj · Jan 13, 2022 · fb5c257 · fb5c257
1 parent 4153c38
commit fb5c257
Showing 1 changed file with 3 additions and 1 deletion.
diff --git a/eventbus/driver/nats.go b/eventbus/driver/nats.go
@@ -294,7 +294,9 @@ func (n *natsStreaming) processEventSourceMsg(m *stan.Msg, msgHolder *eventSourc
 	// New message, set and check
 	msgHolder.msgs[depName] = &eventSourceMessage{seq: m.Sequence, timestamp: m.Timestamp, event: event, lastDeliveredTime: now}
 	msgHolder.parameters[depName] = true
-	msgHolder.latestGoodMsgTimestamp = m.Timestamp
+	if msgHolder.latestGoodMsgTimestamp < m.Timestamp {
+		msgHolder.latestGoodMsgTimestamp = m.Timestamp
+	}
 
 	// Check if there's any stale message being held.
 	// Stale message could be message age has been longer than NATS streaming max message age,