Merge remote-tracking branch 'origin/main' into feat/ecr-support

Author: nicktrn

apps/webapp/app/services/runsReplicationService.server.ts

@@ -462,13 +462,13 @@ export class RunsReplicationService {
 
       // Insert task runs and payloads with retry logic for connection errors
       const [taskRunError, taskRunResult] = await this.#insertWithRetry(
-        () => this.#insertTaskRunInserts(taskRunInserts),
+        (attempt) => this.#insertTaskRunInserts(taskRunInserts, attempt),
         "task run inserts",
         flushId
       );
 
       const [payloadError, payloadResult] = await this.#insertWithRetry(
-        () => this.#insertPayloadInserts(payloadInserts),
+        (attempt) => this.#insertPayloadInserts(payloadInserts, attempt),
         "payload inserts",
         flushId
       );
@@ -502,24 +502,25 @@ export class RunsReplicationService {
 
   // New method to handle inserts with retry logic for connection errors
   async #insertWithRetry<T>(
-    insertFn: () => Promise<T>,
+    insertFn: (attempt: number) => Promise<T>,
     operationName: string,
     flushId: string
   ): Promise<[Error | null, T | null]> {
     let lastError: Error | null = null;
 
     for (let attempt = 1; attempt <= this._insertMaxRetries; attempt++) {
       try {
-        const result = await insertFn();
+        const result = await insertFn(attempt);
         return [null, result];
       } catch (error) {
         lastError = error instanceof Error ? error : new Error(String(error));
 
         // Check if this is a retryable connection error
-        if (this.#isRetryableConnectionError(lastError) && attempt < this._insertMaxRetries) {
+        if (this.#isRetryableConnectionError(lastError)) {
           const delay = this.#calculateConnectionRetryDelay(attempt);
 
-          this.logger.warn(`Retrying ${operationName} due to connection error`, {
+          this.logger.warn(`Retrying RunReplication insert due to connection error`, {
+            operationName,
             flushId,
             attempt,
             maxRetries: this._insertMaxRetries,
@@ -567,7 +568,7 @@ export class RunsReplicationService {
     return delay + jitter;
   }
 
-  async #insertTaskRunInserts(taskRunInserts: TaskRunV2[]) {
+  async #insertTaskRunInserts(taskRunInserts: TaskRunV2[], attempt: number) {
     return await startSpan(this._tracer, "insertTaskRunsInserts", async (span) => {
       const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insert(
         taskRunInserts,
@@ -581,18 +582,20 @@ export class RunsReplicationService {
       );
 
       if (insertError) {
-        this.logger.error("Error inserting task run inserts", {
+        this.logger.error("Error inserting task run inserts attempt", {
           error: insertError,
+          attempt,
         });
 
         recordSpanError(span, insertError);
+        throw insertError;
       }
 
       return insertResult;
     });
   }
 
-  async #insertPayloadInserts(payloadInserts: RawTaskRunPayloadV1[]) {
+  async #insertPayloadInserts(payloadInserts: RawTaskRunPayloadV1[], attempt: number) {
     return await startSpan(this._tracer, "insertPayloadInserts", async (span) => {
       const [insertError, insertResult] = await this.options.clickhouse.taskRuns.insertPayloads(
         payloadInserts,
@@ -606,11 +609,13 @@ export class RunsReplicationService {
       );
 
       if (insertError) {
-        this.logger.error("Error inserting payload inserts", {
+        this.logger.error("Error inserting payload inserts attempt", {
           error: insertError,
+          attempt,
         });
 
         recordSpanError(span, insertError);
+        throw insertError;
       }
 
       return insertResult;

docs/triggering.mdx

@@ -238,7 +238,8 @@ Triggers a single run of a task with the payload you pass in, and any options yo
 </Note>
 
 ```ts ./trigger/my-task.ts
-import { myOtherTask, runs } from "~/trigger/my-other-task";
+import { runs } from "@trigger.dev/sdk/v3";
+import { myOtherTask } from "~/trigger/my-other-task";
 
 export const myTask = task({
   id: "my-task",
@@ -254,7 +255,8 @@ export const myTask = task({
 To pass options to the triggered task, you can use the second argument:
 
 ```ts ./trigger/my-task.ts
-import { myOtherTask, runs } from "~/trigger/my-other-task";
+import { runs } from "@trigger.dev/sdk/v3";
+import { myOtherTask } from "~/trigger/my-other-task";
 
 export const myTask = task({
   id: "my-task",
@@ -272,7 +274,8 @@ export const myTask = task({
 Triggers multiple runs of a single task with the payloads you pass in, and any options you specify.
 
 ```ts /trigger/my-task.ts
-import { myOtherTask, batch } from "~/trigger/my-other-task";
+import { batch } from "@trigger.dev/sdk/v3";
+import { myOtherTask } from "~/trigger/my-other-task";
 
 export const myTask = task({
   id: "my-task",
@@ -288,7 +291,8 @@ export const myTask = task({
 If you need to pass options to `batchTrigger`, you can use the second argument:
 
 ```ts /trigger/my-task.ts
-import { myOtherTask, batch } from "~/trigger/my-other-task";
+import { batch } from "@trigger.dev/sdk/v3";
+import { myOtherTask } from "~/trigger/my-other-task";
 
 export const myTask = task({
   id: "my-task",
@@ -306,7 +310,8 @@ export const myTask = task({
 You can also pass in options for each run in the batch:
 
 ```ts /trigger/my-task.ts
-import { myOtherTask, batch } from "~/trigger/my-other-task";
+import { batch } from "@trigger.dev/sdk/v3";
+import { myOtherTask } from "~/trigger/my-other-task";
 
 export const myTask = task({
   id: "my-task",