Transaction processing and data consistency

Basic Concepts of Transaction Processing

Transaction processing is a core mechanism in database operations, ensuring that a group of operations either all execute successfully or none execute at all. In Koa2 applications, transaction processing is crucial for maintaining data consistency. A typical transaction has ACID properties: Atomicity, Consistency, Isolation, and Durability.

// Example of transaction processing using Sequelize
const { sequelize } = require('./models');

async function transferFunds(senderId, receiverId, amount) {
  const transaction = await sequelize.transaction();
  
  try {
    const sender = await User.findByPk(senderId, { transaction });
    const receiver = await User.findByPk(receiverId, { transaction });
    
    if (sender.balance < amount) {
      throw new Error('Insufficient balance');
    }
    
    sender.balance -= amount;
    receiver.balance += amount;
    
    await sender.save({ transaction });
    await receiver.save({ transaction });
    
    await transaction.commit();
    return { success: true };
  } catch (error) {
    await transaction.rollback();
    return { success: false, error: error.message };
  }
}

Importance of Data Consistency

Data consistency refers to the database maintaining a correct state at all times, without partial updates or contradictory data. In web applications, especially in scenarios involving financial transactions, inventory management, etc., data consistency is particularly important. Inconsistent data can lead to business logic errors, financial losses, or even legal issues.

Consider an e-commerce platform's order processing scenario:

1. Create an order record
2. Deduct inventory
3. Generate a payment record
4. Update user points

These operations must be executed as an atomic unit; otherwise, situations like inventory being deducted but the order not being created may occur.

Transaction Implementation Methods in Koa2

In the Koa2 framework, transactions can be implemented in various ways, depending on the ORM or database driver used. Here are some common methods:

1. Sequelize Transactions:

router.post('/orders', async (ctx) => {
  const t = await sequelize.transaction();
  try {
    const order = await Order.create(ctx.request.body, { transaction: t });
    await Product.decrement('stock', {
      where: { id: order.productId },
      transaction: t
    });
    await t.commit();
    ctx.body = order;
  } catch (err) {
    await t.rollback();
    ctx.status = 500;
    ctx.body = { error: 'Transaction failed' };
  }
});

2. Mongoose Transactions (MongoDB):

const session = await mongoose.startSession();
session.startTransaction();
try {
  const order = new Order({...});
  await order.save({ session });
  
  await Product.updateOne(
    { _id: productId },
    { $inc: { stock: -1 } },
    { session }
  );
  
  await session.commitTransaction();
  ctx.body = order;
} catch (error) {
  await session.abortTransaction();
  ctx.status = 500;
  ctx.body = { error: 'Transaction failed' };
} finally {
  session.endSession();
}

Challenges of Distributed Transactions

In microservices architectures, transactions may span multiple services, introducing additional complexity. Common solutions include:

1. Saga Pattern:

// Example of a Saga coordinator
class OrderSaga {
  async execute(orderData) {
    try {
      // 1. Create order
      const order = await orderService.create(orderData);
      
      // 2. Reserve inventory
      await inventoryService.reserve(order.productId, order.quantity);
      
      // 3. Process payment
      await paymentService.process(order.id, order.total);
      
      // 4. Confirm all operations
      await orderService.confirm(order.id);
      await inventoryService.confirmReservation(order.productId, order.quantity);
      
      return { success: true, order };
    } catch (error) {
      // Compensating actions
      if (order) {
        await orderService.cancel(order.id);
        await inventoryService.cancelReservation(order.productId, order.quantity);
      }
      throw error;
    }
  }
}

2. TCC Pattern (Try-Confirm-Cancel):

// Example of TCC implementation
async function placeOrder(orderData) {
  // Try phase
  const order = await orderService.tryCreate(orderData);
  await inventoryService.tryReserve(order.productId, order.quantity);
  await paymentService.tryCharge(order.userId, order.total);
  
  // Confirm phase
  try {
    await orderService.confirm(order.id);
    await inventoryService.confirmReserve(order.productId, order.quantity);
    await paymentService.confirmCharge(order.userId, order.total);
    return order;
  } catch (error) {
    // Cancel phase
    await orderService.cancel(order.id);
    await inventoryService.cancelReserve(order.productId, order.quantity);
    await paymentService.cancelCharge(order.userId, order.total);
    throw error;
  }
}

Optimistic and Pessimistic Locking

To ensure data consistency in concurrent scenarios, common locking mechanisms include:

1. Optimistic Locking (suitable for read-heavy, write-light scenarios):

// Optimistic locking implementation using version numbers
router.put('/products/:id', async (ctx) => {
  const { id } = ctx.params;
  const { quantity, version } = ctx.request.body;
  
  const result = await Product.update(
    { quantity, version: version + 1 },
    {
      where: {
        id,
        version // Only update if the version matches
      }
    }
  );
  
  if (result[0] === 0) {
    ctx.status = 409;
    ctx.body = { error: 'Conflict - data has been modified' };
  } else {
    ctx.body = { success: true };
  }
});

2. Pessimistic Locking (suitable for write-heavy, read-light scenarios):

// Pessimistic locking using SELECT FOR UPDATE
router.post('/reserve', async (ctx) => {
  const transaction = await sequelize.transaction();
  try {
    const product = await Product.findOne({
      where: { id: ctx.request.body.productId },
      lock: transaction.LOCK.UPDATE,
      transaction
    });
    
    if (product.stock < ctx.request.body.quantity) {
      throw new Error('Insufficient stock');
    }
    
    product.stock -= ctx.request.body.quantity;
    await product.save({ transaction });
    
    await Reservation.create({
      productId: product.id,
      quantity: ctx.request.body.quantity,
      userId: ctx.state.user.id
    }, { transaction });
    
    await transaction.commit();
    ctx.body = { success: true };
  } catch (error) {
    await transaction.rollback();
    ctx.status = 400;
    ctx.body = { error: error.message };
  }
});

Transaction Isolation Levels

Different isolation levels address different concurrency issues:

1. Read Uncommitted - Lowest isolation level, may read uncommitted data
2. Read Committed - Only reads committed data
3. Repeatable Read - Ensures consistent reads within the same transaction
4. Serializable - Highest isolation level, fully serial execution

// Setting transaction isolation level (MySQL example)
const transaction = await sequelize.transaction({
  isolationLevel: Sequelize.Transaction.ISOLATION_LEVELS.SERIALIZABLE
});

// Or setting default isolation level in connection pool configuration
const sequelize = new Sequelize(database, username, password, {
  dialect: 'mysql',
  isolationLevel: Sequelize.Transaction.ISOLATION_LEVELS.REPEATABLE_READ
});

Error Handling and Retry Mechanisms

Robust transaction processing requires comprehensive error handling and retry mechanisms:

// Retry mechanism with exponential backoff
async function withRetry(operation, maxRetries = 3, baseDelay = 100) {
  let attempt = 0;
  
  while (attempt < maxRetries) {
    try {
      return await operation();
    } catch (error) {
      if (!isTransientError(error)) {
        throw error;
      }
      
      attempt++;
      if (attempt >= maxRetries) {
        throw error;
      }
      
      const delay = baseDelay * Math.pow(2, attempt) + Math.random() * 100;
      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }
}

// Usage example
router.post('/transactions', async (ctx) => {
  await withRetry(async () => {
    const transaction = await sequelize.transaction();
    try {
      // Business logic
      await transaction.commit();
    } catch (error) {
      await transaction.rollback();
      throw error;
    }
  });
  
  ctx.body = { success: true };
});

Performance Optimization Considerations

Transaction processing can become a performance bottleneck. Consider the following optimization strategies:

1. Reduce Transaction Scope:

// Bad practice: Entire request in transaction
router.post('/order', async (ctx) => {
  const t = await sequelize.transaction();
  try {
    // Multiple unrelated operations
    const user = await User.updateProfile(ctx.state.user.id, ctx.request.body.user, { transaction: t });
    const order = await Order.create(ctx.request.body.order, { transaction: t });
    await sendNotification(user.email, { transaction: t });
    await t.commit();
    ctx.body = order;
  } catch (error) {
    await t.rollback();
    throw error;
  }
});

// Good practice: Only related operations in transaction
router.post('/order', async (ctx) => {
  // User profile update doesn't need to be in order transaction
  await User.updateProfile(ctx.state.user.id, ctx.request.body.user);
  
  // Only order-related operations in transaction
  const t = await sequelize.transaction();
  try {
    const order = await Order.create(ctx.request.body.order, { transaction: t });
    await Inventory.adjust(order.productId, -order.quantity, { transaction: t });
    await t.commit();
    ctx.body = order;
  } catch (error) {
    await t.rollback();
    throw error;
  }
  
  // Notifications can be processed asynchronously
  sendNotification(ctx.state.user.email).catch(console.error);
});

2. Batch Operation Optimization:

// Inefficient single updates
for (const item of cartItems) {
  await Product.update(
    { stock: sequelize.literal(`stock - ${item.quantity}`) },
    { where: { id: item.productId } }
  );
}

// Efficient batch update
await Product.update(
  { stock: sequelize.literal(`stock - CASE id 
    ${cartItems.map(item => `WHEN ${item.productId} THEN ${item.quantity}`).join(' ')}
    ELSE stock END`) },
  { where: { id: cartItems.map(item => item.productId) } }
);