End Search Natives that got no backend connection as not started

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
 

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{

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

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) {