From a6be166de9c055b39d88876c800abafe871ed37d Mon Sep 17 00:00:00 2001 From: Ilya Brin <464157+ilyabrin@users.noreply.github.com> Date: Sat, 19 Sep 2026 12:25:36 +0300 Subject: [PATCH 1/2] fix: stop the client timeout from aborting uploads and downloads Config.DefaultTimeout lands in http.Client.Timeout, which is an absolute deadline covering the whole request including the body. That is right for the small JSON calls the API is mostly made of, but it also applied to streamed transfers: with the 30s default, every upload or download slower than 30 seconds was aborted mid-flight, no matter how generous a deadline the caller had put on ctx. uploadFileMultipart already worked around this with a private client that carries no Timeout. Lift that into Client.transferClient and a transferContext helper, and use both from uploadFileSingle and DownloadFileToPath, so every path that streams a body is bounded by the context instead of the wall clock. Callers who set their own deadline keep it; those who do not fall back to 30 minutes, as chunked uploads already did. --- client.go | 42 ++++++++- download.go | 13 ++- transfer_timeout_test.go | 181 +++++++++++++++++++++++++++++++++++++++ upload.go | 38 ++++---- 4 files changed, 247 insertions(+), 27 deletions(-) create mode 100644 transfer_timeout_test.go diff --git a/client.go b/client.go index b8b7b92..bb8b456 100644 --- a/client.go +++ b/client.go @@ -352,6 +352,8 @@ func (c *Client) handleResponse(resp *http.Response, expectedCodes []int) (*Erro } } + errorResponse.StatusCode = resp.StatusCode + return &errorResponse, fmt.Errorf("request failed with status %d: %s", resp.StatusCode, errorResponse.Error) } @@ -374,8 +376,12 @@ func requestJSON[T any](ctx context.Context, c *Client, method HttpMethod, endpo if !slices.Contains(okCodes, resp.StatusCode) { var errorResponse ErrorResponse if err := json.NewDecoder(resp.Body).Decode(&errorResponse); err != nil { - return nil, &ErrorResponse{Error: fmt.Sprintf("failed to decode error response: %v", err)} + return nil, &ErrorResponse{ + Error: fmt.Sprintf("failed to decode error response: %v", err), + StatusCode: resp.StatusCode, + } } + errorResponse.StatusCode = resp.StatusCode return nil, &errorResponse } @@ -406,3 +412,37 @@ func (c *Client) safeDecodeJSON(resp *http.Response, target interface{}) error { return nil } + +// transferTimeout bounds a bulk transfer when the caller did not set a +// deadline of their own. +const transferTimeout = 30 * time.Minute + +// transferClient returns an HTTP client suited to streaming a file body. +// +// c.HTTPClient carries Config.DefaultTimeout, which is an absolute deadline +// covering the whole request including the body. That is right for the small +// JSON calls the API is mostly made of, but it silently kills any upload or +// download that outlives it — a 30s default aborts every transfer slower than +// 30 seconds, no matter how generous a deadline the caller put on ctx. +// +// The returned client shares this client's transport (so connection pooling +// and TLS settings are preserved) but carries no Timeout of its own: the +// transfer is bounded by the request context instead. Mutating +// c.HTTPClient.Timeout in place would be a data race with concurrent callers. +func (c *Client) transferClient() *http.Client { + return &http.Client{ + Transport: c.HTTPClient.Transport, + CheckRedirect: c.HTTPClient.CheckRedirect, + Jar: c.HTTPClient.Jar, + } +} + +// transferContext returns ctx bounded by transferTimeout when the caller did +// not set a deadline of their own, together with its cancel func. The cancel +// func is always non-nil and safe to defer. +func transferContext(ctx context.Context) (context.Context, context.CancelFunc) { + if _, hasDeadline := ctx.Deadline(); hasDeadline { + return ctx, func() {} + } + return context.WithTimeout(ctx, transferTimeout) +} diff --git a/download.go b/download.go index b00e6b3..62facb4 100644 --- a/download.go +++ b/download.go @@ -59,13 +59,20 @@ func (c *Client) DownloadFileToPath(ctx context.Context, remotePath string, loca c.Logger.Debug("Received download link: %s", link.Href) - // Step 2: Execute the download request - req, err := http.NewRequestWithContext(ctx, http.MethodGet, link.Href, nil) + // Step 2: Execute the download request. + // + // The body is streamed, so this must not run on c.HTTPClient: its Timeout + // is an absolute deadline over the whole request and would abort any + // download slower than Config.DefaultTimeout. Bound it by the context. + dlCtx, cancel := transferContext(ctx) + defer cancel() + + req, err := http.NewRequestWithContext(dlCtx, http.MethodGet, link.Href, nil) if err != nil { return fmt.Errorf("failed to create download request: %w", err) } - resp, err := c.HTTPClient.Do(req) + resp, err := c.transferClient().Do(req) if err != nil { c.Logger.LogError("file download", err) return fmt.Errorf("download request failed: %w", err) diff --git a/transfer_timeout_test.go b/transfer_timeout_test.go new file mode 100644 index 0000000..0c4b93a --- /dev/null +++ b/transfer_timeout_test.go @@ -0,0 +1,181 @@ +package disk + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" +) + +// slowBody streams a small body in chunks, taking `d` in total. +func slowBody(d time.Duration) http.HandlerFunc { + const chunks = 5 + return func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", "50") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + for i := 0; i < chunks; i++ { + _, _ = w.Write([]byte("0123456789")) + if flusher != nil { + flusher.Flush() + } + time.Sleep(d / chunks) + } + } +} + +// newSlowClient wires a client whose DefaultTimeout is far shorter than the +// transfer takes, which is the shape of the real bug: the 30s default aborts +// any transfer slower than 30 seconds regardless of the caller's context. +func newSlowClient(t *testing.T, apiURL string) *Client { + t.Helper() + cfg := DefaultClientConfig() + cfg.BaseURL = apiURL + "/" + cfg.DefaultTimeout = 150 * time.Millisecond + cfg.MaxRetries = 0 + client, err := NewWithConfig(cfg, "test-token") + if err != nil { + t.Fatal(err) + } + return client +} + +func TestDownloadOutlivesDefaultTimeout(t *testing.T) { + files := httptest.NewServer(slowBody(750 * time.Millisecond)) + defer files.Close() + + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(Link{Href: files.URL, Method: http.MethodGet}) + })) + defer api.Close() + + client := newSlowClient(t, api.URL) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + dst := filepath.Join(t.TempDir(), "out.bin") + if err := client.DownloadFileToPath(ctx, "/big.bin", dst, &DownloadOptions{Overwrite: true}); err != nil { + t.Fatalf("download aborted by the client timeout instead of honouring ctx: %v", err) + } + + got, err := os.ReadFile(dst) // #nosec G304 -- path built by the test + if err != nil { + t.Fatal(err) + } + if len(got) != 50 { + t.Errorf("expected the full 50-byte body, got %d bytes", len(got)) + } +} + +func TestUploadOutlivesDefaultTimeout(t *testing.T) { + var uploaded atomic.Int64 + + upload := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Drain slowly so the request outlives DefaultTimeout. + buf := make([]byte, 16) + for { + n, err := r.Body.Read(buf) + uploaded.Add(int64(n)) + if err != nil { + break + } + time.Sleep(50 * time.Millisecond) + } + w.WriteHeader(http.StatusCreated) + })) + defer upload.Close() + + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "resources") && r.Method == http.MethodGet { + _ = json.NewEncoder(w).Encode(ResourceUploadLink{Href: upload.URL, Method: http.MethodPut}) + return + } + _ = json.NewEncoder(w).Encode(Resource{Name: "big.bin", Path: "disk:/big.bin"}) + })) + defer api.Close() + + client := newSlowClient(t, api.URL) + + src := filepath.Join(t.TempDir(), "big.bin") + payload := strings.Repeat("x", 256) + if err := os.WriteFile(src, []byte(payload), 0o600); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + if _, err := client.UploadFileFromPath(ctx, src, "/big.bin", &UploadOptions{Overwrite: true}); err != nil { + t.Fatalf("upload aborted by the client timeout instead of honouring ctx: %v", err) + } + if got := uploaded.Load(); got != int64(len(payload)) { + t.Errorf("expected the server to receive %d bytes, got %d", len(payload), got) + } +} + +// A caller's own deadline must still win over the transfer fallback. +func TestTransferHonoursCallerDeadline(t *testing.T) { + files := httptest.NewServer(slowBody(2 * time.Second)) + defer files.Close() + + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(Link{Href: files.URL, Method: http.MethodGet}) + })) + defer api.Close() + + client := newSlowClient(t, api.URL) + + ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond) + defer cancel() + + dst := filepath.Join(t.TempDir(), "out.bin") + err := client.DownloadFileToPath(ctx, "/big.bin", dst, &DownloadOptions{Overwrite: true}) + if err == nil { + t.Fatal("expected the caller's short deadline to abort the download") + } +} + +func TestTransferClientHasNoTimeout(t *testing.T) { + client, err := New("token") + if err != nil { + t.Fatal(err) + } + if client.HTTPClient.Timeout == 0 { + t.Fatal("precondition: the default client is expected to carry a timeout") + } + if got := client.transferClient().Timeout; got != 0 { + t.Errorf("transfer client must not carry a wall-clock timeout, got %v", got) + } + if client.transferClient().Transport != client.HTTPClient.Transport { + t.Error("transfer client should reuse the shared transport") + } +} + +func TestTransferContext(t *testing.T) { + // No caller deadline: the fallback applies. + ctx, cancel := transferContext(context.Background()) + defer cancel() + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("expected a fallback deadline") + } + if remaining := time.Until(deadline); remaining < 29*time.Minute { + t.Errorf("fallback deadline too short: %v", remaining) + } + + // Caller deadline: left untouched. + own, cancelOwn := context.WithTimeout(context.Background(), time.Minute) + defer cancelOwn() + got, cancelGot := transferContext(own) + defer cancelGot() + if got != own { + t.Error("expected the caller's context to be returned unchanged") + } +} diff --git a/upload.go b/upload.go index 71a8f3b..203ce49 100644 --- a/upload.go +++ b/upload.go @@ -10,13 +10,8 @@ import ( "os" "path/filepath" "strings" - "time" ) -// multipartUploadTimeout bounds a chunked upload when the caller did not set a -// deadline of their own. -const multipartUploadTimeout = 30 * time.Minute - // UploadProgress represents the progress of an upload operation type UploadProgress struct { BytesUploaded int64 @@ -134,8 +129,15 @@ func (c *Client) uploadFileSingle(ctx context.Context, localPath string, remoteP } } - // Step 4: Create the HTTP request for file upload - req, err := http.NewRequestWithContext(ctx, uploadLink.Method, uploadLink.Href, reader) + // Step 4: Create the HTTP request for file upload. + // + // The body is streamed, so this must not run on c.HTTPClient: its Timeout + // is an absolute deadline over the whole request and would abort any + // upload slower than Config.DefaultTimeout. Bound it by the context. + upCtx, cancel := transferContext(ctx) + defer cancel() + + req, err := http.NewRequestWithContext(upCtx, uploadLink.Method, uploadLink.Href, reader) if err != nil { return nil, fmt.Errorf("failed to create upload request: %w", err) } @@ -155,8 +157,8 @@ func (c *Client) uploadFileSingle(ctx context.Context, localPath string, remoteP c.Logger.Debug("Uploading file with content type: %s", contentType) - // Step 5: Execute the upload using the configured HTTP client - resp, err := c.HTTPClient.Do(req) + // Step 5: Execute the upload using a transfer-scoped HTTP client + resp, err := c.transferClient().Do(req) if err != nil { c.Logger.LogError("file upload", err) return nil, fmt.Errorf("upload request failed: %w", err) @@ -248,21 +250,11 @@ func (c *Client) uploadFileMultipart(ctx context.Context, localPath string, remo c.Logger.Debug("Starting multipart upload with content type: %s", contentType) // Large uploads need far more headroom than the default per-request timeout. - // Use a client that shares this client's transport but is bounded by the - // request context instead, so concurrent callers are not affected — mutating - // c.HTTPClient.Timeout here would be a data race. - uploadClient := &http.Client{ - Transport: c.HTTPClient.Transport, - CheckRedirect: c.HTTPClient.CheckRedirect, - Jar: c.HTTPClient.Jar, - } - if _, hasDeadline := ctx.Deadline(); !hasDeadline { - uploadCtx, cancel := context.WithTimeout(ctx, multipartUploadTimeout) - defer cancel() - req = req.WithContext(uploadCtx) - } + transferCtx, cancel := transferContext(ctx) + defer cancel() + req = req.WithContext(transferCtx) - resp, err := uploadClient.Do(req) + resp, err := c.transferClient().Do(req) if err != nil { c.Logger.LogError("multipart file upload", err) return nil, fmt.Errorf("multipart upload request failed: %w", err) From 4de704373d56d7e4ef16cf0106711b9cec3832c1 Mon Sep 17 00:00:00 2001 From: Ilya Brin <464157+ilyabrin@users.noreply.github.com> Date: Sat, 19 Sep 2026 12:25:36 +0300 Subject: [PATCH 2/2] feat: add CreateDirAll for nested directories The API creates one directory level per request, so CreateDir could not create "newDir/subDir/anotherDir", which was the long-standing todo on it. Add CreateDirAll, which walks the path from the shallowest segment to the deepest and issues one request per missing level, mirroring os.MkdirAll: existing directories are left alone rather than reported as errors. Telling "already exists" apart from a real conflict needs the HTTP status, which ErrorResponse did not carry, so record it in a new StatusCode field (json:"-", filled in by this package) and expose the check as ErrorResponse.AlreadyExists. CreateDir keeps its single-level semantics, matching os.Mkdir. --- createdirall_test.go | 188 +++++++++++++++++++++++++++++++++++++++++++ resources.go | 63 ++++++++++++++- types.go | 5 ++ 3 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 createdirall_test.go diff --git a/createdirall_test.go b/createdirall_test.go new file mode 100644 index 0000000..08fcfd6 --- /dev/null +++ b/createdirall_test.go @@ -0,0 +1,188 @@ +package disk + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" +) + +// dirServer records every path PUT to /resources and can pretend that some of +// them already exist. +type dirServer struct { + mu sync.Mutex + created []string + existing map[string]bool +} + +func (d *dirServer) handler() http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + path := r.URL.Query().Get("path") + + d.mu.Lock() + exists := d.existing[path] + if !exists { + d.created = append(d.created, path) + } + d.mu.Unlock() + + if exists { + w.WriteHeader(http.StatusConflict) + _ = json.NewEncoder(w).Encode(ErrorResponse{ + Message: "Resource already exists", + Error: "DiskPathPointsToExistentDirectoryError", + Description: "Specified path already exists", + }) + return + } + + w.WriteHeader(http.StatusCreated) + _ = json.NewEncoder(w).Encode(Link{Href: "https://example.invalid" + path, Method: http.MethodGet}) + } +} + +func newDirClient(t *testing.T, srv *httptest.Server) *Client { + t.Helper() + cfg := DefaultClientConfig() + cfg.BaseURL = srv.URL + "/" + cfg.MaxRetries = 0 + client, err := NewWithConfig(cfg, "test-token") + if err != nil { + t.Fatal(err) + } + return client +} + +func TestCreateDirAllCreatesEverySegment(t *testing.T) { + d := &dirServer{existing: map[string]bool{}} + srv := httptest.NewServer(d.handler()) + defer srv.Close() + + client := newDirClient(t, srv) + + if errResp := client.CreateDirAll(context.Background(), "/newDir/subDir/anotherDir"); errResp != nil { + t.Fatalf("unexpected error: %v", errResp.Error) + } + + want := []string{"/newDir", "/newDir/subDir", "/newDir/subDir/anotherDir"} + if len(d.created) != len(want) { + t.Fatalf("expected %d requests, got %v", len(want), d.created) + } + for i, path := range want { + if d.created[i] != path { + t.Errorf("request %d: expected %q, got %q", i, path, d.created[i]) + } + } +} + +func TestCreateDirAllSkipsExistingParents(t *testing.T) { + d := &dirServer{existing: map[string]bool{"/photos": true, "/photos/2026": true}} + srv := httptest.NewServer(d.handler()) + defer srv.Close() + + client := newDirClient(t, srv) + + if errResp := client.CreateDirAll(context.Background(), "/photos/2026/summer"); errResp != nil { + t.Fatalf("existing parents must not be an error, got: %v", errResp.Error) + } + if len(d.created) != 1 || d.created[0] != "/photos/2026/summer" { + t.Errorf("expected only the missing leaf to be created, got %v", d.created) + } +} + +func TestCreateDirAllIsIdempotent(t *testing.T) { + d := &dirServer{existing: map[string]bool{"/a": true, "/a/b": true}} + srv := httptest.NewServer(d.handler()) + defer srv.Close() + + client := newDirClient(t, srv) + + if errResp := client.CreateDirAll(context.Background(), "/a/b"); errResp != nil { + t.Fatalf("creating an existing directory must succeed, got: %v", errResp.Error) + } + if len(d.created) != 0 { + t.Errorf("nothing should have been created, got %v", d.created) + } +} + +func TestCreateDirAllPathForms(t *testing.T) { + cases := []struct { + name string + path string + want []string + }{ + {"absolute", "/a/b", []string{"/a", "/a/b"}}, + {"relative", "a/b", []string{"a", "a/b"}}, + {"disk prefix", "disk:/a/b", []string{"/a", "/a/b"}}, + {"redundant slashes", "/a//b/", []string{"/a", "/a/b"}}, + {"single segment", "/a", []string{"/a"}}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + d := &dirServer{existing: map[string]bool{}} + srv := httptest.NewServer(d.handler()) + defer srv.Close() + + if errResp := newDirClient(t, srv).CreateDirAll(context.Background(), tc.path); errResp != nil { + t.Fatalf("unexpected error: %v", errResp.Error) + } + if len(d.created) != len(tc.want) { + t.Fatalf("expected %v, got %v", tc.want, d.created) + } + for i, path := range tc.want { + if d.created[i] != path { + t.Errorf("request %d: expected %q, got %q", i, path, d.created[i]) + } + } + }) + } +} + +func TestCreateDirAllPropagatesRealErrors(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusForbidden) + _ = json.NewEncoder(w).Encode(ErrorResponse{Error: "DiskResourceUploadFailedError", Message: "no"}) + })) + defer srv.Close() + + errResp := newDirClient(t, srv).CreateDirAll(context.Background(), "/a/b") + if errResp == nil { + t.Fatal("expected the 403 to be propagated") + } + if errResp.StatusCode != http.StatusForbidden { + t.Errorf("expected status 403 to be recorded, got %d", errResp.StatusCode) + } + if errResp.AlreadyExists() { + t.Error("a 403 must not be mistaken for an existing directory") + } +} + +func TestCreateDirAllRejectsBadPaths(t *testing.T) { + client, err := New("token") + if err != nil { + t.Fatal(err) + } + for _, path := range []string{"", "/a/../b", "/", "a\x00b"} { + if errResp := client.CreateDirAll(context.Background(), path); errResp == nil { + t.Errorf("expected %q to be rejected", path) + } + } +} + +func TestAlreadyExists(t *testing.T) { + var nilResp *ErrorResponse + if nilResp.AlreadyExists() { + t.Error("a nil error is not an existing directory") + } + conflict := &ErrorResponse{StatusCode: http.StatusConflict, Error: "DiskPathPointsToExistentDirectoryError"} + if !conflict.AlreadyExists() { + t.Error("expected the Yandex existing-directory conflict to be recognised") + } + other := &ErrorResponse{StatusCode: http.StatusConflict, Error: "DiskPathDoesntExistsError"} + if other.AlreadyExists() { + t.Error("a missing-parent conflict is a real error") + } +} diff --git a/resources.go b/resources.go index 289ecf2..d2dbf20 100644 --- a/resources.go +++ b/resources.go @@ -156,7 +156,9 @@ func (c *Client) UpdateMetadata(ctx context.Context, path string, custom_propert } // CreateDir creates a new directory with the specified 'path' name. -// todo: can't create nested dirs like newDir/subDir/anotherDir +// +// Only the final segment is created: the parent must already exist, mirroring +// os.Mkdir. Use CreateDirAll to create a nested path in one call. func (c *Client) CreateDir(ctx context.Context, path string) (*Link, *ErrorResponse) { if len(path) < 1 { return nil, &ErrorResponse{Error: "path cannot be empty"} @@ -168,6 +170,65 @@ func (c *Client) CreateDir(ctx context.Context, path string) (*Link, *ErrorRespo return requestJSON[Link](ctx, c, PUT, "resources?"+query.Encode(), nil, http.StatusCreated) } +// AlreadyExists reports whether this error means the resource is already +// present, which CreateDirAll treats as success rather than failure. +func (e *ErrorResponse) AlreadyExists() bool { + if e == nil { + return false + } + return e.StatusCode == http.StatusConflict && + e.Error == "DiskPathPointsToExistentDirectoryError" +} + +// CreateDirAll creates path along with any missing parent directories, +// mirroring os.MkdirAll. Directories that already exist are left alone and do +// not produce an error; if path already exists as a directory, CreateDirAll +// does nothing and returns nil. +// +// The Yandex Disk API creates one level per request, so this issues one +// request per missing segment, walking from the shallowest to the deepest. +func (c *Client) CreateDirAll(ctx context.Context, path string) *ErrorResponse { + if len(path) < 1 { + return &ErrorResponse{Error: "path cannot be empty"} + } + if err := validatePath(path); err != nil { + return &ErrorResponse{Error: err.Error()} + } + + // "disk:/a/b" and "/a/b" and "a/b" all address the same place; normalise + // to the segments so the prefixes below rebuild a well-formed path. + trimmed := strings.TrimPrefix(path, "disk:") + absolute := strings.HasPrefix(trimmed, "/") + + var segments []string + for _, segment := range strings.Split(trimmed, "/") { + if segment != "" && segment != "." { + segments = append(segments, segment) + } + } + if len(segments) == 0 { + return &ErrorResponse{Error: "path contains no directory names"} + } + + prefix := "" + if absolute { + prefix = "/" + } + + for i, segment := range segments { + if i > 0 { + prefix += "/" + } + prefix += segment + + if _, errResp := c.CreateDir(ctx, prefix); errResp != nil && !errResp.AlreadyExists() { + return errResp + } + } + + return nil +} + func (c *Client) CopyResource(ctx context.Context, from, path string) (*Link, *ErrorResponse) { if len(from) < 1 || len(path) < 1 { return nil, &ErrorResponse{Error: "from and path cannot be empty"} diff --git a/types.go b/types.go index 7b24ef0..66f41ba 100644 --- a/types.go +++ b/types.go @@ -135,6 +135,11 @@ type ErrorResponse struct { Message string `json:"message"` Description string `json:"description"` Error string `json:"error"` + + // StatusCode is the HTTP status that produced this error, or 0 when the + // request never reached a response (connection failure, decode error). + // Not part of the API payload — it is filled in by this package. + StatusCode int `json:"-"` } // TrashResource represents a resource in the trash