Fix MongoDB session lifecycle management in synced adapter

README.md

@@ -70,6 +70,69 @@ const { MongooseAdapter } = require('casbin-mongoose-adapter');
 const adapter = await MongooseAdapter.newFilteredAdapter('mongodb://your_mongodb_uri:27017');
 ```
 
+## Synced Adapter (Transactions)
+
+The Synced Adapter provides transaction support for MongoDB replica sets. This is useful when you need to ensure that multiple policy changes are committed atomically.
+
+**Note:** Transactions require a MongoDB replica set. They will not work with a standalone MongoDB instance.
+
+```javascript
+const { MongooseAdapter } = require('casbin-mongoose-adapter');
+
+// Create a synced adapter with automatic commit and abort
+const adapter = await MongooseAdapter.newSyncedAdapter(
+  'mongodb://your_mongodb_uri:27017?replicaSet=rs0',
+  {},
+  true,  // autoAbort - automatically abort transaction on error
+  true   // autoCommit - automatically commit transaction after each operation
+);
+```
+
+### Manual Transaction Control
+
+You can manually control transactions for more complex scenarios:
+
+```javascript
+const adapter = await MongooseAdapter.newSyncedAdapter(
+  'mongodb://your_mongodb_uri:27017?replicaSet=rs0',
+  {},
+  false,  // autoAbort
+  false   // autoCommit
+);
+
+// Get the current session to perform custom operations
+const session = await adapter.getSession();
+
+// Start a transaction
+await adapter.getTransaction();
+
+// Perform operations...
+await adapter.addPolicy('', 'p', ['alice', 'data1', 'read']);
+await adapter.addPolicy('', 'p', ['bob', 'data2', 'write']);
+
+// You can also perform custom operations with the session
+const CasbinRule = adapter.getCasbinRule();
+await CasbinRule.collection.insertOne(
+  { ptype: 'p', v0: 'charlie', v1: 'data3', v2: 'write' },
+  { session }
+);
+
+// Commit the transaction
+await adapter.commitTransaction();
+
+// Or abort if something went wrong
+// await adapter.abortTransaction();
+```
+
+### Session Lifecycle
+
+The adapter properly manages MongoDB session lifecycle:
+
+- Sessions are automatically created when starting a transaction
+- Sessions are automatically ended after `commitTransaction()` or `abortTransaction()`
+- In autoCommit/autoAbort mode, sessions are ended after each operation
+- You can access the current session using `getSession()` to perform custom operations
+
 ## License
 
 [Apache-2.0](./LICENSE)

examples/session-management.js

@@ -0,0 +1,134 @@
+// Example: Using Session Management with newSyncedAdapter
+// This example demonstrates proper session management for MongoDB transactions
+
+const { MongooseAdapter } = require('casbin-mongoose-adapter');
+
+async function exampleManualTransactionControl () {
+  console.log('Example 1: Manual Transaction Control');
+  console.log('=====================================\n');
+
+  // Create a synced adapter without autoCommit/autoAbort
+  const adapter = await MongooseAdapter.newSyncedAdapter(
+    'mongodb://localhost:27001,localhost:27002/casbin?replicaSet=rs0',
+    {},
+    false, // autoAbort
+    false // autoCommit
+  );
+
+  try {
+    // Get the current session
+    const session = await adapter.getSession();
+    console.log('Session created:', session.id);
+
+    // Start a transaction
+    await adapter.getTransaction();
+    console.log('Transaction started');
+
+    // Perform multiple operations
+    await adapter.addPolicy('', 'p', ['alice', 'data1', 'read']);
+    await adapter.addPolicy('', 'p', ['bob', 'data2', 'write']);
+    console.log('Policies added');
+
+    // You can also perform custom operations with the session
+    const CasbinRule = adapter.getCasbinRule();
+    await CasbinRule.collection.insertOne(
+      { ptype: 'p', v0: 'charlie', v1: 'data3', v2: 'write' },
+      { session }
+    );
+    console.log('Custom operation performed');
+
+    // Commit the transaction
+    await adapter.commitTransaction();
+    console.log('Transaction committed and session ended\n');
+  } catch (error) {
+    console.error('Error occurred:', error.message);
+    // Abort the transaction on error
+    await adapter.abortTransaction();
+    console.log('Transaction aborted and session ended\n');
+  } finally {
+    await adapter.close();
+  }
+}
+
+async function exampleAutoCommit () {
+  console.log('Example 2: Auto Commit Mode');
+  console.log('============================\n');
+
+  // Create a synced adapter with autoCommit enabled
+  const adapter = await MongooseAdapter.newSyncedAdapter(
+    'mongodb://localhost:27001,localhost:27002/casbin?replicaSet=rs0',
+    {},
+    true, // autoAbort
+    true // autoCommit
+  );
+
+  try {
+    // Each operation automatically commits and ends the session
+    await adapter.addPolicy('', 'p', ['alice', 'data1', 'read']);
+    console.log('Policy added and automatically committed');
+
+    await adapter.addPolicy('', 'p', ['bob', 'data2', 'write']);
+    console.log('Policy added and automatically committed\n');
+  } catch (error) {
+    console.error('Error occurred:', error.message);
+    console.log('Transaction automatically aborted\n');
+  } finally {
+    await adapter.close();
+  }
+}
+
+async function exampleSessionReuse () {
+  console.log('Example 3: Session Lifecycle');
+  console.log('=============================\n');
+
+  const adapter = await MongooseAdapter.newSyncedAdapter(
+    'mongodb://localhost:27001,localhost:27002/casbin?replicaSet=rs0',
+    {},
+    false,
+    false
+  );
+
+  try {
+    // First transaction
+    const session1 = await adapter.getTransaction();
+    console.log('First session created:', session1.id);
+
+    await adapter.addPolicy('', 'p', ['alice', 'data1', 'read']);
+    await adapter.commitTransaction();
+    console.log('First transaction committed, session ended');
+
+    // Second transaction - creates a new session
+    const session2 = await adapter.getTransaction();
+    console.log('Second session created:', session2.id);
+    console.log('Sessions are different:', session1.id !== session2.id);
+
+    await adapter.addPolicy('', 'p', ['bob', 'data2', 'write']);
+    await adapter.commitTransaction();
+    console.log('Second transaction committed, session ended\n');
+  } catch (error) {
+    console.error('Error occurred:', error.message);
+  } finally {
+    await adapter.close();
+  }
+}
+
+// Run examples - uncomment to execute
+// async function main () {
+//   try {
+//     await exampleManualTransactionControl();
+//     await exampleAutoCommit();
+//     await exampleSessionReuse();
+//   } catch (error) {
+//     console.error('Failed to run examples:', error.message);
+//     console.log('\nNote: These examples require a MongoDB replica set running at:');
+//     console.log('mongodb://localhost:27001,localhost:27002/casbin?replicaSet=rs0');
+//   }
+// }
+//
+// main();
+
+module.exports = {
+  exampleManualTransactionControl,
+  exampleAutoCommit,
+  exampleSessionReuse
+};

src/adapter.ts

@@ -60,7 +60,7 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
   private options?: ConnectOptions;
   private autoAbort: boolean;
   private autoCommit: boolean;
-  private session: ClientSession;
+  private session?: ClientSession;
   private casbinRule: MongooseModel<IModel>;
 
   /**
@@ -227,9 +227,11 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
    */
   async getSession(): Promise<ClientSession> {
     if (this.isSynced) {
-      return this.session && this.session.inTransaction()
-        ? this.session
-        : this.connection!.startSession();
+      if (this.session && this.session.inTransaction()) {
+        return this.session;
+      }
+      this.session = await this.connection!.startSession();
+      return this.session;
     } else
       throw new InvalidAdapterTypeError(
         'Transactions are only supported by SyncedAdapter. See newSyncedAdapter'
@@ -280,6 +282,8 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
     if (this.isSynced) {
       const session = await this.getSession();
       await session.commitTransaction();
+      await session.endSession();
+      this.session = undefined;
     } else
       throw new InvalidAdapterTypeError(
         'Transactions are only supported by SyncedAdapter. See newSyncedAdapter'
@@ -295,6 +299,8 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
     if (this.isSynced) {
       const session = await this.getSession();
       await session.abortTransaction();
+      await session.endSession();
+      this.session = undefined;
       logPrint('Transaction aborted');
     } else
       throw new InvalidAdapterTypeError(
@@ -355,7 +361,11 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
 
     const lines = await this.casbinRule.find(filter || {}, null, options).lean();
 
-    this.autoCommit && options.session && (await options.session.commitTransaction());
+    if (this.autoCommit && options.session) {
+      await options.session.commitTransaction();
+      await options.session.endSession();
+      this.session = undefined;
+    }
     for (const line of lines) {
       this.loadPolicyLine(line, model);
     }
@@ -434,9 +444,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
 
       await this.casbinRule.collection.insertMany(lines, options);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       console.error(err);
       return false;
     }
@@ -461,9 +479,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
       const line = this.savePolicyLine(pType, rule);
       await line.save(options);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }
@@ -488,9 +514,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
       const promises = rules.map(async (rule) => this.addPolicy(sec, pType, rule));
       await Promise.all(promises);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }
@@ -528,9 +562,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
 
       await this.casbinRule.updateOne({ ptype, v0, v1, v2, v3, v4, v5 }, newModel, options);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }
@@ -553,9 +595,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
 
       await this.casbinRule.deleteMany({ ptype, v0, v1, v2, v3, v4, v5 }, options);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }
@@ -581,9 +631,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
       const promises = rules.map(async (rule) => this.removePolicy(sec, pType, rule));
       await Promise.all(promises);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }
@@ -639,9 +697,17 @@ export class MongooseAdapter implements BatchAdapter, FilteredAdapter, Updatable
 
       await this.casbinRule.deleteMany(where, options);
 
-      this.autoCommit && options.session && (await options.session.commitTransaction());
+      if (this.autoCommit && options.session) {
+        await options.session.commitTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
     } catch (err) {
-      this.autoAbort && options.session && (await options.session.abortTransaction());
+      if (this.autoAbort && options.session) {
+        await options.session.abortTransaction();
+        await options.session.endSession();
+        this.session = undefined;
+      }
       throw err;
     }
   }