Merge pull request #7160 from StarryKira/codex/log-retention-upstream

feat: 支持运维与请求日志的滚动保留设置

backend/internal/repository/dashboard_aggregation_group_usage_test.go

@@ -249,6 +249,18 @@ func TestDashboardAggregationRepositoryCleanupUsageLogsPartitionedSortsAndInvali
 			WillReturnResult(sqlmock.NewResult(0, 0))
 		mock.ExpectCommit()
 	}
+	// The boundary partition is pruned by exact timestamp. tableoid is required
+	// because ctid alone can also identify a retained row in another partition.
+	mock.ExpectBegin()
+	mock.ExpectQuery(`SELECT id FROM usage_group_rollup_state.*FOR UPDATE`).
+		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
+	mock.ExpectQuery(`(?s)SELECT tableoid, ctid.*WHERE created_at < \$1.*WHERE \(tableoid, ctid\) IN.*RETURNING created_at`).
+		WithArgs(cutoff, usageLogsCleanupBatchSize).
+		WillReturnRows(sqlmock.NewRows([]string{"created_at"}).AddRow(cutoff.Add(-time.Hour)))
+	mock.ExpectExec(`UPDATE usage_group_rollup_state`).
+		WithArgs(cutoff.Add(-time.Hour), "Asia/Shanghai").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectCommit()
 	mock.ExpectBegin()
 	mock.ExpectQuery(`SELECT closed_before::text, retained_from.*FOR UPDATE`).
 		WillReturnRows(sqlmock.NewRows([]string{"closed_before", "retained_from", "timezone_name"}).
@@ -271,7 +283,7 @@ func TestDashboardAggregationRepositoryCleanupUsageLogsNonPartitionedFailureRoll
 	mock.ExpectBegin()
 	mock.ExpectQuery(`SELECT id FROM usage_group_rollup_state.*FOR UPDATE`).
 		WillReturnRows(sqlmock.NewRows([]string{"id"}).AddRow(1))
-	mock.ExpectQuery(`(?s)SELECT ctid.*ORDER BY created_at ASC, id ASC.*DELETE FROM usage_logs.*RETURNING created_at`).
+	mock.ExpectQuery(`(?s)SELECT tableoid, ctid.*ORDER BY created_at ASC, id ASC.*DELETE FROM usage_logs.*RETURNING created_at`).
 		WithArgs(cutoff, usageLogsCleanupBatchSize).
 		WillReturnRows(sqlmock.NewRows([]string{"created_at"}).AddRow(deletedAt))
 	mock.ExpectExec(`UPDATE usage_group_rollup_state`).

backend/internal/repository/dashboard_aggregation_repo.go

@@ -239,7 +239,10 @@ func (r *dashboardAggregationRepository) CleanupUsageLogs(ctx context.Context, c
 		if err := r.dropUsageLogsPartitions(ctx, cutoff); err != nil {
 			return err
 		}
-	} else if err := r.cleanupUsageLogsBatches(ctx, cutoff); err != nil {
+	}
+	// Remove the expired part of the boundary partition as well, so a retention
+	// window measured in days does not silently round up to a whole month.
+	if err := r.cleanupUsageLogsBatches(ctx, cutoff); err != nil {
 		return err
 	}
 	return r.SyncGroupUsageRollups(ctx, service.GroupUsageTodayStart(r.now()))
@@ -261,14 +264,14 @@ func (r *dashboardAggregationRepository) cleanupUsageLogsBatches(ctx context.Con
 
 		res, err := r.sql.ExecContext(ctx, `
 			WITH victims AS (
-				SELECT ctid
+				SELECT tableoid, ctid
 				FROM usage_logs
 				WHERE created_at < $1
 				ORDER BY created_at ASC, id ASC
 				LIMIT $2
 			)
 			DELETE FROM usage_logs
-			WHERE ctid IN (SELECT ctid FROM victims)
+			WHERE (tableoid, ctid) IN (SELECT tableoid, ctid FROM victims)
 		`, cutoff.UTC(), usageLogsCleanupBatchSize)
 		if err != nil {
 			return err
@@ -298,14 +301,14 @@ func cleanupUsageLogsBatchWithRollupInvalidation(ctx context.Context, db *sql.DB
 	}
 	rows, err := tx.QueryContext(ctx, `
 		WITH victims AS (
-			SELECT ctid
+			SELECT tableoid, ctid
 			FROM usage_logs
 			WHERE created_at < $1
 			ORDER BY created_at ASC, id ASC
 			LIMIT $2
 		)
 		DELETE FROM usage_logs
-		WHERE ctid IN (SELECT ctid FROM victims)
+		WHERE (tableoid, ctid) IN (SELECT tableoid, ctid FROM victims)
 		RETURNING created_at
 	`, cutoff.UTC(), usageLogsCleanupBatchSize)
 	if err != nil {

backend/internal/repository/request_log_retention_integration_test.go

@@ -0,0 +1,46 @@
+//go:build integration
+
+package repository
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRequestLogRetention_PartitionBoundaryKeepsRecentRows(t *testing.T) {
+	ctx := context.Background()
+	tx := testTx(t)
+	// Temporary tables shadow the real schema and disappear on rollback.
+	_, err := tx.ExecContext(ctx, `
+		CREATE TEMP TABLE usage_logs (id bigint, created_at timestamptz) PARTITION BY RANGE (created_at);
+		CREATE TEMP TABLE retention_july PARTITION OF usage_logs FOR VALUES FROM ('2026-07-01') TO ('2026-08-01');
+		CREATE TEMP TABLE retention_august PARTITION OF usage_logs FOR VALUES FROM ('2026-08-01') TO ('2026-09-01');
+		INSERT INTO usage_logs VALUES
+			(1, '2026-07-17 23:59:59+00'),
+			(2, '2026-07-18 00:00:00+00'),
+			(3, '2026-08-01 00:00:00+00');
+	`)
+	require.NoError(t, err)
+	var collidingCTIDs bool
+	require.NoError(t, tx.QueryRowContext(ctx, `SELECT a.ctid = b.ctid FROM usage_logs a, usage_logs b WHERE a.id = 1 AND b.id = 3`).Scan(&collidingCTIDs))
+	require.True(t, collidingCTIDs, "fixture must exercise identical row locations in different partitions")
+	repo := newDashboardAggregationRepositoryWithSQL(tx)
+	cutoff := time.Date(2026, 7, 18, 0, 0, 0, 0, time.UTC)
+	for range 2 {
+		require.NoError(t, repo.cleanupUsageLogsBatches(ctx, cutoff))
+		rows, err := tx.QueryContext(ctx, `SELECT id FROM usage_logs ORDER BY id`)
+		require.NoError(t, err)
+		var ids []int64
+		for rows.Next() {
+			var id int64
+			require.NoError(t, rows.Scan(&id))
+			ids = append(ids, id)
+		}
+		require.NoError(t, rows.Err())
+		require.NoError(t, rows.Close())
+		require.Equal(t, []int64{2, 3}, ids)
+	}
+}