fix(client): retry watch streams cut mid-frame and word them as a broken connection
- commit
- b217ad38ac23fe53992f324fdbef503639fd2b99
- author
- brice · · 2026-10-10 09:45 UTC
- parent
- 2435de3
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +48 -1
M internal/adapters/client/client_test.go +3 -1
| @@ -92,8 +92,10 @@ func TestDescribe(t *testing.T) { | |||
| 92 | 92 | []string{"permission denied: missing scope repo:admin", "token list"}}, | |
| 93 | 93 | {"already exists", authed, connect.NewError(connect.CodeAlreadyExists, errors.New("name taken")), | |
| 94 | 94 | []string{"already exists: name taken"}}, | |
| 95 | - | {"invalid", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("bad name")), | |
| 95 | + | {"invalid", authed, connect.NewWireError(connect.CodeInvalidArgument, errors.New("bad name")), | |
| 96 | 96 | []string{"invalid request: bad name"}}, | |
| 97 | + | {"invalid raised by the client library", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("protocol error: incomplete envelope")), | |
| 98 | + | []string{"connection to http://srv.test broke", "incomplete envelope"}}, | |
| 97 | 99 | {"precondition", authed, connect.NewError(connect.CodeFailedPrecondition, errors.New("no such branch")), | |
| 98 | 100 | []string{"precondition failed: no such branch"}}, | |
| 99 | 101 | {"unavailable", authed, connect.NewError(connect.CodeUnavailable, errors.New("connection refused")), | |
| @@ -92,8 +92,10 @@ func TestDescribe(t *testing.T) { | |||||
| 92 | []string{"permission denied: missing scope repo:admin", "token list"}}, | 92 | []string{"permission denied: missing scope repo:admin", "token list"}}, | ||
| 93 | {"already exists", authed, connect.NewError(connect.CodeAlreadyExists, errors.New("name taken")), | 93 | {"already exists", authed, connect.NewError(connect.CodeAlreadyExists, errors.New("name taken")), | ||
| 94 | []string{"already exists: name taken"}}, | 94 | []string{"already exists: name taken"}}, | ||
| 95 | - | {"invalid", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("bad name")), | 95 | + | {"invalid", authed, connect.NewWireError(connect.CodeInvalidArgument, errors.New("bad name")), |
| 96 | []string{"invalid request: bad name"}}, | 96 | []string{"invalid request: bad name"}}, | ||
| 97 | + | {"invalid raised by the client library", authed, connect.NewError(connect.CodeInvalidArgument, errors.New("protocol error: incomplete envelope")), | |||
| 98 | + | []string{"connection to http://srv.test broke", "incomplete envelope"}}, | |||
| 97 | {"precondition", authed, connect.NewError(connect.CodeFailedPrecondition, errors.New("no such branch")), | 99 | {"precondition", authed, connect.NewError(connect.CodeFailedPrecondition, errors.New("no such branch")), | ||
| 98 | []string{"precondition failed: no such branch"}}, | 100 | []string{"precondition failed: no such branch"}}, | ||
| 99 | {"unavailable", authed, connect.NewError(connect.CodeUnavailable, errors.New("connection refused")), | 101 | {"unavailable", authed, connect.NewError(connect.CodeUnavailable, errors.New("connection refused")), | ||
M internal/adapters/client/errors.go +8 -0
| @@ -58,6 +58,14 @@ func (c *Client) Describe(err error) error { | |||
| 58 | 58 | case connect.CodeAlreadyExists: | |
| 59 | 59 | text = with("already exists") | |
| 60 | 60 | case connect.CodeInvalidArgument: | |
| 61 | + | if !connect.IsWireError(err) { | |
| 62 | + | // Raised by the client library: the answer was cut or garbled. | |
| 63 | + | text = "the connection to " + c.cfg.URL + " broke or the answer was malformed" | |
| 64 | + | if msg != "" { | |
| 65 | + | text += ": " + msg | |
| 66 | + | } | |
| 67 | + | break | |
| 68 | + | } | |
| 61 | 69 | text = with("invalid request") | |
| 62 | 70 | case connect.CodeFailedPrecondition: | |
| 63 | 71 | text = with("precondition failed") | |
| @@ -58,6 +58,14 @@ func (c *Client) Describe(err error) error { | |||||
| 58 | case connect.CodeAlreadyExists: | 58 | case connect.CodeAlreadyExists: | ||
| 59 | text = with("already exists") | 59 | text = with("already exists") | ||
| 60 | case connect.CodeInvalidArgument: | 60 | case connect.CodeInvalidArgument: | ||
| 61 | + | if !connect.IsWireError(err) { | |||
| 62 | + | // Raised by the client library: the answer was cut or garbled. | |||
| 63 | + | text = "the connection to " + c.cfg.URL + " broke or the answer was malformed" | |||
| 64 | + | if msg != "" { | |||
| 65 | + | text += ": " + msg | |||
| 66 | + | } | |||
| 67 | + | break | |||
| 68 | + | } | |||
| 61 | text = with("invalid request") | 69 | text = with("invalid request") | ||
| 62 | case connect.CodeFailedPrecondition: | 70 | case connect.CodeFailedPrecondition: | ||
| 63 | text = with("precondition failed") | 71 | text = with("precondition failed") | ||
M internal/adapters/client/watch.go +6 -0
| @@ -134,6 +134,12 @@ func retryable(err error) bool { | |||
| 134 | 134 | if !errors.As(err, &ce) { | |
| 135 | 135 | return true | |
| 136 | 136 | } | |
| 137 | + | // A connection cut in the middle of a stream is reported by Connect as an | |
| 138 | + | // invalid_argument "incomplete envelope" error that the server never sent. | |
| 139 | + | // Only an invalid_argument that came over the wire (a bad topic) is final. | |
| 140 | + | if ce.Code() == connect.CodeInvalidArgument && !connect.IsWireError(err) { | |
| 141 | + | return true | |
| 142 | + | } | |
| 137 | 143 | switch ce.Code() { | |
| 138 | 144 | case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded, | |
| 139 | 145 | connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled: | |
| @@ -134,6 +134,12 @@ func retryable(err error) bool { | |||||
| 134 | if !errors.As(err, &ce) { | 134 | if !errors.As(err, &ce) { | ||
| 135 | return true | 135 | return true | ||
| 136 | } | 136 | } | ||
| 137 | + | // A connection cut in the middle of a stream is reported by Connect as an | |||
| 138 | + | // invalid_argument "incomplete envelope" error that the server never sent. | |||
| 139 | + | // Only an invalid_argument that came over the wire (a bad topic) is final. | |||
| 140 | + | if ce.Code() == connect.CodeInvalidArgument && !connect.IsWireError(err) { | |||
| 141 | + | return true | |||
| 142 | + | } | |||
| 137 | switch ce.Code() { | 143 | switch ce.Code() { | ||
| 138 | case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded, | 144 | case connect.CodeUnavailable, connect.CodeUnknown, connect.CodeInternal, connect.CodeDeadlineExceeded, | ||
| 139 | connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled: | 145 | connect.CodeAborted, connect.CodeResourceExhausted, connect.CodeCanceled: | ||
M internal/adapters/client/watch_test.go +31 -0
| @@ -5,6 +5,8 @@ package client_test | |||
| 5 | 5 | import ( | |
| 6 | 6 | "context" | |
| 7 | 7 | "errors" | |
| 8 | + | "net/http" | |
| 9 | + | "net/http/httptest" | |
| 8 | 10 | "sync" | |
| 9 | 11 | "testing" | |
| 10 | 12 | "time" | |
| @@ -205,3 +207,32 @@ func TestWatchResumesFromGivenID(t *testing.T) { | |||
| 205 | 207 | _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil }) | |
| 206 | 208 | assert.Equal(t, "abc", first) | |
| 207 | 209 | } | |
| 210 | + | ||
| 211 | + | // A connection cut mid-stream surfaces as a client-side invalid_argument; it | |
| 212 | + | // must be retried, unlike an invalid_argument the server sent. | |
| 213 | + | func TestWatchRetriesCutConnections(t *testing.T) { | |
| 214 | + | ctx, cancel := context.WithCancel(t.Context()) | |
| 215 | + | defer cancel() | |
| 216 | + | calls := 0 | |
| 217 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |
| 218 | + | calls++ | |
| 219 | + | if calls == 1 { | |
| 220 | + | // Announce a frame and hang up before sending it. | |
| 221 | + | w.Header().Set("Content-Type", "application/connect+proto") | |
| 222 | + | _, _ = w.Write([]byte{0, 0, 0, 0, 9}) | |
| 223 | + | return | |
| 224 | + | } | |
| 225 | + | cancel() | |
| 226 | + | })) | |
| 227 | + | defer srv.Close() | |
| 228 | + | c, err := client.New(client.Config{URL: srv.URL}) | |
| 229 | + | require.NoError(t, err) | |
| 230 | + | var notices []string | |
| 231 | + | require.NoError(t, c.Watch(ctx, client.WatchOptions{ | |
| 232 | + | Sleep: (&sleeps{}).sleep, | |
| 233 | + | OnNotice: func(m string) { notices = append(notices, m) }, | |
| 234 | + | }, func(*ermite.Event) error { return nil })) | |
| 235 | + | assert.GreaterOrEqual(t, calls, 2, "retried") | |
| 236 | + | require.NotEmpty(t, notices) | |
| 237 | + | assert.Contains(t, notices[0], "connection lost") | |
| 238 | + | } | |
| @@ -5,6 +5,8 @@ package client_test | |||||
| 5 | import ( | 5 | import ( | ||
| 6 | "context" | 6 | "context" | ||
| 7 | "errors" | 7 | "errors" | ||
| 8 | + | "net/http" | |||
| 9 | + | "net/http/httptest" | |||
| 8 | "sync" | 10 | "sync" | ||
| 9 | "testing" | 11 | "testing" | ||
| 10 | "time" | 12 | "time" | ||
| @@ -205,3 +207,32 @@ func TestWatchResumesFromGivenID(t *testing.T) { | |||||
| 205 | _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil }) | 207 | _ = c.Watch(t.Context(), client.WatchOptions{ResumeAfter: "abc"}, func(*ermite.Event) error { return nil }) | ||
| 206 | assert.Equal(t, "abc", first) | 208 | assert.Equal(t, "abc", first) | ||
| 207 | } | 209 | } | ||
| 210 | + | ||||
| 211 | + | // A connection cut mid-stream surfaces as a client-side invalid_argument; it | |||
| 212 | + | // must be retried, unlike an invalid_argument the server sent. | |||
| 213 | + | func TestWatchRetriesCutConnections(t *testing.T) { | |||
| 214 | + | ctx, cancel := context.WithCancel(t.Context()) | |||
| 215 | + | defer cancel() | |||
| 216 | + | calls := 0 | |||
| 217 | + | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { | |||
| 218 | + | calls++ | |||
| 219 | + | if calls == 1 { | |||
| 220 | + | // Announce a frame and hang up before sending it. | |||
| 221 | + | w.Header().Set("Content-Type", "application/connect+proto") | |||
| 222 | + | _, _ = w.Write([]byte{0, 0, 0, 0, 9}) | |||
| 223 | + | return | |||
| 224 | + | } | |||
| 225 | + | cancel() | |||
| 226 | + | })) | |||
| 227 | + | defer srv.Close() | |||
| 228 | + | c, err := client.New(client.Config{URL: srv.URL}) | |||
| 229 | + | require.NoError(t, err) | |||
| 230 | + | var notices []string | |||
| 231 | + | require.NoError(t, c.Watch(ctx, client.WatchOptions{ | |||
| 232 | + | Sleep: (&sleeps{}).sleep, | |||
| 233 | + | OnNotice: func(m string) { notices = append(notices, m) }, | |||
| 234 | + | }, func(*ermite.Event) error { return nil })) | |||
| 235 | + | assert.GreaterOrEqual(t, calls, 2, "retried") | |||
| 236 | + | require.NotEmpty(t, notices) | |||
| 237 | + | assert.Contains(t, notices[0], "connection lost") | |||
| 238 | + | } | |||