feat(cache): add ListableCache interface for enumerating cache keys

CHANGELOG.md

@@ -8,6 +8,8 @@ All notable changes to this project will be documented in this file.
 ### Added
 
 - `parse_big_decimal` bloblang method for Kafka Connect / Debezium decimal decoding @aratz-lasa
+- `ListableCache` interface added to the service package. This is an optional interface for caches that are able to enumerate their keys, detected via type assertion on caches obtained from `*Resources.AccessCache` @ecordell
+- `memory`, `file`, `aws_s3` and `gcp_cloud_storage` caches now implement `ListableCache` @ecordell
 
 ## 1.19.0 - 2026-07-10
 

internal/component/cache/cache_metrics.go

@@ -35,13 +35,14 @@ type metricsCache struct {
 }
 
 // MetricsForCache wraps a cache with a struct that adds standard metrics over
-// each method.
+// each method. If the cache implements KeyLister then the returned cache will
+// also implement it.
 func MetricsForCache(c V1, stats metrics.Type) V1 {
 	cacheSuccess := stats.GetCounterVec("cache_success", "operation")
 	cacheError := stats.GetCounterVec("cache_error", "operation")
 	cacheLatency := stats.GetTimerVec("cache_latency_ns", "operation")
 
-	return &metricsCache{
+	mCache := &metricsCache{
 		c: c, sig: shutdown.NewSignaller(),
 
 		mGetNotFound: stats.GetCounterVec("cache_not_found", "operation").With("get"),
@@ -62,6 +63,19 @@ func MetricsForCache(c V1, stats metrics.Type) V1 {
 		mDelSuccess: cacheSuccess.With("delete"),
 		mDelLatency: cacheLatency.With("delete"),
 	}
+
+	kl, ok := c.(KeyLister)
+	if !ok {
+		return mCache
+	}
+	return &listableMetricsCache{
+		metricsCache: mCache,
+		kl:           kl,
+
+		mListKeysError:   cacheError.With("list_keys"),
+		mListKeysSuccess: cacheSuccess.With("list_keys"),
+		mListKeysLatency: cacheLatency.With("list_keys"),
+	}
 }
 
 func (a *metricsCache) Get(ctx context.Context, key string) ([]byte, error) {
@@ -135,3 +149,26 @@ func (a *metricsCache) Delete(ctx context.Context, key string) error {
 func (a *metricsCache) Close(ctx context.Context) error {
 	return a.c.Close(ctx)
 }
+
+//------------------------------------------------------------------------------
+
+type listableMetricsCache struct {
+	*metricsCache
+	kl KeyLister
+
+	mListKeysError   metrics.StatCounter
+	mListKeysSuccess metrics.StatCounter
+	mListKeysLatency metrics.StatTimer
+}
+
+func (a *listableMetricsCache) ListKeys(ctx context.Context) ([]string, error) {
+	started := time.Now()
+	keys, err := a.kl.ListKeys(ctx)
+	a.mListKeysLatency.Timing(int64(time.Since(started)))
+	if err != nil {
+		a.mListKeysError.Incr(1)
+	} else {
+		a.mListKeysSuccess.Incr(1)
+	}
+	return keys, err
+}

internal/component/cache/cache_metrics_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 
 	"github.com/warpstreamlabs/bento/internal/component"
 	"github.com/warpstreamlabs/bento/internal/component/metrics"
@@ -253,3 +254,52 @@ func TestCacheAirGapDelete(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, map[string]testCacheItem{}, rl.m)
 }
+
+type listableClosableCache struct {
+	*closableCache
+}
+
+func (c *listableClosableCache) ListKeys(ctx context.Context) ([]string, error) {
+	if c.err != nil {
+		return nil, c.err
+	}
+	keys := make([]string, 0, len(c.m))
+	for k := range c.m {
+		keys = append(keys, k)
+	}
+	return keys, nil
+}
+
+func TestCacheAirGapListKeys(t *testing.T) {
+	ctx := context.Background()
+	rl := &listableClosableCache{
+		closableCache: &closableCache{
+			m: map[string]testCacheItem{
+				"foo": {
+					b: []byte("bar"),
+				},
+				"baz": {
+					b: []byte("qux"),
+				},
+			},
+		},
+	}
+	agrl := MetricsForCache(rl, metrics.Noop())
+
+	kl, ok := agrl.(KeyLister)
+	require.True(t, ok)
+
+	keys, err := kl.ListKeys(ctx)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"foo", "baz"}, keys)
+}
+
+func TestCacheAirGapNotListable(t *testing.T) {
+	rl := &closableCache{
+		m: map[string]testCacheItem{},
+	}
+	agrl := MetricsForCache(rl, metrics.Noop())
+
+	_, ok := agrl.(KeyLister)
+	assert.False(t, ok)
+}

internal/component/cache/interface.go

@@ -11,6 +11,13 @@ type TTLItem struct {
 	TTL   *time.Duration
 }
 
+// KeyLister is an optional interface implemented by caches that are capable
+// of enumerating the keys they hold.
+type KeyLister interface {
+	// ListKeys returns a slice of all keys currently held by the cache.
+	ListKeys(ctx context.Context) ([]string, error)
+}
+
 // V1 Defines a common interface of cache implementations.
 type V1 interface {
 	// Get attempts to locate and return a cached value by its key, returns an

internal/impl/aws/cache_s3.go

@@ -223,6 +223,40 @@ func (s *s3Cache) Delete(ctx context.Context, key string) (err error) {
 	}
 }
 
+func (s *s3Cache) ListKeys(ctx context.Context) ([]string, error) {
+	boff := s.boffPool.Get().(backoff.BackOff)
+	defer func() {
+		boff.Reset()
+		s.boffPool.Put(boff)
+	}()
+
+	var keys []string
+	pager := s3.NewListObjectsV2Paginator(s.s3, &s3.ListObjectsV2Input{
+		Bucket: &s.bucket,
+	})
+	for pager.HasMorePages() {
+		page, err := pager.NextPage(ctx)
+		if err != nil {
+			wait := boff.NextBackOff()
+			if wait == backoff.Stop {
+				return nil, err
+			}
+			select {
+			case <-time.After(wait):
+			case <-ctx.Done():
+				return nil, err
+			}
+			continue
+		}
+		for _, obj := range page.Contents {
+			if obj.Key != nil {
+				keys = append(keys, *obj.Key)
+			}
+		}
+	}
+	return keys, nil
+}
+
 func (s *s3Cache) Close(context.Context) error {
 	return nil
 }

internal/impl/aws/integration_s3_test.go

@@ -379,6 +379,7 @@ cache_resources:
 			integration.CacheTestDoubleAdd(),
 			integration.CacheTestDelete(),
 			integration.CacheTestGetAndSet(1),
+			integration.CacheTestListKeys(5),
 		)
 		suite.Run(
 			t, template,

internal/impl/gcp/cache_cloud_storage.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/storage"
+	"google.golang.org/api/iterator"
 
 	"github.com/warpstreamlabs/bento/public/service"
 )
@@ -129,6 +130,22 @@ func (c *gcpCloudStorageCache) Delete(ctx context.Context, key string) error {
 	return c.bucketHandle.Object(key).Delete(ctx)
 }
 
+func (c *gcpCloudStorageCache) ListKeys(ctx context.Context) ([]string, error) {
+	var keys []string
+	it := c.bucketHandle.Objects(ctx, nil)
+	for {
+		attrs, err := it.Next()
+		if errors.Is(err, iterator.Done) {
+			break
+		}
+		if err != nil {
+			return nil, err
+		}
+		keys = append(keys, attrs.Name)
+	}
+	return keys, nil
+}
+
 func (c *gcpCloudStorageCache) Close(ctx context.Context) error {
 	return nil
 }

internal/impl/io/cache_file.go

@@ -3,6 +3,7 @@ package io
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io/fs"
 	"os"
 	"path/filepath"
@@ -88,6 +89,44 @@ func (f *fileCache) Delete(_ context.Context, key string) error {
 	return f.mgr.FS().Remove(filepath.Join(f.dir, key))
 }
 
+func (f *fileCache) ListKeys(ctx context.Context) ([]string, error) {
+	var keys []string
+	var walk func(rel string) error
+	walk = func(rel string) error {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		dir, err := f.mgr.FS().Open(filepath.Join(f.dir, rel))
+		if err != nil {
+			return err
+		}
+		defer dir.Close()
+
+		rd, ok := dir.(fs.ReadDirFile)
+		if !ok {
+			return fmt.Errorf("unable to list directory: %v", filepath.Join(f.dir, rel))
+		}
+		entries, err := rd.ReadDir(-1)
+		if err != nil {
+			return err
+		}
+		for _, e := range entries {
+			if e.IsDir() {
+				if err := walk(filepath.Join(rel, e.Name())); err != nil {
+					return err
+				}
+			} else {
+				keys = append(keys, filepath.Join(rel, e.Name()))
+			}
+		}
+		return nil
+	}
+	if err := walk(""); err != nil {
+		return nil, err
+	}
+	return keys, nil
+}
+
 func (f *fileCache) Close(context.Context) error {
 	return nil
 }

internal/impl/io/cache_file_test.go

@@ -3,6 +3,7 @@ package io
 import (
 	"context"
 	"os"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -48,3 +49,42 @@ func TestFileCache(t *testing.T) {
 	_, err = c.Get(tCtx, "foo")
 	assert.Equal(t, service.ErrKeyNotFound, err)
 }
+
+func TestFileCacheListKeys(t *testing.T) {
+	dir := t.TempDir()
+
+	tCtx := context.Background()
+	c := newFileCache(dir, service.MockResources())
+
+	var _ service.ListableCache = c
+
+	require.NoError(t, c.Set(tCtx, "foo", []byte("1"), nil))
+	require.NoError(t, c.Set(tCtx, "bar", []byte("2"), nil))
+
+	nestedKey := filepath.Join("nested", "baz")
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "nested"), 0o755))
+	require.NoError(t, c.Set(tCtx, nestedKey, []byte("3"), nil))
+
+	keys, err := c.ListKeys(tCtx)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"foo", "bar", nestedKey}, keys)
+
+	require.NoError(t, c.Delete(tCtx, "foo"))
+
+	keys, err = c.ListKeys(tCtx)
+	require.NoError(t, err)
+	assert.ElementsMatch(t, []string{"bar", nestedKey}, keys)
+}
+
+func TestFileCacheListKeysCancelled(t *testing.T) {
+	dir := t.TempDir()
+
+	c := newFileCache(dir, service.MockResources())
+	require.NoError(t, c.Set(context.Background(), "foo", []byte("1"), nil))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := c.ListKeys(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+}

internal/impl/pure/cache_memory.go

@@ -239,6 +239,21 @@ func (m *memoryCache) Delete(_ context.Context, key string) error {
 	return nil
 }
 
+func (m *memoryCache) ListKeys(_ context.Context) ([]string, error) {
+	var keys []string
+	for _, shard := range m.shards {
+		shard.RLock()
+		for k, v := range shard.items {
+			// Simulate compaction by omitting keys with an expired ttl.
+			if !shard.isExpired(v) {
+				keys = append(keys, k)
+			}
+		}
+		shard.RUnlock()
+	}
+	return keys, nil
+}
+
 func (m *memoryCache) Close(context.Context) error {
 	return nil
 }