fix(reporting): serialise SQS schema updates of concurrent report calculations

application/aam-backend-service/src/main/kotlin/com/aamdigital/aambackendservice/reporting/report/sqs/SqsSchemaService.kt

@@ -1,10 +1,12 @@
 package com.aamdigital.aambackendservice.reporting.report.sqs
 
 import com.aamdigital.aambackendservice.common.couchdb.core.CouchDbClient
+import com.aamdigital.aambackendservice.common.couchdb.core.DefaultCouchDbClient.DefaultCouchDbClientErrorCode
 import com.aamdigital.aambackendservice.common.domain.EntityAttribute
 import com.aamdigital.aambackendservice.common.domain.EntityAttributeType
 import com.aamdigital.aambackendservice.common.domain.EntityConfig
 import com.aamdigital.aambackendservice.common.domain.EntityType
+import com.aamdigital.aambackendservice.common.error.ExternalSystemException
 import com.fasterxml.jackson.annotation.JsonProperty
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
 import org.slf4j.LoggerFactory
@@ -85,6 +87,14 @@ class SqsSchemaService(
 
     fun getSchemaPath(): String = "/$TARGET_DATABASE/$SCHEMA_PATH"
 
+    /**
+     * Writes the SQS schema derived from the current entity config, unless it is already up to date.
+     *
+     * Report calculations run concurrently and each calls this before its queries. Synchronized, so
+     * that after a config change only the first of them writes the schema and the others find it up
+     * to date, instead of overwriting the revision the first one just wrote (a 409 conflict).
+     */
+    @Synchronized
     fun updateSchema() {
         val config =
             couchDbClient.getDatabaseDocument(
@@ -109,12 +119,7 @@ class SqsSchemaService(
 
         val currentSqsSchema =
             try {
-                couchDbClient.getDatabaseDocument(
-                    database = TARGET_DATABASE,
-                    documentId = SCHEMA_PATH,
-                    queryParams = LinkedMultiValueMap(),
-                    kClass = SqsSchema::class
-                )
+                fetchCurrentSchema()
             } catch (ex: Exception) {
                 logger.warn("[SqsSchemaService] No current SQS Schema found. Creating it.", ex)
                 null
@@ -126,12 +131,30 @@ class SqsSchemaService(
             return
         }
 
-        couchDbClient.putDatabaseDocument(
+        try {
+            couchDbClient.putDatabaseDocument(
+                database = TARGET_DATABASE,
+                documentId = SCHEMA_PATH,
+                body = newSqsSchema
+            )
+        } catch (ex: ExternalSystemException) {
+            // another writer (e.g. a second instance of this service) got there first;
+            // fine as long as it wrote the same schema
+            if (ex.code != DefaultCouchDbClientErrorCode.CONFLICT ||
+                fetchCurrentSchema().configVersion != newSqsSchema.configVersion
+            ) {
+                throw ex
+            }
+        }
+    }
+
+    private fun fetchCurrentSchema(): SqsSchema =
+        couchDbClient.getDatabaseDocument(
             database = TARGET_DATABASE,
             documentId = SCHEMA_PATH,
-            body = newSqsSchema
+            queryParams = LinkedMultiValueMap(),
+            kClass = SqsSchema::class
         )
-    }
 
     private fun mapToSqsSchema(entityConfig: EntityConfig): SqsSchema {
         val tables =

application/aam-backend-service/src/test/kotlin/com/aamdigital/aambackendservice/reporting/report/sqs/SqsSchemaServiceTest.kt

@@ -1,11 +1,15 @@
 package com.aamdigital.aambackendservice.reporting.report.sqs
 
 import com.aamdigital.aambackendservice.common.couchdb.core.CouchDbClient
+import com.aamdigital.aambackendservice.common.couchdb.core.DefaultCouchDbClient.DefaultCouchDbClientErrorCode
 import com.aamdigital.aambackendservice.common.couchdb.dto.DocSuccess
 import com.aamdigital.aambackendservice.common.domain.TestErrorCode
+import com.aamdigital.aambackendservice.common.error.ExternalSystemException
 import com.aamdigital.aambackendservice.common.error.NotFoundException
 import com.aamdigital.aambackendservice.common.rest.ObjectMapperConfiguration
 import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatCode
+import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.extension.ExtendWith
@@ -15,8 +19,13 @@ import org.mockito.kotlin.any
 import org.mockito.kotlin.argumentCaptor
 import org.mockito.kotlin.eq
 import org.mockito.kotlin.reset
+import org.mockito.kotlin.times
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
+import java.util.concurrent.CyclicBarrier
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
 
 private const val SCHEMA_DOC_ID = "_design/sqlite:config"
 
@@ -34,6 +43,21 @@ class SqsSchemaServiceTest {
     }
 
     private fun stubConfig(data: Map<String, AppConfigEntry>) {
+        stubConfigDocument(data)
+        whenever(
+            couchDbClient.getDatabaseDocument(
+                database = eq("app"),
+                documentId = eq(SCHEMA_DOC_ID),
+                queryParams = any(),
+                kClass = eq(SqsSchema::class)
+            )
+        ).thenAnswer { throw NotFoundException(message = "not found", code = TestErrorCode.TEST_EXCEPTION) }
+        whenever(
+            couchDbClient.putDatabaseDocument(any(), any(), any())
+        ).thenReturn(DocSuccess(ok = true, id = SCHEMA_DOC_ID, rev = "1-new"))
+    }
+
+    private fun stubConfigDocument(data: Map<String, AppConfigEntry>) {
         whenever(
             couchDbClient.getDatabaseDocument(
                 database = eq("app"),
@@ -42,17 +66,24 @@ class SqsSchemaServiceTest {
                 kClass = eq(AppConfigFile::class)
             )
         ).thenReturn(AppConfigFile(id = "Config:CONFIG_ENTITY", rev = "1-abc", data = data))
+    }
+
+    /** Backs the schema document by [stored], so reads see what was written before. */
+    private fun stubSchemaDocument(
+        stored: AtomicReference<SqsSchema?>,
+        onRead: () -> Unit = {}
+    ) {
         whenever(
             couchDbClient.getDatabaseDocument(
                 database = eq("app"),
                 documentId = eq(SCHEMA_DOC_ID),
                 queryParams = any(),
                 kClass = eq(SqsSchema::class)
             )
-        ).thenAnswer { throw NotFoundException(message = "not found", code = TestErrorCode.TEST_EXCEPTION) }
-        whenever(
-            couchDbClient.putDatabaseDocument(any(), any(), any())
-        ).thenReturn(DocSuccess(ok = true, id = SCHEMA_DOC_ID, rev = "1-new"))
+        ).thenAnswer {
+            onRead()
+            stored.get() ?: throw NotFoundException(message = "not found", code = TestErrorCode.TEST_EXCEPTION)
+        }
     }
 
     private fun capturePublishedSchema(): SqsSchema {
@@ -202,4 +233,72 @@ class SqsSchemaServiceTest {
                 ?.dataType
         ).isEqualTo("string")
     }
+
+    @Test
+    fun `should write a changed schema only once when calculations update it concurrently`() {
+        // given: both calls read the schema before either writes it, unless the reads are serialised
+        stubConfigDocument(emptyMap())
+        val stored = AtomicReference<SqsSchema?>()
+        val bothRead = CyclicBarrier(2)
+        stubSchemaDocument(stored) {
+            runCatching { bothRead.await(500, TimeUnit.MILLISECONDS) }
+        }
+        whenever(couchDbClient.putDatabaseDocument(any(), any(), any())).thenAnswer {
+            stored.set(it.getArgument(2))
+            DocSuccess(ok = true, id = SCHEMA_DOC_ID, rev = "1-new")
+        }
+        val executor = Executors.newFixedThreadPool(2)
+
+        // when
+        try {
+            val calls = List(2) { executor.submit { service.updateSchema() } }
+            calls.forEach { it.get(5, TimeUnit.SECONDS) }
+        } finally {
+            executor.shutdownNow()
+        }
+
+        // then
+        verify(couchDbClient, times(1)).putDatabaseDocument(any(), any(), any())
+    }
+
+    @Test
+    fun `should accept a conflict when the same schema was written concurrently`() {
+        // given
+        stubConfigDocument(emptyMap())
+        val stored = AtomicReference<SqsSchema?>()
+        stubSchemaDocument(stored)
+        whenever(couchDbClient.putDatabaseDocument(any(), any(), any())).thenAnswer {
+            stored.set(it.getArgument(2))
+            throw ExternalSystemException(message = "conflict", code = DefaultCouchDbClientErrorCode.CONFLICT)
+        }
+
+        // when / then
+        assertThatCode { service.updateSchema() }.doesNotThrowAnyException()
+    }
+
+    @Test
+    fun `should rethrow a conflict when a different schema was written concurrently`() {
+        // given
+        stubConfigDocument(emptyMap())
+        val stored = AtomicReference<SqsSchema?>()
+        stubSchemaDocument(stored)
+        whenever(couchDbClient.putDatabaseDocument(any(), any(), any())).thenAnswer {
+            stored.set(
+                SqsSchema(
+                    sql =
+                        SqlObject(
+                            tables = emptyMap(),
+                            indexes = null,
+                            options = SqlOptions(TableName(field = "_id", separator = ":"))
+                        )
+                )
+            )
+            throw ExternalSystemException(message = "conflict", code = DefaultCouchDbClientErrorCode.CONFLICT)
+        }
+
+        // when / then
+        assertThatThrownBy { service.updateSchema() }
+            .isInstanceOf(ExternalSystemException::class.java)
+            .hasFieldOrPropertyWithValue("code", DefaultCouchDbClientErrorCode.CONFLICT)
+    }
 }