diff --git a/go/vt/mysqlctl/azblobbackupstorage/azblob.go b/go/vt/mysqlctl/azblobbackupstorage/azblob.go index bfbc8fd4842..d00c4c9b3ac 100644 --- a/go/vt/mysqlctl/azblobbackupstorage/azblob.go +++ b/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() diff --git a/go/vt/mysqlctl/azblobbackupstorage/azblob_test.go b/go/vt/mysqlctl/azblobbackupstorage/azblob_test.go new file mode 100644 index 00000000000..7bfbe1f7301 --- /dev/null +++ b/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) +} diff --git a/go/vt/mysqlctl/backupstorage/interface.go b/go/vt/mysqlctl/backupstorage/interface.go index d71417f5661..d549c0b0b6e 100644 --- a/go/vt/mysqlctl/backupstorage/interface.go +++ b/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 diff --git a/go/vt/mysqlctl/xtrabackupengine.go b/go/vt/mysqlctl/xtrabackupengine.go index de53662bf33..90277391ee5 100644 --- a/go/vt/mysqlctl/xtrabackupengine.go +++ b/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() diff --git a/go/vt/mysqlctl/xtrabackupengine_test.go b/go/vt/mysqlctl/xtrabackupengine_test.go index 16ec2812c56..17cb1445cc2 100644 --- a/go/vt/mysqlctl/xtrabackupengine_test.go +++ b/go/vt/mysqlctl/xtrabackupengine_test.go @@ -18,11 +18,13 @@ package mysqlctl import ( "bytes" + "context" "crypto/rand" "io" "os" "path" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -31,6 +33,20 @@ import ( tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata" ) +type ( + // ctxAwareCloser simulates a backend (e.g. GCS) whose Close() blocks on + // upload completion and respects context cancellation. + ctxAwareCloser struct { + ctx context.Context + } +) + +func (c ctxAwareCloser) Write(p []byte) (int, error) { return len(p), nil } +func (c ctxAwareCloser) Close() error { + <-c.ctx.Done() + return c.ctx.Err() +} + func TestFindReplicationPosition(t *testing.T) { input := `MySQL binlog position: filename 'vt-0476396352-bin.000005', position '310088991', GTID of the last change '145e508e-ae54-11e9-8ce6-46824dd1815e:1-3, 1e51f8be-ae54-11e9-a7c6-4280a041109b:1-3, @@ -150,3 +166,91 @@ func TestShouldDrainForBackupXtrabackup(t *testing.T) { assert.True(t, be.ShouldDrainForBackup(nil)) assert.True(t, be.ShouldDrainForBackup(&tabletmanagerdatapb.BackupRequest{})) } + +// TestCloseBackupFilesDoesNotCancelContextOnSuccess guards against a +// regression where closeBackupFiles cancelled ctx as soon as all files +// finished closing. On backends like S3 and Ceph, that ctx is also used by a +// background upload goroutine that Close() does not wait for, so cancelling +// it right after Close() returns aborts an otherwise-successful, still +// in-flight upload. A successful close must only stop the watchdog, never +// cancel the context itself — that's left for bh.Wait()/EndBackup() later. +func TestCloseBackupFilesDoesNotCancelContextOnSuccess(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + logger := logutil.NewMemoryLogger() + + destFiles := []io.WriteCloser{nopWriteCloser{}, nopWriteCloser{}} + var finalErr error + + done := make(chan struct{}) + go func() { + defer close(done) + // Use a generous, CI-safe watchdog timeout. The nop closers return + // immediately, so a successful close stops the watchdog long before + // this fires. A large timeout keeps a preempted CI worker from + // spuriously tripping the watchdog and cancelling ctx. + closeBackupFiles(ctx, cancel, 30*time.Second, destFiles, "backup", len(destFiles), logger, &finalErr) + }() + + // closeBackupFiles stops the watchdog before returning, so once it has + // returned the watchdog can no longer fire. Synchronizing on completion + // (rather than sleeping past a wall-clock window) makes the assertions + // below deterministic: with a 30s watchdog timeout, the timer cannot have + // fired during this sub-second test. Poll non-blockingly so a regression + // that hangs closeBackupFiles fails at the deadline instead of blocking on + // <-done indefinitely. + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, 30*time.Second, 10*time.Millisecond) + + require.NoError(t, finalErr) + + // A successful close must leave ctx untouched for the still-in-flight + // upload, and must not have logged a watchdog timeout. + assert.NoError(t, ctx.Err()) + assert.NotContains(t, logger.String(), "Timed out waiting for Close()") +} + +// TestCloseBackupFilesCancelsOnRealTimeout guards the other direction: if +// Close() genuinely hangs past the timeout, the watchdog must still log and +// cancel ctx so a stuck Close() (e.g. GCS's synchronous upload-on-Close) can +// abort instead of hanging forever. +func TestCloseBackupFilesCancelsOnRealTimeout(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + logger := logutil.NewMemoryLogger() + + destFiles := []io.WriteCloser{ctxAwareCloser{ctx: ctx}} + var finalErr error + + done := make(chan struct{}) + go func() { + defer close(done) + closeBackupFiles(ctx, cancel, 50*time.Millisecond, destFiles, "backup", len(destFiles), logger, &finalErr) + }() + + // The ctxAwareCloser blocks until the watchdog cancels ctx, so this path + // is deterministic in outcome — it only needs a generous, CI-safe deadline + // for how long we wait. A resource-starved runner can pause the goroutine + // for multiple seconds before the 50ms watchdog fires, so use 30s. Poll + // non-blockingly so a regression that hangs closeBackupFiles fails the test + // at the deadline instead of blocking on <-done indefinitely. + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, 30*time.Second, 10*time.Millisecond) + + require.ErrorIs(t, finalErr, context.Canceled) + + assert.Contains(t, logger.String(), "Timed out waiting for Close()") + assert.Error(t, ctx.Err()) +}