void recover0(WriteAheadLog deltaWAL, StreamManager streamManager, ObjectManager objectManager,
        Logger logger) throws InterruptedException, ExecutionException {
        List<StreamMetadata> streams = streamManager.getOpeningStreams().get();
        Map<Long, Long> streamEndOffsets = streams.stream().collect(Collectors.toMap(StreamMetadata::streamId, StreamMetadata::endOffset));
        Iterator<RecoverResult> iterator = deltaWAL.recover();

        LogCache.LogCacheBlock cacheBlock;
        List<RuntimeException> exceptions = new ArrayList<>();
        do {
            RecoveryBlockResult result = recoverContinuousRecords(iterator, streamEndOffsets, 1 << 29, logger);
            cacheBlock = result.cacheBlock;
            Optional.ofNullable(result.exception).ifPresent(exceptions::add);
            updateStreamEndOffsets(cacheBlock, streamEndOffsets);
            uploadRecoveredRecords(objectManager, cacheBlock, logger);
        } while (cacheBlock.isFull());

        deltaWAL.reset().get();
        closeStreams(streamManager, streams, streamEndOffsets, logger);

        // fail it if there is any invalid stream.
        if (!exceptions.isEmpty()) {
            throw ExceptionUtil.combine(exceptions);
        }
    }
Recover WAL, upload WAL to S3 and close opening streams.
Note: {@link WriteAheadLog#start()} should be called before this method.
s3stream/src/main/java/com/automq/stream/s3/S3Storage.java