From adb18ec283dd1c8d92df9b2156924d24ec59ac4c Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Tue, 10 Nov 2026 12:28:02 +0800 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=A5=85=20fix:=20say=20where=20a=20Sea?= =?UTF-8?q?rch=20Native=20exchange=20failed=20before=20its=20response?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/operations.md | 6 +- internal/backend/search/native.go | 72 ++++++++++++++- .../backend/search/native_exchange_test.go | 87 +++++++++++++++++++ 3 files changed, 163 insertions(+), 2 deletions(-) create mode 100644 internal/backend/search/native_exchange_test.go diff --git a/docs/operations.md b/docs/operations.md index 2bed6815..1d00178f 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -233,7 +233,11 @@ here, whatever its HTTP status, because Weir does not interpret Native results. the backend was unreachable or dropped the connection, a deadline or cancellation ended the call before or during the response, or the client stopped reading. Such a command may have taken effect. Check backend health and -network paths first, then client deadlines. +network paths first, then client deadlines. A Search Native that failed before +its response head says where in its `UNAVAILABLE` failure: while connecting, +during the TLS handshake, while sending the request or while waiting for the +response head, with `refused`, `reset`, `closed` or `timeout` when known. Each +Search Native opens its own backend connection. `not_started` calls ended before Weir sent them, so the backend never saw them. The request was malformed (an invalid HTTP/1.1 head or Content-Length, invalid diff --git a/internal/backend/search/native.go b/internal/backend/search/native.go index bc30d4c3..181940c5 100644 --- a/internal/backend/search/native.go +++ b/internal/backend/search/native.go @@ -4,14 +4,20 @@ import ( "bufio" "bytes" "context" + "crypto/tls" + "errors" "fmt" "io" "mime" + "net" "net/http" + "net/http/httptrace" "net/url" "sort" "strings" "sync" + "sync/atomic" + "syscall" "github.com/batchstream/weir-protocol/api/protocol" pb "github.com/batchstream/weir-protocol/api/weir/v1" @@ -121,6 +127,8 @@ func (a *Adapter) ExecuteNative(ctx context.Context, command *pb.Command, emit e if err != nil { return protocol.NativeFailure(false, protocol.Fail(pb.FailureCode_INVALID_ARGUMENT, "invalid HTTP request")) } + var phase exchangePhase + request = request.WithContext(httptrace.WithClientTrace(request.Context(), phase.trace())) a.configureRequest(request) for name, values := range d.Header { @@ -138,7 +146,7 @@ func (a *Adapter) ExecuteNative(ctx context.Context, command *pb.Command, emit e } response, err := a.nativeClient.Do(request) if err != nil { - return protocol.NativeFailure(true, nativeExchangeFailure(ctx, "Native HTTP exchange incomplete")) + return protocol.NativeFailure(true, nativeExchangeFailure(ctx, "Native HTTP exchange incomplete "+phase.describe(err))) } defer response.Body.Close() if response.StatusCode < 100 || response.StatusCode > 599 { @@ -221,6 +229,68 @@ func (b *nativeHTTPBody) Close() error { return nil } +// Phases of a Native HTTP exchange before its response head arrives. +const ( + phaseConnecting int32 = iota + phaseHandshake + phaseSending + phaseWaiting +) + +var exchangePhases = [...]string{ + phaseConnecting: "while connecting", + phaseHandshake: "during the TLS handshake", + phaseSending: "while sending the request", + phaseWaiting: "while waiting for the response head", +} + +// exchangePhase records how far a Native HTTP exchange got, so that a failure +// before the response head can say where it happened. The transport reports +// phases from its own goroutines. +type exchangePhase struct { + reached atomic.Int32 +} + +func (p *exchangePhase) trace() *httptrace.ClientTrace { + trace := &httptrace.ClientTrace{ + ConnectDone: func(_, _ string, err error) { + if err == nil { + p.reached.Store(phaseSending) + } + }, + TLSHandshakeStart: func() { p.reached.Store(phaseHandshake) }, + TLSHandshakeDone: func(_ tls.ConnectionState, err error) { + if err == nil { + p.reached.Store(phaseSending) + } + }, + GotConn: func(httptrace.GotConnInfo) { p.reached.Store(phaseSending) }, + WroteRequest: func(httptrace.WroteRequestInfo) { + p.reached.Store(phaseWaiting) + }, + } + return trace +} + +// describe names where and how the exchange failed from a fixed vocabulary. +// It never quotes err, which carries the backend URL. +func (p *exchangePhase) describe(err error) string { + where := exchangePhases[p.reached.Load()] + var network net.Error + switch { + case errors.Is(err, syscall.ECONNREFUSED): + return where + " (refused)" + case errors.Is(err, syscall.ECONNRESET): + return where + " (reset)" + case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF): + return where + " (closed)" + case errors.As(err, &network) && network.Timeout(): + return where + " (timeout)" + default: + return where + } +} + func nativeExchangeFailure(ctx context.Context, message string) *pb.Failure { if ctx.Err() != nil { return protocol.ContextFailure(ctx) diff --git a/internal/backend/search/native_exchange_test.go b/internal/backend/search/native_exchange_test.go new file mode 100644 index 00000000..0246977b --- /dev/null +++ b/internal/backend/search/native_exchange_test.go @@ -0,0 +1,87 @@ +package search + +import ( + "net" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + pb "github.com/batchstream/weir-protocol/api/weir/v1" +) + +// nativeAdapter forwards Natives to url with the production Native transport +// settings and the given connect timeout. +func nativeAdapter(t *testing.T, url string, timeout time.Duration) *Adapter { + t.Helper() + transport := newTransport(timeout, time.Second) + transport.DisableKeepAlives = true + t.Cleanup(transport.CloseIdleConnections) + client := &http.Client{Transport: transport, CheckRedirect: noRedirect} + config := Config{URL: url} + adapter := &Adapter{config: config, nativeClient: client, ctx: t.Context()} + return adapter +} + +func requireExchangeFailure(t *testing.T, adapter *Adapter, want string) { + t.Helper() + request := nativeRequest(t, "records", "GET", "/_search") + end, capture := runNative(t, adapter, request, nil) + if end.Completion != pb.NativeCompletion_RESPONSE_INCOMPLETE || capture.head != nil { + t.Fatal("failed exchange published output", end) + } + failure := end.GetFailure() + if failure.GetCode() != pb.FailureCode_UNAVAILABLE || failure.GetMessage() != want { + t.Fatalf("failure %s %q, want UNAVAILABLE %q", failure.GetCode(), failure.GetMessage(), want) + } + if strings.Contains(failure.GetMessage(), "127.0.0.1") { + t.Fatal("failure quotes the backend address", failure.GetMessage()) + } +} + +func TestNativeExchangeFailureNamesARefusedConnection(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + address := listener.Addr().String() + listener.Close() + adapter := nativeAdapter(t, "http://"+address, time.Second) + requireExchangeFailure(t, adapter, "Native HTTP exchange incomplete while connecting (refused)") +} + +func TestNativeExchangeFailureNamesATLSHandshakeTimeout(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { listener.Close() }) + // Accept and never answer, so the client's TLS handshake times out. + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + t.Cleanup(func() { conn.Close() }) + } + }() + adapter := nativeAdapter(t, "https://"+listener.Addr().String(), 200*time.Millisecond) + requireExchangeFailure(t, adapter, "Native HTTP exchange incomplete during the TLS handshake (timeout)") +} + +func TestNativeExchangeFailureNamesAConnectionClosedBeforeTheResponse(t *testing.T) { + handler := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + conn, _, err := w.(http.Hijacker).Hijack() + if err != nil { + t.Error(err) + return + } + conn.Close() + }) + backend := httptest.NewServer(handler) + t.Cleanup(backend.Close) + adapter := nativeAdapter(t, backend.URL, time.Second) + requireExchangeFailure(t, adapter, "Native HTTP exchange incomplete while waiting for the response head (closed)") +} From a00a09b90eada85714fe6fb1b7c6b279e7138fd6 Mon Sep 17 00:00:00 2001 From: Randy <857271370@qq.com> Date: Tue, 10 Nov 2026 12:43:01 +0800 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=90=9B=20fix:=20report=20a=20failed?= =?UTF-8?q?=20request=20write=20as=20sending,=20not=20waiting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit net/http calls WroteRequest even when the write fails, so a connection dropped mid-upload read as waiting for the response head. Advance only on a successful write, drop the connect and handshake-done hooks that a losing parallel dial can report out of order, name a failure while reading the head, and treat a broken pipe as closed. --- docs/operations.md | 9 ++-- internal/backend/search/native.go | 38 ++++++++-------- .../backend/search/native_exchange_test.go | 43 +++++++++++++++++-- 3 files changed, 63 insertions(+), 27 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index 1d00178f..7edbb90d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -234,10 +234,11 @@ the backend was unreachable or dropped the connection, a deadline or cancellation ended the call before or during the response, or the client stopped reading. Such a command may have taken effect. Check backend health and network paths first, then client deadlines. A Search Native that failed before -its response head says where in its `UNAVAILABLE` failure: while connecting, -during the TLS handshake, while sending the request or while waiting for the -response head, with `refused`, `reset`, `closed` or `timeout` when known. Each -Search Native opens its own backend connection. +its response head says where in its `UNAVAILABLE` failure: while connecting +(which includes the DNS lookup), during the TLS handshake, while sending the +request, while waiting for the response head or while reading it, with +`refused`, `reset`, `closed` or `timeout` when known. Each Search Native opens +its own backend connection. `not_started` calls ended before Weir sent them, so the backend never saw them. The request was malformed (an invalid HTTP/1.1 head or Content-Length, invalid diff --git a/internal/backend/search/native.go b/internal/backend/search/native.go index 181940c5..d8c32b39 100644 --- a/internal/backend/search/native.go +++ b/internal/backend/search/native.go @@ -4,7 +4,6 @@ import ( "bufio" "bytes" "context" - "crypto/tls" "errors" "fmt" "io" @@ -123,12 +122,12 @@ func (a *Adapter) ExecuteNative(ctx context.Context, command *pb.Command, emit e if len(nativeBody) != 0 { body = bytes.NewReader(nativeBody) } - request, err := http.NewRequestWithContext(ctx, d.Method, endpoint, body) + var phase exchangePhase + traced := httptrace.WithClientTrace(ctx, phase.trace()) + request, err := http.NewRequestWithContext(traced, d.Method, endpoint, body) if err != nil { return protocol.NativeFailure(false, protocol.Fail(pb.FailureCode_INVALID_ARGUMENT, "invalid HTTP request")) } - var phase exchangePhase - request = request.WithContext(httptrace.WithClientTrace(request.Context(), phase.trace())) a.configureRequest(request) for name, values := range d.Header { @@ -235,6 +234,7 @@ const ( phaseHandshake phaseSending phaseWaiting + phaseReading ) var exchangePhases = [...]string{ @@ -242,32 +242,28 @@ var exchangePhases = [...]string{ phaseHandshake: "during the TLS handshake", phaseSending: "while sending the request", phaseWaiting: "while waiting for the response head", + phaseReading: "while reading the response head", } // exchangePhase records how far a Native HTTP exchange got, so that a failure -// before the response head can say where it happened. The transport reports -// phases from its own goroutines. +// before the response head can say where it happened. Only hooks that run in +// exchange order are used: a losing parallel dial can report its connect +// after the winner's TLS handshake has begun, and net/http reports a failed +// request write too. type exchangePhase struct { reached atomic.Int32 } func (p *exchangePhase) trace() *httptrace.ClientTrace { trace := &httptrace.ClientTrace{ - ConnectDone: func(_, _ string, err error) { - if err == nil { - p.reached.Store(phaseSending) - } - }, TLSHandshakeStart: func() { p.reached.Store(phaseHandshake) }, - TLSHandshakeDone: func(_ tls.ConnectionState, err error) { - if err == nil { - p.reached.Store(phaseSending) + GotConn: func(httptrace.GotConnInfo) { p.reached.Store(phaseSending) }, + WroteRequest: func(info httptrace.WroteRequestInfo) { + if info.Err == nil { + p.reached.Store(phaseWaiting) } }, - GotConn: func(httptrace.GotConnInfo) { p.reached.Store(phaseSending) }, - WroteRequest: func(httptrace.WroteRequestInfo) { - p.reached.Store(phaseWaiting) - }, + GotFirstResponseByte: func() { p.reached.Store(phaseReading) }, } return trace } @@ -276,13 +272,17 @@ func (p *exchangePhase) trace() *httptrace.ClientTrace { // It never quotes err, which carries the backend URL. func (p *exchangePhase) describe(err error) string { where := exchangePhases[p.reached.Load()] + var request *url.Error + if errors.As(err, &request) { + err = request.Err + } var network net.Error switch { case errors.Is(err, syscall.ECONNREFUSED): return where + " (refused)" case errors.Is(err, syscall.ECONNRESET): return where + " (reset)" - case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF): + case errors.Is(err, io.EOF), errors.Is(err, io.ErrUnexpectedEOF), errors.Is(err, syscall.EPIPE): return where + " (closed)" case errors.As(err, &network) && network.Timeout(): return where + " (timeout)" diff --git a/internal/backend/search/native_exchange_test.go b/internal/backend/search/native_exchange_test.go index 0246977b..bdba2cc0 100644 --- a/internal/backend/search/native_exchange_test.go +++ b/internal/backend/search/native_exchange_test.go @@ -1,6 +1,8 @@ package search import ( + "bytes" + "io" "net" "net/http" "net/http/httptest" @@ -35,9 +37,6 @@ func requireExchangeFailure(t *testing.T, adapter *Adapter, want string) { if failure.GetCode() != pb.FailureCode_UNAVAILABLE || failure.GetMessage() != want { t.Fatalf("failure %s %q, want UNAVAILABLE %q", failure.GetCode(), failure.GetMessage(), want) } - if strings.Contains(failure.GetMessage(), "127.0.0.1") { - t.Fatal("failure quotes the backend address", failure.GetMessage()) - } } func TestNativeExchangeFailureNamesARefusedConnection(t *testing.T) { @@ -59,12 +58,16 @@ func TestNativeExchangeFailureNamesATLSHandshakeTimeout(t *testing.T) { t.Cleanup(func() { listener.Close() }) // Accept and never answer, so the client's TLS handshake times out. go func() { + var held []net.Conn for { conn, err := listener.Accept() if err != nil { + for _, conn := range held { + conn.Close() + } return } - t.Cleanup(func() { conn.Close() }) + held = append(held, conn) } }() adapter := nativeAdapter(t, "https://"+listener.Addr().String(), 200*time.Millisecond) @@ -85,3 +88,35 @@ func TestNativeExchangeFailureNamesAConnectionClosedBeforeTheResponse(t *testing adapter := nativeAdapter(t, backend.URL, time.Second) requireExchangeFailure(t, adapter, "Native HTTP exchange incomplete while waiting for the response head (closed)") } + +func TestNativeExchangeFailureNamesARequestCutOffWhileSending(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { listener.Close() }) + // Read the start of the request, then drop the connection mid-upload. + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + buffer := make([]byte, 4096) + _, _ = io.ReadFull(conn, buffer) + conn.Close() + } + }() + adapter := nativeAdapter(t, "http://"+listener.Addr().String(), time.Second) + request := nativeRequest(t, "records", "POST", "/_search") + body := bytes.Repeat([]byte("x"), 32<<20) + end, capture := runNative(t, adapter, request, body) + if end.Completion != pb.NativeCompletion_RESPONSE_INCOMPLETE || capture.head != nil { + t.Fatal("failed exchange published output", end) + } + failure := end.GetFailure() + want := "Native HTTP exchange incomplete while sending the request" + if failure.GetCode() != pb.FailureCode_UNAVAILABLE || !strings.HasPrefix(failure.GetMessage(), want) { + t.Fatalf("failure %s %q, want UNAVAILABLE %q", failure.GetCode(), failure.GetMessage(), want) + } +}