chore: Write marker files for sidecars (#18916)

Signed-off-by: Neeharika-Sompalli <[email protected]>

Author: Neeharika-Sompalli

hedera-node/hapi-utils/src/main/java/com/hedera/node/app/hapi/utils/exports/recordstreaming/RecordStreamingUtils.java

@@ -141,6 +141,10 @@ public static boolean isRecordFile(final String file) {
         return isRelevant(file) && !file.contains(SIDECAR_ONLY_TOKEN);
     }
 
+    public static boolean isSidecarMarkerFile(final String file) {
+        return file.contains(SIDECAR_ONLY_TOKEN) && file.contains(".mf");
+    }
+
     public static boolean isSidecarFile(final String file) {
         return isRelevant(file) && file.contains(SIDECAR_ONLY_TOKEN);
     }

hedera-node/hapi-utils/src/test/resources/forensics/CaseOfTheAbsentResult/node0/sidecar/2022-12-05T14_23_46.192841556Z_01.mf

@@ -376,12 +376,14 @@ private void closeSidecarFileWriter() {
                 sidecarFileWriter.close();
                 // get the sidecar hash
                 final Bytes sidecarHash = sidecarFileWriter.fileHash();
-                // create and add sidecar metadata to record file
+                // create and add sidecar metadata to the record file
                 if (sidecarMetadata == null) sidecarMetadata = new ArrayList<>();
                 sidecarMetadata.add(new SidecarMetadata(
                         new HashObject(HashAlgorithm.SHA_384, (int) sidecarHash.length(), sidecarHash),
                         sidecarFileWriter.id(),
                         sidecarFileWriter.types()));
+                // Add a marker file for the sidecar when it is closed
+                writeSidecarMarkerFile();
             }
         } catch (final IOException e) {
             // NOTE: Writing sidecar files really is best-effort, if it doesn't happen, we're OK with just logging the
@@ -390,6 +392,25 @@ private void closeSidecarFileWriter() {
         }
     }
 
+    /**
+     * Write a marker file for the sidecar file.
+     * This is used to indicate that the sidecar file has been closed and is ready to be uploaded.
+     * The marker file is named the same as the sidecar file, but with a .mf extension.
+     * The contents of the marker file are the hash of the sidecar file.
+     *
+     * @throws IOException if there was a problem writing the marker file
+     */
+    private void writeSidecarMarkerFile() throws IOException {
+        final Path sidecarPath = getSidecarFilePath(sidecarFileWriter.id());
+        final Path markerPath =
+                sidecarPath.resolveSibling(sidecarPath.getFileName().toString().replace(".rcd.gz", ".mf"));
+        if (Files.exists(markerPath)) {
+            logger.debug("Side‑car marker already exists: {}", markerPath);
+        } else {
+            Files.createFile(markerPath);
+        }
+    }
+
     /**
      * Get the record file path for a record file with the given consensus time
      *

hedera-node/hedera-app/src/main/java/com/hedera/node/app/records/impl/producers/formats/v6/BlockRecordWriterV6.java

@@ -360,7 +360,11 @@ void writingTest(final List<SingleTransactionRecord> singleTransactionRecords) t
 
             // Check that the sidecar file exists (if the block records produced any sidecars)
             final var sidecarPath = recordPath.getParent().resolve("sidecar/2018-08-24T16_25_42.000000890Z_01.rcd.gz");
+            final var sidecarMarker = recordPath.getParent().resolve("sidecar/2018-08-24T16_25_42.000000890Z_01.mf");
             assertThat(Files.exists(sidecarPath)).isEqualTo(hasSidecars);
+            assertThat(Files.exists(sidecarMarker)).isEqualTo(hasSidecars);
+            assertThat(anyMarkerFilesExist(recordPath.getParent().resolve("sidecar")))
+                    .isEqualTo(hasSidecars);
         }
 
         @Test
@@ -413,6 +417,8 @@ void badSidecarsIsNotFatal() throws IOException {
 
             assertThat(logCaptor.warnLogs()).hasSizeGreaterThan(0);
             assertThat(logCaptor.warnLogs()).allMatch(msg -> msg.contains("sidecar"));
+            assertThat(anyMarkerFilesExist(recordPath.getParent().resolve("sidecar")))
+                    .isFalse();
         }
     }
 
@@ -463,6 +469,17 @@ void cannotWriteToRecordFile() throws IOException {
             assertThat(logCaptor.warnLogs())
                     .matches(logs -> logs.getFirst().contains("Error closing sidecar file")
                             && logs.getLast().contains("Error closing record file"));
+            assertThat(anyMarkerFilesExist(recordPath.getParent().resolve("sidecar")))
+                    .isFalse();
+        }
+    }
+
+    private boolean anyMarkerFilesExist(Path dir) {
+        if (!dir.getFileSystem().isOpen() || Files.notExists(dir)) return false;
+        try (Stream<Path> paths = Files.walk(dir, 2)) {
+            return paths.anyMatch(p -> p.toString().endsWith(".mf"));
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
         }
     }
 }

hedera-node/hedera-app/src/test/java/com/hedera/node/app/workflows/handle/record/impl/producers/formats/v6/BlockRecordWriterV6Test.java

@@ -1,6 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 package com.hedera.services.bdd.junit.support;
 
+import static com.hedera.node.app.hapi.utils.exports.recordstreaming.RecordStreamingUtils.SIDECAR_ONLY_TOKEN;
 import static java.util.Comparator.comparing;
 
 import com.hedera.hapi.block.stream.Block;
@@ -275,6 +276,8 @@ public static long extractBlockNumber(@NonNull final String fileName) {
      * @return true if the file is a block marker file, false otherwise
      */
     public static boolean isBlockMarkerFile(@NonNull final File file) {
-        return file.isFile() && file.getName().endsWith(".mf");
+        return file.isFile()
+                && file.getName().endsWith(".mf")
+                && !file.getName().contains(SIDECAR_ONLY_TOKEN);
     }
 }

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/junit/support/BlockStreamAccess.java

@@ -2,7 +2,7 @@
 package com.hedera.services.bdd.junit.support;
 
 import static com.hedera.node.app.hapi.utils.exports.recordstreaming.RecordStreamingUtils.isRecordFile;
-import static com.hedera.node.app.hapi.utils.exports.recordstreaming.RecordStreamingUtils.isSidecarFile;
+import static com.hedera.node.app.hapi.utils.exports.recordstreaming.RecordStreamingUtils.isSidecarMarkerFile;
 import static com.hedera.services.bdd.junit.support.BlockStreamAccess.isBlockMarkerFile;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
@@ -121,7 +121,14 @@ private void exposeBlock(@NonNull final File file) {
     }
 
     private void exposeSidecars(final File file) {
-        final var contents = StreamFileAccess.ensurePresentSidecarFile(file.getAbsolutePath());
+        // Get Sidecar file path using marker file path
+        final var markerPath = file.toPath();
+        final var baseName = file.getName().replace(".mf", "");
+        final var gzPath = markerPath.resolveSibling(baseName + ".rcd.gz");
+        final var plainPath = markerPath.resolveSibling(baseName + ".rcd");
+        final var sidecarPath = Files.exists(gzPath) ? gzPath : plainPath;
+
+        final var contents = StreamFileAccess.ensurePresentSidecarFile(sidecarPath.toString());
         contents.getSidecarRecordsList().forEach(sidecar -> listeners.forEach(l -> l.onNewSidecar(sidecar)));
     }
 
@@ -138,12 +145,12 @@ private FileType typeOf(final File file) {
         // Ignore empty files, which are likely to be in the process of being written
         if (isBlockMarkerFile(file)) {
             return FileType.BLOCK_FILE;
+        } else if (isSidecarMarkerFile(file.getName())) {
+            return FileType.SIDE_CAR_FILE;
         } else if (file.length() == 0L) {
             return FileType.OTHER;
         } else if (isRecordFile(file.getName())) {
             return FileType.RECORD_STREAM_FILE;
-        } else if (isSidecarFile(file.getName())) {
-            return FileType.SIDE_CAR_FILE;
         } else {
             return FileType.OTHER;
         }

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/junit/support/StreamFileAlterationListener.java

@@ -1208,12 +1208,19 @@ public static HapiSpecOperation remembering(final Map<String, String> props, fin
     /* Stream validation. */
     public static EventualRecordStreamAssertion recordStreamMustIncludeNoFailuresFrom(
             @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion) {
+        return EventualRecordStreamAssertion.eventuallyAssertingNoFailures(assertion)
+                .withBackgroundTraffic();
+    }
+
+    public static EventualRecordStreamAssertion recordStreamMustIncludeNoFailuresWithoutBackgroundTrafficFrom(
+            @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion) {
         return EventualRecordStreamAssertion.eventuallyAssertingNoFailures(assertion);
     }
 
     public static EventualRecordStreamAssertion recordStreamMustIncludePassFrom(
             @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion) {
-        return EventualRecordStreamAssertion.eventuallyAssertingExplicitPass(assertion);
+        return EventualRecordStreamAssertion.eventuallyAssertingExplicitPass(assertion)
+                .withBackgroundTraffic();
     }
 
     /**
@@ -1225,9 +1232,37 @@ public static EventualRecordStreamAssertion recordStreamMustIncludePassFrom(
      */
     public static EventualRecordStreamAssertion recordStreamMustIncludePassFrom(
             @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion, @NonNull final Duration timeout) {
+        return recordStreamMustIncludePassFrom(assertion, timeout, true);
+    }
+
+    /**
+     * Returns an operation that asserts that the record stream must include a pass from the given assertion
+     * before its timeout elapses, and that background traffic is running.
+     * @param assertion the assertion to apply to the record stream
+     * @param timeout the timeout for the assertion
+     * @return the operation that asserts a passing record stream
+     */
+    public static EventualRecordStreamAssertion recordStreamMustIncludePassWithoutBackgroundTrafficFrom(
+            @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion, @NonNull final Duration timeout) {
+        return recordStreamMustIncludePassFrom(assertion, timeout, false);
+    }
+
+    /**
+     * Returns an operation that asserts that the record stream must include a pass from the given assertion
+     * before its timeout elapses, and if the background traffic should be running.
+     * @param assertion the assertion to apply to the record stream
+     * @param timeout the timeout for the assertion
+     * @param needsBackgroundTraffic whether background traffic should be running
+     * @return the operation that asserts a passing record stream
+     */
+    private static EventualRecordStreamAssertion recordStreamMustIncludePassFrom(
+            @NonNull final Function<HapiSpec, RecordStreamAssertion> assertion,
+            @NonNull final Duration timeout,
+            final boolean needsBackgroundTraffic) {
         requireNonNull(assertion);
         requireNonNull(timeout);
-        return EventualRecordStreamAssertion.eventuallyAssertingExplicitPass(assertion, timeout);
+        final var result = EventualRecordStreamAssertion.eventuallyAssertingExplicitPass(assertion, timeout);
+        return needsBackgroundTraffic ? result.withBackgroundTraffic() : result;
     }
 
     /**

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/spec/utilops/UtilVerbs.java

@@ -35,6 +35,8 @@ public class EventualRecordStreamAssertion extends AbstractEventualStreamAsserti
     @Nullable
     private RecordStreamAssertion assertion;
 
+    private boolean needsBackgroundTraffic = false;
+
     /**
      * Returns an {@link EventualRecordStreamAssertion} that will pass as long as the given assertion does not
      * throw an {@link AssertionError} before its timeout.
@@ -43,7 +45,7 @@ public class EventualRecordStreamAssertion extends AbstractEventualStreamAsserti
      */
     public static EventualRecordStreamAssertion eventuallyAssertingNoFailures(
             final Function<HapiSpec, RecordStreamAssertion> assertionFactory) {
-        return new EventualRecordStreamAssertion(assertionFactory, true, false);
+        return new EventualRecordStreamAssertion(assertionFactory, true, false).withBackgroundTraffic();
     }
 
     /**
@@ -67,7 +69,21 @@ public static EventualRecordStreamAssertion eventuallyAssertingExplicitPassWithR
             @NonNull final Function<HapiSpec, RecordStreamAssertion> assertionFactory,
             @NonNull final Duration timeout) {
         requireNonNull(assertionFactory);
-        return new EventualRecordStreamAssertion(assertionFactory, false, timeout, true);
+        return new EventualRecordStreamAssertion(assertionFactory, false, timeout, true).withBackgroundTraffic();
+    }
+
+    @Override
+    public boolean needsBackgroundTraffic() {
+        return needsBackgroundTraffic;
+    }
+
+    /**
+     * Returns an {@link EventualRecordStreamAssertion} enabling background traffic.
+     * @return the eventual record stream assertion with background traffic
+     */
+    public EventualRecordStreamAssertion withBackgroundTraffic() {
+        this.needsBackgroundTraffic = true;
+        return this;
     }
 
     /**

hedera-node/test-clients/src/main/java/com/hedera/services/bdd/spec/utilops/streams/assertions/EventualRecordStreamAssertion.java

@@ -36,7 +36,7 @@
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.doingContextual;
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.newKeyNamed;
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.overridingTwo;
-import static com.hedera.services.bdd.spec.utilops.UtilVerbs.recordStreamMustIncludePassFrom;
+import static com.hedera.services.bdd.spec.utilops.UtilVerbs.recordStreamMustIncludePassWithoutBackgroundTrafficFrom;
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.sourcing;
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.submitModified;
 import static com.hedera.services.bdd.spec.utilops.UtilVerbs.validateChargedUsd;
@@ -158,7 +158,7 @@ final Stream<DynamicTest> keyRotationDoesNotChangeEvmAddress() {
                 .toArray(String[]::new);
         return hapiTest(flatten(
                 cryptoTransfer(tinyBarsFromTo(GENESIS, ADDRESS_BOOK_CONTROL, 1)),
-                recordStreamMustIncludePassFrom(
+                recordStreamMustIncludePassWithoutBackgroundTrafficFrom(
                         visibleNonSyntheticItems(keyRotationsValidator(accountsToHaveKeysRotated), allTxnIds),
                         Duration.ofSeconds(15)),
                 // If the FileAlterationObserver just started the monitor, there's a chance we could miss the