diff --git a/go/porcelain/deploy.go b/go/porcelain/deploy.go index f421e501..ac0761be 100644 --- a/go/porcelain/deploy.go +++ b/go/porcelain/deploy.go @@ -11,6 +11,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "hash" "io" "io/ioutil" "os" @@ -125,9 +126,17 @@ type FileBundle struct { Size *int64 `json:"size,omitempty"` FunctionMetadata *FunctionMetadata - // Path OR Buffer should be populated - Path string + // Path is the location of the file on disk. Uploads always stream from Path. + Path string + + // Deprecated: uploads always stream from Path; this package no longer reads Buffer. It is retained + // only for backwards compatibility with external callers and may be removed in a future release. + // Leave it nil to have the (also deprecated) Read/Seek/Close methods stream from Path instead. Buffer io.ReadSeeker + + // pathReader is lazily opened from Path when Buffer is nil, so the deprecated Read/Seek/Close + // methods keep working for external callers that treat a FileBundle as an io.ReadSeekCloser. + pathReader *os.File } type FunctionMetadata struct { @@ -139,15 +148,46 @@ type toolchainSpec struct { Runtime string `json:"runtime"` } +// Deprecated: read directly from Path (e.g. via os.Open) instead. When Buffer is set, Read reads +// from it; otherwise it streams from Path. Retained for backwards compatibility with external +// callers and may be removed in a future release. func (f *FileBundle) Read(p []byte) (n int, err error) { - return f.Buffer.Read(p) + if f.Buffer != nil { + return f.Buffer.Read(p) + } + if f.pathReader == nil { + if f.pathReader, err = os.Open(f.Path); err != nil { + return 0, err + } + } + return f.pathReader.Read(p) } +// Deprecated: read directly from Path (e.g. via os.Open) instead. When Buffer is set, Seek seeks +// it; otherwise it seeks the stream opened from Path. Retained for backwards compatibility with +// external callers and may be removed in a future release. func (f *FileBundle) Seek(offset int64, whence int) (int64, error) { - return f.Buffer.Seek(offset, whence) + if f.Buffer != nil { + return f.Buffer.Seek(offset, whence) + } + if f.pathReader == nil { + var err error + if f.pathReader, err = os.Open(f.Path); err != nil { + return 0, err + } + } + return f.pathReader.Seek(offset, whence) } +// Deprecated: retained for backwards compatibility with external callers and may be removed in a +// future release. It closes the stream lazily opened from Path by Read/Seek; it never closes a +// caller-supplied Buffer. func (f *FileBundle) Close() error { + if f.pathReader != nil { + err := f.pathReader.Close() + f.pathReader = nil + return err + } return nil } @@ -263,7 +303,13 @@ func (n *Netlify) DoDeploy(ctx context.Context, options *DeployOptions, deploy * options.files = files - functions, schedules, functionsConfig, err := bundle(ctx, options.FunctionsDir, options.Observer) + // The temp dir is created lazily, only if a function actually needs to be zipped. Pre-bundled + // .zip/.tar functions stream from their original path and never touch it, so a deploy with no + // unbundled functions creates no temp dir at all. + functionsTmpDir := &lazyTempDir{} + defer functionsTmpDir.remove() + + functions, schedules, functionsConfig, err := bundle(ctx, options.FunctionsDir, functionsTmpDir, options.Observer) if err != nil { if options.Observer != nil { options.Observer.OnFailedWalk() @@ -456,6 +502,7 @@ func (n *Netlify) uploadFiles(ctx context.Context, d *models.Deploy, files *depl log := context.GetLogger(ctx) log.Infof("Uploading %v files", count) + var abortErr error for _, sha := range required { if files, exist := files.Hashed[sha]; exist { file := files[0] @@ -466,7 +513,11 @@ func (n *Netlify) uploadFiles(ctx context.Context, d *models.Deploy, files *depl go n.uploadFile(ctx, d, file, observer, t, timeout, wg, sem, sharedErr, skipRetry) case <-ctx.Done(): log.Info("Context terminated, aborting file upload") - return errors.Wrap(ctx.Err(), "aborted file upload early") + abortErr = errors.Wrap(ctx.Err(), "aborted file upload early") + } + + if abortErr != nil { + break } if len(files) > 1 { @@ -478,8 +529,16 @@ func (n *Netlify) uploadFiles(ctx context.Context, d *models.Deploy, files *depl } } + // Always wait for in-flight uploads to finish before returning. On the ctx.Done() + // path this prevents orphaned uploadFile goroutines from racing against the caller's + // deferred temp-dir cleanup (os.RemoveAll), which would otherwise open files that are + // being deleted and surface spurious "no such file or directory" errors. wg.Wait() + if abortErr != nil { + return abortErr + } + return sharedErr.err } @@ -539,23 +598,25 @@ func (n *Netlify) uploadFile(ctx context.Context, d *models.Deploy, f *FileBundl _, operationError = n.Operations.UploadDeployFile(params, authInfo) } case functionUpload: - params := operations.NewUploadDeployFunctionParams().WithDeployID(d.ID).WithName(f.Name).WithFileBody(f).WithRuntime(&f.Runtime) + var body io.ReadCloser + body, operationError = os.Open(f.Path) + if operationError == nil { + defer body.Close() + params := operations.NewUploadDeployFunctionParams().WithDeployID(d.ID).WithName(f.Name).WithFileBody(body).WithRuntime(&f.Runtime) - if retryCount > 0 { - params = params.WithXNfRetryCount(&retryCount) - } + if retryCount > 0 { + params = params.WithXNfRetryCount(&retryCount) + } - if f.FunctionMetadata != nil { - params = params.WithInvocationMode(&f.FunctionMetadata.InvocationMode) - params = params.WithTimeout(&f.FunctionMetadata.Timeout) - } + if f.FunctionMetadata != nil { + params = params.WithInvocationMode(&f.FunctionMetadata.InvocationMode) + params = params.WithTimeout(&f.FunctionMetadata.Timeout) + } - if timeout != 0 { - params.SetRequestTimeout(timeout) - } - _, operationError = n.Operations.UploadDeployFunction(params, authInfo) - if operationError != nil { - f.Buffer.Seek(0, 0) + if timeout != 0 { + params.SetRequestTimeout(timeout) + } + _, operationError = n.Operations.UploadDeployFunction(params, authInfo) } } @@ -601,6 +662,14 @@ func (n *Netlify) uploadFile(ctx context.Context, d *models.Deploy, f *FileBundl } func createFileBundle(rel, path string) (*FileBundle, error) { + return createFileBundleWithHasher(rel, path, sha1.New()) +} + +func createFunctionFileBundle(rel, path string) (*FileBundle, error) { + return createFileBundleWithHasher(rel, path, sha256.New()) +} + +func createFileBundleWithHasher(rel, path string, s hash.Hash) (*FileBundle, error) { o, err := os.Open(path) if err != nil { return nil, err @@ -612,7 +681,6 @@ func createFileBundle(rel, path string) (*FileBundle, error) { Path: path, } - s := sha1.New() if _, err := io.Copy(s, o); err != nil { return nil, err } @@ -713,7 +781,30 @@ func addInternalFilesToDeploy(dir, internalPath string, files *deployFiles, obse }) } -func bundle(ctx context.Context, functionDir string, observer DeployObserver) (*deployFiles, []*models.FunctionSchedule, map[string]models.FunctionConfig, error) { +type lazyTempDir struct { + root string + path string + created bool +} + +func (l *lazyTempDir) get() (string, error) { + if !l.created { + path, err := os.MkdirTemp(l.root, "netlify-deploy-functions-") + if err != nil { + return "", err + } + l.path, l.created = path, true + } + return l.path, nil +} + +func (l *lazyTempDir) remove() { + if l.created { + os.RemoveAll(l.path) + } +} + +func bundle(ctx context.Context, functionDir string, tmpDir *lazyTempDir, observer DeployObserver) (*deployFiles, []*models.FunctionSchedule, map[string]models.FunctionConfig, error) { if functionDir == "" { return nil, nil, nil, nil } @@ -725,7 +816,7 @@ func bundle(ctx context.Context, functionDir string, observer DeployObserver) (* if err == nil { defer manifestFile.Close() - return bundleFromManifest(ctx, manifestFile, observer) + return bundleFromManifest(ctx, manifestFile, tmpDir, observer) } functions := newDeployFiles() @@ -744,19 +835,19 @@ func bundle(ctx context.Context, functionDir string, observer DeployObserver) (* if err != nil { return nil, nil, nil, err } - file, err := newFunctionFile(filePath, i, runtime, nil, observer) + file, err := newFunctionFile(filePath, i, runtime, nil, tmpDir, observer) if err != nil { return nil, nil, nil, err } functions.Add(file.Name, file) case jsFile(i): - file, err := newFunctionFile(filePath, i, jsRuntime, nil, observer) + file, err := newFunctionFile(filePath, i, jsRuntime, nil, tmpDir, observer) if err != nil { return nil, nil, nil, err } functions.Add(file.Name, file) case goFile(filePath, i, observer): - file, err := newFunctionFile(filePath, i, amazonLinux2, nil, observer) + file, err := newFunctionFile(filePath, i, amazonLinux2, nil, tmpDir, observer) if err != nil { return nil, nil, nil, err } @@ -771,7 +862,7 @@ func bundle(ctx context.Context, functionDir string, observer DeployObserver) (* return functions, nil, nil, nil } -func bundleFromManifest(ctx context.Context, manifestFile *os.File, observer DeployObserver) (*deployFiles, []*models.FunctionSchedule, map[string]models.FunctionConfig, error) { +func bundleFromManifest(ctx context.Context, manifestFile *os.File, tmpDir *lazyTempDir, observer DeployObserver) (*deployFiles, []*models.FunctionSchedule, map[string]models.FunctionConfig, error) { manifestBytes, err := ioutil.ReadAll(manifestFile) if err != nil { return nil, nil, nil, err @@ -808,7 +899,7 @@ func bundleFromManifest(ctx context.Context, manifestFile *os.File, observer Dep InvocationMode: function.InvocationMode, Timeout: function.Timeout, } - file, err := newFunctionFile(function.Path, fileInfo, runtime, &meta, observer) + file, err := newFunctionFile(function.Path, fileInfo, runtime, &meta, tmpDir, observer) if err != nil { return nil, nil, nil, err } @@ -911,60 +1002,79 @@ func readZipRuntime(filePath string) (string, error) { return jsRuntime, nil } -func newFunctionFile(filePath string, i os.FileInfo, runtime string, metadata *FunctionMetadata, observer DeployObserver) (*FileBundle, error) { - file := &FileBundle{ - Name: strings.TrimSuffix(i.Name(), filepath.Ext(i.Name())), - Runtime: runtime, - } - - s := sha256.New() +func newFunctionFile(filePath string, i os.FileInfo, runtime string, metadata *FunctionMetadata, tmpDir *lazyTempDir, observer DeployObserver) (*FileBundle, error) { + var file *FileBundle + var err error - fileEntry, err := os.Open(filePath) + if zipFile(i) || tarFile(i) { + name := strings.TrimSuffix(i.Name(), filepath.Ext(i.Name())) + file, err = createFunctionFileBundle(name, filePath) + } else { + file, err = zipFunctionFile(filePath, i, runtime, tmpDir) + } if err != nil { return nil, err } - defer fileEntry.Close() - var buf io.ReadWriter - - if zipFile(i) || tarFile(i) { - buf = fileEntry - } else { - buf = new(bytes.Buffer) - archive := zip.NewWriter(buf) + file.Runtime = runtime + file.FunctionMetadata = metadata - fileHeader, err := createHeader(archive, i, runtime) - if err != nil { + if observer != nil { + if err := observer.OnSuccessfulStep(file); err != nil { return nil, err } + } - if _, err = io.Copy(fileHeader, fileEntry); err != nil { - return nil, err - } + return file, nil +} - if err := archive.Close(); err != nil { - return nil, err - } +func zipFunctionFile(filePath string, i os.FileInfo, runtime string, tmpDir *lazyTempDir) (*FileBundle, error) { + src, err := os.Open(filePath) + if err != nil { + return nil, err } + defer src.Close() - fileBuffer := new(bytes.Buffer) - m := io.MultiWriter(s, fileBuffer) - - if _, err := io.Copy(m, buf); err != nil { + dir, err := tmpDir.get() + if err != nil { return nil, err } - file.Sum = hex.EncodeToString(s.Sum(nil)) - file.Buffer = bytes.NewReader(fileBuffer.Bytes()) - if observer != nil { - if err := observer.OnSuccessfulStep(file); err != nil { - return nil, err + tmp, err := os.CreateTemp(dir, "function-*.zip") + if err != nil { + return nil, err + } + defer func() { + if tmp != nil { + _ = tmp.Close() } + }() + + s := sha256.New() + archive := zip.NewWriter(io.MultiWriter(tmp, s)) + + fileHeader, err := createHeader(archive, i, runtime) + if err != nil { + return nil, err + } + if _, err := io.Copy(fileHeader, src); err != nil { + return nil, err + } + if err := archive.Close(); err != nil { + return nil, err } - file.FunctionMetadata = metadata + tmpName := tmp.Name() + if err := tmp.Close(); err != nil { + return nil, err + } + tmp = nil - return file, nil + return &FileBundle{ + Name: strings.TrimSuffix(i.Name(), filepath.Ext(i.Name())), + Sum: hex.EncodeToString(s.Sum(nil)), + Path: tmpName, + }, nil } func zipFile(i os.FileInfo) bool { diff --git a/go/porcelain/deploy_test.go b/go/porcelain/deploy_test.go index 7bdf62c0..869c68f2 100644 --- a/go/porcelain/deploy_test.go +++ b/go/porcelain/deploy_test.go @@ -4,6 +4,7 @@ import ( "bytes" gocontext "context" "fmt" + "io" "io/ioutil" "net/http" "net/http/httptest" @@ -12,6 +13,8 @@ import ( "path" "path/filepath" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -86,6 +89,54 @@ func TestAddWithLargeMedia(t *testing.T) { } } +// The exported Read/Seek/Close adapters must behave identically whether the bytes come from a +// caller-supplied Buffer or (with Buffer nil) are streamed from Path. The Path case is also a +// regression guard: it used to nil-panic instead of falling back to Path. +func TestFileBundleReadSeekClose(t *testing.T) { + const contents = "hello deploy world" + + tests := []struct { + name string + newFileBundleUnderTest func(t *testing.T) *FileBundle + }{ + { + name: "streams from Path when Buffer is nil", + newFileBundleUnderTest: func(t *testing.T) *FileBundle { + p := filepath.Join(t.TempDir(), "file.txt") + require.NoError(t, os.WriteFile(p, []byte(contents), 0o600)) + return &FileBundle{Path: p} + }, + }, + { + name: "reads from Buffer when set", + newFileBundleUnderTest: func(t *testing.T) *FileBundle { + return &FileBundle{Buffer: strings.NewReader(contents)} + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + f := test.newFileBundleUnderTest(t) + + got, err := io.ReadAll(f) + require.NoError(t, err) + assert.Equal(t, contents, string(got)) + + // Rewind and re-read, mirroring the upload client's retry contract. + pos, err := f.Seek(0, 0) + require.NoError(t, err) + assert.Equal(t, int64(0), pos) + + got, err = io.ReadAll(f) + require.NoError(t, err) + assert.Equal(t, contents, string(got)) + + assert.NoError(t, f.Close()) + }) + } +} + func TestOpenAPIClientWithWeirdResponse(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { rw.Header().Set("Content-Type", "application/json; charset=utf-8") @@ -386,6 +437,82 @@ func TestUploadFiles_Cancelation(t *testing.T) { require.ErrorIs(t, err, gocontext.Canceled) } +func TestUploadFiles_CancelationWaitsForInFlightUploads(t *testing.T) { + ctx, cancel := gocontext.WithCancel(gocontext.Background()) + + uploadStarted := make(chan struct{}) + releaseUpload := make(chan struct{}) + var startOnce, releaseOnce sync.Once + releaseUploads := func() { releaseOnce.Do(func() { close(releaseUpload) }) } + + var uploadRequests int32 + server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { + atomic.AddInt32(&uploadRequests, 1) + // Signal that the first upload is in flight, then block until the test + // releases it, keeping the goroutine parked while we cancel the context. + startOnce.Do(func() { close(uploadStarted) }) + <-releaseUpload + rw.Header().Set("Content-Type", "application/json; charset=utf-8") + rw.Write([]byte(`{ "state": "uploaded" }`)) + })) + defer server.Close() + // Registered after server.Close so it runs first (LIFO): always unblock the + // parked handler before Close, which otherwise waits on outstanding requests. + // Without this a failing assertion (e.g. the bug being reintroduced) would + // deadlock server.Close and time out instead of failing fast. + defer releaseUploads() + + hu, _ := url.Parse(server.URL) + tr := apiClient.NewWithClient(hu.Host, "/api/v1", []string{"http"}, http.DefaultClient) + client := NewRetryable(tr, strfmt.Default, 1) + client.uploadLimit = 1 // Force the second file to wait on the semaphore. + ctx = context.WithAuthInfo(ctx, apiClient.BearerToken("token")) + + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "foo.html"), []byte("Hello"), 0644)) + require.NoError(t, os.WriteFile(filepath.Join(dir, "bar.html"), []byte("World"), 0644)) + + files, err := walk(dir, nil, false, false) + require.NoError(t, err) + d := &models.Deploy{} + for _, bundle := range files.Files { + d.Required = append(d.Required, bundle.Sum) + } + + returned := make(chan error, 1) + go func() { + returned <- client.uploadFiles(ctx, d, files, nil, fileUpload, time.Minute, false) + }() + + // Wait for the first upload to be in flight, then cancel the deploy. The + // second file's select hits ctx.Done() (the semaphore is still held), so + // uploadFiles breaks out of its loop while the first upload is still parked. + <-uploadStarted + cancel() + + // uploadFiles blocks on wg.Wait() until the in-flight upload + // finishes, so it must NOT have returned while the upload is still parked. + select { + case <-returned: + t.Fatal("uploadFiles returned before in-flight upload finished; orphaned goroutine would race temp-dir cleanup") + case <-time.After(100 * time.Millisecond): + } + + // Let the in-flight upload complete; uploadFiles should now return the + // cancellation error. + releaseUploads() + select { + case err = <-returned: + require.ErrorIs(t, err, gocontext.Canceled) + case <-time.After(5 * time.Second): + t.Fatal("uploadFiles did not return after in-flight upload was released") + } + + // Only the first file should ever hit the server: the second file's upload + // is aborted at the semaphore by the canceled context and must never start. + require.Equal(t, int32(1), atomic.LoadInt32(&uploadRequests), "second file must not be uploaded after cancelation") +} + func TestUploadFiles_Errors(t *testing.T) { ctx := gocontext.Background() @@ -485,7 +612,7 @@ func TestUploadFunctions422Error_SkipsRetry(t *testing.T) { defer os.RemoveAll(dir) require.NoError(t, ioutil.WriteFile(filepath.Join(functionsPath, "foo.js"), []byte("module.exports = () => {}"), 0644)) - files, _, _, err := bundle(ctx, functionsPath, mockObserver{}) + files, _, _, err := bundle(ctx, functionsPath, &lazyTempDir{root: t.TempDir()}, mockObserver{}) require.NoError(t, err) d := &models.Deploy{} for _, bundle := range files.Files { @@ -584,7 +711,7 @@ func TestUploadFiles_SkipEqualFiles(t *testing.T) { require.NoError(t, ioutil.WriteFile(filepath.Join(functionsDir, "a.zip"), bundleBody, 0644)) require.NoError(t, ioutil.WriteFile(filepath.Join(functionsDir, "b.zip"), bundleBody, 0644)) - functions, _, _, err := bundle(ctx, functionsDir, mockObserver{}) + functions, _, _, err := bundle(ctx, functionsDir, &lazyTempDir{root: t.TempDir()}, mockObserver{}) require.NoError(t, err) d := &models.Deploy{} @@ -646,7 +773,7 @@ func TestUploadFunctions_RetryCountHeader(t *testing.T) { defer os.RemoveAll(dir) require.NoError(t, ioutil.WriteFile(filepath.Join(functionsPath, "foo.js"), []byte("module.exports = () => {}"), 0644)) - files, _, _, err := bundle(ctx, functionsPath, mockObserver{}) + files, _, _, err := bundle(ctx, functionsPath, &lazyTempDir{root: t.TempDir()}, mockObserver{}) require.NoError(t, err) d := &models.Deploy{} for _, bundle := range files.Files { @@ -657,7 +784,7 @@ func TestUploadFunctions_RetryCountHeader(t *testing.T) { } func TestBundle(t *testing.T) { - functions, schedules, functionsConfig, err := bundle(gocontext.Background(), "../internal/data", mockObserver{}) + functions, schedules, functionsConfig, err := bundle(gocontext.Background(), "../internal/data", &lazyTempDir{root: t.TempDir()}, mockObserver{}) assert.Nil(t, err) assert.Equal(t, 5, len(functions.Files)) @@ -732,7 +859,7 @@ func TestBundleWithManifest(t *testing.T) { defer os.Remove(manifestPath) assert.Nil(t, err) - functions, schedules, functionsConfig, err := bundle(gocontext.Background(), "../internal/data", mockObserver{}) + functions, schedules, functionsConfig, err := bundle(gocontext.Background(), "../internal/data", &lazyTempDir{root: t.TempDir()}, mockObserver{}) assert.Nil(t, err) assert.Equal(t, 1, len(schedules)) @@ -794,7 +921,7 @@ func TestBundleWithManifestEventSubscriptions(t *testing.T) { assert.Nil(t, err) defer manifestFileHandle.Close() - _, _, functionsConfig, err := bundleFromManifest(gocontext.Background(), manifestFileHandle, mockObserver{}) + _, _, functionsConfig, err := bundleFromManifest(gocontext.Background(), manifestFileHandle, &lazyTempDir{root: t.TempDir()}, mockObserver{}) assert.Nil(t, err) helloJSConfig := functionsConfig["hello-js-function-test"]