feat(snapshots): when calculating checksum the metadata file should b…

…e added at the end

When calculating the checksum while taking the snapshot, the metadata file is added at the end. To ensure the same semantics while verifying the checksums, we the metadata file explicitly while calculating the checksum. Otherwise, depending on the name of metadata file the checksum calculated while taking a snapshot and checksum calculated when verifying the snapshot will be different.

snapshot/src/main/java/io/camunda/zeebe/snapshots/impl/FileBasedSnapshotStore.java

@@ -47,6 +47,12 @@ public final class FileBasedSnapshotStore extends Actor
     implements ConstructableSnapshotStore, ReceivableSnapshotStore {
 
   static final int VERSION = 1;
+
+  // When sorted with other files in the snapshot, the metadata file must be ordered at the end.
+  // This is required for backward compatibility of checksum calculation. Otherwise, the older
+  // versions, which are not aware of the metadata will calculate the checksum using a different
+  // order of files. We can change the name in later versions, because the new checksum calculation
+  // already order the metadata file explicitly instead of using the implicit sort order.
   static final String METADATA_FILE_NAME = "zeebe.metadata";
   // first is the metadata and the second the the received snapshot count
   private static final String RECEIVING_DIR_FORMAT = "%s-%d";

snapshot/src/main/java/io/camunda/zeebe/snapshots/impl/SnapshotChecksum.java

@@ -12,7 +12,6 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.util.List;
-import java.util.stream.Collectors;
 
 final class SnapshotChecksum {
 
@@ -37,14 +36,26 @@ public static SfvChecksum read(final Path checksumPath) throws IOException {
   }
 
   public static SfvChecksum calculate(final Path snapshotDirectory) throws IOException {
-    try (final var fileStream = Files.list(snapshotDirectory).sorted()) {
-      final SfvChecksum sfvChecksum =
-          createCombinedChecksum(fileStream.collect(Collectors.toList()));
-      sfvChecksum.setSnapshotDirectoryComment(snapshotDirectory.toString());
+    try (final var fileStream =
+        Files.list(snapshotDirectory).filter(SnapshotChecksum::isNotMetadataFile).sorted()) {
+      final SfvChecksum sfvChecksum = createCombinedChecksum(fileStream.toList());
+
+      // While persisting transient snapshot, the checksum of metadata file is added at the end.
+      // Hence when we recalculate the checksum, we must follow the same order. Otherwise base on
+      // the file name, the sorted file list will have a differnt order and thus result in a
+      // different checksum.
+      final var metadataFile = snapshotDirectory.resolve(FileBasedSnapshotStore.METADATA_FILE_NAME);
+      if (metadataFile.toFile().exists()) {
+        sfvChecksum.updateFromFile(metadataFile);
+      }
       return sfvChecksum;
     }
   }
 
+  private static boolean isNotMetadataFile(final Path file) {
+    return !file.getFileName().toString().equals(FileBasedSnapshotStore.METADATA_FILE_NAME);
+  }
+
   public static void persist(final Path checksumPath, final SfvChecksum checksum)
       throws IOException {
     try (final RandomAccessFile checksumFile = new RandomAccessFile(checksumPath.toFile(), "rwd")) {

snapshot/src/test/java/io/camunda/zeebe/snapshots/impl/SnapshotChecksumTest.java

@@ -190,4 +190,26 @@ public void shouldWriteTheNumberOfFiles() throws IOException {
     // then
     assertThat(serialized).contains("; number of files used for combined value = 3");
   }
+
+  public void shouldAddChecksumOfMetadataAtTheEnd() throws IOException {
+    // given
+    final var folder = temporaryFolder.newFolder().toPath();
+    createChunk(folder, "file1.txt");
+    createChunk(folder, "file2.txt");
+    createChunk(folder, "file3.txt");
+    final SfvChecksum checksumCalculatedInSteps = SnapshotChecksum.calculate(folder);
+
+    // when
+    createChunk(folder, FileBasedSnapshotStore.METADATA_FILE_NAME);
+    // This is how checksum is calculated when persisting a transient snapshot
+    checksumCalculatedInSteps.updateFromFile(
+        folder.resolve(FileBasedSnapshotStore.METADATA_FILE_NAME));
+
+    // This is how checksum is calculated when verifying
+    final SfvChecksum checksumCalculatedAtOnce = SnapshotChecksum.calculate(folder);
+
+    // then
+    assertThat(checksumCalculatedInSteps.getCombinedValue())
+        .isEqualTo(checksumCalculatedAtOnce.getCombinedValue());
+  }
 }