diff --git a/internal/storage/db/cassandradb/provider.go b/internal/storage/db/cassandradb/provider.go index bf82b985..af9ad6a8 100644 --- a/internal/storage/db/cassandradb/provider.go +++ b/internal/storage/db/cassandradb/provider.go @@ -430,11 +430,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } - // 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"}, 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) err = session.Query(clientCollectionQuery).Exec() @@ -451,11 +446,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the client_id - // index (used by GetClientByClientID and the boot-time reserved-client seed) - // to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.Client, "client_id", 30*time.Second) - // TrustedIssuer table and indexes trustedIssuerCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, client_id text, name text, issuer_url text, key_source_type text, jwks_url text, expected_aud text, subject_claim text, allowed_subjects text, issuer_type text, auth_method text, is_active boolean, enable_token_review boolean, kubernetes_api_server_url text, spiffe_refresh_hint_seconds bigint, trusted_proxy_header text, trusted_proxy_cidrs text, kind text, org_id text, sso_client_id text, sso_client_secret_enc text, sso_scopes text, sso_redirect_uri text, saml_sso_url text, saml_idp_cert_pem text, saml_sp_entity_id text, saml_acs_url text, saml_attribute_mapping text, saml_allow_idp_initiated boolean, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.TrustedIssuer) err = session.Query(trustedIssuerCollectionQuery).Exec() @@ -493,10 +483,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the issuer_url - // index (hot path for client_assertion validation) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.TrustedIssuer, "issuer_url", 30*time.Second) - // SAMLServiceProvider table and indexes (registered downstream SPs; Authorizer as IdP) samlServiceProviderCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, org_id text, name text, entity_id text, acs_url text, sp_cert_pem text, name_id_format text, mapped_attributes text, allow_idp_initiated boolean, is_active boolean, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.SAMLServiceProvider) if err = session.Query(samlServiceProviderCollectionQuery).Exec(); err != nil { @@ -510,10 +496,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err = session.Query(samlServiceProviderEntityIDIndex).Exec(); err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the org_id index - // (used to resolve an incoming AuthnRequest to its registered SP) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.SAMLServiceProvider, "org_id", 30*time.Second) - // SAMLIDPKey table and index (per-org signing keypairs with rotation) samlIDPKeyCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, org_id text, cert_pem text, private_key_enc text, algorithm text, status text, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.SAMLIDPKey) if err = session.Query(samlIDPKeyCollectionQuery).Exec(); err != nil { @@ -523,10 +505,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err = session.Query(samlIDPKeyOrgIDIndex).Exec(); err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the org_id index - // (used to enumerate an org's signing keys for metadata publishing) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.SAMLIDPKey, "org_id", 30*time.Second) - // WebauthnCredential table and indexes webauthnCredentialCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, user_id text, credential_id text, public_key text, sign_count bigint, flags bigint, transports text, aaguid text, name text, created_at bigint, updated_at bigint, last_used_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.WebauthnCredential) err = session.Query(webauthnCredentialCollectionQuery).Exec() @@ -543,10 +521,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the credential_id - // index (hot path for usernameless login) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.WebauthnCredential, "credential_id", 30*time.Second) - // Organization table and index organizationCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, name text, display_name text, enabled boolean, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.Organization) if err = session.Query(organizationCollectionQuery).Exec(); err != nil { @@ -585,14 +559,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for the lookup - // columns used by the uniqueness guard and membership listings. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.Organization, "name", 30*time.Second) - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.OrgMembership, "org_id", 30*time.Second) - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.OrgMembership, "user_id", 30*time.Second) - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.FederatedIdentity, "org_id", 30*time.Second) - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.FederatedIdentity, "user_id", 30*time.Second) - // ScimEndpoint table and index scimEndpointCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, org_id text, token_hash text, enabled boolean, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.ScimEndpoint) if err = session.Query(scimEndpointCollectionQuery).Exec(); err != nil { @@ -602,10 +568,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err = session.Query(scimEndpointOrgIndex).Exec(); err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for org_id (used by - // GetScimEndpointByOrgID and the uniqueness guard) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.ScimEndpoint, "org_id", 30*time.Second) - // ScimGroup table and index. Membership is NOT stored here (it lives in FGA); // org_id indexes the displayName-eq lookup (org_id + display_name ALLOW FILTERING). scimGroupCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, org_id text, display_name text, external_id text, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.ScimGroup) @@ -616,8 +578,6 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err = session.Query(scimGroupOrgIndex).Exec(); err != nil { return nil, err } - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.ScimGroup, "org_id", 30*time.Second) - // OrgDomain table and index. The normalized domain is the partition key, so // INSERT ... IF NOT EXISTS enforces first-writer-wins atomically (LWT). orgDomainCollectionQuery := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text, org_id text, domain text, verified_at bigint, created_at bigint, updated_at bigint, PRIMARY KEY (id))", KeySpace, schemas.Collections.OrgDomain) @@ -628,9 +588,19 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { if err = session.Query(orgDomainOrgIndex).Exec(); err != nil { return nil, err } - // ScyllaDB builds secondary indexes asynchronously; wait for org_id (used by - // ListOrgDomainsByOrg and the cascade) to become queryable. - waitForCassandraSecondaryIndex(session, KeySpace, schemas.Collections.OrgDomain, "org_id", 30*time.Second) + // Every secondary index created above is probed here, once, against a single + // shared budget. ScyllaDB builds secondary indexes asynchronously and a query + // on an indexed column fails until the index is ready, so a provider that + // returns before they answer hands its first caller an intermittent + // index-unavailable error against a freshly provisioned keyspace. + // + // Consolidated at the end rather than one wait per table: the budget is then + // bounded for the whole of startup instead of per call site, and the indexes + // created earliest have had the longest to settle by the time they are + // probed. Nothing inside NewProvider queries an indexed column, so deferring + // costs nothing — the boot-time reserved-client seed runs in cmd/root.go, + // after this returns. + waitForCassandraIndexes(session, KeySpace, cassandraIndexProbes, 90*time.Second) return &provider{ config: cfg, @@ -639,18 +609,43 @@ func NewProvider(cfg *config.Config, deps *Dependencies) (*provider, error) { }, err } -// waitForCassandraSecondaryIndex polls a probe query that requires the given -// column's 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 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) - pollUntilQuerySucceeds(session, probe, time.Now().Add(timeout)) +// indexProbe names one secondary index to wait on, as the (table, column) pair +// its readiness probe queries. +type indexProbe struct { + table string + column string +} + +// cassandraIndexProbes is every secondary index NewProvider waits on before +// returning. It is the subset of created indexes that a caller can hit +// immediately after startup, so a missing entry here is a startup race, not a +// cosmetic omission. +// +// Indexes on bigint columns (audit_log.created_at, session_token.expires_at, +// mfa_session.expires_at) are absent by necessity: the probe restricts on ”, +// which only type-checks against a text column. +var cassandraIndexProbes = []indexProbe{ + {schemas.Collections.AuditLog, "actor_id"}, + {schemas.Collections.AuditLog, "action"}, + {schemas.Collections.AuditLog, "resource_type"}, + {schemas.Collections.AuditLog, "resource_id"}, + {schemas.Collections.Client, "client_id"}, + {schemas.Collections.TrustedIssuer, "issuer_url"}, + {schemas.Collections.SAMLServiceProvider, "org_id"}, + {schemas.Collections.SAMLIDPKey, "org_id"}, + {schemas.Collections.WebauthnCredential, "credential_id"}, + {schemas.Collections.Organization, "name"}, + {schemas.Collections.OrgMembership, "org_id"}, + {schemas.Collections.OrgMembership, "user_id"}, + {schemas.Collections.FederatedIdentity, "org_id"}, + {schemas.Collections.FederatedIdentity, "user_id"}, + {schemas.Collections.ScimEndpoint, "org_id"}, + {schemas.Collections.ScimGroup, "org_id"}, + {schemas.Collections.OrgDomain, "org_id"}, } // 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. +// succeeds or deadline passes. func pollUntilQuerySucceeds(session *cansandraDriver.Session, probe string, deadline time.Time) { delay := 500 * time.Millisecond for { @@ -667,39 +662,52 @@ func pollUntilQuerySucceeds(session *cansandraDriver.Session, probe string, dead } } -// 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. +// waitForCassandraIndexes probes each secondary index until it answers, or until +// that probe's share of the shared budget runs out. ScyllaDB builds secondary +// indexes asynchronously and a query on an indexed column fails 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. +// Every index is probed, not just one per table: Scylla builds them +// 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. +// The budget is split PER PROBE rather than shared as one deadline. With a +// single shared deadline a slow first index consumes the whole budget and the +// remaining probes never run at all — turning a startup race on one column into +// a startup race on every other. +// +// What this detects: the probe fails until the index SCHEMA propagates, which +// for a fresh keyspace is the same moment the (empty) index is usable. It is not +// a view-build gate — an index added to a table that already holds rows answers +// the probe while its backfill is still running. Readiness of a backfill is +// system_distributed.view_build_status, deliberately not consulted here: a cold +// start must not block on a backfill it does not need. // // 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) { - if len(columns) == 0 { +func waitForCassandraIndexes(session *cansandraDriver.Session, keyspace string, probes []indexProbe, timeout time.Duration) { + if len(probes) == 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) - pollUntilQuerySucceeds(session, probe, time.Now().Add(perColumn)) + perProbe := timeout / time.Duration(len(probes)) + for _, p := range probes { + pollUntilQuerySucceeds(session, indexProbeQuery(keyspace, p), time.Now().Add(perProbe)) } } +// indexProbeQuery builds the readiness probe for one index. +// +// It carries NO ALLOW FILTERING, deliberately: that hint is exactly what lets +// Cassandra answer a query on a NON-indexed column, so a probe carrying it +// succeeds whether or not the index exists — the wait returns on its first +// attempt having detected nothing, and every call site silently keeps the +// startup race the wait was written to close. The probe has to be a query that +// FAILS until the index is ready. +func indexProbeQuery(keyspace string, p indexProbe) string { + return fmt.Sprintf("SELECT id FROM %s.%s WHERE %s='' LIMIT 1", keyspace, p.table, p.column) +} + // Close closes the Cassandra session. func (p *provider) Close() error { if p == nil || p.db == nil { diff --git a/internal/storage/db/cassandradb/provider_index_probe_test.go b/internal/storage/db/cassandradb/provider_index_probe_test.go new file mode 100644 index 00000000..e380cb95 --- /dev/null +++ b/internal/storage/db/cassandradb/provider_index_probe_test.go @@ -0,0 +1,43 @@ +package cassandradb + +import ( + "strings" + "testing" + + "github.com/authorizerdev/authorizer/internal/storage/schemas" +) + +// TestIndexProbeQueryOmitsAllowFiltering is the regression guard for the bug where +// the readiness probe carried ALLOW FILTERING. That hint is what lets Cassandra +// answer a query on a non-indexed column, so the probe succeeded on its first +// attempt whether or not the index existed — the wait never waited, and all 17 +// call sites kept the startup race it was written to close. +// +// A probe is only a readiness probe if it FAILS until the index is ready. +func TestIndexProbeQueryOmitsAllowFiltering(t *testing.T) { + q := indexProbeQuery("authorizer", indexProbe{"audit_log", "actor_id"}) + if strings.Contains(strings.ToUpper(q), "ALLOW FILTERING") { + t.Fatalf("probe must not carry ALLOW FILTERING, it defeats its own purpose: %s", q) + } + if want := "SELECT id FROM authorizer.audit_log WHERE actor_id='' LIMIT 1"; q != want { + t.Fatalf("probe shape changed\n got: %s\nwant: %s", q, want) + } +} + +// TestIndexProbesCoverAuditLogFilters keeps the wait list in step with the columns +// ListAuditLogs filters on. An audit filter whose index is not waited for returns +// an index-unavailable error on the first query against a fresh keyspace. +func TestIndexProbesCoverAuditLogFilters(t *testing.T) { + probed := map[indexProbe]bool{} + for _, p := range cassandraIndexProbes { + if probed[p] { + t.Errorf("duplicate probe %v — it would consume two shares of the budget", p) + } + probed[p] = true + } + for _, col := range []string{"actor_id", "action", "resource_type", "resource_id"} { + if !probed[indexProbe{schemas.Collections.AuditLog, col}] { + t.Errorf("audit_log.%s is filterable but never probed", col) + } + } +}