From de7222a88c1fe0261340e2c9d092535a962ba79a Mon Sep 17 00:00:00 2001 From: endigma Date: Fri, 28 Aug 2026 15:06:50 +0100 Subject: [PATCH 1/5] refactor(router): separate APQ storage backends --- router/core/operation_processor.go | 22 +--- router/core/router.go | 39 +++---- .../internal/persistedoperation/apq/client.go | 104 ------------------ .../internal/persistedoperation/apq/memory.go | 51 +++++++++ .../internal/persistedoperation/apq/redis.go | 31 ++++-- .../internal/persistedoperation/apq/store.go | 13 +++ router/internal/persistedoperation/client.go | 36 +++--- .../operationstorage/cache.go | 5 +- 8 files changed, 128 insertions(+), 173 deletions(-) delete mode 100644 router/internal/persistedoperation/apq/client.go create mode 100644 router/internal/persistedoperation/apq/memory.go create mode 100644 router/internal/persistedoperation/apq/store.go diff --git a/router/core/operation_processor.go b/router/core/operation_processor.go index b666f008bf..b0d98fb4c1 100644 --- a/router/core/operation_processor.go +++ b/router/core/operation_processor.go @@ -456,7 +456,7 @@ func (o *OperationKit) FetchPersistedOperation(ctx context.Context, clientInfo * } if fromCache { if fromCacheHasTTL, _ := o.persistedOperationCacheKeyHasTtl(clientInfo.Name, includeOperationName); fromCacheHasTTL { - if err := o.renewAPQTTL(ctx, clientInfo.Name); err != nil { + if err := o.renewAPQTTL(ctx); err != nil { return false, false, err } } @@ -468,7 +468,7 @@ func (o *OperationKit) FetchPersistedOperation(ctx context.Context, clientInfo * isAPQ = true // If the operation was fetched with APQ, save it again to renew the TTL - err := o.operationProcessor.persistedOperationClient.SaveOperation(ctx, clientInfo.Name, o.parsedOperation.GraphQLRequestExtensions.PersistedQuery.Sha256Hash, o.parsedOperation.Request.Query) + err := o.operationProcessor.persistedOperationClient.SaveOperation(ctx, o.parsedOperation.GraphQLRequestExtensions.PersistedQuery.Sha256Hash, o.parsedOperation.Request.Query) if err != nil { return false, true, err } @@ -501,7 +501,7 @@ func (o *OperationKit) FetchPersistedOperation(ctx context.Context, clientInfo * // If the operation was fetched with APQ, save it again to renew the TTL if isAPQ { - if err = o.operationProcessor.persistedOperationClient.SaveOperation(ctx, clientInfo.Name, o.parsedOperation.GraphQLRequestExtensions.PersistedQuery.Sha256Hash, o.parsedOperation.Request.Query); err != nil { + if err = o.operationProcessor.persistedOperationClient.SaveOperation(ctx, o.parsedOperation.GraphQLRequestExtensions.PersistedQuery.Sha256Hash, o.parsedOperation.Request.Query); err != nil { return false, true, err } } @@ -510,21 +510,9 @@ func (o *OperationKit) FetchPersistedOperation(ctx context.Context, clientInfo * return false, isAPQ, nil } -func (o *OperationKit) renewAPQTTL(ctx context.Context, clientName string) error { +func (o *OperationKit) renewAPQTTL(ctx context.Context) error { sha256Hash := o.parsedOperation.GraphQLRequestExtensions.PersistedQuery.Sha256Hash - // Reload the raw APQ body because normalization can remove conditional fields. - operationBody, isAPQ, err := o.operationProcessor.persistedOperationClient.PersistedOperation(ctx, clientName, sha256Hash) - if err != nil { - return err - } - if !isAPQ { - return nil - } - if len(operationBody) == 0 { - return nil - } - - return o.operationProcessor.persistedOperationClient.SaveOperation(ctx, clientName, sha256Hash, string(operationBody)) + return o.operationProcessor.persistedOperationClient.RenewOperation(ctx, sha256Hash) } const ( diff --git a/router/core/router.go b/router/core/router.go index 25aca4e77a..fab5a7b670 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -1380,7 +1380,7 @@ func (r *Router) buildClients(ctx context.Context) error { return err } - apqClient, err := r.buildAPQClient(registry) + apqStore, err := r.buildAPQStore(registry) if err != nil { return err } @@ -1395,7 +1395,7 @@ func (r *Router) buildClients(ctx context.Context) error { pClient = nil } - if pClient != nil || apqClient != nil || pqlStore != nil { + if pClient != nil || apqStore != nil || pqlStore != nil { // For backwards compatibility with cdn config field cacheSize := r.persistedOperationsConfig.Cache.Size.Uint64() if cacheSize <= 0 { @@ -1406,7 +1406,7 @@ func (r *Router) buildClients(ctx context.Context) error { CacheSize: cacheSize, Logger: r.logger, ProviderClient: pClient, - ApqClient: apqClient, + APQStore: apqStore, PQLStore: pqlStore, }) if err != nil { @@ -1503,38 +1503,33 @@ func (r *Router) buildPersistedOpsClient(registry *ProviderRegistry) (persistedo return nil, nil, nil } -// buildAPQClient creates the automatic persisted queries client and its -// optional Redis backing store. -func (r *Router) buildAPQClient(registry *ProviderRegistry) (apq.Client, error) { - var kvClient apq.KVClient +// buildAPQStore creates the automatic persisted queries store. +func (r *Router) buildAPQStore(registry *ProviderRegistry) (apq.Store, error) { + if !r.automaticPersistedQueriesConfig.Enabled { + return nil, nil + } + + ttl := time.Duration(r.automaticPersistedQueriesConfig.Cache.TTL) * time.Second if provider, ok := registry.Redis(r.automaticPersistedQueriesConfig.Storage.ProviderID); ok { - c, err := apq.NewRedisClient(&apq.RedisOptions{ + store, err := apq.NewRedisStore(&apq.RedisOptions{ Logger: r.logger, StorageConfig: &provider, Prefix: r.automaticPersistedQueriesConfig.Storage.ObjectPrefix, + TTL: ttl, }) if err != nil { return nil, err } - kvClient = c r.logger.Info("Use redis as storage provider for automatic persisted operations", zap.String("provider_id", provider.ID), ) + return store, nil } - if !r.automaticPersistedQueriesConfig.Enabled { - return nil, nil - } - - apqClient, err := apq.NewClient(&apq.Options{ - Logger: r.logger, - ApqConfig: &r.automaticPersistedQueriesConfig, - KVClient: kvClient, - }) - if err != nil { - return nil, err - } - return apqClient, nil + return apq.NewMemoryStore( + int64(r.automaticPersistedQueriesConfig.Cache.Size.Uint64()), + ttl, + ) } // buildManifestStore sets up the PQL manifest store and its background poller. diff --git a/router/internal/persistedoperation/apq/client.go b/router/internal/persistedoperation/apq/client.go deleted file mode 100644 index d4c2d69e70..0000000000 --- a/router/internal/persistedoperation/apq/client.go +++ /dev/null @@ -1,104 +0,0 @@ -package apq - -import ( - "context" - "errors" - - "github.com/wundergraph/cosmo/router/internal/persistedoperation/operationstorage" - "github.com/wundergraph/cosmo/router/pkg/config" - "go.uber.org/zap" -) - -type PersistedOperation struct { - Version int `json:"version"` - Body string `json:"body"` -} - -type Client interface { - Enabled() bool - // IsDistributed returns true when the APQ store is shared across router instances (e.g. Redis). - IsDistributed() bool - PersistedOperation(ctx context.Context, clientName string, sha256Hash string) ([]byte, error) - SaveOperation(ctx context.Context, clientName, sha256Hash string, operationBody []byte) error - Close() -} - -type KVClient interface { - // Get retrieves the operation body from the KV store with the given operation hash as the key - Get(ctx context.Context, operationHash string) ([]byte, error) - // Set saves the operation body in the KV store with the given operation hash as the key and the ttl in seconds - Set(ctx context.Context, operationHash string, operationBody []byte, ttl int) error - // Close closes the KV store connection - Close() -} - -type Options struct { - Logger *zap.Logger - ApqConfig *config.AutomaticPersistedQueriesConfig - KVClient KVClient -} - -type client struct { - enabled bool - ttl int - cache *operationstorage.OperationsCache - kvClient KVClient -} - -func NewClient(opts *Options) (Client, error) { - if opts.ApqConfig == nil { - return nil, errors.New("APQ config is nil") - } - cl := &client{ - enabled: opts.ApqConfig.Enabled, - kvClient: opts.KVClient, - ttl: opts.ApqConfig.Cache.TTL, - } - - if opts.ApqConfig == nil { - return nil, errors.New("APQ config is nil") - } else if !opts.ApqConfig.Enabled || opts.KVClient != nil { - return cl, nil - } - - var err error - cl.cache, err = operationstorage.NewOperationsCache(int64(opts.ApqConfig.Cache.Size.Uint64())) - - return cl, err -} - -func (c *client) Enabled() bool { - return c.enabled -} - -func (c *client) IsDistributed() bool { - return c.kvClient != nil -} - -func (c *client) PersistedOperation(ctx context.Context, clientName string, sha256Hash string) ([]byte, error) { - if c.kvClient != nil { - return c.kvClient.Get(ctx, sha256Hash) - } - - // we don't use the client name in the APQ cache, because operations should be persisted across all clients - return c.cache.Get("", sha256Hash), nil -} - -func (c *client) SaveOperation(ctx context.Context, clientName, sha256Hash string, operationBody []byte) error { - if c.kvClient != nil { - return c.kvClient.Set(ctx, sha256Hash, operationBody, c.ttl) - } - - // we don't use the client name in the APQ cache, because operations should be persisted across all clients - c.cache.Set("", sha256Hash, operationBody, c.ttl) - return nil -} - -func (c *client) Close() { - if c.kvClient != nil { - c.kvClient.Close() - } - if c.cache != nil { - c.cache.Cache.Close() - } -} diff --git a/router/internal/persistedoperation/apq/memory.go b/router/internal/persistedoperation/apq/memory.go new file mode 100644 index 0000000000..6399dbb00a --- /dev/null +++ b/router/internal/persistedoperation/apq/memory.go @@ -0,0 +1,51 @@ +package apq + +import ( + "context" + "time" + + "github.com/wundergraph/cosmo/router/internal/persistedoperation/operationstorage" +) + +type memoryStore struct { + cache *operationstorage.OperationsCache + ttl time.Duration +} + +func NewMemoryStore(cacheSize int64, ttl time.Duration) (*memoryStore, error) { + cache, err := operationstorage.NewOperationsCache(cacheSize) + if err != nil { + return nil, err + } + return &memoryStore{ + cache: cache, + ttl: ttl, + }, nil +} + +func (m *memoryStore) Get(_ context.Context, operationHash string) ([]byte, error) { + return m.cache.Get("", operationHash), nil +} + +func (m *memoryStore) Set(_ context.Context, operationHash string, operationBody []byte) error { + m.cache.Set("", operationHash, operationBody, m.ttl) + return nil +} + +func (m *memoryStore) Renew(_ context.Context, operationHash string) error { + operationBody := m.cache.Get("", operationHash) + if len(operationBody) > 0 { + m.cache.Set("", operationHash, operationBody, m.ttl) + } + return nil +} + +func (m *memoryStore) IsDistributed() bool { + return false +} + +func (m *memoryStore) Close() { + if m.cache.Cache != nil { + m.cache.Cache.Close() + } +} diff --git a/router/internal/persistedoperation/apq/redis.go b/router/internal/persistedoperation/apq/redis.go index 67093d91bd..11524b8b38 100644 --- a/router/internal/persistedoperation/apq/redis.go +++ b/router/internal/persistedoperation/apq/redis.go @@ -14,17 +14,17 @@ import ( type RedisOptions struct { Logger *zap.Logger StorageConfig *config.RedisStorageProvider - ApqConfig *config.AutomaticPersistedQueriesConfig Prefix string + TTL time.Duration } -type redisClient struct { - logger *zap.Logger +type redisStore struct { client rd.RDCloser prefix string + ttl time.Duration } -func NewRedisClient(opts *RedisOptions) (KVClient, error) { +func NewRedisStore(opts *RedisOptions) (*redisStore, error) { if opts.StorageConfig == nil { return nil, errors.New("storage config is nil") } @@ -35,16 +35,16 @@ func NewRedisClient(opts *RedisOptions) (KVClient, error) { ClusterEnabled: opts.StorageConfig.ClusterEnabled, }) - rclient := &redisClient{ - logger: opts.Logger, + store := &redisStore{ client: rdb, prefix: opts.Prefix, + ttl: opts.TTL, } - return rclient, err + return store, err } -func (r *redisClient) Get(ctx context.Context, operationHash string) ([]byte, error) { +func (r *redisStore) Get(ctx context.Context, operationHash string) ([]byte, error) { cmd := r.client.Get(ctx, r.prefix+operationHash) if errors.Is(cmd.Err(), redis.Nil) { return nil, nil @@ -52,12 +52,19 @@ func (r *redisClient) Get(ctx context.Context, operationHash string) ([]byte, er return cmd.Bytes() } -func (r *redisClient) Set(ctx context.Context, operationHash string, operationBody []byte, ttl int) error { - ttlD := time.Duration(float64(ttl)) * time.Second - status := r.client.Set(ctx, r.prefix+operationHash, operationBody, ttlD) +func (r *redisStore) Set(ctx context.Context, operationHash string, operationBody []byte) error { + status := r.client.Set(ctx, r.prefix+operationHash, operationBody, r.ttl) return status.Err() } -func (r *redisClient) Close() { +func (r *redisStore) Renew(ctx context.Context, operationHash string) error { + return r.client.Expire(ctx, r.prefix+operationHash, r.ttl).Err() +} + +func (r *redisStore) IsDistributed() bool { + return true +} + +func (r *redisStore) Close() { _ = r.client.Close() } diff --git a/router/internal/persistedoperation/apq/store.go b/router/internal/persistedoperation/apq/store.go new file mode 100644 index 0000000000..1c77ba667f --- /dev/null +++ b/router/internal/persistedoperation/apq/store.go @@ -0,0 +1,13 @@ +package apq + +import ( + "context" +) + +type Store interface { + Get(ctx context.Context, operationHash string) ([]byte, error) + Set(ctx context.Context, operationHash string, operationBody []byte) error + Renew(ctx context.Context, operationHash string) error + IsDistributed() bool + Close() +} diff --git a/router/internal/persistedoperation/client.go b/router/internal/persistedoperation/client.go index 648373123c..25ff6c7336 100644 --- a/router/internal/persistedoperation/client.go +++ b/router/internal/persistedoperation/client.go @@ -37,14 +37,14 @@ type Options struct { Logger *zap.Logger ProviderClient StorageClient - ApqClient apq.Client + APQStore apq.Store PQLStore *pqlmanifest.Store } type Client struct { cache *operationstorage.OperationsCache providerClient StorageClient - apqClient apq.Client + apqStore apq.Store pqlStore *pqlmanifest.Store } @@ -59,14 +59,14 @@ func NewClient(opts *Options) (*Client, error) { return &Client{ providerClient: opts.ProviderClient, cache: cache, - apqClient: opts.ApqClient, + apqStore: opts.APQStore, pqlStore: opts.PQLStore, }, nil } func (c *Client) PersistedOperation(ctx context.Context, clientName string, sha256Hash string) ([]byte, bool, error) { if c.APQEnabled() { - resp, apqErr := c.apqClient.PersistedOperation(ctx, clientName, sha256Hash) + resp, apqErr := c.apqStore.Get(ctx, sha256Hash) if len(resp) > 0 || apqErr != nil { return resp, true, apqErr } @@ -96,12 +96,10 @@ func (c *Client) PersistedOperation(ctx context.Context, clientName string, sha2 return nil, c.APQEnabled(), nil } - var ( - poNotFound *PersistentOperationNotFoundError - ) + var poNotFound *PersistentOperationNotFoundError content, err := c.providerClient.PersistedOperation(ctx, clientName, sha256Hash) - if errors.As(err, &poNotFound) && c.apqClient != nil { + if errors.As(err, &poNotFound) && c.APQEnabled() { // This could well be the first time a client is requesting an APQ operation and the query is attached to the request. Return without error here, and we'll verify the operation later. return content, true, nil } @@ -114,24 +112,32 @@ func (c *Client) PersistedOperation(ctx context.Context, clientName string, sha2 return content, false, nil } -func (c *Client) SaveOperation(ctx context.Context, clientName, sha256Hash, operationBody string) error { - if c.apqClient != nil && c.apqClient.Enabled() { +func (c *Client) SaveOperation(ctx context.Context, sha256Hash, operationBody string) error { + if c.APQEnabled() { // For in-memory APQ, skip saving operations the manifest already has — // the manifest is the authoritative source and avoids redundant cache entries. // For distributed APQ (Redis), always save so all router instances can resolve the operation. - if !c.apqClient.IsDistributed() && c.ManifestEnabled() { + if !c.apqStore.IsDistributed() && c.ManifestEnabled() { if _, found := c.pqlStore.LookupByHash(sha256Hash); found { return nil } } - return c.apqClient.SaveOperation(ctx, clientName, sha256Hash, []byte(operationBody)) + return c.apqStore.Set(ctx, sha256Hash, []byte(operationBody)) } return nil } +func (c *Client) RenewOperation(ctx context.Context, sha256Hash string) error { + if !c.APQEnabled() { + return nil + } + + return c.apqStore.Renew(ctx, sha256Hash) +} + func (c *Client) APQEnabled() bool { - return c.apqClient != nil && c.apqClient.Enabled() + return c.apqStore != nil } // ManifestEnabled returns whether a PQL manifest is configured and loaded. @@ -151,7 +157,7 @@ func (c *Client) Close() { if c.cache != nil && c.cache.Cache != nil { c.cache.Cache.Close() } - if c.apqClient != nil { - c.apqClient.Close() + if c.APQEnabled() { + c.apqStore.Close() } } diff --git a/router/internal/persistedoperation/operationstorage/cache.go b/router/internal/persistedoperation/operationstorage/cache.go index 27152ac356..3a0d115c46 100644 --- a/router/internal/persistedoperation/operationstorage/cache.go +++ b/router/internal/persistedoperation/operationstorage/cache.go @@ -54,11 +54,10 @@ func (c *OperationsCache) Get(clientName string, operationHash string) []byte { return item } -func (c *OperationsCache) Set(clientName, operationHash string, operationBody []byte, ttl int) { +func (c *OperationsCache) Set(clientName, operationHash string, operationBody []byte, ttl time.Duration) { if ttl > 0 { - ttlD := time.Duration(float64(ttl)) * time.Second c.cacheLock.Lock() - c.Cache.SetWithTTL(c.key(clientName, operationHash), operationBody, int64(len(operationBody)), ttlD) + c.Cache.SetWithTTL(c.key(clientName, operationHash), operationBody, int64(len(operationBody)), ttl) c.cacheLock.Unlock() return } From 4dd10c93d2f890d01f20b8f0677fc290c9d41df8 Mon Sep 17 00:00:00 2001 From: endigma Date: Fri, 28 Aug 2026 15:49:07 +0100 Subject: [PATCH 2/5] fix(router): handle persistent APQ storage --- .../internal/persistedoperation/apq/memory.go | 5 ++++ .../persistedoperation/apq/memory_test.go | 25 ++++++++++++++++ .../internal/persistedoperation/apq/redis.go | 3 ++ .../persistedoperation/apq/redis_test.go | 30 +++++++++++++++++++ 4 files changed, 63 insertions(+) create mode 100644 router/internal/persistedoperation/apq/memory_test.go create mode 100644 router/internal/persistedoperation/apq/redis_test.go diff --git a/router/internal/persistedoperation/apq/memory.go b/router/internal/persistedoperation/apq/memory.go index 6399dbb00a..edb9cc9cda 100644 --- a/router/internal/persistedoperation/apq/memory.go +++ b/router/internal/persistedoperation/apq/memory.go @@ -2,6 +2,7 @@ package apq import ( "context" + "errors" "time" "github.com/wundergraph/cosmo/router/internal/persistedoperation/operationstorage" @@ -13,6 +14,10 @@ type memoryStore struct { } func NewMemoryStore(cacheSize int64, ttl time.Duration) (*memoryStore, error) { + if cacheSize <= 0 { + return nil, errors.New("cache size must be positive") + } + cache, err := operationstorage.NewOperationsCache(cacheSize) if err != nil { return nil, err diff --git a/router/internal/persistedoperation/apq/memory_test.go b/router/internal/persistedoperation/apq/memory_test.go new file mode 100644 index 0000000000..4ca45f512d --- /dev/null +++ b/router/internal/persistedoperation/apq/memory_test.go @@ -0,0 +1,25 @@ +package apq + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestNewMemoryStore(t *testing.T) { + t.Run("returns error when cache size is zero", func(t *testing.T) { + store, err := NewMemoryStore(0, time.Minute) + + require.Error(t, err) + require.Nil(t, store) + }) + + t.Run("returns a backed store when cache size is positive", func(t *testing.T) { + store, err := NewMemoryStore(1024*1024, time.Minute) + require.NoError(t, err) + t.Cleanup(store.Close) + + require.NotNil(t, store.cache.Cache) + }) +} diff --git a/router/internal/persistedoperation/apq/redis.go b/router/internal/persistedoperation/apq/redis.go index 11524b8b38..7de1dff0ea 100644 --- a/router/internal/persistedoperation/apq/redis.go +++ b/router/internal/persistedoperation/apq/redis.go @@ -58,6 +58,9 @@ func (r *redisStore) Set(ctx context.Context, operationHash string, operationBod } func (r *redisStore) Renew(ctx context.Context, operationHash string) error { + if r.ttl <= 0 { + return nil + } return r.client.Expire(ctx, r.prefix+operationHash, r.ttl).Err() } diff --git a/router/internal/persistedoperation/apq/redis_test.go b/router/internal/persistedoperation/apq/redis_test.go new file mode 100644 index 0000000000..07dda6bc67 --- /dev/null +++ b/router/internal/persistedoperation/apq/redis_test.go @@ -0,0 +1,30 @@ +package apq + +import ( + "testing" + "time" + + "github.com/alicebob/miniredis/v2" + "github.com/redis/go-redis/v9" + "github.com/stretchr/testify/require" +) + +func TestRedisStoreRenewRefreshesExpirationWhenTTLIsPositive(t *testing.T) { + server := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: server.Addr()}) + t.Cleanup(func() { + require.NoError(t, client.Close()) + }) + + const operationHash = "hash" + require.NoError(t, client.Set(t.Context(), operationHash, "query", time.Minute).Err()) + + store := redisStore{ + client: client, + ttl: 5 * time.Minute, + } + require.NoError(t, store.Renew(t.Context(), operationHash)) + + require.True(t, server.Exists(operationHash)) + require.Equal(t, 5*time.Minute, server.TTL(operationHash)) +} From 4ef54ad912ac436935dddfe06e73c6ca53e4c32a Mon Sep 17 00:00:00 2001 From: endigma Date: Fri, 28 Aug 2026 19:47:33 +0100 Subject: [PATCH 3/5] test(router): configure APQ caches in integration tests --- router-tests/operations/automatic_persisted_queries_test.go | 5 ++++- .../operations/persisted_operations_over_get_test.go | 5 ++++- router-tests/subscriptions/websocket_test.go | 5 ++++- router/internal/persistedoperation/apq/memory_test.go | 2 +- 4 files changed, 13 insertions(+), 4 deletions(-) diff --git a/router-tests/operations/automatic_persisted_queries_test.go b/router-tests/operations/automatic_persisted_queries_test.go index d955198eb9..e9597a6387 100644 --- a/router-tests/operations/automatic_persisted_queries_test.go +++ b/router-tests/operations/automatic_persisted_queries_test.go @@ -25,7 +25,7 @@ func TestAutomaticPersistedQueries(t *testing.T) { t.Run("local cache", func(t *testing.T) { t.Parallel() - t.Run("Sha without query fails", func(t *testing.T) { + t.Run("returns not found when an unknown hash has no query", func(t *testing.T) { t.Parallel() testenv.Run(t, &testenv.Config{ @@ -34,6 +34,9 @@ func TestAutomaticPersistedQueries(t *testing.T) { }, ApqConfig: config.AutomaticPersistedQueriesConfig{ Enabled: true, + Cache: config.AutomaticPersistedQueriesCacheConfig{ + Size: 1024 * 1024, + }, }, }, func(t *testing.T, xEnv *testenv.Environment) { res := xEnv.MakeGraphQLRequestOK(testenv.GraphQLRequest{ diff --git a/router-tests/operations/persisted_operations_over_get_test.go b/router-tests/operations/persisted_operations_over_get_test.go index fc9b00075c..c1589c4d60 100644 --- a/router-tests/operations/persisted_operations_over_get_test.go +++ b/router-tests/operations/persisted_operations_over_get_test.go @@ -88,12 +88,15 @@ func TestPersistedOperationOverGET(t *testing.T) { func TestAutomatedPersistedQueriesOverGET(t *testing.T) { t.Parallel() - t.Run("Operation not found", func(t *testing.T) { + t.Run("returns not found when the hash is unknown", func(t *testing.T) { t.Parallel() testenv.Run(t, &testenv.Config{ ApqConfig: config.AutomaticPersistedQueriesConfig{ Enabled: true, + Cache: config.AutomaticPersistedQueriesCacheConfig{ + Size: 1024 * 1024, + }, }, }, func(t *testing.T, xEnv *testenv.Environment) { header := make(http.Header) diff --git a/router-tests/subscriptions/websocket_test.go b/router-tests/subscriptions/websocket_test.go index be9c26bac4..cdc6cbb9a1 100644 --- a/router-tests/subscriptions/websocket_test.go +++ b/router-tests/subscriptions/websocket_test.go @@ -2383,12 +2383,15 @@ func TestWebSockets(t *testing.T) { }) }) - t.Run("cache poisoning is tried but prevented", func(t *testing.T) { + t.Run("rejects cache poisoning when query and hash differ", func(t *testing.T) { t.Parallel() testenv.Run(t, &testenv.Config{ ApqConfig: config.AutomaticPersistedQueriesConfig{ Enabled: true, + Cache: config.AutomaticPersistedQueriesCacheConfig{ + Size: 1024 * 1024, + }, }, }, func(t *testing.T, xEnv *testenv.Environment) { conn := xEnv.InitGraphQLWebSocketConnection(nil, nil, []byte(`{"graphql-client-name": "my-client"}`)) diff --git a/router/internal/persistedoperation/apq/memory_test.go b/router/internal/persistedoperation/apq/memory_test.go index 4ca45f512d..79faf3c912 100644 --- a/router/internal/persistedoperation/apq/memory_test.go +++ b/router/internal/persistedoperation/apq/memory_test.go @@ -8,7 +8,7 @@ import ( ) func TestNewMemoryStore(t *testing.T) { - t.Run("returns error when cache size is zero", func(t *testing.T) { + t.Run("returns an error when cache size is zero", func(t *testing.T) { store, err := NewMemoryStore(0, time.Minute) require.Error(t, err) From 263d6d9ff1f24bf06a0f2640d477f253de8f1c20 Mon Sep 17 00:00:00 2001 From: endigma Date: Tue, 1 Sep 2026 10:02:24 +0100 Subject: [PATCH 4/5] chore: review improvements --- router/core/router.go | 76 ++++++++++++------- .../internal/persistedoperation/apq/memory.go | 4 +- .../persistedoperation/apq/memory_test.go | 8 +- .../internal/persistedoperation/apq/redis.go | 8 +- .../persistedoperation/apq/redis_test.go | 36 +++++---- .../internal/persistedoperation/apq/store.go | 2 +- router/internal/persistedoperation/client.go | 10 ++- 7 files changed, 94 insertions(+), 50 deletions(-) diff --git a/router/core/router.go b/router/core/router.go index fab5a7b670..b7fd1ee878 100644 --- a/router/core/router.go +++ b/router/core/router.go @@ -463,7 +463,8 @@ func NewRouter(ctx context.Context, opts ...Option) (*Router, error) { } } - r.logger.Warn("No graph token provided. The following Cosmo Cloud features are disabled. Not recommended for Production.", + r.logger.Warn( + "No graph token provided. The following Cosmo Cloud features are disabled. Not recommended for Production.", zap.Strings("features", disabledFeatures), ) } @@ -789,7 +790,8 @@ func (r *Router) initModules(ctx context.Context) error { r.modules = append(r.modules, moduleInstance) - r.logger.Info("Module registered", + r.logger.Info( + "Module registered", zap.String("id", string(moduleInfo.ID)), zap.String("duration", time.Since(now).String()), ) @@ -905,7 +907,8 @@ func (r *Router) bootstrap(ctx context.Context) error { // Only ensure sampling rate if the user exports traces to Cosmo Cloud if cosmoCloudTracingEnabled { if r.traceConfig.Sampler > float64(r.registrationInfo.AccountLimits.TraceSamplingRate) { - r.logger.Warn("Trace sampling rate is higher than account limit. Using account limit instead. Please contact support to increase your account limit.", + r.logger.Warn( + "Trace sampling rate is higher than account limit. Using account limit instead. Please contact support to increase your account limit.", zap.Float64("limit", r.traceConfig.Sampler), zap.String("account_limit", fmt.Sprintf("%.2f", r.registrationInfo.AccountLimits.TraceSamplingRate)), ) @@ -1047,8 +1050,8 @@ func (r *Router) bootstrap(ctx context.Context) error { routerconfig.AssembleConfigRules{ SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, - }) - + }, + ) if err != nil { return fmt.Errorf("failed to assemble static execution config from manifest: %w", err) } @@ -1151,7 +1154,8 @@ func (r *Router) setupTelemetry(ctx context.Context) error { // The metric store will be passed in later when building the graph mux // because each mux has its own metric store // We'll create the exporter when building the mux in buildGraphMux - r.logger.Info("Prometheus schema field usage metrics enabled", + r.logger.Info( + "Prometheus schema field usage metrics enabled", zap.Bool("include_operation_sha", r.metricConfig.Prometheus.PromSchemaFieldUsage.IncludeOperationSha), ) } @@ -1201,7 +1205,8 @@ func (r *Router) setupResponseCache(ctx context.Context) error { case config.ResponseCacheStorageProviderMemory: return r.setupInMemoryResponseCache() default: - return fmt.Errorf("response cache storage provider %q is not supported, use %q or %q", + return fmt.Errorf( + "response cache storage provider %q is not supported, use %q or %q", provider, config.ResponseCacheStorageProviderRedis, config.ResponseCacheStorageProviderMemory, @@ -1224,7 +1229,8 @@ func (r *Router) setupInMemoryResponseCache() error { // Owned from here on by r.responseCache, which Shutdown closes. r.responseCache = cache - r.logger.Info("Response cache enabled", + r.logger.Info( + "Response cache enabled", zap.Duration("fallback_ttl", r.responseCacheConfig.FallbackTTL), zap.String("storage_provider", string(config.ResponseCacheStorageProviderMemory)), zap.Int64("max_entries", r.responseCacheConfig.Storage.MaxEntries), @@ -1238,7 +1244,8 @@ func (r *Router) setupInMemoryResponseCache() error { func (r *Router) setupRedisResponseCache(ctx context.Context) error { providerID := r.responseCacheConfig.Storage.ProviderID if providerID == "" { - return fmt.Errorf("response cache is enabled with the %q storage provider but no storage provider_id is configured; configure one, or set the storage provider to %q to cache in this router's memory instead", + return fmt.Errorf( + "response cache is enabled with the %q storage provider but no storage provider_id is configured; configure one, or set the storage provider to %q to cache in this router's memory instead", config.ResponseCacheStorageProviderRedis, config.ResponseCacheStorageProviderMemory, ) @@ -1273,7 +1280,8 @@ func (r *Router) setupRedisResponseCache(ctx context.Context) error { // Shutdown closes the cache. r.responseCache = cache - r.logger.Info("Response cache enabled", + r.logger.Info( + "Response cache enabled", zap.Duration("fallback_ttl", r.responseCacheConfig.FallbackTTL), zap.String("storage_provider", string(config.ResponseCacheStorageProviderRedis)), zap.String("key_prefix", r.responseCacheConfig.KeyPrefix), @@ -1441,7 +1449,8 @@ func (r *Router) buildPersistedOpsClient(registry *ProviderRegistry) (persistedo return nil, nil, fmt.Errorf("failed to create CDN client: %w", err) } - r.logger.Info("Use CDN as storage provider for persisted operations", + r.logger.Info( + "Use CDN as storage provider for persisted operations", zap.String("provider_id", provider.ID), ) return c, c.ReadManifest, nil @@ -1461,7 +1470,8 @@ func (r *Router) buildPersistedOpsClient(registry *ProviderRegistry) (persistedo return nil, nil, fmt.Errorf("failed to create S3 client: %w", err) } - r.logger.Info("Use S3 as storage provider for persisted operations", + r.logger.Info( + "Use S3 as storage provider for persisted operations", zap.String("provider_id", provider.ID), ) return c, c.ReadManifest, nil @@ -1475,7 +1485,8 @@ func (r *Router) buildPersistedOpsClient(registry *ProviderRegistry) (persistedo return nil, nil, fmt.Errorf("failed to create filesystem client: %w", err) } - r.logger.Info("Use file system as storage provider for persisted operations", + r.logger.Info( + "Use file system as storage provider for persisted operations", zap.String("provider_id", provider.ID), ) // Filesystem does not support manifest fetching. @@ -1494,7 +1505,8 @@ func (r *Router) buildPersistedOpsClient(registry *ProviderRegistry) (persistedo return nil, nil, fmt.Errorf("failed to create CDN client: %w", err) } - r.logger.Debug("Default to Cosmo CDN as persisted operations provider", + r.logger.Debug( + "Default to Cosmo CDN as persisted operations provider", zap.String("url", r.cdnConfig.URL), ) return c, c.ReadManifest, nil @@ -1520,7 +1532,8 @@ func (r *Router) buildAPQStore(registry *ProviderRegistry) (apq.Store, error) { if err != nil { return nil, err } - r.logger.Info("Use redis as storage provider for automatic persisted operations", + r.logger.Info( + "Use redis as storage provider for automatic persisted operations", zap.String("provider_id", provider.ID), ) return store, nil @@ -1572,7 +1585,8 @@ func (r *Router) buildManifestStore(ctx context.Context, registry *ProviderRegis storageProviderID, err) } - r.logger.Info("Loaded PQL manifest from storage provider", + r.logger.Info( + "Loaded PQL manifest from storage provider", zap.String("provider_id", storageProviderID), zap.String("object_path", objectPath), zap.String("revision", pqlStore.Revision()), @@ -1607,7 +1621,8 @@ func (r *Router) buildManifestStore(ctx context.Context, registry *ProviderRegis return nil, fmt.Errorf("failed to fetch initial PQL manifest: %w", err) } - r.logger.Info("Loaded PQL manifest from Cosmo CDN", + r.logger.Info( + "Loaded PQL manifest from Cosmo CDN", zap.String("revision", pqlStore.Revision()), zap.Int("operation_count", pqlStore.OperationCount()), ) @@ -1707,7 +1722,8 @@ func (r *Router) Start(ctx context.Context) error { r.startPQLPoller(ctx) if r.playgroundConfig.Enabled { - r.logger.Info("GraphQL endpoint", + r.logger.Info( + "GraphQL endpoint", zap.String("method", http.MethodPost), zap.String("url", r.graphqlEndpointURL), ) @@ -1722,7 +1738,8 @@ func (r *Router) Start(ctx context.Context) error { } if r.redisClient != nil { - r.logger.Info("Rate limiting enabled", + r.logger.Info( + "Rate limiting enabled", zap.Int("rate", r.rateLimit.SimpleStrategy.Rate), zap.Int("burst", r.rateLimit.SimpleStrategy.Burst), zap.Duration("duration", r.rateLimit.SimpleStrategy.Period), @@ -1748,7 +1765,8 @@ func (r *Router) Start(ctx context.Context) error { // Mark the server as ready r.httpServer.healthcheck.SetReady(true) - r.logger.Info("Server initialized and ready to serve requests", + r.logger.Info( + "Server initialized and ready to serve requests", zap.String("listen_addr", r.listenAddr), zap.Bool("playground", r.playgroundConfig.Enabled), zap.Bool("introspection", r.introspection), @@ -1798,7 +1816,8 @@ func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { r.httpServer.healthcheck.SetReady(true) - r.logger.Info("Server initialized and ready to serve requests", + r.logger.Info( + "Server initialized and ready to serve requests", zap.String("listen_addr", r.listenAddr), zap.Bool("playground", r.playgroundConfig.Enabled), zap.Bool("introspection", r.introspection), @@ -1816,7 +1835,8 @@ func (r *Router) startWithStaticExecutionConfig(ctx context.Context) error { } }() - r.logger.Info("Watching config file for changes. Router will hot-reload automatically without downtime", + r.logger.Info( + "Watching config file for changes. Router will hot-reload automatically without downtime", zap.String("path", path), ) @@ -1859,7 +1879,6 @@ func (r *Router) buildExecutionConfigWatcher(ctx context.Context, ll *zap.Logger } }, }) - if err != nil { return nil, fmt.Errorf("failed to create watcher: %w", err) } @@ -1882,8 +1901,8 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) r.manifestConfig.Path, routerconfig.AssembleConfigRules{ SkipMissingFeatureFlags: r.manifestConfig.SkipMissingFeatureFlags, IgnoredFeatureFlags: r.manifestConfig.IgnoredFeatureFlags, - }) - + }, + ) if err != nil { ll.Error("Failed to assemble static execution config from manifest", zap.Error(err)) return @@ -1897,7 +1916,6 @@ func (r *Router) buildManifestConfigWatcher(ctx context.Context, ll *zap.Logger) } }, }) - if err != nil { return nil, fmt.Errorf("failed to create watcher: %w", err) } @@ -2088,8 +2106,11 @@ func (r *Router) Shutdown(ctx context.Context) error { // Shutdown the CDN operation client and free up resources if r.persistedOperationClient != nil { - r.persistedOperationClient.Close() + if closeErr := r.persistedOperationClient.Close(); closeErr != nil { + err.Append(fmt.Errorf("failed to close persisted operation client: %w", closeErr)) + } } + if r.pqlStore != nil { r.pqlStore.Close() } @@ -2522,7 +2543,6 @@ func NewSubgraphCircuitBreakerOptions(cfg config.TrafficShapingRules) *SubgraphC // Subgraph specific circuit breakers for k, v := range cfg.Subgraphs { entry.SubgraphMap[k] = newCircuitBreakerConfig(v.CircuitBreaker) - } return entry diff --git a/router/internal/persistedoperation/apq/memory.go b/router/internal/persistedoperation/apq/memory.go index edb9cc9cda..415e407f05 100644 --- a/router/internal/persistedoperation/apq/memory.go +++ b/router/internal/persistedoperation/apq/memory.go @@ -49,8 +49,10 @@ func (m *memoryStore) IsDistributed() bool { return false } -func (m *memoryStore) Close() { +func (m *memoryStore) Close() error { if m.cache.Cache != nil { m.cache.Cache.Close() } + + return nil } diff --git a/router/internal/persistedoperation/apq/memory_test.go b/router/internal/persistedoperation/apq/memory_test.go index 79faf3c912..0d9e6b5a11 100644 --- a/router/internal/persistedoperation/apq/memory_test.go +++ b/router/internal/persistedoperation/apq/memory_test.go @@ -8,7 +8,11 @@ import ( ) func TestNewMemoryStore(t *testing.T) { + t.Parallel() + t.Run("returns an error when cache size is zero", func(t *testing.T) { + t.Parallel() + store, err := NewMemoryStore(0, time.Minute) require.Error(t, err) @@ -16,9 +20,11 @@ func TestNewMemoryStore(t *testing.T) { }) t.Run("returns a backed store when cache size is positive", func(t *testing.T) { + t.Parallel() + store, err := NewMemoryStore(1024*1024, time.Minute) require.NoError(t, err) - t.Cleanup(store.Close) + t.Cleanup(func() { _ = store.Close() }) require.NotNil(t, store.cache.Cache) }) diff --git a/router/internal/persistedoperation/apq/redis.go b/router/internal/persistedoperation/apq/redis.go index 7de1dff0ea..7f1ae295d4 100644 --- a/router/internal/persistedoperation/apq/redis.go +++ b/router/internal/persistedoperation/apq/redis.go @@ -68,6 +68,10 @@ func (r *redisStore) IsDistributed() bool { return true } -func (r *redisStore) Close() { - _ = r.client.Close() +func (r *redisStore) Close() error { + if r.client != nil { + return r.client.Close() + } + + return nil } diff --git a/router/internal/persistedoperation/apq/redis_test.go b/router/internal/persistedoperation/apq/redis_test.go index 07dda6bc67..8b273c4ba4 100644 --- a/router/internal/persistedoperation/apq/redis_test.go +++ b/router/internal/persistedoperation/apq/redis_test.go @@ -9,22 +9,28 @@ import ( "github.com/stretchr/testify/require" ) -func TestRedisStoreRenewRefreshesExpirationWhenTTLIsPositive(t *testing.T) { - server := miniredis.RunT(t) - client := redis.NewClient(&redis.Options{Addr: server.Addr()}) - t.Cleanup(func() { - require.NoError(t, client.Close()) - }) +func TestRedisStore(t *testing.T) { + t.Parallel() + + t.Run("renew refreshes expiration when TTL is positive", func(t *testing.T) { + t.Parallel() - const operationHash = "hash" - require.NoError(t, client.Set(t.Context(), operationHash, "query", time.Minute).Err()) + server := miniredis.RunT(t) + client := redis.NewClient(&redis.Options{Addr: server.Addr()}) + t.Cleanup(func() { + require.NoError(t, client.Close()) + }) - store := redisStore{ - client: client, - ttl: 5 * time.Minute, - } - require.NoError(t, store.Renew(t.Context(), operationHash)) + const operationHash = "hash" + require.NoError(t, client.Set(t.Context(), operationHash, "query", time.Minute).Err()) - require.True(t, server.Exists(operationHash)) - require.Equal(t, 5*time.Minute, server.TTL(operationHash)) + store := redisStore{ + client: client, + ttl: 5 * time.Minute, + } + require.NoError(t, store.Renew(t.Context(), operationHash)) + + require.True(t, server.Exists(operationHash)) + require.Equal(t, 5*time.Minute, server.TTL(operationHash)) + }) } diff --git a/router/internal/persistedoperation/apq/store.go b/router/internal/persistedoperation/apq/store.go index 1c77ba667f..f175a46e64 100644 --- a/router/internal/persistedoperation/apq/store.go +++ b/router/internal/persistedoperation/apq/store.go @@ -9,5 +9,5 @@ type Store interface { Set(ctx context.Context, operationHash string, operationBody []byte) error Renew(ctx context.Context, operationHash string) error IsDistributed() bool - Close() + Close() error } diff --git a/router/internal/persistedoperation/client.go b/router/internal/persistedoperation/client.go index 25ff6c7336..00db65d4bf 100644 --- a/router/internal/persistedoperation/client.go +++ b/router/internal/persistedoperation/client.go @@ -150,14 +150,20 @@ func (c *Client) PQLStore() *pqlmanifest.Store { return c.pqlStore } -func (c *Client) Close() { +func (c *Client) Close() error { if c.providerClient != nil { c.providerClient.Close() } + if c.cache != nil && c.cache.Cache != nil { c.cache.Cache.Close() } + if c.APQEnabled() { - c.apqStore.Close() + if err := c.apqStore.Close(); err != nil { + return err + } } + + return nil } From 75fb6691b9d789dc918b0f120daa4529b2eb3d41 Mon Sep 17 00:00:00 2001 From: endigma Date: Tue, 1 Sep 2026 12:37:13 +0100 Subject: [PATCH 5/5] chore: close APQ memory cache unconditionally --- router/internal/persistedoperation/apq/memory.go | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/router/internal/persistedoperation/apq/memory.go b/router/internal/persistedoperation/apq/memory.go index 415e407f05..6331fdc5c5 100644 --- a/router/internal/persistedoperation/apq/memory.go +++ b/router/internal/persistedoperation/apq/memory.go @@ -50,9 +50,6 @@ func (m *memoryStore) IsDistributed() bool { } func (m *memoryStore) Close() error { - if m.cache.Cache != nil { - m.cache.Cache.Close() - } - + m.cache.Cache.Close() return nil }