Realtime streams fix waiting for streams to finished

ericallam · ericallam · commit 84e3e0a7b799 · 2024-12-02T13:37:29.000Z
diff --git a/packages/core/src/v3/runMetadata/manager.ts b/packages/core/src/v3/runMetadata/manager.ts
@@ -250,7 +250,7 @@ export class StandardMetadataManager implements RunMetadataManager {
       return streamInstance;
     } catch (error) {
       // Clean up metadata key if stream creation fails
-      this.deleteKey(`$$stream.${key}`);
+      this.removeFromKey(`$$streams`, key);
       throw error;
     }
   }
@@ -265,7 +265,7 @@ export class StandardMetadataManager implements RunMetadataManager {
       return;
     }
 
-    const promises = Array.from(this.activeStreams.values());
+    const promises = Array.from(this.activeStreams.values()).map((stream) => stream.wait());
 
     try {
       await Promise.race([

Original file line number	Diff line number	Diff line change
`@@ -250,7 +250,7 @@ export class StandardMetadataManager implements RunMetadataManager {`
`250`	`250`	`return streamInstance;`
`251`	`251`	`} catch (error) {`
`252`	`252`	`// Clean up metadata key if stream creation fails`
`253`		- this.deleteKey(`$$stream.${key}`);
	`253`	+ this.removeFromKey(`$$streams`, key);
`254`	`254`	`throw error;`
`255`	`255`	`}`
`256`	`256`	`}`
`@@ -265,7 +265,7 @@ export class StandardMetadataManager implements RunMetadataManager {`
`265`	`265`	`return;`
`266`	`266`	`}`
`267`	`267`
`268`		`- const promises = Array.from(this.activeStreams.values());`
	`268`	`+ const promises = Array.from(this.activeStreams.values()).map((stream) => stream.wait());`
`269`	`269`
`270`	`270`	`try {`
`271`	`271`	`await Promise.race([`