Server Streaming RPCs in Go With Backpressure and Cancel

Some questions have answers that arrive over time: a feed of events, the rows of a large query, the progress of a job, log lines. A unary call has to build the whole answer first. A server streaming rpc returns a sequence: the client sends one request and reads messages until the server is done. In the .proto file it is one keyword, returns (stream Event), and in Go it is a Send method on the server side and a Recv loop on the client side.

Two things make streaming harder than unary calls: what happens when the client is slower than the server, and what happens when either side stops early. This article demonstrates both with a service that can send as fast as the network allows and a client that can be told to stall. It builds on unary RPCs, and the client and bidirectional streaming article covers the other directions. It ran on Go 1.25.6.

The service

syntax = "proto3";
package feed.v1;
option go_package = "example.test/lab/feedpb";

service Feed {
  // Subscribe streams events until the client cancels or the server runs out.
  rpc Subscribe(SubscribeRequest) returns (stream Event);
}

message SubscribeRequest { int32 count = 1; int32 payload_bytes = 2; }
message Event { int64 seq = 1; bytes payload = 2; }

Subscribe takes a count and a payload size and streams that many events. The server implementation is a loop:

//go:build tools

package main

import (
	_ "google.golang.org/grpc/cmd/protoc-gen-go-grpc"
	_ "google.golang.org/protobuf/cmd/protoc-gen-go"
)

package main

import (
	"context"
	"fmt"
	"io"
	"log"
	"net"
	"sync/atomic"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"

	pb "example.test/lab/feedpb"
)

type feed struct {
	pb.UnimplementedFeedServer
	sent     atomic.Int64 // messages whose Send has returned
	finished chan string  // how each Subscribe call ended, as the server saw it
}

func (f *feed) Subscribe(req *pb.SubscribeRequest, stream grpc.ServerStreamingServer[pb.Event]) error {
	payload := make([]byte, req.GetPayloadBytes())
	for i := int64(1); i <= int64(req.GetCount()); i++ {
		// Send blocks when the client is not reading and the HTTP/2 flow control window is full.
		// That blocking is the backpressure: the server cannot run arbitrarily far ahead of a slow client.
		if err := stream.Send(&pb.Event{Seq: i, Payload: payload}); err != nil {
			f.finished <- "Send failed: " + status.Code(err).String()
			return err
		}
		f.sent.Add(1)
		if err := stream.Context().Err(); err != nil {
			f.finished <- "context ended: " + err.Error()
			return status.FromContextError(err).Err()
		}
	}
	f.finished <- "completed"
	return nil
}

// drain reads until the stream ends and returns how it ended.
func drain(stream grpc.ServerStreamingClient[pb.Event]) codes.Code {
	for {
		if _, err := stream.Recv(); err != nil {
			return status.Code(err)
		}
	}
}

func main() {
	lis, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		log.Fatal(err)
	}
	impl := &feed{finished: make(chan string, 4)}
	srv := grpc.NewServer()
	pb.RegisterFeedServer(srv, impl)
	go srv.Serve(lis)
	defer srv.Stop()
	conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
	if err != nil {
		log.Fatal(err)
	}
	defer conn.Close()
	c := pb.NewFeedClient(conn)

	// 1. A client that reads everything.
	stream, err := c.Subscribe(context.Background(), &pb.SubscribeRequest{Count: 1000, PayloadBytes: 1024})
	if err != nil {
		log.Fatal(err)
	}
	n := 0
	for {
		_, err := stream.Recv()
		if err == io.EOF {
			fmt.Println("reading client: io.EOF after", n, "events - the server returned nil")
			break
		}
		if err != nil {
			log.Fatal(err)
		}
		n++
	}
	fmt.Println("server saw:", <-impl.finished)

	// 2. A client that reads one event and then stops reading. The server has 100000 events of 1 KiB to send.
	impl.sent.Store(0)
	ctx, cancel := context.WithCancel(context.Background())
	stream, err = c.Subscribe(ctx, &pb.SubscribeRequest{Count: 100000, PayloadBytes: 1024})
	if err != nil {
		log.Fatal(err)
	}
	stream.Recv() // one event, then nothing
	time.Sleep(300 * time.Millisecond)
	stuck := impl.sent.Load()
	time.Sleep(300 * time.Millisecond)
	fmt.Printf("stalled client: server Send calls completed = %d, and %d after another 300ms (blocked, not buffering 100 MB)\n", stuck, impl.sent.Load())
	fmt.Printf("that is about %d KiB of 1 KiB events accepted by the transport before Send blocked\n", stuck)

	// 3. The client gives up. The blocked Send returns an error and the handler exits.
	cancel()
	fmt.Println("after the client cancelled, server saw:", <-impl.finished)
	fmt.Println("client's Recv calls end with:", drain(stream))

	// 4. A deadline on the whole stream.
	dctx, dcancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
	defer dcancel()
	stream, _ = c.Subscribe(dctx, &pb.SubscribeRequest{Count: 100000, PayloadBytes: 1024})
	stream.Recv()
	time.Sleep(400 * time.Millisecond)
	fmt.Println("stream with a 200ms deadline, client reads on and gets:", drain(stream))
	fmt.Println("server saw:", <-impl.finished)
}

mkdir -p bin
go build -o bin/protoc-gen-go google.golang.org/protobuf/cmd/protoc-gen-go
go build -o bin/protoc-gen-go-grpc google.golang.org/grpc/cmd/protoc-gen-go-grpc
PATH="$PWD/bin:$PATH" protoc --go_out=. --go_opt=module=example.test/lab --go-grpc_out=. --go-grpc_opt=module=example.test/lab proto/*.proto
go run .

reading client: io.EOF after 1000 events - the server returned nil
server saw: completed
stalled client: server Send calls completed = 368, and 368 after another 300ms (blocked, not buffering 100 MB)
that is about 368 KiB of 1 KiB events accepted by the transport before Send blocked
after the client cancelled, server saw: Send failed: Canceled
client's Recv calls end with: Canceled
stream with a 200ms deadline, client reads on and gets: DeadlineExceeded
server saw: Send failed: Canceled

The method receives the request and a generic stream type (grpc.ServerStreamingServer[pb.Event] in current grpc-go), calls stream.Send(event) for each message and returns when done. Returning nil ends the stream with an OK status; returning an error ends it with that status. The handler's context comes from stream.Context().

The end of a stream

The first scenario is the normal one. The client calls Recv in a loop until it gets an error. When the server returns nil, the client's Recv returns io.EOF, not a status error, after delivering all messages: 1000 events here. io.EOF is not a failure; it is the signal for a clean ending. Any other error is a failure, and status.Code(err) tells you which. Forgetting that distinction is the classic mistake: treating io.EOF as an error and logging every successful stream as a failure, or ignoring other errors because "the loop ended".

The server saw completed, as expected.

Backpressure

The second scenario is the one worth understanding. The server wants to send 100,000 events of 1 KiB each, 100 MB in total. The client reads one event and then stops calling Recv.

What should happen? The server must not buffer 100 MB for a client that is not reading. It does not. HTTP/2 has flow control: each stream and each connection has a window of bytes the receiver is willing to accept. When the client's receive buffers fill and nobody reads them, the window closes, the sender's transport stops writing, and Send blocks. The output shows it: the server completed 368 Send calls and then no more. Three hundred milliseconds later the count was still 368. In earlier runs of the same program it was 326 and 350. The exact number depends on window sizes, buffering in the transport and the dynamic window sizing that grpc-go does, so treat it as "a few hundred kilobytes", not a constant.

This blocking is backpressure, and it is a feature. A slow consumer slows the producer to its own pace without any code on your part. The cost shows up where you expect it: a stalled client holds a goroutine and a bit of memory on the server for as long as it stays connected. That is the reason to set deadlines and keepalive (keepalive), and to limit the number of concurrent streams per server.

If your handler produces data from a source that does not wait (a channel fed by a subscription, a database cursor), you decide what happens when Send blocks. Options: let the producer block too (safe, but a slow client can stall a shared source), buffer a bounded amount and drop the oldest or newest messages (good for state updates where only the latest matters), or disconnect clients that fall behind by too much. Never use an unbounded buffer in front of Send; that moves the memory problem from the transport to your code.

Cancellation

The third scenario: the client cancels its context. The server's handler was blocked inside Send. The cancellation arrives as a reset on the HTTP/2 stream, Send returns an error whose code is Canceled, the handler logs it and returns, and the goroutine goes away. The server recorded Send failed: Canceled. On the client side, Recv calls after the cancellation end with Canceled.

So a handler must handle two failure signals. Send returning a non-nil error means the stream is dead; return it. And stream.Context().Done() closing means the same thing from a different direction, which matters when the handler is waiting on something other than Send: a channel receive, a database call, a timer. A handler that generates data slowly would sit in its own wait for a long time after the client left, unless that wait is a select that includes stream.Context().Done():

for {
	select {
	case ev := <-events:
		if err := stream.Send(ev); err != nil {
			return err
		}
	case <-stream.Context().Done():
		return status.FromContextError(stream.Context().Err()).Err()
	}
}

Deadlines on streams

The fourth scenario sets a 200 ms deadline for the whole stream. A deadline applies to the entire call, not to each message. After 200 ms the call is over: the client reading afterwards got DeadlineExceeded, and the server saw its blocked Send fail.

For long-lived streams (a subscription that should run for hours), a single deadline is the wrong tool. Do not set one, and use keepalive and application-level heartbeats to detect dead peers instead. For bounded streams (exporting a report), a deadline is a good safety limit: it guarantees the stream cannot run forever whatever the client does.

Design guidance for streaming rpcs

Choose streaming for sequences, not for single big answers. Streaming a thousand small rows beats returning one 50 MB message, partly because of the default 4 MiB message size limit (message size limits) and partly because the client can start work on the first row.

Add a way to resume. A stream can break midway. Include a sequence number or cursor in each message and accept a "start after" value in the request, so a reconnecting client continues where it stopped. This is the same idea as Last-Event-ID in server-sent events.

Bound everything. A maximum number of messages or bytes per stream, a maximum duration and a maximum number of concurrent streams per client.

Send a final message or trailers with a summary when the client needs to verify it got everything: a count or checksum in the trailing metadata.

Do not hold locks across Send. Send can block for as long as the client takes. A mutex held during that call blocks every other goroutine that needs it. Copy the data out under the lock and send outside it.

Do not call Send from several goroutines. A stream's Send is not safe for concurrent use. Serialise sends through one goroutine or a mutex held only around the call to Send (and then consider the previous point).

Measure memory. The figure of roughly 350 KiB above is per stalled stream. A thousand stalled clients is about 350 MB. Plan limits with that arithmetic.

Testing streams

Test the four scenarios in the lab: a client that reads everything and sees io.EOF, a client that cancels midway and does not leak the handler goroutine, a stalled client where the server's memory stays bounded, and a deadline. Check for leaks with the goroutine count before and after, or use goleak (finding goroutine leaks). Run an in-memory server with bufconn for speed (testing gRPC services). The streaming semantics are documented in the gRPC core concepts guide and the grpc-go package documentation.