From 5f41f6ff34840a94d4487bdfdc5a93e95ae73d9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E7=8E=89=E9=91=AB?= Date: Fri, 24 Jul 2026 18:27:54 +0800 Subject: [PATCH 1/3] refactor: optimize near_deadline automation trigger to process individual rows - split near_deadline scheduled tasks into separate per-row tasks (max 25) - query near-deadline rows using dtable-db API when scheduling - invalidate automation rules when metadata validation fails - remove unused imports and constants - add CONDITION_NEAR_DEADLINE to can_do_actions check --- .../automations/automations_pipeline.py | 79 ++++++++++++++++--- dtable_events/automations/entities.py | 12 ++- 2 files changed, 78 insertions(+), 13 deletions(-) diff --git a/dtable_events/automations/automations_pipeline.py b/dtable_events/automations/automations_pipeline.py index d6f40dd7..acebe900 100644 --- a/dtable_events/automations/automations_pipeline.py +++ b/dtable_events/automations/automations_pipeline.py @@ -1,8 +1,5 @@ import json -import os import time -from copy import deepcopy -from dataclasses import dataclass, field from datetime import datetime, timedelta, date from threading import Thread, Lock @@ -10,7 +7,7 @@ from sqlalchemy import text from dtable_events.app.config import INNER_DTABLE_WEB_SERVICE_URL, AUTOMATION_RATE_LIMIT_PERCENT, \ - AUTOMATION_RATE_LIMIT_WINDOW_SECS, AUTOMATION_WORKERS + AUTOMATION_RATE_LIMIT_WINDOW_SECS, AUTOMATION_WORKERS, INNER_DTABLE_DB_URL from dtable_events.app.event_redis import RedisClient from dtable_events.app.log import auto_rule_logger from dtable_events.automations.automations_stats_manager import AutomationsStatsManager @@ -18,10 +15,14 @@ from dtable_events.db import init_db_session_class from dtable_events.utils import get_dtable_owner_org_id from dtable_events.utils.dtable_web_api import DTableWebAPI +from dtable_events.utils.dtable_db_api import DTableDBAPI +from dtable_events.utils.utils_metadata_cache import get_metadata +from dtable_events.notification_rules.notification_rules_utils import list_rows_near_deadline_with_dtable_db from dtable_events.utils.utils_metric import AUTOMATION_QUEUE_10_METRIC_HELP, AUTOMATION_QUEUE_20_METRIC_HELP, \ AUTOMATION_QUEUE_30_METRIC_HELP, REALTIME_AUTOMATION_RULES_HEARTBEAT_HELP, \ REALTIME_AUTOMATION_RULES_TRIGGERED_COUNT_HELP, SCHEDULED_AUTOMATION_RULES_TRIGGERED_COUNT_HELP, publish_metric -from dtable_events.automations.entities import AutomationResult, AutomationTask, QUEUE_AUTOMATION_TASKS_10, QUEUE_AUTOMATION_TASKS_20, QUEUE_AUTOMATION_TASKS_30 +from dtable_events.automations.entities import AutomationResult, AutomationTask, CONDITION_NEAR_DEADLINE, \ + QUEUE_AUTOMATION_TASKS_10, QUEUE_AUTOMATION_TASKS_20, QUEUE_AUTOMATION_TASKS_30 class RateLimiter: @@ -282,6 +283,67 @@ def scan_rules(self): cached_exceed_keys_set = set() gen_exceed_key = lambda owner, org_id: org_id if org_id != -1 else owner + def put_scheduled_tasks(automation_task): + if automation_task.trigger.get('condition') != CONDITION_NEAR_DEADLINE: + self.put_task(automation_task) + return 1 + + trigger = automation_task.trigger + try: + dtable_metadata = get_metadata(automation_task.dtable_uuid) + dtable_db_api = DTableDBAPI('automation-rule', automation_task.dtable_uuid, INNER_DTABLE_DB_URL) + rows, _, is_valid = list_rows_near_deadline_with_dtable_db( + dtable_metadata, + trigger.get('table_id'), + trigger.get('view_id'), + trigger.get('date_column_name'), + trigger.get('alarm_days'), + dtable_db_api, + ) + except Exception as e: + auto_rule_logger.exception( + 'List near-deadline rows failed rule_id=%s dtable_uuid=%s: %s', + automation_task.rule_id, + automation_task.dtable_uuid, + e, + ) + return 0 + if not is_valid: + db_session.execute( + text('UPDATE dtable_automation_rules SET is_valid=0 WHERE id=:rule_id'), + {'rule_id': automation_task.rule_id}, + ) + return 0 + + table_name = next((table.get('name') for table in dtable_metadata.get('tables', []) + if table.get('_id') == trigger.get('table_id')), '') + task_count = 0 + for row in rows[:25]: + if not row.get('_id'): + continue + row_task = AutomationTask( + rule_id=automation_task.rule_id, + run_condition=automation_task.run_condition, + trigger=automation_task.trigger, + actions=automation_task.actions, + dtable_uuid=automation_task.dtable_uuid, + org_id=automation_task.org_id, + owner=automation_task.owner, + data={ + 'op_type': CONDITION_NEAR_DEADLINE, + 'table_id': trigger.get('table_id'), + 'table_name': table_name, + 'row_id': row.get('_id'), + 'updated_column_keys': [], + 'dtable_uuid': automation_task.dtable_uuid, + 'automation_rule_id': automation_task.rule_id, + }, + with_test=False, + ) + self.put_task(row_task) + task_count += 1 + return task_count + try: for rule in rules: automation_task = AutomationTask( @@ -301,14 +363,13 @@ def scan_rules(self): if exceed_key in cached_exceed_keys_set: continue if isinstance(exceed_key, str) and '@seafile_group' in exceed_key: - self.put_task(automation_task) - self.scheduled_trigger_count += 1 + self.scheduled_trigger_count += put_scheduled_tasks(automation_task) continue if self.automations_stats_manager.is_exceed(db_session, rule.owner, rule.org_id): cached_exceed_keys_set.add(exceed_key) continue - self.put_task(automation_task) - self.scheduled_trigger_count += 1 + self.scheduled_trigger_count += put_scheduled_tasks(automation_task) + db_session.commit() except Exception as e: auto_rule_logger.exception(e) finally: diff --git a/dtable_events/automations/entities.py b/dtable_events/automations/entities.py index 7b1ac6f5..e2ca2828 100644 --- a/dtable_events/automations/entities.py +++ b/dtable_events/automations/entities.py @@ -8,13 +8,12 @@ PER_UPDATE = 'per_update' PER_MONTH = 'per_month' CRON_CONDITIONS = (PER_DAY, PER_WEEK, PER_MONTH) -ALL_CONDITIONS = (PER_DAY, PER_WEEK, PER_MONTH, PER_UPDATE) -CONDITION_ROWS_MODIFIED = 'rows_modified' CONDITION_ROWS_ADDED = 'rows_added' CONDITION_FILTERS_SATISFY = 'filters_satisfy' CONDITION_PERIODICALLY = 'run_periodically' CONDITION_PERIODICALLY_BY_CONDITION = 'run_periodically_by_condition' +CONDITION_NEAR_DEADLINE = 'near_deadline' QUEUE_AUTOMATION_TASKS_10 = 'automation_tasks_10' QUEUE_AUTOMATION_TASKS_20 = 'automation_tasks_20' @@ -45,14 +44,19 @@ def append_warning(self, warning): self.warnings.append(warning) def can_do_actions(self): - if self.trigger.get('condition') not in (CONDITION_FILTERS_SATISFY, CONDITION_PERIODICALLY, CONDITION_ROWS_ADDED, CONDITION_PERIODICALLY_BY_CONDITION): + if self.trigger.get('condition') not in ( + CONDITION_FILTERS_SATISFY, + CONDITION_PERIODICALLY, + CONDITION_ROWS_ADDED, + CONDITION_PERIODICALLY_BY_CONDITION, + CONDITION_NEAR_DEADLINE): return False if self.trigger.get('condition') == CONDITION_ROWS_ADDED: if self.data.get('op_type') not in ['insert_row', 'append_rows', 'insert_rows']: return False - if self.trigger.get('condition') in [CONDITION_FILTERS_SATISFY, CONDITION_ROWS_MODIFIED]: + if self.trigger.get('condition') == CONDITION_FILTERS_SATISFY: if self.data.get('op_type') not in ['modify_row', 'modify_rows', 'add_link', 'update_links', 'update_rows_links', 'remove_link', 'move_group_rows']: return False From 4999eb29964a9aca35fe34b35a90b7f336488b87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E7=8E=89=E9=91=AB?= Date: Sat, 25 Jul 2026 13:27:12 +0800 Subject: [PATCH 2/3] refactor: extract cron time matching logic to separate method - add is_cron_time_matched() method to AutomationTask - check scheduled time before querying rows in near-deadline automation pipeline - refactor can_do_actions() to reuse is_cron_time_matched() --- .../automations/automations_pipeline.py | 5 +++ dtable_events/automations/entities.py | 35 +++++++++---------- 2 files changed, 21 insertions(+), 19 deletions(-) diff --git a/dtable_events/automations/automations_pipeline.py b/dtable_events/automations/automations_pipeline.py index acebe900..34f1a7d9 100644 --- a/dtable_events/automations/automations_pipeline.py +++ b/dtable_events/automations/automations_pipeline.py @@ -288,6 +288,11 @@ def put_scheduled_tasks(automation_task): self.put_task(automation_task) return 1 + # near-deadline rules have to query rows from dtable-db before tasks are put into queue, + # so check the scheduled time here in advance to avoid querying rows every scan + if not automation_task.is_cron_time_matched(): + return 0 + trigger = automation_task.trigger try: dtable_metadata = get_metadata(automation_task.dtable_uuid) diff --git a/dtable_events/automations/entities.py b/dtable_events/automations/entities.py index e2ca2828..53f9b1c8 100644 --- a/dtable_events/automations/entities.py +++ b/dtable_events/automations/entities.py @@ -43,6 +43,21 @@ def to_dict(self): def append_warning(self, warning): self.warnings.append(warning) + def is_cron_time_matched(self): + if self.run_condition not in CRON_CONDITIONS: + return False + cur_datetime = datetime.now() + cur_hour = cur_datetime.hour + cur_week_day = cur_datetime.isoweekday() + cur_month_day = cur_datetime.day + if self.run_condition == PER_DAY: + return cur_hour == self.trigger.get('notify_hour', 12) + if self.run_condition == PER_WEEK: + return cur_hour == self.trigger.get('notify_week_hour', 12) \ + and cur_week_day == self.trigger.get('notify_week_day', 7) + return cur_hour == self.trigger.get('notify_month_hour', 12) \ + and cur_month_day == self.trigger.get('notify_month_day', 1) + def can_do_actions(self): if self.trigger.get('condition') not in ( CONDITION_FILTERS_SATISFY, @@ -64,25 +79,7 @@ def can_do_actions(self): return True if self.run_condition in CRON_CONDITIONS: - cur_datetime = datetime.now() - cur_hour = cur_datetime.hour - cur_week_day = cur_datetime.isoweekday() - cur_month_day = cur_datetime.day - if self.run_condition == PER_DAY: - trigger_hour = self.trigger.get('notify_hour', 12) - if cur_hour != trigger_hour: - return False - elif self.run_condition == PER_WEEK: - trigger_hour = self.trigger.get('notify_week_hour', 12) - trigger_day = self.trigger.get('notify_week_day', 7) - if cur_hour != trigger_hour or cur_week_day != trigger_day: - return False - else: - trigger_hour = self.trigger.get('notify_month_hour', 12) - trigger_day = self.trigger.get('notify_month_day', 1) - if cur_hour != trigger_hour or cur_month_day != trigger_day: - return False - return True + return self.is_cron_time_matched() return False From ee37a2f9904c0d530b507315ac070d5b4a59779d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E7=8E=89=E9=91=AB?= Date: Mon, 27 Jul 2026 13:48:59 +0800 Subject: [PATCH 3/3] revert: remove near-deadline automation row-level task splitting - remove per-row task generation for near_deadline scheduled tasks - restore simple task queuing for all scheduled automation rules - remove dtable-db API integration from automation pipeline - remove CONDITION_NEAR_DEADLINE constant and related logic --- .../automations/automations_pipeline.py | 80 ++----------------- dtable_events/automations/entities.py | 4 +- dtable_events/automations/general_actions.py | 1 - 3 files changed, 7 insertions(+), 78 deletions(-) diff --git a/dtable_events/automations/automations_pipeline.py b/dtable_events/automations/automations_pipeline.py index 34f1a7d9..53b5c30f 100644 --- a/dtable_events/automations/automations_pipeline.py +++ b/dtable_events/automations/automations_pipeline.py @@ -7,7 +7,7 @@ from sqlalchemy import text from dtable_events.app.config import INNER_DTABLE_WEB_SERVICE_URL, AUTOMATION_RATE_LIMIT_PERCENT, \ - AUTOMATION_RATE_LIMIT_WINDOW_SECS, AUTOMATION_WORKERS, INNER_DTABLE_DB_URL + AUTOMATION_RATE_LIMIT_WINDOW_SECS, AUTOMATION_WORKERS from dtable_events.app.event_redis import RedisClient from dtable_events.app.log import auto_rule_logger from dtable_events.automations.automations_stats_manager import AutomationsStatsManager @@ -15,13 +15,10 @@ from dtable_events.db import init_db_session_class from dtable_events.utils import get_dtable_owner_org_id from dtable_events.utils.dtable_web_api import DTableWebAPI -from dtable_events.utils.dtable_db_api import DTableDBAPI -from dtable_events.utils.utils_metadata_cache import get_metadata -from dtable_events.notification_rules.notification_rules_utils import list_rows_near_deadline_with_dtable_db from dtable_events.utils.utils_metric import AUTOMATION_QUEUE_10_METRIC_HELP, AUTOMATION_QUEUE_20_METRIC_HELP, \ AUTOMATION_QUEUE_30_METRIC_HELP, REALTIME_AUTOMATION_RULES_HEARTBEAT_HELP, \ REALTIME_AUTOMATION_RULES_TRIGGERED_COUNT_HELP, SCHEDULED_AUTOMATION_RULES_TRIGGERED_COUNT_HELP, publish_metric -from dtable_events.automations.entities import AutomationResult, AutomationTask, CONDITION_NEAR_DEADLINE, \ +from dtable_events.automations.entities import AutomationResult, AutomationTask, \ QUEUE_AUTOMATION_TASKS_10, QUEUE_AUTOMATION_TASKS_20, QUEUE_AUTOMATION_TASKS_30 @@ -283,72 +280,6 @@ def scan_rules(self): cached_exceed_keys_set = set() gen_exceed_key = lambda owner, org_id: org_id if org_id != -1 else owner - def put_scheduled_tasks(automation_task): - if automation_task.trigger.get('condition') != CONDITION_NEAR_DEADLINE: - self.put_task(automation_task) - return 1 - - # near-deadline rules have to query rows from dtable-db before tasks are put into queue, - # so check the scheduled time here in advance to avoid querying rows every scan - if not automation_task.is_cron_time_matched(): - return 0 - - trigger = automation_task.trigger - try: - dtable_metadata = get_metadata(automation_task.dtable_uuid) - dtable_db_api = DTableDBAPI('automation-rule', automation_task.dtable_uuid, INNER_DTABLE_DB_URL) - rows, _, is_valid = list_rows_near_deadline_with_dtable_db( - dtable_metadata, - trigger.get('table_id'), - trigger.get('view_id'), - trigger.get('date_column_name'), - trigger.get('alarm_days'), - dtable_db_api, - ) - except Exception as e: - auto_rule_logger.exception( - 'List near-deadline rows failed rule_id=%s dtable_uuid=%s: %s', - automation_task.rule_id, - automation_task.dtable_uuid, - e, - ) - return 0 - if not is_valid: - db_session.execute( - text('UPDATE dtable_automation_rules SET is_valid=0 WHERE id=:rule_id'), - {'rule_id': automation_task.rule_id}, - ) - return 0 - - table_name = next((table.get('name') for table in dtable_metadata.get('tables', []) - if table.get('_id') == trigger.get('table_id')), '') - task_count = 0 - for row in rows[:25]: - if not row.get('_id'): - continue - row_task = AutomationTask( - rule_id=automation_task.rule_id, - run_condition=automation_task.run_condition, - trigger=automation_task.trigger, - actions=automation_task.actions, - dtable_uuid=automation_task.dtable_uuid, - org_id=automation_task.org_id, - owner=automation_task.owner, - data={ - 'op_type': CONDITION_NEAR_DEADLINE, - 'table_id': trigger.get('table_id'), - 'table_name': table_name, - 'row_id': row.get('_id'), - 'updated_column_keys': [], - 'dtable_uuid': automation_task.dtable_uuid, - 'automation_rule_id': automation_task.rule_id, - }, - with_test=False, - ) - self.put_task(row_task) - task_count += 1 - return task_count - try: for rule in rules: automation_task = AutomationTask( @@ -368,13 +299,14 @@ def put_scheduled_tasks(automation_task): if exceed_key in cached_exceed_keys_set: continue if isinstance(exceed_key, str) and '@seafile_group' in exceed_key: - self.scheduled_trigger_count += put_scheduled_tasks(automation_task) + self.put_task(automation_task) + self.scheduled_trigger_count += 1 continue if self.automations_stats_manager.is_exceed(db_session, rule.owner, rule.org_id): cached_exceed_keys_set.add(exceed_key) continue - self.scheduled_trigger_count += put_scheduled_tasks(automation_task) - db_session.commit() + self.put_task(automation_task) + self.scheduled_trigger_count += 1 except Exception as e: auto_rule_logger.exception(e) finally: diff --git a/dtable_events/automations/entities.py b/dtable_events/automations/entities.py index 53f9b1c8..4ce6e616 100644 --- a/dtable_events/automations/entities.py +++ b/dtable_events/automations/entities.py @@ -13,7 +13,6 @@ CONDITION_FILTERS_SATISFY = 'filters_satisfy' CONDITION_PERIODICALLY = 'run_periodically' CONDITION_PERIODICALLY_BY_CONDITION = 'run_periodically_by_condition' -CONDITION_NEAR_DEADLINE = 'near_deadline' QUEUE_AUTOMATION_TASKS_10 = 'automation_tasks_10' QUEUE_AUTOMATION_TASKS_20 = 'automation_tasks_20' @@ -63,8 +62,7 @@ def can_do_actions(self): CONDITION_FILTERS_SATISFY, CONDITION_PERIODICALLY, CONDITION_ROWS_ADDED, - CONDITION_PERIODICALLY_BY_CONDITION, - CONDITION_NEAR_DEADLINE): + CONDITION_PERIODICALLY_BY_CONDITION): return False if self.trigger.get('condition') == CONDITION_ROWS_ADDED: diff --git a/dtable_events/automations/general_actions.py b/dtable_events/automations/general_actions.py index 9eafd031..c5112a2c 100644 --- a/dtable_events/automations/general_actions.py +++ b/dtable_events/automations/general_actions.py @@ -34,7 +34,6 @@ CONDITION_ROWS_MODIFIED = 'rows_modified' CONDITION_ROWS_ADDED = 'rows_added' CONDITION_FILTERS_SATISFY = 'filters_satisfy' -CONDITION_NEAR_DEADLINE = 'near_deadline' CONDITION_PERIODICALLY = 'run_periodically' CONDITION_PERIODICALLY_BY_CONDITION = 'run_periodically_by_condition'