From 557899d9ffb8284e21c4c2642e9fab970bc5ec32 Mon Sep 17 00:00:00 2001
From: armory-abedonik <106548537+armory-abedonik@users.noreply.github.com>
Date: Thu, 21 Sep 2023 05:10:30 +0200
Subject: [PATCH] fix: duplicate entry exception for correlation_ids table.
 (#4521)

* fix: duplicate entry exception for correlation_ids table.

* feat: add unit test.

---------

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
(cherry picked from commit 8523343b91d4a17e0faedb76e3ad3574dfb5c2ce)
---
 .../PipelineExecutionRepositoryTck.groovy     | 22 +++++++++++++++++++
 .../jedis/RedisExecutionRepository.java       |  2 +-
 .../persistence/SqlExecutionRepository.kt     | 18 +++++----------
 3 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/PipelineExecutionRepositoryTck.groovy b/orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/PipelineExecutionRepositoryTck.groovy
index 04d20eb8d6..73a4556fcb 100644
--- a/orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/PipelineExecutionRepositoryTck.groovy
+++ b/orca-core-tck/src/main/groovy/com/netflix/spinnaker/orca/pipeline/persistence/PipelineExecutionRepositoryTck.groovy
@@ -550,6 +550,28 @@ abstract class PipelineExecutionRepositoryTck<T extends ExecutionRepository> ext
     status << ExecutionStatus.values()
   }
 
+  def "should return task ref for currently running pipeline by correlation id"() {
+    given:
+    def execution = pipeline {
+      trigger = new DefaultTrigger("manual", "covfefe")
+    }
+    repository().store(execution)
+    repository().updateStatus(execution.type, execution.id, RUNNING)
+
+    when:
+    def result = repository().retrievePipelineForCorrelationId('covfefe')
+
+    then:
+    result.id == execution.id
+
+    when:
+    repository().updateStatus(execution.type, execution.id, SUCCEEDED)
+    repository().retrievePipelineForCorrelationId('covfefe')
+
+    then:
+    thrown(ExecutionNotFoundException)
+  }
+
   def "should return task ref for currently running orchestration by correlation id"() {
     given:
     def execution = orchestration {
diff --git a/orca-redis/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/jedis/RedisExecutionRepository.java b/orca-redis/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/jedis/RedisExecutionRepository.java
index 5eda7e5ad3..70a405e42f 100644
--- a/orca-redis/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/jedis/RedisExecutionRepository.java
+++ b/orca-redis/src/main/java/com/netflix/spinnaker/orca/pipeline/persistence/jedis/RedisExecutionRepository.java
@@ -695,7 +695,7 @@ public PipelineExecution retrieveByCorrelationId(
   @Override
   public PipelineExecution retrievePipelineForCorrelationId(@Nonnull String correlationId)
       throws ExecutionNotFoundException {
-    String key = format("pipelineCorrelation:%s", correlationId);
+    String key = format("correlation:%s", correlationId);
     return getRedisDelegate(key)
         .withCommandsClient(
             correlationRedis -> {
diff --git a/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
index 07dc3e99a1..2a1e689d0a 100644
--- a/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
+++ b/orca-sql/src/main/kotlin/com/netflix/spinnaker/orca/sql/pipeline/persistence/SqlExecutionRepository.kt
@@ -893,18 +893,12 @@ class SqlExecutionRepository(
       }
 
       withPool(poolName) {
-        val exists = ctx.fetchExists(
-          ctx.select()
-            .from("correlation_ids")
-            .where(field("id").eq(execution.trigger.correlationId))
-            .and(executionIdField.eq(execution.id))
-        )
-        if (!exists) {
-          ctx.insertInto(table("correlation_ids"))
-            .columns(field("id"), executionIdField)
-            .values(execution.trigger.correlationId, execution.id)
-            .execute()
-        }
+        ctx.insertInto(table("correlation_ids"))
+          .columns(field("id"), executionIdField)
+          .values(execution.trigger.correlationId, execution.id)
+          .onConflict()
+          .doNothing()
+          .execute()
       }
     }
   }