Skip to content

feat(firestore): configure gRPC message length limits for large documents #18220

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ohmayr wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(firestore): configure gRPC message length limits for large documents #18220

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 13 additions & 2 deletions packages/google-cloud-firestore/google/cloud/firestore_v1/base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,14 @@
_INACTIVE_TXN: str = "There is no active transaction."
_CLIENT_INFO: Any = client_info.ClientInfo(client_library_version=__version__)
_FIRESTORE_EMULATOR_HOST: str = "FIRESTORE_EMULATOR_HOST"
_GRPC_MSG_SIZE_OPTIONS: List[Tuple[str, int]] = [
("grpc.max_send_message_length", -1),
("grpc.max_receive_message_length", -1),
]
_DEFAULT_CHANNEL_OPTIONS: List[Tuple[str, Any]] = [
("grpc.keepalive_time_ms", 30000),
*_GRPC_MSG_SIZE_OPTIONS,
]


class BaseClient(ClientWithProject):
Expand Down Expand Up @@ -173,7 +181,7 @@ def _firestore_api_helper(self, transport, client_class, client_module) -> Any:
channel = transport.create_channel(
self._target,
credentials=self._credentials,
options={"grpc.keepalive_time_ms": 30000}.items(),
options=_DEFAULT_CHANNEL_OPTIONS,
)

self._transport = transport(host=self._target, channel=channel)
Expand Down Expand Up @@ -204,7 +212,10 @@ def _emulator_channel(self, transport):
and getattr(self._credentials, "id_token", None) is not None
):
token = self._credentials.id_token
options = [("Authorization", f"Bearer {token}")]
options = [
("Authorization", f"Bearer {token}"),
*_GRPC_MSG_SIZE_OPTIONS,
]

if "GrpcAsyncIOTransport" in str(transport.__name__):
return grpc.aio.insecure_channel(self._emulator_host, options=options)
Expand Down
34 changes: 34 additions & 0 deletions packages/google-cloud-firestore/tests/system/test_system.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3836,3 +3836,37 @@ def in_transaction(transaction, rollback):
assert len(result) == 1
assert len(result[0]) == 1
assert result[0][0].value == expected


@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
def test_large_document_standard_writes(client, cleanup, database):
"""Test standard write and read operations for large document on Enterprise DB."""
collection_id = "large_docs_" + UNIQUE_RESOURCE_ID
doc_ref = client.collection(collection_id).document("large_doc")
cleanup(doc_ref.delete)

large_payload = "a" * (900 * 1024)
doc_ref.set({"payload": large_payload})

snapshot = doc_ref.get()
assert snapshot.exists
assert snapshot.to_dict() == {"payload": large_payload}


@pytest.mark.parametrize("method", ["execute", "stream"])
@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
def test_large_document_pipeline(client, cleanup, database, method):
"""Test pipeline execution over large document on Enterprise DB."""
collection_id = "large_pipeline_" + UNIQUE_RESOURCE_ID
col_ref = client.collection(collection_id)
doc_ref = col_ref.document("large_doc")
cleanup(doc_ref.delete)

large_payload = "b" * (900 * 1024)
doc_ref.set({"payload": large_payload})

pipeline = client.pipeline().collection(collection_id)
method_under_test = getattr(pipeline, method)

results = list(method_under_test())
assert [doc.data() for doc in results] == [{"payload": large_payload}]
36 changes: 36 additions & 0 deletions packages/google-cloud-firestore/tests/system/test_system_async.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3708,3 +3708,39 @@ async def in_transaction(transaction):
await in_transaction(transaction)
# make sure we didn't skip assertions in inner function
assert inner_fn_ran is True


@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
async def test_large_document_standard_writes_async(client, cleanup, database):
"""Test standard write and read operations for large document on Enterprise DB (async)."""
collection_id = "large_docs_async_" + UNIQUE_RESOURCE_ID
doc_ref = client.collection(collection_id).document("large_doc")
cleanup(doc_ref.delete)

large_payload = "c" * (900 * 1024)
await doc_ref.set({"payload": large_payload})

snapshot = await doc_ref.get()
assert snapshot.exists
assert snapshot.to_dict() == {"payload": large_payload}
Comment on lines +3714 to +3725

@gemini-code-assist gemini-code-assist Bot Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

In async tests, doc_ref.delete is an asynchronous coroutine function. Passing it directly to a synchronous cleanup fixture will result in the coroutine being called but never awaited, which triggers a RuntimeWarning: coroutine 'AsyncDocumentReference.delete' was never awaited and fails to clean up the document in the database. Instead, use a try...finally block to explicitly await the deletion of the document.

Suggested change
async def test_large_document_standard_writes_async(client, cleanup, database):
"""Test standard write and read operations for large document on Enterprise DB (async)."""
collection_id = "large_docs_async_" + UNIQUE_RESOURCE_ID
doc_ref = client.collection(collection_id).document("large_doc")
cleanup(doc_ref.delete)
large_payload = "c" * (900 * 1024)
await doc_ref.set({"payload": large_payload})
snapshot = await doc_ref.get()
assert snapshot.exists
assert snapshot.to_dict() == {"payload": large_payload}
async def test_large_document_standard_writes_async(client, database):
"""Test standard write and read operations for large document on Enterprise DB (async)."""
collection_id = "large_docs_async_" + UNIQUE_RESOURCE_ID
doc_ref = client.collection(collection_id).document("large_doc")
try:
large_payload = "c" * (900 * 1024)
await doc_ref.set({"payload": large_payload})
snapshot = await doc_ref.get()
assert snapshot.exists
assert snapshot.to_dict() == {"payload": large_payload}
finally:
await doc_ref.delete()



@pytest.mark.parametrize("method", ["execute", "stream"])
@pytest.mark.parametrize("database", [FIRESTORE_ENTERPRISE_DB], indirect=True)
async def test_large_document_pipeline_async(client, cleanup, database, method):
"""Test async pipeline execution over large document on Enterprise DB."""
collection_id = "large_pipeline_async_" + UNIQUE_RESOURCE_ID
col_ref = client.collection(collection_id)
doc_ref = col_ref.document("large_doc")
cleanup(doc_ref.delete)

large_payload = "d" * (900 * 1024)
await doc_ref.set({"payload": large_payload})

pipeline = client.pipeline().collection(collection_id)
if method == "execute":
results = await pipeline.execute()
else:
results = [doc async for doc in pipeline.stream()]

assert [doc.data() for doc in results] == [{"payload": large_payload}]
Comment on lines +3730 to +3746

@gemini-code-assist gemini-code-assist Bot Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

Similar to the standard writes test, doc_ref.delete is an asynchronous coroutine function. Passing it to the synchronous cleanup fixture will not await it, leading to un-deleted documents and RuntimeWarnings. Please use a try...finally block to ensure the document is properly cleaned up.

Suggested change
async def test_large_document_pipeline_async(client, cleanup, database, method):
"""Test async pipeline execution over large document on Enterprise DB."""
collection_id = "large_pipeline_async_" + UNIQUE_RESOURCE_ID
col_ref = client.collection(collection_id)
doc_ref = col_ref.document("large_doc")
cleanup(doc_ref.delete)
large_payload = "d" * (900 * 1024)
await doc_ref.set({"payload": large_payload})
pipeline = client.pipeline().collection(collection_id)
if method == "execute":
results = await pipeline.execute()
else:
results = [doc async for doc in pipeline.stream()]
assert [doc.data() for doc in results] == [{"payload": large_payload}]
async def test_large_document_pipeline_async(client, database, method):
"""Test async pipeline execution over large document on Enterprise DB."""
collection_id = "large_pipeline_async_" + UNIQUE_RESOURCE_ID
col_ref = client.collection(collection_id)
doc_ref = col_ref.document("large_doc")
try:
large_payload = "d" * (900 * 1024)
await doc_ref.set({"payload": large_payload})
pipeline = client.pipeline().collection(collection_id)
if method == "execute":
results = await pipeline.execute()
else:
results = [doc async for doc in pipeline.stream()]
assert [doc.data() for doc in results] == [{"payload": large_payload}]
finally:
await doc_ref.delete()

15 changes: 12 additions & 3 deletions packages/google-cloud-firestore/tests/unit/v1/test_base_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,9 +110,13 @@ def test_baseclient__firestore_api_helper_wo_emulator():

assert api is client_class.return_value
assert client._firestore_api_internal is api
channel_options = {"grpc.keepalive_time_ms": 30000}
channel_options = [
("grpc.keepalive_time_ms", 30000),
("grpc.max_send_message_length", -1),
("grpc.max_receive_message_length", -1),
]
transport_class.create_channel.assert_called_once_with(
target, credentials=client._credentials, options=channel_options.items()
target, credentials=client._credentials, options=channel_options
)
transport_class.assert_called_once_with(
host=target,
Expand Down Expand Up @@ -236,7 +240,12 @@ def test_baseclient__emulator_channel():
with mock.patch("grpc.insecure_channel") as insecure_channel:

@parthea parthea Aug 26, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we also need to testFirestoreGrpcAsyncIOTransport

I would expect something like with mock.patch("grpc.aio.insecure_channel") as aio_insecure_channel:

channel = client._emulator_channel(FirestoreGrpcTransport)
insecure_channel.assert_called_once_with(
emulator_host, options=[("Authorization", "Bearer test")]
emulator_host,
options=[
("Authorization", "Bearer test"),
("grpc.max_send_message_length", -1),
("grpc.max_receive_message_length", -1),
],
)


Expand Down
Loading