init: this is spaak's custom trigger version

Author: NERLOE

apps/supervisor/src/workloadManager/kubernetes.ts

@@ -117,6 +117,9 @@ export class KubernetesWorkloadManager implements WorkloadManager {
               "app.kubernetes.io/part-of": "trigger-worker",
               "app.kubernetes.io/component": "create",
             },
+            annotations: {
+              "cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
+            },
           },
           spec: {
             ...this.addPlacementTags(this.#defaultPodSpec, opts.placementTags),

apps/webapp/app/env.server.ts

@@ -1156,6 +1156,7 @@ const EnvironmentSchema = z
       .default("info"),
     RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
     RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
+    RUN_REPLICATION_LEADER_ELECTION_RETRY_INTERVAL_MS: z.coerce.number().int().default(5_000),
     RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
     RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
     RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),

apps/webapp/app/routes/admin.api.v1.runs-replication.create.ts

@@ -22,6 +22,7 @@ const CreateRunReplicationServiceParams = z.object({
   leaderLockExtendIntervalMs: z.number(),
   leaderLockAcquireAdditionalTimeMs: z.number(),
   leaderLockRetryIntervalMs: z.number(),
+  leaderElectionRetryIntervalMs: z.number().optional(),
   ackIntervalSeconds: z.number(),
   waitForAsyncInsert: z.boolean(),
 });
@@ -113,6 +114,7 @@ function createRunReplicationService(params: CreateRunReplicationServiceParams)
     leaderLockExtendIntervalMs: params.leaderLockExtendIntervalMs,
     leaderLockAcquireAdditionalTimeMs: params.leaderLockAcquireAdditionalTimeMs,
     leaderLockRetryIntervalMs: params.leaderLockRetryIntervalMs,
+    leaderElectionRetryIntervalMs: params.leaderElectionRetryIntervalMs,
     ackIntervalSeconds: params.ackIntervalSeconds,
     logLevel: "debug",
     waitForAsyncInsert: params.waitForAsyncInsert,

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

@@ -58,6 +58,7 @@ function initializeRunsReplicationInstance() {
     leaderLockExtendIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS,
     leaderLockAcquireAdditionalTimeMs: env.RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS,
     leaderLockRetryIntervalMs: env.RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS,
+    leaderElectionRetryIntervalMs: env.RUN_REPLICATION_LEADER_ELECTION_RETRY_INTERVAL_MS,
     ackIntervalSeconds: env.RUN_REPLICATION_ACK_INTERVAL_SECONDS,
     logLevel: env.RUN_REPLICATION_LOG_LEVEL,
     waitForAsyncInsert: env.RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT === "1",

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

@@ -57,6 +57,7 @@ export type RunsReplicationServiceOptions = {
   leaderLockExtendIntervalMs?: number;
   leaderLockAcquireAdditionalTimeMs?: number;
   leaderLockRetryIntervalMs?: number;
+  leaderElectionRetryIntervalMs?: number;
   ackIntervalSeconds?: number;
   acknowledgeTimeoutMs?: number;
   logger?: Logger;
@@ -85,6 +86,7 @@ export type RunsReplicationServiceEvents = {
   batchFlushed: [
     { flushId: string; taskRunInserts: TaskRunInsertArray[]; payloadInserts: PayloadInsertArray[] }
   ];
+  leaderElection: [boolean];
 };
 
 export class RunsReplicationService {
@@ -109,6 +111,8 @@ export class RunsReplicationService {
   private _latestCommitEndLsn: string | null = null;
   private _lastAcknowledgedLsn: string | null = null;
   private _acknowledgeInterval: NodeJS.Timeout | null = null;
+  private _leaderElectionRetryTimer: NodeJS.Timeout | null = null;
+  private _leaderElectionRetryIntervalMs: number;
   // Retry configuration
   private _insertMaxRetries: number;
   private _insertBaseDelayMs: number;
@@ -231,21 +235,27 @@ export class RunsReplicationService {
       tracer: options.tracer,
     });
 
-    this._replicationClient.events.on("data", async ({ lsn, log, parseDuration }) => {
-      this.#handleData(lsn, log, parseDuration);
-    });
+    this._replicationClient.events.on(
+      "data",
+      async ({ lsn, log, parseDuration }: { lsn: string; log: PgoutputMessage; parseDuration: bigint }) => {
+        this.#handleData(lsn, log, parseDuration);
+      }
+    );
 
-    this._replicationClient.events.on("heartbeat", async ({ lsn, shouldRespond }) => {
-      if (this._isShuttingDown) return;
-      if (this._isShutDownComplete) return;
+    this._replicationClient.events.on(
+      "heartbeat",
+      async ({ lsn, shouldRespond }: { lsn: string; shouldRespond: boolean }) => {
+        if (this._isShuttingDown) return;
+        if (this._isShutDownComplete) return;
 
-      if (shouldRespond) {
-        this._lastAcknowledgedLsn = lsn;
-        await this._replicationClient.acknowledge(lsn);
+        if (shouldRespond) {
+          this._lastAcknowledgedLsn = lsn;
+          await this._replicationClient.acknowledge(lsn);
+        }
       }
-    });
+    );
 
-    this._replicationClient.events.on("error", (error) => {
+    this._replicationClient.events.on("error", (error: Error) => {
       this.logger.error("Replication client error", {
         error,
       });
@@ -255,18 +265,31 @@ export class RunsReplicationService {
       this.logger.info("Replication client started");
     });
 
-    this._replicationClient.events.on("acknowledge", ({ lsn }) => {
+    this._replicationClient.events.on("acknowledge", ({ lsn }: { lsn: string }) => {
       this.logger.debug("Acknowledged", { lsn });
     });
 
-    this._replicationClient.events.on("leaderElection", (isLeader) => {
+    this._replicationClient.events.on("leaderElection", (isLeader: boolean) => {
       this.logger.info("Leader election", { isLeader });
+      
+      // Forward the event to external listeners
+      this.events.emit("leaderElection", isLeader);
+      
+      if (!isLeader) {
+        // Failed to acquire leadership - reset subscription flag and schedule retry
+        this._isSubscribed = false;
+        this.#scheduleLeaderElectionRetry();
+      } else {
+        // Successfully acquired leadership - clear any pending retries
+        this.#clearLeaderElectionRetry();
+      }
     });
 
     // Initialize retry configuration
     this._insertMaxRetries = options.insertMaxRetries ?? 3;
     this._insertBaseDelayMs = options.insertBaseDelayMs ?? 100;
     this._insertMaxDelayMs = options.insertMaxDelayMs ?? 2000;
+    this._leaderElectionRetryIntervalMs = options.leaderElectionRetryIntervalMs ?? 5_000;
   }
 
   public async shutdown() {
@@ -276,6 +299,11 @@ export class RunsReplicationService {
 
     this.logger.info("Initiating shutdown of runs replication service");
 
+    // Clear any pending leader election retries
+    this.#clearLeaderElectionRetry();
+
+    this._isSubscribed = false;
+
     if (!this._currentTransaction) {
       this.logger.info("No transaction to commit, shutting down immediately");
       await this._replicationClient.stop();
@@ -287,10 +315,23 @@ export class RunsReplicationService {
   }
 
   async start() {
+    // Prevent multiple concurrent start attempts
+    if (this._isSubscribed) {
+      this.logger.debug("Already subscribed, skipping start");
+      return;
+    }
+
+    if (this._isShuttingDown || this._isShutDownComplete) {
+      this.logger.debug("Service is shutting down, skipping start");
+      return;
+    }
+
     this.logger.info("Starting replication client", {
       lastLsn: this._latestCommitEndLsn,
     });
 
+    this._isSubscribed = true;
+
     await this._replicationClient.subscribe(this._latestCommitEndLsn ?? undefined);
 
     this._acknowledgeInterval = setInterval(this.#acknowledgeLatestTransaction.bind(this), 1000);
@@ -300,6 +341,11 @@ export class RunsReplicationService {
   async stop() {
     this.logger.info("Stopping replication client");
 
+    // Clear any pending leader election retries
+    this.#clearLeaderElectionRetry();
+
+    this._isSubscribed = false;
+
     await this._replicationClient.stop();
 
     if (this._acknowledgeInterval) {
@@ -310,6 +356,11 @@ export class RunsReplicationService {
   async teardown() {
     this.logger.info("Teardown replication client");
 
+    // Clear any pending leader election retries
+    this.#clearLeaderElectionRetry();
+
+    this._isSubscribed = false;
+
     await this._replicationClient.teardown();
 
     if (this._acknowledgeInterval) {
@@ -955,6 +1006,42 @@ export class RunsReplicationService {
     return { data: parsedData };
   }
 
+  #scheduleLeaderElectionRetry() {
+    // Clear any existing retry timer
+    this.#clearLeaderElectionRetry();
+
+    // Don't retry if we're shutting down or already shut down
+    if (this._isShuttingDown || this._isShutDownComplete) {
+      return;
+    }
+
+    this.logger.info("Scheduling leader election retry", {
+      retryIntervalMs: this._leaderElectionRetryIntervalMs,
+    });
+
+    this._leaderElectionRetryTimer = setTimeout(async () => {
+      if (this._isShuttingDown || this._isShutDownComplete) {
+        return;
+      }
+
+      this.logger.info("Retrying leader election after failed attempt");
+
+      try {
+        await this.start();
+      } catch (error) {
+        this.logger.error("Error during leader election retry", { error });
+        // If start() fails, the leaderElection event will be emitted with false,
+        // which will trigger another retry via #scheduleLeaderElectionRetry
+      }
+    }, this._leaderElectionRetryIntervalMs);
+  }
+
+  #clearLeaderElectionRetry() {
+    if (this._leaderElectionRetryTimer) {
+      clearTimeout(this._leaderElectionRetryTimer);
+      this._leaderElectionRetryTimer = null;
+    }
+  }
 }
 
 export type ConcurrentFlushSchedulerConfig<T> = {