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