Refa: make RAGFlow more asynchronous 2 (#11664)

### What problem does this PR solve?

Make RAGFlow more asynchronous 2. #11551, #11579, #11619.

### Type of change

- [x] New Feature (non-breaking change which adds functionality)
- [x] Refactoring
- [x] Performance Improvement

api/db/services/document_service.py

@@ -719,10 +719,14 @@ class DocumentService(CommonService):
                 # only for special task and parsed docs and unfinished
                 freeze_progress = special_task_running and doc_progress >= 1 and not finished
                 msg = "\n".join(sorted(msg))
+                begin_at = d.get("process_begin_at")
+                if not begin_at:
+                    begin_at = datetime.now()
+                    # fallback
+                    cls.update_by_id(d["id"], {"process_begin_at": begin_at})
+
                 info = {
-                    "process_duration": datetime.timestamp(
-                        datetime.now()) -
-                                       d["process_begin_at"].timestamp(),
+                    "process_duration": max(datetime.timestamp(datetime.now()) - begin_at.timestamp(), 0),
                     "run": status}
                 if prg != 0 and not freeze_progress:
                     info["progress"] = prg