fix: reporting race condition between flusher and syncer (#6585)

# Description

## Problem

This PR fixes a critical race condition in the reporting system that
causes metrics data loss when transactions take longer than expected to
commit.

  ### The Race Condition

  The issue occurs due to timing between three operations:
1. **Report()** - Writing metrics to the database (with transaction
commit time)
2. **getReports()** - Reading and aggregating reports from a time bucket
  3. **DELETE** - Removing processed reports after aggregation

  ### Data Loss Scenarios

  #### Scenario 1: Data Deleted Before Commit Completes
If a `Report()` transaction (store stage) takes longer than the time
between `getReports()` + `mainLoop()` iteration but **shorter** than
reaching the DELETE operation:

  1. Transaction starts writing metrics for minute bucket `T`
2. `getReports()` reads bucket `T` (doesn't see uncommitted transaction)
  3. Transaction commits (data now in bucket `T`)
  4. DELETE removes all data from bucket `T`
5. **Result: Metrics are deleted without ever being aggregated →
Complete data loss**

  #### Scenario 2: Duplicate Send Causes Data Loss via Deduplication
If the store stage (commit) takes **longer** than even the DELETE
statement:

  1. Transaction starts writing metrics for minute bucket `T`
2. `getReports()` reads bucket `T` (doesn't see uncommitted transaction)
3. Aggregated data sent to reporting service (without slow transaction
data)
  4. DELETE removes data from bucket `T`
  5. Transaction finally commits (data appears in bucket `T`)
6. Next `getReports()` reads bucket `T` again (now includes slow
transaction)
  7. Data sent to reporting service again

However, the reporting service contract expects **minute level
aggregative data** and performs deduplication. When the same time bucket
is sent twice:
  - First send: Missing the slow transaction's data
  - Second send: Contains only the slow transaction's data
- **Result: Reporting service deduplicates and keeps only one, losing
the other → Partial data loss**

  ## Solution

This PR implements transaction tracking to ensure reports are only
flushed when safe:

1. **Track active transactions by time bucket** -
`activeTransactionsByReportedAt` map maintains a count of ongoing
transactions per minute bucket
2. **Block unsafe flushes** - `getReports()` checks if any active
transactions have `reportedAt < bucketEnd` before flushing
3. **Return specific error** - `errBlockedByActiveTransactions` signals
blocking without logging noise
4. **Lifecycle management** - Transaction counter incremented on start,
decremented on completion (via defer)
5. **New metric** -
`StatReportingMainLoopBlockedDueToActiveTransactionsCount` tracks
blocking frequency

  ### Key Implementation Details
  - Mutex-protected concurrent access to transaction tracking map
  - Defer ensures cleanup even on panics/early returns
- Non-intrusive error handling (doesn't spam logs for expected blocking)
  - Proper cleanup (deletes map entry when count reaches 0)


## Linear Ticket

[PIPE-2679](https://linear.app/rudderstack/issue/PIPE-2679/reporting-race-condition-between-flusher-and-syncer)
## Security

- [ ] The code changed/added as part of this pull request won't create
any security issues with how the software is being used.

Author: itsmihir

enterprise/reporting/reporting.go

@@ -31,15 +31,18 @@ import (
 
 const ReportsTable = "reports"
 
+var errBlockedByActiveTransactions = errors.New("blocked by active transactions")
+
 const (
-	StatReportingMainLoopTime              = "reporting_client_main_loop_time"
-	StatReportingGetReportsTime            = "reporting_client_get_reports_time"
-	StatReportingGetReportsCount           = "reporting_client_get_reports_count"
-	StatReportingGetAggregatedReportsTime  = "reporting_client_get_aggregated_reports_time"
-	StatReportingGetAggregatedReportsCount = "reporting_client_get_aggregated_reports_count"
-	StatReportingGetMinReportedAtQueryTime = "reporting_client_get_min_reported_at_query_time"
-	StatReportingGetReportsQueryTime       = "reporting_client_get_reports_query_time"
-	StatReportingVacuumDuration            = "reporting_vacuum_duration"
+	StatReportingMainLoopTime                                = "reporting_client_main_loop_time"
+	StatReportingGetReportsTime                              = "reporting_client_get_reports_time"
+	StatReportingGetReportsCount                             = "reporting_client_get_reports_count"
+	StatReportingGetAggregatedReportsTime                    = "reporting_client_get_aggregated_reports_time"
+	StatReportingGetAggregatedReportsCount                   = "reporting_client_get_aggregated_reports_count"
+	StatReportingGetMinReportedAtQueryTime                   = "reporting_client_get_min_reported_at_query_time"
+	StatReportingGetReportsQueryTime                         = "reporting_client_get_reports_query_time"
+	StatReportingVacuumDuration                              = "reporting_vacuum_duration"
+	StatReportingMainLoopBlockedDueToActiveTransactionsCount = "reporting_main_loop_blocked_due_to_active_transactions_count"
 )
 
 type DefaultReporter struct {
@@ -74,6 +77,9 @@ type DefaultReporter struct {
 	eventNamePrefixLength config.ValueLoader[int]
 	eventNameSuffixLength config.ValueLoader[int]
 	commonClient          *client.Client
+
+	activeTransactionsByReportedAtMutex sync.Mutex
+	activeTransactionsByReportedAt      map[int64]int64
 }
 
 func NewDefaultReporter(ctx context.Context, conf *config.Config, log logger.Logger, configSubscriber *configSubscriber, stats stats.Stats) *DefaultReporter {
@@ -134,6 +140,7 @@ func NewDefaultReporter(ctx context.Context, conf *config.Config, log logger.Log
 		eventNamePrefixLength:                eventNamePrefixLength,
 		eventNameSuffixLength:                eventNameSuffixLength,
 		commonClient:                         client.New(client.RouteMetrics, conf, log, stats),
+		activeTransactionsByReportedAt:       make(map[int64]int64),
 	}
 }
 
@@ -240,6 +247,20 @@ func (r *DefaultReporter) getReports(currentMs, aggregationIntervalMin int64, sy
 		return nil, 0, nil
 	}
 
+	safeToFlushCurrentBucket := true
+	r.activeTransactionsByReportedAtMutex.Lock()
+	for transactionReportedAt := range r.activeTransactionsByReportedAt {
+		if transactionReportedAt < bucketEnd {
+			safeToFlushCurrentBucket = false
+			break
+		}
+	}
+	r.activeTransactionsByReportedAtMutex.Unlock()
+
+	if !safeToFlushCurrentBucket {
+		return nil, 0, errBlockedByActiveTransactions
+	}
+
 	groupByColumns := "workspace_id, namespace, instance_id, source_definition_id, source_category, source_id, destination_definition_id, destination_id, source_task_run_id, source_job_id, source_job_run_id, transformation_id, transformation_version_id, tracking_plan_id, tracking_plan_version, in_pu, pu, status, terminal_state, initial_state, status_code, event_name, event_type, error_type"
 	sqlStatement = fmt.Sprintf(`
     SELECT
@@ -415,6 +436,7 @@ func (r *DefaultReporter) mainLoop(ctx context.Context, c types.SyncerConfig) {
 	getReportsCount := r.stats.NewTaggedStat(StatReportingGetReportsCount, stats.HistogramType, tags)
 	getAggregatedReportsTimer := r.stats.NewTaggedStat(StatReportingGetAggregatedReportsTime, stats.TimerType, tags)
 	getAggregatedReportsCount := r.stats.NewTaggedStat(StatReportingGetAggregatedReportsCount, stats.HistogramType, tags)
+	loopBlockedDueToActiveTransactionsCount := r.stats.NewTaggedStat(StatReportingMainLoopBlockedDueToActiveTransactionsCount, stats.CountType, stats.Tags{"clientName": c.Label})
 
 	r.getMinReportedAtQueryTime = r.stats.NewTaggedStat(StatReportingGetMinReportedAtQueryTime, stats.TimerType, tags)
 	r.getReportsQueryTime = r.stats.NewTaggedStat(StatReportingGetReportsQueryTime, stats.TimerType, tags)
@@ -452,7 +474,11 @@ func (r *DefaultReporter) mainLoop(ctx context.Context, c types.SyncerConfig) {
 			aggregationIntervalMin := int64(aggregationInterval.Load().Minutes())
 			reports, reportedAt, err := r.getReports(currentMin, aggregationIntervalMin, c.ConnInfo)
 			if err != nil {
-				r.log.Errorn("getting reports", obskit.Error(err))
+				if errors.Is(err, errBlockedByActiveTransactions) {
+					loopBlockedDueToActiveTransactionsCount.Increment()
+				} else {
+					r.log.Errorn("getting reports", obskit.Error(err))
+				}
 				select {
 				case <-ctx.Done():
 					r.log.Infon("stopping mainLoop for syncer",
@@ -633,7 +659,25 @@ func (r *DefaultReporter) Report(ctx context.Context, metrics []*types.PUReporte
 	}
 	defer func() { _ = stmt.Close() }()
 
+	r.activeTransactionsByReportedAtMutex.Lock()
 	reportedAt := time.Now().UTC().Unix() / 60
+	r.activeTransactionsByReportedAt[reportedAt]++
+	r.activeTransactionsByReportedAtMutex.Unlock()
+
+	// Cleanup when transaction completes (success or failure)
+	txn.AddFinallyListener(func() {
+		r.activeTransactionsByReportedAtMutex.Lock()
+		if r.activeTransactionsByReportedAt[reportedAt] == 1 {
+			delete(r.activeTransactionsByReportedAt, reportedAt)
+		} else if r.activeTransactionsByReportedAt[reportedAt] > 1 {
+			r.activeTransactionsByReportedAt[reportedAt]--
+		} else {
+			// This should never happen - indicates a bug in transaction tracking
+			r.log.Errorn("active transactions counter is invalid")
+		}
+		r.activeTransactionsByReportedAtMutex.Unlock()
+	})
+
 	for _, metric := range metrics {
 		workspaceID := r.configSubscriber.WorkspaceIDFromSource(metric.SourceID)
 		metric := *metric

utils/tx/tx.go

@@ -3,24 +3,76 @@ package tx
 import "database/sql"
 
 // Tx is a wrapper around sql.Tx that supports registering and executing
-// post-commit actions, a.k.a. success listeners.
+// post-commit actions (success listeners), post-rollback actions (failure listeners),
+// and post-completion actions (finally listeners that run regardless of outcome).
 type Tx struct {
 	*sql.Tx
 	successListeners []func()
+	failureListeners []func()
+	finallyListeners []func()
 }
 
 // AddSuccessListener registers a listener to be executed after the transaction has been committed successfully.
 func (tx *Tx) AddSuccessListener(listener func()) {
 	tx.successListeners = append(tx.successListeners, listener)
 }
 
-// Commit commits the transaction and executes all listeners.
+// AddFailureListener registers a listener to be executed after the transaction has been rolled back.
+func (tx *Tx) AddFailureListener(listener func()) {
+	tx.failureListeners = append(tx.failureListeners, listener)
+}
+
+// AddFinallyListener registers a listener to be executed after the transaction completes,
+// regardless of whether it was committed or rolled back. Finally listeners are executed
+// after success/failure listeners. Useful for cleanup that should always happen.
+func (tx *Tx) AddFinallyListener(listener func()) {
+	tx.finallyListeners = append(tx.finallyListeners, listener)
+}
+
+// Commit commits the transaction and executes all success listeners on success,
+// or failure listeners if the commit fails. Finally listeners are always executed.
+// Failure and finally listeners are cleared to prevent double-firing if Rollback is called afterward.
 func (tx *Tx) Commit() error {
 	err := tx.Tx.Commit()
+
 	if err == nil {
 		for _, successListener := range tx.successListeners {
 			successListener()
 		}
+	} else {
+		for _, failureListener := range tx.failureListeners {
+			failureListener()
+		}
+	}
+	for _, finallyListener := range tx.finallyListeners {
+		finallyListener()
 	}
+
+	// Clear failure and finally listeners to prevent double-firing if Rollback() is called after Commit()
+	// Success listeners don't need clearing as they only fire in Commit() on success
+	tx.failureListeners = nil
+	tx.finallyListeners = nil
+
+	return err
+}
+
+// Rollback rolls back the transaction and executes all failure listeners,
+// followed by finally listeners.
+// Failure and finally listeners are cleared to prevent double-firing if Commit is called afterward.
+func (tx *Tx) Rollback() error {
+	err := tx.Tx.Rollback()
+
+	for _, failureListener := range tx.failureListeners {
+		failureListener()
+	}
+	for _, finallyListener := range tx.finallyListeners {
+		finallyListener()
+	}
+
+	// Clear failure and finally listeners to prevent double-firing if Commit() is called after Rollback()
+	// Success listeners don't need clearing as they won't fire in Commit() after a Rollback()
+	tx.failureListeners = nil
+	tx.finallyListeners = nil
+
 	return err
 }