diff --git a/backend/internal/handler/modeldownload.go b/backend/internal/handler/modeldownload.go index 793162d93..870e0ae91 100644 --- a/backend/internal/handler/modeldownload.go +++ b/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)) } diff --git a/backend/internal/handler/modeldownload_test.go b/backend/internal/handler/modeldownload_test.go index dda0919f2..284ed1920 100644 --- a/backend/internal/handler/modeldownload_test.go +++ b/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 diff --git a/backend/pkg/reconciler/modeldownload-reconciler.go b/backend/pkg/reconciler/modeldownload-reconciler.go index 586211977..1be505735 100644 --- a/backend/pkg/reconciler/modeldownload-reconciler.go +++ b/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. diff --git a/backend/pkg/reconciler/modeldownload-reconciler_test.go b/backend/pkg/reconciler/modeldownload-reconciler_test.go index 7cb977da5..61584a42d 100644 --- a/backend/pkg/reconciler/modeldownload-reconciler_test.go +++ b/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"]}