self._submit(self._persist_run_single, run.copy(deep=True))
    def _on_retriever_end(self, run: Run) -> None:
        """Process the Retriever Run."""
        self._submit(self._update_run_single, run.copy(deep=True))
    def _on_retriever_error(self, run: Run) -> None:
        """Process the Retriever Run upon error."""
        self._submit(self._update_run_single, run.copy(deep=True))
[docs]    def wait_for_futures(self) -> None:
        """Wait for the given futures to complete."""
        futures = list(self._futures)
        wait(futures)
        for future in futures:
            self._futures.remove(future)