如何修复Firestore流查询的grpc超时错误（StatusCode.UNAVAILABLE）

当使用Firestore的stream()方法读取超大集合时，容易触发StatusCode.UNAVAILABLE超时错误，即使需要遍历所有文档，也可以通过以下几种方案解决：

1. 采用游标分页读取 #

单次拉取全量文档会给服务器和客户端带来压力，通过分页（每次固定批量）读取可以有效避免超时。使用limit()和start_after()实现游标分页，循环处理直到所有文档遍历完成：

from firebase_admin import firestore

def find_all_issues(collection: str) -> None:
    fs_client = firestore.client()
    coll_ref = fs_client.collection(collection)
    batch_size = 1000  # 根据集合大小调整批量数
    last_doc = None
    
    while True:
        # 构建分页查询
        query = coll_ref.limit(batch_size)
        if last_doc:
            query = query.start_after(last_doc)
        
        # 获取当前批次的文档
        docs_batch = list(query.stream())
        if not docs_batch:
            break  # 没有更多文档，终止循环
        
        # 处理当前批次的文档
        for doc_snap in docs_batch:
            d = doc_snap.to_dict()
            # 执行你的错误检查逻辑
        
        # 更新游标为当前批次最后一个文档
        last_doc = docs_batch[-1]

2. 调整客户端GRPC超时配置 #

默认的GRPC连接超时可能不足以支撑大集合的流读取，通过自定义客户端配置增大超时时间、优化连接参数：

from firebase_admin import firestore
import grpc

# 自定义GRPC选项，优化连接稳定性和超时
grpc_options = [
    ('grpc.max_receive_message_length', 100 * 1024 * 1024),  # 允许更大的消息体积（100MB）
    ('grpc.keepalive_time_ms', 30000),  # 每30秒发送一次保活包
    ('grpc.keepalive_timeout_ms', 5000),  # 保活超时5秒
    ('grpc.http2.max_pings_without_data', 0),
    ('grpc.http2.min_time_between_pings_ms', 10000),
    ('grpc.http2.min_ping_interval_without_data_ms', 5000)
]

# 使用自定义选项初始化Firestore客户端
fs_client = firestore.Client(options=grpc_options)

def find_all_issues(collection: str) -> None:
    coll_ref = fs_client.collection(collection)
    doc_snapshot_stream = coll_ref.stream()
    for doc_snap in doc_snapshot_stream:
        d = doc_snap.to_dict()
        # 执行错误检查

3. 绕过查询规划器Bug #

如果怀疑是Firestore查询规划器的bug导致超时，可以添加一个无意义的过滤条件，强制查询规划器使用不同的执行路径。例如针对文档ID添加一个恒成立的过滤：

from firebase_admin import firestore

def find_all_issues(collection: str) -> None:
    fs_client = firestore.client()
    # 添加无意义的过滤条件，绕过可能的查询规划器bug
    coll_ref = fs_client.collection(collection).where('__name__', '>=', '')
    doc_snapshot_stream = coll_ref.stream()
    for doc_snap in doc_snapshot_stream:
        d = doc_snap.to_dict()
        # 执行错误检查

4. 分离文档读取与处理逻辑 #

如果你的错误检查逻辑耗时较长，会导致流连接长时间处于空闲状态触发超时。可以先批量读取文档到本地，再异步或批量处理：

from firebase_admin import firestore
import concurrent.futures

def process_doc(doc_snap):
    d = doc_snap.to_dict()
    # 执行错误检查逻辑

def find_all_issues(collection: str) -> None:
    fs_client = firestore.client()
    coll_ref = fs_client.collection(collection)
    batch_size = 500
    last_doc = None
    
    while True:
        query = coll_ref.limit(batch_size)
        if last_doc:
            query = query.start_after(last_doc)
        
        docs_batch = list(query.stream())
        if not docs_batch:
            break
        
        # 使用线程池并行处理文档，避免阻塞流连接
        with concurrent.futures.ThreadPoolExecutor() as executor:
            executor.map(process_doc, docs_batch)
        
        last_doc = docs_batch[-1]

内容的提问来源于stack exchange，提问作者HaroldFinch