feat: Add InlinePcesWriter

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/CommonPcesWriter.java

@@ -116,16 +116,6 @@
      */
     private final double spanOverlapFactor;
 
-    /**
-     * The highest event sequence number that has been written to the stream (but possibly not yet flushed).
-     */
-    private long lastWrittenEvent = -1;
-
-    /**
-     * The highest event sequence number that has been durably flushed to disk.
-     */
-    private long lastFlushedEvent = -1;
-
     /**
      * If true then all added events are new and need to be written to the stream. If false then all added events are
      * already durable and do not need to be written to the stream.
@@ -168,7 +158,7 @@
         averageSpanUtilization = new LongRunningAverage(pcesConfig.spanUtilizationRunningAverageLength());
 
         fileType = eventConfig.useBirthRoundAncientThreshold()
                 ? AncientMode.BIRTH_ROUND_THRESHOLD
                 : AncientMode.GENERATION_THRESHOLD;
     }
 
@@ -188,19 +178,22 @@
      * Inform the preconsensus event writer that a discontinuity has occurred in the preconsensus event stream.
      *
      * @param newOriginRound the round of the state that the new stream will be starting from
+     * @return {@code true} if this method call resulted in the current file being closed
      */
-    public void registerDiscontinuity(@NonNull final Long newOriginRound) {
+    public boolean registerDiscontinuity(@NonNull final Long newOriginRound) {
         if (!streamingNewEvents) {
             logger.error(EXCEPTION.getMarker(), "registerDiscontinuity() called while replaying events");
         }
 
         try {
             if (currentMutableFile != null) {
                 closeFile();
+                return true;
             }
         } finally {
             fileManager.registerDiscontinuity(newOriginRound);
         }
+        return false;
     }
 
     /**
@@ -211,7 +204,7 @@
      */
     public void updateNonAncientEventBoundary(@NonNull final EventWindow nonAncientBoundary) {
         if (nonAncientBoundary.getAncientThreshold() < this.nonAncientBoundary) {
             throw new IllegalArgumentException("Non-ancient boundary cannot be decreased. Current = "
                     + this.nonAncientBoundary + ", requested = " + nonAncientBoundary);
         }
 
@@ -237,10 +230,9 @@
         try {
             previousSpan = currentMutableFile.getUtilizedSpan();
             if (!bootstrapMode) {
                 averageSpanUtilization.add(previousSpan);
             }
             currentMutableFile.close();
-            lastFlushedEvent = lastWrittenEvent;
 
             fileManager.finishedWritingFile(currentMutableFile);
             currentMutableFile = null;
@@ -248,8 +240,8 @@
             // Not strictly required here, but not a bad place to ensure we delete
             // files incrementally (as opposed to deleting a bunch of files all at once).
             pruneOldFiles();
         } catch (final IOException e) {
             throw new UncheckedIOException("unable to prune files", e);
         }
     }
 
@@ -261,13 +253,13 @@
             // Don't attempt to prune files until we are done replaying the event stream (at start up).
             // Files are being iterated on a different thread, and it isn't thread safe to prune files
             // while they are being iterated.
             return;
         }
 
         try {
             fileManager.pruneOldFiles(minimumAncientIdentifierToStore);
         } catch (final IOException e) {
             throw new UncheckedIOException("unable to prune old files", e);
         }
     }
 
@@ -280,18 +272,18 @@
     public boolean prepareOutputStream(@NonNull final PlatformEvent eventToWrite) throws IOException {
         boolean fileClosed = false;
         if (currentMutableFile != null) {
             final boolean fileCanContainEvent =
                     currentMutableFile.canContain(eventToWrite.getAncientIndicator(fileType));
             final boolean fileIsFull =
                     UNIT_BYTES.convertTo(currentMutableFile.fileSize(), UNIT_MEGABYTES) >= preferredFileSizeMegabytes;
 
             if (!fileCanContainEvent || fileIsFull) {
                 closeFile();
                 fileClosed = true;
             }
 
             if (fileIsFull) {
                 bootstrapMode = false;
             }
         }
 
@@ -344,7 +336,7 @@
      * @return the type of the PCES file
      */
     public AncientMode getFileType() {
         return fileType;
     }
 
     /**
@@ -362,45 +354,9 @@
      * @return the current mutable file
      */
     public PcesMutableFile getCurrentMutableFile() {
         return currentMutableFile;
     }
 
-    /**
-     * Get the last written event.
-     *
-     * @return the last written event
-     */
-    public long getLastWrittenEvent() {
-        return lastWrittenEvent;
-    }
-
-    /**
-     * Get the last flushed event.
-     *
-     * @return the last flushed event
-     */
-    public long getLastFlushedEvent() {
-        return lastFlushedEvent;
-    }
-
-    /**
-     * Set the last written event.
-     *
-     * @param lastWrittenEvent the last written event
-     */
-    public void setLastWrittenEvent(final long lastWrittenEvent) {
-        this.lastWrittenEvent = lastWrittenEvent;
-    }
-
-    /**
-     * Set the last flushed event.
-     *
-     * @param lastFlushedEvent the last flushed event
-     */
-    public void setLastFlushedEvent(final long lastFlushedEvent) {
-        this.lastFlushedEvent = lastFlushedEvent;
-    }
-
     /**
      * Close the current mutable file.
      */
@@ -408,8 +364,8 @@
         if (currentMutableFile != null) {
             try {
                 currentMutableFile.close();
             } catch (final IOException e) {
                 throw new UncheckedIOException(e);
             }
         }
     }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/DefaultInlinePcesWriter.java

@@ -27,7 +27,7 @@
 import org.apache.logging.log4j.Logger;
 
 public class DefaultInlinePcesWriter implements InlinePcesWriter {
     private static final Logger logger = LogManager.getLogger(DefaultInlinePcesWriter.class);
 
     private final CommonPcesWriter commonPcesWriter;
     /**
@@ -37,16 +37,16 @@
      * @param fileManager     manages all preconsensus event stream files currently on disk
      */
     public DefaultInlinePcesWriter(
             @NonNull final PlatformContext platformContext, @NonNull final PcesFileManager fileManager) {
         Objects.requireNonNull(platformContext, "platformContext is required");
         Objects.requireNonNull(fileManager, "fileManager is required");
         commonPcesWriter = new CommonPcesWriter(platformContext, fileManager, true);
     }
 
     @Override
     public void beginStreamingNewEvents() {
         commonPcesWriter.beginStreamingNewEvents();
     }
 
     /**
      * {@inheritDoc}
@@ -54,33 +54,25 @@
     @NonNull
     @Override
     public PlatformEvent writeEvent(@NonNull PlatformEvent event) {
-        if (event.getStreamSequenceNumber() == PlatformEvent.NO_STREAM_SEQUENCE_NUMBER) {
-            throw new IllegalStateException("Event must have a valid stream sequence number");
-        }
-
         // if we aren't streaming new events yet, assume that the given event is already durable
         if (!commonPcesWriter.isStreamingNewEvents()) {
-            commonPcesWriter.setLastWrittenEvent(event.getStreamSequenceNumber());
-            commonPcesWriter.setLastFlushedEvent(event.getStreamSequenceNumber());
             return event;
         }
 
-        // don't do anything with ancient events
         if (event.getAncientIndicator(commonPcesWriter.getFileType()) < commonPcesWriter.getNonAncientBoundary()) {
-            throw new IllegalStateException("Ancient events should not be written to the PCES");
+            // don't do anything with ancient events
+            return event;
         }
 
         try {
             commonPcesWriter.prepareOutputStream(event);
             commonPcesWriter.getCurrentMutableFile().writeEvent(event);
-            commonPcesWriter.setLastWrittenEvent(event.getStreamSequenceNumber());
 
             commonPcesWriter.getCurrentMutableFile().flush();
-            commonPcesWriter.setLastFlushedEvent(commonPcesWriter.getLastWrittenEvent());
 
             return event;
         } catch (final IOException e) {
             throw new UncheckedIOException(e);
         }
     }
 
@@ -89,19 +81,19 @@
      */
     @Override
     public void registerDiscontinuity(@NonNull Long newOriginRound) {
         commonPcesWriter.registerDiscontinuity(newOriginRound);
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public void updateNonAncientEventBoundary(@NonNull EventWindow nonAncientBoundary) {
         commonPcesWriter.updateNonAncientEventBoundary(nonAncientBoundary);
     }
 
     @Override
     public void setMinimumAncientIdentifierToStore(@NonNull final Long minimumAncientIdentifierToStore) {
         commonPcesWriter.setMinimumAncientIdentifierToStore(minimumAncientIdentifierToStore);
     }
 }

platform-sdk/swirlds-platform-core/src/main/java/com/swirlds/platform/event/preconsensus/DefaultPcesWriter.java

@@ -38,6 +38,16 @@
 
     private static final Logger logger = LogManager.getLogger(DefaultPcesWriter.class);
 
+    /**
+     * The highest event sequence number that has been written to the stream (but possibly not yet flushed).
+     */
+    private long lastWrittenEvent = -1;
+
+    /**
+     * The highest event sequence number that has been durably flushed to disk.
+     */
+    private long lastFlushedEvent = -1;
+
     private final CommonPcesWriter commonPcesWriter;
 
     /**
@@ -55,10 +65,10 @@
      */
     public DefaultPcesWriter(
             @NonNull final PlatformContext platformContext, @NonNull final PcesFileManager fileManager) {
         Objects.requireNonNull(platformContext, "platformContext is required");
         Objects.requireNonNull(fileManager, "fileManager is required");
 
         commonPcesWriter = new CommonPcesWriter(platformContext, fileManager, false);
     }
 
     /**
@@ -66,7 +76,7 @@
      */
     @Override
     public void beginStreamingNewEvents() {
         commonPcesWriter.beginStreamingNewEvents();
     }
 
     /**
@@ -81,8 +91,8 @@
 
         // if we aren't streaming new events yet, assume that the given event is already durable
         if (!commonPcesWriter.isStreamingNewEvents()) {
-            commonPcesWriter.setLastWrittenEvent(event.getStreamSequenceNumber());
-            commonPcesWriter.setLastFlushedEvent(event.getStreamSequenceNumber());
+            lastWrittenEvent = event.getStreamSequenceNumber();
+            lastFlushedEvent = event.getStreamSequenceNumber();
             return event.getStreamSequenceNumber();
         }
 
@@ -92,13 +102,16 @@
         }
 
         try {
             final boolean fileClosed = commonPcesWriter.prepareOutputStream(event);
+            if (fileClosed) {
+                lastFlushedEvent = lastWrittenEvent;
+            }
             commonPcesWriter.getCurrentMutableFile().writeEvent(event);
-            commonPcesWriter.setLastWrittenEvent(event.getStreamSequenceNumber());
+            lastFlushedEvent = event.getStreamSequenceNumber();
 
             final boolean flushPerformed = processFlushRequests();
 
-            return fileClosed || flushPerformed ? commonPcesWriter.getLastFlushedEvent() : null;
+            return fileClosed || flushPerformed ? lastFlushedEvent : null;
         } catch (final IOException e) {
             throw new UncheckedIOException(e);
         }
@@ -110,8 +123,11 @@
     @Override
     @Nullable
     public Long registerDiscontinuity(@NonNull final Long newOriginRound) {
-        commonPcesWriter.registerDiscontinuity(newOriginRound);
-        return commonPcesWriter.getLastFlushedEvent();
+        final boolean fileClosed = commonPcesWriter.registerDiscontinuity(newOriginRound);
+        if (fileClosed) {
+            lastFlushedEvent = lastWrittenEvent;
+        }
+        return lastFlushedEvent;
     }
 
     /**
@@ -122,7 +138,7 @@
     public Long submitFlushRequest(@NonNull final Long sequenceNumber) {
         flushRequests.add(sequenceNumber);
 
-        return processFlushRequests() ? commonPcesWriter.getLastFlushedEvent() : null;
+        return processFlushRequests() ? lastFlushedEvent : null;
     }
 
     /**
@@ -131,30 +147,30 @@
      * @return true if a flush was performed, otherwise false
      */
     private boolean processFlushRequests() {
         boolean flushRequired = false;
-        while (!flushRequests.isEmpty() && flushRequests.peekFirst() <= commonPcesWriter.getLastWrittenEvent()) {
+        while (!flushRequests.isEmpty() && flushRequests.peekFirst() <= lastWrittenEvent) {
             final long flushRequest = flushRequests.removeFirst();
 
-            if (flushRequest > commonPcesWriter.getLastFlushedEvent()) {
+            if (flushRequest > lastFlushedEvent) {
                 flushRequired = true;
             }
         }
 
         if (flushRequired) {
             if (commonPcesWriter.getCurrentMutableFile() == null) {
                 logger.error(EXCEPTION.getMarker(), "Flush required, but no file is open. This should never happen");
             }
 
             try {
                 commonPcesWriter.getCurrentMutableFile().flush();
             } catch (final IOException e) {
                 throw new UncheckedIOException(e);
             }
 
-            commonPcesWriter.setLastFlushedEvent(commonPcesWriter.getLastWrittenEvent());
+            lastFlushedEvent = lastWrittenEvent;
         }
 
         return flushRequired;
     }
 
     /**
@@ -162,7 +178,7 @@
      */
     @Override
     public void updateNonAncientEventBoundary(@NonNull final EventWindow nonAncientBoundary) {
         commonPcesWriter.updateNonAncientEventBoundary(nonAncientBoundary);
     }
 
     /**
@@ -170,13 +186,13 @@
      */
     @Override
     public void setMinimumAncientIdentifierToStore(@NonNull final Long minimumAncientIdentifierToStore) {
         commonPcesWriter.setMinimumAncientIdentifierToStore(minimumAncientIdentifierToStore);
     }
 
     /**
      * Close the current mutable file.
      */
     public void closeCurrentMutableFile() {
         commonPcesWriter.closeCurrentMutableFile();
     }
 }