Client streaming

Client streaming — вторая стриминговая модель курса и точное зеркало server streaming: клиент отправляет последовательность сообщений — ноль, одно или сколько накопил, — а сервер в конце отвечает ровно одним сообщением. В .proto это слово stream слева от returns: rpc UploadOrderEvents (stream OrderEvent) returns (UploadSummary) читается как «поток OrderEvent на входе, один UploadSummary на выходе». Русскоязычный запрос «стрим клиент go» описывает ровно эту конструкцию: поток ведёт клиент, сервер слушает и отвечает финалом. Заметьте асимметрию кардинальности: со стороны запросов ограничений нет — пустой поток легален, тысяча сообщений легальна; со стороны ответа правило unary из шестого урока действует в полную силу — ровно одно сообщение, и кардинальность здесь проверяется жёстко. Модель прирождена для всех «выгрузок»: батчи метрик, пачки логов, загрузка файла чанками, массовое создание записей — в других кодовых базах тот же метод зовут BatchCreate или UploadMetrics, суть одна: много мелких сообщений, одна сводка. Схема обмена у обоих стриминговых уроков одна и та же — меняются только направления: там сервер говорил, здесь говорит клиент. Контекст вызова и статусные ошибки при этом не изменились вовсе: grpc-timeout едет в заголовках, итоговый статус — в trailers, и это не совпадение, а одна и та же wire-механика из шестого урока, повернутая другой стороной.

Серверный интерфейс grpc.ClientStreamingServer[OrderEvent, UploadSummary] из stream_interfaces.go — с grpc-go 1.64 он дженериковый, под старым именем OrderService_UploadOrderEventsServer теперь прячется псевдоним — несёт два собственных метода плюс встроенный ServerStream с привычными Context, SetHeader, SendHeader и SetTrailer. Recv() (OrderEvent, error) блокируется до очередного сообщения; документация описывает исходы так: «If io.EOF is returned, it indicates the client called CloseAndRecv on its ClientStreamingClient. Any other error indicates the stream was terminated unexpectedly, and the handler method should return, as the stream is no longer usable». Вот он, главный переворот седьмого урока: io.EOF на сервере здесь не антипаттерн, а штатный сигнал — клиент закрыл свою сторону, время отвечать. SendAndClose(UploadSummary) error отправляет единственное сообщение-ответ; доке дословно: «This method must be called once and only once after all request messages have been processed. Recv should not be called after calling SendAndClose». И общий контракт типа: «To terminate the RPC, call SendAndClose and return nil from the method handler or do not call SendAndClose and return an error from the status package». Переводя: успех — SendAndClose плюс возврат nil, отказ — возврат статусной ошибки, хоть до первого Recv, хоть посреди чтения.

Учебная альтернатива — складывать все события в []*pb.OrderEvent, а после io.EOF обработать разом — выглядит невинно и ломается ровно там, ради чего client streaming придуман. Во-первых, память: хендлер держит весь батч целиком, а клиенты бывают быстрые — гигабайт чанков выест лимит задолго до конца выгрузки; агрегация на лету держит лишь сводку и карту виденных заказов, то есть O(уникальным заказам), а не O(событиям). Во-вторых, честность: если вызов умрёт на середине, слайс-версия потеряет всё — потоковая версия хотя бы уже записала в базу то, что успела, если вы пишете события по мере прихода. В-третьих, потолок из шестого урока касается каждого сообщения стрима отдельно, а не вызова: миллион сообщений по килобайту проходит спокойно там, где одно сообщение на миллион килобайт упрётся в maxReceiveMessageSize и умрёт с ResourceExhausted. Канон агрегации задаёт RouteSummary из route_guide: точка за точкой, счётчики растут, итог собирается из накопленного. Единственный честный повод копить слайс — обработке нужен весь батч целиком: сортировка, группировка, дедупликация с перекрёстными ссылками; тогда держите в голове, что размер батча — теперь ваш риск, а не ограничение транспорта.

Клиентский интерфейс grpc.ClientStreamingClient[OrderEvent, UploadSummary] — два метода плюс унаследованные Header, Trailer и Context. Send(OrderEvent) error шлёт очередное сообщение; клиент может звать её сколько нужно, но из одной горутины — конкурентная отправка это территория девятого урока. CloseAndRecv() (UploadSummary, error) закрывает отправляющую сторону и дожидается ответа; под капотом GenericClientStream.CloseAndRecv — ровно два шага: CloseSend(), затем RecvMsg — никакой магии, просто удобная склейка. Документация требует: «This method must be called once and only once after sending all request messages» — один и только один раз, после всех сообщений. И заметьте финальную фразу: «Any error returned is implemented by the status package» — из CloseAndRecv io.EOF не выходит никогда, в отличие от Recv серверного стрима из седьмого урока: здесь статус вызова всегда упакован в статусную ошибку, и status.Code работает без оговорок. Типовой цикл выгрузки читается за секунды: открыть стрим вызовом с ctx, прогнать батч через Send, закрыть CloseAndRecv — и получить сводку либо статусную ошибку. Официальный пример runRecordRoute из route_guide живёт ровно по этому сценарию: context.WithTimeout на десять секунд, цикл Send по точкам, CloseAndRecv в конце.

В unary дедлайн — это время одного обмена. В client streaming под одним ctx лежит всё: каждая Send, каждый Recv на сервере, обработка и финальное ожидание сводки в CloseAndRecv. Заголовок grpc-timeout из шестого урока едет при открытии вызова, и сервер видит его в stream.Context() — того же самого, что был в седьмом: с дедлайном, отменой и метаданными. Истёк срок посреди выгрузки — и обе стороны узнают об этом статусной ошибкой DeadlineExceeded: у клиента её вернёт Send или CloseAndRecv, у сервера — очередной Recv; ветка stream.Context().Done() в хендлере, привычная по подписке WatchOrders, сработает как прежде. Отсюда простое правило проектирования: длительность дедлайна выбирается по размеру батча и скорости сети, а не по времени одного сообщения — выгрузка тысячи событий не должна делить секундный таймаут с unary-запросом. Для выгрузок, которые могут тянуться непредсказуемо, берут context.WithCancel и отменяют по пользовательскому решению, как для вечной подписки из седьмого урока. И предохранитель оттуда же остаётся в силе: вызов обязан уметь кончаться, иначе GracefulStop из серверного урока так и не вернётся — висящий клиентский стрим держит хендлер.

Пока клиент шлёт сообщения, серверный Recv возвращает их без сюрпризов; интерес начинается, когда стрим умирает не по протоколу. Клиент отменил контекст или истёк его дедлайн — очередной Recv вернёт статусную ошибку Canceled или DeadlineExceeded; сеть оборвалась — Unavailable. Отдельный случай — io.ErrUnexpectedEOF: клиент оборвался посреди передачи сообщения, и библиотека честно превращает его в codes.Internal. Заметьте, чего среди этих ошибок нет: io.EOF зарезервирован за half-close — только явное CloseAndRecv клиента даёт его серверу, смешивать «клиент закончил» и «клиент умер» библиотека не станет. Есть и автоматика на стороне сервера: в serverStream.RecvMsg отложенный код при любой не-EOF ошибке сам пишет статус в транспорт — WriteStatus вызывается до того, как хендлер успеет вернуться, так что клиент в большинстве случаев узнает причину даже при молчаливом хендлере. Практический вывод для хендлера совпадает с седьмым уроком: ошибку Recv не глотаем и не пытаемся читать дальше — после сбоя стрим непригоден, документация прямо говорит «the handler method should return». Единственное различие с подпиской: здесь конец потока означается io.EOF из Recv, а не отменой хендлера.

Со стороны запросов клиентский стрим свободен, но сторона ответа — под железным правилом unary: ServerStreams: false в дескрипторе метода означает ровно один ответ, и проверка из шестого урока теперь сторожит именно вас. Хендлер прочитал весь поток и вернул nil, забыв про SendAndClose? На клиенте RecvMsg получает конец данных при статусе OK, а затем та самая проверка срабатывает: код Internal с сообщением «cardinality violation: received no response message from non-server-streaming RPC» — клиент честно сообщает, что не-серверный стрим не прислал ни одного ответа. Второй случай зеркальный: посчитать SendAndClose дважды — второй вызов упрётся в уже отправленный half-close и вернёт хендлеру Internal; дважды одного ответа не бывает. Контраст со стороной запросов подчёркивает замысел: ClientStreams: true не включает никаких проверок — клиент, открывший стрим и немедленно закрывший его без единого сообщения, получит легальную пустую сводку, как получал пустой поток в седьмом уроке. То есть пустота слева — бизнес-решение, пустота справа — ошибка программиста. Запомните это разделение: оно же станет подсказкой при диагностике — Internal про кардинальность всегда указывает на хендлер, а не на сеть.

Критерий тот же, что в шестом уроке: модель должна выражать то, что не выражается запросом-ответом. Client streaming берут, когда у клиента много данных и они появляются по мере работы, а серверу нужен один осмысленный итог: батчи метрик и логов, выгрузка файла чанками, массовая вставка записей, потоковые апдейты состояния. Не берут его там, где хватит unary: если событий три и они уже в памяти, заверните их в repeated-поле и не плодите стрим. Отдельно проговорите то, что часто недооценивают: между открытием вызова и сводкой промежуточных ответов не существует вообще — сервер молчит до самого конца. Если задача «посылать и слышать ответ по ходу» — это не client streaming, а территория девятого урока, где обе стороны говорят одновременно. По той же причине клиент не видит прогресса: пятьдесят процентов выгрузки неотличимы от нуля — если прогресс нужен, заводите отдельный unary-вызов или WatchOrders из седьмого урока рядом. И последнее: client streaming не транзакция. Отмена на середине — не rollback, а просто конец: сервер сохранил то, что успел принять, и сводки не будет; идемпотентность и повторная выгрузка — ваша забота на уровне домена.

Клиентский Send — тот же SendMsg из седьмого урока, но с другой стороны стекла, и его документация читается теми же тремя предложениями. Send блокируется, пока не появится достаточный flow control — место в HTTP/2-окне; быстрый клиент против медленного сервера упрётся в окно и притормозит сам, без всяких искусственных задержек. Send не ждёт приёма: «ушло в транспорт» не значит «сервер прочитал». Сообщение больше maxSendMessageSize вернёт codes.ResourceExhausted с текстом «trying to send message larger than max» — причём это клиентская ошибка, и она приезжает из Send напрямую статусом, без посредничества io.EOF: её «сгенерировал клиент». Из этих трёх предложений следует архитектурная точка урока: единственный источник истины о принятом — сводка сервера. Клиент посчитал тысячу Send, вернувших nil, — но nil лишь про транспорт; сколько событий реально попало в агрегацию, знает только UploadSummary. Поэтому сводка и существует: не для удобства, а потому что подтверждения доставки в стриме нет. Ну и про дисциплину: Send из двух горутин на одном стриме запрещён так же, как на сервере в седьмом уроке, — очередь отправки одна, и право голоса в ней принадлежит одной горутине.

Одно честное замечание про формат: у самоучителя есть песочница с интерпретатором Go, но сеть в браузере недоступна — открыть gRPC-вызов оттуда невозможно, поэтому runner-ов в этом уроке и во всём gRPC-курсе нет. Всё разобранное проверяется локально за минуты: сервер из урока 04, клиентский каркас из пятого урока — и UploadOrderEvents оживёт: срежьте дедлайн до миллисекунды и посмотрите, что вернёт CloseAndRecv, а затем уберите SendAndClose и встретьте cardinality violation лично. Механику этого урока стоит сверить с серверным стримом седьмого урока: те же сообщения, встречные направления, тот же io.EOF — но на другой стороне. Дальше по маршруту: девятый урок — bidi, где обе стороны говорят одновременно и одна труба перестаёт быть монологом; он уже мелькал в седьмом уроке и ждёт своего часа. За ним — metadata и токены в уроке про метаданные, глубокие дедлайны — в уроке про дедлайны, а базовая wire-механика unary — grpc-timeout, trailers, статусные ошибки — остаётся фундаментом всех четырёх моделей. Полный маршрут курса — на странице gRPC-Go.