fix: gate activation MAR to allowed warehouse sources

enterprise/activationrecords/noop.go

@@ -17,7 +17,7 @@ func NewNoopActivationRecordsReporter() *NoopActivationRecordsReporter {
 	return &NoopActivationRecordsReporter{}
 }
 
-func (n *NoopActivationRecordsReporter) GenerateReportsFromJobs([]*jobsdb.JobT, map[string]string) []*ActivationRecord {
+func (n *NoopActivationRecordsReporter) GenerateReportsFromJobs([]*jobsdb.JobT, map[string]SourceMetadata) []*ActivationRecord {
 	return nil
 }
 

enterprise/activationrecords/records_reporter.go

@@ -34,12 +34,29 @@ const (
 	activationRecordsTable = "activation_records_reports"
 
 	// retlSourceCategory is the SourceDefinition.Category of reverse-ETL ("warehouse
-	// actions") sources. MAR meters activation records from these sources only; the
-	// classification is resolved from the source's config category (via the source_id ->
-	// category map passed to GenerateReportsFromJobs), not from the job's source_category param.
+	// actions") sources. MAR meters activation records from these sources only when
+	// the source definition name is also allow-listed; classification is resolved
+	// from backend-config source metadata, not from the job's source_category param.
 	retlSourceCategory = "warehouse"
 )
 
+var defaultAllowedSourceDefinitionNames = []string{
+	"postgres",
+	"redshift",
+	"snowflake",
+	"bigquery",
+	"mysql",
+	"databricks",
+	"trino",
+}
+
+// SourceMetadata is the source-definition metadata needed to classify whether a
+// job belongs to a real reverse-ETL warehouse source for MAR metering.
+type SourceMetadata struct {
+	Category string
+	Name     string
+}
+
 // recordKey is the aggregation grain for activation records: one HLL sketch per
 // (workspace, source, destination). origin is intentionally NOT part of the key —
 // it is constant per source, so it is carried on the report as a plain column
@@ -67,18 +84,19 @@ type ActivationRecord struct {
 
 // ActivationRecordsReporter is the interface to report monthly active records (MAR).
 type ActivationRecordsReporter interface {
-	GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceCategoriesBySourceID map[string]string) []*ActivationRecord
+	GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceMetadataBySourceID map[string]SourceMetadata) []*ActivationRecord
 	ReportActivationRecords(ctx context.Context, reports []*ActivationRecord, tx *txn.Tx) error
 	MigrateDatabase(dbConn string, conf *config.Config) error
 }
 
 // UniqueActivationRecordsReporter implements ActivationRecordsReporter using HLL sketches.
 type UniqueActivationRecordsReporter struct {
-	log         logger.Logger
-	hllSettings *hll.Settings
-	instanceID  string
-	now         func() time.Time
-	stats       stats.Stats
+	log                          logger.Logger
+	hllSettings                  *hll.Settings
+	instanceID                   string
+	now                          func() time.Time
+	stats                        stats.Stats
+	allowedSourceDefinitionNames config.ValueLoader[[]string]
 }
 
 // NewUniqueActivationRecordsReporter constructs a UniqueActivationRecordsReporter.
@@ -100,6 +118,10 @@ func NewUniqueActivationRecordsReporter(log logger.Logger, conf *config.Config,
 		hllSettings: hllSettings,
 		instanceID:  conf.GetStringVar("1", "INSTANCE_ID"),
 		stats:       stats,
+		allowedSourceDefinitionNames: conf.GetReloadableStringSliceVar(
+			defaultAllowedSourceDefinitionNames,
+			"ActivationRecords.allowedSourceDefinitionNames",
+		),
 		now: func() time.Time {
 			return timeutil.Now()
 		},
@@ -131,33 +153,34 @@ func (u *UniqueActivationRecordsReporter) MigrateDatabase(dbConn string, conf *c
 
 // GenerateReportsFromJobs aggregates activation records from a batch of jobs.
 // It is FAIL-CLOSED: jobs missing fingerprint or origin are skipped (counted via stats).
-// sourceCategoriesBySourceID maps a source_id to its SourceDefinition.Category (from the
-// backend config); reverse-ETL classification is done via this map rather than the job's
-// source_category param, which is not populated on every ingestion path.
-func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceCategoriesBySourceID map[string]string) []*ActivationRecord {
+// sourceMetadataBySourceID maps a source_id to source-definition metadata (from
+// backend config); reverse-ETL classification is done via this map rather than
+// the job's source_category param, which is not populated on every ingestion path.
+func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb.JobT, sourceMetadataBySourceID map[string]SourceMetadata) []*ActivationRecord {
 	if len(jobs) == 0 {
 		return nil
 	}
 
 	accumulators := make(map[recordKey]*recordAccumulator)
+	allowedSourceDefinitionNames := normalizedSourceDefinitionNames(u.allowedSourceDefinitionNames.Load())
 
 	for _, job := range jobs {
-		if job.WorkspaceId == "" {
-			u.log.Warnn("workspace_id not found in job", logger.NewIntField("jobId", job.JobID))
-			u.recordSkip("missing_workspace")
-			continue
-		}
-
 		sourceID := jsonparser.GetStringOrEmpty(job.Parameters, "source_id")
 
-		// MAR meters reverse-ETL (warehouse) sources only. Classify by the source's
-		// SourceDefinition.Category from the backend config (looked up by source_id), NOT
+		// MAR meters real reverse-ETL SQL warehouse sources only. Classify by
+		// SourceDefinition metadata from backend config (looked up by source_id), NOT
 		// the job's source_category param: the internal-batch ingestion path leaves that
-		// param empty, so trusting it would under-count. This also prevents a client from
-		// being metered by stamping context.activation.fingerprint on a non-rETL source.
-		// A missing or unknown source_id resolves to "" here and is skipped. Non-rETL is
-		// the expected majority of traffic, so skip it silently — no per-job skip stat.
-		if !strings.EqualFold(sourceCategoriesBySourceID[sourceID], retlSourceCategory) {
+		// param empty, so trusting it would under-count. Requiring both category=warehouse
+		// and an allow-listed source-definition name avoids metering cloud-storage sources
+		// such as S3 that also use category=warehouse. Missing, unknown, non-warehouse, and
+		// warehouse-but-not-allow-listed sources are skipped silently — no per-job skip stat.
+		if !u.isAllowedWarehouseSource(sourceMetadataBySourceID[sourceID], allowedSourceDefinitionNames) {
+			continue
+		}
+
+		if job.WorkspaceId == "" {
+			u.log.Warnn("workspace_id not found in job", logger.NewIntField("jobId", job.JobID))
+			u.recordSkip("missing_workspace")
 			continue
 		}
 
@@ -246,6 +269,29 @@ func (u *UniqueActivationRecordsReporter) GenerateReportsFromJobs(jobs []*jobsdb
 	return reports
 }
 
+func (u *UniqueActivationRecordsReporter) isAllowedWarehouseSource(source SourceMetadata, allowedSourceDefinitionNames map[string]struct{}) bool {
+	if !strings.EqualFold(source.Category, retlSourceCategory) {
+		return false
+	}
+	_, ok := allowedSourceDefinitionNames[normalizeSourceDefinitionName(source.Name)]
+	return ok
+}
+
+func normalizedSourceDefinitionNames(names []string) map[string]struct{} {
+	allowed := make(map[string]struct{}, len(names))
+	for _, name := range names {
+		normalized := normalizeSourceDefinitionName(name)
+		if normalized != "" {
+			allowed[normalized] = struct{}{}
+		}
+	}
+	return allowed
+}
+
+func normalizeSourceDefinitionName(name string) string {
+	return strings.ToLower(strings.TrimSpace(name))
+}
+
 // ReportActivationRecords writes activation records to the database via a COPY statement.
 func (u *UniqueActivationRecordsReporter) ReportActivationRecords(ctx context.Context, reports []*ActivationRecord, tx *txn.Tx) error {
 	if len(reports) == 0 {

enterprise/activationrecords/records_reporter_test.go

@@ -11,14 +11,15 @@ import (
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
+	"github.com/rudderlabs/rudder-go-kit/stats/memstats"
 
 	"github.com/rudderlabs/rudder-server/jobsdb"
 )
 
 func TestUniqueActivationRecordsReporter(t *testing.T) {
 	// prepareJob builds a job with the standard activation payload shape. The Parameters
 	// intentionally omit source_category: on the internal-batch ingestion path it is empty,
-	// and the reporter classifies reverse-ETL sources from the config map (categoriesByID
+	// and the reporter classifies reverse-ETL sources from the config map (metadataByID
 	// below), not the param.
 	prepareJob := func(sourceID, destinationID, fingerprint, origin, workspaceID string) *jobsdb.JobT {
 		return &jobsdb.JobT{
@@ -43,13 +44,14 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 		}
 	}
 
-	// categoriesByID maps source_id -> SourceDefinition.Category, mirroring the map the
-	// processor passes from the backend config. "src1" is a reverse-ETL (warehouse)
-	// source; "src-eventstream" is not. Unknown source_ids resolve to "" and are treated
-	// as non-rETL.
-	categoriesByID := map[string]string{
-		"src1":            "warehouse",
-		"src-eventstream": "eventStream",
+	// metadataByID maps source_id -> SourceDefinition metadata, mirroring the map the
+	// processor passes from the backend config. "src1" is a reverse-ETL SQL warehouse
+	// source; "src-eventstream" is not. Unknown source_ids resolve to zero-value metadata
+	// and are treated as non-rETL.
+	metadataByID := map[string]SourceMetadata{
+		"src1":            {Category: "warehouse", Name: "Snowflake"},
+		"src-eventstream": {Category: "eventStream", Name: "javascript"},
+		"src-s3":          {Category: "warehouse", Name: "s3"},
 	}
 
 	t.Run("constructor validates HLL settings", func(t *testing.T) {
@@ -92,7 +94,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 				},
 			},
 			{
-				name: "missing fingerprint - skip",
+				name: "allow-listed warehouse missing fingerprint - skip",
 				jobs: []*jobsdb.JobT{
 					prepareJob("src1", "dst1", "", "org1", "ws1"),
 				},
@@ -137,6 +139,15 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 					require.Empty(t, reports)
 				},
 			},
+			{
+				name: "warehouse source not in allow-list - skip silently even with fingerprint",
+				jobs: []*jobsdb.JobT{
+					prepareJob("src-s3", "dst1", "fp-cloud-storage", "org1", "ws1"),
+				},
+				verify: func(t *testing.T, reports []*ActivationRecord) {
+					require.Empty(t, reports)
+				},
+			},
 			{
 				name: "config category is authoritative over a spoofed source_category param - skip",
 				jobs: []*jobsdb.JobT{
@@ -204,12 +215,64 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 
 		for _, tc := range testCases {
 			t.Run(tc.name, func(t *testing.T) {
-				reports := reporter.GenerateReportsFromJobs(tc.jobs, categoriesByID)
+				reports := reporter.GenerateReportsFromJobs(tc.jobs, metadataByID)
 				tc.verify(t, reports)
 			})
 		}
 	})
 
+	t.Run("GenerateReportsFromJobs_SourceDefinitionGateStats", func(t *testing.T) {
+		t.Run("warehouse source not in allow-list skips silently without missing fingerprint stat", func(t *testing.T) {
+			statsStore, err := memstats.New()
+			require.NoError(t, err)
+			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore)
+			require.NoError(t, err)
+
+			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{
+				prepareJob("src-s3", "dst1", "", "org1", "ws1"),
+			}, metadataByID)
+
+			require.Empty(t, reports)
+			require.Empty(t, statsStore.GetByName("activation_records_skipped"))
+		})
+
+		t.Run("allow-listed warehouse missing fingerprint increments missing_fingerprint", func(t *testing.T) {
+			statsStore, err := memstats.New()
+			require.NoError(t, err)
+			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.New(), statsStore)
+			require.NoError(t, err)
+
+			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{
+				prepareJob("src1", "dst1", "", "org1", "ws1"),
+			}, metadataByID)
+
+			require.Empty(t, reports)
+			metric := statsStore.Get("activation_records_skipped", stats.Tags{"reason": "missing_fingerprint"})
+			require.NotNil(t, metric)
+			require.Equal(t, float64(1), metric.LastValue())
+		})
+
+		t.Run("allow-list config changes are observed without reconstructing reporter", func(t *testing.T) {
+			conf := config.New()
+			conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"snowflake"})
+			reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, conf, stats.NOP)
+			require.NoError(t, err)
+
+			customMetadataByID := map[string]SourceMetadata{
+				"src-custom": {Category: "warehouse", Name: "custom_sql"},
+			}
+			job := prepareJob("src-custom", "dst1", "fp1", "org1", "ws1")
+
+			require.Empty(t, reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID))
+
+			conf.Set("ActivationRecords.allowedSourceDefinitionNames", []string{"custom_sql"})
+
+			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, customMetadataByID)
+			require.Len(t, reports, 1)
+			require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality())
+		})
+	})
+
 	t.Run("GenerateReportsFromJobs_MultiEventBatch", func(t *testing.T) {
 		reporter, err := NewUniqueActivationRecordsReporter(logger.NOP, config.Default, stats.NOP)
 		require.NoError(t, err)
@@ -231,7 +294,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 
 		t.Run("two distinct fingerprints in same batch => cardinality 2", func(t *testing.T) {
 			job := prepareTwoEventJob("src1", "dst1", "fp-1", "fp-2", "org1", "ws1")
-			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID)
+			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
 			require.Len(t, reports, 1)
 			require.Equal(t, "org1", reports[0].Origin)
 			require.NotNil(t, reports[0].FingerprintHll)
@@ -248,7 +311,7 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 				CustomVal:    "GW",
 				WorkspaceId:  "ws1",
 			}
-			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID)
+			reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
 			require.Len(t, reports, 1)
 			require.Equal(t, uint64(1), reports[0].FingerprintHll.Cardinality())
 		})
@@ -271,11 +334,11 @@ func TestUniqueActivationRecordsReporter(t *testing.T) {
 
 		job := prepareJob("src1", "dst1", "fp1", "org1", "ws1")
 
-		first := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID)
+		first := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
 		require.Len(t, first, 1)
 		require.Equal(t, uint64(1), first[0].FingerprintHll.Cardinality())
 
-		second := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, categoriesByID)
+		second := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, metadataByID)
 		require.Len(t, second, 1)
 		require.Equal(t, uint64(1), second[0].FingerprintHll.Cardinality())
 

enterprise/activationrecords/wire_compat_test.go

@@ -92,7 +92,9 @@ func TestWireCompat(t *testing.T) {
 		}),
 	}
 
-	reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, map[string]string{"src-1": "warehouse"})
+	reports := reporter.GenerateReportsFromJobs([]*jobsdb.JobT{job}, map[string]SourceMetadata{
+		"src-1": {Category: "warehouse", Name: "snowflake"},
+	})
 	require.Len(t, reports, 1)
 
 	// The reporter must resolve the full grain from the gateway's actual on-the-wire

integration_test/activationrecordsreporting/activation_records_reporting_test.go

@@ -315,8 +315,9 @@ func setup(t testing.TB) testConfig {
 	// destination into source.Destinations so the gateway's authDestIDForSource can
 	// resolve the X-Rudder-Destination-Id header into the job's destination_id
 	// parameter (which MAR metering requires). The source carries the "warehouse"
-	// category (reverse-ETL): MAR meters warehouse sources only, so the gateway stamps
-	// source_category=warehouse into the job params for these records to be metered.
+	// category and an allow-listed source definition name (reverse-ETL): MAR meters
+	// only real warehouse sources, so the gateway stamps source_category=warehouse
+	// into the job params and processor source metadata keeps these records in scope.
 	// ConfigBuilder.WithConnection additionally registers the top-level connection the
 	// router requires before it will deliver warehouse-source jobs (router/worker.go).
 	bcServer := backendconfigtest.NewBuilder().
@@ -327,6 +328,7 @@ func setup(t testing.TB) testConfig {
 					WithWorkspaceID(workspaceID).
 					WithID(sourceID).
 					WithSourceCategory("warehouse").
+					WithSourceType("snowflake").
 					WithConnection(
 						backendconfigtest.NewDestinationBuilder("WEBHOOK").
 							WithID(destinationID).