diff --git a/docs/design/20261004-authorization-cache-reuse.md b/docs/design/20261004-authorization-cache-reuse.md new file mode 100644 index 0000000000000..a3c885e525f58 --- /dev/null +++ b/docs/design/20261004-authorization-cache-reuse.md @@ -0,0 +1,93 @@ +# Reuse authorization decisions against current catalog state + +Owner: frontend authorization and disttae catalog reads. Follow-up to #29532. +Initial validation base: `5e965dece9e`; rebased onto `ff53ee2eb0c`. + +## Problem and contract + +#29532 clears session privileges at every statement to observe remote REVOKE +and RESTORE. In the TPCC 10/10 comparison of `96196b4a9c` and `ae56c83b62`, +authorization CPU rises from 23 to 1,025 seconds per five-minute window. +Repeated catalog SQL adds parsing, planning, compilation and allocation. + +Reuse a cached grant only when its catalog dependencies are unchanged at a +snapshot obtainable by a new authorization transaction. An older data transaction +snapshot is insufficient. Existing protected-database, view-chain, active-role +and ownership checks remain the permission decision owners. + +## Ownership and validity + +Disttae supplies a comparable token for the existing session privilege cache: +mo_user, mo_role, mo_user_grant, mo_role_grant, mo_role_privs, mo_database and +mo_tables. The last two cover object ownership and physical catalog table IDs. +Each entry holds the physical table ID, existing shared-state identity and +applied logtail watermark. Copies and GC preserve identity; reconstruction does +not. Tokens retain only tiny identities, never old row/object trees. The applied +watermark covers row inserts/deletes; the last-flush timestamp does not. + +Reject reuse when logtail is not ready. Otherwise acquire a read-only transaction +through TxnClient.New and close it without workspace, writes or storage commit. +This reuses existing freshness, admission, cancellation and close ownership; +it does not claim stronger consistency than current authorization reads. +Propagate its snapshot lower bound to the session so cache-miss background +transactions cannot evaluate permissions before the captured token. + +Capture only ready subscription states with no pending apply and applied +watermarks below the exclusive read snapshot. Missing/future catalogs or a +catalog-generation transition provide no reusable token. Uncertainty clears the +cache and uses existing authorization SQL; actual snapshot/subscription errors +propagate. Capture before evaluation: a later commit must invalidate those +results next time, never stamp them with a newer version. + +Role/identity changes, manual clear and cache-mode changes keep their existing +invalidation. Freshness is checked by the existing privilege and active-role +consumers, before reading cached decisions. This covers nested SET queries and +WITH GRANT OPTION without making grant-free controls depend on catalog health. +Clearing is local; the next authorization revalidates membership in its own RC +transaction. Ordinary SELECT/UPDATE and bound prepared execution check once; +transaction controls and Execute shells do not check. Complex multi-object +operations may require several consuming evaluations. + +EXPLAIN authorizes its inner statement. EXPLAIN EXECUTE binding checks the resolved +AST and plan before releasing the AST. A prepared EXPLAIN EXECUTE rebinds the +inner mutable handle on execution; its cached query cannot establish that binding. +Ordinary prepared SELECT/UPDATE retain their plan and compile reuse. + +View chains still require metadata checks. Negative lookups allocate nothing; +positive scopes are capped at 1,024. Table/view scope storage shares one +implementation; the unused replacement method and unconsumed atomic counters +are removed. + +## Complexity and validation + +There is no persistent epoch, wire/config change, broadcaster, worker or second +permission cache. The hot path exchanges repeated SQL for one read-only snapshot +and seven catalog lookups. Ordinary data writes preserve the token; unrelated +metadata changes can conservatively invalidate it. TTL invalidation would permit +stale grants; an administrator exemption would leave ordinary-user performance +unfixed and redefine catalog-based authorization. + +Main-agent design/self-review: R3 authorization, generation and hot-path changes; +no subagents. Component tests cover watermark boundaries, pending application, +reconstruction, failures and scope capacity. Reuse the authenticated two-CN +fixture for text/SQL-prepared/binary-prepared queries, remote revoke/regrant, +role switching, old data transactions, DROP/recreate and cancellation. The six +existing #29399 restore/PITR tests retain their original behavioral oracles. + +## Local performance boundary + +Same-base one-row workload, three rounds of 1,500 prepared point reads and 300 +BEGIN / SELECT FOR UPDATE / UPDATE / COMMIT transactions. Median microseconds: + +| Workload | Per-statement clear | Unsafe reuse control | This change | +|---|---:|---:|---:| +| Admin point | 336 | 152 | 149 | +| Ordinary point | 1,413 | 156 | 173 | +| Admin transaction | 5,994 | 5,879 | 5,596 | +| Ordinary transaction | 8,907 | 5,767 | 5,625 | + +The unsafe control disables freshness validation only to estimate historical +cache cost; it is not shipped. Transaction latency is near that control; the +ordinary point read still costs about 17 microseconds more. These small macOS +ARM64 measurements, on a machine with other activity, do not prove unchanged +full TPCC performance. Full TPCC comparison remains required to claim parity. diff --git a/pkg/frontend/authenticate.go b/pkg/frontend/authenticate.go index bc4902a446a33..08e0280b44715 100644 --- a/pkg/frontend/authenticate.go +++ b/pkg/frontend/authenticate.go @@ -74,6 +74,7 @@ import ( "github.com/matrixorigin/matrixone/pkg/util/trace" "github.com/matrixorigin/matrixone/pkg/util/trace/impl/motrace" "github.com/matrixorigin/matrixone/pkg/util/trace/impl/motrace/statistic" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae" ) type TenantInfo struct { @@ -2648,6 +2649,8 @@ const ( // privilegeCache cache privileges on table type privilegeCache struct { + catalogVersion disttae.PrivilegeCacheVersion + cachedScopes int // For objectType table // For objectType table *, *.* storeForTable [int(privilegeLevelEnd)]btree.Set[PrivilegeType] @@ -2669,8 +2672,6 @@ type privilegeCache struct { storeForDatabase2 btree.Map[string, *btree.Set[PrivilegeType]] // For objectType account * storeForAccount [int(privilegeLevelEnd)]btree.Set[PrivilegeType] - total atomic.Uint64 - hit atomic.Uint64 // The active primary role is session state, while its grant to the user is // catalog state. Keep the validation in the same cache generation as the @@ -2738,12 +2739,10 @@ func (pc *privilegeCache) setActiveRoleGrantForGeneration( // has checks the cache has privilege on a table func (pc *privilegeCache) has(objTyp objectType, plt privilegeLevelType, dbName, tableName string, priv PrivilegeType) bool { - pc.total.Add(1) - privSet := pc.getPrivilegeSet(objTyp, plt, dbName, tableName) + privSet := pc.getPrivilegeSet(objTyp, plt, dbName, tableName, false) if privSet != nil { for _, p := range privilegeTypeWithCoveringPrivileges(objTyp, priv) { if privSet.Contains(p) { - pc.hit.Add(1) return true } } @@ -2751,57 +2750,27 @@ func (pc *privilegeCache) has(objTyp objectType, plt privilegeLevelType, dbName, return false } -func (pc *privilegeCache) getPrivilegeSet(objTyp objectType, plt privilegeLevelType, dbName, tableName string) *btree.Set[PrivilegeType] { +// Only successful grants allocate a scoped entry. Negative lookups are read-only. +// Bound the retained scope count independently of catalog/connection lifetime. +const maxPrivilegeCacheScopes = 1024 + +func (pc *privilegeCache) getPrivilegeSet(objTyp objectType, plt privilegeLevelType, dbName, tableName string, create bool) *btree.Set[PrivilegeType] { + var databases *btree.Map[string, *btree.Set[PrivilegeType]] + var relations *btree.Map[string, *btree.Map[string, *btree.Set[PrivilegeType]]] switch objTyp { - case objectTypeTable: - switch plt { - case privilegeLevelStarStar, privilegeLevelStar: - return &pc.storeForTable[plt] - case privilegeLevelDatabaseStar: - dbStore, ok1 := pc.storeForTable2.Get(dbName) - if !ok1 { - dbStore = &btree.Set[PrivilegeType]{} - pc.storeForTable2.Set(dbName, dbStore) - } - return dbStore - case privilegeLevelDatabaseTable, privilegeLevelTable: - tableStore, ok1 := pc.storeForTable3.Get(dbName) - if !ok1 { - tableStore = &btree.Map[string, *btree.Set[PrivilegeType]]{} - pc.storeForTable3.Set(dbName, tableStore) - } - privSet, ok2 := tableStore.Get(tableName) - if !ok2 { - privSet = &btree.Set[PrivilegeType]{} - tableStore.Set(tableName, privSet) - } - return privSet - default: - return nil + case objectTypeTable, objectTypeView: + levels := &pc.storeForTable + databases, relations = &pc.storeForTable2, &pc.storeForTable3 + if objTyp == objectTypeView { + levels = &pc.storeForView + databases, relations = &pc.storeForView2, &pc.storeForView3 } - case objectTypeView: switch plt { case privilegeLevelStarStar, privilegeLevelStar: - return &pc.storeForView[plt] + return &levels[plt] case privilegeLevelDatabaseStar: - dbStore, ok1 := pc.storeForView2.Get(dbName) - if !ok1 { - dbStore = &btree.Set[PrivilegeType]{} - pc.storeForView2.Set(dbName, dbStore) - } - return dbStore + relations = nil case privilegeLevelDatabaseTable, privilegeLevelTable: - viewStore, ok1 := pc.storeForView3.Get(dbName) - if !ok1 { - viewStore = &btree.Map[string, *btree.Set[PrivilegeType]]{} - pc.storeForView3.Set(dbName, viewStore) - } - privSet, ok2 := viewStore.Get(tableName) - if !ok2 { - privSet = &btree.Set[PrivilegeType]{} - viewStore.Set(tableName, privSet) - } - return privSet default: return nil } @@ -2810,12 +2779,7 @@ func (pc *privilegeCache) getPrivilegeSet(objTyp objectType, plt privilegeLevelT case privilegeLevelStar, privilegeLevelStarStar: return &pc.storeForDatabase[plt] case privilegeLevelDatabase: - dbStore, ok1 := pc.storeForDatabase2.Get(dbName) - if !ok1 { - dbStore = &btree.Set[PrivilegeType]{} - pc.storeForDatabase2.Set(dbName, dbStore) - } - return dbStore + databases = &pc.storeForDatabase2 default: return nil } @@ -2824,23 +2788,38 @@ func (pc *privilegeCache) getPrivilegeSet(objTyp objectType, plt privilegeLevelT default: return nil } - -} - -// set replaces the privileges by new ones -func (pc *privilegeCache) set(objTyp objectType, plt privilegeLevelType, dbName, tableName string, priv ...PrivilegeType) { - privSet := pc.getPrivilegeSet(objTyp, plt, dbName, tableName) - if privSet != nil { - privSet.Clear() - for _, p := range priv { - privSet.Insert(p) + if create && pc.cachedScopes >= maxPrivilegeCacheScopes { + pc.invalidate() + } + if relations == nil { + set, found := databases.Get(dbName) + if !found && create { + set = &btree.Set[PrivilegeType]{} + databases.Set(dbName, set) + pc.cachedScopes++ } + return set } + tables, found := relations.Get(dbName) + if !found { + if !create { + return nil + } + tables = &btree.Map[string, *btree.Set[PrivilegeType]]{} + relations.Set(dbName, tables) + } + set, found := tables.Get(tableName) + if !found && create { + set = &btree.Set[PrivilegeType]{} + tables.Set(tableName, set) + pc.cachedScopes++ + } + return set } // add puts the privileges without replacing existed ones func (pc *privilegeCache) add(objTyp objectType, plt privilegeLevelType, dbName, tableName string, priv ...PrivilegeType) { - privSet := pc.getPrivilegeSet(objTyp, plt, dbName, tableName) + privSet := pc.getPrivilegeSet(objTyp, plt, dbName, tableName, true) if privSet != nil { for _, p := range priv { privSet.Insert(p) @@ -2853,11 +2832,11 @@ func (pc *privilegeCache) invalidate() { if pc == nil { return } + pc.catalogVersion = disttae.PrivilegeCacheVersion{} + pc.cachedScopes = 0 // Advance first so a validation that started in the old generation can // never become visible even if its atomic Store races with the clear below. pc.activeRoleGrantGeneration.Add(1) - // total := pc.total.Swap(0) - // hit := pc.hit.Swap(0) for i := privilegeLevelStar; i < privilegeLevelEnd; i++ { pc.storeForTable[i].Clear() pc.storeForView[i].Clear() @@ -2870,13 +2849,6 @@ func (pc *privilegeCache) invalidate() { pc.storeForView3.Clear() pc.storeForDatabase2.Clear() pc.activeRoleGrant.Store(nil) - // ratio := float64(0) - // if total == 0 { - // ratio = 0 - // } else { - // ratio = float64(hit) / float64(total) - // } - // logutil.Debugf("-->hit %d total %d ratio %f", hit, total, ratio) } // verifiedRole holds the role info that has been checked @@ -8204,6 +8176,11 @@ func determineUserHasPrivilegeSet(ctx context.Context, ses *Session, priv *privi return false, stats, nil } + // Freshness belongs to authorization consumption, including nested plans. + if err = ses.refreshPrivilegeCache(ctx); err != nil { + return false, stats, err + } + enableCache, err = privilegeCacheIsEnabled(ctx, ses) if err != nil { return false, stats, err @@ -8507,6 +8484,11 @@ func validateActiveRoleGrantForAuthorization( return true, stats, nil } + // Freshness belongs to authorization consumption, including nested plans. + if err = ses.refreshPrivilegeCache(ctx); err != nil { + return false, stats, err + } + enableCache, err := privilegeCacheIsEnabled(ctx, ses) if err != nil { return false, stats, err diff --git a/pkg/frontend/authenticate_test.go b/pkg/frontend/authenticate_test.go index 5ed363e2a7207..dada3e925eb20 100644 --- a/pkg/frontend/authenticate_test.go +++ b/pkg/frontend/authenticate_test.go @@ -2678,10 +2678,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) // Mock getDatabaseOrTableId for scoped levels. if stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_STAR || @@ -2856,10 +2856,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) // Mock getDatabaseOrTableId for scoped levels. if stmt.Level.Level == tree.PRIVILEGE_LEVEL_TYPE_STAR || @@ -3007,8 +3007,8 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkSql, err := getSqlForCheckDatabase(ses.GetTxnHandler().GetTxnCtx(), "db") convey.So(err, convey.ShouldBeNil) @@ -3064,9 +3064,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkTableSql, err := getSqlForCheckDatabaseTable(ctx, "db", "t1") convey.So(err, convey.ShouldBeNil) @@ -3127,9 +3127,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) privType, err := convertAstPrivilegeTypeToPrivilegeType(context.TODO(), stmt.Privileges[0].Type, stmt.ObjType) convey.So(err, convey.ShouldBeNil) @@ -3182,9 +3182,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) privType, err := convertAstPrivilegeTypeToPrivilegeType(context.TODO(), stmt.Privileges[0].Type, stmt.ObjType) convey.So(err, convey.ShouldBeNil) @@ -3238,8 +3238,8 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkSql, err := getSqlForCheckDatabase(ses.GetTxnHandler().GetTxnCtx(), "db1") convey.So(err, convey.ShouldBeNil) @@ -3299,9 +3299,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkTableSql, err := getSqlForCheckDatabaseTable(ctx, "db1", "t1") convey.So(err, convey.ShouldBeNil) @@ -3368,9 +3368,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v1") convey.So(err, convey.ShouldBeNil) @@ -3445,9 +3445,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v2") require.NoError(t, err) @@ -3522,9 +3522,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v3") require.NoError(t, err) @@ -3597,9 +3597,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) privType, err := convertAstPrivilegeTypeToPrivilegeType(context.TODO(), stmt.Privileges[0].Type, stmt.ObjType) require.NoError(t, err) @@ -3651,9 +3651,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v1") convey.So(err, convey.ShouldBeNil) @@ -3726,9 +3726,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v1") convey.So(err, convey.ShouldBeNil) @@ -3813,9 +3813,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkTableSql, err := getSqlForCheckDatabaseTable(ctx, "db", "missing_t") convey.So(err, convey.ShouldBeNil) @@ -3877,9 +3877,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkTableSql, err := getSqlForCheckDatabaseTable(ctx, "db", "missing_t") convey.So(err, convey.ShouldBeNil) @@ -3948,9 +3948,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkSql, err := getSqlForCheckDatabaseView(ctx, "db", "v1") convey.So(err, convey.ShouldBeNil) @@ -4024,9 +4024,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkDbSql, err := getSqlForCheckDatabase(ctx, "db1") convey.So(err, convey.ShouldBeNil) @@ -4083,9 +4083,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkDbSql, err := getSqlForCheckDatabase(ctx, "db1") convey.So(err, convey.ShouldBeNil) @@ -4151,9 +4151,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) globalSql := getSqlForCheckRoleHasPrivilegeWGOOrWithOwnershipWithObjAndExactLevel( int64(PrivilegeTypeDropTable), int64(PrivilegeTypeDatabaseAll), int64(PrivilegeTypeDatabaseOwnership), @@ -4209,9 +4209,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: uint32(internRoleID), } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) globalSql := getSqlForCheckRoleHasPrivilegeWGOOrWithOwnershipWithObjAndExactLevel( int64(PrivilegeTypeDropTable), int64(PrivilegeTypeDatabaseAll), int64(PrivilegeTypeDatabaseOwnership), @@ -4274,9 +4274,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: uint32(internRoleID), } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkTableSql, err := getSqlForCheckDatabaseTable(ctx, "db", "t1") convey.So(err, convey.ShouldBeNil) @@ -4350,9 +4350,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: uint32(internRoleID), } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) checkViewSql, err := getSqlForCheckDatabaseView(ctx, "db", "v1") convey.So(err, convey.ShouldBeNil) @@ -4435,9 +4435,9 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: uint32(internRoleID), } - ses.markActiveRoleGrantValid() ctx := ses.GetTxnHandler().GetTxnCtx() bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) wgoSql := getSqlForCheckRoleHasPrivilegeWGODependsOnPrivType(PrivilegeTypeDropDatabase) bh.sql2result[wgoSql] = newMrsForPrivilegeWGO([][]interface{}{ @@ -4511,10 +4511,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) for _, p := range stmt.Privileges { sql, err := formSqlFromGrantPrivilege(context.TODO(), ses, stmt, p) convey.So(err, convey.ShouldBeNil) @@ -4593,10 +4593,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) for i, p := range stmt.Privileges { sql, err := formSqlFromGrantPrivilege(context.TODO(), ses, stmt, p) convey.So(err, convey.ShouldBeNil) @@ -4666,10 +4666,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) for _, p := range stmt.Privileges { sql, err := formSqlFromGrantPrivilege(context.TODO(), ses, stmt, p) convey.So(err, convey.ShouldBeNil) @@ -4728,10 +4728,10 @@ func Test_determineGrantPrivilege(t *testing.T) { UserID: 1001, DefaultRoleID: 1001, } - ses.markActiveRoleGrantValid() ses.SetDatabaseName("db") //TODO: make sql2result bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(1001, 1001)] = newMrsForCheckUserGrant([][]interface{}{{int64(1001), int64(1001), false}}) for i, p := range stmt.Privileges { sql, err := formSqlFromGrantPrivilege(context.TODO(), ses, stmt, p) convey.So(err, convey.ShouldBeNil) @@ -13952,97 +13952,38 @@ func Test_graph(t *testing.T) { }) } -func Test_cache(t *testing.T) { - type arg struct { - db string - table string - } - cnt := 10 - args := make([]arg, 10) - for i := 0; i < cnt; i++ { - args[i].db = fmt.Sprintf("db%d", i) - args[i].table = fmt.Sprintf("table%d", i) - } - - cache1 := &privilegeCache{} - convey.Convey("has", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeFalse) - } - }) - - //add some privilege - for _, a := range args { - for i := PrivilegeTypeCreateAccount; i < PrivilegeTypeCreateObject; i++ { - cache1.add(objectTypeTable, privilegeLevelStar, a.db, a.table, i) - } - } - - convey.Convey("has2", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeTrue) - ret = cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateObject) - convey.So(ret, convey.ShouldBeFalse) - } - }) - - for _, a := range args { - for i := PrivilegeTypeCreateObject; i < PrivilegeTypeExecute; i++ { - cache1.add(objectTypeTable, privilegeLevelStar, a.db, a.table, i) - } +func TestPrivilegeCacheScopes(t *testing.T) { + cache := &privilegeCache{} + for _, scope := range []struct { + object objectType + level privilegeLevelType + }{ + {objectTypeTable, privilegeLevelStarStar}, + {objectTypeTable, privilegeLevelDatabaseStar}, + {objectTypeTable, privilegeLevelDatabaseTable}, + {objectTypeView, privilegeLevelDatabaseTable}, + {objectTypeDatabase, privilegeLevelDatabase}, + {objectTypeAccount, privilegeLevelStar}, + } { + require.False(t, cache.has(scope.object, scope.level, "db", "t", PrivilegeTypeSelect)) + cache.add(scope.object, scope.level, "db", "t", PrivilegeTypeSelect) + require.True(t, cache.has(scope.object, scope.level, "db", "t", PrivilegeTypeSelect)) + require.False(t, cache.has(scope.object, scope.level, "db", "t", PrivilegeTypeUpdate)) + cache.invalidate() + require.False(t, cache.has(scope.object, scope.level, "db", "t", PrivilegeTypeSelect)) } - - convey.Convey("has3", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeTrue) - ret = cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateObject) - convey.So(ret, convey.ShouldBeTrue) - } - }) - - //set - for _, a := range args { - for i := PrivilegeTypeCreateObject; i < PrivilegeTypeExecute; i++ { - cache1.set(objectTypeTable, privilegeLevelStar, a.db, a.table) - } + for i := range maxPrivilegeCacheScopes + 1 { + name := fmt.Sprint(i) + require.False(t, cache.has(objectTypeTable, privilegeLevelTable, name, "t", PrivilegeTypeSelect)) } - - convey.Convey("has4", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeFalse) - ret = cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateObject) - convey.So(ret, convey.ShouldBeFalse) - } - }) - - for _, a := range args { - for i := PrivilegeTypeCreateAccount; i < PrivilegeTypeExecute; i++ { - cache1.add(objectTypeTable, privilegeLevelStarStar, a.db, a.table, i) - } + require.Zero(t, cache.cachedScopes) + require.Zero(t, cache.storeForTable3.Len(), "cache misses must not retain empty scopes") + for i := range maxPrivilegeCacheScopes + 1 { + cache.add(objectTypeTable, privilegeLevelTable, "db", fmt.Sprint(i), PrivilegeTypeSelect) } - - convey.Convey("has4", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeFalse) - ret = cache1.has(objectTypeTable, privilegeLevelStarStar, a.db, a.table, PrivilegeTypeCreateObject) - convey.So(ret, convey.ShouldBeTrue) - } - }) - - cache1.invalidate() - convey.Convey("has4", t, func() { - for _, a := range args { - ret := cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateAccount) - convey.So(ret, convey.ShouldBeFalse) - ret = cache1.has(objectTypeTable, privilegeLevelStar, a.db, a.table, PrivilegeTypeCreateObject) - convey.So(ret, convey.ShouldBeFalse) - } - }) + require.LessOrEqual(t, cache.cachedScopes, maxPrivilegeCacheScopes) + require.True(t, cache.has(objectTypeTable, privilegeLevelTable, "db", fmt.Sprint(maxPrivilegeCacheScopes), PrivilegeTypeSelect)) + require.False(t, cache.has(objectTypeTable, privilegeLevelTable, "db", "0", PrivilegeTypeSelect)) } func TestActiveRoleGrantCacheLifecycle(t *testing.T) { @@ -14518,7 +14459,7 @@ func TestRevokedActiveRoleCannotUseAuthorizationFallbacks(t *testing.T) { DefaultRole: "reader", DefaultRoleID: 3, }) - ses.GetPrivilegeCache().setActiveRoleGrant(2, 3, false) + ses.GetPrivilegeCache().setActiveRoleGrant(2, 3, true) return ses } @@ -14527,6 +14468,7 @@ func TestRevokedActiveRoleCannotUseAuthorizationFallbacks(t *testing.T) { ses := newRevokedSession(determinePrivilegeSetOfStatement(stmt)) bh := &backgroundExecTest{} bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(3, 2)] = newMrsForCheckUserGrant(nil) bh.sql2result[getSqlForGetOwnerOfDatabase("owned_db")] = newMrsForOwner( [][]interface{}{{int64(3)}}) backgroundExecStub := gostub.StubFunc(&NewBackgroundExec, bh) @@ -14536,7 +14478,7 @@ func TestRevokedActiveRoleCannotUseAuthorizationFallbacks(t *testing.T) { context.Background(), ses, stmt) require.False(t, ok) require.Error(t, err) - require.Empty(t, bh.executedSQLs, "revoked role must not reach ownership fallback") + require.Equal(t, []string{"begin;", getSqlForCheckUserGrantForAuthorization(3, 2), "commit;"}, bh.executedSQLs, "revoked role must not reach ownership fallback") }) t.Run("privilege grant option", func(t *testing.T) { @@ -14548,6 +14490,7 @@ func TestRevokedActiveRoleCannotUseAuthorizationFallbacks(t *testing.T) { ses := newRevokedSession(determinePrivilegeSetOfStatement(stmt)) bh := &backgroundExecTest{} bh.init() + bh.sql2result[getSqlForCheckUserGrantForAuthorization(3, 2)] = newMrsForCheckUserGrant(nil) backgroundExecStub := gostub.StubFunc(&NewBackgroundExec, bh) defer backgroundExecStub.Reset() @@ -14555,7 +14498,7 @@ func TestRevokedActiveRoleCannotUseAuthorizationFallbacks(t *testing.T) { context.Background(), ses, stmt) require.False(t, ok) require.NoError(t, err) - require.Empty(t, bh.executedSQLs, "revoked role must not reach WITH GRANT OPTION traversal") + require.Equal(t, []string{"begin;", getSqlForCheckUserGrantForAuthorization(3, 2), "commit;"}, bh.executedSQLs, "revoked role must not reach WITH GRANT OPTION traversal") }) t.Run("role switch remains available", func(t *testing.T) { diff --git a/pkg/frontend/compiler_context.go b/pkg/frontend/compiler_context.go index da131398adb09..07f7b317497e6 100644 --- a/pkg/frontend/compiler_context.go +++ b/pkg/frontend/compiler_context.go @@ -188,6 +188,14 @@ func (tcc *TxnCompilerContext) InitExecuteStmtParam(execPlan *plan.Execute) (*pl execPlan, "", ) + if err == nil && !tcc.execCtx.ses.IsBackgroundSession() { + // EXPLAIN delegates EXECUTE binding here rather than through the ordinary + // Execute wrapper. Authorize the resolved AST before releasing ownership. + authStats, authErr := authenticateUserCanExecutePrepareOrExecute( + tcc.execCtx.reqCtx, owner, st, p, tcc.execCtx.effectiveTxnDefaultDatabase) + statistic.StatsInfoFromContext(tcc.execCtx.reqCtx).PermissionAuth.Add(&authStats) + err = authErr + } if owned && st != nil { st.Free() st = nil diff --git a/pkg/frontend/computation_wrapper.go b/pkg/frontend/computation_wrapper.go index 5d8adfe88160f..8f8ce818c56e1 100644 --- a/pkg/frontend/computation_wrapper.go +++ b/pkg/frontend/computation_wrapper.go @@ -1331,6 +1331,13 @@ func initExecuteStmtParamWithResolverInSession( if validateNamedSnapshots { change = true } + // A prepared EXPLAIN EXECUTE embeds another mutable prepared handle. Its + // cached query alone cannot prove the current AST/plan binding or grants. + if inner := unwrapExecutableExplainStatement(prepareStmt.PrepareStmt); inner != prepareStmt.PrepareStmt { + if _, execute := inner.(*tree.Execute); execute { + change = true + } + } rebuildEveryExecute := shouldRebuildPreparePlan(false, executionPlan) schemaChanged, schemasValidated, err := validateCapturedPrepareSchemas( owner.GetAccountId(), preparePlan.GetSchemas(), resolve, catalogCache, diff --git a/pkg/frontend/mysql_cmd_executor.go b/pkg/frontend/mysql_cmd_executor.go index 2254ebfbebc99..c9172b0362fba 100644 --- a/pkg/frontend/mysql_cmd_executor.go +++ b/pkg/frontend/mysql_cmd_executor.go @@ -1768,14 +1768,6 @@ func doSetVar( if cache != nil { cache.invalidate() } - // Clearing the cache is also the explicit synchronization point - // for externally changed role membership. Refresh it now, outside - // the caller's transaction snapshot, instead of allowing the next - // authorization check to repopulate the cache from stale state. - _, _, err = validateActiveRoleGrantForAuthorization(execCtx.reqCtx, ses) - if err != nil { - return err - } } err = setVarFunc(assign.System, assign.Global, name, value, sql) if err != nil { @@ -4860,12 +4852,6 @@ func authenticateUserCanExecuteStatement(reqCtx context.Context, ses *Session, s var stats statistic.StatsArray stats.Reset() - // Cache grants only within one statement. A session-local cache cannot - // observe REVOKE or RESTORE committed by another connection or another CN. - if cache := ses.GetPrivilegeCache(); cache != nil { - cache.invalidate() - } - reqCtx, span := trace.Debug(reqCtx, "authenticateUserCanExecuteStatement") defer span.End() if getPu(ses.GetService()).SV.SkipCheckPrivilege { @@ -4950,6 +4936,7 @@ func authenticateCanExecuteStatementAndPlan(reqCtx context.Context, ses *Session if ses.skipAuthForSpecialUser() { return stats, nil } + stmt = unwrapExecutableExplainStatement(stmt) yes, delta, err := authenticateUserCanExecuteStatementWithObjectTypeDatabaseAndTable(reqCtx, ses, stmt, p) if err != nil { return stats, err diff --git a/pkg/frontend/privilege_cache.go b/pkg/frontend/privilege_cache.go new file mode 100644 index 0000000000000..288fb2e0f3fae --- /dev/null +++ b/pkg/frontend/privilege_cache.go @@ -0,0 +1,62 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package frontend + +import ( + "context" + + "github.com/matrixorigin/matrixone/pkg/vm/engine" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae" +) + +func (ses *Session) refreshPrivilegeCache(ctx context.Context) error { + cache := ses.GetPrivilegeCache() + if cache == nil { + return nil + } + if enabled, err := privilegeCacheIsEnabled(ctx, ses); err != nil || !enabled { + cache.invalidate() + return err + } + pu := getPuIfPresent(ses.GetService()) + if pu == nil || ses.GetTenantInfo() == nil { + cache.invalidate() + return nil + } + storage := pu.StorageEngine + if wrapped, ok := storage.(*engine.EntireEngine); ok { + storage = wrapped.Engine + } + eng, ok := storage.(*disttae.Engine) + if !ok { + // Engines without a catalog-version proof retain statement-local grants. + cache.invalidate() + return nil + } + version, minimum, err := eng.GetPrivilegeCacheVersion( + ctx, ses.GetTenantInfo().GetTenantID(), ses.getLastCommitTS()) + if err != nil { + cache.invalidate() + return err + } + ses.updateLastCommitTS(minimum) + if version == (disttae.PrivilegeCacheVersion{}) || version != cache.catalogVersion { + cache.invalidate() + // Capture before evaluation. A concurrent later commit must invalidate + // the results next time, not be stamped onto an older authorization read. + cache.catalogVersion = version + } + return nil +} diff --git a/pkg/frontend/routine_test.go b/pkg/frontend/routine_test.go index 7a47e67a5d579..e118e908c4318 100644 --- a/pkg/frontend/routine_test.go +++ b/pkg/frontend/routine_test.go @@ -1128,7 +1128,7 @@ func TestMigrateConnectionFromMarksTypedSystemSnapshotTooLargeForLegacyReplay(t require.True(t, resp.SystemVariablesReplayable) } -func TestClearPrivilegeCacheRefreshesActiveRoleGrant(t *testing.T) { +func TestClearPrivilegeCacheDefersActiveRoleValidation(t *testing.T) { for _, tc := range []struct { name string catalogErr error @@ -1174,6 +1174,11 @@ func TestClearPrivilegeCacheRefreshesActiveRoleGrant(t *testing.T) { stmt, err := parsers.ParseOne(ctx, dialect.MYSQL, "set session clear_privilege_cache = on", 1) require.NoError(t, err) err = doSetVar(ses, newTestExecCtx(ctx, ctrl), stmt.(*tree.SetVar), "", false) + require.NoError(t, err) + require.Empty(t, bh.executedSQLs, "clearing must not read the catalog") + _, cached := ses.GetPrivilegeCache().getActiveRoleGrant(2, 3) + require.False(t, cached) + _, _, err = validateActiveRoleGrantForAuthorization(ctx, ses) require.True(t, forcedPessimisticRC) require.Contains(t, bh.executedSQLs, roleGrantSQL) if tc.catalogErr != nil { diff --git a/pkg/tests/issues/privilege_cache_test.go b/pkg/tests/issues/privilege_cache_test.go new file mode 100644 index 0000000000000..130db9c215670 --- /dev/null +++ b/pkg/tests/issues/privilege_cache_test.go @@ -0,0 +1,258 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package issues + +import ( + "context" + "database/sql" + "fmt" + "testing" + "time" + + "github.com/matrixorigin/matrixone/pkg/cnservice" + "github.com/matrixorigin/matrixone/pkg/embed" + "github.com/matrixorigin/matrixone/pkg/objectio" + "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + "github.com/matrixorigin/matrixone/pkg/util/fault" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae" + "github.com/stretchr/testify/require" +) + +func TestPrivilegeCacheTracksRemoteCatalogChanges(t *testing.T) { + runAuthenticatedClusterTest(t, func(c embed.Cluster) { + ctx, cancel := context.WithTimeout(t.Context(), 120*time.Second) + defer cancel() + open := func(index int, user string) *sql.Conn { + cn, err := c.GetCNService(index) + require.NoError(t, err) + db, err := sql.Open("mysql", fmt.Sprintf("%s:111@tcp(127.0.0.1:%d)/", user, cn.GetServiceConfig().CN.Frontend.Port)) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, db.Close()) }) + conn, err := db.Conn(ctx) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, conn.Close()) }) + return conn + } + admin := open(0, "dump") + mustExec(t, ctx, admin, "create account auth_cache admin_name 'admin' identified by '111'") + defer func() { + cleanup, stop := context.WithTimeout(context.Background(), 30*time.Second) + defer stop() + mustExec(t, cleanup, admin, "drop account auth_cache") + }() + owner := open(0, "auth_cache#admin#accountadmin") + for _, stmt := range []string{ + "create database app", "create table app.t(id int primary key)", "insert into app.t values (1)", + "create role reader", "create user learner identified by '111' default role reader", + "grant reader to learner", "grant connect on account * to reader", + "grant select on table app.t to reader", + } { + mustExec(t, ctx, owner, stmt) + } + // Each protocol owns a distinct session cache: one query must not refresh + // another protocol's cache and hide a missing execution-time check. + readers := []*sql.Conn{open(1, "auth_cache#learner#reader"), open(1, "auth_cache#learner#reader"), open(1, "auth_cache#learner#reader")} + allReadersExec := func(stmt string) { + t.Helper() + for _, reader := range readers { + mustExec(t, ctx, reader, stmt) + } + } + allReadersExec("set enable_privilege_cache=on") + mustExec(t, ctx, readers[1], "prepare cached_read from 'select id from app.t where id=1'") + prepared, err := readers[2].PrepareContext(ctx, "select id from app.t where id=?") + require.NoError(t, err) + defer prepared.Close() + check := func(allowed bool) { + t.Helper() + var got int + for i, query := range []string{"select id from app.t where id=1", "execute cached_read", "binary"} { + var row *sql.Row + if query == "binary" { + row = prepared.QueryRowContext(ctx, 1) + } else { + row = readers[i].QueryRowContext(ctx, query) + } + err := row.Scan(&got) + if allowed { + require.NoError(t, err) + require.Equal(t, 1, got) + } else { + require.ErrorContains(t, err, "privilege") + } + } + } + check(true) + check(true) + // A data write must not invalidate authorization catalog contents. + cn, err := c.GetCNService(1) + require.NoError(t, err) + eng := cn.RawService().(cnservice.Service).GetEngine().(*disttae.Engine) + var account uint32 + require.NoError(t, admin.QueryRowContext(ctx, "select account_id from mo_catalog.mo_account where account_name='auth_cache'").Scan(&account)) + version, _, err := eng.GetPrivilegeCacheVersion(ctx, account, timestamp.Timestamp{}) + require.NoError(t, err) + require.True(t, version != (disttae.PrivilegeCacheVersion{})) + mustExec(t, ctx, owner, "insert into app.t values (2)") + current, _, err := eng.GetPrivilegeCacheVersion(ctx, account, timestamp.Timestamp{}) + require.NoError(t, err) + require.True(t, version == current) + canceled, stop := context.WithCancel(ctx) + stop() + current, _, err = eng.GetPrivilegeCacheVersion(canceled, account, timestamp.Timestamp{}) + require.ErrorIs(t, err, context.Canceled) + require.True(t, current == (disttae.PrivilegeCacheVersion{})) + allReadersExec("begin") + check(true) + mustExec(t, ctx, owner, "revoke select on table app.t from reader") + check(false) + allReadersExec("rollback") + mustExec(t, ctx, owner, "grant select on table app.t to reader") + check(true) + mustExec(t, ctx, owner, "revoke reader from learner") + check(false) + mustExec(t, ctx, owner, "grant reader to learner") + check(true) + allReadersExec("set role public") + check(false) + allReadersExec("set role reader") + check(true) + mustExec(t, ctx, owner, "drop table app.t") + mustExec(t, ctx, owner, "create table app.t(id int primary key)") + mustExec(t, ctx, owner, "insert into app.t values (1)") + check(false) + mustExec(t, ctx, owner, "grant select on table app.t to reader") + check(true) + for _, setting := range []string{"set clear_privilege_cache=on", "set enable_privilege_cache=off", "set enable_privilege_cache=on"} { + allReadersExec(setting) + check(true) + } + consume := func(conn *sql.Conn, query string) error { + rows, err := conn.QueryContext(ctx, query) + if err != nil { + return err + } + defer rows.Close() + for rows.Next() { + } + return rows.Err() + } + t.Run("nested_authorization", func(t *testing.T) { + // Each shell has its own warm cache before a remote revoke. + for _, query := range []string{ + "set @nested_value=(select id from app.t)", + "explain select id from app.t", + "explain analyze select id from app.t", + "explain phyplan select id from app.t", + "explain analyze force execute nested_read", + "execute nested_set", + "execute nested_explain", + "binary set", + } { + t.Run(query, func(t *testing.T) { + conn := open(1, "auth_cache#learner#reader") + mustExec(t, ctx, conn, "set enable_privilege_cache=on") + var binary *sql.Stmt + switch query { + case "explain analyze force execute nested_read", "execute nested_explain": + mustExec(t, ctx, conn, "prepare nested_read from 'select id from app.t'") + if query == "execute nested_explain" { + mustExec(t, ctx, conn, "prepare nested_explain from 'explain analyze force execute nested_read'") + } + case "execute nested_set": + mustExec(t, ctx, conn, "prepare nested_set from 'set @nested_value=(select id from app.t)'") + case "binary set": + var err error + binary, err = conn.PrepareContext(ctx, "set @nested_value=(select id from app.t)") + require.NoError(t, err) + defer binary.Close() + } + run := func() error { + if binary != nil { + _, err := binary.ExecContext(ctx) + return err + } + return consume(conn, query) + } + require.NoError(t, run()) + mustExec(t, ctx, conn, "set @nested_value=-1") + mustExec(t, ctx, owner, "revoke select on table app.t from reader") + defer mustExec(t, ctx, owner, "grant select on table app.t to reader") + require.ErrorContains(t, run(), "privilege") + var value int + require.NoError(t, conn.QueryRowContext(ctx, "select @nested_value").Scan(&value)) + require.Equal(t, -1, value, "denied evaluation must not assign a value") + }) + } + conn := open(1, "auth_cache#learner#reader") + require.ErrorContains(t, consume(conn, "explain analyze update app.t set id=7"), "privilege") + var id int + require.NoError(t, owner.QueryRowContext(ctx, "select id from app.t").Scan(&id)) + require.Equal(t, 1, id) + }) + t.Run("grant_option_membership", func(t *testing.T) { + mustExec(t, ctx, owner, "create role recipient") + mustExec(t, ctx, owner, "grant select on table app.t to reader with grant option") + conn := open(1, "auth_cache#learner#reader") + const grant = "grant select on table app.t to recipient" + mustExec(t, ctx, conn, grant) + mustExec(t, ctx, owner, "revoke reader from learner") + require.ErrorContains(t, consume(conn, grant), "privilege") + mustExec(t, ctx, owner, "grant reader to learner") + mustExec(t, ctx, conn, grant) + }) + + t.Run("catalog_fault", func(t *testing.T) { + type control struct { + query string + conn *sql.Conn + } + var controls []control + for _, user := range []string{"learner#reader", "admin#accountadmin"} { + for _, query := range []string{"set clear_privilege_cache=on", "set enable_privilege_cache=off", "show warnings", "set clear_privilege_cache=1", "set @constant=1+1"} { + conn := open(1, "auth_cache#"+user) + mustExec(t, ctx, conn, "set enable_privilege_cache=on") + require.NoError(t, consume(conn, "select id from app.t")) + controls = append(controls, control{query, conn}) + } + } + check(true) + if fault.Enable() { + defer fault.Disable() + } + remove, err := objectio.InjectLogging(objectio.FJ_CNSubscribeTableFail, "mo_catalog", "mo_role_privs", 0, true) + require.NoError(t, err) + defer remove() + for i, tc := range controls { + t.Run(fmt.Sprintf("control_%d/%s", i, tc.query), func(t *testing.T) { + require.NoError(t, consume(tc.conn, tc.query)) + }) + } + // Actual grant consumers must fail closed despite a warm cache. + for i, query := range []string{"select id from app.t where id=1", "execute cached_read", "binary"} { + var id int + var err error + if query == "binary" { + err = prepared.QueryRowContext(ctx, 1).Scan(&id) + } else { + err = readers[i].QueryRowContext(ctx, query).Scan(&id) + } + require.ErrorContains(t, err, "injected subscribe table err", query) + } + }) + check(true) + + }) +} diff --git a/pkg/vm/engine/disttae/logtailreplay/partition_state.go b/pkg/vm/engine/disttae/logtailreplay/partition_state.go index 95f810966b7f5..ae091c8daf921 100644 --- a/pkg/vm/engine/disttae/logtailreplay/partition_state.go +++ b/pkg/vm/engine/disttae/logtailreplay/partition_state.go @@ -95,6 +95,22 @@ type PartitionState struct { shared *sharedStates } +// DataVersion identifies the contents of one subscribed partition without +// retaining its rows or objects. GC/copy preserve it; reconstruction does not. +type DataVersion struct { + generation *sharedStates + appliedTo types.TS +} + +func (p *PartitionState) DataVersion() DataVersion { + return DataVersion{generation: p.shared, appliedTo: p.appliedTo} +} + +// VisibleAt excludes mutations ahead of an exclusive transaction snapshot. +func (v DataVersion) VisibleAt(snapshot types.TS) bool { + return v.generation != nil && !v.appliedTo.IsEmpty() && v.appliedTo.LT(&snapshot) +} + // SourceCommitTS is the timestamp an async index watermark must cover at a // snapshot. It combines the partition-state retention boundary with user-data // commit timestamps; it deliberately does not use an ordinary TN object's diff --git a/pkg/vm/engine/disttae/logtailreplay/partition_state_test.go b/pkg/vm/engine/disttae/logtailreplay/partition_state_test.go index 1ef815a7302fe..42c91a9c07c58 100644 --- a/pkg/vm/engine/disttae/logtailreplay/partition_state_test.go +++ b/pkg/vm/engine/disttae/logtailreplay/partition_state_test.go @@ -79,6 +79,26 @@ func TestSourceCommitTSMax(t *testing.T) { require.Equal(t, types.BuildTS(30, 1), info.Max()) } +func TestPartitionDataVersion(t *testing.T) { + state := NewPartitionState("", false, 42, false) + require.False(t, state.DataVersion().VisibleAt(types.MaxTs()), "unapplied state is not a proof") + state.UpdateAppliedTo(types.BuildTS(10, 0)) + version := state.DataVersion() + require.False(t, version.VisibleAt(types.BuildTS(10, 0))) + require.True(t, version.VisibleAt(types.BuildTS(10, 1))) + + copy := state.Copy() + copy.UpdateDuration(types.BuildTS(9, 0), types.MaxTs()) + copy.UpdateAppliedTo(types.BuildTS(9, 0)) + require.Equal(t, version, copy.DataVersion(), "copy/GC and older delivery preserve the version") + copy.UpdateAppliedTo(types.BuildTS(11, 0)) + require.NotEqual(t, version, copy.DataVersion()) + require.Equal(t, version, state.DataVersion(), "published snapshots stay immutable") + rebuilt := NewPartitionState("", false, 42, false) + rebuilt.UpdateAppliedTo(types.BuildTS(10, 0)) + require.False(t, version == rebuilt.DataVersion(), "same watermark in another subscription is not the same catalog") +} + func TestSourceCommitTSAtIncludesPartitionStateStart(t *testing.T) { state := NewPartitionState("", false, 42, false) state.UpdateDuration(types.BuildTS(100, 1), types.MaxTs()) diff --git a/pkg/vm/engine/disttae/privilege_cache.go b/pkg/vm/engine/disttae/privilege_cache.go new file mode 100644 index 0000000000000..44ba5e80ed994 --- /dev/null +++ b/pkg/vm/engine/disttae/privilege_cache.go @@ -0,0 +1,102 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package disttae + +import ( + "context" + + "github.com/matrixorigin/matrixone/pkg/catalog" + "github.com/matrixorigin/matrixone/pkg/container/types" + "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + "github.com/matrixorigin/matrixone/pkg/txn/client" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay" +) + +// PrivilegeCacheVersion covers both grant contents and physical catalog/object +// identity. It holds only small generation identities, never partition trees. +// The zero version is deliberately not a reusable authorization proof. +type PrivilegeCacheVersion struct { + account uint32 + tables [7]struct { + id uint64 + version logtailreplay.DataVersion + } +} + +// GetPrivilegeCacheVersion uses the same snapshot admission as a fresh catalog +// transaction. It does not consult the caller's possibly older data snapshot. +// No permission decisions are made here: frontend remains their sole owner. +func (e *Engine) GetPrivilegeCacheVersion( + ctx context.Context, account uint32, minimum timestamp.Timestamp, +) (PrivilegeCacheVersion, timestamp.Timestamp, error) { + var version PrivilegeCacheVersion + if !e.pClient.receivedLogTailTime.ready.Load() { + return version, minimum, nil + } + op, err := e.cli.New(ctx, minimum, client.WithTxnReadyOnly()) + if err != nil { + return version, minimum, err + } + snapshot := op.Txn().SnapshotTS + // This transaction owns no workspace or writes. Read-only Commit only + // releases its admission; it cannot send a storage commit or take locks. + if err = op.Commit(ctx); err != nil { + return version, minimum, err + } + minimum = snapshot.Prev() + if !e.pClient.receivedLogTailTime.ready.Load() { + return version, minimum, nil + } + + cc := e.GetLatestCatalogCache() + if cc == nil { + return version, minimum, nil + } + version.account = account + for i, name := range [...]string{ + catalog.MO_DATABASE, catalog.MO_TABLES, + "mo_user", "mo_role", "mo_user_grant", "mo_role_grant", "mo_role_privs", + } { + id := uint64(catalog.MO_DATABASE_ID) + partitionAccount := uint64(0) + switch i { + case 0: + case 1: + id = catalog.MO_TABLES_ID + default: + table := cc.GetTableByName(account, catalog.MO_CATALOG_ID, name) + if table == nil || !table.Ts.Less(snapshot) { + return PrivilegeCacheVersion{}, minimum, nil + } + id = table.Id + partitionAccount = uint64(account) + } + var pending bool + ps, err := e.pClient.toSubscribeTable(ctx, partitionAccount, id, name, + catalog.MO_CATALOG_ID, catalog.MO_CATALOG, &pending) + if err != nil { + return PrivilegeCacheVersion{}, minimum, err + } + if pending || ps == nil || !ps.DataVersion().VisibleAt(types.TimestampToTS(snapshot)) { + return PrivilegeCacheVersion{}, minimum, nil + } + version.tables[i].id = id + version.tables[i].version = ps.DataVersion() + } + if !e.pClient.receivedLogTailTime.ready.Load() || e.GetLatestCatalogCache() != cc { + return PrivilegeCacheVersion{}, minimum, nil + } + return version, minimum, nil +} diff --git a/pkg/vm/engine/disttae/privilege_cache_test.go b/pkg/vm/engine/disttae/privilege_cache_test.go new file mode 100644 index 0000000000000..73242aee899d7 --- /dev/null +++ b/pkg/vm/engine/disttae/privilege_cache_test.go @@ -0,0 +1,122 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package disttae + +import ( + "context" + "errors" + "testing" + + "github.com/golang/mock/gomock" + "github.com/matrixorigin/matrixone/pkg/catalog" + "github.com/matrixorigin/matrixone/pkg/container/types" + mock_frontend "github.com/matrixorigin/matrixone/pkg/frontend/test" + "github.com/matrixorigin/matrixone/pkg/pb/timestamp" + "github.com/matrixorigin/matrixone/pkg/pb/txn" + "github.com/matrixorigin/matrixone/pkg/testutil" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/cache" + "github.com/matrixorigin/matrixone/pkg/vm/engine/disttae/logtailreplay" + "github.com/stretchr/testify/require" +) + +func TestPrivilegeCacheVersionRejectsUnprovenCatalog(t *testing.T) { + ctx := context.Background() + snapshot := timestamp.Timestamp{PhysicalTime: 20} + ctrl := gomock.NewController(t) + cli := mock_frontend.NewMockTxnClient(ctrl) + op := mock_frontend.NewMockTxnOperator(ctrl) + cli.EXPECT().New(ctx, timestamp.Timestamp{}, gomock.Any()).Return(op, nil).AnyTimes() + op.EXPECT().Txn().Return(txn.TxnMeta{SnapshotTS: snapshot}).AnyTimes() + op.EXPECT().Commit(ctx).Return(nil).AnyTimes() + e := &Engine{cli: cli, partitions: make(map[[2]uint64]*logtailreplay.Partition)} + e.catalog.Store(cache.NewCatalog()) + e.pClient.eng = e + e.pClient.receivedLogTailTime.ready.Store(true) + e.pClient.subscribed = subscribedTable{eng: e, m: make(map[uint64]*subEntry)} + workspace := &Transaction{proc: testutil.NewProc(t)} + defer workspace.proc.Free() + for i, name := range []string{catalog.MO_DATABASE, catalog.MO_TABLES, "mo_user", "mo_role", "mo_user_grant", "mo_role_grant", "mo_role_privs"} { + id := uint64(i + 1) + account := uint32(0) + if i >= 2 { + id += 10 + account = 7 + insertCatalogTableAt(t, e, workspace, account, catalog.MO_CATALOG_ID, id, 1, catalog.MO_CATALOG, name, types.BuildTS(5, 0)) + } + e.pClient.subscribed.m[id] = &subEntry{dbID: catalog.MO_CATALOG_ID, state: Subscribed} + state, done := e.GetOrCreateLatestPart(ctx, uint64(account), catalog.MO_CATALOG_ID, id).MutateState() + state.UpdateDuration(types.TS{}, types.MaxTs()) + state.UpdateAppliedTo(types.BuildTS(10, 0)) + done() + } + read := func() PrivilegeCacheVersion { + t.Helper() + version, minimum, err := e.GetPrivilegeCacheVersion(ctx, 7, timestamp.Timestamp{}) + require.NoError(t, err) + require.Equal(t, snapshot.Prev(), minimum) + return version + } + original := read() + require.True(t, original != (PrivilegeCacheVersion{})) + require.True(t, original == read()) + e.pClient.subscribed.setTablePendingUpdate(1, 1, timestamp.Timestamp{PhysicalTime: 15}) + require.True(t, read() == (PrivilegeCacheVersion{}), "pending catalog application must not authorize reuse") + e.pClient.subscribed.clearTablePendingUpdate(1, 1, timestamp.Timestamp{PhysicalTime: 15}) + require.True(t, original == read()) + e.pClient.receivedLogTailTime.ready.Store(false) + unready, observed, err := e.GetPrivilegeCacheVersion(ctx, 7, snapshot) + require.NoError(t, err) + require.Equal(t, snapshot, observed) + require.True(t, unready == (PrivilegeCacheVersion{}), "reconnect admission is closed") + e.pClient.receivedLogTailTime.ready.Store(true) + delete(e.partitions, [2]uint64{1, 1}) + state, done := e.GetOrCreateLatestPart(ctx, 0, 1, 1).MutateState() + state.UpdateDuration(types.TS{}, types.MaxTs()) + state.UpdateAppliedTo(types.BuildTS(10, 0)) + done() + rebuilt := read() + require.True(t, rebuilt != (PrivilegeCacheVersion{})) + require.True(t, original != rebuilt, "reconstruction at the same watermark changes identity") + state, done = e.GetOrCreateLatestPart(ctx, 0, 1, 1).MutateState() + state.UpdateAppliedTo(types.TimestampToTS(snapshot)) + done() + require.True(t, read() == (PrivilegeCacheVersion{}), "a future catalog cannot prove an older snapshot") +} + +func TestPrivilegeCacheVersionPreservesSnapshotErrors(t *testing.T) { + for _, atCommit := range []bool{false, true} { + t.Run(map[bool]string{false: "admission", true: "release"}[atCommit], func(t *testing.T) { + ctx := context.Background() + ctrl := gomock.NewController(t) + cli := mock_frontend.NewMockTxnClient(ctrl) + failed := errors.New("snapshot unavailable") + minimum := timestamp.Timestamp{PhysicalTime: 10} + if atCommit { + op := mock_frontend.NewMockTxnOperator(ctrl) + cli.EXPECT().New(ctx, minimum, gomock.Any()).Return(op, nil) + op.EXPECT().Txn().Return(txn.TxnMeta{SnapshotTS: minimum.Next()}) + op.EXPECT().Commit(ctx).Return(failed) + } else { + cli.EXPECT().New(ctx, minimum, gomock.Any()).Return(nil, failed) + } + eng := &Engine{cli: cli} + eng.pClient.receivedLogTailTime.ready.Store(true) + version, observed, err := eng.GetPrivilegeCacheVersion(ctx, 7, minimum) + require.ErrorIs(t, err, failed) + require.Equal(t, minimum, observed) + require.True(t, version == (PrivilegeCacheVersion{})) + }) + } +}