At-Most-Once Delivery

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

// At-most-once delivery: Producer sends without confirmation
class AtMostOnceProducer {
    private broker: MessageBroker;
 
    constructor(broker: MessageBroker) {
        this.broker = broker;
    }
 
    /**
     * Send a message with at-most-once semantics.
     * 
     * Key characteristics:
     * - No acknowledgment awaited
     * - No retry on failure
     * - Message may be lost
     * - Extremely fast
     */
    send(topic: string, message: Message): void {
        // Fire and forget - no waiting for acknowledgment
        this.broker.publish(topic, message);
        
        // Producer immediately continues
        // No way to know if message was received
        // No retry mechanism
    }
}
 
// Usage - the producer doesn't know if this succeeded
producer.send("metrics-topic", {
    type: "cpu_usage",
    value: 45.2,
    timestamp: Date.now()
});
 
// Execution continues immediately regardless of delivery success

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

// Pattern 1: No acknowledgment at all
class FireAndForgetBroker {
    private consumers: Map<string, Consumer[]> = new Map();
 
    publish(topic: string, message: Message): void {
        const topicConsumers = this.consumers.get(topic) || [];
        
        for (const consumer of topicConsumers) {
            // Send and immediately forget
            // No tracking, no retry, no delivery confirmation
            try {
                consumer.receive(message);
            } catch (error) {
                // Log and continue - message is lost
                console.warn(`Delivery failed: ${error.message}`);
            }
        }
        
        // Message is discarded after single delivery attempt
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

// Pattern 2: Acknowledge before processing
class PreAckConsumer {
    private broker: MessageBroker;
 
    async consumeMessage(): Promise<void> {
        const message = await this.broker.receive();
        
        // CRITICAL: Acknowledge BEFORE processing
        // If we crash after this, message is lost
        await this.broker.acknowledge(message.id);
        
        // Now process the message
        // If this throws, the message is already acknowledged
        // and will NOT be redelivered
        try {
            await this.processMessage(message);
        } catch (error) {
            // Message processing failed but it's already ACK'd
            // Message is effectively lost - no retry will occur
            console.error(`Processing failed, message lost: ${error.message}`);
        }
    }
 
    private async processMessage(message: Message): Promise<void> {
        // Business logic here
        // If this crashes, message is gone forever
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

/**
 * Calculate expected message loss for at-most-once delivery
 * 
 * Assume 0.1% loss rate (optimistic for production systems)
 */
function calculateExpectedLoss(messagesPerDay: number, lossRate: number = 0.001): void {
    const dailyLoss = messagesPerDay * lossRate;
    const monthlyLoss = dailyLoss * 30;
    const yearlyLoss = dailyLoss * 365;
 
    console.log(`At ${messagesPerDay.toLocaleString()} messages/day with ${lossRate * 100}% loss rate:`);
    console.log(`  Daily loss: ${dailyLoss.toLocaleString()} messages`);
    console.log(`  Monthly loss: ${monthlyLoss.toLocaleString()} messages`);
    console.log(`  Yearly loss: ${yearlyLoss.toLocaleString()} messages`);
}
 
// Small system: 100K messages/day
calculateExpectedLoss(100_000);
// Daily loss: 100 messages
// Monthly loss: 3,000 messages
// Yearly loss: 36,500 messages
 
// Medium system: 10M messages/day  
calculateExpectedLoss(10_000_000);
// Daily loss: 10,000 messages
// Monthly loss: 300,000 messages
// Yearly loss: 3,650,000 messages
 
// Large system: 1B messages/day
calculateExpectedLoss(1_000_000_000);
// Daily loss: 1,000,000 messages
// Monthly loss: 30,000,000 messages
// Yearly loss: 365,000,000 messages

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

// Latency comparison for different delivery semantics
interface LatencyBreakdown {
    networkLatency: number;           // One-way network time
    serializationTime: number;        // Message encoding
    brokerPersistence?: number;       // Disk write time
    ackRoundTrip?: number;            // Acknowledgment latency
    retryOverhead?: number;           // Average retry cost
}
 
// At-Most-Once Delivery
const atMostOnceLatency: LatencyBreakdown = {
    networkLatency: 1,              // 1ms
    serializationTime: 0.1,         // 0.1ms
    // No persistence
    // No acknowledgment
    // No retry
};
// Total: ~1.1ms per message
 
// At-Least-Once Delivery
const atLeastOnceLatency: LatencyBreakdown = {
    networkLatency: 1,              // 1ms
    serializationTime: 0.1,         // 0.1ms
    brokerPersistence: 5,           // 5ms (SSD fsync)
    ackRoundTrip: 2,                // 2ms (network round-trip)
    retryOverhead: 0.3,             // 0.3ms (amortized retry cost)
};
// Total: ~8.4ms per message
 
// At-most-once is approximately 7.6x faster in this example
const speedup = (1 + 0.1 + 5 + 2 + 0.3) / (1 + 0.1);
console.log(`At-most-once is ~${speedup.toFixed(1)}x faster`);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

// At-most-once is perfect for high-frequency metrics
class MetricsCollector {
    private broker: AtMostOnceBroker;
    private sampleRate: number = 0.01; // 1% sampling
 
    /**
     * Collect request latency metric.
     * 
     * Perfect for at-most-once because:
     * 1. We sample anyway (don't need 100% of data)
     * 2. Aggregate statistics tolerant to missing points
     * 3. High volume (millions/sec) - can't afford ACK overhead
     * 4. Next measurement replaces value of lost one
     */
    recordLatency(endpoint: string, latencyMs: number): void {
        // Sample to reduce volume
        if (Math.random() > this.sampleRate) {
            return;
        }
 
        // Fire-and-forget - perfectly appropriate here
        this.broker.send("metrics.latency", {
            endpoint,
            latencyMs,
            timestamp: Date.now(),
        });
        
        // Don't wait, don't retry, continue serving requests
    }
}
 
// This is called on every request - performance is critical
// Losing 0.1% of already-sampled metrics is completely acceptable

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

// ❌ ANTI-PATTERN: At-most-once for critical business events
class PaymentProcessor {
    private broker: AtMostOnceBroker;
 
    // THIS IS DANGEROUS!
    async processPayment(order: Order, payment: Payment): Promise<void> {
        // Charge the customer
        await this.paymentGateway.charge(payment);
        
        // Fire and forget the fulfillment event
        // ❌ If this message is lost:
        //    - Customer is charged
        //    - Order is never fulfilled
        //    - No automatic recovery
        //    - Manual investigation required
        this.broker.send("fulfillment.orders", {
            orderId: order.id,
            customerId: order.customerId,
            items: order.items,
        });
        
        // Customer is charged but may never receive their order!
    }
}
 
// ✅ CORRECT: Use at-least-once with idempotent consumer
class ReliablePaymentProcessor {
    private broker: AtLeastOnceBroker;
 
    async processPayment(order: Order, payment: Payment): Promise<void> {
        // Use transactional outbox pattern (covered in Module 4)
        await this.transactionalOutbox.executeWithEvent(
            () => this.paymentGateway.charge(payment),
            {
                topic: "fulfillment.orders",
                message: { orderId: order.id, customerId: order.customerId, items: order.items }
            }
        );
        // Now either both succeed or both fail - no lost orders
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

import * as dgram from 'dgram';
 
// UDP provides pure at-most-once semantics at the transport layer
class UDPMetricsSender {
    private socket: dgram.Socket;
    private metricsHost: string;
    private metricsPort: number;
 
    constructor(host: string, port: number) {
        this.socket = dgram.createSocket('udp4');
        this.metricsHost = host;
        this.metricsPort = port;
    }
 
    /**
     * Send a metric via UDP.
     * 
     * - No connection establishment (faster)
     * - No delivery confirmation (lighter)
     * - Packet may be lost, duplicated, or reordered by network
     * - Perfect for high-frequency, loss-tolerant telemetry
     */
    sendMetric(name: string, value: number): void {
        const message = Buffer.from(`${name}:${value}|${Date.now()}`);
        
        // Send and forget - no callback for delivery confirmation
        this.socket.send(message, this.metricsPort, this.metricsHost);
        
        // Execution continues immediately
        // We have no idea if the packet arrived
    }
}
 
// StatsD, Prometheus's pushgateway, and many metrics systems use UDP
const metrics = new UDPMetricsSender('statsd.internal', 8125);
metrics.sendMetric('api.requests.count', 1);
metrics.sendMetric('api.latency.p99', 145);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

import { Kafka } from 'kafkajs';
 
const kafka = new Kafka({
    clientId: 'metrics-producer',
    brokers: ['kafka-1:9092', 'kafka-2:9092'],
});
 
// At-most-once producer configuration
const producer = kafka.producer({
    // Don't wait for any acknowledgments
    // Message may be lost if broker hasn't received it
    acks: 0,  // ← KEY SETTING for at-most-once
    
    // No point in retries without acks
    retry: {
        retries: 0,  // No retries
    },
});
 
// Consumer with auto-commit (at-most-once on consumer side)
const consumer = kafka.consumer({ groupId: 'metrics-consumer' });
 
await consumer.subscribe({ topic: 'metrics' });
 
await consumer.run({
    // Commit offsets as soon as messages are received
    // If processing fails after commit, messages are lost
    autoCommit: true,
    autoCommitInterval: 100, // Aggressive auto-commit
    
    eachMessage: async ({ message }) => {
        // Message offset already committed
        // If we crash here, this message is "lost" (won't be reprocessed)
        await processMetric(message.value);
    },
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

import * as mqtt from 'mqtt';
 
const client = mqtt.connect('mqtt://broker.local');
 
// QoS 0: At-most-once delivery
// - Fastest delivery option
// - No delivery confirmation
// - Message may be lost
// - Perfect for sensor data with high frequency
client.publish(
    'sensors/temperature/room1',
    JSON.stringify({ value: 22.5, unit: 'celsius' }),
    { qos: 0 }  // ← QoS level 0 = at-most-once
);
 
// MQTT QoS levels summary:
// QoS 0: At-most-once  - Fire and forget
// QoS 1: At-least-once - Delivered at least once, may duplicate
// QoS 2: Exactly-once  - Delivered exactly once (4-way handshake)