Skip to content

fix: isolate automation pubsub Redis client and recover stats consumption #900

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
freeplant merged 2 commits into from
Aug 6, 2026
Merged

fix: isolate automation pubsub Redis client and recover stats consumption #900

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
31466e0
fix: isolate automation pubsub Redis client and recover stats consump…
AlexCXC Aug 5, 2026
0cc7185
remove useless code
AlexCXC Aug 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 53 additions & 23 deletions dtable_events/automations/automations_pipeline.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,8 +72,11 @@ def __init__(self):
self.workers = 5
self._db_session_class = init_db_session_class()

self._redis_client = RedisClient(socket_connect_timeout=5, socket_timeout=10,
health_check_interval=30, retry_on_timeout=True)
# Keep pubsub reconnects isolated from regular Redis commands.

@seafile-dev seafile-dev Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Warning] Missing reconnect regression test

Why this matters:
This fix depends on refresh_subscriber() reconnecting only the subscriber client while stats() continues using the command client. The current CI runs SQL tests only, so a future client refactor could silently reintroduce the thread-exit race.

Suggested fix: add a focused unit test that forces a subscriber refresh during stats queue consumption and verifies the command client can still pop and process a result.

self._command_redis_client = RedisClient(socket_connect_timeout=5, socket_timeout=10,
health_check_interval=30, retry_on_timeout=True)
self._subscriber_redis_client = RedisClient(socket_connect_timeout=5, socket_timeout=10,
health_check_interval=30, retry_on_timeout=True)
self.per_update_channel = 'automation-rule-triggered'
self._pubsub_no_message_timeout = 5 * 60

Expand Down Expand Up @@ -121,9 +124,9 @@ def publish_metrics(self):
while True:
publish_metric(self.realtime_trigger_count, 'realtime_automation_triggered_count', REALTIME_AUTOMATION_RULES_TRIGGERED_COUNT_HELP)
publish_metric(self.scheduled_trigger_count, 'scheduled_automation_triggered_count', SCHEDULED_AUTOMATION_RULES_TRIGGERED_COUNT_HELP)
publish_metric(self._redis_client.llen(QUEUE_AUTOMATION_TASKS_10), f'{QUEUE_AUTOMATION_TASKS_10}_size', AUTOMATION_QUEUE_10_METRIC_HELP)
publish_metric(self._redis_client.llen(QUEUE_AUTOMATION_TASKS_20), f'{QUEUE_AUTOMATION_TASKS_20}_size', AUTOMATION_QUEUE_20_METRIC_HELP)
publish_metric(self._redis_client.llen(QUEUE_AUTOMATION_TASKS_30), f'{QUEUE_AUTOMATION_TASKS_30}_size', AUTOMATION_QUEUE_30_METRIC_HELP)
publish_metric(self._command_redis_client.llen(QUEUE_AUTOMATION_TASKS_10), f'{QUEUE_AUTOMATION_TASKS_10}_size', AUTOMATION_QUEUE_10_METRIC_HELP)
publish_metric(self._command_redis_client.llen(QUEUE_AUTOMATION_TASKS_20), f'{QUEUE_AUTOMATION_TASKS_20}_size', AUTOMATION_QUEUE_20_METRIC_HELP)
publish_metric(self._command_redis_client.llen(QUEUE_AUTOMATION_TASKS_30), f'{QUEUE_AUTOMATION_TASKS_30}_size', AUTOMATION_QUEUE_30_METRIC_HELP)
publish_metric(self.realtime_automation_heartbeat, 'realtime_automation_heartbeat', REALTIME_AUTOMATION_RULES_HEARTBEAT_HELP)
time.sleep(10)

Expand Down Expand Up @@ -156,15 +159,15 @@ def put_task(self, automation_task: AutomationTask):
automation_task.with_test,
queue_key
)
self._redis_client.lpush(queue_key, json.dumps(automation_task.to_dict()))
self._command_redis_client.lpush(queue_key, json.dumps(automation_task.to_dict()))

def receive(self):
auto_rule_logger.info(
"Start consuming automation events from Redis: window_secs=%s limit_percent=%s",
self.rate_limiter.window_secs,
self.rate_limiter.percent,
)
subscriber = self._redis_client.get_subscriber(self.per_update_channel)
subscriber = self._subscriber_redis_client.get_subscriber(self.per_update_channel)
last_pubsub_message_time = time.time()
while True:
try:
Expand Down Expand Up @@ -224,7 +227,7 @@ def receive(self):
warnings=automation_task.warnings
)
self.add_exceed_system_resource_limit_entity(automation_task.owner, automation_task.org_id)
self._redis_client.lpush(self.results_queue_key, json.dumps(automation_result.to_dict()))
self._command_redis_client.lpush(self.results_queue_key, json.dumps(automation_result.to_dict()))
continue
if self.automations_stats_manager.is_exceed(db_session, owner_info['owner'], owner_info['org_id']):
auto_rule_logger.info(
Expand All @@ -243,14 +246,15 @@ def receive(self):
else:
if time.time() - last_pubsub_message_time >= self._pubsub_no_message_timeout:
auto_rule_logger.info('no automation message for %ss', self._pubsub_no_message_timeout)
subscriber = self._redis_client.refresh_subscriber(
subscriber = self._subscriber_redis_client.refresh_subscriber(
subscriber, self.per_update_channel, 'no message timeout')
last_pubsub_message_time = time.time()
continue
time.sleep(0.5)
except Exception as e:
auto_rule_logger.error('Redis pubsub receive failed: %s', e)
subscriber = self._redis_client.refresh_subscriber(subscriber, self.per_update_channel, str(e))
subscriber = self._subscriber_redis_client.refresh_subscriber(
subscriber, self.per_update_channel, str(e))
last_pubsub_message_time = time.time()

def scan_rules(self):
Expand Down Expand Up @@ -326,37 +330,58 @@ def timed_job():

sched.start()

def mark_test_task_done(self, task_id, rule_id):
for attempt in range(3):
try:
self._command_redis_client.set(self.get_test_task_cache_key(task_id), 1, timeout=30)
return True
except Exception:
auto_rule_logger.exception(
'Failed to mark test automation task %s as complete (attempt %s/3)',
task_id, attempt + 1)
time.sleep(1)
auto_rule_logger.error(
'Could not mark test automation task %s for rule %s as complete after 3 attempts',
task_id, rule_id)
return False

def stats(self):
auto_rule_logger.info("Start stats thread")
while True:
result_info_str = self._redis_client.rpop(self.results_queue_key)
try:
result_info_str = self._command_redis_client.rpop(self.results_queue_key)
except Exception:
auto_rule_logger.exception('Failed to pop automation result from Redis')
time.sleep(1)
continue
if not result_info_str:
time.sleep(0.2)
continue
result_info = json.loads(result_info_str)
try:
result_info = json.loads(result_info_str)
result_info['trigger_time'] = datetime.fromisoformat(result_info['trigger_time'])
result_info['trigger_date'] = datetime.fromisoformat(result_info['trigger_date'])
result = AutomationResult(**result_info)
except Exception as e:
auto_rule_logger.exception(f'failed to load result {result_info}')
auto_rule_logger.exception(f'failed to load result {result_info_str}')
continue
result_data = result.data if isinstance(result.data, dict) else {}
auto_rule_logger.info(
"Received automation event: rule_id=%s rule_name=%s run_condition=%s dtable_uuid=%s op_type=%s table_id=%s row_id=%s updated_column_keys=%s success=%s run_time=%s test=%s",
"Received automation result: rule_id=%s rule_name=%s run_condition=%s dtable_uuid=%s op_type=%s table_id=%s row_id=%s updated_column_keys=%s success=%s run_time=%s test=%s",
result.rule_id,
result.rule_name,
result.run_condition,
result.dtable_uuid,
result.data['op_type'] if result.data else None,
result.data['table_id'] if result.data else None,
result.data['row_id'] if result.data else None,
result.data['updated_column_keys'] if result.data else None,
result_data.get('op_type'),
result_data.get('table_id'),
result_data.get('row_id'),
result_data.get('updated_column_keys'),
result.success,
result.run_time,
result.with_test
)
if result.with_test:
self._redis_client.set(self.get_test_task_cache_key(result.task_id), 1, timeout=30)
self.mark_test_task_done(result.task_id, result.rule_id)
continue
if result.run_condition == 'per_update' and not result.is_exceed_system_resource_limit:
owner = result.owner
Expand All @@ -369,13 +394,18 @@ def stats(self):
org_id,
self.rate_limiter.get_percent(owner, org_id, self.workers),
)
db_session = self._db_session_class()
db_session = None
try:
db_session = self._db_session_class()
self.automations_stats_manager.update_stats(db_session, result)
except Exception as e:
auto_rule_logger.exception(e)
finally:
db_session.close()
if db_session:
try:
db_session.close()
except Exception:
auto_rule_logger.exception('Failed to close automation stats database session')

def send_exceed_system_resource_limit_notifications(self):
sched = BlockingScheduler()
Expand Down Expand Up @@ -432,10 +462,10 @@ def put_test_task(self, automation_rule_id, task_id):
self.put_task(automation_task)
start_at = time.time()
while True:
done = self._redis_client.get(self.get_test_task_cache_key(automation_task.task_id))
done = self._command_redis_client.get(self.get_test_task_cache_key(automation_task.task_id))
if done:
break
if time.time() - start_at > 15 * 60:
if time.time() - start_at > 10 * 60:

@seafile-dev seafile-dev Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Warning] Shortened test-task deadline

Why this matters:
This unrelated change makes a test automation that completes between 10 and 15 minutes report a timeout, whereas the previous behavior waited 15 minutes. The result can therefore complete successfully but the caller still receives a failure.

Suggested fix: keep the existing 15-minute timeout for this bug fix, or document and coordinate the new 10-minute API behavior with its callers.

raise Exception(f'Wait test automation {automation_task.rule_id} task id {automation_task.task_id} timeout')
time.sleep(1)

Expand Down
7 changes: 0 additions & 7 deletions dtable_events/dtable_io/task_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,13 +244,6 @@ def add_append_excel_csv_upload_file_task(self, username, file_name, dtable_uuid

@log_function_call
def add_run_auto_rule_task(self, automation_rule_id):
# from dtable_events.automations.automations_pipeline import AutomationTask

# task_id = str(uuid.uuid4())
# automation_task = AutomationTask(automation_rule_id, run_condition, json.loads(trigger), json.loads(actions), dtable_uuid, org_id, owner, None, True, task_id)
# task = (
# self.app._automations_pipeline.put_test_task,
# (automation_task,))
task_id = str(uuid.uuid4())
task = (self.app._automations_pipeline.put_test_task, (automation_rule_id, task_id))
self.tasks_queue.put(task_id)
Expand Down
4 changes: 2 additions & 2 deletions dtable_events/utils/dtable_web_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,7 @@ def internal_add_notification(self, to_users, msg_type, detail):
'detail': detail,
'to_users': to_users,
'type': msg_type
}, headers=headers)
}, headers=headers, timeout=30)
return parse_response(resp)

def internal_submit_row_workflow(self, workflow_token, row_id, rule_id=None):
Expand Down Expand Up @@ -253,7 +253,7 @@ def internal_roles(self):
}
token = jwt.encode({'is_internal': True}, DTABLE_PRIVATE_KEY, algorithm='HS256')
headers = {'Authorization': 'Token ' + token}
resp = requests.get(url, headers=headers)
resp = requests.get(url, headers=headers, timeout=30)
return parse_response(resp).get('roles', [])

def internal_storage_quota(self, org_id=None, username=None):
Expand Down
Loading