How to fix TransactionTooLarge: Total size of all transaction operations must be less than 16793600 in MongoDB

const session = client.startSession();
const documents = await collection.find({ status: 'pending' }).toArray();
const batchSize = 100; // Adjust based on your document size

try {
  for (let i = 0; i < documents.length; i += batchSize) {
    const batch = documents.slice(i, i + batchSize);

    session.startTransaction();

    for (const doc of batch) {
      await collection.updateOne(
        { _id: doc._id },
        { $set: { status: 'processed', processedAt: new Date() } },
        { session }
      );
    }

    await session.commitTransaction();
    console.log('Processed batch', i / batchSize + 1);
  }
} catch (error) {
  await session.abortTransaction();
  throw error;
} finally {
  session.endSession();
}

// Estimate document size
const sampleDoc = await collection.findOne();
const estimatedSize = Buffer.byteLength(JSON.stringify(sampleDoc));

// Calculate safe batch size (leave 10% buffer)
const maxTransactionSize = 16793600; // 16MB in bytes
const safeBatchSize = Math.floor((maxTransactionSize * 0.9) / estimatedSize);

console.log('Safe batch size:', safeBatchSize);

async function insertInBatches(documents, collection, batchSize = 100) {
  const session = client.startSession();

  try {
    for (let i = 0; i < documents.length; i += batchSize) {
      const batch = documents.slice(i, i + batchSize);

      session.startTransaction();
      await collection.insertMany(batch, { session });
      await session.commitTransaction();

      console.log('Inserted', i + batch.length, 'documents');
    }
  } catch (error) {
    await session.abortTransaction();
    throw error;
  } finally {
    session.endSession();
  }
}

// Bad: Fetch entire document
const fullDoc = await collection.findOne({ _id: docId });

// Good: Fetch only fields needed for transaction
const minimalDoc = await collection.findOne(
  { _id: docId },
  { projection: { _id: 1, status: 1, version: 1 } }
);

const zlib = require('zlib');
const util = require('util');
const compress = util.promisify(zlib.gzip);
const decompress = util.promisify(zlib.gunzip);

// Store compressed
const largeText = '...very long text...';
const compressed = await compress(largeText);
await collection.updateOne(
  { _id: docId },
  { $set: { content: compressed } },
  { session }
);

// Retrieve and decompress
const doc = await collection.findOne({ _id: docId });
const decompressed = await decompress(doc.content);

// Instead of storing large content in main document
// Main document
{
  _id: 'doc1',
  title: 'Document 1',
  metadata: { author: 'John', created: '2024-01-01' },
  contentId: 'content_doc1' // Reference to separate collection
}

// Content collection
{
  _id: 'content_doc1',
  docId: 'doc1',
  content: '...very long text...',
  contentType: 'text/plain'
}

// Transaction only updates main document
session.startTransaction();
await mainCollection.updateOne(
  { _id: 'doc1' },
  { $set: { 'metadata.updated': new Date() } },
  { session }
);
await session.commitTransaction();

// Store binary data outside transaction when possible
const GridFSBucket = require('mongodb').GridFSBucket;
const bucket = new GridFSBucket(db);

// Upload file to GridFS (not part of transaction)
const uploadStream = bucket.openUploadStream('file.txt');
fs.createReadStream('largefile.txt').pipe(uploadStream);

// Transaction only stores reference
session.startTransaction();
await collection.updateOne(
  { _id: docId },
  { $set: {
    fileId: uploadStream.id,
    fileName: 'file.txt',
    updatedAt: new Date()
  }},
  { session }
);
await session.commitTransaction();

class TransactionManager {
  constructor(client, maxSizeBytes = 16793600) {
    this.client = client;
    this.maxSizeBytes = maxSizeBytes;
  }

  async executeWithBatching(operations, estimateSizeFn) {
    const session = this.client.startSession();
    const batches = this.createBatches(operations, estimateSizeFn);

    try {
      for (const batch of batches) {
        await this.executeBatch(batch, session);
      }
    } catch (error) {
      await session.abortTransaction();
      throw error;
    } finally {
      session.endSession();
    }
  }

  createBatches(operations, estimateSizeFn) {
    const batches = [];
    let currentBatch = [];
    let currentSize = 0;

    for (const op of operations) {
      const opSize = estimateSizeFn(op);

      if (currentSize + opSize > this.maxSizeBytes * 0.9) {
        // Start new batch
        batches.push([...currentBatch]);
        currentBatch = [op];
        currentSize = opSize;
      } else {
        currentBatch.push(op);
        currentSize += opSize;
      }
    }

    if (currentBatch.length > 0) {
      batches.push(currentBatch);
    }

    return batches;
  }

  async executeBatch(batch, session) {
    session.startTransaction();

    try {
      for (const op of batch) {
        const { collection, operation, filter, update, options } = op;

        switch (operation) {
          case 'insertOne':
            await collection.insertOne(update, { ...options, session });
            break;
          case 'updateOne':
            await collection.updateOne(filter, update, { ...options, session });
            break;
          case 'deleteOne':
            await collection.deleteOne(filter, { ...options, session });
            break;
          // Add other operations as needed
        }
      }

      await session.commitTransaction();
    } catch (error) {
      await session.abortTransaction();
      throw error;
    }
  }
}

// Usage example
const manager = new TransactionManager(client);

// Estimate size function
function estimateOperationSize(op) {
  // Simple estimation - adjust based on your data
  const docSize = Buffer.byteLength(JSON.stringify(op.update || op.filter));
  return docSize + 1000; // Add overhead
}

const operations = [
  {
    collection: usersCollection,
    operation: 'updateOne',
    filter: { _id: 'user1' },
    update: { $set: { status: 'active' } }
  },
  // ... more operations
];

await manager.executeWithBatching(operations, estimateOperationSize);

const BSON = require('bson');

function estimateBSONSize(doc) {
  try {
    const buffer = BSON.serialize(doc);
    return buffer.length;
  } catch (error) {
    // Fallback to JSON estimation
    return Buffer.byteLength(JSON.stringify(doc)) * 1.1; // BSON is typically larger
  }
}

// Decorator/wrapper to log transaction sizes
async function executeTransaction(session, operations) {
  const startTime = Date.now();
  let totalSize = 0;

  // Estimate total size
  for (const op of operations) {
    if (op.update) {
      totalSize += Buffer.byteLength(JSON.stringify(op.update));
    }
    if (op.documents) {
      totalSize += Buffer.byteLength(JSON.stringify(op.documents));
    }
  }

  console.log('Transaction starting:', {
    operationCount: operations.length,
    estimatedSize: totalSize,
    sizePercentage: (totalSize / 16793600 * 100).toFixed(1) + '%'
  });

  try {
    session.startTransaction();

    for (const op of operations) {
      // Execute operation
    }

    await session.commitTransaction();

    console.log('Transaction completed:', {
      duration: Date.now() - startTime,
      size: totalSize,
      success: true
    });

  } catch (error) {
    console.error('Transaction failed:', {
      duration: Date.now() - startTime,
      size: totalSize,
      error: error.message
    });
    throw error;
  }
}

// Alert when transactions approach limit
const TRANSACTION_LIMIT = 16793600;
const WARNING_THRESHOLD = TRANSACTION_LIMIT * 0.8; // 80%

function checkTransactionSize(operations) {
  let totalSize = 0;

  for (const op of operations) {
    // Calculate size
    totalSize += calculateOperationSize(op);
  }

  if (totalSize > WARNING_THRESHOLD) {
    console.warn('Transaction size warning:', {
      size: totalSize,
      percentage: (totalSize / TRANSACTION_LIMIT * 100).toFixed(1) + '%',
      operationCount: operations.length
    });

    // Send alert (e.g., to Slack, email, monitoring system)
    // sendAlert({
    //   type: 'TRANSACTION_SIZE_WARNING',
    //   message: 'Transaction size at ' + (totalSize / TRANSACTION_LIMIT * 100).toFixed(1) + '% of limit',
    //   size: totalSize,
    //   limit: TRANSACTION_LIMIT
    // });
  }

  return totalSize;
}

// Store transaction metrics for analysis
const transactionMetrics = [];

async function trackTransactionMetrics(operations, success, duration) {
  const size = estimateTransactionSize(operations);

  transactionMetrics.push({
    timestamp: new Date(),
    size,
    operationCount: operations.length,
    success,
    duration,
    sizePercentage: (size / 16793600 * 100)
  });

  // Keep only last 1000 entries
  if (transactionMetrics.length > 1000) {
    transactionMetrics.shift();
  }

  // Periodic analysis
  if (transactionMetrics.length % 100 === 0) {
    analyzeTransactionPatterns();
  }
}

function analyzeTransactionPatterns() {
  const largeTransactions = transactionMetrics.filter(m => m.sizePercentage > 50);
  const failedTransactions = transactionMetrics.filter(m => !m.success);

  console.log('Transaction analysis:', {
    totalTransactions: transactionMetrics.length,
    largeTransactions: largeTransactions.length,
    failedTransactions: failedTransactions.length,
    avgSize: transactionMetrics.reduce((sum, m) => sum + m.size, 0) / transactionMetrics.length,
    maxSize: Math.max(...transactionMetrics.map(m => m.size))
  });
}

// Check current transaction statistics
const result = await db.adminCommand({
  serverStatus: 1,
  transactions: 1
});

console.log('Transaction stats:', {
  currentActive: result.transactions?.currentActive || 0,
  currentInactive: result.transactions?.currentInactive || 0,
  totalStarted: result.transactions?.totalStarted || 0,
  totalCommitted: result.transactions?.totalCommitted || 0,
  totalAborted: result.transactions?.totalAborted || 0
});

// Instead of large transaction, use change streams
const changeStream = collection.watch();

changeStream.on('change', async (change) => {
  // Process changes asynchronously
  await processChange(change);
});

// Make individual updates without transaction
await collection.updateOne(
  { _id: 'doc1' },
  { $set: { status: 'processing' } }
);

await collection.updateOne(
  { _id: 'doc2' },
  { $set: { status: 'processing' } }
);

// Change stream will pick up both changes

// Instead of one large transaction, use compensating transactions
async function updateWithCompensation(updates) {
  const completed = [];

  try {
    for (const update of updates) {
      const session = client.startSession();

      try {
        session.startTransaction();
        await update.collection.updateOne(
          update.filter,
          update.update,
          { session }
        );
        await session.commitTransaction();
        completed.push(update);
      } catch (error) {
        await session.abortTransaction();

        // Rollback completed updates
        for (const compUpdate of completed.reverse()) {
          await rollbackUpdate(compUpdate);
        }

        throw error;
      } finally {
        session.endSession();
      }
    }
  } catch (error) {
    console.error('Failed with compensation:', error);
    throw error;
  }
}

async function rollbackUpdate(update) {
  // Implement rollback logic for each update type
  const session = client.startSession();

  try {
    session.startTransaction();
    // Reverse the update
    await update.collection.updateOne(
      update.filter,
      { $set: update.previousState },
      { session }
    );
    await session.commitTransaction();
  } catch (rollbackError) {
    await session.abortTransaction();
    console.error('Rollback failed:', rollbackError);
    // Log for manual intervention
  } finally {
    session.endSession();
  }
}

// Manual two-phase commit for cross-document operations
async function twoPhaseCommit(operations) {
  // Phase 1: Prepare
  const prepared = [];

  for (const op of operations) {
    const prepareResult = await prepareOperation(op);
    prepared.push({ ...op, prepareId: prepareResult.id });
  }

  // Phase 2: Commit
  try {
    for (const preparedOp of prepared) {
      await commitOperation(preparedOp);
    }
  } catch (error) {
    // Phase 3: Rollback
    for (const preparedOp of prepared) {
      await rollbackOperation(preparedOp);
    }
    throw error;
  }
}

async function prepareOperation(operation) {
  // Store operation in "prepared" state
  const prepareDoc = {
    operation,
    status: 'prepared',
    preparedAt: new Date(),
    expiresAt: new Date(Date.now() + 300000) // 5 minutes
  };

  const result = await preparedCollection.insertOne(prepareDoc);
  return { id: result.insertedId };
}

// Denormalize data to reduce transaction needs
// Instead of updating multiple related documents:
// Document 1: { _id: 'order1', status: 'pending', items: [...] }
// Document 2: { _id: 'inventory1', quantity: 10 }
// Document 3: { _id: 'payment1', status: 'pending' }

// Use single document with embedded data:
{
  _id: 'order1',
  status: 'processing',
  items: [...],
  inventoryUpdates: [
    { itemId: 'item1', newQuantity: 5 }
  ],
  payment: {
    status: 'processing',
    amount: 100.00
  },
  updatedAt: new Date()
}

// Single transaction updates one document
session.startTransaction();
await ordersCollection.updateOne(
  { _id: 'order1' },
  { $set: {
    status: 'completed',
    'payment.status': 'completed',
    completedAt: new Date()
  }},
  { session }
);
await session.commitTransaction();