Repository: saleor/saleor
Pull request: Recalculate prices with celery beat
Calculate the prices in the background. In the previous solution any action on product, variant, sale, promotion, rule was triggering the action of recalculating prices in the background. In case of huge amount of actions this could trigger a lot of duplicated tasks. We change this approach to only set the `dirty` flag in the DB.  The celery workers defined in celery beat will pick up the PromotionRules/Products which needs price recalculation. With this approach, there is no risk, that multiple mutation calls could fully occupy the celery workers with price recalculations.

The changes provides below functions:
- `mark_active_promotion_rules_as_dirty(channel_ids: Iterable[int])` - mark rules with `variants_dirty`. Only rules with relations to `channel_ids` will be re-calculated - there are no reasons to re-calculated the promotion for channel X, where we update promotion/variant/etc for channel Y
- `mark_promotion_rules_as_dirty(promotion_pks: Iterable[UUID])`  - mark all rules assigned to specific promotions as dirty
- `mark_products_as_dirty(channel_to_product_ids: dict[int, set[int]]):` - accepts dictionary `channelID: {productIDs, ...}`. Based on the input the ProductChannelListing assigned to provided channelID will be marked as a dirty. So instead of processing all channelListings, we will process only the one that needs it. 

resolves: https://github.com/saleor/saleor/issues/13840


DoD:
- Adding/Updating/Removing promotion rules have an impact on variant prices
- Modifing Collection/Category/Product/Variant listed/filtered in category-predicate should have an impact on prices

For user, it should work in the same way as previously, one thing which is changed there is the way how we recalculate the prices. It can take maximal +/- 30 seconds after providing the changes to get updated prices.

# Impact

- [ x ] New migrations
- [ ] New/Updated API fields or mutations
- [ ] Deprecated API fields or mutations
- [ ] Removed API types, fields, or mutations

# Docs

<!-- Docs are stored in a separate repository: https://github.com/saleor/saleor-docs/. -->
<!-- Please provide a link to the PR that updates documentation for your changes. -->
<!-- If changes in docs are not required, please mention that in the description. -->

- [ ] Link to documentation:

# Pull Request Checklist

<!-- Please keep this section. It will make the maintainer's life easier. -->

- [ ] Privileged queries and mutations are either absent or guarded by proper permission checks
- [ ] Database queries are optimized and the number of queries is constant
- [ ] Database migrations are either absent or optimized for zero downtime
- [ ] The changes are covered by test cases
- [ ] All new fields/inputs/mutations have proper labels added (`ADDED_IN_X`, `PREVIEW_FEATURE`, etc.)
- [ ] All migrations have proper dependencies
- [ ] All indexes are added concurrently in migrations
- [ ] All RunSql and RunPython migrations have revert option defined


Problem statement:
Introduce the flag for products to mark the objects that needs recalculation
We should refactor the way that the product prices are recalculated.

### Problem
Currently the discounted price recalculation method is called each time the related sale is changing. This may result in calling recalculation multiple times for single product what is not efficient.

## Solution
We should introduce similar solution as we already have for search index.
Add the flag on `Product` model to mark the products that needs recalculation. Then we should add a task that will be run each 20 seconds (as it's done for search vector, or maybe other frequency - to discuss) and recalculate the discounted prices only for those products.

Hints from issue discussion:
Task functions to refactor:
- `update_products_discounted_prices_for_promotion_task`
- `update_products_discounted_prices_of_promotion_task`
We could try to use a custom Celery scheduler that will schedule the task only if there are products that need recalculating. An example custom schedule based on this logic was added in [this commit](https://github.com/saleor/saleor/pull/14823/commits/c4ed78c6fd984ef271878f811d3087bdf0bc753d#diff-6724102dbc2ea88d9de3ffdef217f9f4ac654211597696902f8aaf0d26bc5637R165).

Commit message: fix tests.

Patch to review:
diff --git a/saleor/discount/migrations/0074_promotionrule_variants_dirty.py b/saleor/discount/migrations/0074_promotionrule_variants_dirty.py
new file mode 100644
index 00000000000..dd6b3a3801c
--- /dev/null
+++ b/saleor/discount/migrations/0074_promotionrule_variants_dirty.py
@@ -0,0 +1,25 @@
+# Generated by Django 3.2.23 on 2024-01-09 08:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("discount", "0073_auto_20231213_1535"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="promotionrule",
+            name="variants_dirty",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.RunSQL(
+            """
+            ALTER TABLE discount_promotionrule
+            ALTER COLUMN variants_dirty
+            SET DEFAULT false;
+            """,
+            migrations.RunSQL.noop,
+        ),
+    ]
diff --git a/saleor/discount/models.py b/saleor/discount/models.py
index 9b6f423d9c2..9dc6a89df72 100644
--- a/saleor/discount/models.py
+++ b/saleor/discount/models.py
@@ -384,6 +384,7 @@ class PromotionRule(models.Model):
         blank=True,
     )
     old_channel_listing_id = models.IntegerField(blank=True, null=True, unique=True)
+    variants_dirty = models.BooleanField(default=False)
 
     class Meta:
         ordering = ("name", "pk")
diff --git a/saleor/discount/tasks.py b/saleor/discount/tasks.py
index 84542395c8f..2cc1b3c7cad 100644
--- a/saleor/discount/tasks.py
+++ b/saleor/discount/tasks.py
@@ -20,6 +20,7 @@
     VariantChannelListingPromotionRule,
 )
 from ..product.tasks import update_products_discounted_prices_for_promotion_task
+from ..product.utils.product import mark_products_for_recalculate_discounted_price
 from ..product.utils.variant_prices import update_discounted_prices_for_promotion
 from ..webhook.event_types import WebhookEventAsyncType
 from ..webhook.utils import get_webhooks_for_event
@@ -308,7 +309,7 @@ def update_discounted_prices_task():
     if products_ids:
         products = Product.objects.filter(id__in=products_ids)
         update_discounted_prices_for_promotion(products)
-        update_discounted_prices_task.delay()
+        mark_products_for_recalculate_discounted_price(products_ids)
 
 
 @app.task(
diff --git a/saleor/discount/utils.py b/saleor/discount/utils.py
index 055509cf7d9..fe32b06d952 100644
--- a/saleor/discount/utils.py
+++ b/saleor/discount/utils.py
@@ -12,6 +12,7 @@
 from uuid import UUID
 
 import graphene
+from django.conf import settings
 from django.db import transaction
 from django.db.models import Exists, F, OuterRef, QuerySet
 from django.utils import timezone
@@ -624,3 +625,21 @@ def update_rule_variant_relation(
         PromotionRuleVariant.objects.bulk_create(
             rules_variants_to_add, ignore_conflicts=True
         )
+
+
+def get_active_promotion_rules(
+    allow_replica: bool = False,
+) -> "QuerySet[PromotionRule]":
+    promotions = Promotion.objects.active()
+    if allow_replica:
+        promotions = promotions.using(settings.DATABASE_CONNECTION_REPLICA_NAME)
+    rules = (
+        PromotionRule.objects.order_by("id")
+        .filter(
+            Exists(promotions.filter(id=OuterRef("promotion_id"))),
+        )
+        .exclude(catalogue_predicate={})
+    )
+    if allow_replica:
+        rules = rules.using(settings.DATABASE_CONNECTION_REPLICA_NAME)
+    return rules
diff --git a/saleor/graphql/discount/mutations/bulk_mutations.py b/saleor/graphql/discount/mutations/bulk_mutations.py
index 6074283f472..4aacb94cf1e 100644
--- a/saleor/graphql/discount/mutations/bulk_mutations.py
+++ b/saleor/graphql/discount/mutations/bulk_mutations.py
@@ -8,7 +8,7 @@
 from ....discount.error_codes import DiscountErrorCode
 from ....discount.utils import get_current_products_for_rules
 from ....permission.enums import DiscountPermissions
-from ....product.tasks import update_discounted_prices_task
+from ....product.utils.product import mark_products_for_recalculate_discounted_price
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.utils import get_webhooks_for_event
 from ...core import ResolveInfo
@@ -97,7 +97,7 @@ def bulk_action(cls, info: ResolveInfo, queryset, /):
             cls.call_event(
                 manager.sale_deleted, sale, catalogue_info, webhooks=webhooks
             )
-        update_discounted_prices_task.delay(list(product_ids))
+        mark_products_for_recalculate_discounted_price(list(product_ids))
 
     @classmethod
     def get_sale_and_rules(cls, qs: QuerySet[models.Promotion]):
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_bulk_delete.py b/saleor/graphql/discount/mutations/promotion/promotion_bulk_delete.py
index 149aabb4e09..8ded93abdaa 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_bulk_delete.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_bulk_delete.py
@@ -4,7 +4,7 @@
 from .....discount import models
 from .....discount.utils import get_current_products_for_rules
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from .....webhook.utils import get_webhooks_for_event
 from ....core import ResolveInfo
@@ -46,8 +46,7 @@ def bulk_action(cls, info: ResolveInfo, queryset, /):
         webhooks = get_webhooks_for_event(WebhookEventAsyncType.PROMOTION_DELETED)
         for promotion in promotions:
             cls.call_event(manager.promotion_deleted, promotion, webhooks=webhooks)
-
-        update_discounted_prices_task.delay(list(product_ids))
+        mark_products_for_recalculate_discounted_price(list(product_ids))
 
     @classmethod
     def get_product_ids(cls, qs: QuerySet[models.Promotion]):
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_create.py b/saleor/graphql/discount/mutations/promotion/promotion_create.py
index f5db32c2f02..5ee4199519a 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_create.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_create.py
@@ -11,7 +11,6 @@
 from .....discount import events, models
 from .....permission.enums import DiscountPermissions
 from .....plugins.manager import PluginsManager
-from .....product.tasks import update_products_discounted_prices_of_promotion_task
 from .....webhook.event_types import WebhookEventAsyncType
 from ....app.dataloaders import get_app_promise
 from ....channel.types import Channel
@@ -176,7 +175,9 @@ def _save_m2m(
         if rules_data := cleaned_data.get("rules"):
             for rule_data in rules_data:
                 channels = rule_data.pop("channels", None)
-                rule = models.PromotionRule(promotion=instance, **rule_data)
+                rule = models.PromotionRule(
+                    promotion=instance, variants_dirty=True, **rule_data
+                )
                 rules_with_channels_to_add.append((rule, channels))
                 rules.append(rule)
             models.PromotionRule.objects.bulk_create(rules)
@@ -200,7 +201,6 @@ def post_save_actions(
         cls.call_event(manager.promotion_created, instance)
         if has_started:
             cls.send_promotion_started_webhook(manager, instance)
-        update_products_discounted_prices_of_promotion_task.delay(instance.pk)
 
     @classmethod
     def has_started(cls, instance: models.Promotion) -> bool:
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_delete.py b/saleor/graphql/discount/mutations/promotion/promotion_delete.py
index bfd8f067fbb..5353d0dede4 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_delete.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_delete.py
@@ -5,7 +5,7 @@
 from .....discount.utils import get_current_products_for_rules
 from .....graphql.core.mutations import ModelDeleteMutation
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....core import ResolveInfo
 from ....core.descriptions import ADDED_IN_317, PREVIEW_FEATURE
@@ -58,5 +58,5 @@ def perform_mutation(  # type: ignore[override]
             response = super().perform_mutation(root, info, id=id)
             instance.id = promotion_id
             cls.call_event(manager.promotion_deleted, instance)
-            update_discounted_prices_task.delay(product_ids)
+            mark_products_for_recalculate_discounted_price(product_ids)
         return response
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_rule_create.py b/saleor/graphql/discount/mutations/promotion/promotion_rule_create.py
index 84ffedb353f..d73e263a5be 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_rule_create.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_rule_create.py
@@ -5,7 +5,7 @@
 
 from .....discount import events, models
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....app.dataloaders import get_app_promise
 from ....core import ResolveInfo
@@ -70,7 +70,7 @@ def clean_input(
     def post_save_action(cls, info: ResolveInfo, instance, cleaned_input):
         products = get_products_for_rule(instance, update_rule_variants=True)
         if products:
-            update_discounted_prices_task.delay(
+            mark_products_for_recalculate_discounted_price(
                 list(products.values_list("id", flat=True))
             )
         clear_promotion_old_sale_id(instance.promotion, save=True)
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_rule_delete.py b/saleor/graphql/discount/mutations/promotion/promotion_rule_delete.py
index 450d96d327c..d81041f6ea0 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_rule_delete.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_rule_delete.py
@@ -4,7 +4,7 @@
 from .....discount.utils import get_current_products_for_rules
 from .....graphql.core.mutations import ModelDeleteMutation
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....app.dataloaders import get_app_promise
 from ....core import ResolveInfo
@@ -66,7 +66,7 @@ def perform_mutation(  # type: ignore[override]
         instance.id = db_id
 
         if product_ids:
-            update_discounted_prices_task.delay(product_ids)
+            mark_products_for_recalculate_discounted_price(product_ids)
 
         app = get_app_promise(info.context).get()
         events.rule_deleted_event(info.context.user, app, [instance])
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_rule_update.py b/saleor/graphql/discount/mutations/promotion/promotion_rule_update.py
index 9992697c2fc..f34d61ce53d 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_rule_update.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_rule_update.py
@@ -5,7 +5,7 @@
 from .....discount import RewardValueType, events, models
 from .....discount.utils import get_current_products_for_rules
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....app.dataloaders import get_app_promise
 from ....core import ResolveInfo
@@ -200,7 +200,9 @@ def post_save_actions(cls, info: ResolveInfo, instance, previous_product_ids):
         products = get_products_for_rule(instance, update_rule_variants=True)
         product_ids = set(products.values_list("id", flat=True)) | previous_product_ids
         if product_ids:
-            update_discounted_prices_task.delay(list(product_ids))
+            mark_products_for_recalculate_discounted_price(
+                list(products.values_list("id", flat=True))
+            )
         clear_promotion_old_sale_id(instance.promotion, save=True)
         app = get_app_promise(info.context).get()
         events.rule_updated_event(info.context.user, app, [instance])
diff --git a/saleor/graphql/discount/mutations/promotion/promotion_update.py b/saleor/graphql/discount/mutations/promotion/promotion_update.py
index 4ebdf9c5cb7..39ef1d62fda 100644
--- a/saleor/graphql/discount/mutations/promotion/promotion_update.py
+++ b/saleor/graphql/discount/mutations/promotion/promotion_update.py
@@ -9,7 +9,6 @@
 from .....discount import events, models
 from .....permission.enums import DiscountPermissions
 from .....plugins.manager import PluginsManager
-from .....product.tasks import update_products_discounted_prices_of_promotion_task
 from .....webhook.event_types import WebhookEventAsyncType
 from ....app.dataloaders import get_app_promise
 from ....core import ResolveInfo
@@ -109,7 +108,7 @@ def post_save_actions(cls, info, cleaned_input, instance, previous_end_date):
         # update the product undiscounted prices for promotion only when
         # start or end date has changed
         if "start_date" in cleaned_input or "end_date" in cleaned_input:
-            update_products_discounted_prices_of_promotion_task.delay(instance.pk)
+            instance.rules.all().update(variants_dirty=True)
 
     @classmethod
     def get_toggle_type(cls, instance, clean_input, previous_end_date) -> Optional[str]:
diff --git a/saleor/graphql/discount/mutations/sale/sale_base_catalogue.py b/saleor/graphql/discount/mutations/sale/sale_base_catalogue.py
index 0a1996f083e..6f28a894372 100644
--- a/saleor/graphql/discount/mutations/sale/sale_base_catalogue.py
+++ b/saleor/graphql/discount/mutations/sale/sale_base_catalogue.py
@@ -7,8 +7,8 @@
 
 from .....discount.error_codes import DiscountErrorCode
 from .....product import models as product_models
-from .....product.tasks import update_discounted_prices_task
 from .....product.utils import get_products_ids_without_variants
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from ....core import ResolveInfo
 from ....core.mutations import BaseMutation
 from ....plugins.dataloaders import get_plugin_manager_promise
@@ -66,7 +66,7 @@ def post_save_actions(
                 product_ids = new_product_ids - previous_product_ids
             else:
                 product_ids = previous_product_ids - new_product_ids
-            update_discounted_prices_task.delay(list(product_ids))
+            mark_products_for_recalculate_discounted_price(list(product_ids))
 
     @classmethod
     def get_product_ids_for_predicate(cls, predicate: dict) -> set[int]:
diff --git a/saleor/graphql/discount/mutations/sale/sale_channel_listing_update.py b/saleor/graphql/discount/mutations/sale/sale_channel_listing_update.py
index f255e281f51..60a23dec53b 100644
--- a/saleor/graphql/discount/mutations/sale/sale_channel_listing_update.py
+++ b/saleor/graphql/discount/mutations/sale/sale_channel_listing_update.py
@@ -11,7 +11,7 @@
 from .....discount.error_codes import DiscountErrorCode
 from .....discount.models import Promotion, PromotionRule
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_products_discounted_prices_of_promotion_task
+from .....product.tasks import mark_promotion_rules_to_recalculate_variants
 from ....channel import ChannelContext
 from ....channel.mutations import BaseChannelListingMutation
 from ....core import ResolveInfo
@@ -223,7 +223,7 @@ def save(
         with traced_atomic_transaction():
             cls.add_channels(promotion, rule, cleaned_input.get("add_channels", []))
             cls.remove_channels(promotion, cleaned_input.get("remove_channels", []))
-            update_products_discounted_prices_of_promotion_task.delay(promotion.pk)
+            mark_promotion_rules_to_recalculate_variants(promotion.pk)
 
     @classmethod
     def get_instance(cls, id):
diff --git a/saleor/graphql/discount/mutations/sale/sale_create.py b/saleor/graphql/discount/mutations/sale/sale_create.py
index 2890575b1ae..32baf4fd027 100644
--- a/saleor/graphql/discount/mutations/sale/sale_create.py
+++ b/saleor/graphql/discount/mutations/sale/sale_create.py
@@ -9,7 +9,7 @@
 from .....discount.error_codes import DiscountErrorCode
 from .....discount.models import Promotion
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_products_discounted_prices_of_promotion_task
+from .....product.tasks import mark_promotion_rules_to_recalculate_variants
 from .....webhook.event_types import WebhookEventAsyncType
 from ....channel import ChannelContext
 from ....core import ResolveInfo
@@ -130,7 +130,7 @@ def perform_mutation(cls, _root, info: ResolveInfo, /, **data):
             )
             manager = get_plugin_manager_promise(info.context).get()
             cls.send_sale_notifications(manager, promotion, predicate)
-            update_products_discounted_prices_of_promotion_task.delay(promotion.pk)
+            mark_promotion_rules_to_recalculate_variants(promotion.pk)
         return response
 
     @classmethod
diff --git a/saleor/graphql/discount/mutations/sale/sale_delete.py b/saleor/graphql/discount/mutations/sale/sale_delete.py
index 8b937d33876..d31b0116aba 100644
--- a/saleor/graphql/discount/mutations/sale/sale_delete.py
+++ b/saleor/graphql/discount/mutations/sale/sale_delete.py
@@ -5,7 +5,7 @@
 from .....discount.error_codes import DiscountErrorCode
 from .....graphql.core.mutations import ModelDeleteMutation
 from .....permission.enums import DiscountPermissions
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....channel import ChannelContext
 from ....core import ResolveInfo
@@ -68,7 +68,7 @@ def perform_mutation(  # type: ignore[override]
 
             manager = get_plugin_manager_promise(info.context).get()
             cls.call_event(manager.sale_deleted, promotion, previous_catalogue)
-            update_discounted_prices_task.delay(list(product_ids))
+            mark_products_for_recalculate_discounted_price(list(product_ids))
 
         return response
 
diff --git a/saleor/graphql/discount/mutations/sale/sale_update.py b/saleor/graphql/discount/mutations/sale/sale_update.py
index 82c53b96676..93dbe34a027 100644
--- a/saleor/graphql/discount/mutations/sale/sale_update.py
+++ b/saleor/graphql/discount/mutations/sale/sale_update.py
@@ -11,7 +11,7 @@
 from .....discount.utils import CATALOGUE_FIELDS
 from .....permission.enums import DiscountPermissions
 from .....product import models as product_models
-from .....product.tasks import update_discounted_prices_task
+from .....product.utils.product import mark_products_for_recalculate_discounted_price
 from .....webhook.event_types import WebhookEventAsyncType
 from ....channel import ChannelContext
 from ....core import ResolveInfo
@@ -185,7 +185,6 @@ def post_save_actions(
             current_catalogue,
             previous_end_date,
         )
-
         if any(
             field in input.keys()
             for field in [*CATALOGUE_FIELDS, "start_date", "end_date", "type"]
@@ -198,7 +197,7 @@ def post_save_actions(
             )
             update_variants_for_promotion(variants, promotion)
             if product_ids | previous_product_ids:
-                update_discounted_prices_task.delay(list(product_ids))
+                mark_products_for_recalculate_discounted_price(list(product_ids))
 
     @classmethod
     def send_sale_notifications(
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_create.py b/saleor/graphql/discount/tests/mutations/test_promotion_create.py
index 538f3b4479b..ecbb53e3561 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_create.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_create.py
@@ -49,13 +49,11 @@
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_created")
 def test_promotion_create_by_staff_user(
     promotion_created_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     description_json,
@@ -160,19 +158,16 @@ def test_promotion_create_by_staff_user(
 
     promotion_created_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_called_once_with(promotion)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_created")
 def test_promotion_create_by_app(
     promotion_created_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     app_api_client,
     permission_manage_discounts,
     description_json,
@@ -240,9 +235,8 @@ def test_promotion_create_by_app(
 
     promotion_created_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_called_once_with(promotion)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
@@ -507,13 +501,11 @@ def test_promotion_create_invalid_percentage_value(
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_created")
 def test_promotion_create_only_name_and_end_date(
     promotion_created_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     app_api_client,
     permission_manage_discounts,
     description_json,
@@ -551,19 +543,16 @@ def test_promotion_create_only_name_and_end_date(
 
     promotion_created_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_called_once_with(promotion)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_created")
 def test_promotion_create_start_date_and_end_date_after_current_date(
     promotion_created_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     description_json,
@@ -668,9 +657,8 @@ def test_promotion_create_start_date_and_end_date_after_current_date(
 
     promotion_created_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_delete.py b/saleor/graphql/discount/tests/mutations/test_promotion_delete.py
index e64f81f4dc5..2402198974c 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_delete.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_delete.py
@@ -4,6 +4,7 @@
 import pytest
 
 from ....tests.utils import assert_no_permission, get_graphql_content
+from ...utils import get_products_for_promotion
 
 PROMOTION_DELETE_MUTATION = """
     mutation promotionDelete($id: ID!) {
@@ -22,11 +23,9 @@
 """
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_deleted")
 def test_promotion_delete_by_staff_user(
     promotion_deleted_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion,
@@ -48,14 +47,13 @@ def test_promotion_delete_by_staff_user(
     with pytest.raises(promotion._meta.model.DoesNotExist):
         promotion.refresh_from_db()
 
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_deleted")
 def test_promotion_delete_by_staff_app(
     promotion_deleted_mock,
-    update_discounted_prices_task_mock,
     app_api_client,
     permission_manage_discounts,
     promotion,
@@ -77,14 +75,14 @@ def test_promotion_delete_by_staff_app(
 
     with pytest.raises(promotion._meta.model.DoesNotExist):
         promotion.refresh_from_db()
-    update_discounted_prices_task_mock.assert_called_once()
+
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_deleted")
 def test_promotion_delete_by_customer(
     promotion_deleted_mock,
-    update_discounted_prices_task_mock,
     api_client,
     promotion,
 ):
@@ -98,4 +96,5 @@ def test_promotion_delete_by_customer(
     assert_no_permission(response)
 
     promotion_deleted_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is False
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_rule_create.py b/saleor/graphql/discount/tests/mutations/test_promotion_rule_create.py
index fd643fc3181..3748297d867 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_rule_create.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_rule_create.py
@@ -8,6 +8,7 @@
 from .....discount.models import PromotionEvent
 from ....tests.utils import assert_no_permission, get_graphql_content
 from ...enums import RewardValueTypeEnum
+from ...utils import get_products_for_promotion
 
 PROMOTION_RULE_CREATE_MUTATION = """
     mutation promotionRuleCreate($input: PromotionRuleCreateInput!) {
@@ -45,9 +46,7 @@
 
 
 @patch("saleor.plugins.manager.PluginsManager.promotion_rule_created")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_create_by_staff_user(
-    update_discounted_prices_task_mock,
     promotion_rule_created_mock,
     staff_api_client,
     permission_group_manage_discounts,
@@ -116,6 +115,7 @@ def test_promotion_rule_create_by_staff_user(
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleCreate"]
     rule_data = data["promotionRule"]
+    product.refresh_from_db()
 
     assert not data["errors"]
     assert rule_data["name"] == name
@@ -126,14 +126,12 @@ def test_promotion_rule_create_by_staff_user(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count + 1
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
     rule = promotion.rules.last()
     promotion_rule_created_mock.assert_called_once_with(rule)
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_create_by_app(
-    update_discounted_prices_task_mock,
     app_api_client,
     permission_manage_discounts,
     description_json,
@@ -187,6 +185,7 @@ def test_promotion_rule_create_by_app(
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleCreate"]
     rule_data = data["promotionRule"]
+    product = category.products.first()
 
     assert not data["errors"]
     assert rule_data["name"] == name
@@ -197,14 +196,10 @@ def test_promotion_rule_create_by_app(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count + 1
-    update_discounted_prices_task_mock.assert_called_once_with(
-        [category.products.first().id]
-    )
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_create_by_customer(
-    update_discounted_prices_task_mock,
     api_client,
     description_json,
     channel_USD,
@@ -248,7 +243,8 @@ def test_promotion_rule_create_by_customer(
     # when
     response = api_client.post_graphql(PROMOTION_RULE_CREATE_MUTATION, variables)
     assert_no_permission(response)
-    update_discounted_prices_task_mock.assert_not_called()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is False
 
 
 def test_promotion_rule_create_missing_catalogue_predicate(
@@ -775,9 +771,7 @@ def test_promotion_rule_create_percentage_value_above_100(
     assert promotion.rules.count() == rules_count
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_create_clears_old_sale_id(
-    update_discounted_prices_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     description_json,
@@ -830,6 +824,7 @@ def test_promotion_rule_create_clears_old_sale_id(
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleCreate"]
     rule_data = data["promotionRule"]
+    product.refresh_from_db()
 
     assert not data["errors"]
     assert rule_data["name"] == name
@@ -840,7 +835,7 @@ def test_promotion_rule_create_clears_old_sale_id(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count + 1
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
+    assert product.discounted_price_dirty is True
 
     promotion.refresh_from_db()
     assert promotion.old_sale_id is None
@@ -898,9 +893,7 @@ def test_promotion_rule_create_events(
 
 
 @patch("saleor.plugins.manager.PluginsManager.promotion_rule_created")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_create_serializable_decimal_in_predicate(
-    update_discounted_prices_task_mock,
     promotion_rule_created_mock,
     staff_api_client,
     permission_group_manage_discounts,
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_rule_delete.py b/saleor/graphql/discount/tests/mutations/test_promotion_rule_delete.py
index 8b051b05e2b..8fabb70799f 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_rule_delete.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_rule_delete.py
@@ -1,11 +1,10 @@
-from unittest.mock import patch
-
 import graphene
 import pytest
 
 from .....discount import PromotionEvents
 from .....discount.models import PromotionEvent
 from ....tests.utils import assert_no_permission, get_graphql_content
+from ...utils import get_products_for_promotion
 
 PROMOTION_RULE_DELETE_MUTATION = """
     mutation promotionRuleDelete($id: ID!) {
@@ -34,11 +33,7 @@
 """
 
 
-@patch("saleor.plugins.manager.PluginsManager.promotion_rule_deleted")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_delete_by_staff_user(
-    update_discounted_prices_task_mock,
-    promotion_rule_deleted_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion,
@@ -58,14 +53,12 @@ def test_promotion_rule_delete_by_staff_user(
     assert data["promotionRule"]["name"] == rule.name
     with pytest.raises(rule._meta.model.DoesNotExist):
         rule.refresh_from_db()
+    product.refresh_from_db()
 
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
-    promotion_rule_deleted_mock.assert_called_once_with(rule)
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_delete_by_staff_app(
-    update_discounted_prices_task_mock,
     app_api_client,
     permission_manage_discounts,
     promotion,
@@ -83,34 +76,33 @@ def test_promotion_rule_delete_by_staff_app(
     )
 
     # then
+    product.refresh_from_db()
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleDelete"]
     assert data["promotionRule"]["name"] == rule.name
     with pytest.raises(rule._meta.model.DoesNotExist):
         rule.refresh_from_db()
 
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
-def test_promotion_rule_delete_by_customer(
-    update_discounted_prices_task_mock, api_client, promotion
-):
+def test_promotion_rule_delete_by_customer(api_client, promotion):
     # given
     rule = promotion.rules.first()
     variables = {"id": graphene.Node.to_global_id("PromotionRule", rule.id)}
+    products = get_products_for_promotion(promotion)
 
     # when
     response = api_client.post_graphql(PROMOTION_RULE_DELETE_MUTATION, variables)
 
     # then
     assert_no_permission(response)
-    update_discounted_prices_task_mock.assert_not_called()
+    for product in products:
+        product.refresh_from_db()
+        assert product.discounted_price_dirty is False
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_delete_clears_old_sale_id(
-    update_discounted_prices_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion_converted_from_sale,
@@ -129,16 +121,16 @@ def test_promotion_delete_clears_old_sale_id(
     response = staff_api_client.post_graphql(PROMOTION_RULE_DELETE_MUTATION, variables)
 
     # then
+    product.refresh_from_db()
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleDelete"]
     assert data["promotionRule"]["name"] == rule.name
     with pytest.raises(rule._meta.model.DoesNotExist):
         rule.refresh_from_db()
 
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
-
     promotion.refresh_from_db()
     assert promotion.old_sale_id is None
+    assert product.discounted_price_dirty is True
 
 
 def test_promotion_rule_delete_events(
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_rule_update.py b/saleor/graphql/discount/tests/mutations/test_promotion_rule_update.py
index 634f142db25..48e94c4b8f9 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_rule_update.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_rule_update.py
@@ -6,8 +6,10 @@
 from .....discount import PromotionEvents, RewardValueType
 from .....discount.error_codes import PromotionRuleUpdateErrorCode
 from .....discount.models import PromotionEvent
+from .....product.models import Product
 from ....tests.utils import assert_no_permission, get_graphql_content
 from ...enums import RewardValueTypeEnum
+from ...utils import get_variants_for_predicate
 
 PROMOTION_RULE_UPDATE_MUTATION = """
     mutation promotionRuleUpdate($id: ID!, $input: PromotionRuleUpdateInput!) {
@@ -45,9 +47,7 @@
 
 
 @patch("saleor.plugins.manager.PluginsManager.promotion_rule_updated")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_update_by_staff_user(
-    update_discounted_prices_task_mock,
     promotion_rule_updated_mock,
     staff_api_client,
     permission_group_manage_discounts,
@@ -129,15 +129,14 @@ def test_promotion_rule_update_by_staff_user(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count
-    update_discounted_prices_task_mock.assert_called_once()
-    args, _kwargs = update_discounted_prices_task_mock.call_args
-    assert set(args[0]) == {product_list[1].id, product_list[2].id}
     promotion_rule_updated_mock.assert_called_once_with(rule)
+    for product in Product.objects.filter(
+        id__in=[product_list[1].id, product_list[2].id]
+    ):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_update_by_app(
-    update_discounted_prices_task_mock,
     app_api_client,
     permission_manage_discounts,
     channel_USD,
@@ -174,6 +173,7 @@ def test_promotion_rule_update_by_app(
     )
 
     # then
+    product.refresh_from_db()
     content = get_graphql_content(response)
     data = content["data"]["promotionRuleUpdate"]
     rule_data = data["promotionRule"]
@@ -188,12 +188,10 @@ def test_promotion_rule_update_by_app(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count
-    update_discounted_prices_task_mock.assert_called_once_with([product.id])
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_update_by_customer(
-    update_discounted_prices_task_mock,
     api_client,
     channel_USD,
     channel_PLN,
@@ -223,7 +221,8 @@ def test_promotion_rule_update_by_customer(
 
     # then
     assert_no_permission(response)
-    update_discounted_prices_task_mock.assert_not_called()
+    for variant in get_variants_for_predicate(rule.catalogue_predicate):
+        assert variant.product.discounted_price_dirty is True
 
 
 def test_promotion_rule_update_duplicates_channels_in_add_and_remove_field(
@@ -639,9 +638,7 @@ def test_promotion_rule_update_reward_value_invalid_percentage_value(
     assert errors[0]["field"] == "rewardValue"
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_promotion_rule_update_clears_old_sale_id(
-    update_discounted_prices_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     channel_USD,
@@ -724,13 +721,15 @@ def test_promotion_rule_update_clears_old_sale_id(
     assert rule_data["rewardValue"] == reward_value
     assert rule_data["promotion"]["id"] == promotion_id
     assert promotion.rules.count() == rules_count
-    update_discounted_prices_task_mock.assert_called_once()
-    args, _kwargs = update_discounted_prices_task_mock.call_args
-    assert set(args[0]) == {product_list[1].id, product_list[2].id}
 
     promotion.refresh_from_db()
     assert promotion.old_sale_id is None
 
+    for product in Product.objects.filter(
+        id__in=[product_list[1].id, product_list[2].id]
+    ):
+        assert product.discounted_price_dirty is True
+
 
 def test_promotion_rule_update_events(
     staff_api_client,
diff --git a/saleor/graphql/discount/tests/mutations/test_promotion_update.py b/saleor/graphql/discount/tests/mutations/test_promotion_update.py
index eff74bb3888..019eb946bb1 100644
--- a/saleor/graphql/discount/tests/mutations/test_promotion_update.py
+++ b/saleor/graphql/discount/tests/mutations/test_promotion_update.py
@@ -38,13 +38,11 @@
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_updated")
 def test_promotion_update_by_staff_user(
     promotion_updated_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion,
@@ -88,19 +86,16 @@ def test_promotion_update_by_staff_user(
 
     promotion_updated_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_called_once_with(promotion)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_ended")
 @patch("saleor.plugins.manager.PluginsManager.promotion_updated")
 def test_promotion_update_by_app(
     promotion_updated_mock,
     promotion_ended_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     app_api_client,
     permission_manage_discounts,
     promotion,
@@ -143,13 +138,11 @@ def test_promotion_update_by_app(
 
     promotion_updated_mock.assert_called_once_with(promotion)
     promotion_ended_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_ended")
 @patch("saleor.plugins.manager.PluginsManager.promotion_updated")
@@ -157,7 +150,6 @@ def test_promotion_update_dates_dont_change(
     promotion_updated_mock,
     promotion_started_mock,
     promotion_ended_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion,
@@ -204,11 +196,11 @@ def test_promotion_update_dates_dont_change(
     promotion_updated_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_not_called()
     promotion_ended_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_ended")
 @patch("saleor.plugins.manager.PluginsManager.promotion_updated")
@@ -216,7 +208,6 @@ def test_promotion_update_by_customer(
     promotion_updated_mock,
     promotion_started_mock,
     promotion_ended_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     api_client,
     promotion,
 ):
@@ -243,7 +234,8 @@ def test_promotion_update_by_customer(
     promotion_updated_mock.assert_not_called()
     promotion_started_mock.assert_not_called()
     promotion_ended_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
 @freeze_time("2020-03-18 12:00:00")
@@ -280,13 +272,11 @@ def test_promotion_update_end_date_before_start_date(
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.promotion_started")
 @patch("saleor.plugins.manager.PluginsManager.promotion_updated")
 def test_promotion_update_clears_old_sale_id(
     promotion_updated_mock,
     promotion_started_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_group_manage_discounts,
     promotion_converted_from_sale,
@@ -330,9 +320,8 @@ def test_promotion_update_clears_old_sale_id(
 
     promotion_updated_mock.assert_called_once_with(promotion)
     promotion_started_mock.assert_called_once_with(promotion)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        promotion.id
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 def test_promotion_update_events(
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_catalogues_add.py b/saleor/graphql/discount/tests/mutations/test_sale_catalogues_add.py
index 48a50568653..8889cbc65da 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_catalogues_add.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_catalogues_add.py
@@ -22,11 +22,9 @@
 """
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_add_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale_with_empty_predicate,
     category,
@@ -78,14 +76,13 @@ def test_sale_add_catalogues(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_add_catalogues_no_changes_in_catalogue(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -135,14 +132,13 @@ def test_sale_add_catalogues_no_changes_in_catalogue(
     assert current_catalogue == previous_catalogue
 
     updated_webhook_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_add_empty_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -183,14 +179,13 @@ def test_sale_add_empty_catalogues(
     assert variant_id in current_catalogue["variants"]
 
     updated_webhook_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_add_empty_catalogues_to_sale_with_empty_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale_with_empty_predicate,
     permission_manage_discounts,
@@ -222,14 +217,11 @@ def test_sale_add_empty_catalogues_to_sale_with_empty_catalogues(
     assert not current_catalogue["variants"]
 
     updated_webhook_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_add_catalogues_no_product_ids_change(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -273,12 +265,11 @@ def test_sale_add_catalogues_no_product_ids_change(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_not_called()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 def test_sale_add_catalogues_with_product_without_variants(
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -315,7 +306,8 @@ def test_sale_add_catalogues_with_product_without_variants(
 
     assert error["code"] == DiscountErrorCode.CANNOT_MANAGE_PRODUCT_WITHOUT_VARIANT.name
     assert error["message"] == "Cannot manage products without variants."
-    update_discounted_prices_task_mock.assert_not_called()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
 
 
 def test_sale_add_catalogues_with_promotion_id(
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_catalogues_remove.py b/saleor/graphql/discount/tests/mutations/test_sale_catalogues_remove.py
index 17e274bc03e..2a3b07a7f9e 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_catalogues_remove.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_catalogues_remove.py
@@ -22,11 +22,9 @@
 """
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_remove_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -93,14 +91,13 @@ def test_sale_remove_catalogues(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_remove_empty_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -156,14 +153,13 @@ def test_sale_remove_empty_catalogues(
     assert variant_id in current_catalogue["variants"]
 
     updated_webhook_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_remove_empty_catalogues_from_sale_with_empty_catalogues(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale_with_empty_predicate,
     permission_manage_discounts,
@@ -201,14 +197,11 @@ def test_sale_remove_empty_catalogues_from_sale_with_empty_catalogues(
     assert not current_catalogue["variants"]
 
     updated_webhook_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_sale_remove_catalogues_no_product_changes(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -249,7 +242,8 @@ def test_sale_remove_catalogues_no_product_changes(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_not_called()
+    variant.product.refresh_from_db()
+    assert variant.product.discounted_price_dirty is False
 
 
 def test_sale_remove_catalogues_with_promotion_id(
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_channel_listing_update.py b/saleor/graphql/discount/tests/mutations/test_sale_channel_listing_update.py
index 9c3d7b72e94..4a922440546 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_channel_listing_update.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_channel_listing_update.py
@@ -1,5 +1,3 @@
-from unittest.mock import patch
-
 import graphene
 
 from .....discount import RewardValueType
@@ -32,12 +30,7 @@
 """
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_add_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -84,17 +77,11 @@ def test_sale_channel_listing_add_channels(
     )
     assert all([rule.old_channel_listing_id for rule in rules])
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_add_multiple_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -145,17 +132,11 @@ def test_sale_channel_listing_add_multiple_channels(
     assert all(old_channel_listing_ids)
     assert len(set(old_channel_listing_ids)) == 3
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -196,17 +177,11 @@ def test_sale_channel_listing_update_channels(
     rules = promotion.rules.all()
     assert len(rules) == 1
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_remove_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale_with_many_channels,
     permission_manage_discounts,
@@ -244,17 +219,11 @@ def test_sale_channel_listing_remove_channels(
     rules = promotion.rules.all()
     assert len(rules) == 1
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_remove_all_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale_with_many_channels,
     permission_manage_discounts,
@@ -300,17 +269,11 @@ def test_sale_channel_listing_remove_all_channels(
     assert rules[0].reward_value_type == reward_value_type
     assert rules[0].catalogue_predicate == predicate
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_add_update_remove_channels(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale_with_many_channels,
     permission_manage_discounts,
@@ -365,9 +328,8 @@ def test_sale_channel_listing_add_update_remove_channels(
     for rule in rules:
         assert len(rule.channels.all()) == 1
 
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
 def test_sale_channel_listing_update_with_negative_discounted_value(
@@ -402,12 +364,7 @@ def test_sale_channel_listing_update_with_negative_discounted_value(
     assert_negative_positive_decimal_value(response)
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_duplicated_ids_in_add_and_remove(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -441,15 +398,12 @@ def test_sale_channel_listing_update_duplicated_ids_in_add_and_remove(
     assert errors[0]["field"] == "input"
     assert errors[0]["code"] == DiscountErrorCode.DUPLICATED_INPUT_ITEM.name
     assert errors[0]["channels"] == [channel_id]
-    mock_update_products_discounted_prices_of_promotion_task.assert_not_called()
+
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_duplicated_channel_in_add(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -485,15 +439,11 @@ def test_sale_channel_listing_update_duplicated_channel_in_add(
     assert errors[0]["field"] == "addChannels"
     assert errors[0]["code"] == DiscountErrorCode.DUPLICATED_INPUT_ITEM.name
     assert errors[0]["channels"] == [channel_id]
-    mock_update_products_discounted_prices_of_promotion_task.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_duplicated_channel_in_remove(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -523,15 +473,11 @@ def test_sale_channel_listing_update_duplicated_channel_in_remove(
     assert errors[0]["field"] == "removeChannels"
     assert errors[0]["code"] == DiscountErrorCode.DUPLICATED_INPUT_ITEM.name
     assert errors[0]["channels"] == [channel_id]
-    mock_update_products_discounted_prices_of_promotion_task.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_with_invalid_decimal_places(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -564,15 +510,11 @@ def test_sale_channel_listing_update_with_invalid_decimal_places(
     assert errors[0]["code"] == DiscountErrorCode.INVALID.name
     assert errors[0]["field"] == "input"
     assert errors[0]["channels"] == [channel_id]
-    mock_update_products_discounted_prices_of_promotion_task.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_update_with_invalid_percentage_value(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -608,7 +550,8 @@ def test_sale_channel_listing_update_with_invalid_percentage_value(
     assert errors[0]["code"] == DiscountErrorCode.INVALID.name
     assert errors[0]["field"] == "input"
     assert errors[0]["channels"] == [channel_id]
-    mock_update_products_discounted_prices_of_promotion_task.assert_not_called()
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is False
 
 
 SALE_AND_SALE_CHANNEL_LISTING_UPDATE_MUTATION = """
@@ -644,12 +587,7 @@ def test_sale_channel_listing_update_with_invalid_percentage_value(
 """
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_invalidate_data_sale_channel_listings_update(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -706,17 +644,11 @@ def test_invalidate_data_sale_channel_listings_update(
 
     # response from the second mutation contains data
     assert channel_listings_data["channelListings"][0]["channel"]["id"] == channel_id
-    mock_update_products_discounted_prices_of_promotion_task.delay.assert_called_once_with(
-        promotion.pk,
-    )
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch(
-    "saleor.graphql.discount.mutations.sale.sale_channel_listing_update"
-    ".update_products_discounted_prices_of_promotion_task"
-)
 def test_sale_channel_listing_remove_all_channels_multiple_times(
-    mock_update_products_discounted_prices_of_promotion_task,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -736,7 +668,6 @@ def test_sale_channel_listing_remove_all_channels_multiple_times(
     discounted = 2
     staff_api_client.user.user_permissions.add(permission_manage_discounts)
     query = SALE_CHANNEL_LISTING_UPDATE_MUTATION
-    mock_update_products_discounted_prices_of_promotion_task.return_value = None
 
     variables_add = {
         "id": sale_id,
@@ -764,6 +695,9 @@ def test_sale_channel_listing_remove_all_channels_multiple_times(
     assert len(rules) == 1
     assert not rules[0].channels.first()
 
+    for rule in promotion.rules.all():
+        assert rule.variants_dirty is True
+
 
 def test_sale_channel_listing_update_not_found_error(
     staff_api_client,
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_create.py b/saleor/graphql/discount/tests/mutations/test_sale_create.py
index 0b375540439..895a1f280e6 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_create.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_create.py
@@ -40,13 +40,11 @@
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_created")
 def test_create_sale(
     created_webhook_mock,
     sale_toggle_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_manage_discounts,
     product_list,
@@ -100,19 +98,17 @@ def test_create_sale(
     )
     created_webhook_mock.assert_called_once_with(sale, current_catalogue)
     sale_toggle_mock.assert_called_once_with(sale, current_catalogue)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        sale.id
-    )
+
+    for rule in sale.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_created")
 def test_create_sale_only_start_date(
     created_webhook_mock,
     sale_toggle_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_manage_discounts,
     product_list,
@@ -154,14 +150,11 @@ def test_create_sale_only_start_date(
     )
     created_webhook_mock.assert_called_once_with(sale, current_catalogue)
     sale_toggle_mock.assert_called_once_with(sale, current_catalogue)
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        sale.id
-    )
+    for rule in sale.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 def test_create_sale_with_end_date_before_startdate(
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_manage_discounts,
 ):
@@ -190,16 +183,13 @@ def test_create_sale_with_end_date_before_startdate(
     assert len(errors) == 1
     assert errors[0]["field"] == "endDate"
     assert errors[0]["code"] == DiscountErrorCode.INVALID.name
-    update_products_discounted_prices_of_promotion_task_mock.assert_not_called()
 
 
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_created")
 def test_create_sale_start_date_and_end_date_before_current_date(
     created_webhook_mock,
     sale_toggle_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_manage_discounts,
     product_list,
@@ -243,18 +233,15 @@ def test_create_sale_start_date_and_end_date_before_current_date(
     )
     created_webhook_mock.assert_called_once_with(sale, current_catalogue)
     sale_toggle_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        sale.id
-    )
+    for rule in sale.rules.all():
+        assert rule.variants_dirty is True
 
 
-@patch("saleor.product.tasks.update_products_discounted_prices_of_promotion_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_created")
 def test_create_sale_start_date_and_end_date_after_current_date(
     created_webhook_mock,
     sale_toggle_mock,
-    update_products_discounted_prices_of_promotion_task_mock,
     staff_api_client,
     permission_manage_discounts,
     product_list,
@@ -298,9 +285,8 @@ def test_create_sale_start_date_and_end_date_after_current_date(
     )
     created_webhook_mock.assert_called_once_with(sale, current_catalogue)
     sale_toggle_mock.assert_not_called()
-    update_products_discounted_prices_of_promotion_task_mock.assert_called_once_with(
-        sale.id
-    )
+    for rule in sale.rules.all():
+        assert rule.variants_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_delete.py b/saleor/graphql/discount/tests/mutations/test_sale_delete.py
index 582abad692b..a341b8cc072 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_delete.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_delete.py
@@ -6,7 +6,10 @@
 from .....discount.error_codes import DiscountErrorCode
 from .....discount.models import Promotion, PromotionRule
 from ....tests.utils import get_graphql_content
-from ...utils import convert_migrated_sale_predicate_to_catalogue_info
+from ...utils import (
+    convert_migrated_sale_predicate_to_catalogue_info,
+    get_variants_for_promotion,
+)
 
 SALE_DELETE_MUTATION = """
     mutation DeleteSale($id: ID!) {
@@ -25,11 +28,9 @@
 """
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_deleted")
 def test_sale_delete_mutation(
     deleted_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -42,6 +43,7 @@ def test_sale_delete_mutation(
         catalogue_predicate
     )
     variables = {"id": graphene.Node.to_global_id("Sale", promotion.old_sale_id)}
+    variants = get_variants_for_promotion(promotion)
 
     # when
     response = staff_api_client.post_graphql(
@@ -61,7 +63,8 @@ def test_sale_delete_mutation(
         promotion.refresh_from_db()
 
     deleted_webhook_mock.assert_called_once_with(promotion, previous_catalogue)
-    update_discounted_prices_task_mock.assert_called_once()
+    for variant in variants.select_related("product"):
+        assert variant.product.discounted_price_dirty is True
 
 
 @patch("saleor.product.tasks.update_discounted_prices_task.delay")
diff --git a/saleor/graphql/discount/tests/mutations/test_sale_update.py b/saleor/graphql/discount/tests/mutations/test_sale_update.py
index 2e2548176a7..c7bca5d2a4c 100644
--- a/saleor/graphql/discount/tests/mutations/test_sale_update.py
+++ b/saleor/graphql/discount/tests/mutations/test_sale_update.py
@@ -9,7 +9,11 @@
 from .....discount.error_codes import DiscountErrorCode
 from ....tests.utils import get_graphql_content
 from ...enums import DiscountValueTypeEnum
-from ...utils import convert_migrated_sale_predicate_to_catalogue_info
+from ...utils import (
+    convert_migrated_sale_predicate_to_catalogue_info,
+    get_products_for_promotion,
+    get_variants_for_predicate,
+)
 
 SALE_UPDATE_MUTATION = """
     mutation  saleUpdate($id: ID!, $input: SaleInput!) {
@@ -30,11 +34,9 @@
 """
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -86,14 +88,16 @@ def test_update_sale(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    variants = get_variants_for_predicate(rule.catalogue_predicate).select_related(
+        "product"
+    )
+    for variant in variants:
+        assert variant.product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_name(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -134,17 +138,16 @@ def test_update_sale_name(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_not_called()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is False
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_start_date_after_current_date_notification_not_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -187,17 +190,16 @@ def test_update_sale_start_date_after_current_date_notification_not_sent(
         promotion, previous_catalogue, current_catalogue
     )
     sale_toggle_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_start_date_before_current_date_notification_already_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -244,17 +246,16 @@ def test_update_sale_start_date_before_current_date_notification_already_sent(
         promotion, previous_catalogue, current_catalogue
     )
     sale_toggle_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_start_date_before_current_date_notification_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -298,17 +299,16 @@ def test_update_sale_start_date_before_current_date_notification_sent(
     )
 
     sale_toggle_mock.assert_called_once_with(promotion, current_catalogue)
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_end_date_after_current_date_notification_not_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -352,17 +352,16 @@ def test_update_sale_end_date_after_current_date_notification_not_sent(
         promotion, previous_catalogue, current_catalogue
     )
     sale_toggle_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_end_date_before_current_date_notification_already_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -408,17 +407,16 @@ def test_update_sale_end_date_before_current_date_notification_already_sent(
         promotion, previous_catalogue, current_catalogue
     )
     sale_toggle_mock.assert_called_once_with(promotion, current_catalogue)
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_end_date_before_current_date_notification_sent(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -462,14 +460,13 @@ def test_update_sale_end_date_before_current_date_notification_sent(
         promotion, previous_catalogue, current_catalogue
     )
     sale_toggle_mock.assert_called_once_with(promotion, current_catalogue)
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_categories(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -509,14 +506,13 @@ def test_update_sale_categories(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_collections(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -558,14 +554,13 @@ def test_update_sale_collections(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_variants(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -607,14 +602,13 @@ def test_update_sale_variants(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_products(
     updated_webhook_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     catalogue_predicate,
@@ -654,17 +648,16 @@ def test_update_sale_products(
     updated_webhook_mock.assert_called_once_with(
         promotion, previous_catalogue, current_catalogue
     )
-    update_discounted_prices_task_mock.assert_called_once()
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is True
 
 
 @freeze_time("2020-03-18 12:00:00")
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_end_date_before_start_date(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -696,7 +689,9 @@ def test_update_sale_end_date_before_start_date(
     assert errors[0]["code"] == DiscountErrorCode.INVALID.name
     updated_webhook_mock.assert_not_called()
     sale_toggle_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is False
 
 
 @freeze_time("2020-03-18 12:00:00")
@@ -755,13 +750,11 @@ def test_update_sale_with_none_values(
     assert not rule.catalogue_predicate
 
 
-@patch("saleor.product.tasks.update_discounted_prices_task.delay")
 @patch("saleor.plugins.manager.PluginsManager.sale_toggle")
 @patch("saleor.plugins.manager.PluginsManager.sale_updated")
 def test_update_sale_with_promotion_id(
     updated_webhook_mock,
     sale_toggle_mock,
-    update_discounted_prices_task_mock,
     staff_api_client,
     promotion_converted_from_sale,
     permission_manage_discounts,
@@ -794,7 +787,9 @@ def test_update_sale_with_promotion_id(
     )
     updated_webhook_mock.assert_not_called()
     sale_toggle_mock.assert_not_called()
-    update_discounted_prices_task_mock.assert_not_called()
+
+    for product in get_products_for_promotion(promotion):
+        assert product.discounted_price_dirty is False
 
 
 def test_update_sale_not_found_error(staff_api_client, permission_manage_discounts):
diff --git a/saleor/graphql/product/bulk_mutations/collection_bulk_delete.py b/saleor/graphql/product/bulk_mutations/collection_bulk_delete.py
index 8b2f02d12c8..d9d710f1d93 100644
--- a/saleor/graphql/product/bulk_mutations/collection_bulk_delete.py
+++ b/saleor/graphql/product/bulk_mutations/collection_bulk_delete.py
@@ -1,9 +1,9 @@
 import graphene
 from django.db.models import Exists, OuterRef
 
+from ....discount.utils import get_active_promotion_rules
 from ....permission.enums import ProductPermissions
 from ....product import models
-from ....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.utils import get_webhooks_for_event
 from ...core.mutations import ModelBulkDeleteMutation
@@ -47,6 +47,4 @@ def bulk_action(cls, info, queryset):
         for product in products:
             cls.call_event(manager.product_updated, product, webhooks=webhooks)
 
-        update_products_discounted_prices_for_promotion_task.delay(
-            [product.id for product in products]
-        )
+        get_active_promotion_rules().update(variants_dirty=True)
diff --git a/saleor/graphql/product/bulk_mutations/product_bulk_create.py b/saleor/graphql/product/bulk_mutations/product_bulk_create.py
index 23fe95f8c64..5c2f1f3d65a 100644
--- a/saleor/graphql/product/bulk_mutations/product_bulk_create.py
+++ b/saleor/graphql/product/bulk_mutations/product_bulk_create.py
@@ -15,11 +15,11 @@
 from ....core.utils import prepare_unique_slug
 from ....core.utils.editorjs import clean_editor_js
 from ....core.utils.validators import get_oembed_data
+from ....discount.utils import get_active_promotion_rules
 from ....permission.enums import ProductPermissions
 from ....product import ProductMediaTypes, models
 from ....product.error_codes import ProductBulkCreateErrorCode
 from ....product.models import CollectionProduct
-from ....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....thumbnail.utils import get_filename_from_url
 from ....warehouse.models import Warehouse
 from ....webhook.event_types import WebhookEventAsyncType
@@ -883,7 +883,7 @@ def post_save_actions(cls, info, products, variants, channels):
         for channel in channels:
             cls.call_event(manager.channel_updated, channel, webhooks=webhooks)
 
-        update_products_discounted_prices_for_promotion_task.delay(product_ids)
+        get_active_promotion_rules().update(variants_dirty=True)
 
     @classmethod
     @traced_atomic_transaction()
diff --git a/saleor/graphql/product/bulk_mutations/product_variant_bulk_create.py b/saleor/graphql/product/bulk_mutations/product_variant_bulk_create.py
index db9e0eb4175..3924b7ff6f6 100644
--- a/saleor/graphql/product/bulk_mutations/product_variant_bulk_create.py
+++ b/saleor/graphql/product/bulk_mutations/product_variant_bulk_create.py
@@ -9,10 +9,10 @@
 
 from ....attribute import AttributeType
 from ....core.tracing import traced_atomic_transaction
+from ....discount.utils import get_active_promotion_rules
 from ....permission.enums import ProductPermissions
 from ....product import models
 from ....product.error_codes import ProductVariantBulkErrorCode
-from ....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....warehouse import models as warehouse_models
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.utils import get_webhooks_for_event
@@ -917,8 +917,9 @@ def prepare_stocks(cls, variant, stocks_input, stocks_to_create):
 
     @classmethod
     def post_save_actions(cls, info, instances, product):
-        # Recalculate the "discounted price" for the parent product
-        update_products_discounted_prices_for_promotion_task.delay([product.pk])
+        # Mark PromotionRule variants dirty
+        # This will finally recalculate discounted prices for products.
+        get_active_promotion_rules().update(variants_dirty=True)
         product.search_index_dirty = True
         product.save(update_fields=["search_index_dirty"])
 
diff --git a/saleor/graphql/product/bulk_mutations/product_variant_bulk_delete.py b/saleor/graphql/product/bulk_mutations/product_variant_bulk_delete.py
index 54772d5c2ee..e54c4b75a99 100644
--- a/saleor/graphql/product/bulk_mutations/product_variant_bulk_delete.py
+++ b/saleor/graphql/product/bulk_mutations/product_variant_bulk_delete.py
@@ -10,13 +10,13 @@
 from ....attribute import models as attribute_models
 from ....core.postgres import FlatConcatSearchVector
 from ....core.tracing import traced_atomic_transaction
+from ....discount.utils import get_active_promotion_rules
 from ....order import events as order_events
 from ....order import models as order_models
 from ....order.tasks import recalculate_orders_task
 from ....permission.enums import ProductPermissions
 from ....product import models
 from ....product.search import prepare_product_search_vector_value
-from ....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.utils import get_webhooks_for_event
 from ...app.dataloaders import get_app_promise
@@ -124,9 +124,9 @@ def perform_mutation(cls, _root, info: ResolveInfo, /, ids=None, skus=None, **da
                     "updated_at",
                 ]
             )
-
-        # Recalculate the "discounted price" for the related products
-        update_products_discounted_prices_for_promotion_task.delay(product_pks)
+        # Mark PromotionRule variants dirty
+        # This will finally recalculate discounted prices for products.
+        get_active_promotion_rules().update(variants_dirty=True)
 
         return response
 
diff --git a/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py b/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
index cb374e6090c..d045781d423 100644
--- a/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
+++ b/saleor/graphql/product/bulk_mutations/product_variant_bulk_update.py
@@ -7,10 +7,10 @@
 from graphene.utils.str_converters import to_camel_case
 
 from ....core.tracing import traced_atomic_transaction
+from ....discount.utils import get_active_promotion_rules
 from ....permission.enums import ProductPermissions
 from ....product import models
 from ....product.error_codes import ProductErrorCode, ProductVariantBulkErrorCode
-from ....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....warehouse import models as warehouse_models
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.utils import get_webhooks_for_event
@@ -699,8 +699,11 @@ def save_variants(cls, variants_data_with_errors_list):
     def post_save_actions(cls, info, instances, product):
         manager = get_plugin_manager_promise(info.context).get()
 
-        # Recalculate the "discounted price" for the parent product
-        update_products_discounted_prices_for_promotion_task.delay([product.pk])
+        # Mark PromotionRule variants dirty
+        # This will finally recalculate discounted prices for products.
+        rules = get_active_promotion_rules()
+        rules.update(variants_dirty=True)
+
         product.search_index_dirty = True
         product.save(update_fields=["search_index_dirty"])
 
diff --git a/saleor/graphql/product/mutations/channels.py b/saleor/graphql/product/mutations/channels.py
index 1281ae7daab..88aa3dc8582 100644
--- a/saleor/graphql/product/mutations/channels.py
+++ b/saleor/graphql/product/mutations/channels.py
@@ -19,7 +19,7 @@
 )
 from ....product.models import Product as ProductModel
 from ....product.models import ProductVariant as ProductVariantModel
-from ....product.tasks import update_discounted_prices_task
+from ....product.utils.product import mark_products_for_recalculate_discounted_price
 from ...channel import ChannelContext
 from ...channel.mutations import BaseChannelListingMutation
 from ...channel.types import Channel
@@ -356,7 +356,7 @@ def save(cls, info: ResolveInfo, product: "ProductModel", cleaned_input: dict):
             cls.update_channels(product, cleaned_input.get("update_channels", []))
             cls.remove_channels(product, cleaned_input.get("remove_channels", []))
             product = ProductModel.objects.prefetched_for_webhook().get(pk=product.pk)
-            update_discounted_prices_task.delay([product.id])
+            mark_products_for_recalculate_discounted_price([product.id])
             manager = get_plugin_manager_promise(info.context).get()
             cls.call_event(manager.product_updated, product)
 
@@ -541,7 +541,7 @@ def save(
                     channel=channel,
                     defaults=defaults,
                 )
-            update_discounted_prices_task.delay([variant.product_id])
+            mark_products_for_recalculate_discounted_price([variant.product_id])
             manager = get_plugin_manager_promise(info.context).get()
             cls.call_event(manager.product_variant_updated, variant)
 
diff --git a/saleor/graphql/product/mutations/collection/collection_add_products.py b/saleor/graphql/product/mutations/collection/collection_add_products.py
index 6bdedac872b..a92efc2b95e 100644
--- a/saleor/graphql/product/mutations/collection/collection_add_products.py
+++ b/saleor/graphql/product/mutations/collection/collection_add_products.py
@@ -2,10 +2,10 @@
 from django.core.exceptions import ValidationError
 
 from .....core.tracing import traced_atomic_transaction
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
 from .....product.error_codes import CollectionErrorCode
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from .....product.utils import get_products_ids_without_variants
 from ....channel import ChannelContext
 from ....core import ResolveInfo
@@ -53,10 +53,10 @@ def perform_mutation(  # type: ignore[override]
         manager = get_plugin_manager_promise(info.context).get()
         with traced_atomic_transaction():
             collection.products.add(*products)
-            # Updated the db entries, recalculating discounts of affected products
-            update_products_discounted_prices_for_promotion_task.delay(
-                [pq.pk for pq in products]
-            )
+            # Mark PromotionRule variants dirty
+            # This will finally recalculate discounted prices for products.
+            rules = get_active_promotion_rules()
+            rules.update(variants_dirty=True)
             for product in products:
                 cls.call_event(manager.product_updated, product)
 
diff --git a/saleor/graphql/product/mutations/collection/collection_create.py b/saleor/graphql/product/mutations/collection/collection_create.py
index 348229fee72..aa7ba5cc1e4 100644
--- a/saleor/graphql/product/mutations/collection/collection_create.py
+++ b/saleor/graphql/product/mutations/collection/collection_create.py
@@ -5,6 +5,7 @@
 from django.core.exceptions import ValidationError
 
 from .....core.utils.date_time import convert_to_utc_date_time
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
 from .....product.error_codes import CollectionErrorCode
@@ -131,6 +132,8 @@ def post_save_action(cls, info: ResolveInfo, instance, cleaned_input):
         for ids_batch in cls.batch_product_ids(product_ids):
             collection_product_updated_task.delay(ids_batch)
 
+        get_active_promotion_rules().update(variants_dirty=True)
+
     @classmethod
     def perform_mutation(cls, _root, info: ResolveInfo, /, **kwargs):
         result = super().perform_mutation(_root, info, **kwargs)
diff --git a/saleor/graphql/product/mutations/collection/collection_delete.py b/saleor/graphql/product/mutations/collection/collection_delete.py
index f987dcda415..5316da89a23 100644
--- a/saleor/graphql/product/mutations/collection/collection_delete.py
+++ b/saleor/graphql/product/mutations/collection/collection_delete.py
@@ -1,11 +1,9 @@
 import graphene
 
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
-from .....product.tasks import (
-    collection_product_updated_task,
-    update_products_discounted_prices_for_promotion_task,
-)
+from .....product.tasks import collection_product_updated_task
 from ....channel import ChannelContext
 from ....core import ResolveInfo
 from ....core.mutations import ModelDeleteMutation
@@ -49,7 +47,8 @@ def perform_mutation(  # type: ignore[override]
 
         for ids_batch in cls.batch_product_ids(product_ids):
             collection_product_updated_task.delay(ids_batch)
-        update_products_discounted_prices_for_promotion_task.delay(product_ids)
+        rules = get_active_promotion_rules()
+        rules.update(variants_dirty=True)
 
         return CollectionDelete(
             collection=ChannelContext(node=result.collection, channel_slug=None)
diff --git a/saleor/graphql/product/mutations/collection/collection_remove_products.py b/saleor/graphql/product/mutations/collection/collection_remove_products.py
index c585106541b..0029dde0ec9 100644
--- a/saleor/graphql/product/mutations/collection/collection_remove_products.py
+++ b/saleor/graphql/product/mutations/collection/collection_remove_products.py
@@ -1,8 +1,8 @@
 import graphene
 
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....channel import ChannelContext
 from ....core import ResolveInfo
 from ....core.doc_category import DOC_CATEGORY_PRODUCTS
@@ -49,10 +49,10 @@ def perform_mutation(  # type: ignore[override]
         manager = get_plugin_manager_promise(info.context).get()
         for product in products:
             cls.call_event(manager.product_updated, product)
-        # Updated the db entries, recalculating discounts of affected products
-        update_products_discounted_prices_for_promotion_task.delay(
-            [p.pk for p in products]
-        )
+        # Mark PromotionRule variants dirty
+        # This will finally recalculate discounted prices for products.
+        get_active_promotion_rules().update(variants_dirty=True)
+
         return CollectionRemoveProducts(
             collection=ChannelContext(node=collection, channel_slug=None)
         )
diff --git a/saleor/graphql/product/mutations/product/product_create.py b/saleor/graphql/product/mutations/product/product_create.py
index 6a360abd284..aadeaf7f712 100644
--- a/saleor/graphql/product/mutations/product/product_create.py
+++ b/saleor/graphql/product/mutations/product/product_create.py
@@ -4,10 +4,10 @@
 from .....attribute import models as attribute_models
 from .....core.tracing import traced_atomic_transaction
 from .....core.utils.editorjs import clean_editor_js
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
 from .....product.error_codes import ProductErrorCode
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....attribute.types import AttributeValueInput
 from ....attribute.utils import AttrValuesInput, ProductAttributeAssignmentMixin
 from ....channel import ChannelContext
@@ -220,7 +220,7 @@ def _save_m2m(cls, _info: ResolveInfo, instance, cleaned_data):
     @classmethod
     def post_save_action(cls, info: ResolveInfo, instance, _cleaned_input):
         product = models.Product.objects.prefetched_for_webhook().get(pk=instance.pk)
-        update_products_discounted_prices_for_promotion_task.delay([instance.id])
+        get_active_promotion_rules().update(variants_dirty=True)
         manager = get_plugin_manager_promise(info.context).get()
         cls.call_event(manager.product_created, product)
 
diff --git a/saleor/graphql/product/mutations/product/product_update.py b/saleor/graphql/product/mutations/product/product_update.py
index 5f74a8a90b7..8a0532cd6ab 100644
--- a/saleor/graphql/product/mutations/product/product_update.py
+++ b/saleor/graphql/product/mutations/product/product_update.py
@@ -1,9 +1,9 @@
 import graphene
 
 from .....attribute import models as attribute_models
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....attribute.utils import AttrValuesInput, ProductAttributeAssignmentMixin
 from ....core import ResolveInfo
 from ....core.descriptions import ADDED_IN_310
@@ -51,7 +51,7 @@ def clean_attributes(
     def post_save_action(cls, info: ResolveInfo, instance, cleaned_input):
         product = models.Product.objects.prefetched_for_webhook().get(pk=instance.pk)
         if "category" in cleaned_input or "collections" in cleaned_input:
-            update_products_discounted_prices_for_promotion_task.delay([instance.id])
+            get_active_promotion_rules().update(variants_dirty=True)
         manager = get_plugin_manager_promise(info.context).get()
         cls.call_event(manager.product_updated, product)
 
diff --git a/saleor/graphql/product/mutations/product_variant/product_variant_create.py b/saleor/graphql/product/mutations/product_variant/product_variant_create.py
index 45f81c85df9..d1fd30a3412 100644
--- a/saleor/graphql/product/mutations/product_variant/product_variant_create.py
+++ b/saleor/graphql/product/mutations/product_variant/product_variant_create.py
@@ -7,10 +7,10 @@
 from .....attribute import AttributeInputType
 from .....attribute import models as attribute_models
 from .....core.tracing import traced_atomic_transaction
+from .....discount.utils import get_active_promotion_rules
 from .....permission.enums import ProductPermissions
 from .....product import models
 from .....product.error_codes import ProductErrorCode
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from .....product.utils.variants import generate_and_set_variant_name
 from ....attribute.types import AttributeValueInput
 from ....attribute.utils import AttributeAssignmentMixin, AttrValuesInput
@@ -314,10 +314,9 @@ def save(cls, info: ResolveInfo, instance, cleaned_input):
             if not instance.product.default_variant:
                 instance.product.default_variant = instance
                 instance.product.save(update_fields=["default_variant", "updated_at"])
-            # Recalculate the "discounted price" for the parent product
-            update_products_discounted_prices_for_promotion_task.delay(
-                [instance.product_id]
-            )
+            # Mark PromotionRule variants dirty
+            # This will finally recalculate discounted prices for products.
+            get_active_promotion_rules().update(variants_dirty=True)
             stocks = cleaned_input.get("stocks")
             if stocks:
                 cls.create_variant_stocks(instance, stocks)
diff --git a/saleor/graphql/product/mutations/product_variant/product_variant_delete.py b/saleor/graphql/product/mutations/product_variant/product_variant_delete.py
index f1fa9ba755e..ffadc3f2f68 100644
--- a/saleor/graphql/product/mutations/product_variant/product_variant_delete.py
+++ b/saleor/graphql/product/mutations/product_variant/product_variant_delete.py
@@ -7,12 +7,12 @@
 from .....attribute import AttributeInputType
 from .....attribute import models as attribute_models
 from .....core.tracing import traced_atomic_transaction
+from .....discount.utils import get_active_promotion_rules
 from .....order import events as order_events
 from .....order import models as order_models
 from .....order.tasks import recalculate_orders_task
 from .....permission.enums import ProductPermissions
 from .....product import models
-from .....product.tasks import update_products_discounted_prices_for_promotion_task
 from ....app.dataloaders import get_app_promise
 from ....channel import ChannelContext
 from ....core import ResolveInfo
@@ -50,10 +50,9 @@ class Meta:
 
     @classmethod
     def success_response(cls, instance):
-        # Update the "discounted_prices" of the parent product
-        update_products_discounted_prices_for_promotion_task.delay(
-            [instance.product_id]
-        )
+        # Mark PromotionRule variants dirty
+        # This will finally recalculate discounted prices for products.
+        get_active_promotion_rules().update(variants_dirty=True)
         product = models.Product.objects.get(id=instance.product_id)
         product.search_index_dirty = True
         product.save(update_fields=["search_index_dirty"])
diff --git a/saleor/graphql/product/tests/test_bulk_delete.py b/saleor/graphql/product/tests/test_bulk_delete.py
index f34b541aef7..bba08fa6d3f 100644
--- a/saleor/graphql/product/tests/test_bulk_delete.py
+++ b/saleor/graphql/product/tests/test_bulk_delete.py
@@ -38,11 +38,7 @@
 """
 
 
-@patch(
-    "saleor.product.tasks.update_products_discounted_prices_for_promotion_task.delay"
-)
 def test_delete_categories(
-    update_products_discounted_price_task_mock,
     staff_api_client,
     category_list,
     product_list,
@@ -75,9 +71,6 @@ def test_delete_categories(
     assert not Category.objects.filter(
         id__in=[category.id for category in category_list]
     ).exists()
-    update_products_discounted_price_task_mock.assert_called_once()
-    args, kwargs = update_products_discounted_price_task_mock.call_args
-    assert set(kwargs["product_ids"]) == {product.id for product in product_list}
 
 
 @patch("saleor.product.utils.get_webhooks_for_event")
@@ -204,9 +197,7 @@ def test_delete_categories_trigger_product_updated_webhook(
     assert product_updated_mock.call_count == 2
 
 
-@patch("saleor.product.utils.update_products_discounted_prices_for_promotion_task")
 def test_delete_categories_with_subcategories_and_products(
-    mock_update_products_discounted_prices_for_promotion_task,
     staff_api_client,
     category_list,
     permission_manage_products,
@@ -259,14 +250,6 @@ def test_delete_categories_with_subcategories_and_products(
         id__in=[category.id for category in category_list]
     ).exists()
 
-    mock_update_products_discounted_prices_for_promotion_task.delay.assert_called_once()
-    (
-        _call_args,
-        call_kwargs,
-    ) = mock_update_products_discounted_prices_for_promotion_task.delay.call_args
-
-    assert set(call_kwargs["product_ids"]) == set([p.pk for p in product_list])
-
     for product in product_list:
         product.refresh_from_db()
         assert not product.category
@@ -279,6 +262,9 @@ def test_delete_categories_with_subcategories_and_products(
         assert not product_channel_listing.published_at
     assert product_channel_listings.count() == 3
 
+    product.refresh_from_db()
+    assert product.discounted_price_dirty is False
+
 
 MUTATION_COLLECTION_BULK_DELETE = """
     mutation collectionBulkDelete($ids: [ID!]!) {
@@ -289,11 +275,7 @@ def test_delete_categories_with_subcategories_and_products(
 """
 
 
-@patch(
-    "saleor.product.tasks.update_products_discounted_prices_for_promotion_task.delay"
-)
 def test_delete_collections(
-    update_products_discounted_price_task_mock,
     staff_api_client,
     collection_list,
     product_list,
@@ -324,9 +306,10 @@ def test_delete_collections(
     assert not Collection.objects.filter(
         id__in=[collection.id for collection in collection_list]
     ).exists()
-    update_products_discounted_price_task_mock.assert_called_once()
-    args = set(update_products_discounted_price_task_mock.call_args.args[0])
-    assert args == {product.id for product in product_list}
+
+    for product in product_list:
+        product.refresh_from_db()
+        assert product.discounted_price_dirty is False
 
 
 def test_delete_collections_with_images(
@@ -991,15 +974,11 @@ def test_delete_product_variants_by_sku(
     "saleor.graphql.product.bulk_mutations."
     "product_variant_bulk_delete.get_webhooks_for_event"
 )
-@patch(
-    "saleor.product.tasks.update_products_discounted_prices_for_promotion_task.delay"
-)
 @patch("saleor.plugins.manager.PluginsManager.product_variant_deleted")
 @patch("saleor.order.tasks.recalculate_orders_task.delay")
 def test_delete_product_variants_by_sku_task_for_recalculate_product_prices_called(
     mocked_recalculate_orders_task,
     product_variant_deleted_webhook_mock,
-    update_products_discounted_price_task_mock,
     mocked_get_webhooks_for_event,
     staff_api_client,
     product_list,
@@ -1036,9 +1015,9 @@ def test_delete_product_variants_by_sku_task_for_recalculate_product_prices_call
         == content["data"]["productVariantBulkDelete"]["count"]
     )
     mocked_recalculate_orders_task.assert_not_called()
-    update_products_discounted_price_task_mock.assert_called_once()
-    args = set(update_products_discounted_price_task_mock.call_args.args[0])
-    assert args == {product.id for product in product_list}
+    for product in product_list:
+        product.refresh_from_db()
+        assert product.discounted_price_dirty is False
 
 
 PRODUCT_VARIANT_BULK_DELETE_MUTATION = """
@@ -1112,15 +1091,11 @@ def test_delete_product_variants(
     "saleor.graphql.product.bulk_mutations."
     "product_variant_bulk_delete.get_webhooks_for_event"
 )
-@patch(
-    "saleor.product.tasks.update_products_discounted_prices_for_promotion_task.delay"
-)
 @patch("saleor.plugins.manager.PluginsManager.product_variant_deleted")
 @patch("saleor.order.tasks.recalculate_orders_task.delay")
 def test_delete_product_variants_task_for_recalculate_product_prices_called(
     mocked_recalculate_orders_task,
     product_variant_deleted_webhook_mock,
-    update_products_discounted_price_task_mock,
     mocked_get_webhooks_for_event,
     staff_api_client,
     product_list,
@@ -1159,9 +1134,9 @@ def test_delete_product_variants_task_for_recalculate_product_prices_called(
         == content["data"]["productVariantBulkDelete"]["count"]
     )
     mocked_recalculate_orders_task.assert_not_called()
-    update_products_discounted_price_task_mock.assert_called_once()
-    args = set(update_products_discounted_price_task_mock.call_args.args[0])
-    assert args == {product.id for product in product_list}
+    for product in product_list:
+        product.refresh_from_db()
+        assert product.discounted_price_dirty is False
 
 
 @patch(
diff --git a/saleor/product/migrations/0191_product_discounted_price_dirty.py b/saleor/product/migrations/0191_product_discounted_price_dirty.py
new file mode 100644
index 00000000000..e32282d613d
--- /dev/null
+++ b/saleor/product/migrations/0191_product_discounted_price_dirty.py
@@ -0,0 +1,25 @@
+# Generated by Django 3.2.23 on 2024-01-09 08:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("product", "0190_merge_20231221_1356"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="product",
+            name="discounted_price_dirty",
+            field=models.BooleanField(default=False),
+        ),
+        migrations.RunSQL(
+            """
+            ALTER TABLE product_product
+            ALTER COLUMN discounted_price_dirty
+            SET DEFAULT false;
+            """,
+            migrations.RunSQL.noop,
+        ),
+    ]
diff --git a/saleor/product/models.py b/saleor/product/models.py
index fe87b8d4ece..2a2a5ce3e68 100644
--- a/saleor/product/models.py
+++ b/saleor/product/models.py
@@ -207,6 +207,7 @@ class Product(SeoModel, ModelWithMetadata, ModelWithExternalReference):
         null=True,
         on_delete=models.SET_NULL,
     )
+    discounted_price_dirty = models.BooleanField(default=False)
 
     objects = managers.ProductManager()
 
diff --git a/saleor/product/tasks.py b/saleor/product/tasks.py
index 970e3af37ba..0d8abe6d5b1 100644
--- a/saleor/product/tasks.py
+++ b/saleor/product/tasks.py
@@ -20,6 +20,7 @@
 from ..webhook.utils import get_webhooks_for_event
 from .models import Product, ProductType, ProductVariant
 from .search import PRODUCTS_BATCH_SIZE, update_products_search_vector
+from .utils.product import mark_products_for_recalculate_discounted_price
 from .utils.variant_prices import update_discounted_prices_for_promotion
 from .utils.variants import (
     fetch_variants_for_promotion_rules,
@@ -100,6 +101,16 @@ def update_products_discounted_prices_of_promotion_task(promotion_pk: UUID):
     update_discounted_prices_task.delay(list(products.values_list("id", flat=True)))
 
 
+@app.task
+def mark_promotion_rules_to_recalculate_variants(promotion_pk: UUID):
+    try:
+        promotion = Promotion.objects.get(pk=promotion_pk)
+    except ObjectDoesNotExist:
+        logging.warning(f"Cannot find promotion with id: {promotion_pk}.")
+        return
+    promotion.rules.all().update(variants_dirty=True)
+
+
 @app.task
 def update_products_discounted_prices_for_promotion_task(
     product_ids: Iterable[int],
@@ -140,6 +151,54 @@ def update_products_discounted_prices_for_promotion_task(
         update_discounted_prices_task.delay(product_ids)
 
 
+@app.task
+def update_promotion_rules_mark_products_for_price_recalculation_task(
+    product_ids: Iterable[int] = [],
+    start_id: Optional[UUID] = None,
+):
+    """Update relations between PromotionRule and ProductVariant.
+
+    Mark the Product discounted prices for recalculation when all PromotionRules
+    are updated.
+    """
+    promotions = Promotion.objects.using(
+        settings.DATABASE_CONNECTION_REPLICA_NAME
+    ).active()
+    kwargs = {"id__gt": start_id} if start_id else {}
+    rules = (
+        PromotionRule.objects.order_by("id")
+        .using(settings.DATABASE_CONNECTION_REPLICA_NAME)
+        .filter(
+            Exists(promotions.filter(id=OuterRef("promotion_id"))),
+            variants_dirty=True,
+            **kwargs,
+        )[:PROMOTION_RULE_BATCH_SIZE]
+    )
+    if ids := list(rules.values_list("pk", flat=True)):
+        qs = PromotionRule.objects.filter(pk__in=ids)
+        product_ids = fetch_variants_for_promotion_rules(rules=qs)
+        qs.update(variants_dirty=False)
+        update_products_discounted_prices_for_promotion_task.delay(product_ids, ids[-1])
+    else:
+        # when all promotion rules variants are up to date, mark products for
+        if product_ids:
+            mark_products_for_recalculate_discounted_price(product_ids)
+
+
+@app.task
+def recalculate_discounted_price_for_products_task():
+    """Recalculate discounted price for products."""
+    products_ids = list(
+        Product.objects.filter(discounted_price_dirty=True)
+        .order_by("id")
+        .values_list("id", flat=True)
+    )[:DISCOUNTED_PRODUCT_BATCH]
+    if products := Product.objects.filter(id__in=products_ids):
+        update_discounted_prices_for_promotion(products)
+        products.update(discounted_price_dirty=False)
+        recalculate_discounted_price_for_products_task.delay()
+
+
 @app.task
 def update_discounted_prices_task(product_ids: Iterable[int]):
     """Update the product discounted prices for given product ids."""
diff --git a/saleor/product/tests/test_managers.py b/saleor/product/tests/test_managers.py
index f197f996eec..3f724bb5a29 100644
--- a/saleor/product/tests/test_managers.py
+++ b/saleor/product/tests/test_managers.py
@@ -1,4 +1,4 @@
-from saleor.product.models import ProductVariant
+from ...product.models import ProductVariant
 
 
 def test_available_quantity_with_no_allocations(variant, allocations):
diff --git a/saleor/product/utils/__init__.py b/saleor/product/utils/__init__.py
index 43719d447e1..0b1150b273e 100644
--- a/saleor/product/utils/__init__.py
+++ b/saleor/product/utils/__init__.py
@@ -4,10 +4,10 @@
 from ...core.taxes import TaxedMoney, zero_taxed_money
 from ...core.tracing import traced_atomic_transaction
 from ...core.utils.events import call_event
+from ...discount.utils import get_active_promotion_rules
 from ...webhook.event_types import WebhookEventAsyncType
 from ...webhook.utils import get_webhooks_for_event
 from ..models import Product, ProductChannelListing
-from ..tasks import update_products_discounted_prices_for_promotion_task
 
 if TYPE_CHECKING:
     from datetime import date, datetime
@@ -65,9 +65,7 @@ def delete_categories(categories_ids: list[Union[str, int]], manager):
     for product in products:
         call_event(manager.product_updated, product, webhooks=webhooks)
 
-    update_products_discounted_prices_for_promotion_task.delay(
-        product_ids=[product.id for product in products]
-    )
+    get_active_promotion_rules().update(variants_dirty=True)
 
 
 def collect_categories_tree_products(category: "Category") -> "QuerySet[Product]":
diff --git a/saleor/product/utils/product.py b/saleor/product/utils/product.py
new file mode 100644
index 00000000000..bc71db5b1ee
--- /dev/null
+++ b/saleor/product/utils/product.py
@@ -0,0 +1,8 @@
+from collections.abc import Iterable
+
+from ...product.models import Product
+
+
+def mark_products_for_recalculate_discounted_price(product_ids: Iterable[int]):
+    """Mark products for recalculate discounted prices."""
+    Product.objects.filter(pk__in=product_ids).update(discounted_price_dirty=True)
diff --git a/saleor/product/utils/variants.py b/saleor/product/utils/variants.py
index 60c0629b74a..1e624bf6067 100644
--- a/saleor/product/utils/variants.py
+++ b/saleor/product/utils/variants.py
@@ -52,13 +52,15 @@ def get_variant_selection_attributes(
     ]
 
 
-def fetch_variants_for_promotion_rules(rules: QuerySet[PromotionRule]):
+def fetch_variants_for_promotion_rules(rules: QuerySet[PromotionRule]) -> list[int]:
     from ...graphql.discount.utils import get_variants_for_predicate
 
     PromotionRuleVariant = PromotionRule.variants.through
     new_rules_variants = []
+    product_ids: list[int] = []
     for rule in list(rules.iterator()):
         variants = get_variants_for_predicate(rule.catalogue_predicate)
+        product_ids += variants.values_list("product_id", flat=True)
         new_rules_variants.extend(
             [
                 PromotionRuleVariant(
@@ -68,3 +70,4 @@ def fetch_variants_for_promotion_rules(rules: QuerySet[PromotionRule]):
             ]
         )
     update_rule_variant_relation(rules, new_rules_variants)
+    return list(set(product_ids))
diff --git a/saleor/settings.py b/saleor/settings.py
index 043e87172b2..126c98e74fa 100644
--- a/saleor/settings.py
+++ b/saleor/settings.py
@@ -627,6 +627,17 @@ def get_url_from_env(name, *, schemes=None) -> Optional[str]:
         "task": "saleor.payment.tasks.transaction_release_funds_for_checkout_task",
         "schedule": timedelta(minutes=10),
     },
+    "update-promotion-rules-and-mark-products-for-price-recalculation": {
+        "task": (
+            "saleor.product.tasks"
+            ".update_promotion_rules_mark_products_for_price_recalculation_task"
+        ),
+        "schedule": timedelta(seconds=10),
+    },
+    "recalculate-discounted-price-for-products": {
+        "task": "saleor.product.tasks.recalculate_discounted_price_for_products_task",
+        "schedule": timedelta(seconds=10),
+    },
 }
 
 # The maximum wait time between each is_due() call on schedulers


Code review comments:
diff_hunk: @@ -47,6 +47,4 @@ def bulk_action(cls, info, queryset):
         for product in products:
             cls.call_event(manager.product_updated, product, webhooks=webhooks)
 
-        update_products_discounted_prices_for_promotion_task.delay(
-            [product.id for product in products]
-        )
+        get_active_promotion_rules().update(variants_dirty=True)
path: saleor/graphql/product/bulk_mutations/collection_bulk_delete.py
text: @user1:
This will make ALL existing rules marked as dirty, which means, that we will re-calculate everything. 

@user2:
Yep, but we did it already and it was done each time any catalogue have been changed. No, we will mark the rules with flag and we will significantly reduce the number of proceeded promotion rules.
When something have changed in catalogues we have no clue which promotion rules should be marked to recalculate, as rules include filters and we do not know if the given product, variant, category, or collection is included in any of this rule until we filter be the rule predicate

diff_hunk: @@ -100,6 +101,16 @@ def update_products_discounted_prices_of_promotion_task(promotion_pk: UUID):
     update_discounted_prices_task.delay(list(products.values_list("id", flat=True)))
 
 
+@app.task
+def mark_promotion_rules_to_recalculate_variants(promotion_pk: UUID):
path: saleor/product/tasks.py
text: @user1:
Do we need to have this as celery task? It looks like you're calling it as normal function.

diff_hunk: @@ -308,7 +309,7 @@ def update_discounted_prices_task():
     if products_ids:
         products = Product.objects.filter(id__in=products_ids)
         update_discounted_prices_for_promotion(products)
-        update_discounted_prices_task.delay()
+        mark_products_for_recalculate_discounted_price(products_ids)
path: saleor/discount/tasks.py
text: @user1:
I might be wrong but it looks like we will end up in infitie loop. We just processed the `products_ids`, and now we will mark it for recalculate