fix: server lifecycle and backup data integrity edge cases

Author: ChecksumDev

src/jobs/backup_job.go

@@ -208,6 +208,11 @@ func (j *BackupCreateJob) Execute(ctx context.Context, reporter ProgressReporter
 		return nil, errors.New("server not found: " + j.serverID)
 	}
 
+	j.serverManager.AcquireBackupLock(j.serverID)
+	defer j.serverManager.ReleaseBackupLock(j.serverID)
+
+	logger.Debug("acquired backup operation lock for server")
+
 	reporter.ReportProgress(20, "Initializing backup adapter...")
 
 	var adapter backup.BackupInterface
@@ -361,6 +366,10 @@ func (j *BackupDeleteJob) Execute(ctx context.Context, reporter ProgressReporter
 		return nil, errors.New("server not found: " + j.serverID)
 	}
 
+	j.serverManager.AcquireBackupLock(j.serverID)
+	defer j.serverManager.ReleaseBackupLock(j.serverID)
+
+	logger.Debug("acquired backup operation lock for server")
 	logger.Info("executing backup delete job")
 
 	switch backup.AdapterType(j.adapterType) {
@@ -698,6 +707,11 @@ func (j *BackupRestoreJob) Execute(ctx context.Context, reporter ProgressReporte
 		return nil, errors.New("server not found: " + j.serverID)
 	}
 
+	j.serverManager.AcquireBackupLock(j.serverID)
+	defer j.serverManager.ReleaseBackupLock(j.serverID)
+
+	logger.Debug("acquired backup operation lock for server")
+
 	s.SetRestoring(true)
 	defer s.SetRestoring(false)
 

src/router/router_server.go

@@ -5,15 +5,18 @@ import (
 	"net/http"
 	"os"
 	"strconv"
+	"time"
 
 	"emperror.dev/errors"
 	"github.com/apex/log"
 	"github.com/gin-gonic/gin"
 
+	"github.com/pyrohost/elytra/src/remote"
 	"github.com/pyrohost/elytra/src/router/downloader"
 	"github.com/pyrohost/elytra/src/router/middleware"
 	"github.com/pyrohost/elytra/src/router/tokens"
 	"github.com/pyrohost/elytra/src/server"
+	"github.com/pyrohost/elytra/src/server/backup"
 	"github.com/pyrohost/elytra/src/server/transfer"
 )
 
@@ -221,6 +224,10 @@ func deleteServer(c *gin.Context) {
 		return
 	}
 
+	// Destroy backup repositories to prevent orphaned data on S3 and local storage.
+	// This runs in the background since repository destruction can take time for large repositories.
+	go destroyServerBackupRepositories(c.Request.Context(), s, middleware.ExtractManager(c).Client())
+
 	// Once the environment is terminated, remove the server files from the system. This is
 	// done in a separate process since failure is not the end of the world and can be
 	// manually cleaned up after the fact.
@@ -243,6 +250,76 @@ func deleteServer(c *gin.Context) {
 	c.Status(http.StatusNoContent)
 }
 
+// destroyServerBackupRepositories destroys all backup repositories (local and S3) for a server.
+func destroyServerBackupRepositories(ctx context.Context, s *server.Server, client remote.Client) {
+	logger := log.WithField("server_id", s.ID())
+
+	repositories := []struct {
+		name       string
+		backupType string
+	}{
+		{"local", "local"},
+		{"s3", "s3"},
+	}
+
+	for _, repoInfo := range repositories {
+		func() {
+			rusticConfig, err := client.GetServerRusticConfig(ctx, s.ID(), repoInfo.backupType)
+			if err != nil {
+				logger.WithFields(log.Fields{
+					"type":  repoInfo.backupType,
+					"error": err,
+				}).Warn("failed to get rustic config for repository destruction, repository may not exist")
+				return
+			}
+
+			var repository backup.Repository
+			var repoErr error
+
+			config := backup.Config{
+				ServerUUID: s.ID(),
+				Password:   rusticConfig.RepositoryPassword,
+				BackupType: repoInfo.backupType,
+			}
+
+			if repoInfo.backupType == "local" {
+				config.LocalPath = rusticConfig.RepositoryPath
+				repository, repoErr = backup.NewLocalRepository(config)
+			} else {
+				config.S3Config = &backup.S3Config{
+					Endpoint:        rusticConfig.S3Credentials.Endpoint,
+					Bucket:          rusticConfig.S3Credentials.Bucket,
+					AccessKeyID:     rusticConfig.S3Credentials.AccessKeyID,
+					SecretAccessKey: rusticConfig.S3Credentials.SecretAccessKey,
+					Region:          rusticConfig.S3Credentials.Region,
+				}
+				config.LocalPath = rusticConfig.RepositoryPath
+				repository, repoErr = backup.NewS3Repository(config)
+			}
+
+			if repoErr != nil {
+				logger.WithFields(log.Fields{
+					"type":  repoInfo.backupType,
+					"error": repoErr,
+				}).Error("failed to create repository instance for destruction")
+				return
+			}
+
+			destroyCtx, cancel := context.WithTimeout(ctx, 10*time.Minute)
+			defer cancel()
+
+			if err := repository.Destroy(destroyCtx); err != nil {
+				logger.WithFields(log.Fields{
+					"type":  repoInfo.backupType,
+					"error": err,
+				}).Error("failed to destroy backup repository")
+			} else {
+				logger.WithField("type", repoInfo.backupType).Info("successfully destroyed backup repository")
+			}
+		}()
+	}
+}
+
 // Adds any of the JTIs passed through in the body to the deny list for the websocket
 // preventing any JWT generated before the current time from being used to connect to
 // the socket or send along commands.

src/server/backup/rustic_local.go

@@ -280,7 +280,9 @@ func (r *LocalRepository) DeleteSnapshot(ctx context.Context, id string) error {
 	deleteCtx, cancel := context.WithTimeout(ctx, 60*time.Second)
 	defer cancel()
 
-	cmd, err := r.buildCommand(deleteCtx, "forget", "--prune", id)
+	// Use forget with --prune and --keep-delete to safely reclaim storage space
+	// The 1-day delay ensures any parallel backups have time to complete and reference shared blobs
+	cmd, err := r.buildCommand(deleteCtx, "forget", "--prune", "--keep-delete", "1d", id)
 	if err != nil {
 		return errors.Wrap(err, "failed to build delete command")
 	}

src/server/backup/rustic_s3.go

@@ -526,9 +526,9 @@ func (r *S3Repository) deleteSnapshotWithTimeout(ctx context.Context, id string,
 	deleteCtx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
-	// Use forget with --prune and --instant-delete to immediately reclaim storage space
-	// Note: --instant-delete is safe here because Elytra controls all backup operations
-	cmd, err := r.buildCommand(deleteCtx, "forget", "--prune", "--instant-delete", id)
+	// Use forget with --prune and --keep-delete to safely reclaim storage space
+	// The 1-day delay ensures any parallel backups have time to complete and reference shared blobs
+	cmd, err := r.buildCommand(deleteCtx, "forget", "--prune", "--keep-delete", "1d", id)
 	if err != nil {
 		return errors.Wrap(err, "failed to build delete command")
 	}
@@ -707,8 +707,14 @@ func (r *S3Repository) RestoreSnapshot(ctx context.Context, snapshotID string, t
 		return errors.Wrap(err, "failed to build restore command")
 	}
 
-	if err := cmd.Run(); err != nil {
-		return errors.Wrap(err, "S3 restore failed")
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		outputStr := string(output)
+		r.logger.WithFields(log.Fields{
+			"snapshot_id": snapshotID,
+			"output":      outputStr,
+		}).Error("rustic restore command failed")
+		return errors.Wrapf(err, "S3 restore failed: %s", outputStr)
 	}
 
 	return nil

src/server/manager.go

@@ -23,9 +23,11 @@ import (
 )
 
 type Manager struct {
-	mu      sync.RWMutex
-	client  remote.Client
-	servers []*Server
+	mu                 sync.RWMutex
+	client             remote.Client
+	servers            []*Server
+	backupOperationsMu sync.Mutex
+	backupOperations   map[string]*sync.Mutex
 }
 
 // NewManager returns a new server manager instance. This will boot up all the
@@ -43,7 +45,10 @@ func NewManager(ctx context.Context, client remote.Client) (*Manager, error) {
 // loading any of the servers from the disk. This allows the caller to set their
 // own servers into the collection as needed.
 func NewEmptyManager(client remote.Client) *Manager {
-	return &Manager{client: client}
+	return &Manager{
+		client:           client,
+		backupOperations: make(map[string]*sync.Mutex),
+	}
 }
 
 // Client returns the HTTP client interface that allows interaction with the
@@ -131,13 +136,27 @@ func (m *Manager) Find(filter func(match *Server) bool) *Server {
 func (m *Manager) Remove(filter func(match *Server) bool) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+
+	// Collect IDs of servers being removed for lock cleanup
+	removedIDs := make([]string, 0)
 	r := make([]*Server, 0)
 	for _, v := range m.servers {
-		if !filter(v) {
+		if filter(v) {
+			removedIDs = append(removedIDs, v.ID())
+		} else {
 			r = append(r, v)
 		}
 	}
 	m.servers = r
+
+	// Clean up backup operation locks for removed servers
+	if len(removedIDs) > 0 {
+		m.backupOperationsMu.Lock()
+		for _, id := range removedIDs {
+			delete(m.backupOperations, id)
+		}
+		m.backupOperationsMu.Unlock()
+	}
 }
 
 // PersistStates writes the current environment states to the disk for each
@@ -280,3 +299,26 @@ func (m *Manager) init(ctx context.Context) error {
 
 	return nil
 }
+
+// AcquireBackupLock acquires a per-server backup operation lock to prevent concurrent backup operations.
+func (m *Manager) AcquireBackupLock(serverID string) {
+	m.backupOperationsMu.Lock()
+	if _, exists := m.backupOperations[serverID]; !exists {
+		m.backupOperations[serverID] = &sync.Mutex{}
+	}
+	mu := m.backupOperations[serverID]
+	m.backupOperationsMu.Unlock()
+
+	mu.Lock()
+}
+
+// ReleaseBackupLock releases a per-server backup operation lock.
+func (m *Manager) ReleaseBackupLock(serverID string) {
+	m.backupOperationsMu.Lock()
+	mu, exists := m.backupOperations[serverID]
+	m.backupOperationsMu.Unlock()
+
+	if exists {
+		mu.Unlock()
+	}
+}