From 5eab35409baeb6773e0734f3dd578f3947ef0e42 Mon Sep 17 00:00:00 2001 From: Lakhan Samani Date: Tue, 29 Sep 2026 12:00:01 +0530 Subject: [PATCH 1/4] fix(storage): apply all audit log filters on cassandra and couchbase MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ListAuditLogRequest advertises resource_type, resource_id, from_timestamp and to_timestamp. Cassandra/ScyllaDB and Couchbase built a WHERE clause from action and actor_id only, silently returning unfiltered rows — an auditor narrowing to one resource or one time window read the result as authoritative. - cassandra: index resource_type and resource_id; ALLOW FILTERING only when a created_at bound or a second restriction is present, so indexed-equality queries keep their index-served plan - cassandra: probe every indexed column for readiness, not just actor_id — Scylla builds indexes concurrently, so the last one created is not the last one ready - couchbase: apply all four as named N1QL parameters Verified by running the new subtests against both backends before the fix (all three fail) and after (pass). --- internal/storage/db/cassandradb/audit_log.go | 71 ++++++++++++++++---- internal/storage/db/cassandradb/provider.go | 45 +++++++++---- internal/storage/db/couchbase/audit_log.go | 36 ++++++++-- internal/storage/provider_test.go | 69 +++++++++++++++++++ 4 files changed, 188 insertions(+), 33 deletions(-) diff --git a/internal/storage/db/cassandradb/audit_log.go b/internal/storage/db/cassandradb/audit_log.go index 664b03b50..f209c26f6 100644 --- a/internal/storage/db/cassandradb/audit_log.go +++ b/internal/storage/db/cassandradb/audit_log.go @@ -3,6 +3,7 @@ package cassandradb import ( "context" "fmt" + "strings" "time" "github.com/gocql/gocql" @@ -44,24 +45,64 @@ func (p *provider) ListAuditLogs(ctx context.Context, pagination *model.Paginati queryBase := fmt.Sprintf("SELECT id, actor_id, actor_type, actor_email, action, resource_type, resource_id, ip_address, user_agent, metadata, created_at FROM %s", KeySpace+"."+schemas.Collections.AuditLog) countBase := fmt.Sprintf("SELECT COUNT(*) FROM %s", KeySpace+"."+schemas.Collections.AuditLog) - whereClause := "" + // Every filter column below is backed by a secondary index (see + // provider.go), so equality restrictions need no ALLOW FILTERING — Scylla + // builds those indexes as materialized views and serves them directly. + // + // The created_at BOUNDS are different: a range on a non-primary-key column + // cannot be served by an index, so it forces ALLOW FILTERING and a scan. + // Only the timestamp bounds set that flag; an indexed-equality-only query + // keeps its existing index-served plan. + // + // ponytail: ALLOW FILTERING for the timestamp range. This is an admin-only, + // rarely-run query and the table is small relative to user data. The upgrade + // path when it stops being cheap is a materialized view keyed on a coarse + // time bucket, not a bigger scan. + clauses := []string{} filterValues := []interface{}{} + needsAllowFiltering := false + + addEq := func(col string, v interface{}) { + clauses = append(clauses, col+"=?") + filterValues = append(filterValues, v) + } if action, ok := filter["action"]; ok && action != "" { - whereClause += " WHERE action=?" - filterValues = append(filterValues, action) + addEq("action", action) } if actorID, ok := filter["actor_id"]; ok && actorID != "" { - if whereClause == "" { - whereClause += " WHERE actor_id=?" - } else { - whereClause += " AND actor_id=?" - } - filterValues = append(filterValues, actorID) + addEq("actor_id", actorID) + } + if resourceType, ok := filter["resource_type"]; ok && resourceType != "" { + addEq("resource_type", resourceType) + } + if resourceID, ok := filter["resource_id"]; ok && resourceID != "" { + addEq("resource_id", resourceID) + } + if fromTimestamp, ok := filter["from_timestamp"]; ok { + clauses = append(clauses, "created_at>=?") + filterValues = append(filterValues, fromTimestamp) + needsAllowFiltering = true + } + if toTimestamp, ok := filter["to_timestamp"]; ok { + clauses = append(clauses, "created_at<=?") + filterValues = append(filterValues, toTimestamp) + needsAllowFiltering = true + } + + whereClause := "" + if len(clauses) > 0 { + whereClause = " WHERE " + strings.Join(clauses, " AND ") + } + // More than one restriction on non-primary-key columns cannot be served by a + // single index either, so Cassandra/Scylla requires the scan hint there too. + if len(clauses) > 1 { + needsAllowFiltering = true + } + if needsAllowFiltering { + whereClause += " ALLOW FILTERING" } - // Count total — equality on indexed columns (action, actor_id) must not - // use ALLOW FILTERING; Scylla builds secondary indexes as materialized views. countQuery := countBase + whereClause err := p.db.Query(countQuery, filterValues...).Consistency(gocql.One).Scan(&paginationClone.Total) if err != nil { @@ -69,7 +110,13 @@ func (p *provider) ListAuditLogs(ctx context.Context, pagination *model.Paginati } // Fetch with pagination - query := queryBase + whereClause + fmt.Sprintf(" LIMIT %d", pagination.Limit+pagination.Offset) + // CQL grammar: LIMIT precedes ALLOW FILTERING, so the hint cannot simply be + // carried along on whereClause here. + query := queryBase + strings.TrimSuffix(whereClause, " ALLOW FILTERING") + + fmt.Sprintf(" LIMIT %d", pagination.Limit+pagination.Offset) + if needsAllowFiltering { + query += " ALLOW FILTERING" + } scanner := p.db.Query(query, filterValues...).Iter().Scanner() counter := int64(0) for scanner.Next() { diff --git a/internal/storage/db/cassandradb/provider.go b/internal/storage/db/cassandradb/provider.go index bcbfbccbb..f728ac3b2 100644 --- a/internal/storage/db/cassandradb/provider.go +++ b/internal/storage/db/cassandradb/provider.go @@ -420,9 +420,20 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } + auditLogResourceTypeIndex := fmt.Sprintf("CREATE INDEX IF NOT EXISTS authorizer_audit_log_resource_type ON %s.%s (resource_type)", KeySpace, schemas.Collections.AuditLog) + err = session.Query(auditLogResourceTypeIndex).Exec() + if err != nil { + return nil, err + } + auditLogResourceIDIndex := fmt.Sprintf("CREATE INDEX IF NOT EXISTS authorizer_audit_log_resource_id ON %s.%s (resource_id)", KeySpace, schemas.Collections.AuditLog) + err = session.Query(auditLogResourceIDIndex).Exec() + if err != nil { + return nil, err + } // ScyllaDB builds secondary indexes asynchronously. Poll with a probe query // that requires the actor_id index until it succeeds instead of a fixed sleep. - waitForCassandraIndexes(session, KeySpace, schemas.Collections.AuditLog, 30*time.Second) + waitForCassandraIndexes(session, KeySpace, schemas.Collections.AuditLog, + []string{"actor_id", "action", "resource_type", "resource_id"}, 30*time.Second) // Client table clientCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, client_id text, kind text, name text, description text, client_secret text, allowed_scopes text, redirect_uris text, grant_types text, token_endpoint_auth_method text, is_active boolean, org_id text, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.Client) @@ -653,20 +664,26 @@ func waitForCassandraSecondaryIndex(session *cansandraDriver.Session, keyspace, // waitForCassandraIndexes polls a probe query that requires the actor_id secondary // index until it succeeds or the timeout is reached. ScyllaDB builds secondary // indexes asynchronously; queries on indexed columns fail until the index is ready. -func waitForCassandraIndexes(session *cansandraDriver.Session, keyspace, table string, timeout time.Duration) { - probe := fmt.Sprintf("SELECT id FROM %s.%s WHERE actor_id='' LIMIT 1", keyspace, table) +// columns are probed independently: Scylla builds secondary indexes +// concurrently, so the last one created is not necessarily the last one ready, +// and waiting on a single column would let a query against another index run +// before that index exists. +func waitForCassandraIndexes(session *cansandraDriver.Session, keyspace, table string, columns []string, timeout time.Duration) { deadline := time.Now().Add(timeout) - delay := 500 * time.Millisecond - for { - if err := session.Query(probe).Exec(); err == nil { - return - } - if time.Now().After(deadline) { - return - } - time.Sleep(delay) - if delay < 3*time.Second { - delay += 500 * time.Millisecond + for _, col := range columns { + probe := fmt.Sprintf("SELECT id FROM %s.%s WHERE %s='' LIMIT 1", keyspace, table, col) + delay := 500 * time.Millisecond + for { + if err := session.Query(probe).Exec(); err == nil { + break + } + if time.Now().After(deadline) { + return + } + time.Sleep(delay) + if delay < 3*time.Second { + delay += 500 * time.Millisecond + } } } } diff --git a/internal/storage/db/couchbase/audit_log.go b/internal/storage/db/couchbase/audit_log.go index 7bd8e4682..52bf50974 100644 --- a/internal/storage/db/couchbase/audit_log.go +++ b/internal/storage/db/couchbase/audit_log.go @@ -3,6 +3,7 @@ package couchbase import ( "context" "fmt" + "strings" "time" "github.com/couchbase/gocb/v2" @@ -43,19 +44,40 @@ func (p *provider) ListAuditLogs(ctx context.Context, pagination *model.Paginati params["offset"] = paginationClone.Offset params["limit"] = paginationClone.Limit - whereClause := "" + // Every filter the ListAuditLogRequest API accepts is applied here. Leaving + // any of them out silently returns unfiltered rows: the caller cannot tell a + // filter was dropped, so an auditor narrowing to one resource or one time + // window would read the result as authoritative. + clauses := []string{} if action, ok := filter["action"]; ok && action != "" { - whereClause += " WHERE action=$action" + clauses = append(clauses, "action=$action") params["action"] = action } if actorID, ok := filter["actor_id"]; ok && actorID != "" { - if whereClause == "" { - whereClause += " WHERE actor_id=$actorID" - } else { - whereClause += " AND actor_id=$actorID" - } + clauses = append(clauses, "actor_id=$actorID") params["actorID"] = actorID } + if resourceType, ok := filter["resource_type"]; ok && resourceType != "" { + clauses = append(clauses, "resource_type=$resourceType") + params["resourceType"] = resourceType + } + if resourceID, ok := filter["resource_id"]; ok && resourceID != "" { + clauses = append(clauses, "resource_id=$resourceID") + params["resourceID"] = resourceID + } + if fromTimestamp, ok := filter["from_timestamp"]; ok { + clauses = append(clauses, "created_at>=$fromTimestamp") + params["fromTimestamp"] = fromTimestamp + } + if toTimestamp, ok := filter["to_timestamp"]; ok { + clauses = append(clauses, "created_at<=$toTimestamp") + params["toTimestamp"] = toTimestamp + } + + whereClause := "" + if len(clauses) > 0 { + whereClause = " WHERE " + strings.Join(clauses, " AND ") + } // Count with filters applied countQuery := fmt.Sprintf("SELECT COUNT(*) as count FROM %s.%s%s", diff --git a/internal/storage/provider_test.go b/internal/storage/provider_test.go index 92bc5a849..b20a59442 100644 --- a/internal/storage/provider_test.go +++ b/internal/storage/provider_test.go @@ -1343,6 +1343,75 @@ func testAuditLogOperations(t *testing.T, ctx context.Context, provider Provider assert.Equal(t, actorID, logs[0].ActorID) }) + t.Run("filter by resource_id", func(t *testing.T) { + resourceID := uuid.New().String() + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: uuid.New().String(), + ActorType: constants.AuditActorTypeAdmin, + Action: constants.AuditAdminFgaTuplesWrittenEvent, + ResourceType: constants.AuditResourceTypeFgaTuple, + ResourceID: resourceID, + })) + + pagination := &model.Pagination{Limit: 10, Offset: 0} + logs, _, err := provider.ListAuditLogs(ctx, pagination, map[string]interface{}{ + "resource_id": resourceID, + }) + require.NoError(t, err) + require.Len(t, logs, 1) + assert.Equal(t, resourceID, logs[0].ResourceID) + }) + + t.Run("filter by resource_type", func(t *testing.T) { + resourceType := "provider_test_rt_" + uuid.New().String()[:8] + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: uuid.New().String(), + ActorType: constants.AuditActorTypeAdmin, + Action: constants.AuditAdminFgaTuplesWrittenEvent, + ResourceType: resourceType, + ResourceID: uuid.New().String(), + })) + + pagination := &model.Pagination{Limit: 10, Offset: 0} + logs, _, err := provider.ListAuditLogs(ctx, pagination, map[string]interface{}{ + "resource_type": resourceType, + }) + require.NoError(t, err) + require.Len(t, logs, 1) + assert.Equal(t, resourceType, logs[0].ResourceType) + }) + + t.Run("filter by timestamp range", func(t *testing.T) { + // Two rows sharing an action, far apart in time. The range must select + // exactly the recent one — a backend that ignores the bounds returns both. + action := "provider_test_ts_" + uuid.New().String()[:8] + now := time.Now().Unix() + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: uuid.New().String(), + ActorType: constants.AuditActorTypeAdmin, + Action: action, + ResourceType: constants.AuditResourceTypeUser, + CreatedAt: now - int64(90*24*time.Hour/time.Second), + })) + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: uuid.New().String(), + ActorType: constants.AuditActorTypeAdmin, + Action: action, + ResourceType: constants.AuditResourceTypeUser, + CreatedAt: now, + })) + + pagination := &model.Pagination{Limit: 10, Offset: 0} + logs, _, err := provider.ListAuditLogs(ctx, pagination, map[string]interface{}{ + "action": action, + "from_timestamp": now - int64(time.Hour/time.Second), + "to_timestamp": now + int64(time.Hour/time.Second), + }) + require.NoError(t, err) + require.Len(t, logs, 1, "timestamp range must exclude the 90-day-old row") + assert.Equal(t, now, logs[0].CreatedAt) + }) + t.Run("list does not mutate caller pagination pointer", func(t *testing.T) { pagination := &model.Pagination{Limit: 10, Offset: 0} _, returnedPag, err := provider.ListAuditLogs(ctx, pagination, map[string]interface{}{}) From 99658aea37c51a79906f1991d4431e902e0175ca Mon Sep 17 00:00:00 2001 From: Lakhan Samani Date: Tue, 29 Sep 2026 12:15:10 +0530 Subject: [PATCH 2/4] test(storage): cover combined indexed-equality audit filters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two indexed equalities with no ALLOW FILTERING are rejected outright by Scylla: InvalidRequest: Cannot execute this query as it might involve data filtering [...] use ALLOW FILTERING That is the shape the pre-fix builder emitted, so _audit_logs(action:, actor_id:) already errored on Scylla before this branch — a second latent bug, invisible because no test combined two filters. Single-column equality is unaffected and still index-served. --- internal/storage/provider_test.go | 38 +++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/internal/storage/provider_test.go b/internal/storage/provider_test.go index b20a59442..b9c98628e 100644 --- a/internal/storage/provider_test.go +++ b/internal/storage/provider_test.go @@ -1381,6 +1381,44 @@ func testAuditLogOperations(t *testing.T, ctx context.Context, provider Provider assert.Equal(t, resourceType, logs[0].ResourceType) }) + t.Run("filter by action and actor_id and resource_id", func(t *testing.T) { + // Several indexed equalities at once. Cassandra/Scylla cannot serve + // more than one secondary index in a single query, so this is the case + // that decides whether the CQL builder needs a scan hint. + action := "provider_test_multi_" + uuid.New().String()[:8] + actorID := uuid.New().String() + resourceID := uuid.New().String() + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: actorID, + ActorType: constants.AuditActorTypeAdmin, + Action: action, + ResourceType: constants.AuditResourceTypeUser, + ResourceID: resourceID, + })) + // Same action, different actor and resource — must be excluded. + require.NoError(t, provider.AddAuditLog(ctx, &schemas.AuditLog{ + ActorID: uuid.New().String(), + ActorType: constants.AuditActorTypeAdmin, + Action: action, + ResourceType: constants.AuditResourceTypeUser, + ResourceID: uuid.New().String(), + })) + + pagination := &model.Pagination{Limit: 10, Offset: 0} + logs, pag, err := provider.ListAuditLogs(ctx, pagination, map[string]interface{}{ + "action": action, + "actor_id": actorID, + "resource_id": resourceID, + }) + require.NoError(t, err) + require.Len(t, logs, 1) + assert.Equal(t, actorID, logs[0].ActorID) + assert.Equal(t, resourceID, logs[0].ResourceID) + // The count query takes a separate code path from the fetch query. + require.NotNil(t, pag) + assert.Equal(t, int64(1), pag.Total) + }) + t.Run("filter by timestamp range", func(t *testing.T) { // Two rows sharing an action, far apart in time. The range must select // exactly the recent one — a backend that ignores the bounds returns both. From 5552e42a823e20d865efbddf444b86ac11474a22 Mon Sep 17 00:00:00 2001 From: Lakhan Samani Date: Tue, 29 Sep 2026 13:04:18 +0530 Subject: [PATCH 3/4] fix(storage): address review on audit filter parity MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - cassandra: check scanner.Err() in ListAuditLogs. A scan dying part-way ends Next() normally, so the call returned a truncated audit page with a nil error while Total reported the real count. Newly reachable now that filters can produce ALLOW FILTERING scans. DeleteAuditLogsBefore in the same file already did this. - cassandra: split the index-wait budget per column. One shared deadline let a slow first index consume it all, after which the remaining columns were never probed — turning a race on one column into a race on every other. - cassandra: extract pollUntilQuerySucceeds so the backoff cannot drift between the single- and multi-column waits. - couchbase: add GSIs for resource_type and resource_id. Without them the new predicates fall back to the primary-index scan while cassandra serves them from an index — the cross-provider parity bug AGENTS.md forbids. Verified via EXPLAIN that the planner picks AuditLogResourceIdIndex with no PrimaryScan. - correct two comments that still described a single actor_id probe, and one that sized ALLOW FILTERING for the timestamp range only when it also covers any two equality filters. --- internal/storage/db/cassandradb/audit_log.go | 23 ++++++-- internal/storage/db/cassandradb/provider.go | 62 ++++++++++++-------- internal/storage/db/couchbase/provider.go | 9 ++- 3 files changed, 64 insertions(+), 30 deletions(-) diff --git a/internal/storage/db/cassandradb/audit_log.go b/internal/storage/db/cassandradb/audit_log.go index f209c26f6..531c7e645 100644 --- a/internal/storage/db/cassandradb/audit_log.go +++ b/internal/storage/db/cassandradb/audit_log.go @@ -54,10 +54,14 @@ func (p *provider) ListAuditLogs(ctx context.Context, pagination *model.Paginati // Only the timestamp bounds set that flag; an indexed-equality-only query // keeps its existing index-served plan. // - // ponytail: ALLOW FILTERING for the timestamp range. This is an admin-only, - // rarely-run query and the table is small relative to user data. The upgrade - // path when it stops being cheap is a materialized view keyed on a coarse - // time bucket, not a bigger scan. + // ponytail: ALLOW FILTERING covers TWO cases, not just the rare one. The + // timestamp range is genuinely rare. Two or more equality filters is NOT — + // "this actor, this action" is an ordinary admin search, and it now scans + // where it previously errored outright. That is still the right trade here + // (admin-only endpoint, paginated, and an erroring filter combination is + // worse than a slow one), but it is a scan, not a free lunch. The upgrade + // path when it stops being cheap is a composite materialized view over the + // filter combinations that actually get used — not a bigger scan. clauses := []string{} filterValues := []interface{}{} needsAllowFiltering := false @@ -134,6 +138,17 @@ func (p *provider) ListAuditLogs(ctx context.Context, pagination *model.Paginati } counter++ } + // A scan that dies part-way — read timeout, coordinator failure — ends + // Next() normally, so without this the call returns a TRUNCATED page with a + // nil error while paginationClone.Total (a separate query) reports the real + // count. An admin would see an incomplete audit trail with no signal it was + // cut short, which is the exact failure this table exists to prevent. Newly + // reachable here: the filters above can now produce ALLOW FILTERING scans, + // where a partial read is far likelier than on an index-served equality. + // DeleteAuditLogsBefore below already does this. + if err := scanner.Err(); err != nil { + return nil, nil, err + } return auditLogs, &paginationClone, nil } diff --git a/internal/storage/db/cassandradb/provider.go b/internal/storage/db/cassandradb/provider.go index f728ac3b2..bf82b985e 100644 --- a/internal/storage/db/cassandradb/provider.go +++ b/internal/storage/db/cassandradb/provider.go @@ -430,10 +430,10 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously. Poll with a probe query - // that requires the actor_id index until it succeeds instead of a fixed sleep. + // ScyllaDB builds secondary indexes asynchronously. Poll each indexed column + // until its index answers, instead of a fixed sleep. waitForCassandraIndexes(session, KeySpace, schemas.Collections.AuditLog, - []string{"actor_id", "action", "resource_type", "resource_id"}, 30*time.Second) + []string{"actor_id", "action", "resource_type", "resource_id"}, 60*time.Second) // Client table clientCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, client_id text, kind text, name text, description text, client_secret text, allowed_scopes text, redirect_uris text, grant_types text, token_endpoint_auth_method text, is_active boolean, org_id text, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.Client) @@ -645,7 +645,13 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { // fail until the index is ready. func waitForCassandraSecondaryIndex(session *cansandraDriver.Session, keyspace, table, column string, timeout time.Duration) { probe := fmt.Sprintf("SELECT id FROM %s.%s WHERE %s='' LIMIT 1 ALLOW FILTERING", keyspace, table, column) - deadline := time.Now().Add(timeout) + pollUntilQuerySucceeds(session, probe, time.Now().Add(timeout)) +} + +// pollUntilQuerySucceeds retries probe with a capped linear backoff until it +// succeeds or deadline passes. Shared so the backoff cannot drift between the +// single-column and multi-column waits. +func pollUntilQuerySucceeds(session *cansandraDriver.Session, probe string, deadline time.Time) { delay := 500 * time.Millisecond for { if err := session.Query(probe).Exec(); err == nil { @@ -661,30 +667,36 @@ func waitForCassandraSecondaryIndex(session *cansandraDriver.Session, keyspace, } } -// waitForCassandraIndexes polls a probe query that requires the actor_id secondary -// index until it succeeds or the timeout is reached. ScyllaDB builds secondary -// indexes asynchronously; queries on indexed columns fail until the index is ready. -// columns are probed independently: Scylla builds secondary indexes -// concurrently, so the last one created is not necessarily the last one ready, -// and waiting on a single column would let a query against another index run -// before that index exists. +// waitForCassandraIndexes probes each indexed column until its secondary index +// answers, or until that column's share of the timeout runs out. ScyllaDB builds +// secondary indexes asynchronously and queries on an indexed column fail until +// the index is ready. +// +// Every column is probed, not just one: Scylla builds indexes concurrently, so +// the last one created is not necessarily the last one ready, and waiting on a +// single column would let a query against another index run before that index +// exists. +// +// The budget is split PER COLUMN rather than shared. With one shared deadline a +// slow first index consumes the whole budget and the remaining columns are never +// probed at all — turning a startup race on one column into a startup race on +// every other, which is worse than the single-column wait this replaced. An +// exhausted column moves on rather than abandoning the rest. +// +// Best-effort by design: this is startup smoothing, not a correctness gate, so an +// expired probe is not an error. func waitForCassandraIndexes(session *cansandraDriver.Session, keyspace, table string, columns []string, timeout time.Duration) { - deadline := time.Now().Add(timeout) + if len(columns) == 0 { + return + } + perColumn := timeout / time.Duration(len(columns)) for _, col := range columns { + // No ALLOW FILTERING here, deliberately: the hint would let the probe + // succeed whether or not the index exists, so the wait would return + // immediately and detect nothing. The probe has to be a query that + // FAILS until the index is ready. probe := fmt.Sprintf("SELECT id FROM %s.%s WHERE %s='' LIMIT 1", keyspace, table, col) - delay := 500 * time.Millisecond - for { - if err := session.Query(probe).Exec(); err == nil { - break - } - if time.Now().After(deadline) { - return - } - time.Sleep(delay) - if delay < 3*time.Second { - delay += 500 * time.Millisecond - } - } + pollUntilQuerySucceeds(session, probe, time.Now().Add(perColumn)) } } diff --git a/internal/storage/db/couchbase/provider.go b/internal/storage/db/couchbase/provider.go index 276e37c19..ea082b203 100644 --- a/internal/storage/db/couchbase/provider.go +++ b/internal/storage/db/couchbase/provider.go @@ -287,7 +287,14 @@ func getIndex(scopeName string) map[string][]string { auditLogIndex1 := fmt.Sprintf("CREATE INDEX AuditLogActorIdIndex ON %s.%s(actor_id)", scopeName, schemas.Collections.AuditLog) auditLogIndex2 := fmt.Sprintf("CREATE INDEX AuditLogActionIndex ON %s.%s(action)", scopeName, schemas.Collections.AuditLog) auditLogIndex3 := fmt.Sprintf("CREATE INDEX AuditLogCreatedAtIndex ON %s.%s(created_at)", scopeName, schemas.Collections.AuditLog) - indices[schemas.Collections.AuditLog] = []string{auditLogIndex1, auditLogIndex2, auditLogIndex3} + // resource_type / resource_id are filterable via ListAuditLogs. Without these + // the predicates still return correct rows, but N1QL falls back to the + // CREATE PRIMARY INDEX scan of the fastest-growing collection in the scope, + // while Cassandra serves the same filter from an index — the "O(1) in one + // backend, full scan in another" parity bug AGENTS.md calls out. + auditLogIndex4 := fmt.Sprintf("CREATE INDEX AuditLogResourceTypeIndex ON %s.%s(resource_type)", scopeName, schemas.Collections.AuditLog) + auditLogIndex5 := fmt.Sprintf("CREATE INDEX AuditLogResourceIdIndex ON %s.%s(resource_id)", scopeName, schemas.Collections.AuditLog) + indices[schemas.Collections.AuditLog] = []string{auditLogIndex1, auditLogIndex2, auditLogIndex3, auditLogIndex4, auditLogIndex5} // TrustedIssuer indexes trustedIssuerIndex1 := fmt.Sprintf("CREATE INDEX TrustedIssuerIssuerURLIndex ON %s.%s(issuer_url)", scopeName, schemas.Collections.TrustedIssuer) From 2bfc305d12fad15b3b3a6d677259fc8701505901 Mon Sep 17 00:00:00 2001 From: Lakhan Samani Date: Tue, 6 Oct 2026 16:04:32 +0530 Subject: [PATCH 4/4] fix(couchbase): keep index creation from aborting startup on upgrade MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in the same path, both only reachable when a NEW secondary index is built against a collection that already holds data — i.e. on upgrade, not on a fresh install. 1. The per-attempt query timeout was gocb's default 75s while the retry budget is 30s, so a single attempt could return past the deadline and the loop gave up. Since the caller turns an index error into `return nil, err`, the server failed to boot. Now the per-attempt timeout is a third of the budget and is passed explicitly, so retries fit inside it. Timing out an attempt is harmless: CREATE INDEX registers the definition server-side and keeps building, so the next attempt returns "already exists" and startup proceeds while the backfill finishes — matching the async DDL the Cassandra provider already relies on. 2. isTransientQueryErr substring-matched "unambiguous timeout", which does not match gocbcore's "ambiguous timeout". A CREATE INDEX that outruns the client is precisely the ambiguous case, because the server may have accepted it — so the error was treated as fatal and never retried. Now checked with errors.Is against gocb.ErrTimeout, the parent both sentinels wrap. Found by a backward-compatibility review of the audit-filter change, which adds the first two indexes this repo has ever created against a populated audit_log. --- internal/storage/db/couchbase/provider.go | 33 ++++++++++- .../db/couchbase/provider_index_retry_test.go | 56 +++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) create mode 100644 internal/storage/db/couchbase/provider_index_retry_test.go diff --git a/internal/storage/db/couchbase/provider.go b/internal/storage/db/couchbase/provider.go index ea082b203..449c66333 100644 --- a/internal/storage/db/couchbase/provider.go +++ b/internal/storage/db/couchbase/provider.go @@ -143,6 +143,17 @@ func isTransientQueryErr(err error) bool { if err == nil { return false } + // errors.Is on the shared parent, not a substring match on the message. + // gocbcore defines ErrAmbiguousTimeout as "ambiguous timeout" and + // ErrUnambiguousTimeout as "unambiguous timeout", both wrapping ErrTimeout. + // A substring test for "unambiguous timeout" therefore MISSES the ambiguous + // one — and a CREATE INDEX that outruns the client is precisely the + // ambiguous case, because the server may well have accepted it. Missing it + // meant no retry, and a non-retried index error aborts provider + // construction, i.e. the server does not start. + if errors.Is(err, gocb.ErrTimeout) { + return true + } msg := strings.ToLower(err.Error()) return strings.Contains(msg, "eof") || strings.Contains(msg, "connection reset") || @@ -155,11 +166,31 @@ func isTransientQueryErr(err error) bool { } // execIndexQuery runs a CREATE INDEX statement with retries for transient query-service errors. +// +// The per-attempt timeout is bounded to a third of the overall budget, and is +// passed explicitly rather than left to gocb's default. gocb's default +// QueryTimeout is 75s while this budget defaults to 30s, so a single attempt +// could outrun the whole deadline and return past it — the retry that would +// then see "already exists" never ran, and the resulting error aborts provider +// construction (the caller does `return nil, err`), so the server fails to +// boot. +// +// That is not hypothetical on upgrade: Couchbase CREATE INDEX without +// defer_build is synchronous, and a newly added index on a collection that +// already holds months of audit rows builds over all of them. Timing out the +// attempt is harmless — the index definition is registered server-side and the +// build continues — so the next attempt returns "already exists" and startup +// proceeds while the backfill finishes in the background. That matches the +// async DDL behaviour the Cassandra provider already relies on. func execIndexQuery(scope *gocb.Scope, query string, timeout time.Duration) error { deadline := time.Now().Add(timeout) + perAttempt := timeout / 3 + if perAttempt <= 0 { + perAttempt = timeout + } delay := 500 * time.Millisecond for { - _, err := scope.Query(query, nil) + _, err := scope.Query(query, &gocb.QueryOptions{Timeout: perAttempt}) if err == nil { return nil } diff --git a/internal/storage/db/couchbase/provider_index_retry_test.go b/internal/storage/db/couchbase/provider_index_retry_test.go new file mode 100644 index 000000000..6af3386a1 --- /dev/null +++ b/internal/storage/db/couchbase/provider_index_retry_test.go @@ -0,0 +1,56 @@ +package couchbase + +import ( + "errors" + "fmt" + "testing" + + "github.com/couchbase/gocb/v2" + "github.com/stretchr/testify/assert" +) + +// A CREATE INDEX that outruns the client is the AMBIGUOUS timeout case — the +// server may have accepted it. If that is not treated as transient, the retry +// that would see "already exists" never runs, and a non-retried index error +// aborts provider construction: the server does not boot. +// +// The original check substring-matched "unambiguous timeout", which does not +// match gocbcore's "ambiguous timeout". This pins both. +func TestIsTransientQueryErr_CoversBothTimeoutKinds(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"ambiguous", gocb.ErrAmbiguousTimeout}, + {"unambiguous", gocb.ErrUnambiguousTimeout}, + {"wrapped ambiguous", fmt.Errorf("create index: %w", gocb.ErrAmbiguousTimeout)}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.True(t, isTransientQueryErr(tc.err), + "%v must be retryable, else index creation aborts startup", tc.err) + }) + } +} + +// Guard the premise of the fix: the two sentinels really do share ErrTimeout, +// and their messages really do differ in a way substring matching gets wrong. +func TestGocbTimeoutSentinels_ShareParentButNotMessage(t *testing.T) { + assert.True(t, errors.Is(gocb.ErrAmbiguousTimeout, gocb.ErrTimeout)) + assert.True(t, errors.Is(gocb.ErrUnambiguousTimeout, gocb.ErrTimeout)) + assert.NotContains(t, gocb.ErrAmbiguousTimeout.Error(), "unambiguous timeout", + "if this ever contains the longer string, the old substring check was fine") +} + +func TestIsTransientQueryErr_NonTransientStaysFatal(t *testing.T) { + assert.False(t, isTransientQueryErr(nil)) + assert.False(t, isTransientQueryErr(errors.New("syntax error in CREATE INDEX"))) +} + +// "already exists" short-circuits before the transient check, so a re-run +// against a database that already has the index is a no-op rather than a retry +// loop. This is what makes repeated boots safe. +func TestIsIndexExistsErr_TolerantOnRestart(t *testing.T) { + assert.True(t, isIndexExistsErr("Index AuditLogResourceIdIndex already exists")) + assert.True(t, isIndexExistsErr("The index #primary already exists")) + assert.False(t, isIndexExistsErr("ambiguous timeout")) +}