This is an automated email from the ASF dual-hosted git repository.
hsluoyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/casbin-mongodb-adapter.git
The following commit(s) were added to refs/heads/master by this push:
new d6bfd5b fix: make SavePolicy atomic and handle empty policies
d6bfd5b is described below
commit d6bfd5b8191326300dfd09e5900a4a73947423df
Author: Yang Luo <[email protected]>
AuthorDate: Wed Aug 5 22:16:33 2026 +0800
fix: make SavePolicy atomic and handle empty policies
---
adapter.go | 76 ++++++++++++++++++++++++++++++++++++---------
adapter_test.go | 96 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++
2 files changed, 158 insertions(+), 14 deletions(-)
diff --git a/adapter.go b/adapter.go
index 37982a1..975ed58 100644
--- a/adapter.go
+++ b/adapter.go
@@ -226,17 +226,6 @@ func (a *adapter) close() {
_ = a.client.Disconnect(ctx)
}
-func (a *adapter) dropTable() error {
- ctx, cancel := context.WithTimeout(context.TODO(), a.timeout)
- defer cancel()
-
- _, err := a.collection.DeleteMany(ctx, bson.D{{}})
- if err != nil {
- return err
- }
- return nil
-}
-
func loadPolicyLine(line CasbinRule, model model.Model) error {
var p = []string{line.PType,
line.V0, line.V1, line.V2, line.V3, line.V4, line.V5}
@@ -325,13 +314,15 @@ func savePolicyLine(ptype string, rule []string)
CasbinRule {
}
// SavePolicy saves policy to database.
+//
+// The stored policy is replaced as a whole. The rules are collected from the
+// model before the collection is touched, and the delete and the insert run
+// inside a transaction when the server supports one, so that a failure can
+// never leave the stored policy erased or half written.
func (a *adapter) SavePolicy(model model.Model) error {
if a.filtered {
return errors.New("cannot save a filtered policy")
}
- if err := a.dropTable(); err != nil {
- return err
- }
var lines []interface{}
@@ -348,9 +339,66 @@ func (a *adapter) SavePolicy(model model.Model) error {
lines = append(lines, &line)
}
}
+
+ err := a.savePolicyTxn(lines)
+ if err == nil {
+ return nil
+ }
+ // (IllegalOperation) Transaction numbers are only allowed on a replica
set member or mongos
+ if mongoErr, ok := err.(mongo.CommandError); !ok || mongoErr.Code != 20
{
+ return err
+ }
+
+ log.Println("[WARNING]: As your mongodb server doesn't allow a replica
set, transaction operation is not supported. So Casbin Adapter will run
non-transactional saving!")
+ return a.savePolicy(lines)
+}
+
+// savePolicyTxn replaces the stored policy inside a transaction, so the old
+// rules are only dropped if the new ones can be written in the same unit.
+func (a *adapter) savePolicyTxn(lines []interface{}) error {
+ ctx, cancel := context.WithTimeout(context.TODO(), a.timeout)
+ defer cancel()
+
+ session, err := a.client.StartSession()
+ if err != nil {
+ return err
+ }
+ defer session.EndSession(context.TODO())
+
+ _, err = session.WithTransaction(ctx, func(sessionCtx context.Context)
(interface{}, error) {
+ return nil, a.replacePolicy(sessionCtx, lines)
+ })
+
+ return err
+}
+
+// savePolicy replaces the stored policy without a transaction. It is the
+// fallback for servers that are neither a replica set member nor a mongos.
+func (a *adapter) savePolicy(lines []interface{}) error {
ctx, cancel := context.WithTimeout(context.TODO(), a.timeout)
defer cancel()
+ return a.replacePolicy(ctx, lines)
+}
+
+// replacePolicy deletes every stored rule and writes the given ones in their
+// place. The caller owns the context so that it can be run inside a session.
+func (a *adapter) replacePolicy(ctx context.Context, lines []interface{})
error {
+ res, err := a.collection.DeleteMany(ctx, bson.D{})
+ if err != nil {
+ return err
+ }
+
+ // Saving a model that holds no rules is a legitimate way to clear the
+ // store, but InsertMany rejects an empty slice before it ever reaches
the
+ // server, so the delete above is all the work there is left to do.
+ if len(lines) == 0 {
+ if res.DeletedCount > 0 {
+ log.Printf("[WARNING]: SavePolicy was called with an
empty policy, %d stored rule(s) were deleted.", res.DeletedCount)
+ }
+ return nil
+ }
+
if _, err := a.collection.InsertMany(ctx, lines); err != nil {
return err
}
diff --git a/adapter_test.go b/adapter_test.go
index 0e6c046..f15ab40 100644
--- a/adapter_test.go
+++ b/adapter_test.go
@@ -636,6 +636,102 @@ func TestUpdateFilteredPoliciesTxn(t *testing.T) {
testGetPolicyWithoutOrder(t, e, [][]string{{"alice", "data1", "write"},
{"bob", "data2", "read"}, {"data2_admin", "data2", "read"}, {"data2_admin",
"data2", "write"}})
}
+// testSavePolicyEmpty checks that clearing every rule and saving succeeds and
+// leaves the collection empty, instead of failing on an empty InsertMany.
+func testSavePolicyEmpty(t *testing.T, dbURL string) {
+ t.Helper()
+
+ initPolicy(t, dbURL)
+
+ a, err := NewAdapter(dbURL)
+ if err != nil {
+ panic(err)
+ }
+
+ e, err := casbin.NewEnforcer("examples/rbac_model.conf", a)
+ if err != nil {
+ panic(err)
+ }
+
+ e.ClearPolicy()
+ if err := e.SavePolicy(); err != nil {
+ t.Fatalf("Expected SavePolicy() of an empty policy to be
successful; got %v", err)
+ }
+
+ adapter := a.(*adapter)
+
+ ctx, cancel := context.WithTimeout(context.TODO(), adapter.timeout)
+ defer cancel()
+
+ count, err := adapter.collection.CountDocuments(ctx, bson.D{})
+ if err != nil {
+ panic(err)
+ }
+ if count != 0 {
+ t.Errorf("Expected the collection to be empty after saving an
empty policy; got %d rule(s)", count)
+ }
+
+ // The empty policy must also survive a round trip through the database.
+ if err := e.LoadPolicy(); err != nil {
+ t.Errorf("Expected LoadPolicy() to be successful; got %v", err)
+ }
+ testGetPolicy(t, e, [][]string{})
+
+ // Saving a non-empty policy again must still work after the collection
was
+ // emptied, so the fix cannot leave the adapter in a dead state.
+ if _, err := e.AddPolicy("alice", "data1", "read"); err != nil {
+ t.Errorf("Expected AddPolicy() to be successful; got %v", err)
+ }
+ if err := e.SavePolicy(); err != nil {
+ t.Errorf("Expected SavePolicy() to be successful; got %v", err)
+ }
+ if err := e.LoadPolicy(); err != nil {
+ t.Errorf("Expected LoadPolicy() to be successful; got %v", err)
+ }
+ testGetPolicy(t, e, [][]string{{"alice", "data1", "read"}})
+}
+
+func TestSavePolicyEmpty(t *testing.T) {
+ testSavePolicyEmpty(t, getDbURL())
+}
+
+func TestSavePolicyEmptyTxn(t *testing.T) {
+ testSavePolicyEmpty(t, getReplicaSetURL())
+}
+
+// TestSavePolicyAtomic checks that a failing save leaves the previously stored
+// policy intact, rather than wiping it on the way to an error.
+func TestSavePolicyAtomic(t *testing.T) {
+ initPolicy(t, getReplicaSetURL())
+
+ a, err := NewAdapter(getReplicaSetURL())
+ if err != nil {
+ panic(err)
+ }
+
+ adapter := a.(*adapter)
+
+ // Two identical rules violate the unique index, so the insert fails
after
+ // the delete has already run inside the same transaction.
+ duplicate := savePolicyLine("p", []string{"alice", "data1", "read"})
+ lines := []interface{}{&duplicate, &duplicate}
+
+ if err := adapter.savePolicyTxn(lines); err == nil {
+ t.Fatal("Expected the save to fail on the unique index, but it
succeeded")
+ }
+
+ ctx, cancel := context.WithTimeout(context.TODO(), adapter.timeout)
+ defer cancel()
+
+ count, err := adapter.collection.CountDocuments(ctx, bson.D{})
+ if err != nil {
+ panic(err)
+ }
+ if count != 4 {
+ t.Errorf("Expected the 4 stored rules to survive a failed save;
got %d rule(s)", count)
+ }
+}
+
func TestSavePolicyPreservesIndexes(t *testing.T) {
// Initialize with some policies
initPolicy(t, getDbURL())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
