New metadata deployed task fix (#1588)

* Add more logging around new metadata system

* Make the authenticated env optional when updating run metadata

apps/webapp/app/routes/api.v1.runs.$runId.metadata.ts

@@ -16,7 +16,7 @@ const { action } = createActionApiRoute(
     method: "PUT",
   },
   async ({ authentication, body, params }) => {
-    const result = await updateMetadataService.call(authentication.environment, params.runId, body);
+    const result = await updateMetadataService.call(params.runId, body, authentication.environment);
 
     if (!result) {
       return json({ error: "Task Run not found" }, { status: 404 });

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

@@ -229,17 +229,21 @@ export class UpdateMetadataService extends BaseService {
   }
 
   public async call(
-    environment: AuthenticatedEnvironment,
     runId: string,
-    body: UpdateMetadataRequestBody
+    body: UpdateMetadataRequestBody,
+    environment?: AuthenticatedEnvironment
   ) {
     const runIdType = runId.startsWith("run_") ? "friendly" : "internal";
 
     const taskRun = await this._prisma.taskRun.findFirst({
-      where: {
-        runtimeEnvironmentId: environment.id,
-        ...(runIdType === "internal" ? { id: runId } : { friendlyId: runId }),
-      },
+      where: environment
+        ? {
+            runtimeEnvironmentId: environment.id,
+            ...(runIdType === "internal" ? { id: runId } : { friendlyId: runId }),
+          }
+        : {
+            ...(runIdType === "internal" ? { id: runId } : { friendlyId: runId }),
+          },
       select: {
         id: true,
         status: true,
@@ -351,6 +355,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 +376,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 +405,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 +443,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]);

apps/webapp/app/v3/services/finalizeTaskRun.server.ts

@@ -70,9 +70,9 @@ export class FinalizeTaskRunService extends BaseService {
       completedAt,
     });
 
-    if (env && metadata) {
+    if (metadata) {
       try {
-        await updateMetadataService.call(env, id, metadata);
+        await updateMetadataService.call(id, metadata, env);
       } catch (e) {
         logger.error("[FinalizeTaskRunService] Failed to update metadata", {
           taskRun: id,

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;
   },
 });