// Licensed to the Apache Software Foundation (ASF) under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you 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 casbin import ( "errors" "sync/atomic" "github.com/casbin/casbin/v3/persist" ) // Commit commits the transaction using a two-phase commit protocol. // Phase 1: Apply all operations to the database // Phase 2: Apply changes to the in-memory model and rebuild role links. func (tx *Transaction) Commit() error { // Try to acquire the commit lock with timeout. if !tryLockWithTimeout(&tx.enforcer.commitLock, tx.startTime, defaultLockTimeout) { _ = tx.txContext.Rollback() tx.enforcer.activeTransactions.Delete(tx.id) return errors.New("transaction timeout: failed to acquire lock") } defer tx.enforcer.commitLock.Unlock() tx.mutex.Lock() defer tx.mutex.Unlock() if tx.committed { return errors.New("transaction already committed") } if tx.rolledBack { return errors.New("transaction already rolled back") } // First check if model version has changed. currentVersion := atomic.LoadInt64(&tx.enforcer.modelVersion) if currentVersion != tx.baseVersion { // Model has been modified, need to check for conflicts. detector := NewConflictDetector( tx.buffer.GetModelSnapshot(), tx.enforcer.model, tx.buffer.GetOperations(), ) if err := detector.DetectConflicts(); err != nil { _ = tx.txContext.Rollback() tx.enforcer.activeTransactions.Delete(tx.id) return err } } // If no operations, just commit the database transaction and clear state. if !tx.buffer.HasOperations() { if err := tx.txContext.Commit(); err != nil { return err } tx.committed = true tx.enforcer.activeTransactions.Delete(tx.id) return nil } // Phase 1: Apply all buffered operations to the database if err := tx.applyOperationsToDatabase(); err != nil { // Rollback database transaction on failure. _ = tx.txContext.Rollback() tx.enforcer.activeTransactions.Delete(tx.id) return err } // Commit database transaction. if err := tx.txContext.Commit(); err != nil { tx.enforcer.activeTransactions.Delete(tx.id) return err } // Phase 2: Apply changes to the in-memory model if err := tx.applyOperationsToModel(); err != nil { // At this point, database is committed but model update failed. // This is a critical error that should not happen in normal circumstances. tx.enforcer.activeTransactions.Delete(tx.id) return errors.New("critical error: database committed but model update failed: " + err.Error()) } // Increment model version number. atomic.AddInt64(&tx.enforcer.modelVersion, 1) tx.committed = true tx.enforcer.activeTransactions.Delete(tx.id) return nil } // Rollback rolls back the transaction. // This will rollback the database transaction and clear the transaction state. func (tx *Transaction) Rollback() error { // Try to acquire the commit lock with timeout. if !tryLockWithTimeout(&tx.enforcer.commitLock, tx.startTime, defaultLockTimeout) { tx.enforcer.activeTransactions.Delete(tx.id) return errors.New("transaction timeout: failed to acquire lock for rollback") } defer tx.enforcer.commitLock.Unlock() tx.mutex.Lock() defer tx.mutex.Unlock() if tx.committed { return errors.New("transaction already committed") } if tx.rolledBack { return errors.New("transaction already rolled back") } // Rollback database transaction. if err := tx.txContext.Rollback(); err != nil { return err } tx.rolledBack = true tx.enforcer.activeTransactions.Delete(tx.id) return nil } // applyOperationsToDatabase applies all buffered operations to the database. func (tx *Transaction) applyOperationsToDatabase() error { operations := tx.buffer.GetOperations() txAdapter := tx.txContext.GetAdapter() for _, op := range operations { switch op.Type { case persist.OperationAdd: if err := tx.applyAddOperationToDatabase(txAdapter, op); err != nil { return err } case persist.OperationRemove: if err := tx.applyRemoveOperationToDatabase(txAdapter, op); err != nil { return err } case persist.OperationUpdate: if err := tx.applyUpdateOperationToDatabase(txAdapter, op); err != nil { return err } } } return nil } // applyAddOperationToDatabase applies an add operation to the database. func (tx *Transaction) applyAddOperationToDatabase(adapter persist.Adapter, op persist.PolicyOperation) error { if batchAdapter, ok := adapter.(persist.BatchAdapter); ok { // Use batch operation if available. return batchAdapter.AddPolicies(op.Section, op.PolicyType, op.Rules) } else { // Fall back to individual operations. for _, rule := range op.Rules { if err := adapter.AddPolicy(op.Section, op.PolicyType, rule); err != nil { return err } } } return nil } // applyRemoveOperationToDatabase applies a remove operation to the database. func (tx *Transaction) applyRemoveOperationToDatabase(adapter persist.Adapter, op persist.PolicyOperation) error { if batchAdapter, ok := adapter.(persist.BatchAdapter); ok { // Use batch operation if available. return batchAdapter.RemovePolicies(op.Section, op.PolicyType, op.Rules) } else { // Fall back to individual operations. for _, rule := range op.Rules { if err := adapter.RemovePolicy(op.Section, op.PolicyType, rule); err != nil { return err } } } return nil } // applyUpdateOperationToDatabase applies an update operation to the database. func (tx *Transaction) applyUpdateOperationToDatabase(adapter persist.Adapter, op persist.PolicyOperation) error { if updateAdapter, ok := adapter.(persist.UpdatableAdapter); ok { // Use update operation if available. return updateAdapter.UpdatePolicies(op.Section, op.PolicyType, op.OldRules, op.Rules) } // Fall back to remove + add. for i, oldRule := range op.OldRules { if err := adapter.RemovePolicy(op.Section, op.PolicyType, oldRule); err != nil { return err } if err := adapter.AddPolicy(op.Section, op.PolicyType, op.Rules[i]); err != nil { return err } } return nil } // applyOperationsToModel applies all buffered operations to the in-memory model. func (tx *Transaction) applyOperationsToModel() error { // Create new model with all operations applied. newModel, err := tx.buffer.ApplyOperationsToModel(tx.buffer.GetModelSnapshot()) if err != nil { return err } // Replace the enforcer's model. tx.enforcer.model = newModel tx.enforcer.invalidateMatcherMap() // Rebuild role links if necessary. if tx.enforcer.autoBuildRoleLinks { // Check if any operations involved grouping policies. operations := tx.buffer.GetOperations() needRoleRebuild := false for _, op := range operations { if op.Section == "g" { needRoleRebuild = true break } } if needRoleRebuild { if err := tx.enforcer.BuildRoleLinks(); err != nil { return err } } } return nil } // IsCommitted returns true if the transaction has been committed. func (tx *Transaction) IsCommitted() bool { tx.mutex.RLock() defer tx.mutex.RUnlock() return tx.committed } // IsRolledBack returns true if the transaction has been rolled back. func (tx *Transaction) IsRolledBack() bool { tx.mutex.RLock() defer tx.mutex.RUnlock() return tx.rolledBack } // IsActive returns true if the transaction is still active (not committed or rolled back). func (tx *Transaction) IsActive() bool { tx.mutex.RLock() defer tx.mutex.RUnlock() return !tx.committed && !tx.rolledBack }