diff --git a/docs/operations.md b/docs/operations.md index 7edbb90..779a97a 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -230,22 +230,28 @@ raises no record alert when its backend fails. A complete response never counts here, whatever its HTTP status, because Weir does not interpret Native results. `response_incomplete` calls started but did not receive a complete response: -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 -(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. +the backend dropped the connection, a deadline or cancellation ended the call +after it was sent, 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 after it had a backend connection but before its +response head says where in its `UNAVAILABLE` failure: while sending the +request, while waiting for the response head or while reading it, with `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 -BSON or a missing `$db`), or, rarely, a cancellation, deadline or shutdown -landed just as the call was dispatched. A Native whose caller gives up while -waiting for a free Native slot is not counted. A lasting `not_started` rate -means clients send invalid Natives; read the failure messages in client logs. +BSON or a missing `$db`), a Search Native got no backend connection, or a +cancellation, deadline or shutdown ended the call before it was sent, which for +a Search Native includes the time spent connecting. A Search Native without a +connection fails `UNAVAILABLE` while connecting (which includes the DNS lookup) +or during the TLS handshake, with `refused`, `reset`, `closed` or `timeout` +when known; the SDK sends such a Native again within its resend budget, since +it took no effect. Lasting causes, such as an unknown host name or an untrusted +certificate, fail the same way on every send. A Native whose caller gives up +while waiting for a free Native slot is not counted. A lasting `not_started` +rate means clients send invalid Natives or the backend is unreachable; read the +failure messages in client logs. ### WeirQueueWaitHigh diff --git a/internal/backend/search/adapter.go b/internal/backend/search/adapter.go index ce773d2..73070ae 100644 --- a/internal/backend/search/adapter.go +++ b/internal/backend/search/adapter.go @@ -98,6 +98,8 @@ func Open(ctx context.Context, cfg Config) (*Adapter, error) { client := &http.Client{Transport: transport, CheckRedirect: noRedirect} nativeTransport := newTransport(settings.ConnectTimeout, settings.IdleTimeout) nativeTransport.TLSClientConfig = tlsConfig + // A Native that failed before getting a connection ends NOT_STARTED, + // which is true only while net/http cannot reuse and replay (native.go). nativeTransport.DisableKeepAlives = true nativeClient := &http.Client{Transport: nativeTransport, CheckRedirect: noRedirect} a := &Adapter{ diff --git a/internal/backend/search/native.go b/internal/backend/search/native.go index d8c32b3..92c8c5e 100644 --- a/internal/backend/search/native.go +++ b/internal/backend/search/native.go @@ -145,7 +145,19 @@ 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 "+phase.describe(err))) + // Nothing is written before the transport hands over a connection, so + // without one the backend never saw the request and a client may send + // it again, whatever its method. This holds because the Native + // transport never reuses a connection (Open sets DisableKeepAlives): + // net/http replays requests only on reused connections. + reached := phase.reached.Load() + connected := reached >= phaseSending + message := "Native HTTP exchange not started " + if connected { + message = "Native HTTP exchange incomplete " + } + failure := nativeExchangeFailure(ctx, message+describeExchange(reached, err)) + return protocol.NativeFailure(connected, failure) } defer response.Body.Close() if response.StatusCode < 100 || response.StatusCode > 599 { @@ -246,32 +258,43 @@ var exchangePhases = [...]string{ } // 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. +// before the response head can say where it happened and whether request bytes +// may have reached the backend. The phase only advances: a dial that net/http +// abandoned can still start its TLS handshake after the exchange went on. +// Only hooks that run in exchange order are used, since a losing parallel dial +// can report its connect late and net/http reports a failed write too. type exchangePhase struct { reached atomic.Int32 } +func (p *exchangePhase) advance(phase int32) { + for { + current := p.reached.Load() + if current >= phase || p.reached.CompareAndSwap(current, phase) { + return + } + } +} + func (p *exchangePhase) trace() *httptrace.ClientTrace { trace := &httptrace.ClientTrace{ - TLSHandshakeStart: func() { p.reached.Store(phaseHandshake) }, - GotConn: func(httptrace.GotConnInfo) { p.reached.Store(phaseSending) }, + TLSHandshakeStart: func() { p.advance(phaseHandshake) }, + GotConn: func(httptrace.GotConnInfo) { p.advance(phaseSending) }, WroteRequest: func(info httptrace.WroteRequestInfo) { if info.Err == nil { - p.reached.Store(phaseWaiting) + p.advance(phaseWaiting) } }, - GotFirstResponseByte: func() { p.reached.Store(phaseReading) }, + GotFirstResponseByte: func() { p.advance(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()] +// describeExchange names where and how an exchange that reached the given +// phase failed, from a fixed vocabulary. It never quotes err, which carries the +// backend URL. +func describeExchange(reached int32, err error) string { + where := exchangePhases[reached] var request *url.Error if errors.As(err, &request) { err = request.Err diff --git a/internal/backend/search/native_exchange_test.go b/internal/backend/search/native_exchange_test.go index bdba2cc..ee0d11d 100644 --- a/internal/backend/search/native_exchange_test.go +++ b/internal/backend/search/native_exchange_test.go @@ -2,15 +2,19 @@ package search import ( "bytes" + "context" "io" "net" "net/http" "net/http/httptest" + "net/http/httptrace" "strings" "testing" "time" pb "github.com/batchstream/weir-protocol/api/weir/v1" + + "github.com/batchstream/weir/internal/testutil" ) // nativeAdapter forwards Natives to url with the production Native transport @@ -26,12 +30,12 @@ func nativeAdapter(t *testing.T, url string, timeout time.Duration) *Adapter { return adapter } -func requireExchangeFailure(t *testing.T, adapter *Adapter, want string) { +func requireExchangeFailure(t *testing.T, adapter *Adapter, completion pb.NativeCompletion, 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) + if end.Completion != completion || capture.head != nil { + t.Fatalf("completion %s, want %s without output", end.Completion, completion) } failure := end.GetFailure() if failure.GetCode() != pb.FailureCode_UNAVAILABLE || failure.GetMessage() != want { @@ -47,16 +51,18 @@ func TestNativeExchangeFailureNamesARefusedConnection(t *testing.T) { address := listener.Addr().String() listener.Close() adapter := nativeAdapter(t, "http://"+address, time.Second) - requireExchangeFailure(t, adapter, "Native HTTP exchange incomplete while connecting (refused)") + requireExchangeFailure(t, adapter, pb.NativeCompletion_NATIVE_NOT_STARTED, "Native HTTP exchange not started while connecting (refused)") } -func TestNativeExchangeFailureNamesATLSHandshakeTimeout(t *testing.T) { +// silentListener accepts connections and never answers, so a client's TLS +// handshake stalls, and returns its address. +func silentListener(t *testing.T) string { + t.Helper() 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 { @@ -70,8 +76,61 @@ func TestNativeExchangeFailureNamesATLSHandshakeTimeout(t *testing.T) { 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)") + return listener.Addr().String() +} + +func TestNativeExchangeFailureNamesATLSHandshakeTimeout(t *testing.T) { + adapter := nativeAdapter(t, "https://"+silentListener(t), 200*time.Millisecond) + requireExchangeFailure(t, adapter, pb.NativeCompletion_NATIVE_NOT_STARTED, "Native HTTP exchange not started during the TLS handshake (timeout)") +} + +func TestNativeExchangeFailureNamesAnUntrustedCertificate(t *testing.T) { + backend := httptest.NewTLSServer(http.NotFoundHandler()) + t.Cleanup(backend.Close) + adapter := nativeAdapter(t, backend.URL, time.Second) + requireExchangeFailure(t, adapter, pb.NativeCompletion_NATIVE_NOT_STARTED, "Native HTTP exchange not started during the TLS handshake") +} + +func TestNativeExchangeFailureNamesAnUnresolvableBackend(t *testing.T) { + adapter := nativeAdapter(t, "http://weir-native-test.invalid", time.Second) + request := nativeRequest(t, "records", "GET", "/_search") + end, capture := runNative(t, adapter, request, nil) + failure := end.GetFailure() + want := "Native HTTP exchange not started while connecting" + if end.Completion != pb.NativeCompletion_NATIVE_NOT_STARTED || capture.head != nil || + failure.GetCode() != pb.FailureCode_UNAVAILABLE || !strings.HasPrefix(failure.GetMessage(), want) { + t.Fatalf("completion %s failure %s %q, want NATIVE_NOT_STARTED UNAVAILABLE %q", end.Completion, failure.GetCode(), failure.GetMessage(), want) + } +} + +// A deadline before the exchange had a connection also ends it not started, +// with the deadline as its failure, since the backend never saw it. +func TestNativeExchangeDeadlineBeforeAConnectionEndsNotStarted(t *testing.T) { + adapter := nativeAdapter(t, "https://"+silentListener(t), 2*time.Second) + request := nativeRequest(t, "records", "GET", "/_search") + setNativeBody(t, request, nil) + capture := &nativeCapture{} + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) + defer cancel() + end := adapter.ExecuteNative(ctx, testutil.NativeCommand(request), capture.Emit) + if end.Completion != pb.NativeCompletion_NATIVE_NOT_STARTED || end.GetFailure().GetCode() != pb.FailureCode_DEADLINE_EXCEEDED || capture.head != nil { + t.Fatalf("completion %s failure %s, want NATIVE_NOT_STARTED DEADLINE_EXCEEDED", end.Completion, end.GetFailure().GetCode()) + } +} + +// A handshake reported after the exchange got its connection, as from a dial +// net/http abandoned, must not make a sent request look unsent. +func TestExchangePhaseNeverMovesBack(t *testing.T) { + phase := &exchangePhase{} + trace := phase.trace() + connection := httptrace.GotConnInfo{} + trace.GotConn(connection) + written := httptrace.WroteRequestInfo{} + trace.WroteRequest(written) + trace.TLSHandshakeStart() + if reached := phase.reached.Load(); reached != phaseWaiting { + t.Fatalf("phase %s after a late handshake, want %s", exchangePhases[reached], exchangePhases[phaseWaiting]) + } } func TestNativeExchangeFailureNamesAConnectionClosedBeforeTheResponse(t *testing.T) { @@ -86,7 +145,7 @@ func TestNativeExchangeFailureNamesAConnectionClosedBeforeTheResponse(t *testing 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)") + requireExchangeFailure(t, adapter, pb.NativeCompletion_RESPONSE_INCOMPLETE, "Native HTTP exchange incomplete while waiting for the response head (closed)") } func TestNativeExchangeFailureNamesARequestCutOffWhileSending(t *testing.T) {