chore: sync release v1.75.3 to main branch (#6985)

# Description

Syncing patch release v1.75.3 to main branch

**↓↓ Please review and edit commit overrides before merging ↓↓**

BEGIN_COMMIT_OVERRIDE
fix: continuous backend config cache cause postgres disk bloat (#6983)
END_COMMIT_OVERRIDE

---------

Co-authored-by: Akash Chetty <achetty.iitr@gmail.com>
Co-authored-by: Akash <achetty@rudderstack.com>
Co-authored-by: rudderstack-github-actions[bot] <236995729+rudderstack-github-actions[bot]@users.noreply.github.com>
Co-authored-by: rudder-harness[bot] <278693763+rudder-harness[bot]@users.noreply.github.com>
Co-authored-by: Aris Tzoumas <atzoumas@rudderstack.com>

Author: 5 people

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [1.75.3](https://github.com/rudderlabs/rudder-server/compare/v1.75.2...v1.75.3) (2026-05-20)
+
+
+### Bug Fixes
+
+* continuous backend config cache cause postgres disk bloat ([#6983](https://github.com/rudderlabs/rudder-server/issues/6983)) ([398f902](https://github.com/rudderlabs/rudder-server/commit/398f9028eb48c1a59fd515a686b38b573f1da6e7))
+
 ## [1.75.2](https://github.com/rudderlabs/rudder-server/compare/v1.75.1...v1.75.2) (2026-05-19)
 
 

backend-config/backend_config_test.go

@@ -454,6 +454,7 @@ func TestWaitForConfig(t *testing.T) {
 func TestCache(t *testing.T) {
 	initBackendConfig()
 	t.Setenv("RSERVER_BACKEND_CONFIG_POLL_INTERVAL", "10ms")
+	t.Setenv("RSERVER_BACKEND_CONFIG_DB_CACHE_WRITE_DEBOUNCE", "10ms")
 	var calls int32
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		defer atomic.AddInt32(&calls, 1)

backend-config/internal/cache/cache.go

@@ -6,8 +6,14 @@ import (
 	"crypto/aes"
 	"crypto/cipher"
 	"crypto/rand"
+	"crypto/sha256"
 	"database/sql"
+	"encoding/hex"
 	"fmt"
+	"sync"
+	"time"
+
+	"github.com/samber/lo"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/jsonrs"
@@ -69,15 +75,37 @@ func Start(ctx context.Context, secret [32]byte, key string, channelProvider fun
 		return nil, err
 	}
 
+	// writeDebounce coalesces a burst of config updates into a single write
+	writeDebounce := config.GetDurationVar(10, time.Second, "BackendConfig.dbCacheWriteDebounce")
 	go func() {
-		// subscribe to config and write to db
-		for config := range channelProvider() {
-			// persist to database
-			err = dbStore.set(ctx, config.Data)
-			if err != nil {
+		var (
+			latestConfig any
+			gotConfig    bool
+			configMu     sync.Mutex
+		)
+		// persist writes the most recently received config to the database
+		persist := func() {
+			configMu.Lock()
+			data, ok := latestConfig, gotConfig
+			configMu.Unlock()
+			if !ok {
+				return
+			}
+			if err := dbStore.set(ctx, data); err != nil {
 				pkgLogger.Errorn("failed writing config to database", obskit.Error(err))
 			}
 		}
+		debouncedPersist, cancelDebounce := lo.NewDebounce(writeDebounce, persist)
+		// subscribe to config and debounce writes to db
+		for update := range channelProvider() {
+			configMu.Lock()
+			latestConfig, gotConfig = update.Data, true
+			configMu.Unlock()
+			debouncedPersist()
+		}
+		// channel closed: flush the latest config before shutting down
+		cancelDebounce()
+		persist()
 		dbStore.Close()
 	}()
 	return &dbStore, nil
@@ -89,6 +117,28 @@ func (db *cacheStore) set(ctx context.Context, config any) error {
 	if err != nil {
 		return fmt.Errorf("failed to marshal config: %w", err)
 	}
+	configSum := sha256.Sum256(configBytes)
+	configHash := hex.EncodeToString(configSum[:])
+
+	// Skip the write if the stored config is already identical. This avoids
+	// encrypting and transferring the (potentially large) config blob to the
+	// database when nothing has changed.
+	var storedHash sql.NullString
+	err = db.QueryRowContext(
+		ctx,
+		`SELECT config_hash FROM config_cache WHERE key = $1`,
+		db.key,
+	).Scan(&storedHash)
+	switch err {
+	case nil:
+		if storedHash.Valid && storedHash.String == configHash {
+			return nil
+		}
+	case sql.ErrNoRows:
+	default:
+		return err
+	}
+
 	// encrypt
 	encrypted, err := db.encryptAES(configBytes)
 	if err != nil {
@@ -97,13 +147,16 @@ func (db *cacheStore) set(ctx context.Context, config any) error {
 	// write to config table
 	_, err = db.ExecContext(
 		ctx,
-		`INSERT INTO config_cache (key, config) VALUES ($1, $2)
+		`INSERT INTO config_cache (key, config, config_hash) VALUES ($1, $2, $3)
 		on conflict (key)
 		do update set
 		config = $2,
-		updated_at = NOW()`,
+		config_hash = $3,
+		updated_at = NOW()
+		where config_cache.config_hash is distinct from excluded.config_hash`,
 		db.key,
 		encrypted,
+		configHash,
 	)
 	return err
 }

backend-config/internal/cache/cache_test.go

@@ -0,0 +1,76 @@
+package cache
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"testing"
+	"time"
+
+	"github.com/ory/dockertest/v3"
+	"github.com/stretchr/testify/require"
+
+	"github.com/rudderlabs/rudder-go-kit/jsonrs"
+	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource/postgres"
+)
+
+// TestCacheStoreSet exercises the upsert: it asserts that writing the
+// same config twice does not rewrite the row (updated_at stays put), while
+// writing a different config does.
+func TestCacheStoreSet(t *testing.T) {
+	pool, err := dockertest.NewPool("")
+	require.NoError(t, err)
+	resourcePostgres, err := postgres.Setup(pool, t)
+	require.NoError(t, err)
+
+	db := resourcePostgres.DB
+	require.NoError(t, migrate(db))
+
+	store := &cacheStore{
+		DB:     db,
+		secret: sha256.Sum256([]byte("secret")),
+		key:    "test-workspace",
+	}
+	ctx := context.Background()
+
+	// expectedHash mirrors the production hashing so the test verifies the
+	// stored value independently.
+	expectedHash := func(config any) string {
+		b, err := jsonrs.Marshal(config)
+		require.NoError(t, err)
+		sum := sha256.Sum256(b)
+		return hex.EncodeToString(sum[:])
+	}
+
+	readRow := func() (updatedAt time.Time, hash string, config []byte) {
+		t.Helper()
+		require.NoError(t, db.QueryRowContext(ctx,
+			`SELECT updated_at, config_hash, config FROM config_cache WHERE key = $1`,
+			store.key,
+		).Scan(&updatedAt, &hash, &config))
+		return updatedAt, hash, config
+	}
+
+	configA := map[string]any{"workspace": "A", "sources": []string{"s1", "s2"}}
+	configB := map[string]any{"workspace": "B", "sources": []string{"s3"}}
+
+	// first write inserts the row
+	require.NoError(t, store.set(ctx, configA))
+	firstUpdatedAt, firstHash, firstConfig := readRow()
+	require.Equal(t, expectedHash(configA), firstHash)
+
+	// writing an identical config must not rewrite the row
+	require.NoError(t, store.set(ctx, configA))
+	sameUpdatedAt, sameHash, sameConfig := readRow()
+	require.Equal(t, firstUpdatedAt, sameUpdatedAt, "updated_at must not change for identical config")
+	require.Equal(t, firstHash, sameHash)
+	require.Equal(t, firstConfig, sameConfig, "config blob must not be rewritten for identical config")
+
+	// writing a different config must rewrite the row and bump updated_at
+	time.Sleep(time.Millisecond) // ensure NOW() is observably later
+	require.NoError(t, store.set(ctx, configB))
+	changedUpdatedAt, changedHash, changedConfig := readRow()
+	require.True(t, changedUpdatedAt.After(firstUpdatedAt), "updated_at must advance when config changes")
+	require.Equal(t, expectedHash(configB), changedHash)
+	require.NotEqual(t, firstConfig, changedConfig, "config blob must be rewritten when config changes")
+}

sql/migrations/config_cache/000002_add_config_hash_and_autovacuum.up.sql

@@ -0,0 +1,7 @@
+ALTER TABLE config_cache ADD COLUMN IF NOT EXISTS config_hash text;
+
+ALTER TABLE config_cache
+SET (
+	toast.autovacuum_vacuum_scale_factor = 0.0,
+	toast.autovacuum_vacuum_threshold = 1
+);