broker/client: quiet more noisy logging

broker/client/list.go

@@ -97,6 +97,9 @@ func (pl *WatchedList) watch() {
 		} else {
 			stream = nil // Must restart.
 
+			if err == io.EOF {
+				attempt = 0 // Clean server-side close.
+			}
 			// Wait for back-off timer or context cancellation.
 			select {
 			case <-pl.ctx.Done():

broker/client/reader.go

@@ -95,10 +95,18 @@ func (r *Reader) Read(p []byte) (n int, err error) {
 
 	// Lazy initialization: begin the Read RPC.
 	if r.stream == nil {
-		if r.stream, err = r.client.Read(
-			pb.WithDispatchItemRoute(r.ctx, r.client, r.Request.Journal.String(), false),
-			&r.Request,
-		); err == nil {
+		var ctx context.Context
+
+		// Prefer a prior response header route, and fall back to the route cache.
+		if r.Response.Header != nil {
+			ctx = pb.WithDispatchRoute(r.ctx, r.Response.Header.Route, pb.ProcessSpec_ID{})
+		} else {
+			ctx = pb.WithDispatchItemRoute(r.ctx, r.client, r.Request.Journal.String(), false)
+		}
+		// Clear last response of previous stream.
+		r.Response = pb.ReadResponse{}
+
+		if r.stream, err = r.client.Read(ctx, &r.Request); err == nil {
 			n, err = r.Read(p) // Recurse to attempt read against opened |r.stream|.
 		} else {
 			err = mapGRPCCtxErr(r.ctx, err)

broker/client/retry_reader.go

@@ -117,7 +117,7 @@ func (rr *RetryReader) Read(p []byte) (n int, err error) {
 			// any data before being closed server-side, so clear `attempts` to
 			// reduce log noise: it's common to next see ErrNotJournalBroker
 			// on broker topology changes or when authorizations are refreshed.
-			attempt = 0
+			attempt = -1
 			squelch = true
 		default:
 		}
@@ -222,6 +222,8 @@ func backoff(attempt int) time.Duration {
 	// involves a couple of Nagle-like read delays (~30ms) as Etcd watch
 	// updates are applied by participants.
 	switch attempt {
+	case -1:
+		return 0
 	case 0, 1:
 		return time.Millisecond * 50
 	case 2, 3: