Skip to content

fix(storage): make cassandra index readiness probe actually wait #810

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lakhansamani merged 1 commit into from
Oct 14, 2026
Merged

fix(storage): make cassandra index readiness probe actually wait #810

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 83 additions & 75 deletions internal/storage/db/cassandradb/provider.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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()
Expand Down Expand Up @@ -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 {
Expand All @@ -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 {
Expand All @@ -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()
Expand All @@ -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 {
Expand Down Expand Up @@ -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 {
Expand All @@ -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)
Expand All @@ -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)
Expand All @@ -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,
Expand All @@ -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 {
Expand All @@ -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 {
Expand Down
43 changes: 43 additions & 0 deletions internal/storage/db/cassandradb/provider_index_probe_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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)
}
}
}
Loading