Repository: LangStream/langstream (language: python)

Issue:
RESOURCE_EXHAUSTED in python-source agent
Processing large data with `python-source` results in `RESOURCE_EXHAUSTED` error for gRPC:
```
Critical failure: gRPC server sent error: RESOURCE_EXHAUSTED: gRPC message exceeds maximum size 4194304: 11646340. Shutting down the runtime...
```

Perhaps some kind of streaming between python runtime and agent service is in order.


Discussion / hints:
How much large is your payload?
Also, what's the next agent in your pipeline? Another python agent or a builtin agent?
Looks like it's 11646340 (~11MB).
Next one is a build-in agent:
```yaml
  - name: "Extract text"
    id: "extract-text"
    type: "text-extractor"
```
Adding stackstrace:
```
java.lang.RuntimeException: gRPC server sent error: RESOURCE_EXHAUSTED: gRPC message exceeds maximum size 4194304: 11646340
	at ai.langstream.agents.grpc.GrpcAgentSource$1.onError(GrpcAgentSource.java:136)
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:487)
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:576)
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:757)
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:736)
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: io.grpc.StatusRuntimeException: RESOURCE_EXHAUSTED: gRPC message exceeds maximum size 4194304: 11646340
	at io.grpc.Status.asRuntimeException(Status.java:539)
	... 10 common frames omitted
```
Making it run as a standalone agent by providing `input`/`output` unfortunately, doesn't solve the issue.
@cbornet is it possible to change the max size of the GRPC payload ?
I'm quite sure it's possible to change max inbound size using `ManagedChannelBuilder` in `PythonGrpcServer.java:90`. Possible also in `insecure_channel` in python server. This however won't solve the root problem. Agent may generate payload much bigger than 11 MB.

This is a design choice. Should payload size be bound to some arbitrary max size, or is there some way of splitting the data on transport?
We can increase the GRPC max message length (up to 2GB it seems).
GRPC itself [should be fine](https://stackoverflow.com/a/58435745).
We're using a local socket so we don't go through HTTP proxies that could feel bad about it.
But with very large documents the contraints will mostly be on the pod memory.
Also note that it's important to have a composable agent afterwards to chunk the data before sending it to the streaming engine as neither Kafka nor Pulsar will be happy with a big payload.

Fix patch:
diff --git a/langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/AbstractGrpcAgent.java b/langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/AbstractGrpcAgent.java
index acdf2fe26..c21d5dcff 100644
--- a/langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/AbstractGrpcAgent.java
+++ b/langstream-agents/langstream-agent-grpc/src/main/java/ai/langstream/agents/grpc/AbstractGrpcAgent.java
@@ -100,7 +100,11 @@ public void start() throws Exception {
         }
         blockingStub =
                 AgentServiceGrpc.newBlockingStub(channel).withDeadlineAfter(30, TimeUnit.SECONDS);
-        asyncStub = AgentServiceGrpc.newStub(channel).withWaitForReady();
+        asyncStub =
+                AgentServiceGrpc.newStub(channel)
+                        .withWaitForReady()
+                        .withMaxInboundMessageSize(Integer.MAX_VALUE)
+                        .withMaxOutboundMessageSize(Integer.MAX_VALUE);
 
         topicProducerWriteResults = new CompletableFuture<>();
         topicProducerWriteResults.complete(
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/grpc_service.py b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/grpc_service.py
index d1e9a28fa..057699a50 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/grpc_service.py
+++ b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/grpc_service.py
@@ -415,7 +415,12 @@ def get_topic_producer(self) -> TopicProducer:
 class AgentServer(object):
     def __init__(self, target: str):
         self.target = target
-        self.grpc_server = grpc.aio.server()
+        self.grpc_server = grpc.aio.server(
+            options=[
+                ("grpc.max_send_message_length", 0x7FFFFFFF),
+                ("grpc.max_receive_message_length", 0x7FFFFFFF),
+            ]
+        )
         self.port = self.grpc_server.add_insecure_port(target)
         self.agent = None
 


Test patch:
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/server_and_stub.py b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/server_and_stub.py
index ea294cbbd..4ff959665 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/server_and_stub.py
+++ b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/server_and_stub.py
@@ -36,7 +36,13 @@ async def __aenter__(self):
         self.server = AgentServer("[::]:0")
         await self.server.init(json.dumps(self.config), json.dumps(self.context))
         await self.server.start()
-        self.channel = grpc.aio.insecure_channel("localhost:%d" % self.server.port)
+        self.channel = grpc.aio.insecure_channel(
+            "localhost:%d" % self.server.port,
+            options=[
+                ("grpc.max_send_message_length", 0x7FFFFFFF),
+                ("grpc.max_receive_message_length", 0x7FFFFFFF),
+            ],
+        )
         self.stub = AgentServiceStub(channel=self.channel)
         return self
 
diff --git a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/test_grpc_processor.py b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/test_grpc_processor.py
index 3692bc432..82eefae20 100644
--- a/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/test_grpc_processor.py
+++ b/langstream-runtime/langstream-runtime-impl/src/main/python/langstream_grpc/tests/test_grpc_processor.py
@@ -201,6 +201,25 @@ async def test_future_record(klass):
             assert response.results[0].records[0].value.string_value == "test"
 
 
+async def test_big_record():
+    async with ServerAndStub(
+        "langstream_grpc.tests.test_grpc_processor.MyProcessor"
+    ) as server_and_stub:
+        long_string = "a" * 10_000_000
+        response: ProcessorResponse
+        async for response in server_and_stub.stub.process(
+            [
+                ProcessorRequest(
+                    records=[GrpcRecord(value=Value(string_value=long_string))]
+                )
+            ]
+        ):
+            assert len(response.results) == 1
+            assert response.results[0].HasField("error") is False
+            assert len(response.results[0].records) == 1
+            assert response.results[0].records[0].value.string_value == long_string
+
+
 async def test_info():
     async with ServerAndStub(
         "langstream_grpc.tests.test_grpc_processor.MyProcessor"