Release 2024.3.0

Before this commit the commands that listed tunnels and tunnel routes would be limited to 1000 results by the server.

Now, the commands will call the endpoints until the result set is exhausted. This can take a long time if there are
thousands of pages available, since each request is executed synchronously.
From a user's perspective, nothing changes.

.github/workflows/check.yaml

@@ -9,10 +9,10 @@ jobs:
     runs-on: ${{ matrix.os }}
     steps:
     - name: Install Go
-      uses: actions/setup-go@v3
+      uses: actions/setup-go@v5
       with:
         go-version: ${{ matrix.go-version }}
     - name: Checkout code
-      uses: actions/checkout@v3
+      uses: actions/checkout@v4
     - name: Test
       run: make test

RELEASE_NOTES

@@ -1,3 +1,15 @@
+2024.3.0
+- 2024-03-14 TUN-8281: Run cloudflared query list tunnels/routes endpoint in a paginated way
+- 2024-03-13 TUN-8297: Improve write timeout logging on safe_stream.go
+- 2024-03-07 TUN-8290: Remove `|| true` from postrm.sh
+- 2024-03-05 TUN-8275: Skip write timeout log on "no network activity"
+- 2024-01-23 Update postrm.sh to fix incomplete uninstall
+- 2024-01-05 fix typo in errcheck for response parsing logic in CreateTunnel routine
+- 2023-12-23 Update linux_service.go
+- 2023-12-07 ci: bump actions/checkout to v4
+- 2023-12-07 ci/check: bump actions/setup-go to v5
+- 2023-04-28 check.yaml: bump actions/setup-go to v4
+
 2024.2.1
 - 2024-02-20 TUN-8242: Update Changes.md file with new remote diagnostics behaviour
 - 2024-02-19 TUN-8238: Fix type mismatch introduced by fast-forward

cfapi/base_client.go

@@ -109,20 +109,34 @@ func (r *RESTClient) sendRequest(method string, url url.URL, body interface{}) (
 	return r.client.Do(req)
 }
 
-func parseResponse(reader io.Reader, data interface{}) error {
+func parseResponseEnvelope(reader io.Reader) (*response, error) {
 	// Schema for Tunnelstore responses in the v1 API.
 	// Roughly, it's a wrapper around a particular result that adds failures/errors/etc
 	var result response
 	// First, parse the wrapper and check the API call succeeded
 	if err := json.NewDecoder(reader).Decode(&result); err != nil {
-		return errors.Wrap(err, "failed to decode response")
+		return nil, errors.Wrap(err, "failed to decode response")
 	}
 	if err := result.checkErrors(); err != nil {
-		return err
+		return nil, err
 	}
 	if !result.Success {
-		return ErrAPINoSuccess
+		return nil, ErrAPINoSuccess
 	}
+
+	return &result, nil
+}
+
+func parseResponse(reader io.Reader, data interface{}) error {
+	result, err := parseResponseEnvelope(reader)
+	if err != nil {
+		return err
+	}
+
+	return parseResponseBody(result, data)
+}
+
+func parseResponseBody(result *response, data interface{}) error {
 	// At this point we know the API call succeeded, so, parse out the inner
 	// result into the datatype provided as a parameter.
 	if err := json.Unmarshal(result.Result, &data); err != nil {
@@ -131,11 +145,58 @@ func parseResponse(reader io.Reader, data interface{}) error {
 	return nil
 }
 
+func fetchExhaustively[T any](requestFn func(int) (*http.Response, error)) ([]*T, error) {
+	page := 0
+	var fullResponse []*T
+
+	for {
+		page += 1
+		envelope, parsedBody, err := fetchPage[T](requestFn, page)
+
+		if err != nil {
+			return nil, errors.Wrap(err, fmt.Sprintf("Error Parsing page %d", page))
+		}
+
+		fullResponse = append(fullResponse, parsedBody...)
+		if envelope.Pagination.Count < envelope.Pagination.PerPage || len(fullResponse) >= envelope.Pagination.TotalCount {
+			break
+		}
+
+	}
+	return fullResponse, nil
+}
+
+func fetchPage[T any](requestFn func(int) (*http.Response, error), page int) (*response, []*T, error) {
+	pageResp, err := requestFn(page)
+	if err != nil {
+		return nil, nil, errors.Wrap(err, "REST request failed")
+	}
+	defer pageResp.Body.Close()
+	if pageResp.StatusCode == http.StatusOK {
+		envelope, err := parseResponseEnvelope(pageResp.Body)
+		if err != nil {
+			return nil, nil, err
+		}
+		var parsedRspBody []*T
+		return envelope, parsedRspBody, parseResponseBody(envelope, &parsedRspBody)
+
+	}
+	return nil, nil, errors.New(fmt.Sprintf("Failed to fetch page. Server returned: %d", pageResp.StatusCode))
+}
+
 type response struct {
-	Success  bool            `json:"success,omitempty"`
-	Errors   []apiErr        `json:"errors,omitempty"`
-	Messages []string        `json:"messages,omitempty"`
-	Result   json.RawMessage `json:"result,omitempty"`
+	Success    bool            `json:"success,omitempty"`
+	Errors     []apiErr        `json:"errors,omitempty"`
+	Messages   []string        `json:"messages,omitempty"`
+	Result     json.RawMessage `json:"result,omitempty"`
+	Pagination Pagination      `json:"result_info,omitempty"`
+}
+
+type Pagination struct {
+	Count      int `json:"count,omitempty"`
+	Page       int `json:"page,omitempty"`
+	PerPage    int `json:"per_page,omitempty"`
+	TotalCount int `json:"total_count,omitempty"`
 }
 
 func (r *response) checkErrors() error {

cfapi/ip_route.go

@@ -137,20 +137,24 @@ type GetRouteByIpParams struct {
 }
 
 // ListRoutes calls the Tunnelstore GET endpoint for all routes under an account.
+// Due to pagination on the server side it will call the endpoint multiple times if needed.
 func (r *RESTClient) ListRoutes(filter *IpRouteFilter) ([]*DetailedRoute, error) {
-	endpoint := r.baseEndpoints.accountRoutes
-	endpoint.RawQuery = filter.Encode()
-	resp, err := r.sendRequest("GET", endpoint, nil)
-	if err != nil {
-		return nil, errors.Wrap(err, "REST request failed")
-	}
-	defer resp.Body.Close()
+	fetchFn := func(page int) (*http.Response, error) {
+		endpoint := r.baseEndpoints.accountRoutes
+		filter.Page(page)
+		endpoint.RawQuery = filter.Encode()
+		rsp, err := r.sendRequest("GET", endpoint, nil)
 
-	if resp.StatusCode == http.StatusOK {
-		return parseListDetailedRoutes(resp.Body)
+		if err != nil {
+			return nil, errors.Wrap(err, "REST request failed")
+		}
+		if rsp.StatusCode != http.StatusOK {
+			rsp.Body.Close()
+			return nil, r.statusCodeToError("list routes", rsp)
+		}
+		return rsp, nil
 	}
-
-	return nil, r.statusCodeToError("list routes", resp)
+	return fetchExhaustively[DetailedRoute](fetchFn)
 }
 
 // AddRoute calls the Tunnelstore POST endpoint for a given route.
@@ -208,12 +212,6 @@ func (r *RESTClient) GetByIP(params GetRouteByIpParams) (DetailedRoute, error) {
 	return DetailedRoute{}, r.statusCodeToError("get route by IP", resp)
 }
 
-func parseListDetailedRoutes(body io.ReadCloser) ([]*DetailedRoute, error) {
-	var routes []*DetailedRoute
-	err := parseResponse(body, &routes)
-	return routes, err
-}
-
 func parseRoute(body io.ReadCloser) (Route, error) {
 	var route Route
 	err := parseResponse(body, &route)

cfapi/ip_route_filter.go

@@ -167,6 +167,10 @@ func (f *IpRouteFilter) MaxFetchSize(max uint) {
 	f.queryParams.Set("per_page", strconv.Itoa(int(max)))
 }
 
+func (f *IpRouteFilter) Page(page int) {
+	f.queryParams.Set("page", strconv.Itoa(page))
+}
+
 func (f IpRouteFilter) Encode() string {
 	return f.queryParams.Encode()
 }

cfapi/tunnel.go

@@ -93,7 +93,7 @@ func (r *RESTClient) CreateTunnel(name string, tunnelSecret []byte) (*TunnelWith
 	switch resp.StatusCode {
 	case http.StatusOK:
 		var tunnel TunnelWithToken
-		if serdeErr := parseResponse(resp.Body, &tunnel); err != nil {
+		if serdeErr := parseResponse(resp.Body, &tunnel); serdeErr != nil {
 			return nil, serdeErr
 		}
 		return &tunnel, nil
@@ -177,25 +177,22 @@ func (r *RESTClient) DeleteTunnel(tunnelID uuid.UUID, cascade bool) error {
 }
 
 func (r *RESTClient) ListTunnels(filter *TunnelFilter) ([]*Tunnel, error) {
-	endpoint := r.baseEndpoints.accountLevel
-	endpoint.RawQuery = filter.encode()
-	resp, err := r.sendRequest("GET", endpoint, nil)
-	if err != nil {
-		return nil, errors.Wrap(err, "REST request failed")
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode == http.StatusOK {
-		return parseListTunnels(resp.Body)
+	fetchFn := func(page int) (*http.Response, error) {
+		endpoint := r.baseEndpoints.accountLevel
+		filter.Page(page)
+		endpoint.RawQuery = filter.encode()
+		rsp, err := r.sendRequest("GET", endpoint, nil)
+		if err != nil {
+			return nil, errors.Wrap(err, "REST request failed")
+		}
+		if rsp.StatusCode != http.StatusOK {
+			rsp.Body.Close()
+			return nil, r.statusCodeToError("list tunnels", rsp)
+		}
+		return rsp, nil
 	}
 
-	return nil, r.statusCodeToError("list tunnels", resp)
-}
-
-func parseListTunnels(body io.ReadCloser) ([]*Tunnel, error) {
-	var tunnels []*Tunnel
-	err := parseResponse(body, &tunnels)
-	return tunnels, err
+	return fetchExhaustively[Tunnel](fetchFn)
 }
 
 func (r *RESTClient) ListActiveClients(tunnelID uuid.UUID) ([]*ActiveClient, error) {

cfapi/tunnel_filter.go

@@ -50,6 +50,10 @@ func (f *TunnelFilter) MaxFetchSize(max uint) {
 	f.queryParams.Set("per_page", strconv.Itoa(int(max)))
 }
 
+func (f *TunnelFilter) Page(page int) {
+	f.queryParams.Set("page", strconv.Itoa(page))
+}
+
 func (f TunnelFilter) encode() string {
 	return f.queryParams.Encode()
 }

cfapi/tunnel_test.go

@@ -3,7 +3,6 @@ package cfapi
 import (
 	"bytes"
 	"fmt"
-	"io"
 	"net"
 	"reflect"
 	"strings"
@@ -16,52 +15,6 @@ import (
 
 var loc, _ = time.LoadLocation("UTC")
 
-func Test_parseListTunnels(t *testing.T) {
-	type args struct {
-		body string
-	}
-	tests := []struct {
-		name    string
-		args    args
-		want    []*Tunnel
-		wantErr bool
-	}{
-		{
-			name: "empty list",
-			args: args{body: `{"success": true, "result": []}`},
-			want: []*Tunnel{},
-		},
-		{
-			name:    "success is false",
-			args:    args{body: `{"success": false, "result": []}`},
-			wantErr: true,
-		},
-		{
-			name:    "errors are present",
-			args:    args{body: `{"errors": [{"code": 1003, "message":"An A, AAAA or CNAME record already exists with that host"}], "result": []}`},
-			wantErr: true,
-		},
-		{
-			name:    "invalid response",
-			args:    args{body: `abc`},
-			wantErr: true,
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			body := io.NopCloser(bytes.NewReader([]byte(tt.args.body)))
-			got, err := parseListTunnels(body)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("parseListTunnels() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("parseListTunnels() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
 func Test_unmarshalTunnel(t *testing.T) {
 	type args struct {
 		body string

cmd/cloudflared/linux_service.go

@@ -55,7 +55,8 @@ var systemdAllTemplates = map[string]ServiceTemplate{
 		Path: fmt.Sprintf("/etc/systemd/system/%s", cloudflaredService),
 		Content: `[Unit]
 Description=cloudflared
-After=network.target
+After=network-online.target
+Wants=network-online.target
 
 [Service]
 TimeoutStartSec=0
@@ -72,7 +73,8 @@ WantedBy=multi-user.target
 		Path: fmt.Sprintf("/etc/systemd/system/%s", cloudflaredUpdateService),
 		Content: `[Unit]
 Description=Update cloudflared
-After=network.target
+After=network-online.target
+Wants=network-online.target
 
 [Service]
 ExecStart=/bin/bash -c '{{ .Path }} update; code=$?; if [ $code -eq 11 ]; then systemctl restart cloudflared; exit 0; fi; exit $code'

postrm.sh

@@ -1,4 +1,4 @@
 #!/bin/bash
 set -eu
-rm /usr/local/bin/cloudflared
-rm /usr/local/etc/cloudflared/.installedFromPackageManager || true
+rm -f /usr/local/bin/cloudflared
+rm -f /usr/local/etc/cloudflared/.installedFromPackageManager

quic/safe_stream.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"net"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/quic-go/quic-go"
@@ -11,11 +12,15 @@ import (
 	"github.com/rs/zerolog/log"
 )
 
+// The error that is throw by the writer when there is `no network activity`.
+var idleTimeoutError = quic.IdleTimeoutError{}
+
 type SafeStreamCloser struct {
 	lock         sync.Mutex
 	stream       quic.Stream
 	writeTimeout time.Duration
 	log          *zerolog.Logger
+	closing      atomic.Bool
 }
 
 func NewSafeStreamCloser(stream quic.Stream, writeTimeout time.Duration, log *zerolog.Logger) *SafeStreamCloser {
@@ -41,24 +46,35 @@ func (s *SafeStreamCloser) Write(p []byte) (n int, err error) {
 	}
 	nBytes, err := s.stream.Write(p)
 	if err != nil {
-		s.handleTimeout(err)
+		s.handleWriteError(err)
 	}
 
 	return nBytes, err
 }
 
 // Handles the timeout error in case it happened, by canceling the stream write.
-func (s *SafeStreamCloser) handleTimeout(err error) {
+func (s *SafeStreamCloser) handleWriteError(err error) {
+	// If we are closing the stream we just ignore any write error.
+	if s.closing.Load() {
+		return
+	}
 	var netErr net.Error
 	if errors.As(err, &netErr) {
 		if netErr.Timeout() {
-			s.log.Error().Err(netErr).Msg("Closing quic stream due to timeout while writing")
+			// We don't need to log if what cause the timeout was no network activity.
+			if !errors.Is(netErr, &idleTimeoutError) {
+				s.log.Error().Err(netErr).Msg("Closing quic stream due to timeout while writing")
+			}
+			// We need to explicitly cancel the write so that it frees all buffers.
 			s.stream.CancelWrite(0)
 		}
 	}
 }
 
 func (s *SafeStreamCloser) Close() error {
+	// Set this stream to a closing state.
+	s.closing.Store(true)
+
 	// Make sure a possible writer does not block the lock forever. We need it, so we can close the writer
 	// side of the stream safely.
 	_ = s.stream.SetWriteDeadline(time.Now())