FIX (chunking): Add backuping chunk by chunk without buffering in RAM and improve cancelation process

Author: RostislavDugin

backend/internal/features/backups/backups/backup_context_manager.go

@@ -2,46 +2,59 @@ package backups
 
 import (
 	"context"
-	"errors"
 	"sync"
 
 	"github.com/google/uuid"
 )
 
 type BackupContextManager struct {
-	mu          sync.RWMutex
-	cancelFuncs map[uuid.UUID]context.CancelFunc
+	mu               sync.RWMutex
+	cancelFuncs      map[uuid.UUID]context.CancelFunc
+	cancelledBackups map[uuid.UUID]bool
 }
 
 func NewBackupContextManager() *BackupContextManager {
 	return &BackupContextManager{
-		cancelFuncs: make(map[uuid.UUID]context.CancelFunc),
+		cancelFuncs:      make(map[uuid.UUID]context.CancelFunc),
+		cancelledBackups: make(map[uuid.UUID]bool),
 	}
 }
 
 func (m *BackupContextManager) RegisterBackup(backupID uuid.UUID, cancelFunc context.CancelFunc) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.cancelFuncs[backupID] = cancelFunc
+	delete(m.cancelledBackups, backupID)
 }
 
 func (m *BackupContextManager) CancelBackup(backupID uuid.UUID) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
+	if m.cancelledBackups[backupID] {
+		return nil
+	}
+
 	cancelFunc, exists := m.cancelFuncs[backupID]
-	if !exists {
-		return errors.New("backup is not in progress or already completed")
+	if exists {
+		cancelFunc()
+		delete(m.cancelFuncs, backupID)
 	}
 
-	cancelFunc()
-	delete(m.cancelFuncs, backupID)
+	m.cancelledBackups[backupID] = true
 
 	return nil
 }
 
+func (m *BackupContextManager) IsCancelled(backupID uuid.UUID) bool {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.cancelledBackups[backupID]
+}
+
 func (m *BackupContextManager) UnregisterBackup(backupID uuid.UUID) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	delete(m.cancelFuncs, backupID)
+	delete(m.cancelledBackups, backupID)
 }

backend/internal/features/backups/backups/controller_test.go

@@ -1,6 +1,7 @@
 package backups
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -701,7 +702,7 @@ func createTestBackup(
 	dummyContent := []byte("dummy backup content for testing")
 	reader := strings.NewReader(string(dummyContent))
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
-	if err := storages[0].SaveFile(encryption.GetFieldEncryptor(), logger, backup.ID, reader); err != nil {
+	if err := storages[0].SaveFile(context.Background(), encryption.GetFieldEncryptor(), logger, backup.ID, reader); err != nil {
 		panic(fmt.Sprintf("Failed to create test backup file: %v", err))
 	}
 

backend/internal/features/backups/backups/service.go

@@ -275,7 +275,12 @@ func (s *BackupService) MakeBackup(databaseID uuid.UUID, isLastTry bool) {
 		errMsg := err.Error()
 
 		// Check if backup was cancelled (not due to shutdown)
-		if strings.Contains(errMsg, "backup cancelled") && !strings.Contains(errMsg, "shutdown") {
+		isCancelled := strings.Contains(errMsg, "backup cancelled") ||
+			strings.Contains(errMsg, "context canceled") ||
+			errors.Is(err, context.Canceled)
+		isShutdown := strings.Contains(errMsg, "shutdown")
+
+		if isCancelled && !isShutdown {
 			backup.Status = BackupStatusCanceled
 			backup.BackupDurationMs = time.Since(start).Milliseconds()
 			backup.BackupSizeMb = 0

backend/internal/features/backups/backups/usecases/postgresql/create_backup_uc.go

@@ -45,6 +45,11 @@ type CreatePostgresqlBackupUsecase struct {
 	fieldEncryptor   encryption.FieldEncryptor
 }
 
+type writeResult struct {
+	bytesWritten int
+	writeErr     error
+}
+
 // Execute creates a backup of the database
 func (uc *CreatePostgresqlBackupUsecase) Execute(
 	ctx context.Context,
@@ -172,7 +177,7 @@ func (uc *CreatePostgresqlBackupUsecase) streamToStorage(
 	// Start streaming into storage in its own goroutine
 	saveErrCh := make(chan error, 1)
 	go func() {
-		saveErr := storage.SaveFile(uc.fieldEncryptor, uc.logger, backupID, storageReader)
+		saveErr := storage.SaveFile(ctx, uc.fieldEncryptor, uc.logger, backupID, storageReader)
 		saveErrCh <- saveErr
 	}()
 
@@ -195,12 +200,10 @@ func (uc *CreatePostgresqlBackupUsecase) streamToStorage(
 		copyResultCh <- err
 	}()
 
-	// Wait for the copy to finish first, then the dump process
 	copyErr := <-copyResultCh
 	bytesWritten := <-bytesWrittenCh
 	waitErr := cmd.Wait()
 
-	// Check for shutdown or cancellation before finalizing
 	select {
 	case <-ctx.Done():
 		uc.cleanupOnCancellation(encryptionWriter, storageWriter, saveErrCh)
@@ -213,7 +216,6 @@ func (uc *CreatePostgresqlBackupUsecase) streamToStorage(
 		return nil, err
 	}
 
-	// Wait until storage ends reading
 	saveErr := <-saveErrCh
 	stderrOutput := <-stderrCh
 
@@ -267,7 +269,23 @@ func (uc *CreatePostgresqlBackupUsecase) copyWithShutdownCheck(
 
 		bytesRead, readErr := src.Read(buf)
 		if bytesRead > 0 {
-			bytesWritten, writeErr := dst.Write(buf[0:bytesRead])
+			writeResultCh := make(chan writeResult, 1)
+			go func() {
+				bytesWritten, writeErr := dst.Write(buf[0:bytesRead])
+				writeResultCh <- writeResult{bytesWritten, writeErr}
+			}()
+
+			var bytesWritten int
+			var writeErr error
+
+			select {
+			case <-ctx.Done():
+				return totalBytesWritten, fmt.Errorf("copy cancelled during write: %w", ctx.Err())
+			case result := <-writeResultCh:
+				bytesWritten = result.bytesWritten
+				writeErr = result.writeErr
+			}
+
 			if bytesWritten < 0 || bytesRead < bytesWritten {
 				bytesWritten = 0
 				if writeErr == nil {
@@ -354,6 +372,9 @@ func (uc *CreatePostgresqlBackupUsecase) createBackupContext(
 			select {
 			case <-ctx.Done():
 				return
+			case <-parentCtx.Done():
+				cancel()
+				return
 			case <-ticker.C:
 				if config.IsShouldShutdown() {
 					cancel()

backend/internal/features/restores/controller_test.go

@@ -1,6 +1,7 @@
 package restores
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -340,7 +341,7 @@ func createTestBackup(
 	dummyContent := []byte("dummy backup content for testing")
 	reader := strings.NewReader(string(dummyContent))
 	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
-	if err := storages[0].SaveFile(fieldEncryptor, logger, backup.ID, reader); err != nil {
+	if err := storages[0].SaveFile(context.Background(), fieldEncryptor, logger, backup.ID, reader); err != nil {
 		panic(fmt.Sprintf("Failed to create test backup file: %v", err))
 	}
 

backend/internal/features/storages/interfaces.go

@@ -1,6 +1,7 @@
 package storages
 
 import (
+	"context"
 	"io"
 	"log/slog"
 	"postgresus-backend/internal/util/encryption"
@@ -10,6 +11,7 @@ import (
 
 type StorageFileSaver interface {
 	SaveFile(
+		ctx context.Context,
 		encryptor encryption.FieldEncryptor,
 		logger *slog.Logger,
 		fileID uuid.UUID,

backend/internal/features/storages/model.go

@@ -1,6 +1,7 @@
 package storages
 
 import (
+	"context"
 	"errors"
 	"io"
 	"log/slog"
@@ -30,12 +31,13 @@ type Storage struct {
 }
 
 func (s *Storage) SaveFile(
+	ctx context.Context,
 	encryptor encryption.FieldEncryptor,
 	logger *slog.Logger,
 	fileID uuid.UUID,
 	file io.Reader,
 ) error {
-	err := s.getSpecificStorage().SaveFile(encryptor, logger, fileID, file)
+	err := s.getSpecificStorage().SaveFile(ctx, encryptor, logger, fileID, file)
 	if err != nil {
 		lastSaveError := err.Error()
 		s.LastSaveError = &lastSaveError

backend/internal/features/storages/model_test.go

@@ -167,6 +167,7 @@ func Test_Storage_BasicOperations(t *testing.T) {
 				fileID := uuid.New()
 
 				err = tc.storage.SaveFile(
+					context.Background(),
 					encryptor,
 					logger.GetLogger(),
 					fileID,
@@ -189,6 +190,7 @@ func Test_Storage_BasicOperations(t *testing.T) {
 
 				fileID := uuid.New()
 				err = tc.storage.SaveFile(
+					context.Background(),
 					encryptor,
 					logger.GetLogger(),
 					fileID,
@@ -238,7 +240,7 @@ func setupS3Container(ctx context.Context) (*S3Container, error) {
 	secretKey := "testpassword"
 	bucketName := "test-bucket"
 	region := "us-east-1"
-	endpoint := fmt.Sprintf("localhost:%s", env.TestMinioPort)
+	endpoint := fmt.Sprintf("127.0.0.1:%s", env.TestMinioPort)
 
 	// Create MinIO client and ensure bucket exists
 	minioClient, err := minio.New(endpoint, &minio.Options{