timeout bounds transaction work before commit begins. The deadline is checked before dispatching commit, then its timer stops. Commit, abort and session end are serialized. For direct Transaction construction, configure driver commit limits with transactionOptions.maxCommitTimeMS. The high-level withTransaction() / startSession() options currently do not forward maxCommitTimeMS; timeout does not replace a driver/server commit limit. An unknown commit result conservatively invalidates affected caches and preserves the original error without reporting rollback or replaying the business callback.
monSQLize wraps MongoDB driver sessions with withTransaction() and startSession() helpers. The ACID boundary is MongoDB's transaction/session boundary; monSQLize adds retry, timeout, statistics, and cache-invalidation coordination around that driver behavior.
Managed transaction writes stop being accepted as soon as commit or abort begins, including lazy aggregate writes and batch subwrites. An unknown commit result retries commit without rerunning the business callback; errors after a successful commit also do not rerun it. Native sessions passed directly to the MongoDB driver keep their native behavior.
- ✅ Automatic Transaction Management (withTransaction - Recommended)
- ✅ Manual transaction management (startSession - Advanced usage)
- ✅ Transaction-aware cache policy (session reads bypass shared query cache; writes record post-commit invalidations)
- ✅ Process-local cache barrier/lock (short-lived protection around cache invalidation)
- ✅ Auto-Retry (Auto-retry on transient errors)
- ✅ Work timeout (abort work that exceeds its deadline before commit begins)
- ✅ Monitoring indicators (execution time, success rate, etc.)
- ✅ Read attention/read preference/causal consistency
- ✅ MongoDB 4.0+ replica set or sharded cluster
- ✅ Node.js 18+
import MonSQLize from 'monsqlize';
const msq = new MonSQLize({
type: 'mongodb',
databaseName: 'mydb',
config: {
uri: 'mongodb://localhost:27017?replicaSet=rs0', // Must be a replica set
options: {}
},
cache: {
defaultTtl: 60000, //Cache for 60 seconds
maxEntries: 1000 //Up to 1000 items
}
});
await msq.connect();
const { collection } = await msq.connect();The simplest way to automatically manage commits, rollbacks and retries:
//Example 1: Transfer
await msq.withTransaction(async (tx) => {
const accounts = collection('accounts');
//Debit Alice
await accounts.updateOne(
{ userId: 'alice' },
{ $inc: { balance: -100 } },
{ session: tx.session } //🔑 Pass in session
);
//Add money to Bob
await accounts.updateOne(
{ userId: 'bob' },
{ $inc: { balance: 100 } },
{ session: tx.session }
);
//✅ Success: Automatic submission
//❌ Failure: automatic rollback
});//Example 2: Inventory deduction + create order
const orderId = await msq.withTransaction(async (tx) => {
const inventory = collection('inventory');
const orders = collection('orders');
//Check inventory
const product = await inventory.findOne(
{ productId: 'SKU123' },
{ session: tx.session }
);
if (product.stock < 10) {
throw new Error('Insufficient stock');
}
//deduction inventory
await inventory.updateOne(
{ productId: 'SKU123' },
{ $inc: { stock: -10 } },
{ session: tx.session }
);
//Create order
const order = {
orderId: 'ORDER001',
productId: 'SKU123',
quantity: 10,
createdAt: new Date()
};
await orders.insertOne(order, { session: tx.session });
return order.orderId;
});
console.log('Order created successfully:', orderId);Use when you need fine control over the transaction life cycle:
const tx = await msq.startSession();
try {
await tx.start();
//perform operations
await collection('accounts').updateOne(
{ userId: 'alice' },
{ $inc: { balance: -100 } },
{ session: tx.session }
);
//Manual submission
await tx.commit();
} catch (error) {
//Manual rollback
await tx.abort();
throw error;
} finally {
//Release resources
await tx.end();
}const msq = new MonSQLize({
//...basic configuration
transaction: {
// Reads with a session bypass shared query cache, even when cache is specified.
// Writes record invalidation intents for commit.
//Whether to enable automatic retry (default: true)
enableRetry: true,
//Maximum number of retries (default: 3)
maxRetries: 3,
//Retry delay (ms, default: 100)
retryDelay: 100,
//Retry backoff factor (default: 2)
retryBackoff: 2,
//Default work deadline before commit (milliseconds, default: 30000)
defaultTimeout: 30000,
//Default read following (optional)
defaultReadConcern: { level: 'majority' },
//Write follow by default (optional)
defaultWriteConcern: { w: 'majority' },
//Cache lock maximum duration (ms, default: 300000)
lockMaxDuration: 300000,
//Cache lock cleanup interval (milliseconds, default: 10000)
lockCleanupInterval: 10000
}
});await msq.withTransaction(async (tx) => {
//transaction logic
}, {
//Read attention level
readConcern: { level: 'snapshot' },
//write level of concern
writeConcern: { w: 'majority' },
//Read preferences
readPreference: 'primary',
//Causal consistency (default: true)
causalConsistency: true,
//Work deadline before commit (milliseconds)
timeout: 60000,
//Maximum number of retries (overrides global configuration)
maxRetries: 5
});Automatically manage transactions (recommended).
Parameters:
callback(tx): transaction callback functiontx.session: MongoDB session objecttx.id: transaction unique identifiertx.state: Transaction status ('pending' | 'active' | 'committed' | 'aborted')
options: transaction options (optional)
Returns: Promise - returns the return value of callback
Example:
const result = await msq.withTransaction(async (tx) => {
//Operations within a transaction must be passed into the session
await collection('users').updateOne(
{ _id: 1 },
{ $set: { name: 'Alice' } },
{ session: tx.session }
);
return { success: true };
});Create a manual transaction session.
Parameters:
options: Transaction options (same as withTransaction)
Return: Promise
Transaction instance method:
start(): Start transactioncommit(): Commit transactionabort(): rollback transactionend(): End the session after pending commit/abort work; abort an active transaction whose outcome is known. It does not commit.
Example:
const tx = await msq.startSession({
readConcern: { level: 'majority' },
timeout: 60000
});
try {
await tx.start();
//perform operations
await collection('accounts').updateOne(
{ _id: 1 },
{ $inc: { balance: -100 } },
{ session: tx.session }
);
await tx.commit();
} catch (error) {
await tx.abort();
throw error;
} finally {
await tx.end();
}Design Concept: Transactions pursue data consistency, and caching pursues performance. By default, intra-transaction operations do not use caching, ensuring data accuracy.
await msq.withTransaction(async (tx) => {
//✅ Intra-transaction query: read directly from the database, without using cache
const user = await collection('users').findOne(
{ _id: 1 },
{ session: tx.session }
//No need to specify cache: 0, no caching by default
);
//✅ Intra-transaction writing: record a cache invalidation intent only when configured
await collection('users').updateOne(
{ _id: 1 },
{ $set: { balance: 100 } },
{
session: tx.session,
cache: {
invalidate: [{
operation: 'findOne',
query: { _id: 1 },
options: { cache: 5000 }
}]
}
}
);
});
// Cache invalidation is flushed after commit success or an unknown commit result.
// If the transaction aborts with a known outcome, pending intents are discarded.
// Post-commit cache invalidation is best-effort; a cache failure does not roll back the database commit.
//✅ Query outside transaction: use cache normally
const user = await collection('users').findOne(
{ _id: 1 },
{ cache: 60000 } //Cache for 60 seconds
);When an operation receives session: tx.session, monSQLize forwards the session to the MongoDB driver and skips the shared query-result cache. There is no separate transaction-cache isolation flag in the public API. Keep repeated transaction reads inside the driver snapshot, or move cacheable reads outside the transaction when that is safe for your workflow.
Function: Record explicit cache invalidation intents during the transaction and keep a short-lived, process-local barrier until those intents are flushed after commit or discarded on abort.
await msq.withTransaction(async (tx) => {
//1. Update data
await collection('users').updateOne(
{ _id: 1 },
{ $set: { balance: 100 } },
{
session: tx.session,
cache: {
invalidate: [{
operation: 'findOne',
query: { _id: 1 },
options: { cache: 5000 }
}]
}
}
);
//🔒 Records the configured cache invalidation intent in this process
//2. Other reads in this process bypass or avoid refilling affected cache keys
//3. Transaction submission
await tx.commit();
//🔓 Release process-local lock + flush recorded invalidations
});The cache lock manager is process-local. Cross-instance cache coherence depends on post-commit distributed invalidation when configured, and remains best-effort/eventual rather than an atomic cache/DB commit.
| Strategy | Advantages | Disadvantages | Applicable scenarios |
|---|---|---|---|
| No caching (default) | High data consistency and simplicity | Slightly lower performance | Most scenarios |
| Session reads | Uses MongoDB snapshot semantics | Does not use shared query cache | Reads inside the transaction |
| Cache barrier/lock | Reduces stale cache refill windows in this process | Cross-instance coherence is still eventual | Automatically used for transaction invalidation |
Important: Transaction callbacks must be idempotent as automatic retries may be possible.
//✅ Good design: use unique identifiers
await msq.withTransaction(async (tx) => {
await collection('orders').insertOne({
orderId: 'ORDER_' + Date.now(), //Unique ID
status: 'pending'
}, { session: tx.session });
});
//❌ Bad design: relying on external state
let counter = 0;
await msq.withTransaction(async (tx) => {
counter++; //Retrying will cause the counter to be incremented multiple times
await collection('logs').insertOne({
logId: counter //Not idempotent
}, { session: tx.session });
});//Short transactions (recommended)
await msq.withTransaction(async (tx) => {
//Simple operation
}, { timeout: 5000 });
//Long transactions (use with caution)
await msq.withTransaction(async (tx) => {
//Complex operations
}, { timeout: 60000 }); //Work deadline only; MongoDB server limits are independenttry {
await msq.withTransaction(async (tx) => {
//business logic
const user = await collection('users').findOne(
{ _id: 1 },
{ session: tx.session }
);
if (!user) {
throw new Error('User does not exist');
}
//More actions...
});
} catch (error) {
if (error.message === 'User does not exist') {
//business error
console.error('Business error:', error.message);
} else if (error.errorLabels?.includes('TransientTransactionError')) {
//MongoDB transient error (automatically retried)
console.error('Transaction failed:', error.message);
} else {
//Other errors
console.error('Unknown error:', error);
}
}//✅ Good practice: Verify first, then transaction
async function transfer(fromId, toId, amount) {
//1. Pre-check outside the transaction (fail fast)
const fromUser = await collection('users').findOne({ _id: fromId });
if (!fromUser || fromUser.balance < amount) {
throw new Error('Insufficient balance');
}
//2. Execution within a transaction
await msq.withTransaction(async (tx) => {
await collection('users').updateOne(
{ _id: fromId },
{ $inc: { balance: -amount } },
{ session: tx.session }
);
await collection('users').updateOne(
{ _id: toId },
{ $inc: { balance: amount } },
{ session: tx.session }
);
});
}
//❌ Bad practice: all logic within transactions
await msq.withTransaction(async (tx) => {
//Complex business logic
//multiple network calls
//...(transaction occupied for a long time)
});const tx = await msq.startSession();
try {
await tx.start();
console.log('Transaction starts:', tx.id);
//business logic
await collection('users').updateOne(
{ _id: 1 },
{ $set: { lastLogin: new Date() } },
{ session: tx.session }
);
await tx.commit();
console.log('Transaction submitted successfully:', tx.id);
} catch (error) {
await tx.abort();
console.error('Transaction rollback:', tx.id, error);
throw error;
} finally {
await tx.end();
}A: This is the default behavior by design. Reason:
- Data consistency first - Transactions pursue accuracy, cache may have delays
- Avoid dirty reads - The latest data should be read within the transaction
- Simplified use - Users do not need to consider caching issues
If performance optimization is required, the cache option can be enabled explicitly.
A: Use withTransaction (auto) in most cases. Consider manual operation in the following situations:
- Need to make complex judgments before starting a transaction
- Additional verification is required before committing
- Need for fine-grained control of transaction life cycle
A: Check the following points:
- **Is MongoDB a replica set? ** - Single node does not support transactions
- **Is the connection string correct? ** - requires
?replicaSet=rs0 - **Is the timeout reasonable? ** - Default 30 seconds
- **Is the callback idempotent? ** - may retry multiple times
withTransaction() retries transient transaction errors and retries commitTransaction() when the driver marks the result as UnknownTransactionCommitResult.
A: The impact is minimal. Reason:
- The lock is only in effect for the duration of the transaction (usually a short period of time)
- The lock is process-local memory (no network I/O)
- Checking of locks is very fast (O(1) hash lookup)
The cache lock is process-local. Cross-instance cache coherence is handled by distributed invalidation after commit on a best-effort basis; use application/framework-level coordination when a critical section must be mutually exclusive across processes.
A: Yes, but there are restrictions:
- ✅ Multiple databases within the same MongoDB cluster
- ❌ Across MongoDB clusters (not supported)
await msq.withTransaction(async (tx) => {
//Operation db1.users
await collection('users').updateOne(
{ _id: 1 },
{ $set: { status: 'active' } },
{ session: tx.session }
);
//Operate db2.logs (you need to obtain the collection of db2 first)
const db2 = msq.db('db2');
await db2.collection('logs').insertOne({
action: 'user_activated',
userId: 1
}, { session: tx.session });
});A: MongoDB automatically detects and throws WriteConflict errors. monSQLize will automatically retry if enableRetry is enabled.
//✅ Good practice
const validated = await preValidate(); //Outside affairs
if (validated) {
await msq.withTransaction(async (tx) => {
//Core operations only
});
}
//❌ Bad practice
await msq.withTransaction(async (tx) => {
await validate(); //within affairs
await doWork(); //within affairs
});//✅ Good Practice: Use Bulk API
await msq.withTransaction(async (tx) => {
await collection('users').updateMany(
{ status: 'inactive' },
{ $set: { status: 'deleted' } },
{ session: tx.session }
);
});
//❌ Bad practice: looping single updates
await msq.withTransaction(async (tx) => {
for (const user of users) {
await collection('users').updateOne(
{ _id: user._id },
{ $set: { status: 'deleted' } },
{ session: tx.session }
);
}
});//✅ Make sure the query field is indexed
await collection('users').createIndex({ userId: 1 });
await msq.withTransaction(async (tx) => {
//Query using indexed fields (fast)
const user = await collection('users').findOne(
{ userId: 'alice' }, //There is an index
{ session: tx.session }
);
});//Get transaction statistics
const stats = msq.getTransactionStats();
if (stats) {
console.log('Transaction statistics:', {
total: stats.totalTransactions,
successful: stats.successfulTransactions,
failed: stats.failedTransactions,
averageDuration: `${stats.averageDuration}ms`
});
}