anchor.diff:              cts.Cancel()
   
               [| stopProcessingLogSub; heartbeatLogSub; heartbeatSub; checkpointLogSub; fetchSub;
  -                receivedLogSub; processingLogSub; processingSub; processedLogSub; (cts :> IDisposable) |]
  +                receivedLogSub; processingLogSub; processingSub; processedLogSub; retryInitSub;
  +                (cts :> IDisposable) |]
               |> Array.iter (fun x -> x.Dispose())
   
               logDebug "Disposed." [||]
  
positive.code:                 | Failure exn -> initializationFailedEvent.Trigger(exn)
                  | Success status -> 
                      match status with
                      | NotFound -> logWarn "Shard is not found. Please check if it was manually deleted from DynamoDB." [||]
                      | Closed   -> shardClosedEvent.Trigger()
                      | New(workerId', _) when workerId' = app.WorkerId -> 
                          // the shard has not been processed before, so start from the oldest record
  
negative.code:         }
  
      // keep retrying failed initializations until it succeeds
      let retryInitSub = initializationFailedEvent.Publish
                          .TakeUntil(initializedEvent.Publish)
                          .Subscribe(fun _ -> Async.Start(init, cts.Token))
  
      do Async.Start(init, cts.Token)
  
  