Add more logging around new metadata system

Author: ericallam

apps/webapp/app/services/metadata/updateMetadata.server.ts

@@ -351,6 +351,15 @@ export class UpdateMetadataService extends BaseService {
       });
 
       if (result.count === 0) {
+        if (this.flushLoggingEnabled) {
+          logger.debug(
+            `[UpdateMetadataService][updateRunMetadataWithOperations] Optimistic lock failed for run ${runId}`,
+            {
+              metadataVersion: run.metadataVersion,
+            }
+          );
+        }
+
         // If this was our last attempt, buffer the operations and return optimistically
         if (attempts === MAX_RETRIES) {
           this.#ingestRunOperations(runId, operations);
@@ -363,6 +372,15 @@ export class UpdateMetadataService extends BaseService {
         continue;
       }
 
+      if (this.flushLoggingEnabled) {
+        logger.debug(
+          `[UpdateMetadataService][updateRunMetadataWithOperations] Updated metadata for run ${runId}`,
+          {
+            metadata: applyResults.newMetadata,
+          }
+        );
+      }
+
       // Success! Return the new metadata
       return applyResults.newMetadata;
     }
@@ -383,10 +401,15 @@ export class UpdateMetadataService extends BaseService {
       metadataPacket.data !== "{}" ||
       (existingMetadata.data && metadataPacket.data !== existingMetadata.data)
     ) {
-      logger.debug(`Updating metadata directly for run`, {
-        metadata: metadataPacket.data,
-        runId,
-      });
+      if (this.flushLoggingEnabled) {
+        logger.debug(
+          `[UpdateMetadataService][updateRunMetadataDirectly] Updating metadata directly for run`,
+          {
+            metadata: metadataPacket.data,
+            runId,
+          }
+        );
+      }
 
       // Update the metadata without version check
       await this._prisma.taskRun.update({
@@ -416,6 +439,13 @@ export class UpdateMetadataService extends BaseService {
       };
     });
 
+    if (this.flushLoggingEnabled) {
+      logger.debug(`[UpdateMetadataService] Ingesting operations for run`, {
+        runId,
+        bufferedOperations,
+      });
+    }
+
     const existingBufferedOperations = this._bufferedOperations.get(runId) ?? [];
 
     this._bufferedOperations.set(runId, [...existingBufferedOperations, ...bufferedOperations]);

references/nextjs-realtime/src/trigger/csv.ts

@@ -62,16 +62,6 @@ export const handleCSVUpload = schemaTask({
     const successfulRows = results.runs.filter((r) => r.ok);
     const failedRows = results.runs.filter((r) => !r.ok);
 
-    const firstSuccessfulRow = successfulRows[0];
-
-    if (firstSuccessfulRow) {
-      const stream = await metadata.fetchStream<string>(firstSuccessfulRow.id);
-
-      for await (const value of stream) {
-        logger.info(`Stream value from ${firstSuccessfulRow.id}`, { value });
-      }
-    }
-
     return {
       file,
       rows,
@@ -93,14 +83,6 @@ export const handleCSVRow = schemaTask({
 
     metadata.parent.increment("processedRows", 1).append("rowRuns", ctx.run.id);
 
-    await metadata.parent.stream(
-      ctx.run.id,
-      (async function* () {
-        yield "hello";
-        yield "world";
-      })()
-    );
-
     return row;
   },
 });