Data consistency and transaction processing

Basic Concepts of Data Consistency

Data consistency refers to the ability of a database to maintain data correctness and integrity at any point in time. In Mongoose, data consistency is primarily reflected in three aspects: model definition, data validation, and transaction handling. For example, when defining a user model, you can enforce field types and required properties through the Schema:

const userSchema = new mongoose.Schema({
  username: {
    type: String,
    required: true,
    unique: true
  },
  age: {
    type: Number,
    min: 18,
    max: 120
  },
  email: {
    type: String,
    match: /^\S+@\S+\.\S+$/
  }
});

Transaction Handling in Mongoose

MongoDB introduced support for multi-document transactions starting from version 4.0, and Mongoose provides a more user-friendly API on top of this. Typical use cases for transactions include bank transfers, order creation, and other business scenarios requiring atomic execution of multiple operations:

const session = await mongoose.startSession();
session.startTransaction();

try {
  const fromAccount = await Account.findOne({ _id: 'A' }).session(session);
  const toAccount = await Account.findOne({ _id: 'B' }).session(session);
  
  fromAccount.balance -= 100;
  toAccount.balance += 100;
  
  await fromAccount.save();
  await toAccount.save();
  
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
  throw error;
} finally {
  session.endSession();
}

Optimistic Concurrency Control

Mongoose implements optimistic locking through version numbers (the __v field) to prevent data inconsistencies caused by concurrent updates:

const product = await Product.findById('someId');
product.price = 200;

// Simulate concurrent modification
const concurrentProduct = await Product.findById('someId');
concurrentProduct.stock -= 1;
await concurrentProduct.save();

try {
  await product.save(); // Throws VersionError
} catch (err) {
  if (err instanceof mongoose.Error.VersionError) {
    // Handle version conflict
  }
}

Middleware and Data Consistency

Mongoose middleware (pre/post hooks) can execute custom logic before or after operations to ensure data consistency:

orderSchema.pre('save', async function() {
  const product = await Product.findById(this.productId);
  if (product.stock < this.quantity) {
    throw new Error('Insufficient stock');
  }
});

orderSchema.post('save', async function() {
  await Product.updateOne(
    { _id: this.productId },
    { $inc: { stock: -this.quantity } }
  );
});

Data Consistency in Bulk Operations

For bulk operations, using bulkWrite ensures better performance and data consistency:

await Character.bulkWrite([
  {
    updateOne: {
      filter: { name: '张三' },
      update: { $set: { age: 30 } }
    }
  },
  {
    updateOne: {
      filter: { name: '李四' },
      update: { $inc: { score: 5 } }
    }
  }
]);

Challenges in Distributed Environments

In distributed systems, Mongoose needs to work with MongoDB's sharded clusters to handle cross-shard transactions:

const session = await mongoose.startSession();
session.startTransaction({
  readConcern: { level: 'snapshot' },
  writeConcern: { w: 'majority' }
});

try {
  // Cross-shard operations
  await Order.create([{ ... }], { session });
  await Inventory.updateOne({ ... }, { $inc: { ... } }, { session });
  
  await session.commitTransaction();
} catch (error) {
  await session.abortTransaction();
}

Error Handling and Retry Mechanisms

Temporary failures like network fluctuations require implementing automatic retry logic:

const withTransactionRetry = async (txnFunc, session) => {
  let retryCount = 0;
  const MAX_RETRIES = 3;
  
  while (true) {
    try {
      return await txnFunc(session);
    } catch (err) {
      if (err.hasErrorLabel('TransientTransactionError') && 
          retryCount < MAX_RETRIES) {
        retryCount++;
        await new Promise(resolve => setTimeout(resolve, 100 * retryCount));
        continue;
      }
      throw err;
    }
  }
};

Consistency Considerations in Read-Write Separation

When using MongoDB's readPreference, be mindful of data synchronization delays:

// Write using the primary node
await User.create({ name: '王五' }, { writeConcern: { w: 'majority' } });

// Read allowing secondary nodes
const users = await User.find().read('secondaryPreferred');

Atomicity in Data Migration

Large-scale data migrations require ensuring atomicity and rollback capabilities:

const migrationSession = await mongoose.startSession();
migrationSession.startTransaction();

try {
  await OldModel.find().session(migrationSession).cursor()
    .eachAsync(async (doc) => {
      await NewModel.create(transform(doc), { session: migrationSession });
      await OldModel.deleteOne({ _id: doc._id }).session(migrationSession);
    }, { parallel: 5 });
    
  await migrationSession.commitTransaction();
} catch (error) {
  await migrationSession.abortTransaction();
}