From aca2cf5d02e5238c279bc569c21af2034adf0c41 Mon Sep 17 00:00:00 2001 From: M_Kececi Date: Wed, 24 Jun 2026 17:39:38 +0300 Subject: [PATCH] Merge remote-tracking branch 'origin/master' --- svc/product_series_auto_scheduler.go | 56 ++++++++++++++++++++++++ svc/queries/product_series_auto_infra.go | 3 ++ 2 files changed, 59 insertions(+) diff --git a/svc/product_series_auto_scheduler.go b/svc/product_series_auto_scheduler.go index 15f5262..c39a73d 100644 --- a/svc/product_series_auto_scheduler.go +++ b/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 { diff --git a/svc/queries/product_series_auto_infra.go b/svc/queries/product_series_auto_infra.go index d5d679a..7114624 100644 --- a/svc/queries/product_series_auto_infra.go +++ b/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 {