notification-service: accept epoch timestamps and drop poison SQS messages

docs/runbooks/notification-processing-failure.md

@@ -22,12 +22,26 @@
    ```
    redis-cli EXISTS chaos:notification-service:consumer_strict_schema
    ```
-
-<!-- TODO: Complete investigation steps -->
+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
 
-<!-- TODO -->
+1. If the chaos flag is set, clear it: `redis-cli DEL chaos:notification-service:consumer_strict_schema`
+   (or `scripts/inject-bug.sh <ID> 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 <dlq-arn> --destination-arn <queue-arn>`.
+4. Confirm `notifications_processing_errors_total` stops increasing and the alert resolves.
 
 ## Post-Incident
 

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")

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<SqsNotificationMessage>(body)
+            json.decodeFromString<SqsNotificationMessage>(body)
         } catch (_: Exception) {
             try {
                 // Try unwrapping SNS envelope
-                val snsWrapper = parser.decodeFromString<SnsEnvelope>(body)
-                parser.decodeFromString<SqsNotificationMessage>(snsWrapper.Message)
+                val snsWrapper = json.decodeFromString<SnsEnvelope>(body)
+                json.decodeFromString<SqsNotificationMessage>(snsWrapper.Message)
             } catch (e: Exception) {
                 logger.error(e) { "Failed to parse message body" }
                 null

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<String> {
+    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,

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<SqsNotificationMessage>(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<DeleteMessageRequest>()
+        coVerify(exactly = 2) { sqsClient.deleteMessage(capture(deleted)) }
+        assertEquals(setOf("rh-poison-1", "rh-ok-1"), deleted.map { it.receiptHandle }.toSet())
+
+        val processed = slot<SqsNotificationMessage>()
+        coVerify(exactly = 1) { notificationService.processEvent(capture(processed)) }
+        assertEquals("f-1", processed.captured.fileId)
+    }
+
     @Test
     fun `parseMessage handles missing optional fields`() {
         val body = """