Function TestLiveStoreConsumeDropsOldRecords

modules/livestore/live_store_test.go:461–512

(t *testing.T)

}

func TestLiveStoreConsumeDropsOldRecords(t *testing.T) {
	// default live store uses the default complete block timeout
	ls, _ := defaultLiveStore(t, t.TempDir())

	// Reset metrics
	metricRecordsProcessed.Reset()
	metricRecordsDropped.Reset()

	now := time.Now()
	older := now.Add(-1 * (defaultCompleteBlockTimeout + time.Second))
	newer := now.Add(-1 * (defaultCompleteBlockTimeout - time.Second))

	// Create test records - some old, some new
	records := []*kgo.Record{
		{
			Key:       []byte("tenant1"),
			Timestamp: older, // Too old (older than CompleteBlockTimeout)
			Value:     createValidPushRequest(t),
		},
		{
			Key:       []byte("tenant1"),
			Timestamp: newer, // Valid (newer than CompleteBlockTimeout)
			Value:     createValidPushRequest(t),
		},
		{
			Key:       []byte("tenant2"),
			Timestamp: older, // Too old
			Value:     createValidPushRequest(t),
		},
		{
			Key:       []byte("tenant2"),
			Timestamp: newer, // Valid
			Value:     createValidPushRequest(t),
		},
	}

	// Call consume
	_, err := ls.consume(context.Background(), createRecordIter(records), now)
	require.NoError(t, err)

	// Verify metrics
	// Should have processed 2 valid records (1 per tenant)
	require.Equal(t, float64(1), test.MustGetCounterValue(metricRecordsProcessed.WithLabelValues("tenant1")))
	require.Equal(t, float64(1), test.MustGetCounterValue(metricRecordsProcessed.WithLabelValues("tenant2")))

	// Should have dropped 2 old records (1 per tenant)
	require.Equal(t, float64(1), test.MustGetCounterValue(metricRecordsDropped.WithLabelValues("tenant1", "too_old")))
	require.Equal(t, float64(1), test.MustGetCounterValue(metricRecordsDropped.WithLabelValues("tenant2", "too_old")))

	err = services.StopAndAwaitTerminated(t.Context(), ls)
	require.NoError(t, err)
}

// TestLiveStoreConsumeTracksPartitionLag verifies that partition lag is tracked
// for all records regardless of outcome: too old (dropped), decode failure, or