From 1d8d350591e12a3d3051ea0d98ca92aa387161ed Mon Sep 17 00:00:00 2001 From: taitelee Date: Tue, 9 Jun 2026 22:57:13 -0400 Subject: [PATCH 01/21] feat(cache): namespace-versioned cache keys for queries and pipes --- internal/cache/local.go | 34 +++++++++++++++ internal/cache/version_manager.go | 70 +++++++++++++++++++++++++++++-- 2 files changed, 101 insertions(+), 3 deletions(-) diff --git a/internal/cache/local.go b/internal/cache/local.go index 4faea349..a68f05a2 100644 --- a/internal/cache/local.go +++ b/internal/cache/local.go @@ -57,6 +57,40 @@ func (l *LocalCache) InvalidateCache(_ context.Context, versionKeys []string) (u return uint64(len(versionKeys)), nil } +// getRaw fetches a value by an already-built cache key (no version folding). The +// pipe layer folds a multi-namespace key via the VersionManager, then calls this. +// Returns nil, 0, nil on miss. +func (l *LocalCache) getRaw(key string) ([]byte, time.Duration, error) { + val, found := l.cache.Get(key) + if !found { + return nil, 0, nil + } + remaining, _ := l.cache.GetTTL(key) + return val, remaining, nil +} + +// setRaw stores a value under an already-built cache key (no version folding). +func (l *LocalCache) setRaw(key string, value []byte, ttl time.Duration) error { + if ok := l.cache.SetWithTTL(key, value, int64(len(value)), ttl); !ok { + return fmt.Errorf("cache admission rejected for key %q", key) + } + return nil +} + +// GetQuery looks up a cached query RESULT by its sha (hash of SQL+params) and the +// namespaces it depends on. This is the general namespace-cache read, used by BOTH +// structured queries (which pass one Namespace) and pipes (which pass several). +// Returns nil, 0, nil on miss. +func (l *LocalCache) GetQuery(_ context.Context, sha string, deps []Namespace) ([]byte, time.Duration, error) { + return l.getRaw(l.versionManager.QueryKey(sha, deps)) +} + +// SetQuery stores a query result under the folded key for its dependency +// namespaces. General write used by both structured queries and pipes. +func (l *LocalCache) SetQuery(_ context.Context, sha string, deps []Namespace, value []byte, ttl time.Duration) error { + return l.setRaw(l.versionManager.QueryKey(sha, deps), value, ttl) +} + // Wait blocks until all buffered writes have been applied. // Exposed for testing; production callers rarely need this. func (l *LocalCache) Wait() { diff --git a/internal/cache/version_manager.go b/internal/cache/version_manager.go index 3b435fc1..9fbe351b 100644 --- a/internal/cache/version_manager.go +++ b/internal/cache/version_manager.go @@ -2,6 +2,8 @@ package cache import ( "fmt" + "sort" + "strings" "sync" "github.com/nats-io/nats.go" @@ -13,15 +15,21 @@ import ( type VersionManager struct { mu sync.RWMutex versions map[string]uint64 - conn *nats.Conn + + tableVersions map[string]uint64 // -> table_version + namespaceVersions map[string]uint64 //
.. -> namespace_version + + conn *nats.Conn } // NewVersionManager initializes the thread-safe version store. // Optionally initialized with a NATS connection, so that each version manager on every distributed server can keep in sync – NOT IMPLEMENTED, just wired in func NewVersionManager(conn *nats.Conn) *VersionManager { return &VersionManager{ - versions: make(map[string]uint64), - conn: conn, + versions: make(map[string]uint64), + tableVersions: make(map[string]uint64), + namespaceVersions: make(map[string]uint64), + conn: conn, } } @@ -47,3 +55,59 @@ func (vm *VersionManager) IncrementVersion(versionKey string) { defer vm.mu.Unlock() vm.versions[versionKey]++ } + +type Namespace struct { + Table string + Scope string +} + +// namespaceKeyLocked builds the namespace-table key; caller must hold vm.mu. +func (vm *VersionManager) namespaceKeyLocked(table, scope string) string { + return fmt.Sprintf("%s.%d.%s", table, vm.tableVersions[table], scope) +} + +// NamespaceKey renders the namespace-table key for (table, scope) at the table's +// current version: "
.." (scopeless scope is "", so +// e.g. "
.."). +func (vm *VersionManager) NamespaceKey(table, scope string) string { + vm.mu.RLock() + defer vm.mu.RUnlock() + return vm.namespaceKeyLocked(table, scope) +} + +// QueryKey builds the queries-table key for a result that depends on deps: the +// query's sha (hash of SQL+params) folded with every dependency's namespace key +// AND its namespace version, so a bump of any dependency misses the key. A +// structured query passes one Namespace; a pipe passes several. Deps are sorted +// so their order never changes the key. +func (vm *VersionManager) QueryKey(sha string, deps []Namespace) string { + vm.mu.RLock() + defer vm.mu.RUnlock() + segs := make([]string, len(deps)) + for i, d := range deps { + nsKey := vm.namespaceKeyLocked(d.Table, d.Scope) + segs[i] = fmt.Sprintf("%s.%d", nsKey, vm.namespaceVersions[nsKey]) + } + sort.Strings(segs) + return sha + "|" + strings.Join(segs, "|") +} + +// BumpTable advances a table's version, orphaning every namespace — and every +// cached query — that depends on the table, in one step (the whole-table nuke). +func (vm *VersionManager) BumpTable(table string) { + vm.mu.Lock() + defer vm.mu.Unlock() + vm.tableVersions[table]++ +} + +// BumpNamespace advances one (table, scope) namespace plus the table's whole-table +// (empty-scope) view, since a write to a named scope also changes the whole-table +// result; other scopes' cached queries stay valid. +func (vm *VersionManager) BumpNamespace(table, scope string) { + vm.mu.Lock() + defer vm.mu.Unlock() + vm.namespaceVersions[vm.namespaceKeyLocked(table, scope)]++ + if scope != "" { + vm.namespaceVersions[vm.namespaceKeyLocked(table, "")]++ + } +} From 218cad04acfaabf6ccd77b2c1de11ef611bab55d Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 10 Jun 2026 09:17:19 -0400 Subject: [PATCH 02/21] condensed new namespace query functions --- internal/cache/local.go | 36 ++++++++++++++---------------------- 1 file changed, 14 insertions(+), 22 deletions(-) diff --git a/internal/cache/local.go b/internal/cache/local.go index a68f05a2..08a873b9 100644 --- a/internal/cache/local.go +++ b/internal/cache/local.go @@ -57,38 +57,30 @@ func (l *LocalCache) InvalidateCache(_ context.Context, versionKeys []string) (u return uint64(len(versionKeys)), nil } -// getRaw fetches a value by an already-built cache key (no version folding). The -// pipe layer folds a multi-namespace key via the VersionManager, then calls this. -// Returns nil, 0, nil on miss. -func (l *LocalCache) getRaw(key string) ([]byte, time.Duration, error) { - val, found := l.cache.Get(key) - if !found { - return nil, 0, nil - } - remaining, _ := l.cache.GetTTL(key) - return val, remaining, nil -} - -// setRaw stores a value under an already-built cache key (no version folding). -func (l *LocalCache) setRaw(key string, value []byte, ttl time.Duration) error { - if ok := l.cache.SetWithTTL(key, value, int64(len(value)), ttl); !ok { - return fmt.Errorf("cache admission rejected for key %q", key) - } - return nil -} - // GetQuery looks up a cached query RESULT by its sha (hash of SQL+params) and the // namespaces it depends on. This is the general namespace-cache read, used by BOTH // structured queries (which pass one Namespace) and pipes (which pass several). // Returns nil, 0, nil on miss. func (l *LocalCache) GetQuery(_ context.Context, sha string, deps []Namespace) ([]byte, time.Duration, error) { - return l.getRaw(l.versionManager.QueryKey(sha, deps)) + cacheKey := l.versionManager.QueryKey(sha, deps) + + val, found := l.cache.Get(cacheKey) + if !found { + return nil, 0, nil + } + remaining, _ := l.cache.GetTTL(cacheKey) + return val, remaining, nil } // SetQuery stores a query result under the folded key for its dependency // namespaces. General write used by both structured queries and pipes. func (l *LocalCache) SetQuery(_ context.Context, sha string, deps []Namespace, value []byte, ttl time.Duration) error { - return l.setRaw(l.versionManager.QueryKey(sha, deps), value, ttl) + cacheKey := l.versionManager.QueryKey(sha, deps) + + if ok := l.cache.SetWithTTL(cacheKey, value, int64(len(value)), ttl); !ok { + return fmt.Errorf("cache admission rejected for key %q", cacheKey) + } + return nil } // Wait blocks until all buffered writes have been applied. From 289fa1126e34627cd6cdf4ca9ad5d1e74edceb51 Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 10 Jun 2026 11:26:37 -0400 Subject: [PATCH 03/21] refactor(cache): cut over to namespace-versioned Get/Set/Invalidate --- internal/api/pipes.go | 17 ++- internal/api/structured_query.go | 6 +- internal/cache/cache.go | 33 ++---- internal/cache/cache_test.go | 10 -- internal/cache/local.go | 59 ++++------ internal/cache/local_test.go | 46 ++++---- internal/cache/version_manager.go | 37 ++----- internal/cache/version_manager_test.go | 68 ++++++++---- internal/ingest/worker.go | 26 +++-- internal/ingest/worker_test.go | 145 ++++++++++++------------- internal/testutil/mocks.go | 20 ++-- 11 files changed, 218 insertions(+), 249 deletions(-) diff --git a/internal/api/pipes.go b/internal/api/pipes.go index 4949cd82..93c6823f 100644 --- a/internal/api/pipes.go +++ b/internal/api/pipes.go @@ -12,7 +12,6 @@ import ( "github.com/Wave-RF/WaveHouse/internal/cache" "github.com/Wave-RF/WaveHouse/internal/pipes" "github.com/Wave-RF/WaveHouse/internal/policy" - "github.com/Wave-RF/WaveHouse/internal/query" "github.com/go-chi/chi/v5" "golang.org/x/sync/singleflight" ) @@ -137,16 +136,14 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) { return } - // TODO: scope impl - scope := "" - safePipeName := query.SafeEncodeNATS(name) - - // TODO: current pipe impl doesn't have a list of tables/scopes, so ingest worker cannot invalidate it - - // Cache. + // Cache. A pipe can read several tables, but the current pipe impl doesn't + // expose its table/scope dependencies, so we pass no deps: the result is keyed + // by sha alone (TTL-only) and the ingest worker cannot version-invalidate it. + // TODO: once pipes expose their tables/scopes, pass them as deps here so writes + // invalidate cached pipe results. cacheKey := queryCacheKey(sql, params) if h.Cache != nil { - if data, _, err := h.Cache.Get(r.Context(), cacheKey, safePipeName, scope); err == nil && data != nil { + if data, _, err := h.Cache.Get(r.Context(), cacheKey, nil); err == nil && data != nil { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Cache", "HIT") _, _ = w.Write(data) @@ -177,7 +174,7 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) { ttl := cache.QueryTimeToTTL(queryDuration) if h.Cache != nil { - _ = h.Cache.Set(r.Context(), cacheKey, safePipeName, scope, data, ttl) + _ = h.Cache.Set(r.Context(), cacheKey, nil, data, ttl) } return data, nil }) diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index dc0b6678..9188ccd6 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -126,10 +126,12 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) // TODO: impl scope scope := "" safeTableName := query.SafeEncodeNATS(table) + // A structured query reads one table, so it depends on a single namespace. + deps := []cache.Namespace{{Table: safeTableName, Scope: scope}} // Try cache. if h.Cache != nil { - if data, _, err := h.Cache.Get(r.Context(), cacheKey, safeTableName, scope); err == nil && data != nil { + if data, _, err := h.Cache.Get(r.Context(), cacheKey, deps); err == nil && data != nil { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Cache", "HIT") _, _ = w.Write(data) //nolint:gosec // G705 XSS only JSON @@ -165,7 +167,7 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) ttl := cache.QueryTimeToTTL(queryDuration) if h.Cache != nil { - _ = h.Cache.Set(r.Context(), cacheKey, safeTableName, scope, data, ttl) + _ = h.Cache.Set(r.Context(), cacheKey, deps, data, ttl) } return data, nil }) diff --git a/internal/cache/cache.go b/internal/cache/cache.go index 2966968a..90401d62 100644 --- a/internal/cache/cache.go +++ b/internal/cache/cache.go @@ -2,30 +2,28 @@ package cache import ( "context" - "fmt" "time" ) -// Cache provides key-value storage with TTL support. +// Cache provides versioned query-result storage with TTL support. type Cache interface { - // Get retrieves a value and its remaining TTL. Returns nil, 0, nil on miss. - // key is the hashed cache key value, namespace is the table or pipe, and scope applies roles - Get(ctx context.Context, key string, namespace string, scope string) ([]byte, time.Duration, error) + // Get retrieves a cached query result and its remaining TTL. sha is the hash of + // the SQL+params; deps are the namespaces the result depends on (one for a + // structured query, several for a pipe). Returns nil, 0, nil on miss. + Get(ctx context.Context, sha string, deps []Namespace) ([]byte, time.Duration, error) // TODO: TTL should be set based on query execution time - // Set stores a value with the given TTL. - // key is the hashed cache key value, namespace is the table or pipe, and scope applies roles - Set(ctx context.Context, key string, namespace string, scope string, value []byte, ttl time.Duration) error + // Set stores a query result keyed by sha + its dependency namespaces. + Set(ctx context.Context, sha string, deps []Namespace, value []byte, ttl time.Duration) error // TODO: option to prefetch pipes when invalidated? // TODO: AST query builder needs to give us a deterministic key or bypass cache entirely - // Version Registry - // Version operations accept a scope (e.g., "org_123") - // If scope is empty, it bumps the global table version - - // InvalidateCache invalidates the cache for a given list of versionKeys - InvalidateCache(ctx context.Context, versionKeys []string) (uint64, error) + // Invalidate bumps the version for each namespace, orphaning every cached query + // that depends on it. A namespace with an empty Scope bumps the whole table + // (every scope); a non-empty Scope bumps just that scope plus the whole-table + // view. Returns the number of namespaces processed. + Invalidate(ctx context.Context, namespaces []Namespace) (uint64, error) // TODO: for local cache, we can just store the versions in memory, but for distributed/L2 cache, we will need to be able to either have stored procedures/pipelines etc to query them and attach them to a query, or sync them to each edge api server. @@ -53,10 +51,3 @@ func QueryTimeToTTL(queryTime time.Duration) time.Duration { return ttl } - -func generateVersionKey(namespace string, scope string) string { - if scope == "" { - return namespace - } - return fmt.Sprintf("%s.%s", namespace, scope) -} diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 2a761b0f..2b909aa4 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -50,13 +50,3 @@ func TestQueryTimeToTTL(t *testing.T) { }) } } - -func TestGenerateVersionKey(t *testing.T) { - t.Parallel() - - // Test empty scope - assert.Equal(t, "users", generateVersionKey("users", "")) - - // Test populated scope - assert.Equal(t, "users.org_123", generateVersionKey("users", "org_123")) -} diff --git a/internal/cache/local.go b/internal/cache/local.go index 08a873b9..e5a5b3bf 100644 --- a/internal/cache/local.go +++ b/internal/cache/local.go @@ -28,40 +28,10 @@ func NewLocal(maxCost int64) (*LocalCache, error) { return &LocalCache{cache: cache, versionManager: vm}, nil } -func (l *LocalCache) Get(_ context.Context, key string, namespace string, scope string) ([]byte, time.Duration, error) { - cacheKey := l.versionManager.GetCacheKey(key, namespace, scope) - - val, foundVal := l.cache.Get(cacheKey) - if !foundVal { - return nil, 0, nil - } - remaining, _ := l.cache.GetTTL(cacheKey) - - return val, remaining, nil -} - -func (l *LocalCache) Set(_ context.Context, key string, namespace string, scope string, value []byte, ttl time.Duration) error { - cacheKey := l.versionManager.GetCacheKey(key, namespace, scope) - - // set cost = 0 for dynamic cost evaluation - if ok := l.cache.SetWithTTL(cacheKey, value, int64(len(value)), ttl); !ok { - return fmt.Errorf("cache admission rejected for key %q", cacheKey) - } - return nil -} - -func (l *LocalCache) InvalidateCache(_ context.Context, versionKeys []string) (uint64, error) { - for _, key := range versionKeys { - l.versionManager.IncrementVersion(key) - } - return uint64(len(versionKeys)), nil -} - -// GetQuery looks up a cached query RESULT by its sha (hash of SQL+params) and the -// namespaces it depends on. This is the general namespace-cache read, used by BOTH -// structured queries (which pass one Namespace) and pipes (which pass several). -// Returns nil, 0, nil on miss. -func (l *LocalCache) GetQuery(_ context.Context, sha string, deps []Namespace) ([]byte, time.Duration, error) { +// Get looks up a cached query RESULT by its sha (hash of SQL+params) and the +// namespaces it depends on. Used by BOTH structured queries (which pass one +// Namespace) and pipes (which pass several). Returns nil, 0, nil on miss. +func (l *LocalCache) Get(_ context.Context, sha string, deps []Namespace) ([]byte, time.Duration, error) { cacheKey := l.versionManager.QueryKey(sha, deps) val, found := l.cache.Get(cacheKey) @@ -72,9 +42,9 @@ func (l *LocalCache) GetQuery(_ context.Context, sha string, deps []Namespace) ( return val, remaining, nil } -// SetQuery stores a query result under the folded key for its dependency -// namespaces. General write used by both structured queries and pipes. -func (l *LocalCache) SetQuery(_ context.Context, sha string, deps []Namespace, value []byte, ttl time.Duration) error { +// Set stores a query result under the folded key for its dependency namespaces. +// Used by both structured queries and pipes. +func (l *LocalCache) Set(_ context.Context, sha string, deps []Namespace, value []byte, ttl time.Duration) error { cacheKey := l.versionManager.QueryKey(sha, deps) if ok := l.cache.SetWithTTL(cacheKey, value, int64(len(value)), ttl); !ok { @@ -83,6 +53,21 @@ func (l *LocalCache) SetQuery(_ context.Context, sha string, deps []Namespace, v return nil } +// Invalidate bumps the version for each namespace, instantly orphaning every +// cached query that depends on it. An empty Scope bumps the whole table (every +// scope at once); a non-empty Scope bumps just that scope plus the whole-table +// view. Returns the number of namespaces processed. +func (l *LocalCache) Invalidate(_ context.Context, namespaces []Namespace) (uint64, error) { + for _, ns := range namespaces { + if ns.Scope == "" { + l.versionManager.BumpTable(ns.Table) + } else { + l.versionManager.BumpNamespace(ns.Table, ns.Scope) + } + } + return uint64(len(namespaces)), nil +} + // Wait blocks until all buffered writes have been applied. // Exposed for testing; production callers rarely need this. func (l *LocalCache) Wait() { diff --git a/internal/cache/local_test.go b/internal/cache/local_test.go index da44a531..e050874b 100644 --- a/internal/cache/local_test.go +++ b/internal/cache/local_test.go @@ -15,7 +15,7 @@ func TestLocalCache_GetMiss(t *testing.T) { require.NoError(t, err) defer func() { _ = c.Close() }() - val, ttl, err := c.Get(context.Background(), "missing", "", "") + val, ttl, err := c.Get(context.Background(), "missing", []Namespace{{Table: "table"}}) assert.NoError(t, err) assert.Nil(t, val) assert.Zero(t, ttl) @@ -28,13 +28,14 @@ func TestLocalCache_SetAndGet(t *testing.T) { defer func() { _ = c.Close() }() ctx := context.Background() - err = c.Set(ctx, "key1", "table", "scope", []byte("hello"), 10*time.Second) + deps := []Namespace{{Table: "table", Scope: "scope"}} + err = c.Set(ctx, "key1", deps, []byte("hello"), 10*time.Second) assert.NoError(t, err) // Ristretto uses async admission — wait briefly for it to be admitted. c.Wait() - val, ttl, err := c.Get(ctx, "key1", "table", "scope") + val, ttl, err := c.Get(ctx, "key1", deps) assert.NoError(t, err) assert.Equal(t, []byte("hello"), val) assert.True(t, ttl > 0, "expected positive remaining TTL") @@ -47,15 +48,16 @@ func TestLocalCache_ExpiredKey(t *testing.T) { defer func() { _ = c.Close() }() ctx := context.Background() + deps := []Namespace{{Table: "table"}} // Set with very short TTL. - err = c.Set(ctx, "expires", "table", "", []byte("data"), 1*time.Millisecond) + err = c.Set(ctx, "expires", deps, []byte("data"), 1*time.Millisecond) assert.NoError(t, err) // Ensure async admission completes, then wait for expiry. c.Wait() time.Sleep(50 * time.Millisecond) - val, _, err := c.Get(ctx, "expires", "table", "") + val, _, err := c.Get(ctx, "expires", deps) assert.NoError(t, err) assert.Nil(t, val, "expected nil for expired key") } @@ -67,12 +69,13 @@ func TestLocalCache_Overwrite(t *testing.T) { defer func() { _ = c.Close() }() ctx := context.Background() - require.NoError(t, c.Set(ctx, "key", "table", "", []byte("v1"), 10*time.Second)) + deps := []Namespace{{Table: "table"}} + require.NoError(t, c.Set(ctx, "key", deps, []byte("v1"), 10*time.Second)) c.Wait() - require.NoError(t, c.Set(ctx, "key", "table", "", []byte("v2"), 10*time.Second)) + require.NoError(t, c.Set(ctx, "key", deps, []byte("v2"), 10*time.Second)) c.Wait() - val, _, err := c.Get(ctx, "key", "table", "") + val, _, err := c.Get(ctx, "key", deps) assert.NoError(t, err) assert.Equal(t, []byte("v2"), val) } @@ -84,13 +87,14 @@ func TestLocalCache_ZeroTTL(t *testing.T) { defer func() { _ = c.Close() }() ctx := context.Background() - err = c.Set(ctx, "notimed", "table", "", []byte("data"), 0) + deps := []Namespace{{Table: "table"}} + err = c.Set(ctx, "notimed", deps, []byte("data"), 0) assert.NoError(t, err) c.Wait() time.Sleep(10 * time.Millisecond) // arbitrary tiny sleep to see its still here after - val, ttl, err := c.Get(ctx, "notimed", "table", "") + val, ttl, err := c.Get(ctx, "notimed", deps) assert.NoError(t, err) if val != nil { assert.Equal(t, []byte("data"), val) @@ -98,35 +102,33 @@ func TestLocalCache_ZeroTTL(t *testing.T) { } } -func TestLocalCache_InvalidateCache(t *testing.T) { +func TestLocalCache_Invalidate(t *testing.T) { t.Parallel() c, err := NewLocal(1 << 20) require.NoError(t, err) defer func() { _ = c.Close() }() ctx := context.Background() + deps := []Namespace{{Table: "users", Scope: "org_1"}} // Set value - err = c.Set(ctx, "queryHash", "users", "org_1", []byte("my_data"), 10*time.Second) + err = c.Set(ctx, "queryHash", deps, []byte("my_data"), 10*time.Second) assert.NoError(t, err) c.Wait() // Ensure readable - val, _, err := c.Get(ctx, "queryHash", "users", "org_1") + val, _, err := c.Get(ctx, "queryHash", deps) assert.NoError(t, err) assert.Equal(t, []byte("my_data"), val) - // Invalidate cache for the scope - var versionKeys []string - versionKeys = append(versionKeys, "users") - versionKeys = append(versionKeys, "users.org_1") - count, err := c.InvalidateCache(ctx, versionKeys) + // Invalidate the (users, org_1) namespace. + count, err := c.Invalidate(ctx, deps) assert.NoError(t, err) - assert.Equal(t, uint64(2), count) // Returns len(keys): "users" and "users.org_1" + assert.Equal(t, uint64(1), count) - // Try fetching again - // Since the underlying cache key includes the version (which was just incremented), this should result in a cache miss - valAfter, ttlAfter, errAfter := c.Get(ctx, "queryHash", "users", "org_1") + // The folded key embeds the namespace version, which was just bumped, so this + // must now miss. + valAfter, ttlAfter, errAfter := c.Get(ctx, "queryHash", deps) assert.NoError(t, errAfter) assert.Nil(t, valAfter) assert.Zero(t, ttlAfter) diff --git a/internal/cache/version_manager.go b/internal/cache/version_manager.go index 9fbe351b..c76b2b3c 100644 --- a/internal/cache/version_manager.go +++ b/internal/cache/version_manager.go @@ -13,8 +13,7 @@ import ( // It uses a standard map because versions must NEVER be evicted under memory pressure. // TODO: this potentially could be bad/dangerous with a low amount of RAM available/high memory pressure AND a TON of tables/scopes per table... will need to work out eventually type VersionManager struct { - mu sync.RWMutex - versions map[string]uint64 + mu sync.RWMutex tableVersions map[string]uint64 //
-> table_version namespaceVersions map[string]uint64 //
.. -> namespace_version @@ -26,36 +25,12 @@ type VersionManager struct { // Optionally initialized with a NATS connection, so that each version manager on every distributed server can keep in sync – NOT IMPLEMENTED, just wired in func NewVersionManager(conn *nats.Conn) *VersionManager { return &VersionManager{ - versions: make(map[string]uint64), tableVersions: make(map[string]uint64), namespaceVersions: make(map[string]uint64), conn: conn, } } -func (vm *VersionManager) GetCacheKey(queryHash, namespace, scope string) string { - versionKey := generateVersionKey(namespace, scope) - version := vm.GetVersion(versionKey) - return fmt.Sprintf("%s.%d:%s", versionKey, version, queryHash) -} - -// Returns 0 if it has never been set. -func (vm *VersionManager) GetVersion(versionKey string) uint64 { - vm.mu.RLock() - defer vm.mu.RUnlock() - version := vm.versions[versionKey] // Go maps safely return the zero-value (0) if the key doesn't exist - - return version -} - -// IncrementVersion bumps the version string, instantly invalidating previous cache keys. -func (vm *VersionManager) IncrementVersion(versionKey string) { - // TODO: NATS Core broadcasting to keep keys in sync? Or just use L2 instead (just means round trip flights for versions w/o pipelines etc) - vm.mu.Lock() - defer vm.mu.Unlock() - vm.versions[versionKey]++ -} - type Namespace struct { Table string Scope string @@ -81,12 +56,18 @@ func (vm *VersionManager) NamespaceKey(table, scope string) string { // structured query passes one Namespace; a pipe passes several. Deps are sorted // so their order never changes the key. func (vm *VersionManager) QueryKey(sha string, deps []Namespace) string { - vm.mu.RLock() - defer vm.mu.RUnlock() segs := make([]string, len(deps)) + // Lock per dependency rather than across the whole loop: each dep's table + + // namespace versions are read together (consistent for that dep), but we don't + // hold the lock across all deps. A concurrent bump can land between deps, but the + // key is already a racy snapshot (versions can move between building it and using + // it), so cross-dep consistency buys nothing. Crucially, the sort/join run with + // no lock held. for i, d := range deps { + vm.mu.RLock() nsKey := vm.namespaceKeyLocked(d.Table, d.Scope) segs[i] = fmt.Sprintf("%s.%d", nsKey, vm.namespaceVersions[nsKey]) + vm.mu.RUnlock() } sort.Strings(segs) return sha + "|" + strings.Join(segs, "|") diff --git a/internal/cache/version_manager_test.go b/internal/cache/version_manager_test.go index 636311cb..e0b50ba4 100644 --- a/internal/cache/version_manager_test.go +++ b/internal/cache/version_manager_test.go @@ -6,41 +6,67 @@ import ( "github.com/stretchr/testify/assert" ) -func TestVersionManager_GetVersion(t *testing.T) { +func TestVersionManager_NamespaceKey(t *testing.T) { t.Parallel() vm := NewVersionManager(nil) - // Should return 0 for uninitialized keys - assert.Equal(t, uint64(0), vm.GetVersion("non_existent_table")) + // Default table version (0); a scopeless namespace renders a trailing dot. + assert.Equal(t, "users.0.", vm.NamespaceKey("users", "")) + assert.Equal(t, "users.0.org_1", vm.NamespaceKey("users", "org_1")) + + // The table version is embedded in every namespace key for that table, so a + // BumpTable is reflected across all scopes at once. + vm.BumpTable("users") + assert.Equal(t, "users.1.", vm.NamespaceKey("users", "")) + assert.Equal(t, "users.1.org_1", vm.NamespaceKey("users", "org_1")) +} + +func TestVersionManager_QueryKey(t *testing.T) { + t.Parallel() + vm := NewVersionManager(nil) + + // One dependency at default versions: sha |
.... + key := vm.QueryKey("hash123", []Namespace{{Table: "users", Scope: "org_1"}}) + assert.Equal(t, "hash123|users.0.org_1.0", key) + + // Dependency order must not change the key (segments are sorted). + deps1 := []Namespace{{Table: "a"}, {Table: "b"}} + deps2 := []Namespace{{Table: "b"}, {Table: "a"}} + assert.Equal(t, vm.QueryKey("h", deps1), vm.QueryKey("h", deps2)) } -func TestVersionManager_IncrementVersion(t *testing.T) { +func TestVersionManager_BumpTable(t *testing.T) { t.Parallel() vm := NewVersionManager(nil) - // Increment and check - vm.IncrementVersion("users") - assert.Equal(t, uint64(1), vm.GetVersion("users")) + users := []Namespace{{Table: "users", Scope: "org_1"}} + orders := []Namespace{{Table: "orders", Scope: "org_1"}} + + usersBefore := vm.QueryKey("h", users) + ordersBefore := vm.QueryKey("h", orders) - // Increment again - vm.IncrementVersion("users") - assert.Equal(t, uint64(2), vm.GetVersion("users")) + // Bumping a table changes the key for that table but leaves other tables alone. + vm.BumpTable("users") + assert.NotEqual(t, usersBefore, vm.QueryKey("h", users)) + assert.Equal(t, ordersBefore, vm.QueryKey("h", orders)) } -func TestVersionManager_GetCacheKey(t *testing.T) { +func TestVersionManager_BumpNamespace(t *testing.T) { t.Parallel() vm := NewVersionManager(nil) - // Default version (0) - key1 := vm.GetCacheKey("hash123", "users", "org_1") - assert.Equal(t, "users.org_1.0:hash123", key1) + scoped := []Namespace{{Table: "users", Scope: "org_1"}} + wholeTable := []Namespace{{Table: "users"}} + otherScope := []Namespace{{Table: "users", Scope: "org_2"}} - // Incrementing the version should change the resulting cache key - vm.IncrementVersion("users.org_1") - key2 := vm.GetCacheKey("hash123", "users", "org_1") - assert.Equal(t, "users.org_1.1:hash123", key2) + scopedBefore := vm.QueryKey("h", scoped) + wholeBefore := vm.QueryKey("h", wholeTable) + otherBefore := vm.QueryKey("h", otherScope) - // Global scope key - keyGlobal := vm.GetCacheKey("hash456", "users", "") - assert.Equal(t, "users.0:hash456", keyGlobal) + // Bumping (users, org_1) changes that scope AND the whole-table view, but leaves + // every other scope valid. + vm.BumpNamespace("users", "org_1") + assert.NotEqual(t, scopedBefore, vm.QueryKey("h", scoped)) + assert.NotEqual(t, wholeBefore, vm.QueryKey("h", wholeTable)) + assert.Equal(t, otherBefore, vm.QueryKey("h", otherScope)) } diff --git a/internal/ingest/worker.go b/internal/ingest/worker.go index 0d06e6a3..c20ef320 100644 --- a/internal/ingest/worker.go +++ b/internal/ingest/worker.go @@ -459,24 +459,28 @@ func (w *IngestWorker) insertToClickHouse(ctx context.Context, tableName string, } func (w *IngestWorker) handleSuccess(ctx context.Context, tableName string, msgs []parsedMsg) { - // Pre-allocate the set using the length of msgs (+1 for tableName itself) to prevent expensive rehashing - seenSubjects := make(map[string]struct{}, len(msgs)+1) - versionKeys := make([]string, 0, len(msgs)+1) - + // Invalidate the namespaces this batch touched: the encoded table paired with + // each distinct scope. Invalidate routes an empty scope to a whole-table bump and + // a non-empty scope to a per-scope bump (which also covers the whole-table view). + // Pre-allocate using len(msgs) to avoid rehashing. encodedTable := query.SafeEncodeNATS(tableName) - seenSubjects[encodedTable] = struct{}{} - versionKeys = append(versionKeys, encodedTable) + seenScopes := make(map[string]struct{}, len(msgs)) + namespaces := make([]cache.Namespace, 0, len(msgs)) for _, pm := range msgs { - if _, exists := seenSubjects[pm.natsSafeSubject]; !exists { - seenSubjects[pm.natsSafeSubject] = struct{}{} - versionKeys = append(versionKeys, pm.natsSafeSubject) + if _, exists := seenScopes[pm.scope]; exists { + continue } + seenScopes[pm.scope] = struct{}{} + namespaces = append(namespaces, cache.Namespace{ + Table: encodedTable, + Scope: query.SafeEncodeNATS(pm.scope), + }) } - if len(versionKeys) > 0 { + if len(namespaces) > 0 { invCtx := trace.ContextWithSpanContext(context.WithoutCancel(ctx), trace.SpanContextFromContext(ctx)) - _, err := w.cache.InvalidateCache(invCtx, versionKeys) + _, err := w.cache.Invalidate(invCtx, namespaces) if err != nil { w.logger.ErrorContext(invCtx, "failed to invalidate cache after insert - your cache is holding stale data now!", "table", tableName, "error", err) } diff --git a/internal/ingest/worker_test.go b/internal/ingest/worker_test.go index 18bfba0e..1a214b9b 100644 --- a/internal/ingest/worker_test.go +++ b/internal/ingest/worker_test.go @@ -221,21 +221,15 @@ func TestStartIngestWorker_EndToEnd(t *testing.T) { assert.Equal(t, "events", params[0]) gotMu.Unlock() - // Cache invalidation should run with the table + the (encoded) subject. + // Cache invalidation should run with the table+scope namespace. require.Eventually(t, func() bool { - keys := cache.GetKeys() - hasTable := false - hasSubject := false - for _, k := range keys { - if k == "events" { - hasTable = true - } - if k == "events.org_42" { - hasSubject = true + for _, ns := range cache.GetNamespaces() { + if ns.Table == "events" && ns.Scope == "org_42" { + return true } } - return hasTable && hasSubject - }, 4*time.Second, 25*time.Millisecond, "cache must be invalidated for table + subject") + return false + }, 4*time.Second, 25*time.Millisecond, "cache must be invalidated for the table+scope namespace") // Shutdown (cancel + drain) is handled by t.Cleanup above. } @@ -455,91 +449,86 @@ func TestInsertToClickHouse_NetworkError(t *testing.T) { func TestHandleSuccess(t *testing.T) { t.Parallel() - // natsSafeSubject is what the worker derives by stripping "ingest." from - // the NATS subject the producer publishes to (see internal/api/ingest.go). - // In production that is always "[.]", - // so e.g. table "events" + scope "org_1" yields natsSafeSubject "events.org_1". - // handleSuccess invalidates the encoded table key + each unique scoped key, - // deduping when the scopeless subject already equals the table key. + // handleSuccess invalidates one namespace per distinct scope in the batch: the + // encoded table paired with the encoded scope (envelope.scope). Invalidate turns + // an empty scope into a whole-table bump and a non-empty scope into a per-scope + // bump, so the worker only needs to emit the table+scope pairs it saw. tests := []struct { - name string - table string - natsSafeSubjects []string // realistic shape: "[.]" - cacheErr error // applied via MockCache.InvErr before the call - msgDoubleAckErr error // applied to every MockJetStreamMsg - wantCacheKeys []string // expected (ElementsMatch) keys in cache.GetKeys() + name string + table string + scopes []string // raw per-message scopes (envelope.scope) + cacheErr error // applied via MockCache.InvErr before the call + msgDoubleAckErr error // applied to every MockJetStreamMsg + wantNamespaces []cache.Namespace }{ { - name: "invalidates table + each unique scope", - table: "events", - natsSafeSubjects: []string{"events.org_1", "events.org_2"}, - wantCacheKeys: []string{"events", "events.org_1", "events.org_2"}, + name: "invalidates each unique scope", + table: "events", + scopes: []string{"org_1", "org_2"}, + wantNamespaces: []cache.Namespace{{Table: "events", Scope: "org_1"}, {Table: "events", Scope: "org_2"}}, }, { - // 3 msgs, 2 distinct scoped subjects → table + 2 scoped = 3 keys. - name: "deduplicates repeated scope keys", - table: "events", - natsSafeSubjects: []string{"events.org_1", "events.org_1", "events.org_2"}, - wantCacheKeys: []string{"events", "events.org_1", "events.org_2"}, + // 3 msgs, 2 distinct scopes → 2 namespaces. + name: "deduplicates repeated scopes", + table: "events", + scopes: []string{"org_1", "org_1", "org_2"}, + wantNamespaces: []cache.Namespace{{Table: "events", Scope: "org_1"}, {Table: "events", Scope: "org_2"}}, }, { - // Scopeless: producer publishes to "ingest.
", so - // natsSafeSubject equals the encoded table → one key. - name: "scopeless subject dedups against table key", - table: "events", - natsSafeSubjects: []string{"events"}, - wantCacheKeys: []string{"events"}, + // Scopeless message → empty-scope namespace (a whole-table bump). + name: "scopeless message yields whole-table namespace", + table: "events", + scopes: []string{""}, + wantNamespaces: []cache.Namespace{{Table: "events", Scope: ""}}, }, { - // Table name containing a '.' is percent-encoded on both sides of - // the version key — keys must use the encoded form for cache hits - // to line up with the reader (internal/api/structured_query.go). - name: "table with special chars is percent-encoded", - table: "events.staging", - natsSafeSubjects: []string{"events%2Estaging.org_1"}, - wantCacheKeys: []string{"events%2Estaging", "events%2Estaging.org_1"}, + // Table and scope are percent-encoded so keys line up with the reader + // (internal/api/structured_query.go), which encodes the table too. + name: "table and scope are percent-encoded", + table: "events.staging", + scopes: []string{"org.1"}, + wantNamespaces: []cache.Namespace{{Table: "events%2Estaging", Scope: "org%2E1"}}, }, { // Cache failure must not prevent ack — failure is logged, non-fatal. - name: "cache error still acks", - table: "events", - natsSafeSubjects: []string{"events.org_1"}, - cacheErr: errors.New("cache backend down"), - wantCacheKeys: []string{"events", "events.org_1"}, + name: "cache error still acks", + table: "events", + scopes: []string{"org_1"}, + cacheErr: errors.New("cache backend down"), + wantNamespaces: []cache.Namespace{{Table: "events", Scope: "org_1"}}, }, { // DoubleAck error is logged but DoubleAcked flag still flips. - name: "double ack error does not panic", - table: "events", - natsSafeSubjects: []string{"events.org_1"}, - msgDoubleAckErr: errors.New("server unavailable"), - wantCacheKeys: []string{"events", "events.org_1"}, + name: "double ack error does not panic", + table: "events", + scopes: []string{"org_1"}, + msgDoubleAckErr: errors.New("server unavailable"), + wantNamespaces: []cache.Namespace{{Table: "events", Scope: "org_1"}}, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() - w, _, cache, wait := newTestWorker(&testutil.MockRoundTripper{}) - cache.InvErr = tt.cacheErr - - msgs := make([]*testutil.MockJetStreamMsg, len(tt.natsSafeSubjects)) - parsed := make([]parsedMsg, len(tt.natsSafeSubjects)) - for i, sub := range tt.natsSafeSubjects { - // MsgSubject intentionally left blank — handleSuccess reads - // natsSafeSubject off parsedMsg, not the NATS msg itself. + w, _, mc, wait := newTestWorker(&testutil.MockRoundTripper{}) + mc.InvErr = tt.cacheErr + + msgs := make([]*testutil.MockJetStreamMsg, len(tt.scopes)) + parsed := make([]parsedMsg, len(tt.scopes)) + for i, scope := range tt.scopes { + // handleSuccess reads scope off parsedMsg, not the NATS msg itself. msgs[i] = &testutil.MockJetStreamMsg{DoubleAckErr: tt.msgDoubleAckErr} - parsed[i] = parsedMsg{natsMsg: msgs[i], natsSafeSubject: sub} + parsed[i] = parsedMsg{natsMsg: msgs[i], scope: scope} } w.handleSuccess(context.Background(), tt.table, parsed) wait() - assert.ElementsMatch(t, tt.wantCacheKeys, cache.GetKeys()) + assert.ElementsMatch(t, tt.wantNamespaces, mc.GetNamespaces()) for i, m := range msgs { assert.True(t, m.DoubleAcked.Load(), - "msg %d (natsSafeSubject=%q) must be DoubleAcked", i, tt.natsSafeSubjects[i]) + "msg %d (scope=%q) must be DoubleAcked", i, tt.scopes[i]) } }) } @@ -711,7 +700,7 @@ func TestFlushTable_HappyPath(t *testing.T) { return &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewBufferString("OK"))}, nil }, } - w, _, cache, wait := newTestWorker(rt) + w, _, mc, wait := newTestWorker(rt) m1 := newIngestMsg(t, "events", "org_1", map[string]any{"id": 1}) m2 := newIngestMsg(t, "events", "org_1", map[string]any{"id": 2}) @@ -726,16 +715,15 @@ func TestFlushTable_HappyPath(t *testing.T) { assert.True(t, m1.DoubleAcked.Load()) assert.True(t, m2.DoubleAcked.Load()) - // Cache invalidation: table-level + the shared scoped key. + // Cache invalidation: one namespace for the single shared scope. assert.ElementsMatch(t, - []string{"events", "events.org_1"}, - cache.GetKeys(), + []cache.Namespace{{Table: "events", Scope: "org_1"}}, + mc.GetNamespaces(), ) } // TestFlushTable_MultiScope covers a single table with multiple scopes: one bulk -// HTTP insert, but each unique scope still gets its own cache version key bumped -// alongside the global table key. +// HTTP insert, but each unique scope still gets its own namespace invalidated. func TestFlushTable_MultiScope(t *testing.T) { t.Parallel() rt := &testutil.MockRoundTripper{ @@ -743,7 +731,7 @@ func TestFlushTable_MultiScope(t *testing.T) { return &http.Response{StatusCode: 200, Body: io.NopCloser(bytes.NewBufferString("OK"))}, nil }, } - w, _, cache, wait := newTestWorker(rt) + w, _, mc, wait := newTestWorker(rt) msgs := parseAll(t, w, newIngestMsg(t, "events", "org_1", map[string]any{"id": 1}), @@ -757,10 +745,13 @@ func TestFlushTable_MultiScope(t *testing.T) { // One bulk HTTP request despite multiple scopes. assert.Equal(t, int32(1), rt.Hits()) - // Cache: table key + each unique scope key, deduped. + // Cache: one namespace per unique scope, deduped. assert.ElementsMatch(t, - []string{"events", "events.org_1", "events.org_2"}, - cache.GetKeys(), + []cache.Namespace{ + {Table: "events", Scope: "org_1"}, + {Table: "events", Scope: "org_2"}, + }, + mc.GetNamespaces(), ) } diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index 9f039620..eb7ab654 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -114,25 +114,25 @@ func (m *MockDeduplicator) Close() error { return nil } // ── Mock Cache ─────────────────────────────────────────────────── -// MockCache implements cache.Cache and records invalidation keys for testing. +// MockCache implements cache.Cache and records invalidated namespaces for testing. type MockCache struct { - cache.Cache // Embed to satisfy remaining interface methods silently - InvKeys []string - InvErr error - mu sync.Mutex + cache.Cache // Embed to satisfy remaining interface methods silently + InvNamespaces []cache.Namespace + InvErr error + mu sync.Mutex } -func (m *MockCache) InvalidateCache(ctx context.Context, keys []string) (uint64, error) { +func (m *MockCache) Invalidate(_ context.Context, namespaces []cache.Namespace) (uint64, error) { m.mu.Lock() defer m.mu.Unlock() - m.InvKeys = append(m.InvKeys, keys...) - return uint64(len(keys)), m.InvErr + m.InvNamespaces = append(m.InvNamespaces, namespaces...) + return uint64(len(namespaces)), m.InvErr } -func (m *MockCache) GetKeys() []string { +func (m *MockCache) GetNamespaces() []cache.Namespace { m.mu.Lock() defer m.mu.Unlock() - return append([]string(nil), m.InvKeys...) // Return copy + return append([]cache.Namespace(nil), m.InvNamespaces...) // Return copy } // ── Mock jetstream.Msg ─────────────────────────────────────────── From 13427bc913ac0ef180b0b5f7f4493b0762c93d1f Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 10 Jun 2026 11:28:24 -0400 Subject: [PATCH 04/21] make fix --- internal/api/structured_query.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index 9188ccd6..776567c3 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -134,7 +134,7 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) if data, _, err := h.Cache.Get(r.Context(), cacheKey, deps); err == nil && data != nil { w.Header().Set("Content-Type", "application/json") w.Header().Set("X-Cache", "HIT") - _, _ = w.Write(data) //nolint:gosec // G705 XSS only JSON + _, _ = w.Write(data) return } } From 1eeb929e30622912ccc4e456243d5a0bf6b8c638 Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 10 Jun 2026 14:23:56 -0400 Subject: [PATCH 05/21] perf(cache): skip redundant per-scope bumps on whole-table invalidation --- internal/cache/local.go | 22 ++++++++++++++++++++-- internal/cache/local_test.go | 30 ++++++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 2 deletions(-) diff --git a/internal/cache/local.go b/internal/cache/local.go index e5a5b3bf..a1db3a5f 100644 --- a/internal/cache/local.go +++ b/internal/cache/local.go @@ -57,14 +57,32 @@ func (l *LocalCache) Set(_ context.Context, sha string, deps []Namespace, value // cached query that depends on it. An empty Scope bumps the whole table (every // scope at once); a non-empty Scope bumps just that scope plus the whole-table // view. Returns the number of namespaces processed. +// +// A whole-table bump subsumes every per-scope bump for that table — the table +// version is embedded in every namespace key, so bumping it already invalidates +// all scopes. So we bump tables first, then bump only the scopes whose table +// wasn't wholesale-bumped, skipping the redundant work (and the wasted +// namespaceVersions entries it would otherwise create). func (l *LocalCache) Invalidate(_ context.Context, namespaces []Namespace) (uint64, error) { + bumpedTables := make(map[string]struct{}) for _, ns := range namespaces { if ns.Scope == "" { + if _, done := bumpedTables[ns.Table]; done { + continue + } l.versionManager.BumpTable(ns.Table) - } else { - l.versionManager.BumpNamespace(ns.Table, ns.Scope) + bumpedTables[ns.Table] = struct{}{} } } + for _, ns := range namespaces { + if ns.Scope == "" { + continue + } + if _, whole := bumpedTables[ns.Table]; whole { + continue // table bump already covers this scope + } + l.versionManager.BumpNamespace(ns.Table, ns.Scope) + } return uint64(len(namespaces)), nil } diff --git a/internal/cache/local_test.go b/internal/cache/local_test.go index e050874b..943e73b9 100644 --- a/internal/cache/local_test.go +++ b/internal/cache/local_test.go @@ -133,3 +133,33 @@ func TestLocalCache_Invalidate(t *testing.T) { assert.Nil(t, valAfter) assert.Zero(t, ttlAfter) } + +// A batch that bumps a whole table must invalidate that table's scoped entries +// via the table bump alone — the redundant per-scope bumps are skipped, but the +// scoped entry must still miss. +func TestLocalCache_Invalidate_WholeTableSubsumesScopes(t *testing.T) { + t.Parallel() + c, err := NewLocal(1 << 20) + require.NoError(t, err) + defer func() { _ = c.Close() }() + + ctx := context.Background() + scoped := []Namespace{{Table: "events", Scope: "org_1"}} + + require.NoError(t, c.Set(ctx, "q", scoped, []byte("v1"), 10*time.Second)) + c.Wait() + val, _, err := c.Get(ctx, "q", scoped) + require.NoError(t, err) + require.Equal(t, []byte("v1"), val) + + // Mixed batch: whole-table entry + a scoped entry for the same table. + _, err = c.Invalidate(ctx, []Namespace{ + {Table: "events", Scope: ""}, + {Table: "events", Scope: "org_1"}, + }) + require.NoError(t, err) + + after, _, err := c.Get(ctx, "q", scoped) + assert.NoError(t, err) + assert.Nil(t, after, "whole-table bump must invalidate the scoped entry") +} From 377636c34958fd9aa8eef136bca0e9b5f72edad0 Mon Sep 17 00:00:00 2001 From: taitelee Date: Wed, 10 Jun 2026 14:52:36 -0400 Subject: [PATCH 06/21] docs(changelog): add Unreleased entry for the cache cutover; encode query scope --- CHANGELOG.md | 1 + internal/api/structured_query.go | 5 ++++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 25cd275b..0ad90b41 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- **Read-cache invalidation cut over to namespace-versioned keys, replacing the flat per-namespace version map** (`internal/cache/cache.go`, `internal/cache/local.go`, `internal/cache/version_manager.go`, `internal/api/structured_query.go`, `internal/api/pipes.go`, `internal/ingest/worker.go`, `internal/testutil/mocks.go`, plus tests in `internal/cache/{local,version_manager}_test.go` and `internal/ingest/worker_test.go`): the `Cache` interface is now keyed by what a result *depends on* rather than one opaque string. `Get`/`Set` take a `sha` (hash of SQL+params) plus the `[]Namespace` (`{table, scope}` pairs) the result reads — one for a structured query, several for a (future) pipe — and `QueryKey` folds the table version and per-namespace version of every dependency into the key, so a write to any dependency misses it. Invalidation is two-level: `BumpTable(table)` advances a table version that is embedded in every namespace key, invalidating all scopes of that table in O(1); `BumpNamespace(table, scope)` advances one scope plus the whole-table view, leaving other scopes' cached entries valid. The old `GetCacheKey`/`GetVersion`/`IncrementVersion`/`InvalidateCache` flat-`versions` path is removed, and the structured-query handler and ingest worker are wired onto the new path via `Cache.Invalidate([]Namespace)` (empty scope → `BumpTable`, scoped → `BumpNamespace`); a whole-table bump subsumes any same-batch per-scope bump, so `Invalidate` skips that redundant work. Pipes currently pass no dependencies (TTL-only invalidation) until they can report the tables they read ([#178](https://github.com/Wave-RF/WaveHouse/issues/178)). - **Docs "Last updated" footer shows each page's real last-edit date, and robots.txt declares open AI content signals** (`.github/workflows/ci.yml`, `docs/public/robots.txt`): Starlight's `lastUpdated` reads git history, but CI checked out with the default shallow (depth-1) clone, so every page silently fell back to the build time — the live site showed the same "Last updated" on every page regardless of when it actually changed. CI now checks out with `fetch-depth: 0`. Separately, `robots.txt` declares open Content-Signals (`search=yes, ai-input=yes, ai-train=yes`): WaveHouse is open source, so agents are welcome to read, index, answer from, and train on these docs. (A markdown-twin `Link` header — the other half of the agent-discoverability pass — is being upstreamed to [`cloudflare-md-router`](https://github.com/Wave-RF/cloudflare-md-router) instead, where the twin concept lives.) - **Mermaid diagrams are click-to-zoom, with a top-level house style to author them vertically** (`docs/src/components/MermaidZoom.astro` (new), `docs/src/components/Footer.astro`, `docs/src/styles/global.css`, `AGENTS.md`): wide `flowchart LR` diagrams shrink to ~36% in the content column and are hard to read inline. Clicking any diagram now opens it in a lightbox — small diagrams scale up to fill the viewport (capped 2.5×), wide ones show at natural size with pan-scroll for full detail; Esc / backdrop / button closes, keyboard-focusable with Enter. View-transition-safe (one delegated `document` listener guarded by a window flag, re-enhanced on `astro:page-load`). The clone is re-id'd so its id-scoped inline `