dcrouter/test/suite/smtpclient_performance/test.cperf-02.message-throughput.ts

import { expect, tap } from '@git.zone/tstest/tapbundle';
import * as plugins from './plugins.js';
import { createTestServer } from '../../helpers/server.loader.js';
import { createSmtpClient } from '../../helpers/smtp.client.js';

tap.test('CPERF-02: should achieve optimal message throughput', async (tools) => {
  const testId = 'CPERF-02-message-throughput';
  console.log(`\n${testId}: Testing message throughput performance...`);

  let scenarioCount = 0;

  // Scenario 1: Sequential message throughput
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing sequential message throughput`);
    
    let messageCount = 0;
    const startTime = Date.now();
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        console.log('  [Server] Client connected');
        socket.write('220 throughput.example.com ESMTP\r\n');
        
        socket.on('data', (data) => {
          const command = data.toString().trim();
          
          if (command.startsWith('EHLO')) {
            socket.write('250-throughput.example.com\r\n');
            socket.write('250-PIPELINING\r\n');
            socket.write('250 OK\r\n');
          } else if (command.startsWith('MAIL FROM:')) {
            socket.write('250 OK\r\n');
          } else if (command.startsWith('RCPT TO:')) {
            socket.write('250 OK\r\n');
          } else if (command === 'DATA') {
            socket.write('354 Start mail input\r\n');
          } else if (command === '.') {
            messageCount++;
            const elapsed = Date.now() - startTime;
            const rate = (messageCount / elapsed) * 1000;
            socket.write(`250 OK: Message ${messageCount} (${rate.toFixed(1)} msg/sec)\r\n`);
          } else if (command === 'QUIT') {
            socket.write('221 Bye\r\n');
            socket.end();
          }
        });
      }
    });

    const smtpClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false
    });

    const messageCount_ = 20;
    const messages = Array(messageCount_).fill(null).map((_, i) => 
      new plugins.smartmail.Email({
        from: 'sender@example.com',
        to: [`recipient${i + 1}@example.com`],
        subject: `Sequential throughput test ${i + 1}`,
        text: `Testing sequential message sending - message ${i + 1}`
      })
    );

    console.log(`  Sending ${messageCount_} messages sequentially...`);
    const sequentialStart = Date.now();
    
    for (const message of messages) {
      await smtpClient.sendMail(message);
    }
    
    const sequentialTime = Date.now() - sequentialStart;
    const sequentialRate = (messageCount_ / sequentialTime) * 1000;
    
    console.log(`  Sequential throughput: ${sequentialRate.toFixed(2)} messages/second`);
    console.log(`  Total time: ${sequentialTime}ms for ${messageCount_} messages`);
    
    expect(sequentialRate).toBeGreaterThan(1); // At least 1 message per second
    expect(messageCount).toBe(messageCount_);

    await testServer.server.close();
  })();

  // Scenario 2: Concurrent message throughput
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing concurrent message throughput`);
    
    let messageCount = 0;
    const startTime = Date.now();
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        console.log('  [Server] Client connected');
        socket.write('220 concurrent.example.com ESMTP\r\n');
        
        socket.on('data', (data) => {
          const command = data.toString().trim();
          
          if (command.startsWith('EHLO')) {
            socket.write('250-concurrent.example.com\r\n');
            socket.write('250 OK\r\n');
          } else if (command.startsWith('MAIL FROM:')) {
            socket.write('250 OK\r\n');
          } else if (command.startsWith('RCPT TO:')) {
            socket.write('250 OK\r\n');
          } else if (command === 'DATA') {
            socket.write('354 Start mail input\r\n');
          } else if (command === '.') {
            messageCount++;
            const elapsed = Date.now() - startTime;
            const rate = (messageCount / elapsed) * 1000;
            socket.write(`250 OK: Message ${messageCount} (${rate.toFixed(1)} msg/sec)\r\n`);
          } else if (command === 'QUIT') {
            socket.write('221 Bye\r\n');
            socket.end();
          }
        });
      }
    });

    const smtpClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false
    });

    const messageCount_ = 30;
    const messages = Array(messageCount_).fill(null).map((_, i) => 
      new plugins.smartmail.Email({
        from: 'sender@example.com',
        to: [`recipient${i + 1}@example.com`],
        subject: `Concurrent throughput test ${i + 1}`,
        text: `Testing concurrent message sending - message ${i + 1}`
      })
    );

    console.log(`  Sending ${messageCount_} messages concurrently...`);
    const concurrentStart = Date.now();
    
    const results = await Promise.all(
      messages.map(message => smtpClient.sendMail(message))
    );
    
    const concurrentTime = Date.now() - concurrentStart;
    const concurrentRate = (messageCount_ / concurrentTime) * 1000;
    
    console.log(`  Concurrent throughput: ${concurrentRate.toFixed(2)} messages/second`);
    console.log(`  Total time: ${concurrentTime}ms for ${messageCount_} messages`);
    
    expect(concurrentRate).toBeGreaterThan(5); // Should be faster than sequential
    expect(results.length).toBe(messageCount_);

    await testServer.server.close();
  })();

  // Scenario 3: Pipelined message throughput
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing pipelined message throughput`);
    
    let messageCount = 0;
    const messageBuffer: string[] = [];
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        console.log('  [Server] Client connected');
        socket.write('220 pipeline.example.com ESMTP\r\n');
        
        socket.on('data', (data) => {
          const commands = data.toString().split('\r\n').filter(cmd => cmd.length > 0);
          
          // Process pipelined commands
          if (commands.length > 1) {
            console.log(`  [Server] Received ${commands.length} pipelined commands`);
          }
          
          commands.forEach(command => {
            if (command.startsWith('EHLO')) {
              socket.write('250-pipeline.example.com\r\n');
              socket.write('250-PIPELINING\r\n');
              socket.write('250 OK\r\n');
            } else if (command.startsWith('MAIL FROM:')) {
              socket.write('250 OK\r\n');
            } else if (command.startsWith('RCPT TO:')) {
              socket.write('250 OK\r\n');
            } else if (command === 'DATA') {
              socket.write('354 Start mail input\r\n');
            } else if (command === '.') {
              messageCount++;
              socket.write(`250 OK: Pipelined message ${messageCount}\r\n`);
            } else if (command === 'QUIT') {
              socket.write('221 Bye\r\n');
              socket.end();
            }
          });
        });
      }
    });

    const smtpClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false,
      pipelining: true
    });

    const messageCount_ = 25;
    const messages = Array(messageCount_).fill(null).map((_, i) => 
      new plugins.smartmail.Email({
        from: 'sender@example.com',
        to: [`recipient${i + 1}@example.com`],
        subject: `Pipelined throughput test ${i + 1}`,
        text: `Testing pipelined message sending - message ${i + 1}`
      })
    );

    console.log(`  Sending ${messageCount_} messages with pipelining...`);
    const pipelineStart = Date.now();
    
    const results = await Promise.all(
      messages.map(message => smtpClient.sendMail(message))
    );
    
    const pipelineTime = Date.now() - pipelineStart;
    const pipelineRate = (messageCount_ / pipelineTime) * 1000;
    
    console.log(`  Pipelined throughput: ${pipelineRate.toFixed(2)} messages/second`);
    console.log(`  Total time: ${pipelineTime}ms for ${messageCount_} messages`);
    
    expect(pipelineRate).toBeGreaterThan(3); // Should benefit from pipelining
    expect(results.length).toBe(messageCount_);

    await testServer.server.close();
  })();

  // Scenario 4: Connection pooling throughput
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing connection pooling throughput`);
    
    let connectionCount = 0;
    let messageCount = 0;
    const connectionMessages = new Map<any, number>();
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        connectionCount++;
        const connId = connectionCount;
        connectionMessages.set(socket, 0);
        
        console.log(`  [Server] Connection ${connId} established`);
        socket.write('220 pool.example.com ESMTP\r\n');
        
        socket.on('close', () => {
          const msgCount = connectionMessages.get(socket) || 0;
          connectionMessages.delete(socket);
          console.log(`  [Server] Connection ${connId} closed after ${msgCount} messages`);
        });
        
        socket.on('data', (data) => {
          const command = data.toString().trim();
          
          if (command.startsWith('EHLO')) {
            socket.write('250-pool.example.com\r\n');
            socket.write('250 OK\r\n');
          } else if (command.startsWith('MAIL FROM:')) {
            socket.write('250 OK\r\n');
          } else if (command.startsWith('RCPT TO:')) {
            socket.write('250 OK\r\n');
          } else if (command === 'DATA') {
            socket.write('354 Start mail input\r\n');
          } else if (command === '.') {
            messageCount++;
            const msgCount = (connectionMessages.get(socket) || 0) + 1;
            connectionMessages.set(socket, msgCount);
            socket.write(`250 OK: Message ${messageCount} on connection ${connId}\r\n`);
          } else if (command === 'QUIT') {
            socket.write('221 Bye\r\n');
            socket.end();
          }
        });
      }
    });

    const pooledClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false,
      pool: true,
      maxConnections: 5,
      maxMessages: 100
    });

    const messageCount_ = 40;
    const messages = Array(messageCount_).fill(null).map((_, i) => 
      new plugins.smartmail.Email({
        from: 'sender@example.com',
        to: [`recipient${i + 1}@example.com`],
        subject: `Pooled throughput test ${i + 1}`,
        text: `Testing connection pooling - message ${i + 1}`
      })
    );

    console.log(`  Sending ${messageCount_} messages with connection pooling...`);
    const poolStart = Date.now();
    
    const results = await Promise.all(
      messages.map(message => pooledClient.sendMail(message))
    );
    
    const poolTime = Date.now() - poolStart;
    const poolRate = (messageCount_ / poolTime) * 1000;
    
    console.log(`  Pooled throughput: ${poolRate.toFixed(2)} messages/second`);
    console.log(`  Total time: ${poolTime}ms for ${messageCount_} messages`);
    console.log(`  Used ${connectionCount} connections for ${messageCount_} messages`);
    
    expect(poolRate).toBeGreaterThan(8); // Should be faster with pooling
    expect(results.length).toBe(messageCount_);
    expect(connectionCount).toBeGreaterThan(1);
    expect(connectionCount).toBeLessThanOrEqual(5);

    await pooledClient.close();
    await testServer.server.close();
  })();

  // Scenario 5: Variable message size throughput
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing variable message size throughput`);
    
    let totalBytes = 0;
    let messageCount = 0;
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        console.log('  [Server] Client connected');
        socket.write('220 variable.example.com ESMTP\r\n');
        
        let inData = false;
        let messageSize = 0;
        
        socket.on('data', (data) => {
          if (inData) {
            messageSize += data.length;
            if (data.toString().includes('\r\n.\r\n')) {
              inData = false;
              messageCount++;
              totalBytes += messageSize;
              const avgSize = Math.round(totalBytes / messageCount);
              socket.write(`250 OK: Message ${messageCount} (${messageSize} bytes, avg: ${avgSize})\r\n`);
              messageSize = 0;
            }
            return;
          }
          
          const command = data.toString().trim();
          
          if (command.startsWith('EHLO')) {
            socket.write('250-variable.example.com\r\n');
            socket.write('250 OK\r\n');
          } else if (command.startsWith('MAIL FROM:')) {
            socket.write('250 OK\r\n');
          } else if (command.startsWith('RCPT TO:')) {
            socket.write('250 OK\r\n');
          } else if (command === 'DATA') {
            socket.write('354 Start mail input\r\n');
            inData = true;
            messageSize = 0;
          } else if (command === 'QUIT') {
            socket.write('221 Bye\r\n');
            socket.end();
          }
        });
      }
    });

    const smtpClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false
    });

    // Create messages of varying sizes
    const messageSizes = [
      { size: 'small', content: 'Short message' },
      { size: 'medium', content: 'Medium message: ' + 'x'.repeat(1000) },
      { size: 'large', content: 'Large message: ' + 'x'.repeat(10000) },
      { size: 'extra-large', content: 'Extra large message: ' + 'x'.repeat(50000) }
    ];
    
    const messages = [];
    for (let i = 0; i < 20; i++) {
      const sizeType = messageSizes[i % messageSizes.length];
      messages.push(new plugins.smartmail.Email({
        from: 'sender@example.com',
        to: [`recipient${i + 1}@example.com`],
        subject: `Variable size test ${i + 1} (${sizeType.size})`,
        text: sizeType.content
      }));
    }

    console.log(`  Sending ${messages.length} messages of varying sizes...`);
    const variableStart = Date.now();
    
    const results = await Promise.all(
      messages.map(message => smtpClient.sendMail(message))
    );
    
    const variableTime = Date.now() - variableStart;
    const variableRate = (messages.length / variableTime) * 1000;
    const bytesPerSecond = (totalBytes / variableTime) * 1000;
    
    console.log(`  Variable size throughput: ${variableRate.toFixed(2)} messages/second`);
    console.log(`  Data throughput: ${(bytesPerSecond / 1024).toFixed(2)} KB/second`);
    console.log(`  Average message size: ${Math.round(totalBytes / messages.length)} bytes`);
    
    expect(variableRate).toBeGreaterThan(2);
    expect(results.length).toBe(messages.length);

    await testServer.server.close();
  })();

  // Scenario 6: Sustained throughput over time
  await (async () => {
    scenarioCount++;
    console.log(`\nScenario ${scenarioCount}: Testing sustained throughput over time`);
    
    let messageCount = 0;
    const timestamps: number[] = [];
    
    const testServer = await createTestServer({
      onConnection: async (socket) => {
        console.log('  [Server] Client connected');
        socket.write('220 sustained.example.com ESMTP\r\n');
        
        socket.on('data', (data) => {
          const command = data.toString().trim();
          
          if (command.startsWith('EHLO')) {
            socket.write('250-sustained.example.com\r\n');
            socket.write('250 OK\r\n');
          } else if (command.startsWith('MAIL FROM:')) {
            socket.write('250 OK\r\n');
          } else if (command.startsWith('RCPT TO:')) {
            socket.write('250 OK\r\n');
          } else if (command === 'DATA') {
            socket.write('354 Start mail input\r\n');
          } else if (command === '.') {
            messageCount++;
            timestamps.push(Date.now());
            socket.write(`250 OK: Sustained message ${messageCount}\r\n`);
          } else if (command === 'QUIT') {
            socket.write('221 Bye\r\n');
            socket.end();
          }
        });
      }
    });

    const smtpClient = createSmtpClient({
      host: testServer.hostname,
      port: testServer.port,
      secure: false,
      pool: true,
      maxConnections: 3
    });

    const totalMessages = 30;
    const batchSize = 5;
    const batchDelay = 500; // 500ms between batches
    
    console.log(`  Sending ${totalMessages} messages in batches of ${batchSize}...`);
    const sustainedStart = Date.now();
    
    for (let batch = 0; batch < totalMessages / batchSize; batch++) {
      const batchMessages = Array(batchSize).fill(null).map((_, i) => {
        const msgIndex = batch * batchSize + i + 1;
        return new plugins.smartmail.Email({
          from: 'sender@example.com',
          to: [`recipient${msgIndex}@example.com`],
          subject: `Sustained test batch ${batch + 1} message ${i + 1}`,
          text: `Testing sustained throughput - message ${msgIndex}`
        });
      });
      
      // Send batch concurrently
      await Promise.all(
        batchMessages.map(message => smtpClient.sendMail(message))
      );
      
      console.log(`    Batch ${batch + 1} completed (${(batch + 1) * batchSize} messages total)`);
      
      // Delay between batches (except last)
      if (batch < (totalMessages / batchSize) - 1) {
        await new Promise(resolve => setTimeout(resolve, batchDelay));
      }
    }
    
    const sustainedTime = Date.now() - sustainedStart;
    const sustainedRate = (totalMessages / sustainedTime) * 1000;
    
    // Calculate rate stability
    const windowSize = 5;
    const rates: number[] = [];
    for (let i = windowSize; i < timestamps.length; i++) {
      const windowStart = timestamps[i - windowSize];
      const windowEnd = timestamps[i];
      const windowRate = (windowSize / (windowEnd - windowStart)) * 1000;
      rates.push(windowRate);
    }
    
    const avgRate = rates.reduce((a, b) => a + b, 0) / rates.length;
    const rateVariance = rates.reduce((acc, rate) => acc + Math.pow(rate - avgRate, 2), 0) / rates.length;
    const rateStdDev = Math.sqrt(rateVariance);
    
    console.log(`  Sustained throughput: ${sustainedRate.toFixed(2)} messages/second`);
    console.log(`  Average windowed rate: ${avgRate.toFixed(2)} ± ${rateStdDev.toFixed(2)} msg/sec`);
    console.log(`  Rate stability: ${((1 - rateStdDev / avgRate) * 100).toFixed(1)}%`);
    
    expect(sustainedRate).toBeGreaterThan(3);
    expect(rateStdDev / avgRate).toBeLessThan(0.5); // Coefficient of variation < 50%

    await smtpClient.close();
    await testServer.server.close();
  })();

  console.log(`\n${testId}: All ${scenarioCount} message throughput scenarios tested ✓`);
});