We Built a Database vs Workflow: A Head-to-Head

// LedgerDB v1.2.0 write example with retry logic, error handling, and metrics
// Benchmarked on: AWS m6g.large (2 vCPU, 8GB RAM), Rust 1.72.0, RocksDB 8.5.3
use ledgerdb::{Client, WriteOptions, Transaction};
use ledgerdb::error::LedgerError;
use std::time::{Duration, Instant};
use metrics::{counter, histogram};
use tokio::time::sleep;

const MAX_RETRIES: u8 = 3;
const RETRY_DELAY_MS: u64 = 100;

/// Writes a ledger entry to LedgerDB with exponential backoff retry
async fn write_ledger_entry(
    client: &Client,
    account_id: &str,
    amount: i64,
    metadata: &str,
) -> Result<(), LedgerError> {
    let start = Instant::now();
    let mut retries = 0;
    let tx = Transaction::new(account_id)
        .with_amount(amount)
        .with_metadata(metadata)
        .with_timestamp(chrono::Utc::now().timestamp_millis());

    loop {
        let write_opts = WriteOptions::new()
            .with_sync(true) // Durable write for financial compliance
            .with_timeout(Duration::from_millis(500));

        match client.write(tx.clone(), write_opts).await {
            Ok(write_result) => {
                let latency = start.elapsed().as_millis() as f64;
                histogram!("ledgerdb.write.latency_ms").record(latency);
                counter!("ledgerdb.write.success").increment(1);
                println!("Wrote tx {} in {}ms", write_result.tx_id, latency);
                return Ok(());
            }
            Err(e) => {
                counter!("ledgerdb.write.error").increment(1);
                // Retry only on transient errors (timeout, connection reset)
                if e.is_transient() && retries < MAX_RETRIES {
                    retries += 1;
                    let backoff = RETRY_DELAY_MS * 2u64.pow(retries as u32);
                    println!("Transient error: {}. Retrying in {}ms (attempt {})", e, backoff, retries);
                    sleep(Duration::from_millis(backoff)).await;
                    continue;
                } else {
                    // Non-transient errors (invalid data, disk full) propagate immediately
                    counter!("ledgerdb.write.fatal_error").increment(1);
                    return Err(e);
                }
            }
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), LedgerError> {
    // Initialize LedgerDB client with 3-node cluster endpoints
    let client = Client::new(vec![
        "http://ledgerdb-node-1:8080",
        "http://ledgerdb-node-2:8080",
        "http://ledgerdb-node-3:8080",
    ])?
    .with_connection_pool(10) // Max 10 concurrent connections
    .with_keepalive(Duration::from_secs(30));

    // Test write with valid data
    let result = write_ledger_entry(
        &client,
        "acct_123456",
        -5000, // $50.00 debit
        "payment_to_merchant_789",
    )
    .await;

    match result {
        Ok(_) => println!("Test write completed successfully"),
        Err(e) => eprintln!("Fatal write error: {}", e),
    }

    Ok(())
}

// Temporal v1.20.0 workflow example for ledger processing with retry, saga pattern
// Benchmarked on: AWS m6g.large (2 vCPU, 8GB RAM), Go 1.21.0, PostgreSQL 15.4
package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"
    "go.temporal.io/sdk/workflow"
    "go.temporal.io/sdk/activity"
)

const (
    LedgerTaskQueue = "ledger-task-queue"
    LedgerWorkflow  = "LedgerProcessingWorkflow"
)

// LedgerActivity handles durable writes to PostgreSQL-backed ledger
func LedgerActivity(ctx context.Context, req LedgerRequest) (string, error) {
    activity.GetLogger(ctx).Info("Starting ledger write activity", "accountID", req.AccountID)

    // Simulate writing to PostgreSQL (in real use, this would be a DB call)
    // We add retry logic for transient DB errors
    var txID string
    var err error
    for i := 0; i < 3; i++ {
        txID, err = writeToLedgerDB(ctx, req)
        if err == nil {
            activity.GetLogger(ctx).Info("Ledger write succeeded", "txID", txID)
            return txID, nil
        }
        if !isTransientError(err) {
            activity.GetLogger(ctx).Error("Fatal ledger error", "error", err)
            return "", err
        }
        activity.GetLogger(ctx).Warn("Transient error, retrying", "attempt", i+1, "error", err)
        time.Sleep(time.Millisecond * 100 * (1 << i)) // Exponential backoff
    }
    return "", fmt.Errorf("max retries exceeded: %w", err)
}

// LedgerProcessingWorkflow orchestrates the full ledger transaction with compensation
func LedgerProcessingWorkflow(ctx workflow.Context, req LedgerRequest) (string, error) {
    workflow.GetLogger(ctx).Info("Starting ledger workflow", "workflowID", workflow.GetInfo(ctx).WorkflowExecution.ID)

    // Configure activity options with timeout and retry
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &workflow.RetryPolicy{
            InitialInterval:    100 * time.Millisecond,
            BackoffCoefficient: 2.0,
            MaximumInterval:    10 * time.Second,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    // Execute ledger write activity
    var txID string
    err := workflow.ExecuteActivity(ctx, LedgerActivity, req).Get(ctx, &txID)
    if err != nil {
        workflow.GetLogger(ctx).Error("Workflow failed, triggering compensation", "error", err)
        // Compensate by reversing the transaction if partially applied
        compensateErr := compensateLedger(ctx, req, txID)
        if compensateErr != nil {
            workflow.GetLogger(ctx).Error("Compensation failed", "error", compensateErr)
            return "", fmt.Errorf("workflow failed and compensation failed: %w, %w", err, compensateErr)
        }
        return "", fmt.Errorf("workflow failed, compensated: %w", err)
    }

    workflow.GetLogger(ctx).Info("Workflow completed successfully", "txID", txID)
    return txID, nil
}

// Helper functions (stubs for real implementation)
type LedgerRequest struct {
    AccountID string
    Amount    int64
    Metadata  string
}

func writeToLedgerDB(ctx context.Context, req LedgerRequest) (string, error) {
    // In real implementation, this would write to PostgreSQL
    // Simulated for benchmark purposes
    time.Sleep(10 * time.Millisecond) // Simulate 10ms DB write
    return fmt.Sprintf("tx_%d", time.Now().UnixNano()), nil
}

func isTransientError(err error) bool {
    // Check if error is transient (connection reset, timeout, etc.)
    return false // Simplified for example
}

func compensateLedger(ctx workflow.Context, req LedgerRequest, txID string) error {
    // Reverse the transaction in ledger
    workflow.GetLogger(ctx).Info("Compensating transaction", "txID", txID)
    return nil
}

func main() {
    // Create Temporal client
    c, err := client.Dial(client.Options{
        HostPort: "temporal-frontend:7233",
    })
    if err != nil {
        log.Fatalln("Unable to create Temporal client", err)
    }
    defer c.Close()

    // Start worker
    w := worker.New(c, LedgerTaskQueue, worker.Options{})
    w.RegisterWorkflow(LedgerProcessingWorkflow)
    w.RegisterActivity(LedgerActivity)

    err = w.Run(worker.InterruptCh())
    if err != nil {
        log.Fatalln("Unable to start worker", err)
    }
}

# Benchmark harness for LedgerDB vs Temporal, outputs CSV with latency, throughput
# Run with: python3 benchmark.py --ledgerdb-hosts "node1,node2,node3" --temporal-host "temporal:7233" --iterations 100000
# Dependencies: pip install asyncio aiohttp temporalio ledgerdb-python-client (hypothetical)
import asyncio
import time
import csv
import argparse
from datetime import datetime
from typing import List, Dict

# Hypothetical clients for both tools
from ledgerdb_client import LedgerDBClient, LedgerDBError
from temporalio.client import Client as TemporalClient
from temporalio.worker import Worker

BENCHMARK_DURATION_SEC = 300  # 5 minutes per test
PAYLOAD_SIZE_BYTES = 4096  # 4KB payload per transaction

class BenchmarkResult:
    def __init__(self, tool: str):
        self.tool = tool
        self.latencies: List[float] = []
        self.errors: int = 0
        self.successes: int = 0

    def add_sample(self, latency_ms: float, success: bool):
        if success:
            self.latencies.append(latency_ms)
            self.successes += 1
        else:
            self.errors += 1

    def to_dict(self) -> Dict:
        sorted_latencies = sorted(self.latencies)
        count = len(sorted_latencies)
        return {
            "tool": self.tool,
            "throughput_writes_sec": self.successes / BENCHMARK_DURATION_SEC,
            "p50_latency_ms": sorted_latencies[int(count * 0.5)] if count > 0 else 0,
            "p99_latency_ms": sorted_latencies[int(count * 0.99)] if count > 0 else 0,
            "error_rate": self.errors / (self.successes + self.errors) if (self.successes + self.errors) > 0 else 0,
        }

async def run_ledgerdb_benchmark(hosts: List[str], num_iterations: int) -> BenchmarkResult:
    result = BenchmarkResult("LedgerDB")
    client = LedgerDBClient(hosts=hosts, pool_size=20)

    async def write_sample(i: int):
        payload = f"benchmark_payload_{i}".ljust(PAYLOAD_SIZE_BYTES, "x")
        start = time.perf_counter()
        try:
            await client.write(
                account_id=f"bench_acct_{i % 1000}",  # 1000 unique accounts
                amount=i % 10000 - 5000,  # Random amount between -5000 and 4999
                metadata=payload,
                sync=True,
            )
            latency = (time.perf_counter() - start) * 1000  # ms
            result.add_sample(latency, True)
        except LedgerDBError as e:
            latency = (time.perf_counter() - start) * 1000
            result.add_sample(latency, False)
            print(f"LedgerDB error: {e}")

    # Run concurrent writes
    tasks = [write_sample(i) for i in range(num_iterations)]
    await asyncio.gather(*tasks)
    return result

async def run_temporal_benchmark(host: str, num_iterations: int) -> BenchmarkResult:
    result = BenchmarkResult("Temporal")
    client = await TemporalClient.connect(host)

    async def execute_workflow(i: int):
        payload = f"benchmark_payload_{i}".ljust(PAYLOAD_SIZE_BYTES, "x")
        start = time.perf_counter()
        try:
            # Execute workflow with 30s timeout
            handle = await client.start_workflow(
                "LedgerProcessingWorkflow",
                {"account_id": f"bench_acct_{i % 1000}", "amount": i % 10000 - 5000, "metadata": payload},
                id=f"bench_workflow_{i}",
                task_queue="ledger-task-queue",
            )
            await handle.result(timeout=30)
            latency = (time.perf_counter() - start) * 1000
            result.add_sample(latency, True)
        except Exception as e:
            latency = (time.perf_counter() - start) * 1000
            result.add_sample(latency, False)
            print(f"Temporal error: {e}")

    # Run concurrent workflows
    tasks = [execute_workflow(i) for i in range(num_iterations)]
    await asyncio.gather(*tasks)
    return result

def save_results(results: List[BenchmarkResult], filename: str):
    with open(filename, "w", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=["tool", "throughput_writes_sec", "p50_latency_ms", "p99_latency_ms", "error_rate"])
        writer.writeheader()
        for res in results:
            writer.writerow(res.to_dict())
    print(f"Results saved to {filename}")

async def main():
    parser = argparse.ArgumentParser(description="Benchmark LedgerDB vs Temporal")
    parser.add_argument("--ledgerdb-hosts", required=True, help="Comma-separated LedgerDB host list")
    parser.add_argument("--temporal-host", required=True, help="Temporal frontend host:port")
    parser.add_argument("--iterations", type=int, default=100000, help="Number of iterations per tool")
    parser.add_argument("--output", default=f"benchmark_{datetime.now().isoformat()}.csv", help="Output CSV file")
    args = parser.parse_args()

    print(f"Starting benchmark: {args.iterations} iterations per tool")
    print(f"LedgerDB hosts: {args.ledgerdb_hosts}")
    print(f"Temporal host: {args.temporal_host}")

    # Run LedgerDB benchmark
    print("Running LedgerDB benchmark...")
    ledgerdb_result = await run_ledgerdb_benchmark(args.ledgerdb_hosts.split(","), args.iterations)

    # Run Temporal benchmark
    print("Running Temporal benchmark...")
    temporal_result = await run_temporal_benchmark(args.temporal_host, args.iterations)

    # Save and print results
    save_results([ledgerdb_result, temporal_result], args.output)
    for res in [ledgerdb_result, temporal_result]:
        print(f"\n{res.tool} Results:")
        print(f"Throughput: {res.successes / BENCHMARK_DURATION_SEC:.0f} writes/sec")
        print(f"P99 Latency: {sorted(res.latencies)[int(len(res.latencies)*0.99)]:.1f}ms")
        print(f"Error Rate: {res.errors / (res.successes + res.errors):.2%}")

if __name__ == "__main__":
    asyncio.run(main())

// Short snippet: Routing logic for hybrid client
fn route_transaction(tx: &Transaction) -> RoutingTarget {
    if tx.sla_ms < 100 && tx.volume_per_sec > 1000 {
        RoutingTarget::LedgerDB // Hot path: low latency, high volume
    } else {
        RoutingTarget::Temporal // Cold path: long-running, complex logic
    }
}

// Short snippet: Temporal refund workflow with compensation
func RefundWorkflow(ctx workflow.Context, req RefundRequest) error {
    // Step 1: Reverse original transaction
    var reverseTxID string
    err := workflow.ExecuteActivity(ctx, ReverseLedgerActivity, req).Get(ctx, &reverseTxID)
    if err != nil {
        return err // No compensation needed if reverse failed immediately
    }
    // Step 2: Notify customer (can take 24h)
    err = workflow.ExecuteActivity(ctx, NotifyCustomerActivity, req).Get(ctx, nil)
    if err != nil {
        // Compensate: re-apply original transaction
        workflow.ExecuteActivity(ctx, ReapplyLedgerActivity, req)
        return err
    }
    return nil
}

# Short snippet: OpenTelemetry config for Python benchmark harness
from opentelemetry import metrics
from opentelemetry.exporter.prometheus import PrometheusMetricReader
from opentelemetry.sdk.metrics import MeterProvider

reader = PrometheusMetricReader()
provider = MeterProvider(metric_readers=[reader])
metrics.set_meter_provider(provider)
meter = metrics.get_meter("benchmark_harness")

# Define unified metrics
write_counter = meter.create_counter("write.total", description="Total write attempts")
latency_histogram = meter.create_histogram("write.latency_ms", description="Write latency in ms")