diff --git a/docs/operations.md b/docs/operations.md index 2bed681..7edbb90 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -233,7 +233,12 @@ 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 +(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 bc30d4c..d8c32b3 100644 --- a/internal/backend/search/native.go +++ b/internal/backend/search/native.go @@ -4,14 +4,19 @@ import ( "bufio" "bytes" "context" + "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" @@ -117,7 +122,9 @@ 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")) } @@ -138,7 +145,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 +228,69 @@ 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 + phaseReading +) + +var exchangePhases = [...]string{ + phaseConnecting: "while connecting", + 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. 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{ + TLSHandshakeStart: func() { p.reached.Store(phaseHandshake) }, + GotConn: func(httptrace.GotConnInfo) { p.reached.Store(phaseSending) }, + WroteRequest: func(info httptrace.WroteRequestInfo) { + if info.Err == nil { + p.reached.Store(phaseWaiting) + } + }, + GotFirstResponseByte: func() { p.reached.Store(phaseReading) }, + } + 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 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), errors.Is(err, syscall.EPIPE): + 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 0000000..bdba2cc --- /dev/null +++ b/internal/backend/search/native_exchange_test.go @@ -0,0 +1,122 @@ +package search + +import ( + "bytes" + "io" + "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) + } +} + +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() { + var held []net.Conn + for { + conn, err := listener.Accept() + if err != nil { + for _, conn := range held { + conn.Close() + } + return + } + held = append(held, conn) + } + }() + 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)") +} + +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) + } +}