Say where a Search Native exchange failed before its response

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

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)

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)
+	}
+}