fix(market-data-collector): task retry subscription complete (#357)

No-Trade-No-Life · Dec 12, 2023 · 02ef5d5 · 02ef5d5
1 parent 5077d80
commit 02ef5d5
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 19 deletions.
diff --git a/apps/market-data-collector/src/index.ts b/apps/market-data-collector/src/index.ts
@@ -221,7 +221,6 @@ const runTask = (psr: IPullSourceRelation) =>
     };
 
     // Metrics State
-    // TODO: optimize the performance of Metric.set
     subs.push(interval(10_000).subscribe(reportStatus));
 
     // Wait for current_back_off_time to start
@@ -296,25 +295,30 @@ const runTask = (psr: IPullSourceRelation) =>
       copyDataAction$
         .pipe(
           mergeMap(() =>
-            term.copyDataRecords({
-              type: 'period',
-              tags: {
-                datasource_id: psr.datasource_id,
-                product_id: psr.product_id,
-                period_in_sec: '' + psr.period_in_sec,
-              },
-              time_range: [lastTime, Date.now()],
-              receiver_terminal_id: STORAGE_TERMINAL_ID,
-            }),
+            term
+              .copyDataRecords({
+                type: 'period',
+                tags: {
+                  datasource_id: psr.datasource_id,
+                  product_id: psr.product_id,
+                  period_in_sec: '' + psr.period_in_sec,
+                },
+                time_range: [lastTime, Date.now()],
+                receiver_terminal_id: STORAGE_TERMINAL_ID,
+              })
+              .pipe(
+                tap(() => {
+                  taskComplete$.next();
+                }),
+                // ISSUE: catch error will replace the whole stream with EMPTY, therefore it must be placed inside mergeMap
+                // so that the outer stream subscription will not be affected
+                catchError((e) => {
+                  err = e;
+                  taskError$.next();
+                  return EMPTY;
+                }),
+              ),
           ),
-          tap(() => {
-            taskComplete$.next();
-          }),
-          catchError((e, caught$) => {
-            err = e;
-            taskError$.next();
-            return EMPTY;
-          }),
         )
         .subscribe(),
     );
@@ -355,6 +359,7 @@ const runTask = (psr: IPullSourceRelation) =>
     subs.push(
       taskFinalize$.subscribe(() => {
         if (status === 'error') {
+          console.info(formatTime(Date.now()), `TaskRetry`, title);
           timer(current_back_off_time).subscribe(() => {
             taskScheduled$.next();
           });

diff --git a/common/changes/@yuants/app-market-data-collector/2023-12-12-11-44.json b/common/changes/@yuants/app-market-data-collector/2023-12-12-11-44.json
@@ -0,0 +1,10 @@
+{
+  "changes": [
+    {
+      "packageName": "@yuants/app-market-data-collector",
+      "comment": "fix: task error retry",
+      "type": "patch"
+    }
+  ],
+  "packageName": "@yuants/app-market-data-collector"
+}