Use nanosecond precision of mtime for deterministic timestamps

Incorporate the hash in nanoseconds of timestamp.
RIPE-NCC · Aug 28, 2023 · adfc7ba · adfc7ba
1 parent 6e50bb3
commit adfc7ba
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 10 deletions.
diff --git a/src/main/java/net/ripe/rpki/rsyncit/rrdp/RRDPFetcherMetrics.java b/src/main/java/net/ripe/rpki/rsyncit/rrdp/RRDPFetcherMetrics.java
@@ -3,6 +3,7 @@
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Gauge;
 import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
 
 import java.util.concurrent.atomic.AtomicInteger;
 
@@ -13,6 +14,8 @@ public final class RRDPFetcherMetrics {
     private final Counter timeoutUpdates;
     private final Counter objectFailures;
 
+    public final Timer objectConstructionTimer;
+
     public RRDPFetcherMetrics(MeterRegistry meterRegistry) {
         successfulUpdates = buildCounter("success", meterRegistry);
         failedUpdates = buildCounter("failed", meterRegistry);
@@ -25,6 +28,10 @@ public RRDPFetcherMetrics(MeterRegistry meterRegistry) {
         Gauge.builder("rsyncit.fetcher.rrdp.serial", rrdpSerial::get)
             .description("Serial of the RRDP notification.xml at the given URL")
             .register(meterRegistry);
+
+        objectConstructionTimer = Timer.builder("rsyncit.fetcher.parsing")
+                .description("Time spent parsing objects for the last run")
+                .register(meterRegistry);
     }
 
     public void success(int serial) {

diff --git a/src/main/java/net/ripe/rpki/rsyncit/rrdp/RrdpFetcher.java b/src/main/java/net/ripe/rpki/rsyncit/rrdp/RrdpFetcher.java
@@ -210,6 +210,9 @@ private ProcessPublishElementResult processPublishElements(Element doc, Instant
         // to the closest hour -- it is unlikely that different instances will have clocks off by a lot,
         // so rounding down to an hour should generate the same timestamps _most of the time_.
         //
+        if (lastModified == null) {
+            log.info("No last-modified header in response: Using current hour as timestamp");
+        }
         var defaultTimestamp = lastModified != null ? lastModified : Instant.now().truncatedTo(ChronoUnit.HOURS);
 
         // This timestamp is only needed for marking objects in the timestamp cache.
@@ -223,8 +226,9 @@ private ProcessPublishElementResult processPublishElements(Element doc, Instant
             .mapToObj(publishedObjects::item)
             .toList();
 
-        var t = Time.timed(() -> objectItems
-            .parallelStream()
+        var objects = metrics.objectConstructionTimer.record(() -> objectItems
+            .stream().unordered()
+            .parallel()
             .map(item -> {
                 var objectUri = item.getAttributes().getNamedItem("uri").getNodeValue();
                 var content = item.getTextContent();
@@ -241,10 +245,10 @@ private ProcessPublishElementResult processPublishElements(Element doc, Instant
                     //
                     // Cache the timestamp per hash do avoid re-parsing every object in the snapshot every time.
                     //
-                    final Instant createAt = state.cacheTimestamps(Sha256.asString(hash), now,
-                        () -> spiceWithHash(getTimestampForObject(objectUri, decoded, defaultTimestamp), hash));
+                    final Instant modificationTime = state.cacheTimestamps(Sha256.asString(hash), now,
+                        () -> incorporateHashInTimestamp(getTimestampForObject(objectUri, decoded, defaultTimestamp), hash));
 
-                    return new RpkiObject(URI.create(objectUri), decoded, createAt);
+                    return new RpkiObject(URI.create(objectUri), decoded, modificationTime);
                 } catch (RuntimeException e) {
                     metrics.badObject();
                     log.error("Cannot decode object data for URI {}\n{}", objectUri, content);
@@ -270,8 +274,7 @@ private ProcessPublishElementResult processPublishElements(Element doc, Instant
             })
             .collect(Collectors.toList()));
 
-        var objects = t.getResult();
-        log.info("Constructed {} objects in {}ms", objects.size(), t.getTime());
+        log.info("Parsed {} objects", objects.size());
         return new ProcessPublishElementResult(objects, collisionCount.get());
     }
 
@@ -332,9 +335,9 @@ private static void checkResult(String objectUri, ValidationResult result) {
      * This MAY help for the corner case of objects having second-accuracy timestamps
      * and the timestatmp in seconds being the same for multiple objects.
      */
-    private Instant spiceWithHash(Instant t, byte[] hash) {
-        final BigInteger ms = new BigInteger(hash).mod(BigInteger.valueOf(1000L));
-        return t.truncatedTo(ChronoUnit.SECONDS).plusMillis(ms.longValue());
+    private Instant incorporateHashInTimestamp(Instant t, byte[] hash) {
+        final BigInteger ms = new BigInteger(hash).mod(BigInteger.valueOf(1000_000_000L));
+        return t.truncatedTo(ChronoUnit.SECONDS).plusNanos(ms.longValue());
     }
 
     record NotificationXml(String sessionId, Integer serial, String snapshotUrl, String expectedSnapshotHash) {