the status remains as "running"

[zhymm1314]

When I use Temporal deployed on Alibaba Cloud SAE, there is a situation where the workflow gets stuck, and the status remains as "running". Of course, this happens roughly once in every 100 cases during my batch testing

[zhymm1314]

[zhymm1314]

I actually set the timeout to 10 seconds, but a long time has already passed. wuwuwu～

[gow]

@zhymm1314 it's difficult to identify the issue with the screenshot and provided information. Could you please provide the following information:

• Server version
• Any relevant logs from the server for this workflow run.
• Persistence setup (ie which DB)
• Server config (anything you think might be relevant)
• Worker logs

[zhymm1314]

lib-temporal.zip

@gow
I've tested several times and found that this issue didn't occur on machines with a high core count (128 cores, 256GB RAM), but it almost always occurred on machines with 2 cores and 4GB RAM. I suspect it's related to the number of cores on the machine.

Tested on: 2 cores, 4GB RAM machines
Producers: 500
Consumers: 40 concurrent users.

Below is the demo I used to reproduce the issue.

Could you please take a look at this problem? It's really important to me.

This is our deployment configuration, docker-compose.yml. The database we're using is a cloud database, version MySQL 8.0.
docker-compose.yml

version: "3.8"

services:
temporal:
    image: temporalio/auto-setup:1.29.2
    container_name: temporal
    networks:
      - xlab-net
    environment:
      DB: mysql8
      MYSQL_USER: temporal_test
      MYSQL_PWD: xxxxx
      DBNAME: temporal_temp
      MYSQL_SEEDS: xxxxx
      VISIBILITY_DBNAME: temporal_temp_visibility
      VISIBILITY_DB_PORT: 3306
      ES_VERSION: v7
      DEFAULT_NAMESPACE_RETENTION: 72h
    ports:
      - "7233:7233"

  temporal-ui:
    image: temporalio/ui:2.44.0
    container_name: temporal-ui
    networks:
      - xlab-net
    environment:
      TEMPORAL_ADDRESS: temporal:7233
      TEMPORAL_NAMESPACE: default
    ports:
      - "8080:8080"
networks:
  xlab-net:
    external: true

this is python code =>
1.workflow_simple.py

"""
简化版 Workflow - 不使用子 Workflow，直接调用 Activity

依赖关系：A -> 并行执行 BCDE
"""

from datetime import timedelta
from typing import Any

from activities import (
    execute_task_a,
    execute_task_b,
    execute_task_c,
    execute_task_d,
    execute_task_e,
)
from temporalio import workflow
from temporalio.common import RetryPolicy

# 通用重试策略
COMMON_RETRY_POLICY = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(seconds=30),
    maximum_attempts=3,
)


@workflow.defn
class MainWorkflowSimple:
    """主工作流（简化版）：A -> 并行执行 BCDE，不使用子 Workflow"""

    @workflow.run
    async def run(self, task_id: str) -> dict[str, Any]:
        """
        简化版执行流程：
        1. 直接调用 Activity A
        2. 并行调用 Activity B、C、D、E
        
        不使用子 Workflow，避免创建开销和死锁
        """
        try:
            # 步骤1：执行 A (直接调用 Activity)
            workflow.set_current_details("进度: 1/2 - 正在执行任务 A")
            workflow.logger.info("[MainWorkflowSimple] 执行 Activity A")
            
            result_a = await workflow.execute_activity(
                execute_task_a,
                args=[task_id],
                start_to_close_timeout=timedelta(seconds=60),      # 执行超时 60 秒
                heartbeat_timeout=timedelta(seconds=30),           # 心跳超时 30 秒
                retry_policy=COMMON_RETRY_POLICY,
                # 不设置 schedule_to_start_timeout，允许无限排队
                # 如果 Worker 停止，heartbeat_timeout 会检测到并重试
            )

            # 步骤2：并行执行 BCDE (直接调用 Activity)
            workflow.set_current_details("进度: 2/2 - A 完成，正在并行执行 B, C, D, E")
            workflow.logger.info("[MainWorkflowSimple] 并行执行 Activities BCDE")

            import asyncio

            # 使用 asyncio.gather 并行执行
            results = await asyncio.gather(
                workflow.execute_activity(
                    execute_task_b,
                    args=[task_id, result_a.get("b_input", "")],
                    start_to_close_timeout=timedelta(seconds=60),
                    heartbeat_timeout=timedelta(seconds=30),
                    retry_policy=COMMON_RETRY_POLICY,
                ),
                workflow.execute_activity(
                    execute_task_c,
                    args=[task_id, result_a.get("c_input", "")],
                    start_to_close_timeout=timedelta(seconds=60),
                    heartbeat_timeout=timedelta(seconds=30),
                    retry_policy=COMMON_RETRY_POLICY,
                ),
                workflow.execute_activity(
                    execute_task_d,
                    args=[task_id, result_a.get("d_input", "")],
                    start_to_close_timeout=timedelta(seconds=60),
                    heartbeat_timeout=timedelta(seconds=30),
                    retry_policy=COMMON_RETRY_POLICY,
                ),
                workflow.execute_activity(
                    execute_task_e,
                    args=[task_id, result_a.get("e_input", "")],
                    start_to_close_timeout=timedelta(seconds=60),
                    heartbeat_timeout=timedelta(seconds=30),
                    retry_policy=COMMON_RETRY_POLICY,
                ),
                return_exceptions=True,  # 允许部分失败
            )

            result_b, result_c, result_d, result_e = results

            # 检查是否有失败的
            for i, (name, result) in enumerate([("B", result_b), ("C", result_c), ("D", result_d), ("E", result_e)]):
                if isinstance(result, Exception):
                    workflow.logger.error(f"[MainWorkflowSimple] Activity {name} 失败: {result}")

            final_results = {
                "A": result_a,
                "B": result_b if not isinstance(result_b, Exception) else {"error": str(result_b)},
                "C": result_c if not isinstance(result_c, Exception) else {"error": str(result_c)},
                "D": result_d if not isinstance(result_d, Exception) else {"error": str(result_d)},
                "E": result_e if not isinstance(result_e, Exception) else {"error": str(result_e)},
            }

            workflow.logger.info("[MainWorkflowSimple] 全部完成")
            return final_results

        except Exception as e:
            workflow.logger.error(f"[MainWorkflowSimple] 错误: {e}")
            raise

2.worker_simple.py

"""
简化版 Worker - 只使用 Activity，不使用子 Workflow

运行命令:
    cd libs/lib-temporal
    export TEMPORAL_HOST=xxxxx
    export TEMPORAL_NAMESPACE=default
    uv run examples/storylab_demo/worker_simple.py
"""

import asyncio
import logging

from activities import (
    execute_task_a,
    execute_task_b,
    execute_task_c,
    execute_task_d,
    execute_task_e,
)
from definitions import TASK_QUEUE
from workflows_simple import MainWorkflowSimple

from lib_temporal import BaseWorkerManager, create_client


class SimpleWorkerManager(BaseWorkerManager):
    task_queue = TASK_QUEUE
    
    # 只注册主 Workflow，不注册子 Workflow
    workflows = [
        MainWorkflowSimple,
    ]
    
    # 注册所有 Activities
    activities = [
        execute_task_a,
        execute_task_b,
        execute_task_c,
        execute_task_d,
        execute_task_e,
    ]
    
    # 使用中等并发数（测试发现高并发会有问题）
    use_tuner = False
    max_concurrent_workflow_tasks = 40   # 中等并发，避免问题
    max_concurrent_activities = 40       # 中等并发，足够用且稳定


async def main():
    # 配置日志（启用 DEBUG 级别，查看 Worker 详细行为）
    logging.basicConfig(
        level=logging.DEBUG,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )

    print("=" * 60)
    print("🚀 启动简化版 Worker（只用 Activity）")
    print("=" * 60)
    print(f"正在连接到 Temporal...")
    client = await create_client()
    print(f"✓ 连接成功")
    print()

    print(f"正在启动 Worker，监听队列: {TASK_QUEUE}")
    print(f"注册 Workflows: {len(SimpleWorkerManager.workflows)} 个（只有 MainWorkflowSimple）")
    print(f"注册 Activities: {len(SimpleWorkerManager.activities)} 个")
    print(f"⚠️  并发配置（故意降低）: Workflow Tasks=5, Activities=3")
    print(f"⚠️  目的：让问题更容易复现")
    print("=" * 60)
    
    manager = SimpleWorkerManager(client)
    await manager.start()

    print("✓ Worker 已启动，按 Ctrl+C 停止...")

    # 维持运行直到被中断
    try:
        await asyncio.Future()
    except asyncio.CancelledError:
        pass
    finally:
        await manager.shutdown()


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n✓ Worker 已停止")

3.producer_simple.py

"""
简化版投递脚本

使用 MainWorkflowSimple（不使用子 Workflow）

运行命令:
    cd libs/lib-temporal
    export TEMPORAL_HOST=xxxxx
    export TEMPORAL_NAMESPACE=default
    
    # 单个任务
    uv run examples/storylab_demo/producer_simple.py
    
    # 批量 5 个
    uv run examples/storylab_demo/producer_simple.py --batch 5 --concurrency 5
"""

import asyncio
import uuid
import argparse

from definitions import TASK_QUEUE
from workflows_simple import MainWorkflowSimple

from lib_temporal.client import create_client


async def main():
    """启动单个 MainWorkflowSimple"""
    client = await create_client()

    task_id = str(uuid.uuid4())
    print(f"🚀 启动 MainWorkflowSimple: {task_id}")

    handle = await client.start_workflow(
        MainWorkflowSimple.run,
        task_id,
        id=f"workflow-simple-{task_id}",
        task_queue=TASK_QUEUE,
    )

    print(f"✓ 工作流已启动: {handle.id}")


async def run_batch(count: int, concurrency: int = 5):
    """批量启动多个 MainWorkflowSimple"""
    client = await create_client()
    
    print("=" * 60)
    print(f"🚀 批量启动 {count} 个 MainWorkflowSimple (并发度: {concurrency})")
    print("=" * 60)
    print()

    semaphore = asyncio.Semaphore(concurrency)
    success_count = 0
    fail_count = 0

    async def start_one(index: int):
        nonlocal success_count, fail_count
        async with semaphore:
            try:
                task_id = str(uuid.uuid4())
                await client.start_workflow(
                    MainWorkflowSimple.run,
                    task_id,
                    id=f"workflow-simple-{task_id}",
                    task_queue=TASK_QUEUE,
                )
                success_count += 1
                if success_count % 10 == 0 or success_count == count:
                    print(f"  ... 进度: {success_count}/{count}")
            except Exception as e:
                fail_count += 1
                print(f"  ❌ 启动失败 (Index: {index}): {e}")

    # 并发启动
    tasks = [start_one(i) for i in range(count)]
    await asyncio.gather(*tasks)
    
    print()
    print("=" * 60)
    print(f"✅ 批量启动完成: 成功 {success_count}, 失败 {fail_count}")
    print("=" * 60)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--batch", type=int, help="批量启动数量", default=0)
    parser.add_argument("--concurrency", type=int, help="并发度", default=5)
    args = parser.parse_args()

    if args.batch > 0:
        asyncio.run(run_batch(args.batch, args.concurrency))
    else:
        asyncio.run(main())