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

# Description

The backend config cache persists the workspace config to postgres on
every config update. On every update the row was rewritten regardless of
whether the config value had actually changed or not. In postgres an
`UPDATE` always writes a new tuple version — even when the new values
equal the old ones — leaving the previous one dead; and since the large
`config` value lives in TOAST, each rewrite churns the TOAST table too.

This becomes pathological when:

- **There are a lot of continuous configuration changes**: every change
is another full-row rewrite.
- **Multiple pods try to cache the same config**: each pod independently
writes the same config to the same row, so a single logical change
becomes one write per pod, most of them re-writing a value identical to
what another pod just stored.

Dead tuples accumulated faster than autovacuum reclaimed them, steadily
bloating postgres disk usage.

### Changes

- **Conditional upsert via content hash** — a new `config_hash` column
stores a SHA-256 hex digest of the marshalled config. The upsert now
carries `WHERE config_cache.config_hash IS DISTINCT FROM
excluded.config_hash`, so when the stored config is already identical
the row is not rewritten and `updated_at` is not bumped. This
de-duplicates both no-op updates and redundant cross-pod writes.
- **Skip redundant work** — `set` does a cheap `SELECT config_hash`
first and returns early on a match, so a pod that finds the config
already cached skips encrypting and transferring the large config blob
entirely.
- **Debounce writes** — a burst of config updates is coalesced into a
single write per pod, controlled by `BackendConfig.dbCacheWriteDebounce`
(default `10s`).
- **Aggressive TOAST autovacuum** — the migration tunes `config_cache`
so dead tuples from any remaining rewrites are reclaimed promptly
(`toast.autovacuum_vacuum_scale_factor = 0`,
`toast.autovacuum_vacuum_threshold = 1`).

## Linear Ticket

resolves PIPE-3018

## Security

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

Co-authored-by: Aris Tzoumas <atzoumas@rudderstack.com>

Author: rudder-devbox[bot]

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
+);