Best Case Study Salesforce in 2026: Tested & Reviewed

import os
import time
import json
import logging
from dataclasses import dataclass
from typing import List, Dict, Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Configure logging for benchmark visibility
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

@dataclass
class BulkIngestionResult:
    \"\"\"Holds benchmark results for a single Bulk API 2.0 ingestion run\"\"\"
    job_id: str
    records_processed: int
    success_count: int
    error_count: int
    p50_latency_ms: float
    p99_latency_ms: float
    total_cost_usd: float

class SalesforceBulkBenchmarker:
    \"\"\"Benchmarks Salesforce Bulk API 2.0 performance for 2026 org configurations\"\"\"

    def __init__(self, instance_url: str, access_token: str, api_version: str = \"v248.0\"):
        self.base_url = f\"{instance_url}/services/data/{api_version}\"
        self.headers = {
            \"Authorization\": f\"Bearer {access_token}\",
            \"Content-Type\": \"application/json\"
        }
        # Configure retry logic for transient Salesforce errors (rate limits, 503s)
        self.session = requests.Session()
        retry_strategy = Retry(
            total=5,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=[\"POST\", \"GET\", \"PATCH\"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        self.session.mount(\"https://\", adapter)
        self.session.headers.update(self.headers)

    def create_bulk_job(self, object_name: str, operation: str = \"insert\") -> Optional[str]:
        \"\"\"Creates a new Bulk API 2.0 job, returns job ID or None on failure\"\"\"
        job_payload = {
            \"object\": object_name,
            \"contentType\": \"JSON\",
            \"operation\": operation,
            \"lineEnding\": \"LF\"
        }
        try:
            response = self.session.post(
                f\"{self.base_url}/jobs/ingest\",
                json=job_payload,
                timeout=10
            )
            response.raise_for_status()
            job_id = response.json().get(\"id\")
            logger.info(f\"Created Bulk Job {job_id} for {object_name}\")
            return job_id
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to create Bulk Job: {str(e)}\")
            if response := getattr(e, \"response\", None):
                logger.error(f\"Response body: {response.text}\")
            return None

    def upload_job_data(self, job_id: str, records: List[Dict]) -> bool:
        \"\"\"Uploads JSON payload to an existing Bulk Job, returns success status\"\"\"
        upload_url = f\"{self.base_url}/jobs/ingest/{job_id}/batches\"
        # Split records into 10MB chunks to align with Salesforce 2026 max batch size
        chunk_size = 10000  # ~10MB for average contact records
        for i in range(0, len(records), chunk_size):
            chunk = records[i:i+chunk_size]
            try:
                response = self.session.put(
                    upload_url,
                    data=json.dumps(chunk),
                    headers={\"Content-Type\": \"application/json\"},
                    timeout=30
                )
                response.raise_for_status()
                logger.info(f\"Uploaded chunk {i//chunk_size + 1} for job {job_id}\")
            except requests.exceptions.RequestException as e:
                logger.error(f\"Failed to upload chunk for job {job_id}: {str(e)}\")
                return False
        return True

    def close_job_and_wait(self, job_id: str, poll_interval: int = 5) -> Optional[Dict]:
        \"\"\"Closes a Bulk Job and polls until completion, returns job status or None\"\"\"
        try:
            # Close the job to start processing
            close_resp = self.session.patch(
                f\"{self.base_url}/jobs/ingest/{job_id}\",
                json={\"state\": \"UploadComplete\"},
                timeout=10
            )
            close_resp.raise_for_status()

            # Poll for job completion
            start_time = time.time()
            while True:
                status_resp = self.session.get(
                    f\"{self.base_url}/jobs/ingest/{job_id}\",
                    timeout=10
                )
                status_resp.raise_for_status()
                status = status_resp.json()
                if status.get(\"state\") in [\"JobComplete\", \"Failed\"]:
                    elapsed = (time.time() - start_time) * 1000
                    logger.info(f\"Job {job_id} finished in {elapsed:.2f}ms with state {status.get('state')}\")
                    return status
                logger.info(f\"Job {job_id} state: {status.get('state')}, polling again...\")
                time.sleep(poll_interval)
        except requests.exceptions.RequestException as e:
            logger.error(f\"Failed to close/wait for job {job_id}: {str(e)}\")
            return None

    def run_benchmark(self, object_name: str, record_count: int = 100000) -> Optional[BulkIngestionResult]:
        \"\"\"Runs a full benchmark for ingesting N records, returns result or None\"\"\"
        # Generate dummy contact records for benchmarking
        records = [
            {
                \"FirstName\": f\"Benchmark{i}\",
                \"LastName\": \"Salesforce2026\",
                \"Email\": f\"benchmark{i}@test-salesforce-2026.com\"
            }
            for i in range(record_count)
        ]

        start_time = time.time()
        job_id = self.create_bulk_job(object_name)
        if not job_id:
            return None

        if not self.upload_job_data(job_id, records):
            return None

        final_status = self.close_job_and_wait(job_id)
        if not final_status:
            return None

        total_time_ms = (time.time() - start_time) * 1000
        # Calculate p50/p99 latency (simplified for example, real benchmark would use all record latencies)
        p50_latency = total_time_ms * 0.5
        p99_latency = total_time_ms * 0.99
        # Salesforce 2026 Bulk API cost: $0.0001 per 1000 records
        total_cost = (record_count / 1000) * 0.0001

        return BulkIngestionResult(
            job_id=job_id,
            records_processed=record_count,
            success_count=final_status.get(\"numberRecordsProcessed\", 0),
            error_count=final_status.get(\"numberRecordsFailed\", 0),
            p50_latency_ms=p50_latency,
            p99_latency_ms=p99_latency,
            total_cost_usd=total_cost
        )

if __name__ == \"__main__\":
    # Load credentials from environment variables (never hardcode in production!)
    required_vars = [\"SF_INSTANCE_URL\", \"SF_ACCESS_TOKEN\"]
    for var in required_vars:
        if var not in os.environ:
            raise ValueError(f\"Missing required environment variable: {var}\")

    benchmarker = SalesforceBulkBenchmarker(
        instance_url=os.environ[\"SF_INSTANCE_URL\"],
        access_token=os.environ[\"SF_ACCESS_TOKEN\"]
    )

    logger.info(\"Starting 100k record Bulk API 2.0 benchmark...\")
    result = benchmarker.run_benchmark(object_name=\"Contact\", record_count=100000)

    if result:
        print(json.dumps({
            \"job_id\": result.job_id,
            \"records_processed\": result.records_processed,
            \"success_rate\": f\"{(result.success_count / result.records_processed) * 100:.2f}%\",
            \"p50_latency_ms\": result.p50_latency_ms,
            \"p99_latency_ms\": result.p99_latency_ms,
            \"total_cost_usd\": f\"${result.total_cost_usd:.4f}\"
        }, indent=2))
    else:
        logger.error(\"Benchmark failed to complete\")
        exit(1)

import { AuthInfo, Connection, EventBus, EventBusConfig } from '@salesforce/core';
import { Logger, LogLevel } from '@salesforce/kit';
import { DeadLetterQueue } from './dead-letter-queue';
import { MetricsCollector } from './metrics-collector';
import { DateTime } from 'luxon';

// Initialize logger for production-grade visibility
Logger.setLevel(LogLevel.INFO);
const logger = Logger.child('SalesforceEventBusBenchmark');

// Configuration for 2026 Salesforce Event Bus (v248.0 compliant)
interface EventBusBenchmarkConfig {
  username: string;
  eventName: string;
  batchSize: number;
  maxRetries: number;
  pollIntervalMs: number;
}

class SalesforceEventBusSubscriber {
  private connection: Connection;
  private eventBus: EventBus;
  private metrics: MetricsCollector;
  private dlq: DeadLetterQueue;
  private config: EventBusBenchmarkConfig;
  private isRunning: boolean = false;

  constructor(config: EventBusBenchmarkConfig) {
    this.config = config;
    this.metrics = new MetricsCollector();
    this.dlq = new DeadLetterQueue();
  }

  /**
   * Initializes Salesforce connection using JWT auth (2026 recommended pattern)
   */
  async init(): Promise {
    try {
      const authInfo = await AuthInfo.create({
        username: this.config.username,
        oauth2Options: {
          // Loaded from environment variables in production
          clientId: process.env.SF_CLIENT_ID!,
          privateKey: process.env.SF_PRIVATE_KEY!,
          loginUrl: process.env.SF_LOGIN_URL || 'https://login.salesforce.com'
        }
      });
      await authInfo.save();
      this.connection = await Connection.create({ authInfo });
      logger.info(`Connected to Salesforce org ${this.connection.getAuthInfoFields().orgId}`);

      // Configure Event Bus with 2026 retry and batch settings
      const eventBusConfig: EventBusConfig = {
        eventName: this.config.eventName,
        batchSize: this.config.batchSize,
        pollInterval: this.config.pollIntervalMs,
        retryConfig: {
          maxRetries: this.config.maxRetries,
          backoffFactor: 2,
          initialDelayMs: 1000
        }
      };
      this.eventBus = await EventBus.create({ connection, eventBusConfig });
      logger.info(`Event Bus initialized for event ${this.config.eventName}`);
    } catch (error) {
      logger.error(`Failed to initialize Salesforce connection: ${error.message}`);
      throw error;
    }
  }

  /**
   * Starts subscribing to events and processing them with benchmark metrics
   */
  async start(): Promise {
    if (this.isRunning) {
      logger.warn('Event Bus subscriber is already running');
      return;
    }
    this.isRunning = true;
    logger.info(`Starting event subscription for ${this.config.eventName}`);

    try {
      await this.eventBus.subscribe(async (events) => {
        const processingStart = DateTime.now().toMillis();
        logger.info(`Received ${events.length} events to process`);

        for (const event of events) {
          try {
            // Process individual event (example: update Salesforce record)
            await this.processEvent(event);
            this.metrics.recordSuccess(event.replayId);
          } catch (error) {
            logger.error(`Failed to process event ${event.replayId}: ${error.message}`);
            this.metrics.recordFailure(event.replayId);
            // Send to DLQ after max retries exhausted (handled by EventBus retry config)
            await this.dlq.enqueue(event, error.message);
          }
        }

        const processingEnd = DateTime.now().toMillis();
        const latencyMs = processingEnd - processingStart;
        this.metrics.recordBatchLatency(latencyMs, events.length);
        logger.info(`Processed batch of ${events.length} events in ${latencyMs}ms`);
      });

      // Keep process running until SIGINT/SIGTERM
      process.on('SIGINT', () => this.stop());
      process.on('SIGTERM', () => this.stop());
    } catch (error) {
      logger.error(`Subscription failed: ${error.message}`);
      this.isRunning = false;
      throw error;
    }
  }

  /**
   * Processes a single Salesforce event with business logic
   */
  private async processEvent(event: any): Promise {
    // Example: Update a Contact record based on event payload
    const { contactId, newEmail } = event.payload;
    if (!contactId || !newEmail) {
      throw new Error('Missing required payload fields: contactId, newEmail');
    }

    const updateResult = await this.connection.sobject('Contact').update({
      Id: contactId,
      Email: newEmail
    });

    if (!updateResult.success) {
      throw new Error(`Contact update failed: ${JSON.stringify(updateResult.errors)}`);
    }
    logger.debug(`Updated Contact ${contactId} with new email ${newEmail}`);
  }

  /**
   * Stops the subscriber and flushes metrics
   */
  async stop(): Promise {
    if (!this.isRunning) return;
    logger.info('Stopping event subscriber...');
    this.isRunning = false;
    await this.eventBus.close();
    const summary = this.metrics.getSummary();
    logger.info(`Benchmark summary: ${JSON.stringify(summary, null, 2)}`);
    await this.dlq.flush();
    process.exit(0);
  }
}

// Run benchmark if executed directly
if (require.main === module) {
  const config: EventBusBenchmarkConfig = {
    username: process.env.SF_USERNAME!,
    eventName: 'ContactEmailUpdated__e',
    batchSize: 200, // 2026 Event Bus max batch size for standard events
    maxRetries: 3,
    pollIntervalMs: 500
  };

  const subscriber = new SalesforceEventBusSubscriber(config);
  subscriber.init()
    .then(() => subscriber.start())
    .catch((error) => {
      logger.error(`Fatal error: ${error.message}`);
      process.exit(1);
    });
}

import com.salesforce.grpc.bridge.v1.SalesforceGrpc;
import com.salesforce.grpc.bridge.v1.QueryRequest;
import com.salesforce.grpc.bridge.v1.QueryResponse;
import com.salesforce.grpc.bridge.v1.AuthRequest;
import com.salesforce.grpc.bridge.v1.AuthResponse;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.StatusRuntimeException;
import io.grpc.netty.shaded.io.grpc.netty.NettyChannelBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Benchmarks Salesforce gRPC Bridge (v1.2.0) query performance for 2026 orgs
 * Requires salesforce/grpc-salesforce-bridge running locally
 */
public class SalesforceGrpcBenchmark {
    private static final Logger logger = LoggerFactory.getLogger(SalesforceGrpcBenchmark.class);
    private static final String BRIDGE_HOST = System.getenv().getOrDefault(\"GRPC_BRIDGE_HOST\", \"localhost\");
    private static final int BRIDGE_PORT = Integer.parseInt(System.getenv().getOrDefault(\"GRPC_BRIDGE_PORT\", \"50051\"));
    private static final int WARMUP_QUERIES = 100;
    private static final int BENCHMARK_QUERIES = 1000;
    private static final String SOQL_QUERY = \"SELECT Id, Name, Email FROM Contact WHERE CreatedDate > LAST_N_DAYS:7 LIMIT 100\";

    private final ManagedChannel channel;
    private final SalesforceGrpc.SalesforceBlockingStub blockingStub;
    private final AtomicInteger successCount = new AtomicInteger(0);
    private final AtomicInteger failureCount = new AtomicInteger(0);
    private final List latencies = new ArrayList<>();

    public SalesforceGrpcBenchmark() {
        // Initialize gRPC channel with 2026 recommended TLS settings (disable for local dev only)
        this.channel = NettyChannelBuilder.forAddress(BRIDGE_HOST, BRIDGE_PORT)
                .usePlaintext() // Enable TLS in production with .sslContext()
                .build();
        this.blockingStub = SalesforceGrpc.newBlockingStub(channel);
    }

    /**
     * Authenticates with the gRPC bridge using a Salesforce access token
     */
    private String authenticate(String accessToken) throws StatusRuntimeException {
        AuthRequest authRequest = AuthRequest.newBuilder()
                .setAccessToken(accessToken)
                .build();
        AuthResponse authResponse = blockingStub.authenticate(authRequest);
        if (!authResponse.getSuccess()) {
            throw new RuntimeException(\"gRPC authentication failed: \" + authResponse.getErrorMessage());
        }
        logger.info(\"Authenticated with gRPC bridge, session ID: {}\", authResponse.getSessionId());
        return authResponse.getSessionId();
    }

    /**
     * Executes a single SOQL query via gRPC and records latency
     */
    private void executeQuery(String sessionId, String soql) {
        QueryRequest request = QueryRequest.newBuilder()
                .setSessionId(sessionId)
                .setSoql(soql)
                .build();

        long start = System.currentTimeMillis();
        try {
            QueryResponse response = blockingStub.executeQuery(request);
            long latency = System.currentTimeMillis() - start;
            latencies.add(latency);

            if (response.getSuccess()) {
                successCount.incrementAndGet();
                logger.debug(\"Query returned {} records in {}ms\", response.getRecordsCount(), latency);
            } else {
                failureCount.incrementAndGet();
                logger.error(\"Query failed: {}\", response.getErrorMessage());
            }
        } catch (StatusRuntimeException e) {
            failureCount.incrementAndGet();
            long latency = System.currentTimeMillis() - start;
            latencies.add(latency);
            logger.error(\"gRPC call failed: {}\", e.getStatus(), e);
        }
    }

    /**
     * Runs the full benchmark: warmup, then measured queries
     */
    public void runBenchmark(String accessToken) {
        try {
            String sessionId = authenticate(accessToken);
            logger.info(\"Starting warmup: {} queries\", WARMUP_QUERIES);
            for (int i = 0; i < WARMUP_QUERIES; i++) {
                executeQuery(sessionId, SOQL_QUERY);
            }
            logger.info(\"Warmup complete. Starting benchmark: {} queries\", BENCHMARK_QUERIES);

            latencies.clear();
            successCount.set(0);
            failureCount.set(0);

            for (int i = 0; i < BENCHMARK_QUERIES; i++) {
                executeQuery(sessionId, SOQL_QUERY);
            }

            printResults();
        } catch (Exception e) {
            logger.error(\"Benchmark failed\", e);
        } finally {
            shutdown();
        }
    }

    /**
     * Calculates and prints benchmark results
     */
    private void printResults() {
        if (latencies.isEmpty()) {
            logger.warn(\"No latencies recorded, skipping results\");
            return;
        }

        List sorted = new ArrayList<>(latencies);
        sorted.sort(Long::compareTo);
        double p50 = sorted.get((int) (sorted.size() * 0.5));
        double p99 = sorted.get((int) (sorted.size() * 0.99));
        double avg = latencies.stream().mapToLong(Long::longValue).average().orElse(0);
        double successRate = (successCount.get() / (double) BENCHMARK_QUERIES) * 100;

        logger.info(\"=== gRPC Bridge Benchmark Results ===\");
        logger.info(\"Total Queries: {}\", BENCHMARK_QUERIES);
        logger.info(\"Success Rate: {}%\", String.format(\"%.2f\", successRate));
        logger.info(\"P50 Latency: {}ms\", p50);
        logger.info(\"P99 Latency: {}ms\", p99);
        logger.info(\"Average Latency: {}ms\", String.format(\"%.2f\", avg));
        logger.info(\"Failure Count: {}\", failureCount.get());
    }

    /**
     * Shuts down the gRPC channel
     */
    private void shutdown() {
        try {
            channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
            logger.info(\"gRPC channel shut down\");
        } catch (InterruptedException e) {
            logger.error(\"Failed to shut down channel\", e);
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        if (args.length < 1) {
            logger.error(\"Usage: java SalesforceGrpcBenchmark \");
            System.exit(1);
        }
        String accessToken = args[0];
        SalesforceGrpcBenchmark benchmark = new SalesforceGrpcBenchmark();
        benchmark.runBenchmark(accessToken);
    }
}

// Node.js example using the official tooling client for Bulk API 2.0
const { ToolingClient } = require('@salesforce/tooling-api-client');
const client = new ToolingClient({
  accessToken: process.env.SF_ACCESS_TOKEN,
  instanceUrl: process.env.SF_INSTANCE_URL,
  apiVersion: 'v248.0'
});

// Create and execute a bulk job with automatic retry handling
async function runBulkJob() {
  const job = await client.bulk.createJob({
    object: 'Contact',
    operation: 'insert',
    contentType: 'JSON'
  });
  await job.uploadData([{ FirstName: 'Test', LastName: 'User' }]);
  const result = await job.waitForCompletion();
  console.log(`Processed ${result.numberRecordsProcessed} records`);
}

// Python example adding OpenTelemetry tracing to Salesforce REST calls
from opentelemetry import trace
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from salesforce_bulk import SalesforceBulk

# Enable automatic instrumentation for all requests calls
RequestsInstrumentor().instrument()

tracer = trace.get_tracer("salesforce.integrations")

def update_contact(contact_id, new_email):
    with tracer.start_as_current_span("salesforce.contact.update") as span:
        span.set_attribute("salesforce.object", "Contact")
        span.set_attribute("salesforce.record_id", contact_id)

        bulk = SalesforceBulk(
            session_id=os.environ["SF_ACCESS_TOKEN"],
            host=os.environ["SF_INSTANCE_URL"]
        )
        # Bulk API calls are automatically traced via RequestsInstrumentor
        job = bulk.create_insert_job("Contact", contentType="JSON")
        batch = bulk.post_batch(job, [{"Id": contact_id, "Email": new_email}])
        bulk.wait_for_batch(job, batch)

// Java example querying Salesforce via gRPC bridge
SalesforceGrpc.SalesforceBlockingStub stub = SalesforceGrpc.newBlockingStub(channel);
QueryRequest request = QueryRequest.newBuilder()
    .setSessionId(sessionId)
    .setSoql("SELECT Id, Name FROM Account LIMIT 1000")
    .build();
QueryResponse response = stub.executeQuery(request);
if (response.getSuccess()) {
    for (Record record : response.getRecordsList()) {
        System.out.println(record.getFieldsMap().get("Name").getStringValue());
    }
}