Merge branch 'main' into xrfxlp/device-count-labels

Author: XRFXLP

fault-remediation/pkg/reconciler/reconciler.go

@@ -1088,10 +1088,7 @@ func (r *FaultRemediationReconciler) parseHealthEvent(ctx context.Context, event
 		metrics.ProcessingErrors.WithLabelValues("extract_id_error", "unknown").Inc()
 		slog.ErrorContext(ctx, "Error extracting document ID", "error", err)
 
-		if markErr := watcherInstance.MarkProcessed(context.Background(), eventWithToken.ResumeToken); markErr != nil {
-			metrics.ProcessingErrors.WithLabelValues("mark_processed_error", "unknown").Inc()
-			slog.Error("Error updating resume token", "error", markErr)
-		}
+		_ = safeMarkProcessed(context.Background(), watcherInstance, eventWithToken.ResumeToken, "unknown")
 
 		return result, fmt.Errorf("error extracting document ID: %w", err)
 	}

fault-remediation/pkg/reconciler/reconciler_test.go

@@ -1534,3 +1534,20 @@ func TestDeletedNodeCancellationEventMarkedTerminal(t *testing.T) {
 	_, markProcessedCount, _, _ := mockWatcher.GetCallCounts()
 	assert.Equal(t, 1, markProcessedCount, "expected resume token to be marked processed")
 }
+
+func TestParseHealthEventShouldNotMarkColdStartEventProcessedIfDocumentIDExtractionFails(t *testing.T) {
+	ctx := context.Background()
+	nodeName := "cold-start-node"
+	event := testRawHealthEvent("cold-start-event", nodeName, protos.RecommendedAction_RESTART_BM)
+	delete(event, "_id")
+
+	eventWithToken := datastore.EventWithToken{Event: event}
+	mockWatcher := &MockChangeStreamWatcher{}
+	reconciler := &FaultRemediationReconciler{}
+
+	_, err := reconciler.parseHealthEvent(ctx, eventWithToken, mockWatcher)
+
+	assert.ErrorContains(t, err, "error extracting document ID")
+	_, markProcessedCount, _, _ := mockWatcher.GetCallCounts()
+	assert.Equal(t, 0, markProcessedCount, "expected cold-start event without resume token to not be marked processed")
+}