为什么重新启动http请求后zlib解压会中断？

Question

我有一个 python 脚本，它使用urllib3“流式传输”一个非常大的 gzip 文件并将其输入到zlib.decompressobj。这个 zlib 解压对象被配置为读取 gzip 压缩。如果这个初始 http 连接被中断，那么zlib.decompressobj 在连接“恢复”后开始抛出错误。如果您想切入正题，请参阅下面的源代码。

尽管脚本使用Range 标头指定先前读取的字节数来启动新的http 连接，但仍会发生这些错误。它从连接断开时出现的已完成读取点恢复。我相信这个任意的恢复点是我问题的根源。

如果我不尝试解压缩 urllib3 正在读取的数据块，而是将它们写入文件，那么一切正常。即使有中断，也无需尝试解压缩流，一切正常。完成的存档是有效的，它与浏览器下载的文件大小相同，.gz 文件的 MD5 哈希值与我直接使用 Chrome 下载的文件相同。

另一方面，如果我尝试解压缩中断后进入的数据块，即使指定了Range 标头，zlib 库也会引发各种错误。最近的是Error -3 while decompressing data: invalid block type

补充说明：

• 我正在使用的站点将Accept-Range 标志设置为bytes，这意味着我能够将修改后的Range 标头提交到服务器。
• 我不在此脚本中使用requests 库，因为它最终管理urllib3。我转而直接使用urllib3 试图切断中间人。
• 此脚本过于简化了我的最终目标，即直接从托管位置流式传输压缩数据，丰富它并将其存储在本地网络上的MySQL 数据库中。
  • 我在 docker 容器内的资源严重受限，该处理将在其中进行。
  • 这个问题的起源出现在我大约 3 周前提出的一个问题中：requests.iter_content() thinks file is complete but it's not
• urllib3（和requests）库遇到的最常见问题是IncompleteRead(self._fp_bytes_read, self.length_remaining) 错误。
  • 仅当urllib3 库已被修补以在发生不完整读取时引发异常时才会出现此错误。

我的最佳猜测：

我猜想输入到zlib.decompressobj 的数据流中断导致zlib 以某种方式丢失上下文并开始尝试在一个奇怪的位置再次解压缩数据。有时它会恢复，但是数据流是乱码，这让我相信用作新的Range 标头的字节位置落在了一些字节的前面，然后这些字节被错误地解释为标头。我不知道如何解决这个问题，我已经尝试解决了几个星期。即使发生中断，数据在整个下载（完成前未解压缩）时仍然有效这一事实使我相信zlib 中的一些“上下文丢失”是原因。

源代码：（已更新为包含“缓冲区”）

这段代码有点乱，请见谅。此外，这个目标 gzip 文件比我将使用的实际文件小很多。此外，大约一个月后，Rapid7 将不再提供此示例中的目标文件。如果适合您，您可以选择替换其他 .gz 文件。

import urllib3
import certifi
import inspect
import os
import time
import zlib

def patch_urllib3():
    """Set urllib3's enforce_content_length to True by default."""
    previous_init = urllib3.HTTPResponse.__init__
    def new_init(self, *args, **kwargs):
        previous_init(self, *args, enforce_content_length = True, **kwargs)
    urllib3.HTTPResponse.__init__ = new_init

#Path the urllib3 module to throw an exception for IncompleteRead
patch_urllib3()

#Set the target URL
url = "https://opendata.rapid7.com/sonar.http/2021-11-27-1638020044-http_get_8899.json.gz"

#Set the local filename
local_filename = '2021-11-27-1638020044-http_get_8899_script.json.gz'

#Configure the PoolManager to handle https (I think...)
http = urllib3.PoolManager(ca_certs=certifi.where())

#Initiate start bytes at 0 then update as download occurs
sum_bytes_read=0
session_bytes_read=0
total_bytes_read=0
#Dummy variable to silence console output from file write
writer=0

#Set zlib window bits to 16 bits for gzip decompression
decompressor = zlib.decompressobj(zlib.MAX_WBITS|16)

#Build a buffer list
buf_list=[]
i=0
while True:
    print("Building request. Bytes read:",total_bytes_read)
    resp = http.request(
        'GET',
        url,
        timeout=urllib3.Timeout(connect=15, read=40),
        preload_content=False)
    print("Setting headers.")
    #This header should cause the request to resume at "total_bytes_read"
    resp.headers['Range'] = 'bytes=%s' % (total_bytes_read)
    print("Local filename:",local_filename)
    #If file already exists then append to it
    if os.path.exists(local_filename):
        print("File already exists.")
        try:
            print("Starting appended download.")
            with open(local_filename, 'ab') as f:
                for chunk in resp.stream(2048):
                    buf_list.append(chunk)
                    #Use i to offset the chunk being read from the "buffer"
                    #I.E. load 3 chunks (0,1,2) in the buffer list before starting to read from it
                    if i >2: 
                        buffered_chunk=buf_list.pop(0)
                        writer=f.write(buffered_chunk)
                        #Comment out the below line to stop the error from occurring.
                        #File download should complete successfully even if interrupted when the following line is commented out.
                        decompressed_chunk=decompressor.decompress(buffered_chunk)
                    #Increment i so that the buffer list will fill before reading from it
                    i=i+1
                    session_bytes_read = resp._fp_bytes_read
                    #Sum bytes read is an updated value that isn't stored. It is only used for console print 
                    sum_bytes_read = total_bytes_read + session_bytes_read
                    print("[+] Bytes read:",str(format(sum_bytes_read, ",")), end='\r')
            print("\nAppended download complete.")
            break
        except Exception as e:
            print(e)
            #Add to total bytes read to current session bytes each time the loop needs to repeat
            total_bytes_read=total_bytes_read+session_bytes_read
            print("Bytes Read:",total_bytes_read)
            #Mod the total_bytes back to the nearest chunk size so it can be - re-requested
            total_bytes_read=total_bytes_read-(total_bytes_read%2048)-2048
            print("Rounded bytes Read:",total_bytes_read)
            #Pop the last entry off of the buffer since it may be incomplete
            buf_list.pop()
            #reset i so that the buffer has to rebuilt
            i=0
            print("Sleeping for 30 seconds before re-attempt...")
            time.sleep(30)
    #If file doesn't already exist then write to it directly
    else:
        print("File does not exist.")
        try:
            print("Starting initial download.")
            with open(local_filename, 'wb') as f:
                for chunk in resp.stream(2048):
                    buf_list.append(chunk)
                    #Use i to offset the chunk being read from the "buffer"
                    #I.E. load 3 chunks (0,1,2) in the buffer list before starting to read from it
                    if i > 2: 
                        buffered_chunk=buf_list.pop(0)
                        #print("Buffered Chunk",str(i-2),"-",buffered_chunk)
                        writer=f.write(buffered_chunk)
                        decompressed_chunk=decompressor.decompress(buffered_chunk)
                    #Increment i so that the buffer list will fill before reading from it
                    i=i+1
                    session_bytes_read = resp._fp_bytes_read
                    print("[+] Bytes read:",str(format(session_bytes_read, ",")), end='\r')
            print("\nInitial download complete.")
            break
        except Exception as e:
            print(e)
            #Set the total bytes read equal to the session bytes since this is the first failure
            total_bytes_read=session_bytes_read
            print("Bytes Read:",total_bytes_read)
            #Mod the total_bytes back to the nearest chunk size so it can be - re-requested
            total_bytes_read=total_bytes_read-(total_bytes_read%2048)-2048
            print("Rounded bytes Read:",total_bytes_read)
            #Pop the last entry off of the buffer since it may be incomplete
            buf_list.pop()
            #reset i so that the buffer has to rebuilt
            i=0
            print("Sleeping for 30 seconds before re-attempt...")
            time.sleep(30)
    print("Looping...")
#Finish writing from buffer into file
#BE SURE TO SET TO "APPEND" with "ab" or you will overwrite the start of the file
f = open(local_filename, 'ab')
print("[+] Finishing write from buffer.")
while not len(buf_list) == 0:
    buffered_chunk=buf_list.pop(0)
    writer=f.write(buffered_chunk)
    decompressed_chunk=decompressor.decompress(buffered_chunk)
#Flush and close the file
f.flush()
f.close()
resp.release_conn()

重现错误

• 要重现错误，请执行以下操作：
  • 运行脚本并开始下载
  • 确保第 65 行 decompressed_chunk=decompressor.decompress(chunk)没有被注释掉
  • 在引发异常之前关闭网络连接
  • 立即重新打开网络连接。

如果从脚本中删除decompressor.decompress(chunk) 行，那么它将下载文件，并且可以成功地从文件本身解压缩数据。但是，如果出现第 65 行并发生中断，则 zlib 库将无法继续解压缩数据流。我需要解压缩数据流，因为我无法存储我尝试使用的实际文件。

有什么方法可以防止这种情况发生吗？我现在尝试添加一个存储块的“缓冲区”列表；脚本在失败后丢弃最后一个块并移回文件中“失败”块之前的点。我能够重新建立连接，甚至可以正确拉回所有数据，但即使使用“缓冲区”，我解压缩流的能力也会中断。我一定不能以某种方式顺利地将数据恢复回缓冲区。

可视化：

为了更好地描述我正在尝试做的事情，我很快将这些放在一起......

我敢打赌，马克·阿德勒正躲在某个地方……

Answer 1

r+b 不附加。为此，您需要使用ab。似乎在重试时，您从头开始再次读取整个 gzip 文件。使用r+b，通过覆盖之前读取的内容，该文件将正确写入您的输出文件。

但是，您将初始读取提供给解压缩器，然后再次开始文件。毫不奇怪，解压缩器很快就会检测到无效的压缩数据。