fix(model): make model download pause race-safe

backend/internal/handler/modeldownload.go

@@ -375,6 +375,25 @@ func updateDownloadAndReleaseQuota(
 	})
 }
 
+func pauseDownloadAndReleaseQuota(ctx context.Context, downloadID uint) error {
+	db := query.ModelDownload.WithContext(ctx).UnderlyingDB().Session(&gorm.Session{NewDB: true})
+	return db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		result := tx.Model(&model.ModelDownload{}).
+			Where("id = ? AND status = ?", downloadID, model.ModelDownloadStatusDownloading).
+			Updates(map[string]any{
+				"status":  model.ModelDownloadStatusPaused,
+				"message": "Download paused by user",
+			})
+		if result.Error != nil {
+			return result.Error
+		}
+		if result.RowsAffected == 0 {
+			return bizerr.Conflict.ResourceStatusError.New("only downloading tasks can be paused")
+		}
+		return service.ReleaseModelDownloadQuotaReservation(ctx, tx, downloadID)
+	})
+}
+
 // checkLogicalDownloadConflict blocks a new source/revision when a historical
 // failed or soft-deleted record already owns storage for the same public model.
 // Ready and ongoing records are handled earlier and reused instead.
@@ -1116,24 +1135,26 @@ func (mgr *ModelDownloadMgr) PauseDownload(c *gin.Context) {
 		return
 	}
 
-	// Pausing is implemented by deleting the Job. Persist logs before removal.
+	// Persist the user's pause intent before deleting the Job. Kubernetes deletion
+	// is asynchronous, so the reconciler must be able to observe Paused before it
+	// handles the Job deletion event.
 	mgr.captureJobLogsToRecord(c, download)
-	if err := mgr.deleteDownloadJob(c, download.JobName); err != nil {
-		resputil.HandleError(c, err)
-		return
-	}
-
-	updates := map[string]any{
-		"status":  model.ModelDownloadStatusPaused,
-		"message": "Download paused by user",
-	}
-	if err := updateDownloadAndReleaseQuota(c, download.ID, updates); err != nil {
+	if err := pauseDownloadAndReleaseQuota(c, download.ID); err != nil {
+		if errors.Is(err, bizerr.Conflict.Base) {
+			resputil.HandleError(c, err)
+			return
+		}
 		resputil.HandleError(c, bizerr.Internal.DatabaseError.Wrap(err, "update paused download failed"))
 		return
 	}
 
 	download.Status = model.ModelDownloadStatusPaused
 	download.Message = "Download paused by user"
+	if err := mgr.deleteDownloadJob(c, download.JobName); err != nil {
+		// Paused is the durable desired state. Keep it instead of rolling back to
+		// Downloading; the reconciler will retry the idempotent Job cleanup.
+		klog.Warningf("download %d was paused but Job %q cleanup failed: %v", download.ID, download.JobName, err)
+	}
 
 	resputil.Success(c, convertDownloadToResp(download, token))
 }

backend/internal/handler/modeldownload_test.go

@@ -196,6 +196,57 @@ func TestRetryUpdateDuplicateIsConflict(t *testing.T) {
 	}
 }
 
+func TestPauseDownloadTransitionIsConditionalAndReleasesQuota(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open("file:pause_download_transition?mode=memory&cache=shared"), &gorm.Config{
+		DisableForeignKeyConstraintWhenMigrating: true,
+		IgnoreRelationshipsWhenMigrating:         true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&model.ModelDownload{}, &model.ModelDownloadSubmission{}); err != nil {
+		t.Fatal(err)
+	}
+	query.SetDefault(db)
+
+	download := model.ModelDownload{
+		Name: "owner/pause", Source: model.ModelSourceModelScope,
+		Category: model.DownloadCategoryModel, Revision: "main", Path: "public/Models/owner/pause",
+		Status: model.ModelDownloadStatusDownloading, JobName: "pause-job", CreatorID: 7,
+	}
+	if err := db.Create(&download).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&model.ModelDownloadSubmission{
+		UserID: 7, ModelDownloadID: download.ID,
+		Action: model.ModelDownloadSubmissionCreate, Status: model.ModelDownloadSubmissionReserved,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	if err := pauseDownloadAndReleaseQuota(t.Context(), download.ID); err != nil {
+		t.Fatal(err)
+	}
+	var stored model.ModelDownload
+	if err := db.First(&stored, download.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if stored.Status != model.ModelDownloadStatusPaused || stored.Message != "Download paused by user" {
+		t.Fatalf("unexpected paused download: %#v", stored)
+	}
+	var submission model.ModelDownloadSubmission
+	if err := db.Where("model_download_id = ?", download.ID).First(&submission).Error; err != nil {
+		t.Fatal(err)
+	}
+	if submission.Status != model.ModelDownloadSubmissionReleased || submission.CompletedAt != nil {
+		t.Fatalf("pause did not release quota reservation: %#v", submission)
+	}
+
+	if err := pauseDownloadAndReleaseQuota(t.Context(), download.ID); !errors.Is(err, bizerr.Conflict.Base) {
+		t.Fatalf("second pause error = %v, want status conflict", err)
+	}
+}
+
 func TestRestoredReadyDownloadDoesNotSubmitJob(t *testing.T) {
 	for _, testCase := range []struct {
 		status model.ModelDownloadStatus

backend/pkg/reconciler/modeldownload-reconciler.go

@@ -165,6 +165,18 @@ func (r *ModelDownloadReconciler) fetchDownloadRecord(
 
 func (r *ModelDownloadReconciler) syncDownloadWithJob(
 	ctx context.Context, job *batchv1.Job, download *model.ModelDownload, logger logr.Logger,
+) (ctrl.Result, error) {
+	// Paused is a user-requested desired state, not a status inferred from the
+	// Kubernetes Job. Never let an already queued Job event move it back to
+	// Downloading or Failed; only converge the runtime by removing the Job.
+	if download.Status == model.ModelDownloadStatusPaused {
+		return r.ensurePausedJobDeleted(ctx, job, logger)
+	}
+	return r.syncActiveDownloadWithJob(ctx, job, download, logger)
+}
+
+func (r *ModelDownloadReconciler) syncActiveDownloadWithJob(
+	ctx context.Context, job *batchv1.Job, download *model.ModelDownload, logger logr.Logger,
 ) (ctrl.Result, error) {
 	oldStatus := download.Status
 	newStatus := r.getJobStatus(job)
@@ -222,6 +234,21 @@ func (r *ModelDownloadReconciler) syncDownloadWithJob(
 	return ctrl.Result{}, nil
 }
 
+func (r *ModelDownloadReconciler) ensurePausedJobDeleted(
+	ctx context.Context, job *batchv1.Job, logger logr.Logger,
+) (ctrl.Result, error) {
+	if !job.DeletionTimestamp.IsZero() {
+		return ctrl.Result{}, nil
+	}
+	if err := r.Delete(ctx, job, client.PropagationPolicy(metav1.DeletePropagationForeground)); err != nil &&
+		!k8serrors.IsNotFound(err) {
+		logger.Error(err, "failed to delete Job for paused download", "jobName", job.Name)
+		return ctrl.Result{}, err
+	}
+	logger.Info("deleted Job for paused download", "jobName", job.Name)
+	return ctrl.Result{}, nil
+}
+
 type repositoryMetadata struct {
 	DisplayName    string   `json:"display_name"`
 	Description    string   `json:"description"`
@@ -398,38 +425,66 @@ func resolveRepositoryLogo(
 func (r *ModelDownloadReconciler) handleJobNotFound(ctx context.Context, jobName string) (ctrl.Result, error) {
 	logger := log.FromContext(ctx)
 
-	q := query.ModelDownload
-	download, err := q.WithContext(ctx).Where(q.JobName.Eq(jobName)).First()
-	if err != nil {
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			// Job和数据库记录都不存在，这是正常的（可能是旧Job或已清理的任务）
-			logger.V(1).Info("Job not found in both k8s and database", "jobName", jobName)
-			return ctrl.Result{}, nil
-		}
-		logger.Error(err, "unable to fetch download record")
-		return ctrl.Result{Requeue: true}, err
-	}
-
-	logger.Info("Job not found in k8s but exists in database", "jobName", jobName, "status", download.Status)
-
-	// If already in terminal state, no update needed
-	if download.Status == model.ModelDownloadStatusReady ||
-		download.Status == model.ModelDownloadStatusFailed ||
-		download.Status == model.ModelDownloadStatusPaused {
+	status, transitioned, err := r.failMissingJobIfActive(ctx, jobName)
+	if errors.Is(err, gorm.ErrRecordNotFound) {
+		logger.V(1).Info("Job not found in both k8s and database", "jobName", jobName)
 		return ctrl.Result{}, nil
 	}
-
-	// Job deleted but status not terminal, mark as failed
-	if err := r.updateDownloadStatus(ctx, download, model.ModelDownloadStatusFailed); err != nil {
-		logger.Error(err, "failed to update download status to failed")
+	if err != nil {
+		logger.Error(err, "failed to reconcile missing download Job", "jobName", jobName)
 		return ctrl.Result{Requeue: true}, err
 	}
-
-	_, _ = q.WithContext(ctx).Where(q.ID.Eq(download.ID)).Update(q.Message, "Job was deleted")
+	logger.Info("Job not found in k8s but exists in database", "jobName", jobName, "status", status,
+		"markedFailed", transitioned)
 
 	return ctrl.Result{}, nil
 }
 
+// failMissingJobIfActive atomically distinguishes an unexpected Job deletion
+// from a concurrent pause. The row is re-read inside the transaction so a
+// stale reconcile event cannot overwrite a newer Paused state.
+func (r *ModelDownloadReconciler) failMissingJobIfActive(
+	ctx context.Context, jobName string,
+) (model.ModelDownloadStatus, bool, error) {
+	var download model.ModelDownload
+	transitioned := false
+	db := query.ModelDownload.WithContext(ctx).UnderlyingDB().Session(&gorm.Session{NewDB: true})
+	err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).
+			Where("job_name = ?", jobName).
+			First(&download).Error; err != nil {
+			return err
+		}
+		if download.Status != model.ModelDownloadStatusPending &&
+			download.Status != model.ModelDownloadStatusDownloading {
+			return nil
+		}
+
+		result := tx.Model(&model.ModelDownload{}).
+			Where("id = ? AND status IN ?", download.ID, []model.ModelDownloadStatus{
+				model.ModelDownloadStatusPending,
+				model.ModelDownloadStatusDownloading,
+			}).
+			Updates(map[string]any{
+				"status":  model.ModelDownloadStatusFailed,
+				"message": "Job was deleted",
+			})
+		if result.Error != nil {
+			return result.Error
+		}
+		if result.RowsAffected == 0 {
+			return nil
+		}
+		if err := service.ReleaseModelDownloadQuotaReservation(ctx, tx, download.ID); err != nil {
+			return err
+		}
+		download.Status = model.ModelDownloadStatusFailed
+		transitioned = true
+		return nil
+	})
+	return download.Status, transitioned, err
+}
+
 func (r *ModelDownloadReconciler) getJobStatus(job *batchv1.Job) model.ModelDownloadStatus {
 	// Prefer terminal Job conditions so that retries (BackoffLimit > 0) do not
 	// flip the record to Failed while attempts are still being made.

backend/pkg/reconciler/modeldownload-reconciler_test.go

@@ -21,11 +21,16 @@ import (
 	"time"
 	"unicode/utf8"
 
+	"github.com/go-logr/logr"
 	"gorm.io/datatypes"
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"
 	batchv1 "k8s.io/api/batch/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	"github.com/raids-lab/crater/dao/model"
 	"github.com/raids-lab/crater/dao/query"
@@ -103,6 +108,112 @@ func TestFinalLogsAreCurrentRejectsProgressSnapshot(t *testing.T) {
 	}
 }
 
+func TestPausedDownloadDeletesLiveJobWithoutChangingState(t *testing.T) {
+	scheme := runtime.NewScheme()
+	if err := batchv1.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Name: "paused-download", Namespace: "jobs"}}
+	reconciler := &ModelDownloadReconciler{
+		Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(job).Build(),
+	}
+	download := &model.ModelDownload{Status: model.ModelDownloadStatusPaused}
+
+	result, err := reconciler.syncDownloadWithJob(t.Context(), job, download, logr.Discard())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.RequeueAfter != 0 {
+		t.Fatalf("syncDownloadWithJob() result = %#v, want no requeue", result)
+	}
+	if download.Status != model.ModelDownloadStatusPaused {
+		t.Fatalf("paused download status changed to %s", download.Status)
+	}
+	var deleted batchv1.Job
+	err = reconciler.Get(t.Context(), types.NamespacedName{Name: job.Name, Namespace: job.Namespace}, &deleted)
+	if !k8serrors.IsNotFound(err) {
+		t.Fatalf("paused download Job still exists or lookup failed: %v", err)
+	}
+}
+
+func newMissingJobTransitionTestDB(t *testing.T) *gorm.DB {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{
+		DisableForeignKeyConstraintWhenMigrating: true,
+		IgnoreRelationshipsWhenMigrating:         true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&model.ModelDownload{}, &model.ModelDownloadSubmission{}); err != nil {
+		t.Fatal(err)
+	}
+	query.SetDefault(db)
+	return db
+}
+
+func TestMissingJobTransitionPreservesPausedDownload(t *testing.T) {
+	db := newMissingJobTransitionTestDB(t)
+	reconciler := &ModelDownloadReconciler{}
+
+	paused := model.ModelDownload{
+		Name: "owner/paused", Source: model.ModelSourceModelScope,
+		Category: model.DownloadCategoryModel, Revision: "main", Path: "public/Models/owner/paused",
+		Status: model.ModelDownloadStatusPaused, Message: "Download paused by user", JobName: "paused-job", CreatorID: 7,
+	}
+	if err := db.Create(&paused).Error; err != nil {
+		t.Fatal(err)
+	}
+	status, transitioned, err := reconciler.failMissingJobIfActive(t.Context(), paused.JobName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if transitioned || status != model.ModelDownloadStatusPaused {
+		t.Fatalf("paused missing Job transition = %t/%s, want false/Paused", transitioned, status)
+	}
+	var storedPaused model.ModelDownload
+	if err := db.First(&storedPaused, paused.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if storedPaused.Status != model.ModelDownloadStatusPaused || storedPaused.Message != "Download paused by user" {
+		t.Fatalf("paused download was overwritten: %#v", storedPaused)
+	}
+}
+
+func TestMissingJobTransitionFailsActiveDownload(t *testing.T) {
+	db := newMissingJobTransitionTestDB(t)
+	reconciler := &ModelDownloadReconciler{}
+	active := model.ModelDownload{
+		Name: "owner/active", Source: model.ModelSourceModelScope,
+		Category: model.DownloadCategoryModel, Revision: "main", Path: "public/Models/owner/active",
+		Status: model.ModelDownloadStatusDownloading, JobName: "active-job", CreatorID: 7,
+	}
+	if err := db.Create(&active).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&model.ModelDownloadSubmission{
+		UserID: 7, ModelDownloadID: active.ID,
+		Action: model.ModelDownloadSubmissionCreate, Status: model.ModelDownloadSubmissionReserved,
+	}).Error; err != nil {
+		t.Fatal(err)
+	}
+	status, transitioned, err := reconciler.failMissingJobIfActive(t.Context(), active.JobName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !transitioned || status != model.ModelDownloadStatusFailed {
+		t.Fatalf("active missing Job transition = %t/%s, want true/Failed", transitioned, status)
+	}
+	var storedActive model.ModelDownload
+	if err := db.First(&storedActive, active.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if storedActive.Status != model.ModelDownloadStatusFailed || storedActive.Message != "Job was deleted" {
+		t.Fatalf("unexpected active missing Job state: %#v", storedActive)
+	}
+	assertQuotaSubmissionSettlement(t, db, active.ID, model.ModelDownloadSubmissionReleased, false)
+}
+
 func TestParseRepositoryMetadata(t *testing.T) {
 	metadata := parseRepositoryMetadata(`noise
 [META] {"downloads":4235273,"likes":333,"updated_at":"2025-07-26T16:12:41Z","tags":["text-generation"]}