Platform Events process old data

Question

Sometimes when I publish platform event on change of certain Record, Platform Event Trigger fires with old data. If I publish event on insert, sometimes record is not visible for query.

To test it on clean Salesforce Org I created new scratch org without any custom logic. I followed those steps:

1. Create Platform Event (PE__e) with fields for WorkOrder Id and for information whether it is Insert or Update
2. Create Trigger on Work Order to publish platform events on After Insert/Update
3. Create object LOG with fields for Work Order Description and for trigger context.
4. Create Trigger for PE__e which is querying Work Orders with given ids, and then inserts all logs with queried Work Order Description and Trigger context

5. Run following code:

   List<WorkOrder> wos = new List<WorkOrder>();
   
   for(Integer i = 0; i < 100; i++){
       WorkOrder wo = new WorkOrder();
       wo.Description = 'A_' + i;
       wo.Status = 'New';
       wos.add(wo);
   }
   
   insert wos;
   

6. Run following code:

      List<WorkOrder> wors = [SELECT Id, Status, Description FROM WorkOrder]; 
      for(WorkOrder wo : wors){
          wo.Status = 'In Progress';
          wo.Description = wo.Description + '|a';
      }
   
      update wors;
   

7. Repeat step 6.

Result:

As you can see some of records are processed with updated Description and some are not. Is there any soloution for this? Or maybe it is expected behavior?

Work Order Trigger:

trigger WOR_TH on WorkOrder (
    after insert,
    after update) {
    List<PF__e> pfs = new List<PF__e>();
    for(WorkOrder wo : Trigger.new){
        if(Trigger.isInsert){
            pfs.add(new PF__e(Data__c = wo.Id, Additional__c = 'Insert'));   
        } 
        if(Trigger.isUpdate){
            pfs.add(new PF__e(Data__c = wo.Id, Additional__c = 'Update'));   
        } 
    }

    System.debug(pfs);

    EventBus.publish(pfs);
}

PF__e trigger:

trigger PF_TH on PF__e (after insert) {
    String query = 'SELECT ';
    try{
        List<Log__c> logs = new List<Log__c>();
        Map<Id, String> ids = new Map<Id, String>();

        for(PF__e e : Trigger.new){
            ids.put(e.Data__c, e.ADDITIONAL__c);
        }
        insert new Log__c(log__c = JSON.serialize(ids));

        Schema.DescribeSObjectResult a_desc = WorkOrder.sObjectType.getDescribe(); 
        Map<String, Schema.SObjectField> a_fields = a_desc.fields.getMap();
        for(Schema.sObjectField fld : a_fields.values()){ 
            query += fld.getDescribe().getName() + ',';
        }

        query = query.removeEnd(',');
        List<Id> idsList = new List<Id>(ids.keySet());
        query += ' FROM WorkOrder WHERE Id in :idsList';

        for(WorkOrder wo : Database.query(query)){
            Log__c log = new Log__c();        
            log.log__c = wo.Description;
            log.ADDITIONAL__c = ids.get(wo.Id);
            logs.add(log);
        }

        insert logs;
    }catch(Exception e){
        insert new Log__c(log__c = e.getMessage() + ' || ' + e.getStackTraceString() + ' || ' + query);
    }
}

Answer 1

Sometimes I have witnessed platform events being fired and their trigger is run even before the original transaction who fired the event hasn't committed data back to the database.

For an insert, if you can't find that record in the database, you can throw RetryableException. That trigger will run automatically after some time.

throw new EventBus.RetryableException(
                     'Record yet to be committed back to the database');

For an Update: You can pass LastmodifiedDate as a field on platform event, when you query that record if lastModifiedDate is lower than what present in your platform event's field retry it again. This will make sure that trigger doesn't run on old data.

SRC: https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_subscribe_apex_refire.htm

EDIT: I raised this on Platform Events Chatter group and got response as its the desired behavior, and they have included this as in consideration. The consideration says to use future/queuable or scheduler to publish platform event which for me defeats the purpose.

https://developer.salesforce.com/docs/atlas.en-us.platform_events.meta/platform_events/platform_events_considerations_decoupled_processes.htm

Answer 2

For the sake of knowledge, seems that the newly introduced in Summer 19' Publish Behavior does solve this issue:

https://releasenotes.docs.salesforce.com/en-us/summer19/release-notes/rn_platfom_event_post_commit.htm