refactor: removing threading in bbdev

Author: shirohasuki

api/motia.config.ts

@@ -3,8 +3,7 @@ import endpointPlugin from '@motiadev/plugin-endpoint/plugin'
 import logsPlugin from '@motiadev/plugin-logs/plugin'
 import observabilityPlugin from '@motiadev/plugin-observability/plugin'
 import statesPlugin from '@motiadev/plugin-states/plugin'
-import bullmqPlugin from '@motiadev/plugin-bullmq/plugin'
 
 export default defineConfig({
-  plugins: [observabilityPlugin, statesPlugin, endpointPlugin, logsPlugin, bullmqPlugin],
+  plugins: [observabilityPlugin, statesPlugin, endpointPlugin, logsPlugin],
 })

bbdev

@@ -449,41 +449,11 @@ if __name__ == "__main__":
                 available_port = find_available_port(start_port=5100, end_port=5500)
             print(f"Starting server on port {available_port}...")
             proc = subprocess.Popen(
-                ["pnpm", "dev", "--port", str(available_port)],
-                cwd=workflow_dir,
-                start_new_session=True,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.PIPE,
+                ["pnpm", "dev", "--port", str(available_port)], cwd=workflow_dir
             )
 
-            # Forward stdout/stderr in background threads; stopped before
-            # shutdown to suppress BullMQ teardown errors and pnpm's
-            # ELIFECYCLE message (Motia framework limitation).
-            import threading
-
-            output_active = threading.Event()
-            output_active.set()
-
-            def forward_stream(stream, dest):
-                for line in iter(stream.readline, b""):
-                    if not output_active.is_set():
-                        break
-                    dest.write(line)
-                    dest.flush()
-
-            threading.Thread(
-                target=forward_stream,
-                args=(proc.stdout, sys.stdout.buffer),
-                daemon=True,
-            ).start()
-            threading.Thread(
-                target=forward_stream,
-                args=(proc.stderr, sys.stderr.buffer),
-                daemon=True,
-            ).start()
-
             # Wait for service to start
-            max_retries = 600
+            max_retries = 30
             for i in range(max_retries):
                 try:
                     # Disable proxy, connect directly to localhost
@@ -536,22 +506,10 @@ if __name__ == "__main__":
             )
 
             # 3. Shutdown service ================================
-            output_active.clear()
-            import signal
-
+            # Give observability plugin time to finish async operations (e.g., Redis writes)
             time.sleep(1)
-            try:
-                os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
-            except ProcessLookupError:
-                pass
-            try:
-                proc.wait(timeout=10)
-            except subprocess.TimeoutExpired:
-                try:
-                    os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-                except ProcessLookupError:
-                    pass
-                proc.wait()
+            proc.terminate()
+            proc.wait()
             print(
                 f"\nTask completed. Command running on http://localhost:{available_port} is finished"
             )