Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: Check for clock skew with response header #113

Merged
merged 3 commits into from
Dec 4, 2024
Merged

chore: Check for clock skew with response header #113

Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eef8b04
chore: Check for clock skew with response header
patriknw Dec 3, 2024
5a376e9
Instant.now instead of InstantFactory
patriknw Dec 3, 2024
c6a46b5
tolerance config
patriknw Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 45 additions & 2 deletions core/src/main/scala/akka/persistence/dynamodb/internal/JournalDao.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,15 +6,19 @@ package akka.persistence.dynamodb.internal

import java.nio.ByteBuffer
import java.time.Instant
import java.time.format.DateTimeFormatter
import java.util.concurrent.CompletionException
import java.util.Base64
import java.util.Locale
import java.util.{ HashMap => JHashMap }
import java.util.UUID
import java.util.concurrent.atomic.AtomicLong

import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.jdk.CollectionConverters._
import scala.jdk.FutureConverters._
import scala.util.control.NonFatal

import akka.Done
import akka.actor.typed.ActorSystem
Expand All @@ -25,6 +29,7 @@ import akka.persistence.typed.PersistenceId
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import software.amazon.awssdk.core.SdkBytes
import software.amazon.awssdk.core.SdkResponse
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient
import software.amazon.awssdk.services.dynamodb.model.AttributeValue
import software.amazon.awssdk.services.dynamodb.model.Delete
Expand Down Expand Up @@ -58,6 +63,34 @@ import software.amazon.awssdk.services.dynamodb.model.Update

private implicit val ec: ExecutionContext = system.executionContext

private val dateHeaderLogCounter = new AtomicLong
private val dateHeaderParser = DateTimeFormatter.ofPattern("EEE, dd MMM yyyy HH:mm:ss z", Locale.US)

private def checkClockSkew(response: SdkResponse): Unit = {
try {
if (dateHeaderLogCounter.getAndIncrement() % 1000 == 0) {
val dateHeaderOpt = response.sdkHttpResponse().firstMatchingHeader("Date")
if (dateHeaderOpt.isPresent) {
val dateHeader = dateHeaderOpt.get
val awsInstant = Instant.from(dateHeaderParser.parse(dateHeader))
val now = Instant.now()
// The Date header only has precision of seconds so this is just a rough check
if (math.abs(java.time.Duration.between(awsInstant, now).toMillis) >= 2000) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Make the tolerance configurable?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, also good to be able to disable that way:c6a46b5

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

👍🏼

log.warn(
"Possible clock skew, make sure clock synchronization is installed. " +
"Local time [{}] vs DynamoDB response time [{}]",
now,
awsInstant)
}
}
}
} catch {
case NonFatal(exc) =>
log.warn("check clock skew failed", exc)
}

}

def writeEvents(events: Seq[SerializedJournalItem]): Future[Done] = {
require(events.nonEmpty)

Expand Down Expand Up @@ -118,7 +151,14 @@ import software.amazon.awssdk.services.dynamodb.model.Update
response.consumedCapacity.capacityUnits)
}
}
result.map(_ => Done)(ExecutionContext.parasitic)
result
.map { response =>
checkClockSkew(response)
Done
}(ExecutionContext.parasitic)
.recoverWith { case c: CompletionException =>
Future.failed(c.getCause)
}(ExecutionContext.parasitic)
} else {
val writeItems =
events.map { item =>
Expand Down Expand Up @@ -160,7 +200,10 @@ import software.amazon.awssdk.services.dynamodb.model.Update
}
}
result
.map(_ => Done)(ExecutionContext.parasitic)
.map { response =>
checkClockSkew(response)
Done
}(ExecutionContext.parasitic)
.recoverWith { case c: CompletionException =>
Future.failed(c.getCause)
}(ExecutionContext.parasitic)
Expand Down
Loading