stale lrps claimed state is retained

Author: kart2bc

controllers/lrp_convergence_controller.go

@@ -179,6 +179,16 @@ func (h *LRPConvergenceController) ConvergeLRPs(ctx context.Context) {
 	for _, lrpKey := range convergenceResult.UnstartedLRPKeys {
 		dereferencedKey := *lrpKey
 		works = append(works, func() {
+			actualLRP, err := h.lrpDB.GetActualLRP(ctx, logger, dereferencedKey.Key)
+			if err != nil {
+				logger.Error("cannot-find-actual-lrp", err, lager.Data{"key": dereferencedKey})
+				return
+			}
+			// no action taken if an unstarted lrp is in claimed state
+			if actualLRP.State == models.ActualLRPStateClaimed {
+				return
+			}
+
 			before, after, err := h.lrpDB.UnclaimActualLRP(ctx, logger, dereferencedKey.Key)
 			if err != nil && err != models.ErrActualLRPCannotBeUnclaimed {
 				logger.Error("cannot-unclaim-lrp", err, lager.Data{"key": dereferencedKey})

db/actual_lrp_db.go

@@ -14,6 +14,7 @@ type ActualLRPDB interface {
 	ActualLRPsByProcessGuids(ctx context.Context, logger lager.Logger, filter models.ActualLRPsByProcessGuidsFilter) ([]*models.ActualLRP, error)
 	CreateUnclaimedActualLRP(ctx context.Context, logger lager.Logger, key *models.ActualLRPKey) (after *models.ActualLRP, err error)
 	UnclaimActualLRP(ctx context.Context, logger lager.Logger, key *models.ActualLRPKey) (before *models.ActualLRP, after *models.ActualLRP, err error)
+	GetActualLRP(ctx context.Context, logger lager.Logger, key *models.ActualLRPKey) (actualLRP *models.ActualLRP, err error)
 	ClaimActualLRP(ctx context.Context, logger lager.Logger, processGuid string, index int32, instanceKey *models.ActualLRPInstanceKey) (before *models.ActualLRP, after *models.ActualLRP, err error)
 	StartActualLRP(ctx context.Context,
 		logger lager.Logger,

db/sqldb/actual_lrp_db.go

@@ -212,7 +212,7 @@ func (db *SQLDB) UnclaimActualLRP(ctx context.Context, logger lager.Logger, key
 		}
 		beforeActualLRP = *actualLRP
 
-		if actualLRP.State == models.ActualLRPStateUnclaimed || actualLRP.State == models.ActualLRPStateClaimed {
+		if actualLRP.State == models.ActualLRPStateUnclaimed {
 			logger.Debug("already-" + actualLRP.State)
 			return models.ErrActualLRPCannotBeUnclaimed
 		}
@@ -253,6 +253,27 @@ func (db *SQLDB) UnclaimActualLRP(ctx context.Context, logger lager.Logger, key
 	return &beforeActualLRP, actualLRP, err
 }
 
+func (db *SQLDB) GetActualLRP(ctx context.Context, logger lager.Logger, key *models.ActualLRPKey) (*models.ActualLRP, error) {
+	logger = logger.Session("db-get-actual-lrp", lager.Data{"key": key})
+	logger.Info("starting")
+	defer logger.Info("complete")
+
+	var actualLRP *models.ActualLRP
+	processGuid := key.ProcessGuid
+	index := key.Index
+
+	err := db.transact(ctx, logger, func(logger lager.Logger, tx helpers.Tx) error {
+		var err error
+		actualLRP, err = db.fetchActualLRPForUpdate(ctx, logger, processGuid, index, models.ActualLRP_Ordinary, tx)
+		if err != nil {
+			logger.Error("failed-fetching-actual-lrp-for-share", err)
+			return err
+		}
+		return nil
+	})
+	return actualLRP, err
+}
+
 func (db *SQLDB) ClaimActualLRP(ctx context.Context, logger lager.Logger, processGuid string, index int32, instanceKey *models.ActualLRPInstanceKey) (*models.ActualLRP, *models.ActualLRP, error) {
 	logger = logger.Session("db-claim-actual-lrp", lager.Data{"process_guid": processGuid, "index": index, "instance_key": instanceKey})
 	logger.Info("starting")