fix(sql): fix potential race condition between ack and push (#65)

keiko-sql/src/main/kotlin/com/netflix/spinnaker/q/sql/SqlQueue.kt

@@ -632,44 +632,31 @@ class SqlQueue(
   }
 
   private fun ackMessage(fingerprint: String) {
-    /* TODO: since this isn't transactional, we should periodically check for and cleanup
-     *  any old messages with fingerprints not represented on the queue or in unacked.
-     */
     withRetry(RetryCategory.WRITE) {
       jooq.deleteFrom(unackedTable)
         .where(fingerprintField.eq(fingerprint))
         .execute()
     }
 
-    var changed = 0
-
     withRetry(RetryCategory.WRITE) {
       jooq.transaction { config ->
         val txn = DSL.using(config)
 
-        val id = txn.select(idField)
-          .from(queueTable)
+        val changed = txn.update(queueTable)
+          .set(lockedField, "0")
           .where(fingerprintField.eq(fingerprint))
-          .limit(1)
-          .fetchInto(String::class.java)
+          .execute()
 
-        if (id.isNotEmpty()) {
-          changed = txn.update(queueTable)
-            .set(lockedField, "0")
-            .where(idField.eq(id.first()))
+        // TODO: consider async scheduled cleanup of messagesTable if having this in the
+        //  ack txn results in frequent deadlocks.
+        if (changed == 0) {
+          txn.deleteFrom(messagesTable)
+            .where(fingerprintField.eq(fingerprint))
             .execute()
         }
       }
     }
 
-    if (changed == 0) {
-      withRetry(RetryCategory.WRITE) {
-        jooq.deleteFrom(messagesTable)
-          .where(fingerprintField.eq(fingerprint))
-          .execute()
-      }
-    }
-
     fire(MessageAcknowledged)
   }