mysqlctl: prevent closeBackupFiles from cancelling context on successful file close, protecting in-flight S3/Ceph uploads

go/vt/mysqlctl/azblobbackupstorage/azblob.go

@@ -250,7 +250,15 @@ func (bh *AZBlobBackupHandle) AddFile(ctx context.Context, filename string, file
 	reader, writer := io.Pipe()
 
 	bh.waitGroup.Go(func() {
-		_, err := azblob.UploadStreamToBlockBlob(bh.ctx, reader, blockBlobURL, azblob.UploadStreamToBlockBlobOptions{
+		// The upload must honor both the per-file context and the handle-level
+		// abort context: callers cancel the per-file ctx to stop a failing
+		// backup promptly, while AbortBackup cancels bh.ctx. Waiting on this
+		// upload (via Wait) without honoring the per-file ctx would otherwise
+		// let a stalled upload block the caller from ever reaching AbortBackup.
+		uploadCtx, cleanup := mergeCancel(bh.ctx, ctx)
+		defer cleanup()
+
+		_, err := azblob.UploadStreamToBlockBlob(uploadCtx, reader, blockBlobURL, azblob.UploadStreamToBlockBlobOptions{
 			BufferSize: azBlobBufferSize.Get(),
 			MaxBuffers: azBlobParallelism.Get(),
 		})
@@ -263,6 +271,21 @@ func (bh *AZBlobBackupHandle) AddFile(ctx context.Context, filename string, file
 	return writer, nil
 }
 
+// mergeCancel returns a context derived from parent that is also cancelled when
+// other is cancelled, along with a cleanup function that must be called when
+// the work is done to release resources and stop watching other. Context values
+// come from parent only. If other is cancelled with a cause, mergeCancel
+// preserves that cause so callers see why the context was cancelled rather
+// than a bare context.Canceled.
+func mergeCancel(parent, other context.Context) (context.Context, func()) {
+	ctx, cancel := context.WithCancelCause(parent)
+	stop := context.AfterFunc(other, func() { cancel(context.Cause(other)) })
+	return ctx, func() {
+		stop()
+		cancel(nil)
+	}
+}
+
 // Wait implements BackupHandle.
 func (bh *AZBlobBackupHandle) Wait() {
 	bh.waitGroup.Wait()

go/vt/mysqlctl/azblobbackupstorage/azblob_test.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2026 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package azblobbackupstorage
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMergeCancelCancelsOnOther guards the fix that makes an Azure upload honor
+// the per-file context passed to AddFile in addition to the handle abort
+// context. Cancelling the "other" (per-file) context must cancel the merged
+// context, even though the merged context is derived from the parent (handle)
+// context.
+func TestMergeCancelCancelsOnOther(t *testing.T) {
+	parent := t.Context()
+	other, cancelOther := context.WithCancel(t.Context())
+
+	ctx, cleanup := mergeCancel(parent, other)
+	defer cleanup()
+
+	require.NoError(t, ctx.Err())
+
+	cancelOther()
+
+	assert.Eventually(t, func() bool {
+		return ctx.Err() != nil
+	}, 30*time.Second, 10*time.Millisecond)
+	assert.ErrorIs(t, ctx.Err(), context.Canceled)
+}
+
+// TestMergeCancelCancelsOnParent verifies the merged context still honors the
+// parent (handle abort) context, which is what AbortBackup cancels.
+func TestMergeCancelCancelsOnParent(t *testing.T) {
+	parent, cancelParent := context.WithCancel(t.Context())
+	other := t.Context()
+
+	ctx, cleanup := mergeCancel(parent, other)
+	defer cleanup()
+
+	require.NoError(t, ctx.Err())
+
+	cancelParent()
+
+	assert.Eventually(t, func() bool {
+		return ctx.Err() != nil
+	}, 30*time.Second, 10*time.Millisecond)
+	assert.ErrorIs(t, ctx.Err(), context.Canceled)
+}
+
+// TestMergeCancelCleanupCancels verifies the cleanup function cancels the
+// merged context so a completed upload doesn't leak it.
+func TestMergeCancelCleanupCancels(t *testing.T) {
+	ctx, cleanup := mergeCancel(t.Context(), t.Context())
+	require.NoError(t, ctx.Err())
+
+	cleanup()
+
+	assert.ErrorIs(t, ctx.Err(), context.Canceled)
+}

go/vt/mysqlctl/backupstorage/interface.go

@@ -66,8 +66,7 @@ type BackupHandle interface {
 	// characters and hyphens.
 	// It should be thread safe, it is possible to call AddFile in
 	// multiple go routines once a backup has been started.
-	// The context is valid for the duration of the writes, until the
-	// WriteCloser is closed.
+	// The context is valid until Wait() returns, even for async backends.
 	// filesize should not be treated as an exact value but rather
 	// as an approximate value.
 	// A filesize of -1 should be treated as a special value indicating that

go/vt/mysqlctl/xtrabackupengine.go

@@ -172,6 +172,85 @@ func closeFile(wc io.WriteCloser, fileName string, logger logutil.Logger, finalE
 	}
 }
 
+// closeBackupFiles closes destFiles, renaming per-stripe as needed. A
+// background watchdog cancels ctx via cancel if the Close() calls don't
+// return within timeout. On some backends (e.g. S3, Ceph) ctx is also used by
+// a background upload goroutine that outlives Close(), so cancel must only be
+// called on an actual timeout, never just because closing finished — doing so
+// would abort an otherwise-successful upload that hasn't been drained yet by
+// bh.Wait()/EndBackup().
+func closeBackupFiles(ctx context.Context, cancel context.CancelFunc, timeout time.Duration, destFiles []io.WriteCloser, backupFileName string, numStripes int, logger logutil.Logger, finalErr *error) {
+	done := make(chan struct{})
+	// watchdogDone is closed when the watchdog goroutine exits, so we can wait
+	// for it to fully finish before returning and be sure it will make no
+	// further calls to cancel().
+	watchdogDone := make(chan struct{})
+	// closingFinished, guarded by mu, records that every Close() call returned.
+	// The watchdog checks it under the same lock before cancelling, so once
+	// closing is marked finished the watchdog cannot cancel.
+	//
+	// This narrows but cannot fully eliminate the boundary race: "Close()
+	// returned" only becomes observable via the statement that sets this flag
+	// (below), so a timer that fires in the few-instruction gap between the
+	// final Close() returning and mu.Lock() can still win the lock, see false,
+	// and cancel(). That gap is irreducible — no completion signal (flag,
+	// channel close/send, timer.Stop()) can land atomically with Close()
+	// returning. The consequence, given the caller drains and inspects uploads
+	// via bh.Wait()/bh.Error() before writing the MANIFEST, is at worst a rare
+	// spurious backup failure that gets retried — never the recreate-after-
+	// abort corruption this whole change exists to prevent.
+	var mu sync.Mutex
+	closingFinished := false
+	go func() {
+		defer close(watchdogDone)
+		timer := time.NewTimer(timeout)
+
+		select {
+		case <-done:
+			timer.Stop()
+		case <-timer.C:
+			mu.Lock()
+			defer mu.Unlock()
+			if closingFinished {
+				// Closing finished just as the timer fired; don't cancel an
+				// otherwise-successful, still-in-flight upload.
+				return
+			}
+			logger.Errorf("Timed out waiting for Close() on backup file to complete")
+			// Cancelling the Context that was originally passed to bh.AddFile()
+			// should hopefully cause Close() calls on the file that AddFile()
+			// returned to abort. If the underlying implementation doesn't
+			// respect cancellation of the AddFile() Context while inside
+			// Close(), then we just hang because it's unsafe to return and
+			// leave Close() running indefinitely in the background.
+			cancel()
+		}
+	}()
+
+	filename := backupFileName
+	for i, file := range destFiles {
+		if numStripes > 1 {
+			filename = stripeFileName(backupFileName, i)
+		}
+		closeFile(file, filename, logger, finalErr)
+	}
+
+	// Mark closing finished under the lock before signalling the watchdog, so a
+	// timer that fires at this instant observes the flag and suppresses its
+	// cancel() instead of aborting a successful close.
+	mu.Lock()
+	closingFinished = true
+	mu.Unlock()
+	// Signal the watchdog to stop waiting now that closing has finished. This
+	// must not call cancel(): closing successfully doesn't mean any
+	// background upload started by bh.AddFile() has finished too.
+	close(done)
+	// Wait for the watchdog to fully exit before returning, so no cancel() can
+	// fire after this function returns and abort an in-flight S3/Ceph upload
+	// that hasn't been drained yet by bh.Wait()/EndBackup().
+	<-watchdogDone
+}
+
 // ExecuteBackup runs a backup based on given params. This could be a full or incremental backup.
 // The function returns a BackupResult that indicates the usability of the backup, and an overall error.
 func (be *XtrabackupEngine) ExecuteBackup(ctx context.Context, params BackupParams, bh backupstorage.BackupHandle) (BackupResult, error) {
@@ -336,45 +415,40 @@ func (be *XtrabackupEngine) backupFiles(
 	// This context also allows us to immediately abort AddFiles if we encountered
 	// an error in this function.
 	addFilesCtx, cancelAddFiles := context.WithCancel(ctx)
+	// This must run after closeBackupFiles below, so it's deferred first:
+	// defers run LIFO, and closeBackupFiles needs addFilesCtx to still be live
+	// while it closes destFiles.
 	defer func() {
 		if finalErr != nil {
+			// We're already failing: cancel first so any straggling
+			// background uploads (S3, Ceph) stop as promptly as possible,
+			// then drain them. Without this wait, the caller could invoke
+			// AbortBackup() while an upload is still in flight; AbortBackup
+			// only removes objects that already exist in storage, so an
+			// upload landing after that removal would recreate part of the
+			// backup we're trying to discard.
 			cancelAddFiles()
+			bh.Wait()
+			return
 		}
+
+		// Drain any in-flight background uploads (S3, Ceph) and surface
+		// their errors before letting the caller proceed to write the
+		// MANIFEST, for the same reason: the MANIFEST must not be written
+		// (and the backup must not be reported usable) while an upload that
+		// could still fail is outstanding.
+		bh.Wait()
+		if err := bh.Error(); err != nil {
+			finalErr = vterrors.Wrap(err, "error uploading backup file")
+		}
+		cancelAddFiles()
 	}()
 
 	destFiles, err := addStripeFiles(addFilesCtx, params, bh, backupFileName, numStripes)
 	if err != nil {
 		return replicationPosition, vterrors.Wrapf(err, "cannot create backup file %v", backupFileName)
 	}
-	defer func() {
-		// Impose a timeout on the process of closing files.
-		go func() {
-			timer := time.NewTimer(closeTimeout)
-
-			select {
-			case <-addFilesCtx.Done():
-				timer.Stop()
-				return
-			case <-timer.C:
-				params.Logger.Errorf("Timed out waiting for Close() on backup file to complete")
-				// Cancelling the Context that was originally passed to bh.AddFile()
-				// should hopefully cause Close() calls on the file that AddFile()
-				// returned to abort. If the underlying implementation doesn't
-				// respect cancellation of the AddFile() Context while inside
-				// Close(), then we just hang because it's unsafe to return and
-				// leave Close() running indefinitely in the background.
-				cancelAddFiles()
-			}
-		}()
-
-		filename := backupFileName
-		for i, file := range destFiles {
-			if numStripes > 1 {
-				filename = stripeFileName(backupFileName, i)
-			}
-			closeFile(file, filename, params.Logger, &finalErr)
-		}
-	}()
+	defer closeBackupFiles(addFilesCtx, cancelAddFiles, closeTimeout, destFiles, backupFileName, numStripes, params.Logger, &finalErr)
 
 	backupCmd := exec.CommandContext(ctx, backupProgram, flagsToExec...)
 	backupOut, err := backupCmd.StdoutPipe()