Platform Events with Idempotent Consumers and Durable Replay Options (Salesforce)

In Salesforce event-driven systems, Platform Events are delivered at least once — meaning the same message can show up multiple times or even arrive out of order. That’s completely fine, as long as your consumers are idempotent (safe to process more than once) and you have a durable replay strategy to recover cleanly from downtime without missing anything.

This guide walks through practical techniques to make your events safe, resilient, and replayable — including how to add deduplication keys, persist checkpoints, and subscribe with replay support.

---

1) Idempotency: Achieving “Exactly-Once” Results on an At-Least-Once Bus

Every reliable event consumer needs a way to detect duplicates and ensure business logic only runs once per unique event.

Deduplication key:
Each event should carry a unique and deterministic identifier — something like OrderId + Version, or a producer-generated UUID.

Idempotency store:
Before making any database updates or callouts, your consumer should attempt to insert this key into a unique indexed store.

• If the insert succeeds → proceed with business logic.

• If it fails (duplicate key) → safely skip the event.

Transactional boundary:
Commit the idempotency claim and the business side effects in the same transaction. That way, you never risk applying one without the other.

Prefer natural keys over hashes:
Meaningful keys that can be recomputed by the producer are easier to reason about and less prone to collision than payload hashes.

---

2) Durable Replay and Retention Options

Salesforce Platform Events keep messages for a fixed retention window (typically 72 hours), each identified by a ReplayId that increases with every new event. When a subscriber reconnects, it can choose a replay option:

• LATEST: Start from new events only.

• EARLIEST: Replay everything still within retention.

• CUSTOM: Resume from a stored ReplayId checkpoint.

Client options:

• CometD (Streaming API): Use the Replay Extension to set the ReplayId per channel.

• Pub/Sub API (gRPC/REST): Set replayPreset or replayId directly. It’s faster, supports batching, and handles back-pressure better.

Checkpointing:
Always save the last successfully processed ReplayId after finishing business logic.
On restart, subscribe from that checkpoint to continue seamlessly. Even if duplicates are reprocessed, they’re handled safely by your idempotent consumer.

---

3) Ordering and Concurrency

Platform Events don’t guarantee global ordering, but you can design around it:

• Partition by key (e.g., one stream per OrderId) to keep related events in order.

• Serialize critical updates by locking on a single Idempotency__c row or a parent record if per-key ordering is essential.

• Avoid global serialization — it destroys throughput. Only lock per business key when truly necessary.

---

4) Error Handling Best Practices

A robust consumer separates retryable and terminal errors:

• Retry transient issues like timeouts or record locks.

• Log and dead-letter validation or data errors for later review.

Add observability:
Track how many events you’ve processed, the last ReplayId, timestamps, and the most recent error.
This helps confirm forward progress and simplifies troubleshooting.

---

Real-World Example (with Code)

Scenario

Let’s say your fulfillment system publishes Order_Event__e messages for order creation, updates, and shipments.
On the receiving end, an Apex trigger consumes those events and updates the related Order__c records.
We’ll make this consumer idempotent with a custom store and also demonstrate a durable replay subscriber using the Pub/Sub API.

---

A) Platform Event Definition & Publisher (Producer)

// Order_Event__e fields (example):
// – OrderId__c (Text, required)
// – EventType__c (Text: ‘Created’ | ‘Updated’ | ‘Shipped’, required)
// – DedupKey__c (Text, required, unique upstream if possible)
// – Payload__c (Long Text Area or JSON)

// Order_Event__e fields (example):
// - OrderId__c (Text, required)
// - EventType__c (Text: 'Created'|'Updated'|'Shipped', required)
// - DedupKey__c (Text, required, unique upstream if possible)
// - Payload__c (Long Text Area or JSON)

// Example publish
public with sharing class OrderEventPublisher {
    public static void publishOrderEvent(String orderId, String type, String dedupKey, Object payload) {
        Order_Event__e evt = new Order_Event__e(
            OrderId__c   = orderId,
            EventType__c = type,
            DedupKey__c  = dedupKey,
            Payload__c   = JSON.serialize(payload)
        );
        Database.SaveResult sr = EventBus.publish(evt);
        System.assert(sr.isSuccess(), 'Failed to publish PE: ' + sr);
    }
}

B) Idempotency Store (Custom Object)

// Custom object: Idempotency__c
// Fields:
// - Key__c (Text, External ID, Unique)
// - OrderId__c (Lookup or Text)
// - ConsumedAt__c (Datetime)
// - ReplayId__c (Text)  // optional: for observability

C) Apex Platform Event Trigger (Idempotent Consumer)

// OrderEventTrigger.trigger (PE triggers are AFTER INSERT)
trigger OrderEventTrigger on Order_Event__e (after insert) {
    List<Idempotency__c> toClaim = new List<Idempotency__c>();
    Set<String> keys = new Set<String>();
    for (Order_Event__e e : Trigger.New) {
        if (String.isBlank(e.DedupKey__c)) continue;
        keys.add(e.DedupKey__c);
        toClaim.add(new Idempotency__c(
            Key__c = e.DedupKey__c,
            OrderId__c = e.OrderId__c,
            ConsumedAt__c = System.now()
            // ReplayId__c cannot be read in Apex; leave blank or populate via middleware
        ));
    }

    // Try to claim keys. Unique constraint prevents double-processing.
    Database.SaveResult[] claims = Database.insert(toClaim, /* allOrNone */ false);

    // Build a map from DedupKey to event for business processing
    Map<String, Order_Event__e> byKey = new Map<String, Order_Event__e>();
    for (Order_Event__e e : Trigger.New) if (e.DedupKey__c != null) byKey.put(e.DedupKey__c, e);

    List<Order__c> updates = new List<Order__c>();
    for (Integer i = 0; i < claims.size(); i++) {
        Database.SaveResult r = claims[i];
        Idempotency__c claimed = toClaim[i];
        if (!r.isSuccess()) {
            // Duplicate key → already processed. Skip safely.
            Boolean duplicate = false;
            for (Database.Error err : r.getErrors())
                if (err.getStatusCode() == StatusCode.DUPLICATE_VALUE) duplicate = true;
            if (duplicate) continue; else throw new DmlException('Idempotency claim failed: ' + r);
        }

        // First claim: safe to apply business side effects
        Order_Event__e evt = byKey.get(claimed.Key__c);
        if (evt == null) continue;

        if (evt.EventType__c == 'Created') {
            updates.add(new Order__c(
                External_Id__c = evt.OrderId__c,
                Status__c = 'Open'
            ));
        } else if (evt.EventType__c == 'Shipped') {
            updates.add(new Order__c(
                External_Id__c = evt.OrderId__c,
                Status__c = 'Shipped',
                Shipped_At__c = System.now()
            ));
        } else if (evt.EventType__c == 'Updated') {
            // parse JSON payload minimally
            Map<String, Object> p = (Map<String, Object>) JSON.deserializeUntyped(evt.Payload__c);
            updates.add(new Order__c(
                External_Id__c = evt.OrderId__c,
                Amount__c = (Decimal) p.get('amount')
            ));
        }
    }

    if (!updates.isEmpty()) {
        // Upsert by external id avoids duplicates and is bulk-safe
        Database.UpsertResult[] res = Database.upsert(updates, Order__c.Fields.External_Id__c, false);
        // Optional: collect and log row-level errors instead of throwing
    }
}

Why it’s idempotent:
The consumer inserts a unique DedupKey__c before doing anything else. If the same message replays, the insert fails with a DUPLICATE_VALUE, and the event is skipped safely — no side effects.

---

D) Durable Replay Subscriber (Pub/Sub API – Node.js Pseudo-Code)

This pattern is ideal for middleware or long-running worker services. It stores a replay checkpoint and resumes from the last processed ReplayId when restarted.

// Pseudo-code for Pub/Sub API (Node.js-like)
const channel = '/event/Order_Event__e';
let checkpoint = await repo.loadReplayId(channel); // e.g., from Postgres or Custom Metadata

const request = {
  topicName: channel,
  numRequested: 100,
  replayId: checkpoint || null,     // when null, use preset below
  replayPreset: checkpoint ? 'CUSTOM' : 'LATEST' // or 'EARLIEST'
};

const stream = await pubsub.subscribe(request);

for await (const msg of stream) {
  const replayId = msg.replayId; // Buffer/bytes
  const evt = JSON.parse(msg.payload.Payload__c);

  // Start a DB tx; write idempotency key first
  const claimed = await repo.tryClaim(evt.DedupKey__c);
  if (claimed) {
    await applyBusinessChanges(evt); // Upserts in your target system
    await repo.saveCheckpoint(channel, replayId); // commit checkpoint AFTER success
  } else {
    // Already processed → still advance the checkpoint
    await repo.saveCheckpoint(channel, replayId);
  }
}

Key Takeaways

• Checkpoint after success: Only advance your ReplayId once business work completes.

• CUSTOM replay allows subscribers to resume exactly where they left off.

• Batch size and flow control help tune throughput while avoiding overload.

• 

---

Final Thoughts

Design for at-least-once delivery by making every consumer idempotent.
Use a unique deduplication key and a claim-then-act approach to guarantee safe processing.
Persist replay checkpoints and subscribe with CUSTOM replay to recover cleanly after outages.
Serialize only where necessary and log enough metrics to prove progress.

When done right, Platform Events become a rock-solid backbone for Salesforce event-driven architectures — reliable, replayable, and safe at scale.