Fix(datapipe): Do not set the last_run field as part of status update but

rather explicitly at start of analysis.

Author: kpom-specter

cmd/api/src/daemons/datapipe/pipeline.go

@@ -207,6 +207,11 @@ func (s *BHCEPipeline) Analyze(ctx context.Context) error {
 
 		defer measure.LogAndMeasure(slog.LevelInfo, "Graph Analysis")()
 
+		// Record the last time we started an analysis run
+		if err := s.db.SetLastAnalysisStartTime(ctx); err != nil {
+			return fmt.Errorf("update last analysis start time: %v", err)
+		}
+
 		if err := RunAnalysisOperations(ctx, s.db, s.graphdb, s.cfg); err != nil {
 			if errors.Is(err, ErrAnalysisFailed) {
 				s.jobService.FailAnalyzedIngestJobs()

cmd/api/src/database/datapipestatus.go

@@ -25,29 +25,26 @@ import (
 
 type DatapipeStatusData interface {
 	UpdateLastAnalysisCompleteTime(ctx context.Context) error
+	SetLastAnalysisStartTime(ctx context.Context) error
 	SetDatapipeStatus(ctx context.Context, status model.DatapipeStatus) error
 	GetDatapipeStatus(ctx context.Context) (model.DatapipeStatusWrapper, error)
 }
 
+// This should be called at the end of a successful analysis run (not always every analysis)
 func (s *BloodhoundDB) UpdateLastAnalysisCompleteTime(ctx context.Context) error {
 	now := time.Now().UTC()
 	return s.db.WithContext(ctx).Exec("UPDATE datapipe_status SET updated_at = ?, last_complete_analysis_at = ?", now, now).Error
 }
 
+// This should be called at the start of analysis processing (not every datapipe tick, but start of real work)
+func (s *BloodhoundDB) SetLastAnalysisStartTime(ctx context.Context) error {
+	now := time.Now().UTC()
+	return s.db.WithContext(ctx).Exec("UPDATE datapipe_status SET updated_at = ?, last_analysis_run_at = ?", now, now).Error
+}
+
 func (s *BloodhoundDB) SetDatapipeStatus(ctx context.Context, status model.DatapipeStatus) error {
 	now := time.Now().UTC()
-	// All queries will update the status and table update time
-	updateSql := "UPDATE datapipe_status SET status = ?, updated_at = ?"
-
-	if status == model.DatapipeStatusAnalyzing {
-		// Updates last run anytime we start analysis
-		updateSql += ", last_analysis_run_at = ?;"
-		return s.db.WithContext(ctx).Exec(updateSql, status, now, now).Error
-	} else {
-		// Otherwise, only update status and last update to the table
-		updateSql += ";"
-		return s.db.WithContext(ctx).Exec(updateSql, status, now).Error
-	}
+	return s.db.WithContext(ctx).Exec("UPDATE datapipe_status SET status = ?, updated_at = ?;", status, now).Error
 }
 
 func (s *BloodhoundDB) GetDatapipeStatus(ctx context.Context) (model.DatapipeStatusWrapper, error) {

cmd/api/src/database/datapipestatus_integration_test.go

@@ -42,8 +42,9 @@ func TestDatapipeStatus(t *testing.T) {
 	oldAnalysisTime := status.LastAnalysisRunAt
 	err = db.SetDatapipeStatus(testCtx, model.DatapipeStatusAnalyzing)
 	require.Nil(t, err)
-	status, err = db.GetDatapipeStatus(testCtx)
+	err = db.SetLastAnalysisStartTime(testCtx)
 	require.Nil(t, err)
+	status, err = db.GetDatapipeStatus(testCtx)
 	assert.Equal(t, model.DatapipeStatusAnalyzing, status.Status)
 	assert.True(t, oldAnalysisTime.Before(status.LastAnalysisRunAt))
 	assert.True(t, status.LastCompleteAnalysisAt.IsZero())