[streaming] source fault tolerance PersistenceStorage wip