Merge remote-tracking branch 'origin/master'

svc/product_series_auto_scheduler.go

@@ -646,12 +646,38 @@ func productSeriesApplyVariant(ctx context.Context, pg *sql.DB, v productSeriesA
 						}
 					}
 					if anyStillValid {
+						_ = productSeriesResetInvalidStreak(ctx, pg, productSeriesAutoKey(v.ProductCode, v.ColorCode, v.Dim3Code))
 						if productSeriesFallbackLogEnabled() {
 							log.Printf("[ProductSeriesFallback] already_exists product=%s color=%s dim3=%s fallback=%s(%d)", strings.TrimSpace(v.ProductCode), strings.TrimSpace(v.ColorCode), strings.TrimSpace(v.Dim3Code), strings.TrimSpace(fallbackCode), fallbackID)
 						}
 						// keep existing manual/previous assignment; nothing to do
 						return 0, 0, nil
 					}
+
+					// Guardrail: if stock snapshots are partial/transient, "existing invalid" can flap.
+					// Require N consecutive invalid observations before overwriting existing mapping.
+					confirmations := envIntRange("PRODUCT_SERIES_EXISTING_INVALID_CONFIRMATIONS", 3, 1, 20)
+					rowKey := productSeriesAutoKey(v.ProductCode, v.ColorCode, v.Dim3Code)
+					streak, err := productSeriesBumpInvalidStreak(ctx, pg, rowKey)
+					if err != nil {
+						return 0, 1, err
+					}
+					if streak < confirmations {
+						if productSeriesFallbackLogEnabled() {
+							log.Printf("[ProductSeriesFallback] existing_invalid_defer product=%s color=%s dim3=%s streak=%d/%d fallback=%s(%d)",
+								strings.TrimSpace(v.ProductCode),
+								strings.TrimSpace(v.ColorCode),
+								strings.TrimSpace(v.Dim3Code),
+								streak,
+								confirmations,
+								strings.TrimSpace(fallbackCode),
+								fallbackID,
+							)
+						}
+						// keep existing mapping for now
+						return 0, 0, nil
+					}
+
 					if productSeriesFallbackLogEnabled() {
 						log.Printf("[ProductSeriesFallback] existing_invalid_apply product=%s color=%s dim3=%s fallback=%s(%d)", strings.TrimSpace(v.ProductCode), strings.TrimSpace(v.ColorCode), strings.TrimSpace(v.Dim3Code), strings.TrimSpace(fallbackCode), fallbackID)
 					}
@@ -719,6 +745,8 @@ ON CONFLICT (row_key) DO UPDATE
 SET last_apply_at=EXCLUDED.last_apply_at,
     last_apply_kind=EXCLUDED.last_apply_kind,
     last_series_ids=EXCLUDED.last_series_ids,
+    invalid_streak=0,
+    last_invalid_at=NULL,
     updated_at=now()
 `, rowKey, kind, strings.Join(ids, ","))
 
@@ -751,6 +779,34 @@ WHERE row_key=$1
 	return k, t, nil
 }
 
+func productSeriesBumpInvalidStreak(ctx context.Context, pg *sql.DB, rowKey string) (int, error) {
+	var next int
+	err := pg.QueryRowContext(ctx, `
+INSERT INTO mk_product_series_apply_state (row_key, last_apply_at, last_apply_kind, last_series_ids, invalid_streak, last_invalid_at, updated_at)
+VALUES ($1, now(), '', '', 1, now(), now())
+ON CONFLICT (row_key) DO UPDATE
+SET invalid_streak=mk_product_series_apply_state.invalid_streak + 1,
+    last_invalid_at=now(),
+    updated_at=now()
+RETURNING invalid_streak
+`, rowKey).Scan(&next)
+	if err != nil {
+		return 0, err
+	}
+	return next, nil
+}
+
+func productSeriesResetInvalidStreak(ctx context.Context, pg *sql.DB, rowKey string) error {
+	_, err := pg.ExecContext(ctx, `
+UPDATE mk_product_series_apply_state
+SET invalid_streak=0,
+    last_invalid_at=NULL,
+    updated_at=now()
+WHERE row_key=$1
+`, rowKey)
+	return err
+}
+
 func productSeriesFindRuleByID(rules []productSeriesAutoRule, seriesID int64) (productSeriesAutoRule, bool) {
 	for _, r := range rules {
 		if r.SeriesID == seriesID {

svc/queries/product_series_auto_infra.go

@@ -80,6 +80,9 @@ CREATE TABLE IF NOT EXISTS mk_product_series_apply_state (
   CONSTRAINT ck_mk_product_series_apply_state_kind CHECK (last_apply_kind IN ('', 'rules', 'fallback'))
 )`,
 		`CREATE INDEX IF NOT EXISTS ix_mk_product_series_apply_state_updated ON mk_product_series_apply_state (updated_at DESC)`,
+		// Backward compatible schema upgrades.
+		`ALTER TABLE mk_product_series_apply_state ADD COLUMN IF NOT EXISTS invalid_streak INTEGER NOT NULL DEFAULT 0`,
+		`ALTER TABLE mk_product_series_apply_state ADD COLUMN IF NOT EXISTS last_invalid_at TIMESTAMPTZ`,
 	}
 	for _, stmt := range stmts {
 		if _, err := pg.Exec(stmt); err != nil {