diff --git a/docs/runbooks/notification-processing-failure.md b/docs/runbooks/notification-processing-failure.md index 83d4e1099..f324610a6 100644 --- a/docs/runbooks/notification-processing-failure.md +++ b/docs/runbooks/notification-processing-failure.md @@ -22,12 +22,26 @@ ``` redis-cli EXISTS chaos:notification-service:consumer_strict_schema ``` - - +3. Inspect the payload of a stuck message and compare it with `SqsNotificationMessage` + (`services/notification-service/src/main/kotlin/com/otterworks/notification/model/NotificationEvent.kt`): + ``` + aws sqs receive-message --queue-url "$SQS_QUEUE_URL" --max-number-of-messages 1 --visibility-timeout 0 + ``` + Legacy producers send `timestamp` as a Unix epoch number rather than an RFC 3339 string. +4. Check `ApproximateNumberOfMessages` / `ApproximateAgeOfOldestMessage` on the queue and the + depth of the `-dlq` queue to size the backlog. ## Resolution Steps - +1. If the chaos flag is set, clear it: `redis-cli DEL chaos:notification-service:consumer_strict_schema` + (or `scripts/inject-bug.sh reset` for a tenant). +2. The consumer accepts both RFC 3339 and epoch `timestamp` values and deletes messages that + cannot be deserialized instead of leaving them to cycle through the visibility timeout. If + a new schema mismatch appears, extend the model/serializer in `NotificationEvent.kt` and + redeploy; the backlog drains on its own once the consumer accepts the payload. +3. Redrive the DLQ once the consumer is healthy: + `aws sqs start-message-move-task --source-arn --destination-arn `. +4. Confirm `notifications_processing_errors_total` stops increasing and the alert resolves. ## Post-Incident diff --git a/services/notification-service/build.gradle.kts b/services/notification-service/build.gradle.kts index 778f9fdbd..d109a5ee7 100644 --- a/services/notification-service/build.gradle.kts +++ b/services/notification-service/build.gradle.kts @@ -70,9 +70,6 @@ dependencies { implementation("io.opentelemetry:opentelemetry-sdk:1.36.0") implementation("io.opentelemetry:opentelemetry-exporter-otlp:1.36.0") - // Redis (chaos flag checks) - implementation("redis.clients:jedis:5.1.3") - // Testing testImplementation("io.ktor:ktor-server-tests-jvm:$ktorVersion") testImplementation("io.ktor:ktor-server-test-host:$ktorVersion") diff --git a/services/notification-service/src/main/kotlin/com/otterworks/notification/consumer/SqsConsumer.kt b/services/notification-service/src/main/kotlin/com/otterworks/notification/consumer/SqsConsumer.kt index 56d823a23..6f271c312 100644 --- a/services/notification-service/src/main/kotlin/com/otterworks/notification/consumer/SqsConsumer.kt +++ b/services/notification-service/src/main/kotlin/com/otterworks/notification/consumer/SqsConsumer.kt @@ -14,26 +14,9 @@ import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import kotlinx.serialization.json.Json import mu.KotlinLogging -import redis.clients.jedis.JedisPool -import redis.clients.jedis.JedisPoolConfig private val logger = KotlinLogging.logger {} -// Lazy Redis pool for chaos flag checks. -private val redisPool: JedisPool by lazy { - val host = System.getenv("REDIS_HOST") ?: "localhost" - val port = System.getenv("REDIS_PORT")?.toIntOrNull() ?: 6379 - JedisPool(JedisPoolConfig(), host, port, 1000) -} - -private fun chaosActive(flag: String): Boolean { - return try { - redisPool.resource.use { jedis -> jedis.exists(flag) } - } catch (e: Exception) { - false - } -} - class SqsConsumer( private val sqsClient: SqsClient, private val notificationService: NotificationService, @@ -42,24 +25,11 @@ class SqsConsumer( ) { private val processingErrorsCounter: Counter? = meterRegistry?.counter("notifications.processing.errors") - // Standard lenient parser used in normal operation. private val json = Json { ignoreUnknownKeys = true isLenient = true } - // CHAOS: strict parser that rejects messages whose timestamp field is not - // a valid RFC 3339 string. Legacy events emitted by older service versions - // use Unix epoch integers for timestamps, which are rejected here. - // When the chaos flag is active, every such message throws - // SerializationException, is never deleted from the queue, and becomes - // visible again after the SQS visibility timeout — causing queue depth to - // climb indefinitely while the consumer appears healthy. - private val strictJson = Json { - ignoreUnknownKeys = false - isLenient = false - } - suspend fun startPolling() = coroutineScope { logger.info { "Starting SQS consumer polling: ${config.sqsQueueUrl}" } @@ -86,18 +56,16 @@ class SqsConsumer( if (event != null) { notificationService.processEvent(event) - - val deleteRequest = DeleteMessageRequest { - queueUrl = config.sqsQueueUrl - receiptHandle = msg.receiptHandle - } - sqsClient.deleteMessage(deleteRequest) - logger.debug { "Deleted SQS message: ${msg.messageId}" } + deleteMessage(msg.messageId, msg.receiptHandle) } else { + // A message that cannot be deserialized will never succeed on + // redelivery; drop it so it does not cycle through the queue. processingErrorsCounter?.increment() - logger.warn { "Failed to parse SQS message: ${msg.messageId}" } + logger.warn { "Discarding unparseable SQS message: ${msg.messageId}" } + deleteMessage(msg.messageId, msg.receiptHandle) } } catch (e: Exception) { + processingErrorsCounter?.increment() logger.error(e) { "Error processing SQS message: ${msg.messageId}" } } } @@ -113,16 +81,24 @@ class SqsConsumer( } } + private suspend fun deleteMessage(messageId: String?, handle: String?) { + val deleteRequest = DeleteMessageRequest { + queueUrl = config.sqsQueueUrl + receiptHandle = handle + } + sqsClient.deleteMessage(deleteRequest) + logger.debug { "Deleted SQS message: $messageId" } + } + internal fun parseMessage(body: String): SqsNotificationMessage? { - val parser = if (chaosActive("chaos:notification-service:consumer_strict_schema")) strictJson else json return try { // Try parsing as direct message first - parser.decodeFromString(body) + json.decodeFromString(body) } catch (_: Exception) { try { // Try unwrapping SNS envelope - val snsWrapper = parser.decodeFromString(body) - parser.decodeFromString(snsWrapper.Message) + val snsWrapper = json.decodeFromString(body) + json.decodeFromString(snsWrapper.Message) } catch (e: Exception) { logger.error(e) { "Failed to parse message body" } null diff --git a/services/notification-service/src/main/kotlin/com/otterworks/notification/model/NotificationEvent.kt b/services/notification-service/src/main/kotlin/com/otterworks/notification/model/NotificationEvent.kt index eeebd8744..ee6073325 100644 --- a/services/notification-service/src/main/kotlin/com/otterworks/notification/model/NotificationEvent.kt +++ b/services/notification-service/src/main/kotlin/com/otterworks/notification/model/NotificationEvent.kt @@ -1,6 +1,19 @@ package com.otterworks.notification.model +import kotlinx.serialization.KSerializer import kotlinx.serialization.Serializable +import kotlinx.serialization.SerializationException +import kotlinx.serialization.descriptors.PrimitiveKind +import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor +import kotlinx.serialization.descriptors.SerialDescriptor +import kotlinx.serialization.encoding.Decoder +import kotlinx.serialization.encoding.Encoder +import kotlinx.serialization.json.JsonDecoder +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.doubleOrNull +import kotlinx.serialization.json.longOrNull +import java.time.Instant +import java.time.format.DateTimeFormatter @Serializable enum class EventType { @@ -39,9 +52,46 @@ data class SqsNotificationMessage( val userId: String = "", val actorId: String = "", val mentionedUserId: String = "", + @Serializable(with = EventTimestampSerializer::class) val timestamp: String, ) +/** + * Accepts the event timestamp either as an RFC 3339 string (current producers) or as a + * Unix epoch number in seconds or milliseconds (legacy producers), always yielding an + * ISO-8601 UTC string. + */ +object EventTimestampSerializer : KSerializer { + private const val EPOCH_MILLIS_THRESHOLD = 100_000_000_000L + + override val descriptor: SerialDescriptor = + PrimitiveSerialDescriptor("com.otterworks.notification.EventTimestamp", PrimitiveKind.STRING) + + override fun deserialize(decoder: Decoder): String { + val jsonDecoder = decoder as? JsonDecoder + ?: return decoder.decodeString() + val element = jsonDecoder.decodeJsonElement() + val primitive = element as? JsonPrimitive + ?: throw SerializationException("timestamp must be a string or number, got $element") + if (primitive.isString) return primitive.content + val epoch = primitive.longOrNull + ?: primitive.doubleOrNull?.toLong() + ?: throw SerializationException("timestamp must be a string or number, got $element") + return fromEpoch(epoch) + } + + override fun serialize(encoder: Encoder, value: String) = encoder.encodeString(value) + + fun fromEpoch(epoch: Long): String { + val instant = if (epoch >= EPOCH_MILLIS_THRESHOLD) { + Instant.ofEpochMilli(epoch) + } else { + Instant.ofEpochSecond(epoch) + } + return DateTimeFormatter.ISO_INSTANT.format(instant) + } +} + @Serializable data class Notification( val id: String, diff --git a/services/notification-service/src/test/kotlin/com/otterworks/notification/consumer/SqsConsumerTest.kt b/services/notification-service/src/test/kotlin/com/otterworks/notification/consumer/SqsConsumerTest.kt index c3397475e..7ef616aa1 100644 --- a/services/notification-service/src/test/kotlin/com/otterworks/notification/consumer/SqsConsumerTest.kt +++ b/services/notification-service/src/test/kotlin/com/otterworks/notification/consumer/SqsConsumerTest.kt @@ -1,9 +1,20 @@ package com.otterworks.notification.consumer import aws.sdk.kotlin.services.sqs.SqsClient +import aws.sdk.kotlin.services.sqs.model.DeleteMessageRequest +import aws.sdk.kotlin.services.sqs.model.Message +import aws.sdk.kotlin.services.sqs.model.ReceiveMessageResponse import com.otterworks.notification.config.AppConfig +import com.otterworks.notification.model.SqsNotificationMessage import com.otterworks.notification.service.NotificationService +import io.mockk.coEvery +import io.mockk.coVerify import io.mockk.mockk +import io.mockk.slot +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.advanceTimeBy +import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.Json import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertNotNull @@ -120,6 +131,88 @@ class SqsConsumerTest { assertEquals("doc-789", event.documentId) } + @Test + fun `parseMessage accepts legacy epoch-seconds timestamp`() { + val body = """ + { + "eventType": "file_shared", + "fileId": "file-123", + "ownerId": "owner-1", + "sharedWithUserId": "user-2", + "timestamp": 1704067200 + } + """.trimIndent() + + val event = consumer.parseMessage(body) + + assertNotNull(event) + assertEquals("file_shared", event.eventType) + assertEquals("2024-01-01T00:00:00Z", event.timestamp) + } + + @Test + fun `parseMessage accepts legacy epoch-millis timestamp`() { + val body = """{"eventType":"document_edited","userId":"user-1","documentId":"doc-1","timestamp":1704067200123}""" + + val event = consumer.parseMessage(body) + + assertNotNull(event) + assertEquals("2024-01-01T00:00:00.123Z", event.timestamp) + } + + @Test + fun `parseMessage accepts SNS-wrapped legacy epoch timestamp`() { + val innerMessage = """{"eventType":"comment_added","userId":"user-1","actorId":"actor-1","documentId":"doc-1","commentId":"c-1","timestamp":1704067200}""" + val escapedInner = innerMessage.replace("\"", "\\\"") + val body = """{"Type":"Notification","MessageId":"msg-1","Message":"$escapedInner"}""" + + val event = consumer.parseMessage(body) + + assertNotNull(event) + assertEquals("comment_added", event.eventType) + assertEquals("2024-01-01T00:00:00Z", event.timestamp) + } + + @Test + fun `epoch timestamps decode even with a strict non-lenient parser`() { + val strict = Json { isLenient = false; ignoreUnknownKeys = false } + val body = """{"eventType":"file_shared","timestamp":1704067200}""" + + val event = strict.decodeFromString(body) + + assertEquals("2024-01-01T00:00:00Z", event.timestamp) + } + + @Test + fun `startPolling deletes unparseable messages instead of leaving them on the queue`() = runTest { + val poison = Message { + messageId = "poison-1" + receiptHandle = "rh-poison-1" + body = "not json at all" + } + val valid = Message { + messageId = "ok-1" + receiptHandle = "rh-ok-1" + body = """{"eventType":"file_shared","fileId":"f-1","timestamp":1704067200}""" + } + coEvery { sqsClient.receiveMessage(any()) } returnsMany listOf( + ReceiveMessageResponse { messages = listOf(poison, valid) }, + ReceiveMessageResponse { messages = emptyList() }, + ) + + val job = launch { consumer.startPolling() } + advanceTimeBy(config.sqsPollIntervalMs * 3) + job.cancel() + + val deleted = mutableListOf() + coVerify(exactly = 2) { sqsClient.deleteMessage(capture(deleted)) } + assertEquals(setOf("rh-poison-1", "rh-ok-1"), deleted.map { it.receiptHandle }.toSet()) + + val processed = slot() + coVerify(exactly = 1) { notificationService.processEvent(capture(processed)) } + assertEquals("f-1", processed.captured.fileId) + } + @Test fun `parseMessage handles missing optional fields`() { val body = """