fix(feature): persist failed status when scheduled flag change execution fails (#2709)

Author: cre8ivejp

pkg/feature/api/scheduled_feature_change.go

@@ -22,7 +22,10 @@ import (
 	"time"
 
 	"go.uber.org/zap"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 
+	"github.com/bucketeer-io/bucketeer/v2/pkg/api/api"
 	domainevent "github.com/bucketeer-io/bucketeer/v2/pkg/domainevent/domain"
 	"github.com/bucketeer-io/bucketeer/v2/pkg/feature/domain"
 	"github.com/bucketeer-io/bucketeer/v2/pkg/feature/scheduled"
@@ -645,6 +648,13 @@ func (s *FeatureService) ExecuteScheduledFlagChange(
 
 	var sfc *domain.ScheduledFlagChange
 	var event *eventproto.Event
+	// Failure reason for permanent (non-retryable) failures. It must be
+	// persisted OUTSIDE the transaction: returning an error from the
+	// transaction callback rolls back everything, including any status
+	// update written inside it. Persisting FAILED after the rollback
+	// prevents the batch executor from retrying the same broken schedule
+	// forever.
+	var failureReason string
 
 	err = s.dbClient.RunInTransactionV2(ctx, func(ctxWithTx context.Context) error {
 		var err error
@@ -673,17 +683,26 @@ func (s *FeatureService) ExecuteScheduledFlagChange(
 		feature, err := s.featureStorage.GetFeature(ctxWithTx, sfc.FeatureId, req.EnvironmentId)
 		if err != nil {
 			if errors.Is(err, v2fs.ErrFeatureNotFound) {
-				sfc.MarkFailed("Feature not found")
-				_ = s.scheduledFlagChangeStorage.UpdateScheduledFlagChange(ctxWithTx, sfc)
+				failureReason = "Feature not found"
 				return statusFeatureNotFound.Err()
 			}
-			return err
+			s.logger.Error(
+				"Failed to get feature for scheduled change execution",
+				log.FieldsFromIncomingContext(ctx).AddFields(
+					zap.Error(err),
+					zap.String("id", req.Id),
+					zap.String("featureId", sfc.FeatureId),
+					zap.String("environmentId", req.EnvironmentId),
+				)...,
+			)
+			return statusInternal.Err()
 		}
 
 		// Validate references still exist
 		if err := s.validateScheduledChangePayload(ctxWithTx, sfc.Payload, feature.Feature, req.EnvironmentId); err != nil {
-			sfc.MarkFailed(err.Error())
-			_ = s.scheduledFlagChangeStorage.UpdateScheduledFlagChange(ctxWithTx, sfc)
+			if isPermanentScheduledChangeError(err) {
+				failureReason = err.Error()
+			}
 			return err
 		}
 
@@ -693,8 +712,9 @@ func (s *FeatureService) ExecuteScheduledFlagChange(
 
 		event, _, err = s.updateFeatureWithinTransaction(ctxWithTx, editor, updateReq)
 		if err != nil {
-			sfc.MarkFailed(err.Error())
-			_ = s.scheduledFlagChangeStorage.UpdateScheduledFlagChange(ctxWithTx, sfc)
+			if isPermanentScheduledChangeError(err) {
+				failureReason = err.Error()
+			}
 			return err
 		}
 
@@ -705,6 +725,9 @@ func (s *FeatureService) ExecuteScheduledFlagChange(
 	})
 
 	if err != nil {
+		if failureReason != "" && sfc != nil {
+			s.markScheduledFlagChangeFailed(ctx, sfc, failureReason, req.EnvironmentId)
+		}
 		s.logger.Error(
 			"Failed to execute scheduled flag change",
 			log.FieldsFromIncomingContext(ctx).AddFields(
@@ -734,6 +757,64 @@ func (s *FeatureService) ExecuteScheduledFlagChange(
 	}, nil
 }
 
+// isPermanentScheduledChangeError reports whether an execution error is
+// permanent (caused by the schedule's payload or the flag's current state)
+// as opposed to transient (storage/infrastructure). Only permanent errors
+// should mark the schedule FAILED; transient errors leave it PENDING so the
+// batch executor retries it.
+func isPermanentScheduledChangeError(err error) bool {
+	st, ok := status.FromError(err)
+	if !ok {
+		// Not a gRPC status error: domain validation errors (pkg/error
+		// BktError) are mapped to their equivalent gRPC code; raw storage
+		// errors map to Unknown and are treated as transient.
+		st = api.NewGRPCStatus(err)
+	}
+	switch st.Code() {
+	case codes.InvalidArgument,
+		codes.NotFound,
+		codes.AlreadyExists,
+		codes.FailedPrecondition,
+		codes.OutOfRange:
+		return true
+	default:
+		return false
+	}
+}
+
+// markScheduledFlagChangeFailed persists the FAILED status in its own write,
+// outside any (rolled back) transaction, so the schedule is not retried by
+// the batch executor. It uses a detached context so the write still succeeds
+// when the caller's context is already cancelled or past its deadline
+// (e.g. the batch executor's timeout).
+func (s *FeatureService) markScheduledFlagChangeFailed(
+	ctx context.Context,
+	sfc *domain.ScheduledFlagChange,
+	reason, environmentID string,
+) {
+	// WithoutCancel detaches cancellation/deadline (so the write succeeds
+	// even if the caller's context is already cancelled) while preserving
+	// request-scoped values such as trace IDs. The caller's context never
+	// carries a transaction: RunInTransactionV2 only injects it into the
+	// callback's context, so this write always goes to the pool.
+	writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
+	defer cancel()
+
+	sfc.MarkFailed(reason)
+	if err := s.scheduledFlagChangeStorage.UpdateScheduledFlagChange(writeCtx, sfc); err != nil {
+		s.logger.Error(
+			"Failed to mark scheduled flag change as failed",
+			log.FieldsFromIncomingContext(ctx).AddFields(
+				zap.Error(err),
+				zap.String("id", sfc.Id),
+				zap.String("featureId", sfc.FeatureId),
+				zap.String("environmentId", environmentID),
+				zap.String("failureReason", reason),
+			)...,
+		)
+	}
+}
+
 func (s *FeatureService) GetScheduledFlagChangeSummary(
 	ctx context.Context,
 	req *ftproto.GetScheduledFlagChangeSummaryRequest,
@@ -928,7 +1009,13 @@ func (s *FeatureService) validateScheduledChangePayload(
 			}
 			prereqFeature, err := s.featureStorage.GetFeature(ctx, pc.Prerequisite.FeatureId, environmentID)
 			if err != nil {
-				return statusInvalidPrerequisiteReference.Err()
+				if errors.Is(err, v2fs.ErrFeatureNotFound) {
+					return statusInvalidPrerequisiteReference.Err()
+				}
+				// Transient storage error: don't misreport it as an invalid
+				// reference, or execution would permanently mark the schedule
+				// FAILED instead of retrying.
+				return statusInternal.Err()
 			}
 			if !domain.VariationExists(prereqFeature.Feature, pc.Prerequisite.VariationId) {
 				return statusInvalidPrerequisiteReference.Err()

pkg/feature/api/scheduled_feature_change_test.go

@@ -16,6 +16,7 @@ package api
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -415,6 +416,159 @@ func TestDeleteScheduledFlagChange_Success(t *testing.T) {
 	assert.NotNil(t, resp)
 }
 
+func TestExecuteScheduledFlagChange_ValidationFailureMarksFailedOutsideTransaction(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	service := createFeatureServiceWithGetAccountByEnvironmentMock(
+		ctrl,
+		accountproto.AccountV2_Role_Organization_MEMBER,
+		accountproto.AccountV2_Role_Environment_EDITOR,
+	)
+
+	dbClient := service.dbClient.(*databasemock.MockClient)
+	featureStorage := service.featureStorage.(*mock.MockFeatureStorage)
+	scheduledStorage := service.scheduledFlagChangeStorage.(*mock.MockScheduledFlagChangeStorage)
+
+	feature := &domain.Feature{
+		Feature: &featureproto.Feature{
+			Id:      "feature-id",
+			Name:    "Test Feature",
+			Version: 1,
+			Variations: []*featureproto.Variation{
+				{Id: "var-1", Name: "Variation 1", Value: "true"},
+			},
+			// No rules: the scheduled payload references a rule that no longer exists
+		},
+	}
+
+	sfc := &featureproto.ScheduledFlagChange{
+		Id:            "sfc-id",
+		FeatureId:     "feature-id",
+		EnvironmentId: "ns0",
+		ScheduledAt:   time.Now().Add(-time.Minute).Unix(),
+		Status:        featureproto.ScheduledFlagChangeStatus_SCHEDULED_FLAG_CHANGE_STATUS_PENDING,
+		Payload: &featureproto.ScheduledChangePayload{
+			RuleChanges: []*featureproto.RuleChange{
+				{
+					ChangeType: featureproto.ChangeType_UPDATE,
+					Rule:       &featureproto.Rule{Id: "deleted-rule-id"},
+				},
+			},
+		},
+	}
+
+	// Simulate real transaction semantics: the callback error propagates
+	// and everything written inside the transaction is rolled back.
+	// A marker is added to the transaction context so we can verify the
+	// FAILED write does NOT happen on it (i.e. not inside the transaction).
+	type txCtxKey struct{}
+	var ctxWithTx context.Context
+	dbClient.EXPECT().RunInTransactionV2(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, f func(context.Context) error) error {
+			ctxWithTx = context.WithValue(ctx, txCtxKey{}, true)
+			return f(ctxWithTx)
+		},
+	)
+	scheduledStorage.EXPECT().GetScheduledFlagChange(gomock.Any(), "sfc-id", "ns0").
+		Return(&domain.ScheduledFlagChange{ScheduledFlagChange: sfc}, nil)
+	featureStorage.EXPECT().GetFeature(gomock.Any(), "feature-id", "ns0").Return(feature, nil)
+
+	// The FAILED status must be persisted after (outside) the failed transaction,
+	// otherwise the rollback would discard it and the batch executor would
+	// retry the same broken schedule forever.
+	scheduledStorage.EXPECT().UpdateScheduledFlagChange(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(updateCtx context.Context, updated *domain.ScheduledFlagChange) error {
+			// The write must not use the transaction context, or it would
+			// be rolled back together with the failed transaction.
+			require.NotNil(t, ctxWithTx)
+			assert.Nil(t, updateCtx.Value(txCtxKey{}))
+			assert.Equal(
+				t,
+				featureproto.ScheduledFlagChangeStatus_SCHEDULED_FLAG_CHANGE_STATUS_FAILED,
+				updated.Status,
+			)
+			assert.NotEmpty(t, updated.FailureReason)
+			return nil
+		})
+
+	ctx := createContextWithToken()
+	_, err := service.ExecuteScheduledFlagChange(ctx, &featureproto.ExecuteScheduledFlagChangeRequest{
+		EnvironmentId: "ns0",
+		Id:            "sfc-id",
+	})
+	assert.Equal(t, statusInvalidRuleReference.Err(), err)
+}
+
+func TestExecuteScheduledFlagChange_TransientErrorKeepsSchedulePending(t *testing.T) {
+	t.Parallel()
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	service := createFeatureServiceWithGetAccountByEnvironmentMock(
+		ctrl,
+		accountproto.AccountV2_Role_Organization_MEMBER,
+		accountproto.AccountV2_Role_Environment_EDITOR,
+	)
+
+	dbClient := service.dbClient.(*databasemock.MockClient)
+	featureStorage := service.featureStorage.(*mock.MockFeatureStorage)
+	scheduledStorage := service.scheduledFlagChangeStorage.(*mock.MockScheduledFlagChangeStorage)
+
+	feature := &domain.Feature{
+		Feature: &featureproto.Feature{
+			Id:      "feature-id",
+			Name:    "Test Feature",
+			Version: 1,
+			Variations: []*featureproto.Variation{
+				{Id: "var-1", Name: "Variation 1", Value: "true"},
+			},
+		},
+	}
+
+	sfc := &featureproto.ScheduledFlagChange{
+		Id:            "sfc-id",
+		FeatureId:     "feature-id",
+		EnvironmentId: "ns0",
+		ScheduledAt:   time.Now().Add(-time.Minute).Unix(),
+		Status:        featureproto.ScheduledFlagChangeStatus_SCHEDULED_FLAG_CHANGE_STATUS_PENDING,
+		Payload: &featureproto.ScheduledChangePayload{
+			PrerequisiteChanges: []*featureproto.PrerequisiteChange{
+				{
+					ChangeType: featureproto.ChangeType_CREATE,
+					Prerequisite: &featureproto.Prerequisite{
+						FeatureId:   "prereq-feature-id",
+						VariationId: "prereq-var-1",
+					},
+				},
+			},
+		},
+	}
+
+	dbClient.EXPECT().RunInTransactionV2(gomock.Any(), gomock.Any()).DoAndReturn(
+		func(ctx context.Context, f func(context.Context) error) error {
+			return f(ctx)
+		},
+	)
+	scheduledStorage.EXPECT().GetScheduledFlagChange(gomock.Any(), "sfc-id", "ns0").
+		Return(&domain.ScheduledFlagChange{ScheduledFlagChange: sfc}, nil)
+	featureStorage.EXPECT().GetFeature(gomock.Any(), "feature-id", "ns0").Return(feature, nil)
+	// The prerequisite lookup fails with a transient storage error (not "not found").
+	featureStorage.EXPECT().GetFeature(gomock.Any(), "prereq-feature-id", "ns0").
+		Return(nil, errors.New("db connection lost"))
+
+	// No UpdateScheduledFlagChange expectation: a transient error must NOT
+	// mark the schedule FAILED, so it stays PENDING and is retried later.
+
+	ctx := createContextWithToken()
+	_, err := service.ExecuteScheduledFlagChange(ctx, &featureproto.ExecuteScheduledFlagChangeRequest{
+		EnvironmentId: "ns0",
+		Id:            "sfc-id",
+	})
+	assert.Equal(t, statusInternal.Err(), err)
+}
+
 func TestGetScheduledFlagChangeSummary_Success(t *testing.T) {
 	t.Parallel()
 	ctrl := gomock.NewController(t)