feat: add timeout for slow instances

CHANGELOG.md

@@ -15,6 +15,9 @@ All notable changes to `src-cli` are documented in this file.
 
 ### Changed
 
+- HTTP requests now fail instead of hanging forever if the server does not start responding within 1 minute. Set the `SRC_RESPONSE_HEADER_TIMEOUT` environment variable to change this timeout, or to `0` to disable it. Responses that stream data for a long time (for example, large search job results) are not affected.
+- `src search-jobs logs` and `src search-jobs results` now use the standard API client, gaining proxy support, `-insecure-skip-verify`, and cross-host redirect protection, and now report an error on non-200 responses instead of writing the error page into the output.
+
 ### Removed
 
 - Removed `src sbom` and `src signature` commands. SBOMs and container signatures are no longer published as of Sourcegraph 7.1.0.

README.md

@@ -197,6 +197,10 @@ mv /usr/local/bin/src /usr/local/bin/src-cli
 
 You can then invoke it via `src-cli`.
 
+## Timeouts
+
+`src` waits up to 1 minute for the server to start responding to a request. To change this, set the `SRC_RESPONSE_HEADER_TIMEOUT` environment variable to a duration such as `30s` or `10m`, or to `0` to disable the timeout. This timeout only applies until the server sends its response headers — responses that stream data for a long time, such as large search job results, are not interrupted.
+
 ## Telemetry
 
 `src` includes the operating system and architecture in the `User-Agent` header sent to Sourcegraph. For example, running `src` version 3.21.10 on an x86-64 Linux host will result in this header:

cmd/src/search_jobs.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"io"
+	"net/http"
 	"strings"
 
 	"github.com/sourcegraph/src-cli/internal/api"
@@ -167,6 +169,31 @@ func parseSearchJobsArgs(flagSet *flag.FlagSet, args []string) error {
 	return nil
 }
 
+// fetchSearchJobFile downloads a file (logs or results) belonging to a search
+// job. The request goes through the API client so that it picks up the
+// configured transport (timeouts, proxy, TLS and redirect handling).
+func fetchSearchJobFile(client api.Client, fileURL string) (io.ReadCloser, error) {
+	req, err := http.NewRequest("GET", fileURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Add("Authorization", "token "+cfg.accessToken)
+
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
+		resp.Body.Close()
+		return nil, fmt.Errorf("error: %s\n\n%s", resp.Status, body)
+	}
+
+	return resp.Body, nil
+}
+
 // validateJobID validates that a job ID was provided
 func validateJobID(args []string) (string, error) {
 	if len(args) != 1 {

cmd/src/search_jobs_logs.go

@@ -4,32 +4,19 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"net/http"
 	"os"
 
 	"github.com/sourcegraph/src-cli/internal/api"
 	"github.com/sourcegraph/src-cli/internal/cmderrors"
 )
 
 // fetchJobLogs retrieves logs for a search job from its log URL
-func fetchJobLogs(jobID string, logURL string) (io.ReadCloser, error) {
+func fetchJobLogs(client api.Client, jobID string, logURL string) (io.ReadCloser, error) {
 	if logURL == "" {
 		return nil, fmt.Errorf("no logs URL found for search job %s", jobID)
 	}
 
-	req, err := http.NewRequest("GET", logURL, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	req.Header.Add("Authorization", "token "+cfg.accessToken)
-
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return nil, err
-	}
-
-	return resp.Body, nil
+	return fetchSearchJobFile(client, logURL)
 }
 
 func outputLogs(logs io.Reader, outputPath string) error {
@@ -88,7 +75,7 @@ func init() {
 			return fmt.Errorf("no job found with ID %s", jobID)
 		}
 
-		logsData, err := fetchJobLogs(jobID, job.LogURL)
+		logsData, err := fetchJobLogs(client, jobID, job.LogURL)
 		if err != nil {
 			return err
 		}

cmd/src/search_jobs_results.go

@@ -4,32 +4,19 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"net/http"
 	"os"
 
 	"github.com/sourcegraph/src-cli/internal/api"
 	"github.com/sourcegraph/src-cli/internal/cmderrors"
 )
 
 // fetchJobResults retrieves results for a search job from its results URL
-func fetchJobResults(jobID string, resultsURL string) (io.ReadCloser, error) {
+func fetchJobResults(client api.Client, jobID string, resultsURL string) (io.ReadCloser, error) {
 	if resultsURL == "" {
 		return nil, fmt.Errorf("no results URL found for search job %s", jobID)
 	}
 
-	req, err := http.NewRequest("GET", resultsURL, nil)
-	if err != nil {
-		return nil, err
-	}
-
-	req.Header.Add("Authorization", "token "+cfg.accessToken)
-
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		return nil, err
-	}
-
-	return resp.Body, nil
+	return fetchSearchJobFile(client, resultsURL)
 }
 
 // outputResults writes results to either a file or stdout
@@ -90,7 +77,7 @@ func init() {
 			return fmt.Errorf("no job found with ID %s", jobID)
 		}
 
-		resultsData, err := fetchJobResults(jobID, job.URL)
+		resultsData, err := fetchJobResults(client, jobID, job.URL)
 		if err != nil {
 			return err
 		}

cmd/src/search_jobs_test.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+
+	"github.com/sourcegraph/src-cli/internal/api"
+)
+
+func TestFetchSearchJobFile(t *testing.T) {
+	var gotAuth string
+	mux := http.NewServeMux()
+	mux.HandleFunc("/ok", func(w http.ResponseWriter, r *http.Request) {
+		gotAuth = r.Header.Get("Authorization")
+		_, _ = w.Write([]byte("results data"))
+	})
+	mux.HandleFunc("/fail", func(w http.ResponseWriter, r *http.Request) {
+		http.Error(w, "something went wrong", http.StatusInternalServerError)
+	})
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	endpointURL, err := url.Parse(server.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg = &config{
+		endpointURL: endpointURL,
+		accessToken: "test-token",
+	}
+	defer func() { cfg = nil }()
+
+	client := api.NewClient(api.ClientOpts{EndpointURL: endpointURL, Out: io.Discard})
+
+	t.Run("success", func(t *testing.T) {
+		body, err := fetchSearchJobFile(client, server.URL+"/ok")
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer body.Close()
+
+		data, err := io.ReadAll(body)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if string(data) != "results data" {
+			t.Fatalf("got body %q, want %q", data, "results data")
+		}
+		if gotAuth != "token test-token" {
+			t.Fatalf("got Authorization header %q, want %q", gotAuth, "token test-token")
+		}
+	})
+
+	t.Run("non-200 response", func(t *testing.T) {
+		_, err := fetchSearchJobFile(client, server.URL+"/fail")
+		if err == nil {
+			t.Fatal("expected an error for a non-200 response, got nil")
+		}
+		if !strings.Contains(err.Error(), "something went wrong") {
+			t.Fatalf("error %q does not contain the response body", err)
+		}
+	})
+}

internal/api/api.go

@@ -12,6 +12,7 @@ import (
 	"net/url"
 	"os"
 	"runtime"
+	"time"
 
 	ioaux "github.com/jig/teereadcloser"
 	"github.com/kballard/go-shellquote"
@@ -95,10 +96,36 @@ type ClientOpts struct {
 // ErrCIAccessTokenRequired indicates SRC_ACCESS_TOKEN must be set when CI=true.
 var ErrCIAccessTokenRequired = errors.New("SRC_ACCESS_TOKEN must be set when CI=true")
 
+// defaultResponseHeaderTimeout bounds how long we wait for a server to start
+// responding. It is deliberately generous because some GraphQL queries take a
+// long time server-side before the first response byte is written.
+const defaultResponseHeaderTimeout = 1 * time.Minute
+
+// responseHeaderTimeout returns the timeout to wait for a server's response
+// headers, honoring the SRC_RESPONSE_HEADER_TIMEOUT environment variable (a Go
+// duration string; "0" disables the timeout).
+func responseHeaderTimeout() time.Duration {
+	if v := os.Getenv("SRC_RESPONSE_HEADER_TIMEOUT"); v != "" {
+		if d, err := time.ParseDuration(v); err == nil && d >= 0 {
+			return d
+		}
+	}
+	return defaultResponseHeaderTimeout
+}
+
+// BaseTransport returns a clone of http.DefaultTransport (which carries dial
+// and TLS handshake timeouts) with a response header timeout applied, so that
+// an unresponsive server cannot stall requests indefinitely.
+func BaseTransport() *http.Transport {
+	tp := http.DefaultTransport.(*http.Transport).Clone()
+	tp.ResponseHeaderTimeout = responseHeaderTimeout()
+	return tp
+}
+
 func buildTransport(opts ClientOpts, flags *Flags) http.RoundTripper {
 	var transport http.RoundTripper
 	{
-		tp := http.DefaultTransport.(*http.Transport).Clone()
+		tp := BaseTransport()
 
 		if flags.insecureSkipVerify != nil && *flags.insecureSkipVerify {
 			tp.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
@@ -138,6 +165,10 @@ func NewClient(opts ClientOpts) Client {
 
 	transport := buildTransport(opts, flags)
 
+	// Note: no Client.Timeout is set on purpose. It would cap the entire
+	// request including reading the response body, but downloads (search job
+	// results, batch change archives, ...) may legitimately stream for a very
+	// long time. The transport's connection-phase timeouts bound the rest.
 	httpClient := &http.Client{
 		Transport:     transport,
 		CheckRedirect: checkRedirect,

internal/api/timeout_test.go

@@ -0,0 +1,107 @@
+package api
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+)
+
+func newTestClient(t *testing.T, serverURL string) Client {
+	t.Helper()
+	endpointURL, err := url.Parse(serverURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return NewClient(ClientOpts{EndpointURL: endpointURL, Out: io.Discard})
+}
+
+func TestUnresponsiveServerTimesOut(t *testing.T) {
+	t.Setenv("SRC_RESPONSE_HEADER_TIMEOUT", "100ms")
+
+	block := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-block
+	}))
+	defer server.Close()
+	defer close(block)
+
+	client := newTestClient(t, server.URL)
+	req, err := client.NewHTTPRequest(context.Background(), http.MethodGet, "", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	start := time.Now()
+	resp, err := client.Do(req)
+	if err == nil {
+		resp.Body.Close()
+		t.Fatal("expected a timeout error, got nil")
+	}
+	if elapsed := time.Since(start); elapsed > 5*time.Second {
+		t.Fatalf("request took %s, expected it to fail within the response header timeout", elapsed)
+	}
+}
+
+func TestSlowStreamingDownloadSucceeds(t *testing.T) {
+	t.Setenv("SRC_RESPONSE_HEADER_TIMEOUT", "200ms")
+
+	const chunks = 5
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		flusher := w.(http.Flusher)
+		w.WriteHeader(http.StatusOK)
+		flusher.Flush()
+		// Stream the body for much longer than the response header timeout.
+		for range chunks {
+			time.Sleep(100 * time.Millisecond)
+			_, _ = w.Write([]byte("chunk"))
+			flusher.Flush()
+		}
+	}))
+	defer server.Close()
+
+	client := newTestClient(t, server.URL)
+	req, err := client.NewHTTPRequest(context.Background(), http.MethodGet, "", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading a slow streaming body failed: %v", err)
+	}
+	if want := len("chunk") * chunks; len(body) != want {
+		t.Fatalf("got %d body bytes, want %d", len(body), want)
+	}
+}
+
+func TestResponseHeaderTimeoutEnv(t *testing.T) {
+	tests := []struct {
+		value string
+		want  time.Duration
+	}{
+		{value: "", want: defaultResponseHeaderTimeout},
+		{value: "30s", want: 30 * time.Second},
+		{value: "0", want: 0},
+		{value: "garbage", want: defaultResponseHeaderTimeout},
+		{value: "-5s", want: defaultResponseHeaderTimeout},
+	}
+
+	for _, test := range tests {
+		t.Run(test.value, func(t *testing.T) {
+			t.Setenv("SRC_RESPONSE_HEADER_TIMEOUT", test.value)
+			if got := responseHeaderTimeout(); got != test.want {
+				t.Fatalf("got %s, want %s", got, test.want)
+			}
+		})
+	}
+}